diff --git a/CLAUDE.md b/CLAUDE.md index 4dd15ce..d6f65b5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -65,6 +65,17 @@ twenty fresh samples exist, which is `MIN_TIP_SAMPLES` doing its job: nominal an labelled nominal beats measured and wrong. The same applies to an ordinary burst import, where the next head is several heights on with no elapsed time. +**Fixing the code did not fix the rows already written.** `at_tip` in the table +is whatever the process that wrote it believed, and every process before the +receipt-stamping fix believed a drained channel was a tip observation. Those rows +are still there, and `restore_ticker_and_timing` replayed them on every start — +Heisenberg reported **0.054 s/block** while producing one block every five +minutes, entirely from restored samples, with the running code already correct. +Restored samples are now checked against the one clock the observer cannot fake: +at the tip a block is seen roughly when it is authored, so the observed span +should resemble the authored span. Propagation and skew are seconds; a catch-up +is a factor. A batch that fails goes unrestored and the interval starts nominal. + **`observed_at` is stamped where the head arrives, not where it is handled.** Every head costs two RPC round trips inside `record`, so on a remote endpoint the 64-deep head channel backs up and drains in a burst — and a batch stamped with diff --git a/crates/blackbeard-api/src/ingest.rs b/crates/blackbeard-api/src/ingest.rs index 0d9a5b8..6149d42 100644 --- a/crates/blackbeard-api/src/ingest.rs +++ b/crates/blackbeard-api/src/ingest.rs @@ -239,6 +239,32 @@ pub async fn warm_start(chain: &Arc, store: &Store, ticker_blocks: restore_ticker_and_timing(chain, store, ticker_blocks).await; } +/// Whether a batch of stored tip observations can be believed as timings. +/// +/// Compares the span the observer recorded against the span the chain recorded +/// for the same blocks. Anything the observer saw at the tip was seen roughly +/// when it was authored, so the two agree to within propagation and clock skew. +/// A batch written while catching up does not: its heights span hours of chain +/// time and seconds of wall clock. +/// +/// `true` when there is nothing to judge — too few samples, or no authored +/// timestamps — because the existing guards downstream (`MIN_TIP_SAMPLES`) then +/// decide, and this is not the place to invent a second opinion. +fn tip_samples_agree_with_the_chain(samples: &[&Observed]) -> bool { + let (Some(first), Some(last)) = (samples.first(), samples.last()) else { + return true; + }; + let (Some(first_at), Some(last_at)) = (first.authored_at, last.authored_at) else { + return true; + }; + let authored = (last_at - first_at).as_seconds_f64(); + let observed = (last.observed_at - first.observed_at).as_seconds_f64(); + if authored <= 0.0 { + return true; + } + observed >= authored / 2.0 +} + /// How stale a tip observation may be and still seed the measured interval. /// /// The interval is `span / height difference`, which stays arithmetically valid @@ -284,11 +310,40 @@ async fn restore_ticker_and_timing(chain: &Arc, store: &Store, tic let mut ticker = Vec::new(); let mut previous: Option> = None; + // Rows this process did not write, checked against the chain's own clock + // before they are trusted to time anything. + // + // `at_tip` in the database is whatever the process that wrote the row + // believed at the time, and processes before #14 believed every head-stream + // block was a tip observation — including a batch drained from a backed-up + // channel, which shares an `observed_at` while its heights march on. Those + // rows are still in the table and restoring them reproduces the bug on + // every start: Heisenberg reported **0.054 s/block** while producing one + // block every five minutes, entirely from restored samples. + // + // The test is the one thing the observer cannot fake. At the tip, blocks + // are seen roughly as they are authored, so the span the observer recorded + // should resemble the span the *chain* recorded. Propagation and clock skew + // are seconds; a catch-up is a factor. Half is generous to the first and + // fatal to the second. + let candidates: Vec<_> = oldest_first + .iter() + .filter(|(observed, at_tip)| *at_tip && observed.observed_at >= cutoff) + .map(|(observed, _)| observed) + .collect(); + let trustworthy = tip_samples_agree_with_the_chain(&candidates); + if !trustworthy && !candidates.is_empty() { + tracing::info!( + chain = %id, samples = candidates.len(), + "warm start: tip samples disagree with the chain's clock; timing starts fresh" + ); + } + { let mut inner = chain.write(); - for (observed, at_tip) in &oldest_first { - if *at_tip && observed.observed_at >= cutoff { - inner.window.push(observed.clone(), true); + if trustworthy { + for observed in &candidates { + inner.window.push((*observed).clone(), true); restored_samples += 1; } } @@ -1447,6 +1502,45 @@ fn from_millis(ms: u64) -> Option> { mod tests { use super::*; + /// Rows written while catching up must not come back as timings. + /// + /// #14: `at_tip` in the table is whatever the writing process believed, and + /// every process before the fix believed a drained channel was a tip + /// observation. Heisenberg reported 0.054 s/block while producing one block + /// every five minutes, entirely from restored rows — the code was fixed and + /// the data was not. + #[test] + fn restored_tip_samples_are_checked_against_the_chains_clock() { + let start = Utc::now(); + let row = |h: u64, authored: i64, observed: i64| Observed { + height: h, + miner: blackbeard_entities::MinerId("m".into()), + observed_at: start + chrono::TimeDelta::seconds(observed), + authored_at: Some(start + chrono::TimeDelta::seconds(authored)), + difficulty: None, + }; + + // Seen as authored: five minutes a block, both clocks agreeing. + let honest: Vec = (0..20) + .map(|i| row(i, i as i64 * 300, i as i64 * 300)) + .collect(); + let honest: Vec<&Observed> = honest.iter().collect(); + assert!(tip_samples_agree_with_the_chain(&honest)); + + // The same blocks, drained from a backlog in two seconds. + let caught_up: Vec = (0..20) + .map(|i| row(i, i as i64 * 300, i as i64 / 10)) + .collect(); + let caught_up: Vec<&Observed> = caught_up.iter().collect(); + assert!(!tip_samples_agree_with_the_chain(&caught_up)); + + // Nothing to judge: let the downstream guards decide rather than + // inventing a second opinion here. + assert!(tip_samples_agree_with_the_chain(&[])); + let one = row(1, 0, 0); + assert!(tip_samples_agree_with_the_chain(&[&one])); + } + #[test] fn a_plausible_timestamp_converts() { let t = from_millis(1_757_000_000_000).unwrap();