Store each relation child in its own key (kaspanet#325)

* Store each relation child in its own key

* Database version upgrade logic

* Improve staging relations perf

* Implement CachedDbSetAccess

* Use BlockHasher for children store

* Pass children readlock

* clippy fix

* Use default Debug impl for ReadLock

* Address review comments

* Remove relations rwlock

* Get rid of relations service

* Use RefCell instead of Mutex in MemoryRelationsStore and StagingRelationsStore

* fix clippy warnings

* fix simpa for low delay values

* Improve delete_children n StagingRelationsStore

* Suggestion for removing the need for `RefCell` (#4)

* prep for ref-cell removal

* remove ref-cell from StagingRelationsStore

* remove ref-cell from MemoryRelationsStore

* add comment

* update comment

* flatten staging (semantic change only)

* unify deletions

* use correct prefix

* bug fix: add to child deletions even if not removed from insertions

* remove unused API method

* fix user msg

* add simpa as test

* Revert "Get rid of relations service"

This reverts commit e8f61b1.

* Revert "Remove relations rwlock"

This reverts commit 0a4c5dd.

* Remove redundant ChildKey

* set access:
- cache the new item only if the set entry already exists in the cache
- fix bug in delete_bucket where set was emptied in cache but the entry was not removed

* bug fix: make sure to propagate key not found err if staging has no data for this hash

* clean

* Remove redundant comment

---------

Co-authored-by: Michael Sutton <[email protected]>

components/consensusmanager/src/session.rs

@@ -308,7 +308,7 @@ impl ConsensusSessionOwned {
         self.clone().spawn_blocking(move |c| c.get_ghostdag_data(hash)).await
     }
 
-    pub async fn async_get_block_children(&self, hash: Hash) -> Option<Arc<Vec<Hash>>> {
+    pub async fn async_get_block_children(&self, hash: Hash) -> Option<Vec<Hash>> {
         self.clone().spawn_blocking(move |c| c.get_block_children(hash)).await
     }
 

consensus/core/src/api/mod.rs

@@ -242,7 +242,7 @@ pub trait ConsensusApi: Send + Sync {
         unimplemented!()
     }
 
-    fn get_block_children(&self, hash: Hash) -> Option<Arc<Vec<Hash>>> {
+    fn get_block_children(&self, hash: Hash) -> Option<Vec<Hash>> {
         unimplemented!()
     }
 

consensus/src/consensus/factory.rs

@@ -40,7 +40,7 @@ pub enum ConsensusEntryType {
     New(ConsensusEntry),
 }
 
-#[derive(Serialize, Deserialize, Clone, Default)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct MultiConsensusMetadata {
     current_consensus_key: Option<u64>,
     staging_consensus_key: Option<u64>,
@@ -54,6 +54,20 @@ pub struct MultiConsensusMetadata {
     version: u32,
 }
 
+const LATEST_DB_VERSION: u32 = 1;
+impl Default for MultiConsensusMetadata {
+    fn default() -> Self {
+        Self {
+            current_consensus_key: Default::default(),
+            staging_consensus_key: Default::default(),
+            max_key_used: Default::default(),
+            is_archival_node: Default::default(),
+            props: Default::default(),
+            version: LATEST_DB_VERSION,
+        }
+    }
+}
+
 #[derive(Clone)]
 pub struct MultiConsensusManagementStore {
     db: Arc<DB>,
@@ -199,6 +213,14 @@ impl MultiConsensusManagementStore {
             self.metadata.write(BatchDbWriter::new(&mut batch), &metadata).unwrap();
         }
     }
+
+    pub fn should_upgrade(&self) -> StoreResult<bool> {
+        match self.metadata.read() {
+            Ok(data) => Ok(data.version != LATEST_DB_VERSION),
+            Err(StoreError::KeyNotFound(_)) => Ok(false),
+            Err(err) => Err(err),
+        }
+    }
 }
 
 pub struct Factory {

consensus/src/consensus/mod.rs

@@ -759,8 +759,12 @@ impl ConsensusApi for Consensus {
         Ok((&*ghostdag).into())
     }
 
-    fn get_block_children(&self, hash: Hash) -> Option<Arc<Vec<Hash>>> {
-        self.services.relations_service.get_children(hash).unwrap_option()
+    fn get_block_children(&self, hash: Hash) -> Option<Vec<Hash>> {
+        self.services
+            .relations_service
+            .get_children(hash)
+            .unwrap_option()
+            .map(|children| children.read().iter().copied().collect_vec())
     }
 
     fn get_block_parents(&self, hash: Hash) -> Option<Arc<Vec<Hash>>> {

consensus/src/model/services/relations.rs

@@ -1,5 +1,6 @@
 use crate::model::stores::relations::RelationsStoreReader;
-use kaspa_database::prelude::StoreError;
+use kaspa_consensus_core::BlockHashSet;
+use kaspa_database::prelude::{ReadLock, StoreError, StoreResult};
 use kaspa_hashes::Hash;
 use parking_lot::RwLock;
 use std::sync::Arc;
@@ -22,7 +23,7 @@ impl<T: RelationsStoreReader> RelationsStoreReader for MTRelationsService<T> {
         self.store.read()[self.level].get_parents(hash)
     }
 
-    fn get_children(&self, hash: Hash) -> Result<kaspa_consensus_core::blockhash::BlockHashes, StoreError> {
+    fn get_children(&self, hash: Hash) -> StoreResult<ReadLock<BlockHashSet>> {
         self.store.read()[self.level].get_children(hash)
     }
 

consensus/src/model/stores/children.rs

@@ -0,0 +1,79 @@
+use kaspa_consensus_core::BlockHashSet;
+use kaspa_consensus_core::BlockHasher;
+use kaspa_consensus_core::BlockLevel;
+use kaspa_database::prelude::BatchDbWriter;
+use kaspa_database::prelude::CachedDbSetAccess;
+use kaspa_database::prelude::DbWriter;
+use kaspa_database::prelude::ReadLock;
+use kaspa_database::prelude::StoreError;
+use kaspa_database::prelude::StoreResult;
+use kaspa_database::prelude::DB;
+use kaspa_database::registry::DatabaseStorePrefixes;
+use kaspa_hashes::Hash;
+use rocksdb::WriteBatch;
+use std::sync::Arc;
+
+pub trait ChildrenStoreReader {
+    fn get(&self, hash: Hash) -> StoreResult<ReadLock<BlockHashSet>>;
+}
+
+pub trait ChildrenStore {
+    fn insert_child(&mut self, writer: impl DbWriter, parent: Hash, child: Hash) -> Result<(), StoreError>;
+    fn delete_child(&mut self, writer: impl DbWriter, parent: Hash, child: Hash) -> Result<(), StoreError>;
+}
+
+/// A DB + cache implementation of `DbChildrenStore` trait, with concurrency support.
+#[derive(Clone)]
+pub struct DbChildrenStore {
+    db: Arc<DB>,
+    access: CachedDbSetAccess<Hash, Hash, BlockHasher, BlockHasher>,
+}
+
+impl DbChildrenStore {
+    pub fn new(db: Arc<DB>, level: BlockLevel, cache_size: u64) -> Self {
+        let lvl_bytes = level.to_le_bytes();
+        Self {
+            db: Arc::clone(&db),
+            access: CachedDbSetAccess::new(
+                db,
+                cache_size,
+                DatabaseStorePrefixes::RelationsChildren.into_iter().chain(lvl_bytes).collect(),
+            ),
+        }
+    }
+
+    pub fn with_prefix(db: Arc<DB>, prefix: &[u8], cache_size: u64) -> Self {
+        let db_prefix = prefix.iter().copied().chain(DatabaseStorePrefixes::RelationsChildren).collect();
+        Self { db: Arc::clone(&db), access: CachedDbSetAccess::new(db, cache_size, db_prefix) }
+    }
+
+    pub fn insert_batch(&self, batch: &mut WriteBatch, parent: Hash, child: Hash) -> Result<(), StoreError> {
+        self.access.write(BatchDbWriter::new(batch), parent, child)?;
+        Ok(())
+    }
+
+    pub(crate) fn delete_children(&self, mut writer: impl DbWriter, parent: Hash) -> Result<(), StoreError> {
+        self.access.delete_bucket(&mut writer, parent)
+    }
+
+    pub(crate) fn prefix(&self) -> &[u8] {
+        self.access.prefix()
+    }
+}
+
+impl ChildrenStoreReader for DbChildrenStore {
+    fn get(&self, parent: Hash) -> StoreResult<ReadLock<BlockHashSet>> {
+        self.access.read(parent)
+    }
+}
+
+impl ChildrenStore for DbChildrenStore {
+    fn insert_child(&mut self, writer: impl DbWriter, parent: Hash, child: Hash) -> Result<(), StoreError> {
+        self.access.write(writer, parent, child)?;
+        Ok(())
+    }
+
+    fn delete_child(&mut self, writer: impl DbWriter, parent: Hash, child: Hash) -> Result<(), StoreError> {
+        self.access.delete(writer, parent, child)
+    }
+}

consensus/src/model/stores/mod.rs

@@ -1,6 +1,7 @@
 pub mod acceptance_data;
 pub mod block_transactions;
 pub mod block_window_cache;
+pub mod children;
 pub mod daa;
 pub mod selected_chain;
 use std::{fmt::Display, mem::size_of};