accumulator: replay missed membership notifications via persisted watermark

A channel created or joined while the daemon was backfilling, reconnecting,
or briefly down never appeared until a process restart: the live-membership
subscription (44100/44101, which are persisted and p-gated) anchored at
"now", taken only after discovery + channel backfill + the profile backfill
— so a notification emitted in that window was never queried again.

The daemon now persists a membership watermark (meta table), advanced when
discovery completes and past each handled notification. The subscription
anchors at the watermark, so missed notifications replay on subscribe and
trigger the existing clean resync; replays terminate because every resync's
discovery re-advances the watermark. One subscription per channel is
unchanged (a resync is a fresh connection).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Riley Crane <rileycrane@squareup.com>

crates/buzz-accumulator/src/daemon/store.rs

@@ -634,6 +634,30 @@ impl Store {
             .await?
             .get::<i64, _>("n"))
     }
+
+    // ---- meta ----
+
+    /// Reads one daemon bookkeeping value (`meta` table).
+    pub async fn get_meta(&self, key: &str) -> Result<Option<String>, sqlx::Error> {
+        Ok(sqlx::query("SELECT value FROM meta WHERE key = ?")
+            .bind(key)
+            .fetch_optional(&self.pool)
+            .await?
+            .map(|r| r.get("value")))
+    }
+
+    /// Writes one daemon bookkeeping value (`meta` table), replacing any prior.
+    pub async fn set_meta(&self, key: &str, value: &str) -> Result<(), sqlx::Error> {
+        sqlx::query(
+            "INSERT INTO meta (key, value) VALUES (?, ?)
+             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+        )
+        .bind(key)
+        .bind(value)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
 }
 
 /// Base signal query: columns + window + selection predicates, no ordering.
@@ -1033,6 +1057,31 @@ mod tests {
         assert_eq!(store.artifacts("f").await.expect("chain").len(), 1);
     }
 
+    #[tokio::test]
+    async fn meta_roundtrip_and_overwrite() {
+        let store = Store::open(":memory:").await.expect("open");
+        assert_eq!(
+            store.get_meta("membership_watermark").await.expect("get"),
+            None
+        );
+        store
+            .set_meta("membership_watermark", "100")
+            .await
+            .expect("set");
+        assert_eq!(
+            store.get_meta("membership_watermark").await.expect("get"),
+            Some("100".into())
+        );
+        store
+            .set_meta("membership_watermark", "200")
+            .await
+            .expect("overwrite");
+        assert_eq!(
+            store.get_meta("membership_watermark").await.expect("get"),
+            Some("200".into())
+        );
+    }
+
     #[tokio::test]
     async fn profile_newest_wins() {
         let store = Store::open(":memory:").await.expect("open");

crates/buzz-accumulator/src/daemon/sync.rs

@@ -104,11 +104,32 @@ async fn sync_once(
     info!(relay = %cfg.relay_url, "connected and authenticated");
 
     discover_channels(&mut conn, cfg, store, registry).await?;
+    // Discovery just read the authoritative rosters, so membership is known
+    // current as of `now` (captured before the roster query). Advancing the
+    // watermark here is what lets the live-membership subscription anchor at
+    // it instead of "now": a 44100/44101 emitted while this connection was
+    // still backfilling — or while the daemon was disconnected — replays on
+    // subscribe and triggers the resync it would otherwise have missed.
+    store
+        .set_meta(MEMBERSHIP_WATERMARK_KEY, &now.to_string())
+        .await?;
     backfill(&mut conn, store, registry).await?;
     backfill_profiles(&mut conn, store).await?;
     live_tail(&mut conn, cfg, store, registry).await
 }
 
+/// Meta key holding the unix-seconds timestamp up to which membership
+/// notifications are known to be reflected in the channel table.
+const MEMBERSHIP_WATERMARK_KEY: &str = "membership_watermark";
+
+/// `since` anchor for the live membership subscription: the persisted
+/// watermark when one exists (replaying anything missed while unsubscribed),
+/// otherwise "now" — discovery has just run either way, the watermark simply
+/// wasn't persisted yet on first boot. Skew keeps the boundary inclusive.
+pub(crate) fn membership_since(watermark: Option<i64>, now: i64) -> i64 {
+    (watermark.unwrap_or(now) - SINCE_SKEW_SECS).max(0)
+}
+
 /// Discovers every channel the key is a member of (39002 → 39000) and
 /// registers the non-archived ones.
 async fn discover_channels(
@@ -393,6 +414,16 @@ async fn live_tail(
     ]))
     .await?;
     tokio::time::sleep(REQ_PACING).await;
+    // Membership notifications are persisted and p-gated, so anchoring at the
+    // watermark (not "now") replays anything emitted while this daemon was
+    // backfilling, reconnecting, or down since the last discovery — e.g. a
+    // channel the person created during startup. Each replayed or live event
+    // triggers the same clean resync, and the resync's discovery advances the
+    // watermark, so replays terminate.
+    let watermark = store
+        .get_meta(MEMBERSHIP_WATERMARK_KEY)
+        .await?
+        .and_then(|v| v.parse::<i64>().ok());
     conn.send_raw(&json!([
         "REQ", "live-membership",
         {
@@ -401,7 +432,7 @@ async fn live_tail(
                 buzz_core::kind::KIND_MEMBER_REMOVED_NOTIFICATION
             ],
             "#p": [me],
-            "since": now - SINCE_SKEW_SECS
+            "since": membership_since(watermark, now)
         }
     ]))
     .await?;
@@ -421,6 +452,16 @@ async fn live_tail(
                 registry.saw_event(chrono::Utc::now().timestamp());
                 if subscription_id == "live-membership" {
                     // Membership changed; the clean move is a full resync.
+                    // Advance the watermark past this notification first —
+                    // the resync's discovery reads the authoritative rosters
+                    // regardless, and this keeps a notification stamped ahead
+                    // of our clock from replaying on every reconnect.
+                    let advance = (event.created_at.as_secs() as i64)
+                        .saturating_add(1)
+                        .max(now);
+                    store
+                        .set_meta(MEMBERSHIP_WATERMARK_KEY, &advance.to_string())
+                        .await?;
                     info!(kind = %event.kind, "membership changed; resyncing");
                     return Ok(());
                 }
@@ -626,6 +667,21 @@ mod tests {
         assert_eq!(backoff_delay(99), Duration::from_secs(60));
     }
 
+    #[test]
+    fn membership_anchor_replays_from_the_watermark() {
+        // With a persisted watermark, the subscription reaches back to it —
+        // notifications emitted while unsubscribed (startup backfill window,
+        // disconnect, downtime) replay instead of vanishing.
+        assert_eq!(
+            membership_since(Some(1_000), 5_000),
+            1_000 - SINCE_SKEW_SECS
+        );
+        // First boot (no watermark yet): discovery just ran, anchor at now.
+        assert_eq!(membership_since(None, 5_000), 5_000 - SINCE_SKEW_SECS);
+        // Never underflows.
+        assert_eq!(membership_since(Some(2), 5), 0);
+    }
+
     #[test]
     fn profile_name_prefers_display_name() {
         assert_eq!(