From 4046b96bf552cf2154f2e29cb13d581d43970325 Mon Sep 17 00:00:00 2001 From: Coulm Date: Sat, 24 Oct 2026 13:47:15 +0200 Subject: [PATCH 1/2] fix(events): publish the refs a folded checkpoint skipped A checkpoint can fold log entries the events bridge has not read yet; the bridge then detects a gap and skipped them, so consumers lost the push. Store the refs visible at the published seq in the cursor and, on a gap, diff them against the current refs and emit the missing create/update/delete events with the checkpoint seq (so (repo, seq, ref) dedup still holds on retry). Older cursors without refs keep the previous skip behaviour and are populated on the next successful pass. --- crates/walgit-server/src/bridge.rs | 71 ++++++++++++++++++++------ crates/walgit-server/src/events.rs | 80 ++++++++++++++++++++++++++++++ 2 files changed, 135 insertions(+), 16 deletions(-) diff --git a/crates/walgit-server/src/bridge.rs b/crates/walgit-server/src/bridge.rs index 349ff181..eaee91ce 100644 --- a/crates/walgit-server/src/bridge.rs +++ b/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, } /// 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::>(); + let mut batch = Vec::new(); + events::refs_from_gap(id, &previous, ¤t.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::>(); + 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 diff --git a/crates/walgit-server/src/events.rs b/crates/walgit-server/src/events.rs index 4d4d77bb..e204b5a4 100644 --- a/crates/walgit-server/src/events.rs +++ b/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, +) { + 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 ¤t { + 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` From ff6714f532ffe18b9d7baf797f1405595f6737b3 Mon Sep 17 00:00:00 2001 From: Coulm Date: Sat, 24 Oct 2026 14:23:24 +0200 Subject: [PATCH 2/2] test(events): cover the gap ref diff --- crates/walgit-server/src/events.rs | 47 ++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/crates/walgit-server/src/events.rs b/crates/walgit-server/src/events.rs index e204b5a4..3e8a3dc8 100644 --- a/crates/walgit-server/src/events.rs +++ b/crates/walgit-server/src/events.rs @@ -413,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, ¤t, 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, ¤t, 8, &mut out); + assert!(out.is_empty()); + } }