From 761f36298216566917e6beb6e658e1009925a64c Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Wed, 15 Jul 2026 13:10:26 +0300 Subject: [PATCH] refactor(hotblocks): consolidate head/finalized-head into one owner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Head and finalized-head were held twice in memory — as WriteController fields and as the DatasetController watch channels — kept in sync by hand through WriteCtx.notify_*, so a write that forgot a notify would silently stall wait_for_block. Give the writer the watch senders and route every mutation through set_head/set_finalized_head, which update the field and publish together, after the commit: a write can no longer forget to publish, and a published watermark is always already durable (INV-31/CN-4). The senders live on the controller and are cloned into each rebuilt writer, so seeding moves into WriteController::new and restart re-seed becomes automatic (INV-40/CN-9). The plain head field stays as the writer's committed-state mirror (WP-1) — reading it back from the channel would pin a watch read-lock across the RocksDB transaction in finalize(). WriteCtx dissolves into WriteController, removing the write.write. indirection. New unit tests cover the two properties the conformance matrix flags as untested: a committed transition publishes exactly the durable watermark (INV-30/31) and a rebuilt writer re-seeds from storage (INV-40/CN-9), plus the head-only-progress dedup that keeps finalized waiters from waking. Co-Authored-By: Claude Opus 4.8 --- .../dataset_controller/dataset_controller.rs | 261 ++-------- .../dataset_controller/write_controller.rs | 453 +++++++++++++++--- 2 files changed, 436 insertions(+), 278 deletions(-) diff --git a/crates/hotblocks/src/dataset_controller/dataset_controller.rs b/crates/hotblocks/src/dataset_controller/dataset_controller.rs index 9925d8e..1ec67ed 100644 --- a/crates/hotblocks/src/dataset_controller/dataset_controller.rs +++ b/crates/hotblocks/src/dataset_controller/dataset_controller.rs @@ -1,24 +1,15 @@ -use std::{ - collections::BTreeMap, - ops::Add, - time::{Duration, Instant as StdInstant} -}; +use std::{ops::Add, time::Duration}; use anyhow::{Context, anyhow}; use futures::{FutureExt, StreamExt, future::BoxFuture, stream::FuturesUnordered}; use sqd_data_client::reqwest::ReqwestDataClient; use sqd_primitives::{BlockNumber, BlockRef, TransactionRef}; -use sqd_storage::db::{Chunk, CompactionStatus, DatasetId, HashIndexWriteMetrics}; +use sqd_storage::db::{CompactionStatus, DatasetId}; use tokio::{select, task::JoinHandle, time::Instant}; use tracing::{Instrument, debug, error, info, info_span, instrument, warn}; use crate::{ - dataset_controller::{ - ingest::ingest, - ingest_generic::{IngestMessage, NewChunk}, - write_controller::WriteController - }, - metrics::{WriteStage, report_hash_index_write_metrics, report_write_duration}, + dataset_controller::{ingest::ingest, ingest_generic::IngestMessage, write_controller::WriteController}, types::{DBRef, DatasetKind, RetentionStrategy} }; @@ -51,22 +42,26 @@ impl DatasetController { max_blocks: Option, data_sources: Vec ) -> anyhow::Result { - let mut write = WriteController::new(db.clone(), dataset_id, dataset_kind)?; + let (head_sender, head_receiver) = tokio::sync::watch::channel(None); + let (finalized_head_sender, finalized_head_receiver) = tokio::sync::watch::channel(None); + + // Channels live on the controller so they outlive writer restarts; each + // rebuilt writer gets a sender clone and seeds it from storage. + let mut write = WriteController::new( + db.clone(), + dataset_id, + dataset_kind, + head_sender.clone(), + finalized_head_sender.clone() + )?; if let RetentionStrategy::FromBlock { number, parent_hash } = &retention { - observe_storage_write(dataset_id, WriteStage::Retention, |metrics| { - write.init_retention(*number, parent_hash.clone(), metrics) - })?; + write.init_retention(*number, parent_hash.clone())?; } let (retention_sender, retention_recv) = tokio::sync::watch::channel(retention); - let (head_sender, head_receiver) = tokio::sync::watch::channel(None); - let (finalized_head_sender, finalized_head_receiver) = tokio::sync::watch::channel(None); let (compaction_enabled_sender, compaction_enabled_receiver) = tokio::sync::watch::channel(false); - let _ = head_sender.send(write.head().cloned()); - let _ = finalized_head_sender.send(write.finalized_head().cloned()); - let ctl = Ctl { db: db.clone(), dataset_id, @@ -174,131 +169,6 @@ impl DatasetController { } } -struct WriteCtx { - db: DBRef, - dataset_id: DatasetId, - write: WriteController, - head_sender: tokio::sync::watch::Sender>, - finalized_head_sender: tokio::sync::watch::Sender> -} - -impl WriteCtx { - fn handle_ingest_msg(&mut self, msg: IngestMessage, head: Option) -> anyhow::Result<()> { - match msg { - IngestMessage::FinalizedHead(head) => { - self.write.finalize(&head)?; - self.notify_finalized_head(); - } - IngestMessage::NewChunk(new_chunk) => { - let ctx = format!("failed to write new chunk {}", new_chunk); - self.write_new_chunk(new_chunk).context(ctx)?; - self.notify_head(); - self.notify_finalized_head(); - if let Some(n) = head { - let first_chunk_head = self.write.first_chunk_head().map(|h| h.number); - if let Some(floor) = trim_floor(first_chunk_head, self.write.next_block(), n) { - self.retain(floor, None)?; - } - } - } - IngestMessage::Fork { - prev_blocks, - rollback_sender - } => { - self.write.compute_rollback(&prev_blocks).map(|rollback| { - let _ = rollback_sender.send(rollback); - })?; - } - } - Ok(()) - } - - fn write_new_chunk(&mut self, mut new_chunk: NewChunk) -> anyhow::Result<()> { - let desc = self.write.dataset_kind().dataset_description(); - let started = StdInstant::now(); - let tables: anyhow::Result<_> = (|| { - let mut tables = BTreeMap::new(); - - for (name, prepared) in new_chunk.tables.iter_mut() { - let mut builder = self.db.new_table_builder(prepared.schema()); - - if let Some(table_desc) = desc.tables.get(name) { - for (&col, opts) in table_desc.options.column_options.iter() { - if opts.stats_enable { - builder.add_stat_by_name(col)?; - } - } - } - - prepared.read(&mut builder, 0, prepared.num_rows())?; - - tables.insert(name.to_string(), builder.finish()?); - } - - Ok(tables) - })(); - report_write_duration(self.dataset_id, WriteStage::Tables, started.elapsed(), tables.is_ok()); - let tables = tables?; - - let chunk = Chunk::V1 { - parent_block_hash: new_chunk.parent_block_hash, - first_block: new_chunk.first_block, - last_block: new_chunk.last_block, - last_block_hash: new_chunk.last_block_hash, - first_block_time: new_chunk.first_block_time, - last_block_time: new_chunk.last_block_time, - tables - }; - - observe_storage_write(self.dataset_id, WriteStage::Commit, |metrics| { - self.write.new_chunk(new_chunk.finalized_head.as_ref(), &chunk, metrics) - }) - } - - fn retain(&mut self, from_block: BlockNumber, parent_hash: Option) -> anyhow::Result<()> { - observe_storage_write(self.dataset_id, WriteStage::Retention, |metrics| { - self.write.retain(from_block, parent_hash, metrics) - })?; - self.notify_finalized_head(); - self.notify_head(); - Ok(()) - } - - fn notify_head(&self) { - send_if_new(&self.head_sender, self.write.head().cloned()); - } - - fn notify_finalized_head(&self) { - send_if_new(&self.finalized_head_sender, self.write.finalized_head().cloned()) - } - - fn starts_at(&self, block_number: BlockNumber, parent_hash: &Option) -> bool { - self.write.start_block() == block_number - && self.write.start_block_parent_hash() == parent_hash.as_ref().map(String::as_str) - } -} - -fn send_if_new(sender: &tokio::sync::watch::Sender, value: T) { - sender.send_if_modified(|current| { - if current == &value { - false - } else { - *current = value; - true - } - }); -} - -// Returns the new floor when the tail has to be trimmed to keep -// `max_blocks` behind the tip, or `None` when the window still fits. -// -// `max_blocks` is a soft limit. Since `retain()` only drops whole chunks, trimming -// may keep a part of the first chunk. -fn trim_floor(first_chunk_head: Option, next_block: BlockNumber, max_blocks: u64) -> Option { - let first_chunk_head = first_chunk_head?; - (next_block - first_chunk_head > max_blocks).then(|| next_block - max_blocks) -} - enum State { Idle, Init { @@ -386,7 +256,7 @@ impl Ctl { } async fn write_epoch(&mut self, maybe_write: Option) -> anyhow::Result<()> { - let mut write = self.new_write_ctx(maybe_write).await?; + let mut write = self.new_write(maybe_write).await?; macro_rules! blocking { ($body:expr) => { @@ -408,7 +278,7 @@ impl Ctl { } RetentionStrategy::Head(n) => State::Init { head: Some(n) }, RetentionStrategy::None => { - if write.write.head().is_some() { + if write.head().is_some() { State::Init { head: self.max_blocks } } else { State::Idle @@ -443,9 +313,9 @@ impl Ctl { }, top = future => { let n = *head; - let top = write.write.head().map_or(top, |h| h.number.max(top)); + let top = write.head().map_or(top, |h| h.number.max(top)); let first_block = top.saturating_sub(n); - if first_block > write.write.start_block() { + if first_block > write.start_block() { blocking! { write.retain(first_block, None) }?; @@ -515,25 +385,29 @@ impl Ctl { // Don't allow auto-moving the floor backwards because it causes a full resync. fn clamp_floor( &self, - write: &WriteCtx, + write: &WriteController, number: BlockNumber, parent_hash: Option ) -> (BlockNumber, Option) { - if self.max_blocks.is_some() && number < write.write.start_block() { - (write.write.start_block(), None) + if self.max_blocks.is_some() && number < write.start_block() { + (write.start_block(), None) } else { (number, parent_hash) } } - async fn handle_retention_change(&mut self, state: &mut State, mut write: WriteCtx) -> anyhow::Result { + async fn handle_retention_change( + &mut self, + state: &mut State, + mut write: WriteController + ) -> anyhow::Result { // need this variable to please the compiler let retention = self.retention_recv.borrow_and_update().clone(); match retention { RetentionStrategy::FromBlock { number, parent_hash } => { let (number, parent_hash) = self.clamp_floor(&write, number, parent_hash); - let will_erase_head = write.write.head().map_or(false, |h| h.number < number) || // FromBlock is greater than current head, so everything is cleared - write.write.start_block() > number; // FromBlock is less than current front, dropping everything by design + let will_erase_head = write.head().map_or(false, |h| h.number < number) || // FromBlock is greater than current head, so everything is cleared + write.start_block() > number; // FromBlock is less than current front, dropping everything by design blocking_write!(write, write.retain(number, parent_hash))?; match state { State::Ingest { .. } if !will_erase_head => {} // Keep ingesting, head is valid @@ -550,7 +424,7 @@ impl Ctl { Ok(write) } - fn spawn_ingest(&self, write: &WriteCtx) -> IngestHandle { + fn spawn_ingest(&self, write: &WriteController) -> IngestHandle { let (msg_sender, msg_recv) = tokio::sync::mpsc::channel(1); let ingest_span = info_span!("ingest"); @@ -561,8 +435,8 @@ impl Ctl { msg_sender, self.data_sources.clone(), self.dataset_kind, - write.write.next_block(), - write.write.head_hash() + write.next_block(), + write.head_hash() ) .instrument(ingest_span) ); @@ -570,47 +444,27 @@ impl Ctl { IngestHandle { msg_recv, task } } - async fn new_write_ctx(&self, maybe_write: Option) -> anyhow::Result { + async fn new_write(&self, maybe_write: Option) -> anyhow::Result { + if let Some(write) = maybe_write { + return Ok(write); + } + let db = self.db.clone(); let dataset_id = self.dataset_id; let dataset_kind = self.dataset_kind; + let head_sender = self.head_sender.clone(); + let finalized_head_sender = self.finalized_head_sender.clone(); - let write = if let Some(write) = maybe_write { - write - } else { - let span = tracing::Span::current(); - tokio::task::spawn_blocking(move || { - let _entered = span.enter(); - WriteController::new(db, dataset_id, dataset_kind) - }) - .await - .context("write init task panicked")?? - }; - - Ok(WriteCtx { - db: self.db.clone(), - dataset_id: self.dataset_id, - write, - head_sender: self.head_sender.clone(), - finalized_head_sender: self.finalized_head_sender.clone() + let span = tracing::Span::current(); + tokio::task::spawn_blocking(move || { + let _entered = span.enter(); + WriteController::new(db, dataset_id, dataset_kind, head_sender, finalized_head_sender) }) + .await + .context("write init task panicked")? } } -fn observe_storage_write( - dataset_id: DatasetId, - stage: WriteStage, - write: impl FnOnce(&mut HashIndexWriteMetrics) -> anyhow::Result -) -> anyhow::Result { - let mut hash_metrics = HashIndexWriteMetrics::default(); - let started = StdInstant::now(); - let result = write(&mut hash_metrics); - let success = result.is_ok(); - report_write_duration(dataset_id, stage, started.elapsed(), success); - report_hash_index_write_metrics(dataset_id, &hash_metrics, success); - result -} - async fn fetch_chain_top(clients: Vec) -> BlockNumber { let mut calls: FuturesUnordered<_> = (0..clients.len()).map(|i| call_client(&clients, i, false)).collect(); @@ -737,26 +591,3 @@ async fn compaction_loop(db: DBRef, dataset_id: DatasetId, mut enabled: tokio::s } } } - -#[cfg(test)] -mod tests { - use super::trim_floor; - - #[test] - fn nothing_is_trimmed_while_the_window_fits() { - assert_eq!(trim_floor(None, 500, 100), None); - // The whole dataset is one chunk [0..50], well inside the cap. - assert_eq!(trim_floor(Some(50), 51, 100), None); - // Exactly at the cap: the first chunk still has a block in the window. - assert_eq!(trim_floor(Some(0), 100, 100), None); - } - - #[test] - fn the_tail_is_trimmed_once_the_first_chunk_leaves_the_window() { - // First chunk ends at 0, so trimming starts one block past the cap. - assert_eq!(trim_floor(Some(0), 101, 100), Some(1)); - // The soft-limit overshoot: [0..150K] under a 100K cap survives until 250K. - assert_eq!(trim_floor(Some(150_000), 250_000, 100_000), None); - assert_eq!(trim_floor(Some(150_000), 250_001, 100_000), Some(150_001)); - } -} diff --git a/crates/hotblocks/src/dataset_controller/write_controller.rs b/crates/hotblocks/src/dataset_controller/write_controller.rs index 7f31563..813c2c8 100644 --- a/crates/hotblocks/src/dataset_controller/write_controller.rs +++ b/crates/hotblocks/src/dataset_controller/write_controller.rs @@ -1,9 +1,16 @@ -use anyhow::{anyhow, bail, ensure}; +use std::{collections::BTreeMap, time::Instant as StdInstant}; + +use anyhow::{Context, anyhow, bail, ensure}; use sqd_primitives::{BlockNumber, BlockRef}; use sqd_storage::db::{Chunk as StorageChunk, Chunk, DatasetId, HashIndexWriteMetrics}; +use tokio::sync::watch; use tracing::{debug, field::valuable, info, instrument, warn}; -use crate::types::{DBRef, DatasetKind}; +use crate::{ + dataset_controller::ingest_generic::{IngestMessage, NewChunk}, + metrics::{WriteStage, report_hash_index_write_metrics, report_write_duration}, + types::{DBRef, DatasetKind} +}; #[derive(Debug)] pub struct Rollback { @@ -11,6 +18,10 @@ pub struct Rollback { pub parent_block_hash: Option } +/// Single writer for a dataset. Owns head/finalized-head as its working copy of +/// committed state (WP-1) and publishes them through `set_head`/ +/// `set_finalized_head`, which update field and channel together only after the +/// commit — so a published watermark is always already durable (INV-31/CN-4). #[derive(Debug)] pub struct WriteController { db: DBRef, @@ -20,11 +31,19 @@ pub struct WriteController { parent_block_hash: Option, first_chunk_head: Option, head: Option, - finalized_head: Option + finalized_head: Option, + head_sender: watch::Sender>, + finalized_head_sender: watch::Sender> } impl WriteController { - pub fn new(db: DBRef, dataset_id: DatasetId, dataset_kind: DatasetKind) -> anyhow::Result { + pub fn new( + db: DBRef, + dataset_id: DatasetId, + dataset_kind: DatasetKind, + head_sender: watch::Sender>, + finalized_head_sender: watch::Sender> + ) -> anyhow::Result { db.create_dataset_if_not_exists(dataset_id, dataset_kind.storage_kind())?; let snapshot = db.snapshot(); @@ -32,7 +51,7 @@ impl WriteController { let first_chunk = snapshot.get_first_chunk(dataset_id)?; let last_chunk = snapshot.get_last_chunk(dataset_id)?; - Ok(Self { + let this = Self { db: db.clone(), dataset_id, dataset_kind, @@ -40,8 +59,16 @@ impl WriteController { parent_block_hash: first_chunk.as_ref().map(|c| c.last_block_hash().to_string()), first_chunk_head: first_chunk.as_ref().map(get_chunk_head), head: last_chunk.as_ref().map(get_chunk_head), - finalized_head: label.and_then(|l| l.finalized_head().cloned()) - }) + finalized_head: label.and_then(|l| l.finalized_head().cloned()), + head_sender, + finalized_head_sender + }; + + // Reseed subscribers to committed state (CN-9: recovery on writer rebuild). + this.publish_head(); + this.publish_finalized_head(); + + Ok(this) } pub fn dataset_kind(&self) -> DatasetKind { @@ -71,14 +98,30 @@ impl WriteController { self.head.as_ref() } - pub fn finalized_head(&self) -> Option<&BlockRef> { - self.finalized_head.as_ref() - } - pub fn first_chunk_head(&self) -> Option<&BlockRef> { self.first_chunk_head.as_ref() } + /// Publish only after the commit that produced `head`, never inside the txn + /// closure (INV-31: a published watermark must already be durable). + fn set_head(&mut self, head: Option) { + self.head = head; + self.publish_head(); + } + + fn set_finalized_head(&mut self, finalized_head: Option) { + self.finalized_head = finalized_head; + self.publish_finalized_head(); + } + + fn publish_head(&self) { + publish(&self.head_sender, self.head.clone()); + } + + fn publish_finalized_head(&self) { + publish(&self.finalized_head_sender, self.finalized_head.clone()); + } + pub fn compute_rollback(&self, mut prev: &[BlockRef]) -> anyhow::Result { // FIXME: self.first_block rollback limit ensure!(!prev.is_empty(), "no previous blocks where provided"); @@ -237,8 +280,8 @@ impl WriteController { head, finalized_head } => { - self.head = Some(get_chunk_head(&head)); - self.finalized_head = finalized_head; + self.set_head(Some(get_chunk_head(&head))); + self.set_finalized_head(finalized_head); self.first_chunk_head = Some(get_chunk_head(&first_chunk)); info!( "retained blocks from {} to {}", @@ -269,27 +312,23 @@ impl WriteController { } fn clear_heads(&mut self) { - self.head = None; - self.finalized_head = None; + self.set_head(None); + self.set_finalized_head(None); self.first_chunk_head = None; } - pub fn retain( - &mut self, - from_block: BlockNumber, - parent_block_hash: Option, - metrics: &mut HashIndexWriteMetrics - ) -> anyhow::Result<()> { - self._retain(from_block, parent_block_hash, true, metrics) + pub fn retain(&mut self, from_block: BlockNumber, parent_block_hash: Option) -> anyhow::Result<()> { + let dataset_id = self.dataset_id; + observe_storage_write(dataset_id, WriteStage::Retention, |metrics| { + self._retain(from_block, parent_block_hash, true, metrics) + }) } - pub fn init_retention( - &mut self, - from_block: BlockNumber, - parent_block_hash: Option, - metrics: &mut HashIndexWriteMetrics - ) -> anyhow::Result<()> { - self._retain(from_block, parent_block_hash, false, metrics) + pub fn init_retention(&mut self, from_block: BlockNumber, parent_block_hash: Option) -> anyhow::Result<()> { + let dataset_id = self.dataset_id; + observe_storage_write(dataset_id, WriteStage::Retention, |metrics| { + self._retain(from_block, parent_block_hash, false, metrics) + }) } #[instrument(skip_all, fields( @@ -342,7 +381,7 @@ impl WriteController { block_hash = new_head.hash, "saved new finalized head" ); - self.finalized_head = Some(new_head); + self.set_finalized_head(Some(new_head)); } else { debug!("finalized head was ignored") } @@ -356,44 +395,42 @@ impl WriteController { last_block_hash = %chunk.last_block_hash(), finalized_head = valuable(&finalized_head), ))] - pub fn new_chunk( - &mut self, - finalized_head: Option<&BlockRef>, - chunk: &StorageChunk, - metrics: &mut HashIndexWriteMetrics - ) -> anyhow::Result<()> { + pub fn new_chunk(&mut self, finalized_head: Option<&BlockRef>, chunk: &StorageChunk) -> anyhow::Result<()> { // FIXME: accept self.first_block rollback limit - let finalized_head = self - .db - .update_dataset_with_hash_index_metrics(self.dataset_id, metrics, |tx| { - let new_finalized_head = match (finalized_head, tx.label().finalized_head()) { - (Some(new), None) => Some(new), - (Some(new), Some(current)) if new.number >= current.number => Some(new), - (_, Some(current)) if current.number < chunk.first_block() => Some(current), - (_, Some(_)) => bail!( - "can't fork safely, because fork base is below the current finalized head \ - and finalized head of the data pack is below the current" - ), - (None, None) => None - }; - - let new_finalized_head = new_finalized_head.map(|head| { - if head.number < chunk.last_block() { - head.clone() - } else { - get_chunk_head(&chunk) - } - }); + let dataset_id = self.dataset_id; + let finalized_head = observe_storage_write(dataset_id, WriteStage::Commit, |metrics| { + self.db + .update_dataset_with_hash_index_metrics(dataset_id, metrics, |tx| { + let new_finalized_head = match (finalized_head, tx.label().finalized_head()) { + (Some(new), None) => Some(new), + (Some(new), Some(current)) if new.number >= current.number => Some(new), + (_, Some(current)) if current.number < chunk.first_block() => Some(current), + (_, Some(_)) => bail!( + "can't fork safely, because fork base is below the current finalized head \ + and finalized head of the data pack is below the current" + ), + (None, None) => None + }; + + let new_finalized_head = new_finalized_head.map(|head| { + if head.number < chunk.last_block() { + head.clone() + } else { + get_chunk_head(&chunk) + } + }); - tx.set_finalized_head(new_finalized_head.clone()); - tx.insert_fork(chunk)?; - Ok(new_finalized_head) - })?; + tx.set_finalized_head(new_finalized_head.clone()); + tx.insert_fork(chunk)?; + Ok(new_finalized_head) + }) + })?; debug!(finalized_head = valuable(&finalized_head), "saved new chunk"); - self.finalized_head = finalized_head; - self.head = Some(get_chunk_head(&chunk)); + // Head before finalized, so a subscriber never observes finalized > head (INV-5). + self.set_head(Some(get_chunk_head(&chunk))); + self.set_finalized_head(finalized_head); if self .first_chunk_head .as_ref() @@ -404,6 +441,79 @@ impl WriteController { Ok(()) } + + /// `retain_from_head` is the `Head(n)` window size, if set; on EXTEND the + /// window is trimmed to keep at most `n` blocks behind the head. + pub fn handle_ingest_msg(&mut self, msg: IngestMessage, retain_from_head: Option) -> anyhow::Result<()> { + match msg { + IngestMessage::FinalizedHead(finalized_head) => { + self.finalize(&finalized_head)?; + } + IngestMessage::NewChunk(new_chunk) => { + let ctx = format!("failed to write new chunk {}", new_chunk); + self.write_new_chunk(new_chunk).context(ctx)?; + if let Some(n) = retain_from_head { + let first_chunk_head = self.first_chunk_head().map(|h| h.number); + if let Some(floor) = trim_floor(first_chunk_head, self.next_block(), n) { + self.retain(floor, None)?; + } + } + } + IngestMessage::Fork { + prev_blocks, + rollback_sender + } => { + self.compute_rollback(&prev_blocks).map(|rollback| { + let _ = rollback_sender.send(rollback); + })?; + } + } + Ok(()) + } + + fn write_new_chunk(&mut self, mut new_chunk: NewChunk) -> anyhow::Result<()> { + let desc = self.dataset_kind().dataset_description(); + let started = StdInstant::now(); + let tables: anyhow::Result<_> = (|| { + let mut tables = BTreeMap::new(); + + for (name, prepared) in new_chunk.tables.iter_mut() { + let mut builder = self.db.new_table_builder(prepared.schema()); + + if let Some(table_desc) = desc.tables.get(name) { + for (&col, opts) in table_desc.options.column_options.iter() { + if opts.stats_enable { + builder.add_stat_by_name(col)?; + } + } + } + + prepared.read(&mut builder, 0, prepared.num_rows())?; + + tables.insert(name.to_string(), builder.finish()?); + } + + Ok(tables) + })(); + report_write_duration(self.dataset_id, WriteStage::Tables, started.elapsed(), tables.is_ok()); + let tables = tables?; + + let chunk = Chunk::V1 { + parent_block_hash: new_chunk.parent_block_hash, + first_block: new_chunk.first_block, + last_block: new_chunk.last_block, + last_block_hash: new_chunk.last_block_hash, + first_block_time: new_chunk.first_block_time, + last_block_time: new_chunk.last_block_time, + tables + }; + + self.new_chunk(new_chunk.finalized_head.as_ref(), &chunk) + } + + pub fn starts_at(&self, block_number: BlockNumber, parent_hash: &Option) -> bool { + self.start_block() == block_number && self.start_block_parent_hash() == parent_hash.as_ref().map(String::as_str) + } } fn get_chunk_head(chunk: &Chunk) -> BlockRef { @@ -412,3 +522,220 @@ fn get_chunk_head(chunk: &Chunk) -> BlockRef { hash: chunk.last_block_hash().to_string() } } + +// Returns the new floor when the tail has to be trimmed to keep +// `max_blocks` behind the tip, or `None` when the window still fits. +// +// `max_blocks` is a soft limit. Since `retain()` only drops whole chunks, trimming +// may keep a part of the first chunk. +fn trim_floor(first_chunk_head: Option, next_block: BlockNumber, max_blocks: u64) -> Option { + let first_chunk_head = first_chunk_head?; + (next_block - first_chunk_head > max_blocks).then(|| next_block - max_blocks) +} + +/// Publish only on a real change: a no-op FINALIZE must not wake +/// `wait_for_finalized_block` waiters. +fn publish(sender: &watch::Sender>, value: Option) { + sender.send_if_modified(|current| { + if *current == value { + false + } else { + *current = value; + true + } + }); +} + +/// Time a storage write and report its duration plus the hash-index counters the +/// storage transaction fills into `metrics`. +fn observe_storage_write( + dataset_id: DatasetId, + stage: WriteStage, + write: impl FnOnce(&mut HashIndexWriteMetrics) -> anyhow::Result +) -> anyhow::Result { + let mut hash_metrics = HashIndexWriteMetrics::default(); + let started = StdInstant::now(); + let result = write(&mut hash_metrics); + let success = result.is_ok(); + report_write_duration(dataset_id, stage, started.elapsed(), success); + report_hash_index_write_metrics(dataset_id, &hash_metrics, success); + result +} + +#[cfg(test)] +mod tests { + use std::{collections::BTreeMap, sync::Arc}; + + use sqd_primitives::BlockRef; + use sqd_storage::db::{Chunk, DatabaseSettings, DatasetId}; + use tokio::sync::watch; + + use super::{WriteController, trim_floor}; + use crate::types::{DBRef, DatasetKind}; + + #[test] + fn nothing_is_trimmed_while_the_window_fits() { + assert_eq!(trim_floor(None, 500, 100), None); + // The whole dataset is one chunk [0..50], well inside the cap. + assert_eq!(trim_floor(Some(50), 51, 100), None); + // Exactly at the cap: the first chunk still has a block in the window. + assert_eq!(trim_floor(Some(0), 100, 100), None); + } + + #[test] + fn the_tail_is_trimmed_once_the_first_chunk_leaves_the_window() { + // First chunk ends at 0, so trimming starts one block past the cap. + assert_eq!(trim_floor(Some(0), 101, 100), Some(1)); + // The soft-limit overshoot: [0..150K] under a 100K cap survives until 250K. + assert_eq!(trim_floor(Some(150_000), 250_000, 100_000), None); + assert_eq!(trim_floor(Some(150_000), 250_001, 100_000), Some(150_001)); + } + + fn block(number: u64, hash: &str) -> BlockRef { + BlockRef { + number, + hash: hash.to_string() + } + } + + /// Head/linkage metadata only — no Arrow tables (an empty table set skips + /// hash indexing; the head comes from `last_block`/`last_block_hash`). + fn chunk(first: u64, last: u64, last_hash: &str, parent_hash: &str) -> Chunk { + Chunk::V1 { + first_block: first, + last_block: last, + last_block_hash: last_hash.to_string(), + parent_block_hash: parent_hash.to_string(), + first_block_time: None, + last_block_time: None, + tables: BTreeMap::new() + } + } + + struct Fixture { + db: DBRef, + dataset_id: DatasetId, + head_rx: watch::Receiver>, + fin_rx: watch::Receiver>, + wc: WriteController, + // Dropped last so RocksDB closes before the directory is removed. + _dir: tempfile::TempDir + } + + fn fixture() -> Fixture { + let dir = tempfile::tempdir().unwrap(); + let db: DBRef = Arc::new(DatabaseSettings::default().open(dir.path()).unwrap()); + let dataset_id = DatasetId::from_str("evm-test"); + let (head_tx, head_rx) = watch::channel(None); + let (fin_tx, fin_rx) = watch::channel(None); + let wc = WriteController::new(db.clone(), dataset_id, DatasetKind::Evm, head_tx, fin_tx).unwrap(); + Fixture { + db, + dataset_id, + head_rx, + fin_rx, + wc, + _dir: dir + } + } + + // INV-30/31: the published head equals what is durable in storage. + #[test] + fn new_chunk_publishes_committed_head() { + let mut f = fixture(); + assert_eq!(*f.head_rx.borrow(), None); + + f.wc.new_chunk(None, &chunk(1, 10, "h10", "h0")).unwrap(); + + assert_eq!(*f.head_rx.borrow(), Some(block(10, "h10"))); + let stored = f.db.snapshot().get_last_chunk(f.dataset_id).unwrap().unwrap(); + assert_eq!(stored.last_block(), 10); + assert_eq!(stored.last_block_hash(), "h10"); + } + + // INV-30: the published finalized head equals the storage label. + #[test] + fn finalize_publishes_committed_finalized_head() { + let mut f = fixture(); + f.wc.new_chunk(None, &chunk(1, 10, "h10", "h0")).unwrap(); + assert_eq!(*f.fin_rx.borrow(), None); + + f.wc.finalize(&block(5, "h5")).unwrap(); + + assert_eq!(*f.fin_rx.borrow(), Some(block(5, "h5"))); + let label = f.db.snapshot().get_label(f.dataset_id).unwrap().unwrap(); + assert_eq!(label.finalized_head(), Some(&block(5, "h5"))); + } + + // INV-40/CN-9: a rebuilt writer reseeds subscribers from committed storage. + #[test] + fn rebuilt_writer_reseeds_watermarks_from_storage() { + let mut f = fixture(); + f.wc.new_chunk(None, &chunk(1, 10, "h10", "h0")).unwrap(); + f.wc.finalize(&block(5, "h5")).unwrap(); + drop(f.wc); + + let (head_tx, head_rx) = watch::channel(None); + let (fin_tx, fin_rx) = watch::channel(None); + let _wc = WriteController::new(f.db.clone(), f.dataset_id, DatasetKind::Evm, head_tx, fin_tx).unwrap(); + + assert_eq!(*head_rx.borrow(), Some(block(10, "h10"))); + assert_eq!(*fin_rx.borrow(), Some(block(5, "h5"))); + } + + // Head-only progress must not fire the finalized channel — no spurious + // `wait_for_finalized_block` wakeups. Guards the `publish` dedup. + #[test] + fn head_only_progress_does_not_wake_finalized_waiters() { + let mut f = fixture(); + f.wc.new_chunk(None, &chunk(1, 10, "h10", "h0")).unwrap(); + // observe the current (still-None) finalized head + assert_eq!(*f.fin_rx.borrow_and_update(), None); + + f.wc.new_chunk(None, &chunk(11, 20, "h20", "h10")).unwrap(); + + assert_eq!(*f.head_rx.borrow(), Some(block(20, "h20"))); + assert!( + !f.fin_rx.has_changed().unwrap(), + "finalized head must stay unchanged while unfinalized blocks arrive" + ); + } + + // Perf probe over the real read (`get_head` == borrow+clone) and write + // (commit→set_head→publish) paths. Run: + // cargo test -p sqd-hotblocks --bin sqd-hotblocks -- --ignored --nocapture watermark_hotpath + #[test] + #[ignore] + fn watermark_hotpath_throughput() { + use std::time::Instant; + + let mut f = fixture(); + f.wc.new_chunk(None, &chunk(1, 10, "h10", "h0")).unwrap(); + + let reads = 2_000_000u32; + let t = Instant::now(); + let mut last = None; + for _ in 0..reads { + last = f.head_rx.borrow().clone(); + } + let read_ns = t.elapsed().as_nanos() as f64 / reads as f64; + assert_eq!(last, Some(block(10, "h10"))); + eprintln!("watermark read: {read_ns:.1} ns/op"); + + let commits = 5_000u64; + let mut parent = "h10".to_string(); + let mut last_block = 10u64; + let t = Instant::now(); + for _ in 0..commits { + let first = last_block + 1; + let last = last_block + 10; + let hash = format!("h{last}"); + f.wc.new_chunk(None, &chunk(first, last, &hash, &parent)).unwrap(); + parent = hash; + last_block = last; + } + let commit_us = t.elapsed().as_micros() as f64 / commits as f64; + eprintln!("chunk commit + publish: {commit_us:.1} us/op"); + assert_eq!(*f.head_rx.borrow(), Some(block(last_block, &parent))); + } +}