feat: persist stackerdb versioning for signers/miner

* feat: persist latest chunk version in a new miner_db
* test: add test case for quicker versioning uptake

Author: kantai

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to the versioning scheme outlined in the [README.md](README.md).
 
+## [Unreleased]
+
+### Added
+- Persisted tracking of StackerDB slot versions for mining. This improves miner p2p performance.
+
 ## [3.1.0.0.9]
 
 ### Added

stacks-signer/CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to the versioning scheme outlined in the [README.md](README.md).
 
+## [Unreleased]
+
+### Added
+- Persisted tracking of StackerDB slot versions. This improves signer p2p performance.
+
 ## [3.1.0.0.9.0]
 
 ### Changed

stacks-signer/src/client/mod.rs

@@ -49,6 +49,9 @@ pub enum ClientError {
     /// Failed to sign stacker-db chunk
     #[error("Failed to sign stacker-db chunk: {0}")]
     FailToSign(#[from] StackerDBError),
+    /// Failed on a DBError
+    #[error("SignerDB database error: {0}")]
+    SignerDBError(#[from] blockstack_lib::util_lib::db::Error),
     /// Stacker-db instance rejected the chunk
     #[error("Stacker-db rejected the chunk. Reason: {0}")]
     PutChunkRejected(String),

stacks-signer/src/client/stackerdb.rs

@@ -16,6 +16,7 @@
 //
 use blockstack_lib::net::api::poststackerdbchunk::StackerDBErrorCodes;
 use clarity::codec::read_next;
+use clarity::types::chainstate::StacksPublicKey;
 use hashbrown::HashMap;
 use libsigner::{MessageSlotID, SignerMessage, SignerSession, StackerDBSession};
 use libstackerdb::{StackerDBChunkAckData, StackerDBChunkData};
@@ -25,6 +26,7 @@ use stacks_common::{debug, info, warn};
 
 use crate::client::{retry_with_exponential_backoff, ClientError};
 use crate::config::{SignerConfig, SignerConfigMode};
+use crate::signerdb::SignerDb;
 
 /// The signer StackerDB slot ID, purposefully wrapped to prevent conflation with SignerID
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Copy, PartialOrd, Ord)]
@@ -50,13 +52,13 @@ pub struct StackerDB<M: MessageSlotID + std::cmp::Eq> {
     signers_message_stackerdb_sessions: HashMap<M, StackerDBSession>,
     /// The private key used in all stacks node communications
     stacks_private_key: StacksPrivateKey,
-    /// A map of a message ID to last chunk version for each session
-    slot_versions: HashMap<M, HashMap<SignerSlotID, u32>>,
     /// The running mode of the stackerdb (whether the signer is running in dry-run or
     ///  normal operation)
     mode: StackerDBMode,
     /// The reward cycle of the connecting signer
     reward_cycle: u64,
+    /// signerdb connection
+    signer_db: SignerDb,
 }
 
 impl<M: MessageSlotID + 'static> From<&SignerConfig> for StackerDB<M> {
@@ -69,12 +71,14 @@ impl<M: MessageSlotID + 'static> From<&SignerConfig> for StackerDB<M> {
                 signer_slot_id: *signer_slot_id,
             },
         };
+        let signer_db = SignerDb::new(&config.db_path).expect("Failed to connect to SignerDb");
 
         Self::new(
             &config.node_host,
             config.stacks_private_key,
             config.mainnet,
             config.reward_cycle,
+            signer_db,
             mode,
         )
     }
@@ -89,12 +93,14 @@ impl<M: MessageSlotID + 'static> StackerDB<M> {
         is_mainnet: bool,
         reward_cycle: u64,
         signer_slot_id: SignerSlotID,
+        signer_db: SignerDb,
     ) -> Self {
         Self::new(
             host,
             stacks_private_key,
             is_mainnet,
             reward_cycle,
+            signer_db,
             StackerDBMode::Normal { signer_slot_id },
         )
     }
