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..3e8a3dc8 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` @@ -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, ¤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()); + } }