feat(hotblocks): make the spill bound configurable

crates/hotblocks/Cargo.toml

@@ -9,7 +9,7 @@ async-stream = "0.3.6"
 axum = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true, features = ["std"] }
-clap =  { workspace = true, features = ["derive"] }
+clap =  { workspace = true, features = ["derive", "env"] }
 flate2 = { workspace = true }
 zstd = "0.13"
 futures = { workspace = true }

crates/hotblocks/spec/14-parameters.md

@@ -27,7 +27,7 @@ ran against.
 | Parameter | Role | Observed | Target |
 |---|---|---|---|
 | `P-BATCH-ROWS` | batch flush bound, rows (WP-3, HZ-6) | 200 000 rows | keep |
-| `P-BATCH-BYTES` | batch flush bound, bytes (WP-3, PF-1) | ~30 MB (soft) | hard ceiling ⚠ (GAP-13) |
+| `P-BATCH-BYTES` | batch flush bound, bytes (WP-3, PF-1) | `--spill-bound-bytes`, 30 MB by default (soft) | hard ceiling ⚠ (GAP-13) |
 | `P-MAX-BLOCK-BYTES` | ceiling on one block's encoded size, enforced at ingest (WP-2 rejection + source fault); what makes the read-side "+ one block" allowance (RP-17, INV-25) and PF-1's ceiling finite | absent — a single oversized block stores (the batch bound is soft) and must later be emitted whole (GAP-37) | define ⚠ |
 | `P-FORK-CONSENSUS` | arbitration timeout before accepting a fork signal (WP-4) | 2 s | keep |
 | `P-SOURCE-BACKOFF` | per-source retry backoff schedule (WP-17, FM-SRC-1) | 0→10 s exponential steps | keep |

crates/hotblocks/src/cli.rs

@@ -113,6 +113,17 @@ pub struct CLI {
     #[arg(long)]
     pub transaction_hash_index: bool,
 
+    /// Chunks that never grew past this many buffered bytes are prepared in
+    /// memory at flush; larger ones spill to temp files while accumulating.
+    /// 0 forces the temp-file path for every chunk.
+    #[arg(
+        long,
+        value_name = "BYTES",
+        env = "SQD_SPILL_BOUND_BYTES",
+        default_value_t = crate::dataset_controller::DEFAULT_SPILL_BOUND_BYTES
+    )]
+    pub spill_bound_bytes: usize,
+
     /// Known client IDs for metrics labeling. Client IDs not in this list
     /// will be reported as "unknown" to prevent metrics cardinality abuse.
     #[arg(long = "known-client", value_name = "ID")]
