Quota follow up: Buddy group accounting mode, refactoring, logging

Cargo.toml

@@ -17,7 +17,7 @@ itertools = "0"
 libc = "0"
 log = { version = "0", features = ["std"] }
 prost = "0.14"
-protobuf = { git = "https://github.com/thinkparq/protobuf", rev = "4d5e5db085065acbbaa5bb76ce4b81d6d733e446" }
+protobuf = { git = "https://github.com/thinkparq/protobuf", rev = "25d79b6293ee46e1fddf832874836472951aeb29" }
 regex = "1"
 ring = "0"
 rusqlite = { version = "0", features = ["bundled", "vtab", "array", "fallible_uint"] }

mgmtd/src/app.rs

@@ -10,7 +10,7 @@ use anyhow::Result;
 use protobuf::license::GetCertDataResult;
 pub(crate) use runtime::RuntimeApp;
 use rusqlite::{Connection, Transaction};
-use shared::bee_msg::Msg;
+use shared::bee_msg::{Header, Msg};
 use shared::bee_serde::{Deserializable, Serializable};
 use shared::types::{NodeId, NodeType, Uid};
 use std::fmt::Debug;
@@ -50,6 +50,13 @@ pub(crate) trait App: Debug + Clone + Send + 'static {
     ) -> impl Future<Output = Result<R>> + Send;
 
     // BeeMsg communication
