From 6aa6e37b01b3358142c1919ce75e5256dafc1065 Mon Sep 17 00:00:00 2001
From: Riley Crane <rileycrane@squareup.com>
Date: Mon, 31 Aug 2026 16:26:36 -0700
Subject: [PATCH] fix(accumulator): honest status on resumed runs

A restart with a completed mirror skipped backfill without telling the
status registry, so /status reported backfill_complete=false forever
and newest_ts stayed null until a live event happened to land. Done
channels now mark themselves at startup, and live subscriptions seed
newest_ts from the mirror.

Verified live: restart on a warm mirror reports backfill_complete=true
and per-channel recency within seconds, zero pages re-fetched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Riley Crane <rileycrane@squareup.com>
---
 crates/buzz-accumulator/src/daemon/sync.rs | 21 ++++++++++++++++-----
 1 file changed, 16 insertions(+), 5 deletions(-)

diff --git a/crates/buzz-accumulator/src/daemon/sync.rs b/crates/buzz-accumulator/src/daemon/sync.rs
index 2d9cb228af..e65e59ddad 100644
--- a/crates/buzz-accumulator/src/daemon/sync.rs
+++ b/crates/buzz-accumulator/src/daemon/sync.rs
@@ -205,7 +205,12 @@ async fn backfill(
     registry: &StatusRegistry,
 ) -> anyhow::Result<()> {
     let channels = store.channels().await?;
-    for ch in channels.iter().filter(|c| c.active && !c.backfill_done) {
+    for ch in channels.iter().filter(|c| c.active) {
+        if ch.backfill_done {
+            // Already complete in a prior run; reflect that in status.
+            registry.channel(&ch.id, |c| c.backfill = "done".into());
+            continue;
+        }
         let mut cursor = ch
             .backfill_cursor
             .unwrap_or_else(|| chrono::Utc::now().timestamp());
@@ -345,9 +350,8 @@ async fn live_tail(
     let channels = store.channels().await?;
     for ch in channels.iter().filter(|c| c.active) {
         tokio::time::sleep(REQ_PACING).await;
-        let since = store
-            .newest_ts(&ch.id)
-            .await?
+        let newest = store.newest_ts(&ch.id).await?;
+        let since = newest
             .unwrap_or(ch.discovered_at)
             .saturating_sub(SINCE_SKEW_SECS);
         conn.send_raw(&json!([
@@ -356,7 +360,14 @@ async fn live_tail(
             { "#h": [ch.id], "since": since }
         ]))
         .await?;
-        registry.channel(&ch.id, |c| c.live = true);
+        registry.channel(&ch.id, |c| {
+            c.live = true;
+            // Seed from the mirror so a resumed run shows real recency
+            // before the first live event lands.
+            if let Some(ts) = newest {
+                c.newest_ts = Some(c.newest_ts.unwrap_or(ts).max(ts));
+            }
+        });
     }
     tokio::time::sleep(REQ_PACING).await;
     conn.send_raw(&json!([