Merge pull request #6106 from kantai/feat/miners-versioning

Feat: persist stackerdb versioning for signers and miner

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,21 @@ 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(|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 +190,10 @@ 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)?;
                 return Ok(chunk_ack);
             } else {
                 warn!("Chunk rejected by stackerdb: {chunk_ack:?}");
@@ -208,15 +203,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,
+                            )?;
                         } 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,
+                            )?;
                         }
                     }
                     _ => {

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,

stackslib/src/burnchains/bitcoin/blocks.rs

@@ -285,6 +285,9 @@ impl BitcoinBlockParser {
                     test_debug!("Data output does not use a standard OP_RETURN");
                     return None;
                 }
+                if data.len() <= MAGIC_BYTES_LENGTH {
+                    return None;
+                }
                 if !data.starts_with(self.magic_bytes.as_bytes()) {
                     test_debug!("Data output does not start with magic bytes");
                     return None;

stackslib/src/net/api/tests/mod.rs

@@ -16,6 +16,8 @@
 
 use std::net::SocketAddr;
 use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::{Duration, Instant};
 
 use clarity::vm::costs::ExecutionCost;
 use clarity::vm::types::{QualifiedContractIdentifier, StacksAddressExtensions};
@@ -1060,16 +1062,20 @@ impl<'a> TestRPC<'a> {
     }
 
     pub fn run(self, requests: Vec<StacksHttpRequest>) -> Vec<StacksHttpResponse> {
-        self.run_with_observer(requests, None)
+        self.run_with_observer(requests, None, |_, _| true)
     }
 
     /// Run zero or more HTTP requests on this setup RPC test harness.
     /// Return the list of responses.
-    pub fn run_with_observer(
+    pub fn run_with_observer<F>(
         self,
         requests: Vec<StacksHttpRequest>,
         event_observer: Option<&dyn MemPoolEventDispatcher>,
-    ) -> Vec<StacksHttpResponse> {
+        wait_for: F,
+    ) -> Vec<StacksHttpResponse>
+    where
+        F: Fn(&mut TestPeer, &mut TestPeer) -> bool,
+    {
         let mut peer_1 = self.peer_1;
         let mut peer_2 = self.peer_2;
         let peer_1_indexer = self.peer_1_indexer;
@@ -1079,7 +1085,14 @@ impl<'a> TestRPC<'a> {
         let unconfirmed_state = self.unconfirmed_state;
 
         let mut responses = vec![];
-        for request in requests.into_iter() {
+        for (ix, request) in requests.into_iter().enumerate() {
+            let start = Instant::now();
+            while !wait_for(&mut peer_1, &mut peer_2) {
+                if start.elapsed() > Duration::from_secs(120) {
+                    panic!("Timed out waiting for wait_for check to pass");
+                }
+                thread::sleep(Duration::from_secs(1));
+            }
             peer_1.refresh_burnchain_view();
             peer_2.refresh_burnchain_view();
 

stackslib/src/net/api/tests/postblock_proposal.rs

@@ -46,7 +46,7 @@ use crate::net::api::*;
 use crate::net::connection::ConnectionOptions;
 use crate::net::httpcore::{RPCRequestHandler, StacksHttp, StacksHttpRequest};
 use crate::net::relay::Relayer;
-use crate::net::test::TestEventObserver;
+use crate::net::test::{TestEventObserver, TestPeer};
 use crate::net::ProtocolFamily;
 
 #[warn(unused)]
@@ -395,7 +395,11 @@ fn test_try_make_response() {
     let proposal_observer = Arc::clone(&observer.proposal_observer);
 
     info!("Run requests with observer");
-    let responses = rpc_test.run_with_observer(requests, Some(&observer));
+    let wait_for = |peer_1: &mut TestPeer, peer_2: &mut TestPeer| {
+        !peer_1.network.is_proposal_thread_running() && !peer_2.network.is_proposal_thread_running()
+    };
+
+    let responses = rpc_test.run_with_observer(requests, Some(&observer), wait_for);
 
     for response in responses.iter().take(3) {
         assert_eq!(response.preamble().status_code, 202);
@@ -567,8 +571,12 @@ fn replay_validation_test(
     let observer = ProposalTestObserver::new();
     let proposal_observer = Arc::clone(&observer.proposal_observer);
 
+    let wait_for = |peer_1: &mut TestPeer, peer_2: &mut TestPeer| {
+        !peer_1.network.is_proposal_thread_running() && !peer_2.network.is_proposal_thread_running()
+    };
+
     info!("Run request with observer for replay mismatch test");
-    let responses = rpc_test.run_with_observer(requests, Some(&observer));
+    let responses = rpc_test.run_with_observer(requests, Some(&observer), wait_for);
 
     // Expect 202 Accepted initially
     assert_eq!(responses[0].preamble().status_code, 202);