+    //
+    /// Send a [Msg] to a node via TCP and receive the response
+    fn request_with_header<M: Msg + Serializable, R: Msg + Deserializable>(
+        &self,
+        node_uid: Uid,
+        msg: &M,
+    ) -> impl Future<Output = Result<(R, Header)>> + Send;
 
     /// Send a [Msg] to a node via TCP and receive the response
     fn request<M: Msg + Serializable, R: Msg + Deserializable>(

mgmtd/src/app/runtime.rs

@@ -116,12 +116,20 @@ impl App for RuntimeApp {
         Connections::conn(&self.db, op).await
     }
 
+    async fn request_with_header<M: Msg + Serializable, R: Msg + Deserializable>(
+        &self,
+        node_uid: Uid,
+        msg: &M,
+    ) -> Result<(R, Header)> {
+        Pool::request(&self.conn, node_uid, msg).await
+    }
+
     async fn request<M: Msg + Serializable, R: Msg + Deserializable>(
         &self,
         node_uid: Uid,
         msg: &M,
     ) -> Result<R> {
-        Pool::request(&self.conn, node_uid, msg).await
+        Pool::request(&self.conn, node_uid, msg).await.map(|e| e.0)
     }
 
     async fn send_notifications<M: Msg + Serializable>(

mgmtd/src/app/test.rs

@@ -127,6 +127,14 @@ impl App for TestApp {
         Connections::conn(&self.db, op).await
     }
 
+    async fn request_with_header<M: Msg + Serializable, R: Msg + Deserializable>(
+        &self,
+        node_uid: Uid,
+        msg: &M,
+    ) -> Result<(R, Header)> {
+        Ok((self.request(node_uid, msg).await?, Header::default()))
+    }
+
     async fn request<M: Msg + Serializable, R: Msg + Deserializable>(
         &self,
         _node_uid: Uid,

mgmtd/src/bee_msg/change_target_consistency_states.rs

@@ -65,8 +65,8 @@ doesn't match stored state {old_stored}, no consistency state changes will be ma
             })
             .await?;
 
-        log::debug!(
-            "Updated target states for {:?} targets {:?}, {} consistency states and {reachabilities_changed} reachability states changed",
+        log::info!(
+            "Updated {:?} targets' ({:?}) states: {} consistency states and {reachabilities_changed} reachability states changed",
             self.node_type,
             self.target_ids,
             consistencies_changed.unwrap_or(0)

mgmtd/src/bee_msg/request_exceeded_quota.rs

@@ -30,20 +30,13 @@ impl HandleWithResponse for RequestExceededQuota {
                     )?
                 };
 
+                // Query the exceeded ids matching the request
                 let exceeded_quota_ids = tx.query_map_collect(
-                    sql!(
-                        "SELECT DISTINCT e.quota_id FROM quota_usage AS e
-                        INNER JOIN targets AS st USING(node_type, target_id)
-                        LEFT JOIN quota_default_limits AS d USING(id_type, quota_type, pool_id)
-                        LEFT JOIN quota_limits AS l USING(quota_id, id_type, quota_type, pool_id)
-                        WHERE e.id_type = ?1 AND e.quota_type = ?2 AND st.pool_id = ?3
-                        GROUP BY e.quota_id, e.id_type, e.quota_type, st.pool_id
-                        HAVING SUM(e.value) > COALESCE(l.value, d.value)"
-                    ),
+                    crate::quota::EXCEEDED_QUOTA_IDS_SQL,
                     params![
                         self.id_type.sql_variant(),
                         self.quota_type.sql_variant(),
-                        pool_id
+                        pool_id,
                     ],
                     |row| row.get(0),
                 )?;
@@ -84,7 +77,7 @@ mod test {
                     pool_id: 1,
                     target_id: 0,
                 },
-                &[2, 4, 10],
+                &[2, 4, 10, 51],
             ),
             (
                 RequestExceededQuota {

mgmtd/src/bee_msg/set_storage_target_info.rs

@@ -15,6 +15,8 @@ impl HandleWithResponse for SetStorageTargetInfo {
         fail_on_pre_shutdown(app)?;
 
         let node_type = self.node_type;
+        let target_ids: Vec<_> = self.info.iter().map(|e| e.target_id).collect();
+
         app.write_tx(move |tx| {
             db::target::get_and_update_capacities(
                 tx,
@@ -34,7 +36,7 @@ impl HandleWithResponse for SetStorageTargetInfo {
         })
         .await?;
 
-        log::debug!("Updated {node_type:?} target info");
+        log::info!("Updated {node_type:?} targets' ({target_ids:?}) info and capacities");
 
         // in the old mgmtd, a notice to refresh cap pools is sent out here if a cap pool
         // changed I consider this being to expensive to check here and just don't

mgmtd/src/config.rs

@@ -451,6 +451,10 @@ impl Config {
             bail!("Provided file system UUID is not a valid v4 UUID");
         }
 
+        if self.connection_limit < 1 {
+            bail!("Connection limit cannot be smaller than 1");
+        }
+
         if self.quota_enforce && !self.quota_enable {
             bail!("Quota enforcement requires quota being enabled");
         }

mgmtd/src/db/buddy_group.rs

@@ -48,6 +48,7 @@ pub(crate) fn insert(
     node_type: NodeTypeServer,
     p_target_id: TargetId,
     s_target_id: TargetId,
+    quota_accounting: Option<BuddyGroupQuotaAccounting>,
 ) -> Result<(Uid, BuddyGroupId)> {
     let group_id = if group_id == 0 {
         misc::find_new_id(tx, "buddy_groups", "group_id", node_type.into(), 1..=0xFFFF)?
@@ -128,20 +129,33 @@ pub(crate) fn insert(
         None
     };
 
+    // Quota accounting only applies to storage groups
+    let quota_accounting = match (node_type, quota_accounting) {
+        (NodeTypeServer::Meta, None) => None,
+        (NodeTypeServer::Meta, Some(_)) => {
+            bail!("The quota accounting mode can only be set for storage buddy groups")
+        }
+        (NodeTypeServer::Storage, None) => {
+            bail!("The quota accounting mode must be set for storage buddy groups");
+        }
+        (NodeTypeServer::Storage, Some(q)) => Some(q),
+    };
+
     // Insert generic buddy group
     tx.execute(
         sql!(
             "INSERT INTO buddy_groups
-            (group_uid, node_type, group_id, p_target_id, s_target_id, pool_id)
-            VALUES (?1, ?2, ?3, ?4, ?5, ?6)"
+            (group_uid, node_type, group_id, p_target_id, s_target_id, pool_id, quota_accounting)
+            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"
         ),
         params![
             new_uid,
             node_type.sql_variant(),
             group_id,
             p_target_id,
             s_target_id,
-            pool_id
+            pool_id,
+            quota_accounting.map(|e| e.sql_variant())
         ],
     )?;
 
@@ -298,6 +312,7 @@ mod test {
                 NodeTypeServer::Meta,
                 3,
                 4,
+                None,
             )
             .unwrap();
             super::insert(
@@ -307,13 +322,14 @@ mod test {
                 NodeTypeServer::Storage,
                 3,
                 7,
+                None,
             )
             .unwrap_err();
 
             let meta_groups = get_with_type(tx, NodeTypeServer::Meta).unwrap();
             let storage_groups = get_with_type(tx, NodeTypeServer::Storage).unwrap();
 
-            assert_eq!(2, meta_groups.len());
+            assert_eq!(3, meta_groups.len());
             assert_eq!(2, storage_groups.len());
             assert!(meta_groups.iter().any(|e| e.0 == 1234));
         })

mgmtd/src/db/import_v7.rs

@@ -255,6 +255,7 @@ fn buddy_groups(tx: &Transaction, f: &Path, nt: NodeTypeServer) -> Result<()> {
             nt,
             BuddyGroupId::from_str_radix(p_id.trim(), 16)?,
             BuddyGroupId::from_str_radix(s_id.trim(), 16)?,
+            matches!(nt, NodeTypeServer::Storage).then_some(BuddyGroupQuotaAccounting::Both),
         )?;
     }
 

mgmtd/src/db/misc.rs

@@ -132,7 +132,7 @@ mod test {
             // New max id
             let new_id =
                 super::find_new_id(tx, "targets", "target_id", NodeType::Meta, 1..=100).unwrap();
-            assert_eq!(new_id, 5);
+            assert_eq!(new_id, 6);
             // New min ID in a non-empty range
             let new_id =
                 super::find_new_id(tx, "targets", "target_id", NodeType::Meta, 0..=4).unwrap();

mgmtd/src/db/node.rs

@@ -208,7 +208,7 @@ mod test {
     #[test]
     fn insert_get_delete() {
         with_test_data(|tx| {
-            assert_eq!(5, get_with_type(tx, NodeType::Meta).unwrap().len());
+            assert_eq!(7, get_with_type(tx, NodeType::Meta).unwrap().len());
             let node = insert(
                 tx,
                 1234,
@@ -233,11 +233,11 @@ mod test {
                 10000,
             )
             .unwrap_err();
-            assert_eq!(6, get_with_type(tx, NodeType::Meta).unwrap().len());
+            assert_eq!(8, get_with_type(tx, NodeType::Meta).unwrap().len());
 
             delete(tx, node.uid).unwrap();
             delete(tx, node.uid).unwrap_err();
-            assert_eq!(5, get_with_type(tx, NodeType::Meta).unwrap().len());
+            assert_eq!(7, get_with_type(tx, NodeType::Meta).unwrap().len());
         });
     }
 

mgmtd/src/db/schema/7.sql

@@ -0,0 +1,7 @@
+-- This can be NULL as it only applies to storage groups
+ALTER table buddy_groups ADD COLUMN quota_accounting INTEGER;
+-- Make sure existing mirrored targets don't change behavior automatically
+UPDATE buddy_groups SET quota_accounting = 2 WHERE node_type = 2;
+
+ALTER table buddy_groups ADD CONSTRAINT quota_accounting_null
+CHECK ((node_type == 2) == (quota_accounting IS NOT NULL));

mgmtd/src/db/schema/test_data.sql

@@ -11,6 +11,8 @@ INSERT INTO entities (uid, entity_type, alias) VALUES
     (101002, 1, "meta_node_2"),
     (101003, 1, "meta_node_3"),
     (101004, 1, "meta_node_4"),
+    (101005, 1, "meta_node_5"),
+    (101013, 1, "meta_node_13"),
     (101099, 1, "meta_node_no_target"),
     (102001, 1, "storage_node_1"),
     (102002, 1, "storage_node_2"),
@@ -27,6 +29,8 @@ INSERT INTO nodes (node_uid, node_id, node_type, port, last_contact) VALUES
     (101002, 2, 1, 8005, DATETIME("NOW")),
     (101003, 3, 1, 8005, DATETIME("NOW")),
     (101004, 4, 1, 8005, DATETIME("NOW")),
+    (101005, 5, 1, 8005, DATETIME("NOW")),
+    (101013, 13, 1, 8005, DATETIME("NOW")),
 
     (101099, 99, 1, 8005, DATETIME("NOW")),
 
@@ -83,6 +87,8 @@ INSERT INTO entities (uid, entity_type, alias) VALUES
     (201002, 2, "meta_target_2"),
     (201003, 2, "meta_target_3"),
     (201004, 2, "meta_target_4"),
+    (201005, 2, "meta_target_5"),
+    (201013, 2, "meta_target_13"),
 
     (202001, 2, "storage_target_1"),
     (202002, 2, "storage_target_2"),
@@ -110,6 +116,8 @@ free_space, free_inodes, consistency, last_update) VALUES
     (201002, 1, 2, 2, NULL, 1000000, 1000000, 550000, 550000, 1, DATETIME("NOW")),
     (201003, 1, 3, 3, NULL, 1000000, 1000000, 550000, 550000, 1, DATETIME("NOW")),
     (201004, 1, 4, 4, NULL, 1000000, 1000000, 450000, 450000, 1, DATETIME("NOW")),
+    (201005, 1, 5, 3, NULL, 1000000, 1000000, 550000, 550000, 1, DATETIME("NOW")),
+    (201013, 1, 13, 4, NULL, 1000000, 1000000, 450000, 450000, 1, DATETIME("NOW")),
 
     (202001, 2, 1, 1, 1, 1000000, 1000000, 450000, 450000, 1, DATETIME("NOW")),
     (202002, 2, 2, 1, 2, 1000000, 1000000, 500000, 500000, 1, DATETIME("NOW")),
@@ -133,14 +141,17 @@ free_space, free_inodes, consistency, last_update) VALUES
 
 INSERT INTO entities (uid, entity_type, alias) VALUES
     (301001, 4, "meta_buddy_group_1"),
+    (301002, 4, "meta_buddy_group_2"),
     (302001, 4, "storage_buddy_group_1"),
     (302002, 4, "storage_buddy_group_2")
 ;
 
-INSERT INTO buddy_groups (group_uid, node_type, group_id, p_target_id, s_target_id, pool_id) VALUES
-    (301001, 1, 1, 1, 2, NULL),
-    (302001, 2, 1, 1, 5, 1),
-    (302002, 2, 2, 9, 13, 1)
+INSERT INTO buddy_groups (group_uid, node_type, group_id, p_target_id, s_target_id, pool_id, quota_accounting) VALUES
+    (301001, 1, 1, 1, 2, NULL, NULL),
+    -- this meta buddy group tests target id separation from storage buddy groups
+    (301002, 1, 2, 5, 13, NULL, NULL),
+    (302001, 2, 1, 1, 5, 1, 1),
+    (302002, 2, 2, 9, 13, 1, 2)
 ;
 
 
@@ -192,5 +203,13 @@ INSERT INTO quota_usage (quota_id, id_type, quota_type, target_id, value) VALUES
     (10, 2, 1, 2, 999999999),
     (10, 1, 2, 2, 999999999),
     (10, 2, 2, 2, 999999999),
-    (20, 1, 1, 2, 101)
+    (20, 1, 1, 2, 101),
+    -- 1 + 5 has quota accounting mode "primary", counted once, not exceeeded
+    (50, 1, 1, 1, 800),
+    (50, 1, 1, 5, 800),
+    -- 9 + 13 has quota accounting mode "both", counted twice, exceeded
+    (51, 1, 1, 9, 800),
+    (51, 1, 1, 13, 800),
+    -- target 13 exists both as meta and storage, this tests their correct separation - not exceeded
+    (52, 1, 1, 13, 800)
 ;