From 7ee5ffadab7daf6b316d115b16561d6539439e56 Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Wed, 22 Jul 2026 11:25:06 +0300 Subject: [PATCH] feat(hotblocks): make the spill bound configurable The bound decides which chunks are prepared in memory and which build an out-of-core processor, so it is the one dial that trades ingest memory against the temp-file path #98 was written to avoid. Raising it past a dataset's flush size is what keeps that dataset off temp files; 0 puts every chunk back on them, which is the kill switch if the in-memory path ever misbehaves in production. Neither is reachable while the bound is a constant, and both are wanted per stack -- block sizes differ by an order of magnitude between them. The default stays 30 MB, so an unset flag renders exactly as before. Sizing it up is not free: the buffer is per dataset and mainnet runs 45 of them, so the worst case scales with the bound. Restores the knob from the first draft of #98, dropped there because nothing needed it yet. --- crates/hotblocks/Cargo.toml | 2 +- crates/hotblocks/spec/14-parameters.md | 2 +- crates/hotblocks/src/cli.rs | 13 ++++++- crates/hotblocks/src/data_service.rs | 14 ++++++-- .../dataset_controller/dataset_controller.rs | 12 ++++--- .../src/dataset_controller/ingest.rs | 5 +-- .../src/dataset_controller/ingest_generic.rs | 34 +++++++++++-------- .../hotblocks/src/dataset_controller/mod.rs | 1 + 8 files changed, 58 insertions(+), 25 deletions(-) diff --git a/crates/hotblocks/Cargo.toml b/crates/hotblocks/Cargo.toml index cd2d395f..03ad3da4 100644 --- a/crates/hotblocks/Cargo.toml +++ b/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 } diff --git a/crates/hotblocks/spec/14-parameters.md b/crates/hotblocks/spec/14-parameters.md index 48f40094..9cd102d4 100644 --- a/crates/hotblocks/spec/14-parameters.md +++ b/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 | diff --git a/crates/hotblocks/src/cli.rs b/crates/hotblocks/src/cli.rs index ede190f0..d00f69af 100644 --- a/crates/hotblocks/src/cli.rs +++ b/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)?; diff --git a/crates/hotblocks/src/data_service.rs b/crates/hotblocks/src/data_service.rs index 5abe8c19..e83de07e 100644 --- a/crates/hotblocks/src/data_service.rs +++ b/crates/hotblocks/src/data_service.rs @@ -27,7 +27,8 @@ impl DataService { pub async fn start( db: DBRef, datasets: BTreeMap, - disk_reclaim: bool + disk_reclaim: bool, + spill_bound_bytes: usize ) -> anyhow::Result { let unconfigured: Vec = 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) }) diff --git a/crates/hotblocks/src/dataset_controller/dataset_controller.rs b/crates/hotblocks/src/dataset_controller/dataset_controller.rs index 26b73e07..e5d14774 100644 --- a/crates/hotblocks/src/dataset_controller/dataset_controller.rs +++ b/crates/hotblocks/src/dataset_controller/dataset_controller.rs @@ -40,7 +40,8 @@ impl DatasetController { dataset_kind: DatasetKind, retention: RetentionStrategy, max_blocks: Option, - data_sources: Vec + data_sources: Vec, + spill_bound_bytes: usize ) -> anyhow::Result { 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, retention_recv: tokio::sync::watch::Receiver, head_sender: tokio::sync::watch::Sender>, - finalized_head_sender: tokio::sync::watch::Sender> + finalized_head_sender: tokio::sync::watch::Sender>, + 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) ); diff --git a/crates/hotblocks/src/dataset_controller/ingest.rs b/crates/hotblocks/src/dataset_controller/ingest.rs index b52f93ea..e16228dc 100644 --- a/crates/hotblocks/src/dataset_controller/ingest.rs +++ b/crates/hotblocks/src/dataset_controller/ingest.rs @@ -17,13 +17,14 @@ pub fn ingest<'a, 'b>( sources: Vec, 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() }}; diff --git a/crates/hotblocks/src/dataset_controller/ingest_generic.rs b/crates/hotblocks/src/dataset_controller/ingest_generic.rs index c4046986..c98b5e59 100644 --- a/crates/hotblocks/src/dataset_controller/ingest_generic.rs +++ b/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 { builder: CB, - processor: Option + processor: Option, + spill_bound_bytes: usize } impl DataBuilder { - 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 DataBuilder { } pub fn finish(&mut self) -> anyhow::Result { - 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 + message_sender: tokio::sync::mpsc::Sender, + 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 { - let mut b = DataBuilder::new(HyperliquidFillsChunkBuilder::new()); + const TEST_BOUND: usize = 256 * 1024; + + fn unspilled(bound: usize, fill_over_bytes: usize) -> DataBuilder { + 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,7 +384,7 @@ 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"); @@ -386,7 +392,7 @@ mod tests { #[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"); diff --git a/crates/hotblocks/src/dataset_controller/mod.rs b/crates/hotblocks/src/dataset_controller/mod.rs index 9fe3502a..129f6fe5 100644 --- a/crates/hotblocks/src/dataset_controller/mod.rs +++ b/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;