feat: stackerdb bandwidth config

changelog.d/stackerdb-bandwith-config.changed

@@ -0,0 +1,4 @@
+* Exposed connection options `max_stackerdb_push_bandwidth`, `max_nakamoto_block_push_bandwidth`, `max_transaction_push_bandwidth` as TOML configuration to set bandwidth cap on message reception.
+* Set connection option `max_stackerdb_push_bandwidth` default to 4 MB/seconds
+* Enforced StackerDB message chunk-size check against replica configuration instead of statically against `STACKERDB_MAX_CHUNK_SIZE`
+* Improved StackerDB configuration creation for miner to set chunk-size to `STACKERDB_MAX_CHUNK_SIZE` instead of `MAX_PAYLOAD_LEN`

stacks-common/src/util/macros.rs

@@ -19,6 +19,21 @@ pub fn is_big_endian() -> bool {
     u32::from_be(0x1Au32) == 0x1Au32
 }
 
+/// One mebibyte (1024 * 1024), for expressing sizes like `MB!(1)`.
+///
+/// Binary byte-size unit. Expands to a plain integer arithmetic expression, so
+/// the result type is inferred from the usage site. Usable for `u32`, `u64`,
+/// `usize`, etc. without casts.
+///
+/// Overflow behavior is identical to writing N * 1024 * 1024 directly and
+/// depends on the inferred integer type.
+#[macro_export]
+macro_rules! MB {
+    ($n:expr) => {
+        ($n * 1024 * 1024)
+    };
+}
+
 /// Define an iterable enum: an enum where each variant is an atomic
 /// type (i.e., has no paramters), and the variants can be iterated over
 /// with an Enum::ALL const
@@ -810,6 +825,7 @@ mod tests {
         assert_eq!(Some(MyEnum::Variant2), MyEnum::lookup_by_name("variant2"));
         assert_eq!(None, MyEnum::lookup_by_name("inexistent"));
     }