@@ -169,7 +180,7 @@ impl CLI {
             .filter_map(|(id, cfg)| matches!(cfg.retention_strategy, RetentionConfig::Api { .. }).then_some(*id))
             .collect();
 
-        let data_service = DataService::start(db.clone(), datasets, self.startup_disk_reclaim)
+        let data_service = DataService::start(db.clone(), datasets, self.startup_disk_reclaim, self.spill_bound_bytes)
             .await
             .map(Arc::new)?;
 

crates/hotblocks/src/data_service.rs

@@ -27,7 +27,8 @@ impl DataService {
     pub async fn start(
         db: DBRef,
         datasets: BTreeMap<DatasetId, DatasetConfig>,
-        disk_reclaim: bool
+        disk_reclaim: bool,
+        spill_bound_bytes: usize
     ) -> anyhow::Result<Self> {
         let unconfigured: Vec<DatasetId> = db
             .get_all_datasets()?
@@ -75,7 +76,16 @@ impl DataService {
                 };
 
                 tokio::task::spawn_blocking(move || {
-                    DatasetController::new(db, dataset_id, cfg.kind, retention, max_blocks, data_sources).map(|c| {
+                    DatasetController::new(
+                        db,
+                        dataset_id,
+                        cfg.kind,
+                        retention,
+                        max_blocks,
+                        data_sources,
+                        spill_bound_bytes
+                    )
+                    .map(|c| {
                         c.enable_compaction(!cfg.disable_compaction);
                         Arc::new(c)
                     })

crates/hotblocks/src/dataset_controller/dataset_controller.rs

@@ -40,7 +40,8 @@ impl DatasetController {
         dataset_kind: DatasetKind,
         retention: RetentionStrategy,
         max_blocks: Option<u64>,
-        data_sources: Vec<ReqwestDataClient>
+        data_sources: Vec<ReqwestDataClient>,
+        spill_bound_bytes: usize
     ) -> anyhow::Result<Self> {
         let (head_sender, head_receiver) = tokio::sync::watch::channel(None);
         let (finalized_head_sender, finalized_head_receiver) = tokio::sync::watch::channel(None);
@@ -70,7 +71,8 @@ impl DatasetController {
             max_blocks,
             retention_recv,
             head_sender,
-            finalized_head_sender
+            finalized_head_sender,
+            spill_bound_bytes
         };
 
         let task = tokio::spawn(ctl.run(write).in_current_span());
@@ -210,7 +212,8 @@ struct Ctl {
     max_blocks: Option<u64>,
     retention_recv: tokio::sync::watch::Receiver<RetentionStrategy>,
     head_sender: tokio::sync::watch::Sender<Option<BlockRef>>,
-    finalized_head_sender: tokio::sync::watch::Sender<Option<BlockRef>>
+    finalized_head_sender: tokio::sync::watch::Sender<Option<BlockRef>>,
+    spill_bound_bytes: usize
 }
 
 macro_rules! warn_on_tx_restart {
@@ -437,7 +440,8 @@ impl Ctl {
                 self.data_sources.clone(),
                 self.dataset_kind,
                 write.next_block(),
-                write.head_hash()
+                write.head_hash(),
+                self.spill_bound_bytes
             )
             .instrument(ingest_span)
         );

crates/hotblocks/src/dataset_controller/ingest.rs

@@ -17,13 +17,14 @@ pub fn ingest<'a, 'b>(
     sources: Vec<ReqwestDataClient>,
     dataset_kind: DatasetKind,
     first_block: BlockNumber,
-    parent_block_hash: Option<&'a str>
+    parent_block_hash: Option<&'a str>,
+    spill_bound_bytes: usize
 ) -> BoxFuture<'b, anyhow::Result<()>> {
     macro_rules! run {
         ($builder:expr) => {{
             let mut data_source = StandardDataSource::new(sources, from_json_bytes);
             data_source.set_position(first_block, parent_block_hash);
-            IngestGeneric::new(dataset_id, data_source, $builder, message_sender)
+            IngestGeneric::new(dataset_id, data_source, $builder, message_sender, spill_bound_bytes)
                 .run()
                 .boxed()
         }};

crates/hotblocks/src/dataset_controller/ingest_generic.rs

@@ -50,19 +50,22 @@ impl Display for NewChunk {
     }
 }
 
-/// `maybe_flush` spills builder contents to the processor beyond this size.
-const SPILL_BOUND_BYTES: usize = 30 * 1024 * 1024;
+/// Default for `--spill-bound-bytes`: `maybe_flush` spills builder contents to the
+/// processor beyond this size.
+pub(crate) const DEFAULT_SPILL_BOUND_BYTES: usize = 30 * 1024 * 1024;
 
 struct DataBuilder<CB> {
     builder: CB,
-    processor: Option<ChunkProcessor>
+    processor: Option<ChunkProcessor>,
+    spill_bound_bytes: usize
 }
 
 impl<CB: BlockChunkBuilder> DataBuilder<CB> {
-    pub fn new(builder: CB) -> Self {
+    pub fn new(builder: CB, spill_bound_bytes: usize) -> Self {
         Self {
             builder,
-            processor: None
+            processor: None,
+            spill_bound_bytes
         }
     }
 
@@ -89,7 +92,7 @@ impl<CB: BlockChunkBuilder> DataBuilder<CB> {
     }
 
     pub fn finish(&mut self) -> anyhow::Result<PreparedChunk> {
-        if self.processor.is_none() && self.builder.byte_size() <= SPILL_BOUND_BYTES {
+        if self.processor.is_none() && self.builder.byte_size() <= self.spill_bound_bytes {
             // no spill and within the spill bound — skip the temp files. The bound is
             // re-checked here: a row-count-triggered flush can carry an oversized final
             // block that maybe_flush's byte check never saw.
@@ -130,14 +133,15 @@ where
         dataset_id: DatasetId,
         data_source: DS,
         chunk_builder: CB,
-        message_sender: tokio::sync::mpsc::Sender<IngestMessage>
+        message_sender: tokio::sync::mpsc::Sender<IngestMessage>,
+        spill_bound_bytes: usize
     ) -> Self {
         let first_block = data_source.get_next_block();
         Self {
             dataset_id,
             message_sender,
             data_source,
-            builder: Some(DataBuilder::new(chunk_builder)),
+            builder: Some(DataBuilder::new(chunk_builder, spill_bound_bytes)),
             finalized_head: None,
             buffered_blocks: 0,
             parent_block_hash: String::new(),
@@ -234,7 +238,7 @@ where
         if self.builder_ref().num_rows() > 200_000 {
             return self.flush().await;
         }
-        if self.builder_ref().in_memory_buffered_bytes() > SPILL_BOUND_BYTES {
+        if self.builder_ref().in_memory_buffered_bytes() > self.builder_ref().spill_bound_bytes {
             return self.with_blocking_builder(|b| b.flush_to_processor()).await;
         }
         Ok(())
@@ -363,10 +367,12 @@ mod tests {
         .unwrap()
     }
 
-    fn unspilled(over_bytes: usize) -> DataBuilder<HyperliquidFillsChunkBuilder> {
-        let mut b = DataBuilder::new(HyperliquidFillsChunkBuilder::new());
+    const TEST_BOUND: usize = 256 * 1024;
+
+    fn unspilled(bound: usize, fill_over_bytes: usize) -> DataBuilder<HyperliquidFillsChunkBuilder> {
+        let mut b = DataBuilder::new(HyperliquidFillsChunkBuilder::new(), bound);
         let block = fills_block();
-        while b.in_memory_buffered_bytes() <= over_bytes {
+        while b.in_memory_buffered_bytes() <= fill_over_bytes {
             b.push_block(&block).unwrap();
         }
         b
@@ -378,15 +384,15 @@ mod tests {
     /// `into_processor`, which in-memory tables refuse.
     #[test]
     fn oversized_unspilled_chunk_spills_at_finish() {
-        let mut b = unspilled(SPILL_BOUND_BYTES);
+        let mut b = unspilled(TEST_BOUND, TEST_BOUND);
         let mut chunk = b.finish().unwrap();
         let (_, table) = chunk.pop_first().unwrap();
         assert!(table.into_processor().is_ok(), "expected the spill path");
     }
 
     #[test]
     fn bounded_unspilled_chunk_is_prepared_in_memory() {
-        let mut b = unspilled(1);
+        let mut b = unspilled(TEST_BOUND, 1);
         let mut chunk = b.finish().unwrap();
         let (_, table) = chunk.pop_first().unwrap();
         assert!(table.into_processor().is_err(), "expected the in-memory path");

crates/hotblocks/src/dataset_controller/mod.rs

@@ -4,3 +4,4 @@ mod ingest_generic;
 mod write_controller;
 
 pub use dataset_controller::DatasetController;
+pub(crate) use ingest_generic::DEFAULT_SPILL_BOUND_BYTES;