jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 21:11:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(accumulator): honest status on resumed runs

Browse Source 
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>
This commit is contained in:
Riley Crane
2026-08-31 16:26:36 -07:00
committed by Sol
parent e987232711
commit 6aa6e37b01
1 changed files with 16 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
crates/buzz-accumulator/src/daemon/sync.rs
View File

@@ -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!([