events: publish the refs a folded checkpoint skipped

crates/walgit-server/src/bridge.rs

@@ -32,10 +32,20 @@ use crate::events::{self, RefEvent, Sink};
 
 const CURSOR_KEY: &str = "events/cursor.json";
 
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+struct PublishedRef {
+    name: String,
+    oid: String,
+}
+
 #[derive(Debug, serde::Serialize, serde::Deserialize)]
 struct Cursor {
     published_seq: u64,
     updated_at: String,
+    /// Refs as of `published_seq`. Older cursors have none: the bridge then
+    /// cannot reconstruct a gap, and stores them on its next successful pass.
+    #[serde(default)]
+    refs: Vec<PublishedRef>,
 }
 
 /// What one `catch_up` did.
@@ -113,29 +123,41 @@ impl Bridge {
         let (cursor, version) = match store.get_bytes(CURSOR_KEY).await? {
             Some((meta, bytes)) => {
                 let c: Cursor = serde_json::from_slice(&bytes).context("events/cursor.json")?;
-                (Some(c.published_seq), Some(meta.version))
+                (Some(c), Some(meta.version))
             }
             None => (None, None),
         };
-        let mut from = cursor.unwrap_or(readable_from);
+        // The local refs after `sync_refs`: used both to fill a gap and to
+        // remember what was published for the next gap.
+        let current = handle.local().refs().context("events bridge: local refs")?;
+        let mut from = cursor.as_ref().map_or(readable_from, |c| c.published_seq);
+        let mut emitted = 0usize;
         let mut gap = None;
         if from < readable_from {
             gap = Some([from + 1, readable_from]);
             metrics::counter!("events_bridge_gap_total", "repo" => id.to_string())
                 .increment(readable_from - from);
             tracing::warn!(repo = %id, first = from + 1, last = readable_from,
-                "events bridge: entries folded into a checkpoint before they were published; consumers backfill from the WAL");
+                "events bridge: entries folded into a checkpoint before they were published; publishing the refs that changed since the cursor");
+            if let Some(previous) = &cursor {
+                let previous = previous
+                    .refs
+                    .iter()
+                    .map(|r| (r.name.clone(), r.oid.clone()))
+                    .collect::<Vec<_>>();
+                let mut batch = Vec::new();
+                events::refs_from_gap(id, &previous, &current.refs, readable_from, &mut batch);
+                if !batch.is_empty() {
+                    // A failing sink returns here: cursor untouched, retried on
+                    // the next wake-up with the same synthetic events.
+                    self.publish(&batch).await?;
+                    emitted += batch.len();
+                }
+            }
             from = readable_from;
         }
         metrics::gauge!("events_bridge_lag_entries", "repo" => id.to_string())
             .set((head - from) as f64);