@@ -105,6 +111,7 @@ impl<M: MessageSlotID + 'static> StackerDB<M> {
         stacks_private_key: StacksPrivateKey,
         is_mainnet: bool,
         reward_cycle: u64,
+        signer_db: SignerDb,
         signer_mode: StackerDBMode,
     ) -> Self {
         let mut signers_message_stackerdb_sessions = HashMap::new();
@@ -117,9 +124,9 @@ impl<M: MessageSlotID + 'static> StackerDB<M> {
         Self {
             signers_message_stackerdb_sessions,
             stacks_private_key,
-            slot_versions: HashMap::new(),
             mode: signer_mode,
             reward_cycle,
+            signer_db,
         }
     }
 
@@ -160,28 +167,22 @@ impl<M: MessageSlotID + 'static> StackerDB<M> {
                 code: None,
             });
         };
+        let signer_pk = StacksPublicKey::from_private(&self.stacks_private_key);
         loop {
-            let mut slot_version = if let Some(versions) = self.slot_versions.get_mut(msg_id) {
-                if let Some(version) = versions.get(slot_id) {
-                    *version
-                } else {
-                    versions.insert(*slot_id, 0);
-                    1
-                }
-            } else {
-                let mut versions = HashMap::new();
-                versions.insert(*slot_id, 0);
-                self.slot_versions.insert(*msg_id, versions);
-                1
-            };
-
-            let mut chunk = StackerDBChunkData::new(slot_id.0, slot_version, message_bytes.clone());
-            chunk.sign(&self.stacks_private_key)?;
+            let slot_version = self
+                .signer_db
+                .get_latest_chunk_version(&signer_pk, slot_id.0)
+                .map_err(ClientError::SignerDBError)?
+                .map(|x| x.saturating_add(1))
+                .unwrap_or(0);
 
             let Some(session) = self.signers_message_stackerdb_sessions.get_mut(msg_id) else {
                 panic!("FATAL: would loop forever trying to send a message with ID {msg_id:?}, for which we don't have a session");
             };
 
+            let mut chunk = StackerDBChunkData::new(slot_id.0, slot_version, message_bytes.clone());
+            chunk.sign(&self.stacks_private_key)?;
+
             debug!(
                 "Sending a chunk to stackerdb slot ID {slot_id} with version {slot_version} and message ID {msg_id:?} to contract {:?}!\n{chunk:?}",
                 &session.stackerdb_contract_id
@@ -190,15 +191,11 @@ impl<M: MessageSlotID + 'static> StackerDB<M> {
             let send_request = || session.put_chunk(&chunk).map_err(backoff::Error::transient);
             let chunk_ack: StackerDBChunkAckData = retry_with_exponential_backoff(send_request)?;
 
-            if let Some(versions) = self.slot_versions.get_mut(msg_id) {
-                // NOTE: per the above, this is always executed
-                versions.insert(*slot_id, slot_version.saturating_add(1));
-            } else {
-                return Err(ClientError::NotConnected);
-            }
-
             if chunk_ack.accepted {
                 debug!("Chunk accepted by stackerdb: {chunk_ack:?}");
+                self.signer_db
+                    .set_latest_chunk_version(&signer_pk, slot_id.0, slot_version)
+                    .map_err(ClientError::SignerDBError)?;
                 return Ok(chunk_ack);
             } else {
                 warn!("Chunk rejected by stackerdb: {chunk_ack:?}");
@@ -208,15 +205,18 @@ impl<M: MessageSlotID + 'static> StackerDB<M> {
                     Some(StackerDBErrorCodes::DataAlreadyExists) => {
                         if let Some(slot_metadata) = chunk_ack.metadata {
                             warn!("Failed to send message to stackerdb due to wrong version number. Attempted {}. Expected {}. Retrying...", slot_version, slot_metadata.slot_version);
-                            slot_version = slot_metadata.slot_version;
+                            self.signer_db
+                                .set_latest_chunk_version(
+                                    &signer_pk,
+                                    slot_id.0,
+                                    slot_metadata.slot_version,
+                                )
+                                .map_err(ClientError::SignerDBError)?;
                         } else {
                             warn!("Failed to send message to stackerdb due to wrong version number. Attempted {}. Expected unknown version number. Incrementing and retrying...", slot_version);
-                        }
-                        if let Some(versions) = self.slot_versions.get_mut(msg_id) {
-                            // NOTE: per the above, this is always executed
-                            versions.insert(*slot_id, slot_version.saturating_add(1));
-                        } else {
-                            return Err(ClientError::NotConnected);
+                            self.signer_db
+                                .set_latest_chunk_version(&signer_pk, slot_id.0, slot_version)
+                                .map_err(ClientError::SignerDBError)?;
                         }
                     }
                     _ => {

stacks-signer/src/signerdb.rs

@@ -27,7 +27,7 @@ use blockstack_lib::util_lib::db::{
     query_row, query_rows, sqlite_open, table_exists, tx_begin_immediate, u64_to_sql,
     Error as DBError, FromRow,
 };
-use clarity::types::chainstate::{BurnchainHeaderHash, StacksAddress};
+use clarity::types::chainstate::{BurnchainHeaderHash, StacksAddress, StacksPublicKey};
 use clarity::types::Address;
 use libsigner::v0::messages::{RejectReason, RejectReasonPrefix, StateMachineUpdate};
 use libsigner::BlockProposal;
@@ -603,6 +603,14 @@ static ADD_PARENT_BURN_BLOCK_HASH_INDEX: &str = r#"
 CREATE INDEX IF NOT EXISTS burn_blocks_parent_burn_block_hash_idx on burn_blocks (parent_burn_block_hash);
 "#;
 
+static CREATE_STACKERDB_TRACKING: &str = "
+CREATE TABLE stackerdb_tracking(
+   public_key TEXT NOT NULL,
+   slot_id INTEGER NOT NULL,
+   slot_version INTEGER NOT NULL,
+   PRIMARY KEY (public_key, slot_id)
+) STRICT;";
+
 static SCHEMA_1: &[&str] = &[
     DROP_SCHEMA_0,
     CREATE_DB_CONFIG,
@@ -695,6 +703,11 @@ static SCHEMA_13: &[&str] = &[
     "INSERT INTO db_config (version) VALUES (13);",
 ];
 
+static SCHEMA_14: &[&str] = &[
+    CREATE_STACKERDB_TRACKING,
+    "INSERT INTO db_config (version) VALUES (14);",
+];
+
 impl SignerDb {
     /// The current schema version used in this build of the signer binary.
     pub const SCHEMA_VERSION: u32 = 12;
@@ -909,6 +922,20 @@ impl SignerDb {
         Ok(())
     }
 
+    /// Migrate from schema 13 to schema 14
+    fn schema_14_migration(tx: &Transaction) -> Result<(), DBError> {
+        if Self::get_schema_version(tx)? >= 14 {
+            // no migration necessary
+            return Ok(());
+        }
+
+        for statement in SCHEMA_14.iter() {
+            tx.execute_batch(statement)?;
+        }
+
+        Ok(())
+    }
+
     /// Register custom scalar functions used by the database
     fn register_scalar_functions(&self) -> Result<(), DBError> {
         // Register helper function for determining if a block is a tenure change transaction
@@ -955,7 +982,8 @@ impl SignerDb {
                 10 => Self::schema_11_migration(&sql_tx)?,
                 11 => Self::schema_12_migration(&sql_tx)?,
                 12 => Self::schema_13_migration(&sql_tx)?,
-                13 => break,
+                13 => Self::schema_14_migration(&sql_tx)?,
+                14 => break,
                 x => return Err(DBError::Other(format!(
                     "Database schema is newer than supported by this binary. Expected version = {}, Database version = {x}",
                     Self::SCHEMA_VERSION,
@@ -975,6 +1003,36 @@ impl SignerDb {
         )
     }
 
+    /// Get the latest known version from the db for the given slot_id/pk pair
+    pub fn get_latest_chunk_version(
+        &self,
+        pk: &StacksPublicKey,
+        slot_id: u32,
+    ) -> Result<Option<u32>, DBError> {
+        self.db
+            .query_row(
+                "SELECT slot_version FROM stackerdb_tracking WHERE public_key = ? AND slot_id = ?",
+                params![pk.to_hex(), slot_id],
+                |row| row.get(0),
+            )
+            .optional()
+            .map_err(DBError::from)
+    }
+
+    /// Set the latest known version for the given slot_id/pk pair
+    pub fn set_latest_chunk_version(
+        &self,
+        pk: &StacksPublicKey,
+        slot_id: u32,
+        slot_version: u32,
+    ) -> Result<(), DBError> {
+        self.db.execute(
+            "INSERT OR REPLACE INTO stackerdb_tracking (public_key, slot_id, slot_version) VALUES (?, ?, ?)",
+            params![pk.to_hex(), slot_id, slot_version],
+        )?;
+        Ok(())
+    }
+
     /// Get the signer state for the provided reward cycle if it exists in the database
     pub fn get_encrypted_signer_state(
         &self,

testnet/stacks-node/src/nakamoto_node.rs

@@ -43,6 +43,7 @@ use crate::run_loop::nakamoto::{Globals, RunLoop};
 use crate::run_loop::RegisteredKey;
 
 pub mod miner;
+pub mod miner_db;
 pub mod peer;
 pub mod relayer;
 pub mod signer_coordinator;

testnet/stacks-node/src/nakamoto_node/miner.rs

@@ -58,6 +58,7 @@ use stacks_common::types::{PrivateKey, StacksEpochId};
 use stacks_common::util::tests::TestFlag;
 use stacks_common::util::vrf::VRFProof;
 
+use super::miner_db::MinerDB;
 use super::relayer::{MinerStopHandle, RelayerThread};
 use super::{Config, Error as NakamotoNodeError, EventDispatcher, Keychain};
 use crate::nakamoto_node::signer_coordinator::SignerCoordinator;
@@ -227,6 +228,8 @@ pub struct BlockMinerThread {
     abort_flag: Arc<AtomicBool>,
     /// Should the nonce cache be reset before mining the next block?
     reset_nonce_cache: bool,
+    /// Storage for persisting non-confidential miner information
+    miner_db: MinerDB,
 }
 
 impl BlockMinerThread {
@@ -239,8 +242,8 @@ impl BlockMinerThread {
         parent_tenure_id: StacksBlockId,
         burn_tip_at_start: &ConsensusHash,
         reason: MinerReason,
-    ) -> BlockMinerThread {
-        BlockMinerThread {
+    ) -> Result<BlockMinerThread, NakamotoNodeError> {
+        Ok(BlockMinerThread {
             config: rt.config.clone(),
             globals: rt.globals.clone(),
             keychain: rt.keychain.clone(),
@@ -261,7 +264,8 @@ impl BlockMinerThread {
             tenure_cost: ExecutionCost::ZERO,
             tenure_budget: ExecutionCost::ZERO,
             reset_nonce_cache: true,
-        }
+            miner_db: MinerDB::open_with_config(&rt.config)?,
+        })
     }
 
     pub fn get_abort_flag(&self) -> Arc<AtomicBool> {
@@ -843,6 +847,7 @@ impl BlockMinerThread {
             stackerdbs,
             &self.globals.counters,
             &self.burn_election_block,
+            &self.miner_db,
         )
     }
 
@@ -1058,6 +1063,7 @@ impl BlockMinerThread {
             chain_state.mainnet,
             &mut miners_session,
             &self.burn_election_block.consensus_hash,
+            &self.miner_db,
         )
     }