fix: don't restore tip samples the chain's own clock contradicts

I closed #14 on an incomplete fix. Heisenberg still reports **0.054 s/block**
while producing one block every five minutes — 5,500x too fast — with the
running code already correct.

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` replays them on every
start, so the bug comes back from storage each time a chain is restarted after a
catch-up. Heisenberg produces a block every five minutes, which is far too slow
to accumulate twenty fresh samples before the next restart, so it never got the
chance to correct itself.

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 span the observer
recorded should resemble the span the chain recorded for the same blocks.
Propagation and skew are seconds; a catch-up is a factor. A batch that fails is
left unrestored and timing starts fresh — nominal, and labelled nominal.

The threshold is deliberately loose (half) because it only has to separate two
populations that differ by orders of magnitude, and being generous to honest
skew costs nothing: the penalty for a false negative is one window of nominal
hashrate, and for a false positive it is a headline wrong by 5,000x.

Reopens #14 — closing it again when all three chains have held a plausible
interval across a restart, rather than when the code looks right.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jp6a8EDar9ueEhAxzep4V5

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

crates/blackbeard-api/src/ingest.rs

@@ -239,6 +239,32 @@ pub async fn warm_start(chain: &Arc<ChainRuntime>, 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<ChainRuntime>, store: &Store, tic
     let mut ticker = Vec::new();
     let mut previous: Option<DateTime<Utc>> = 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<DateTime<Utc>> {
 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<Observed> = (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<Observed> = (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();