+
     #[test]
     fn test_macro_define_named_enum_with_docs() {
         define_named_enum!(
@@ -831,4 +847,25 @@ mod tests {
         assert_eq!(Some(MyEnum::Variant2), MyEnum::lookup_by_name("variant2"));
         assert_eq!(None, MyEnum::lookup_by_name("inexistent"));
     }
+
+    #[test]
+    fn test_macro_mb() {
+        // Expands to the right value, and the result type is inferred from the
+        // usage site without any casts.
+        let as_u32: u32 = MB!(4);
+        let as_u64: u64 = MB!(4);
+        let as_usize: usize = MB!(4);
+        assert_eq!(4_194_304, as_u32);
+        assert_eq!(4_194_304, as_u64);
+        assert_eq!(4_194_304, as_usize);
+
+        // Zero and the unit value.
+        assert_eq!(0u64, MB!(0));
+        assert_eq!(1_048_576u64, MB!(1));
+
+        // The argument is grouped, so a compound expression is multiplied as a
+        // whole, and the macro composes inside a larger expression.
+        assert_eq!(4_194_304u64, MB!(1 + 3));
+        assert_eq!(1_048_577u32, 1 + MB!(1));
+    }
 }

stackslib/src/chainstate/nakamoto/mod.rs

@@ -23,13 +23,13 @@ use clarity::vm::events::{STXEventType, STXMintEventData, StacksTransactionEvent
 use clarity::vm::types::PrincipalData;
 use clarity::vm::{ClarityVersion, Value};
 use lazy_static::lazy_static;
+use libstackerdb::STACKERDB_MAX_CHUNK_SIZE;
 use rusqlite::types::{FromSql, FromSqlError, FromSqlResult, ToSql, ToSqlOutput};
 use rusqlite::{params, Connection, OptionalExtension};
 use sha2::{Digest as Sha2Digest, Sha512_256};
 use stacks_common::bitvec::BitVec;
 use stacks_common::codec::{
     read_next, write_next, Error as CodecError, StacksMessageCodec, MAX_MESSAGE_LEN,
-    MAX_PAYLOAD_LEN,
 };
 use stacks_common::consts::{FIRST_BURNCHAIN_CONSENSUS_HASH, FIRST_STACKS_BLOCK_HASH};
 use stacks_common::types::chainstate::{
@@ -5244,7 +5244,7 @@ impl NakamotoChainState {
 
         Ok((
             StackerDBConfig {
-                chunk_size: MAX_PAYLOAD_LEN.into(),
+                chunk_size: STACKERDB_MAX_CHUNK_SIZE.into(),
                 signers,
                 write_freq: 0,
                 max_writes: u32::MAX,  // no limit on number of writes

stackslib/src/chainstate/nakamoto/tests/mod.rs

@@ -21,7 +21,7 @@ use clarity::util::secp256k1::Secp256k1PrivateKey;
 use clarity::vm::costs::ExecutionCost;
 use clarity::vm::types::StacksAddressExtensions;
 use clarity::vm::{ClarityName, ContractName, Value};
-use libstackerdb::StackerDBChunkData;
+use libstackerdb::{StackerDBChunkData, STACKERDB_MAX_CHUNK_SIZE};
 use rand::distributions::Standard;
 use rand::{thread_rng, Rng, RngCore};
 use rusqlite::params;
@@ -2236,6 +2236,41 @@ fn test_make_miners_stackerdb_config() {
         let stackerdb_config = NakamotoChainState::make_miners_stackerdb_config(sort_db, &tip)
             .unwrap()
             .0;
+        assert_eq!(
+            stackerdb_config.chunk_size,
+            u64::from(STACKERDB_MAX_CHUNK_SIZE),
+            "chunk_size config matches the stackerdb wire cap"
+        );
+        assert_eq!(
+            stackerdb_config.write_freq, 0,
+            "write_freq config has no minimum write interval"
+        );
+        assert_eq!(
+            stackerdb_config.max_writes,
+            u32::MAX,
+            "max_writes config has no write-count limit"
+        );
+        assert_eq!(
+            stackerdb_config.max_neighbors, 200,
+            "max_neighbors config should be 200"
+        );
+        assert!(
+            stackerdb_config.hint_replicas.is_empty(),
+            "hint_replicas config must not provide hint"
+        );
+        assert_eq!(
+            2,
+            stackerdb_config.signers.len(),
+            "signers config must always have exactly two signers (latest and previous sortition winner)"
+        );
+        for (idx, (_addr, slot_count)) in stackerdb_config.signers.iter().enumerate() {
+            // note: addresses checked later in the test
+            assert_eq!(
+                *slot_count, MINER_SLOT_COUNT,
+                "signer at index {idx} must have exactly MINER_SLOT_COUNT slots"
+            );
+        }
+
         eprintln!(
             "stackerdb_config at i = {} (sorition? {}): {:?}",
             &i, sortition, &stackerdb_config

stackslib/src/config/mod.rs

@@ -3742,6 +3742,27 @@ pub struct ConnectionOptionsFile {
     /// @default: [`DEFAULT_PROPOSAL_MEMORY_BYTES`]
     /// @units: bytes
     pub block_proposal_max_tx_mem_bytes: Option<u64>,
+
+    /// Maximum bytes/sec a single peer may push as transactions before being NACKed
+    /// with Throttled. Zero disables the cap.
+    /// ---
+    /// @default: `0` (disabled)
+    /// @units: bytes/second
+    pub max_transaction_push_bandwidth: Option<u64>,
+
+    /// Maximum bytes/sec a single peer may push as StackerDB chunks before being
+    /// NACKed with Throttled. Zero disables the cap.
+    /// ---
+    /// @default: `4_194_304` (4 MB/sec)
+    /// @units: bytes/second
+    pub max_stackerdb_push_bandwidth: Option<u64>,
+
+    /// Maximum bytes/sec a single peer may push as Nakamoto blocks before being
+    /// NACKed with Throttled. Zero disables the cap.
+    /// ---
+    /// @default: `0` (disabled)
+    /// @units: bytes/second
+    pub max_nakamoto_block_push_bandwidth: Option<u64>,
 }
 
 impl ConnectionOptionsFile {
@@ -3905,6 +3926,15 @@ impl ConnectionOptionsFile {
             block_proposal_max_tx_mem_bytes: self
                 .block_proposal_max_tx_mem_bytes
                 .unwrap_or(default.block_proposal_max_tx_mem_bytes),
+            max_transaction_push_bandwidth: self
+                .max_transaction_push_bandwidth
+                .unwrap_or(default.max_transaction_push_bandwidth),
+            max_stackerdb_push_bandwidth: self
+                .max_stackerdb_push_bandwidth
+                .unwrap_or(default.max_stackerdb_push_bandwidth),
+            max_nakamoto_block_push_bandwidth: self
+                .max_nakamoto_block_push_bandwidth
+                .unwrap_or(default.max_nakamoto_block_push_bandwidth),
             ..default
         })
     }
@@ -5102,4 +5132,35 @@ mod tests {
             "internal default migrate setting"
         );
     }
+
+    #[test]
+    fn test_load_push_bandwidth_fields_config() {
+        // check defaults for omitted fields
+        let config = utils::config_from_valid_string("");
+        assert_eq!(0, config.connection_options.max_transaction_push_bandwidth,);
+        assert_eq!(
+            MB!(4),
+            config.connection_options.max_stackerdb_push_bandwidth,
+        );
+        assert_eq!(
+            0,
+            config.connection_options.max_nakamoto_block_push_bandwidth,
+        );
+
+        // Check values for configured fields
+        let config = utils::config_from_valid_string(
+            r#"
+            [connection_options]
+            max_transaction_push_bandwidth = 10
+            max_stackerdb_push_bandwidth = 20
+            max_nakamoto_block_push_bandwidth = 30
+            "#,
+        );
+        assert_eq!(10, config.connection_options.max_transaction_push_bandwidth,);
+        assert_eq!(20, config.connection_options.max_stackerdb_push_bandwidth,);
+        assert_eq!(
+            30,
+            config.connection_options.max_nakamoto_block_push_bandwidth,
+        );
+    }
 }

stackslib/src/net/connection.rs

@@ -389,10 +389,15 @@ pub struct ConnectionOptions {
     pub max_attachment_retry_count: u64,
     pub read_only_call_limit: ExecutionCost,
     pub maximum_call_argument_size: u32,
+    /// maximum bytes/sec a single peer may push as Stacks 2.x Blocks before being NACKed
     pub max_block_push_bandwidth: u64,
+    /// maximum bytes/sec a single peer may push as Stacks 2.x Microblocks before being NACKed
     pub max_microblocks_push_bandwidth: u64,
+    /// maximum bytes/sec a single peer may push as Transaction messages before being NACKed
     pub max_transaction_push_bandwidth: u64,
+    /// maximum bytes/sec a single peer may push as StackerDB chunks before being NACKed
     pub max_stackerdb_push_bandwidth: u64,
+    /// maximum bytes/sec a single peer may push as Nakamoto Block messages before being NACKed
     pub max_nakamoto_block_push_bandwidth: u64,
     pub max_sockets: usize,
     pub public_ip_address: Option<(PeerAddress, u16)>,
@@ -557,7 +562,7 @@ impl std::default::Default for ConnectionOptions {
             max_block_push_bandwidth: 0, // infinite upload bandwidth allowed
             max_microblocks_push_bandwidth: 0, // infinite upload bandwidth allowed
             max_transaction_push_bandwidth: 0, // infinite upload bandwidth allowed
-            max_stackerdb_push_bandwidth: 0, // infinite upload bandwidth allowed
+            max_stackerdb_push_bandwidth: MB!(4), // 4 MB/sec upload bandwidth allowed
             max_nakamoto_block_push_bandwidth: 0, // infinite upload bandwidth allowed
             max_sockets: 800,            // maximum number of client sockets we'll ever register
             public_ip_address: None,     // resolve it at runtime by default

stackslib/src/net/stackerdb/db.rs

@@ -17,7 +17,7 @@ use std::path::Path;
 use std::{fs, io};
 
 use clarity::vm::types::QualifiedContractIdentifier;
-use libstackerdb::{SlotMetadata, STACKERDB_MAX_CHUNK_SIZE};
+use libstackerdb::SlotMetadata;
 use rusqlite::{params, OpenFlags, OptionalExtension, Row};
 use stacks_common::types::chainstate::StacksAddress;
 use stacks_common::types::sqlite::NO_PARAMS;
@@ -403,7 +403,8 @@ impl StackerDBTx<'_> {
         slot_desc: &SlotMetadata,
         chunk: &[u8],
     ) -> Result<(), net_error> {
-        if chunk.len() > STACKERDB_MAX_CHUNK_SIZE as usize {
+        // Check per-replica chunk-size cap.
+        if (chunk.len() as u64) > self.config.chunk_size {
             return Err(net_error::StackerDBChunkTooBig(chunk.len()));
         }
 

stackslib/src/net/stackerdb/mod.rs

@@ -187,7 +187,14 @@ impl StackerDBConfig {
     /// Config that does nothing
     pub fn noop() -> StackerDBConfig {
         StackerDBConfig {
-            chunk_size: u64::MAX,
+            // Cap the chunk size at the protocol-wide maximum rather than u64::MAX.
+            // `noop()` is used as a fallback whenever a replica's real config can't be
+            // loaded (see `create_or_reconfigure_stackerdbs`), and it can transiently
+            // overwrite a good in-memory config on a failed refresh. Since the DB slots
+            // persist independently of the config, writes can still land on existing
+            // slots while `noop()` is active, so its chunk-size limit must never be
+            // looser than `STACKERDB_MAX_CHUNK_SIZE`.
+            chunk_size: STACKERDB_MAX_CHUNK_SIZE as u64,
             write_freq: 0,
             max_writes: u32::MAX,
             hint_replicas: vec![],
@@ -565,18 +572,33 @@ impl PeerNetwork {
         })
     }
 
-    /// Validate chunk data -- either pushed to us, or downloaded.
+    /// Validate chunk data either downloaded (with [`StackerDBSync::validate_downloaded_chunk`]), or
+    /// pushed to us (with [`PeerNetwork::handle_unsolicited_StackerDBPushChunk`])
+    ///
     /// NOTE: does not check write frequency, since the caller has different ways of doing this.
-    /// Returns Ok(true) if the chunk is valid
-    /// Returns Ok(false) if the chunk is invalid
-    /// Returns Err(..) on DB error
+    /// Returns:  
+    /// - Ok(true) if the chunk is valid
+    /// - Ok(false) if the chunk is invalid
+    /// - Err(..) on DB error
     pub fn validate_received_chunk(
         &self,
         smart_contract_id: &QualifiedContractIdentifier,
         config: &StackerDBConfig,
         data: &StackerDBChunkData,
         expected_versions: &[u32],
     ) -> Result<bool, net_error> {
+        // validate -- must not exceed this replica's configured chunk size.
+        if (data.data.len() as u64) > config.chunk_size {
+            info!(
+                "Received StackerDBChunk for {} ID {}, which is oversized: {} bytes (max {} bytes)",
+                smart_contract_id,
+                data.slot_id,
+                data.data.len(),
+                config.chunk_size
+            );
+            return Ok(false);
+        }
+
         // validate -- must be a valid chunk
         let Some(expected_version) = expected_versions.get(data.slot_id as usize) else {
             info!(

stackslib/src/net/stackerdb/tests/db.rs

@@ -19,6 +19,7 @@ use std::path::Path;
 
 use clarity::vm::types::QualifiedContractIdentifier;
 use clarity::vm::ContractName;
+use libstackerdb::STACKERDB_MAX_CHUNK_SIZE;
 use stacks_common::address::{AddressHashMode, C32_ADDRESS_VERSION_MAINNET_SINGLESIG};
 use stacks_common::types::chainstate::{StacksAddress, StacksPrivateKey, StacksPublicKey};
 use stacks_common::util::hash::{Hash160, Sha512Trunc256Sum};
@@ -737,4 +738,76 @@ fn test_reconfigure_stackerdb() {
     }
 }
 
-// TODO: max chunk size
+/// [`StackerDBTx::try_replace_chunk`] must reject any chunk whose length exceeds the
+/// per-replica `chunk_size` configuration.
+#[test]
+fn test_try_replace_chunk_enforces_config_chunk_size() {
+    let path = "/tmp/stackerdb-tests/test_try_replace_chunk_chunk_size.sqlite";
+    setup_test_path(path);
+
+    let sc = QualifiedContractIdentifier::new(
+        StacksAddress::new(0x01, Hash160([0x01; 20]))
+            .unwrap()
+            .into(),
+        ContractName::try_from("db1").unwrap(),
+    );
+
+    let mut db = StackerDBs::connect(path, true).unwrap();
+
+    let mut db_config = StackerDBConfig::noop();
+    db_config.chunk_size = 256;
+    let tx = db.tx_begin(db_config.clone()).unwrap();
+
+    let pk = StacksPrivateKey::random();
+    let addr = StacksAddress::from_public_keys(
+        C32_ADDRESS_VERSION_MAINNET_SINGLESIG,
+        &AddressHashMode::SerializeP2PKH,
+        1,
+        &vec![StacksPublicKey::from_private(&pk)],
+    )
+    .unwrap();
+    tx.create_stackerdb(&sc, &[(addr, 1)]).unwrap();
+
+    // Case 1: within chunk_size, must pass the size gate
+    // (and succeed end-to-end given a valid signer/version).
+    let mut at_max = StackerDBChunkData {
+        slot_id: 0,
+        slot_version: 1,
+        sig: MessageSignature::empty(),
+        data: vec![0xcd; db_config.chunk_size as usize],
+    };
+    at_max.sign(&pk).unwrap();
+    tx.try_replace_chunk(&sc, &at_max.get_slot_metadata(), &at_max.data)
+        .expect("a chunk at exactly config.chunk_size must be accepted");
+
+    // Case 2: oversized by one byte over the cap must be rejected
+    // regardless of whether the signature would otherwise validate.
+    let mut oversized = StackerDBChunkData {
+        slot_id: 0,
+        slot_version: 1,
+        sig: MessageSignature::empty(),
+        data: vec![0xab; (db_config.chunk_size as usize) + 1],
+    };
+    oversized.sign(&pk).unwrap();
+
+    let err = tx
+        .try_replace_chunk(&sc, &oversized.get_slot_metadata(), &oversized.data)
+        .unwrap_err();
+    assert_eq!(
+        net_error::StackerDBChunkTooBig(oversized.data.len()),
+        err,
+        "should be chunk too big, but got {err:?}"
+    );
+}
+
+#[test]
+fn test_stackerdb_noop_config() {
+    let config = StackerDBConfig::noop();
+    assert_eq!(STACKERDB_MAX_CHUNK_SIZE as u64, config.chunk_size);
+    assert_eq!(0, config.write_freq);
+    assert_eq!(u32::MAX, config.max_writes);
+    assert!(config.hint_replicas.is_empty());
+    assert_eq!(8, config.max_neighbors);
+    assert!(config.signers.is_empty());
+    assert_eq!(0, config.num_slots());
+}