-        let mut report = CatchUp {
-            repo: id.to_string(),
-            from_seq: from,
-            head_seq: head,
-            emitted: 0,
-            gap,
-        };
 
         if head > from {
             let entries = handle.read_log(from + 1, Some(head)).await?;
@@ -144,13 +166,25 @@ impl Bridge {
             // A failing sink returns here: cursor untouched, retried on the
             // next wake-up.
             self.publish(&batch).await?;
-            report.emitted = batch.len();
+            emitted += batch.len();
         }
 
-        if cursor != Some(head) {
+        let refs = current
+            .refs
+            .iter()
+            .map(|r| PublishedRef {
+                name: r.name.clone(),
+                oid: r.oid.clone(),
+            })
+            .collect::<Vec<_>>();
+        let cursor_moved = cursor
+            .as_ref()
+            .map_or(true, |c| c.published_seq != head || c.refs != refs);
+        if cursor_moved {
             let body = serde_json::to_vec(&Cursor {
                 published_seq: head,
                 updated_at: Utc::now().to_rfc3339(),
+                refs,
             })?;
             let mode = match version {
                 Some(v) => PutMode::Update(v),
@@ -167,11 +201,16 @@ impl Bridge {
             }
         }
         metrics::gauge!("events_bridge_lag_entries", "repo" => id.to_string()).set(0.0);
-        if report.emitted > 0 {
-            tracing::info!(repo = %id, from = from, head = head, emitted = report.emitted,
-                "events bridge: published");
+        if emitted > 0 {
+            tracing::info!(repo = %id, from = from, head = head, emitted, "events bridge: published");
         }
-        Ok(report)
+        Ok(CatchUp {
+            repo: id.to_string(),
+            from_seq: from,
+            head_seq: head,
+            emitted,
+            gap,
+        })
     }
 
     /// Every repo (a `list` + one conditional manifest GET each): the backstop

crates/walgit-server/src/events.rs

@@ -156,6 +156,86 @@ pub(crate) fn refs_from_entries(repo: &RepoId, entries: &[LogEntry], out: &mut V
     }
 }
 
+/// `ref` events for the refs that differ between the last published snapshot
+/// and the current refs.
+///
+/// The bridge uses this when a checkpoint folded log entries it never read: the
+/// original entries are gone, but the cursor stored the refs that were visible
+/// at `published_seq`. The synthetic events carry the checkpoint's seq, so a
+/// consumer's `(repo, seq, ref_name)` dedup still holds on a retry.
+pub(crate) fn refs_from_gap(
+    repo: &RepoId,
+    previous: &[(String, String)],
+    current: &[walgit_git::Ref],
+    seq: u64,
+    out: &mut Vec<RefEvent>,
+) {
+    let previous: std::collections::BTreeMap<&str, &str> = previous
+        .iter()
+        .map(|(name, oid)| (name.as_str(), oid.as_str()))
+        .collect();
+    let current: std::collections::BTreeMap<&str, &str> = current
+        .iter()
+        .map(|r| (r.name.as_str(), r.oid.as_str()))
+        .collect();
+    for (name, oid) in &current {
+        match previous.get(*name) {
+            Some(old) if old == oid => {}
+            Some(old) => out.push(snapshot_event(Action::Update, name, old, oid, seq, repo)),
+            None if name.starts_with("refs/") => {
+                out.push(snapshot_event(
+                    Action::Create,
+                    name,
+                    &zero_like(oid),
+                    oid,
+                    seq,
+                    repo,
+                ));
+            }
+            None => {}
+        }
+    }
+    for (name, old) in &previous {
+        if !current.contains_key(*name) {
+            out.push(snapshot_event(
+                Action::Delete,
+                name,
+                old,
+                &zero_like(old),
+                seq,
+                repo,
+            ));
+        }
+    }
+    out.sort_by(|a, b| a.ref_name.cmp(&b.ref_name));
+}
+
+fn snapshot_event(
+    action: Action,
+    name: &str,
+    old: &str,
+    new: &str,
+    seq: u64,
+    repo: &RepoId,
+) -> RefEvent {
+    RefEvent {
+        action,
+        ref_type: ref_type(name),
+        ref_name: name.to_owned(),
+        old: old.to_owned(),
+        new: new.to_owned(),
+        pusher: "walgit".to_owned(),
+        correlation_id: String::new(),
+        repo: repo.to_string(),
+        walgit: WalgitExt {
+            schema_version: 1,
+            seq: seq.to_string(),
+            entry_kind: "ref_update".to_owned(),
+            request_id: String::new(),
+        },
+    }
+}
+
 /// Where the bridge publishes: a webhook receives each batch as a JSON array,
 /// `POST`ed with `Content-Type: application/json`, a `X-Walgit-Delivery` id
 /// (sha1 of the body — the consumer's dedup key) and, when `events.webhook_secret`
@@ -333,4 +413,51 @@ mod tests {
             (SHA, "0".repeat(40).as_str())
         );
     }
+
+    fn reference(name: &str, oid: &str) -> walgit_git::Ref {
+        walgit_git::Ref {
+            name: name.to_owned(),
+            oid: oid.to_owned(),
+            peeled: String::new(),
+        }
+    }
+
+    #[test]
+    fn gap_events_cover_updates_creates_and_deletes() {
+        let repo = walgit_git::RepoId::new("t", "r").unwrap();
+        let previous = vec![
+            ("refs/heads/main".to_owned(), "aaaa".to_owned()),
+            ("refs/heads/gone".to_owned(), "bbbb".to_owned()),
+        ];
+        let current = vec![
+            reference("refs/heads/main", "cccc"),
+            reference("refs/heads/new", "dddd"),
+        ];
+        let mut out = Vec::new();
+        refs_from_gap(&repo, &previous, &current, 7, &mut out);
+        assert_eq!(out.len(), 3);
+        assert!(out.iter().any(|e| {
+            e.action == Action::Update && e.ref_name == "refs/heads/main" && e.new == "cccc"
+        }));
+        assert!(out.iter().any(|e| {
+            e.action == Action::Create
+                && e.ref_name == "refs/heads/new"
+                && e.old.chars().all(|c| c == '0')
+        }));
+        assert!(
+            out.iter()
+                .any(|e| { e.action == Action::Delete && e.ref_name == "refs/heads/gone" })
+        );
+        assert!(out.iter().all(|e| e.walgit.seq == "7"));
+    }
+
+    #[test]
+    fn gap_events_skip_unchanged_refs() {
+        let repo = walgit_git::RepoId::new("t", "r").unwrap();
+        let previous = vec![("refs/heads/main".to_owned(), "aaaa".to_owned())];
+        let current = vec![reference("refs/heads/main", "aaaa")];
+        let mut out = Vec::new();
+        refs_from_gap(&repo, &previous, &current, 8, &mut out);
+        assert!(out.is_empty());
+    }
 }