fix(wal): read the log and replay refs through the caller's manifest

crates/walgit-cli/src/settings_cmd.rs

@@ -87,7 +87,7 @@ pub async fn run(action: SettingsAction, cfg: &Arc<Config>) -> Result<()> {
             h.sync_refs().await?;
             let m = h.manifest();
             let from = m.min_seq.max(1);
-            let entries = h.read_log(from, None).await?;
+            let entries = h.read_log(&m, from, None).await?;
             let mut n = 0;
             for e in entries
                 .iter()

crates/walgit-cli/src/wal_cmd.rs

@@ -23,7 +23,8 @@ pub async fn run(action: WalAction, cfg: &Arc<Config>) -> Result<()> {
             let (owner, name) = parse_repo_id(&repo)?;
             let id = walgit_git::RepoId::new(owner, name)?;
             let handle = registry.open(&id).await?;
-            let entries = handle.read_log(from.unwrap_or(0), to).await?;
+            let manifest = handle.fresh_manifest().await?;
+            let entries = handle.read_log(&manifest, from.unwrap_or(0), to).await?;
 
             if entries.is_empty() {
                 println!("(no WAL entries)");
@@ -136,7 +137,8 @@ pub async fn run(action: WalAction, cfg: &Arc<Config>) -> Result<()> {
             let (owner, name) = parse_repo_id(&repo)?;
             let id = walgit_git::RepoId::new(owner, name)?;
             let handle = registry.open(&id).await?;
-            let entries = handle.read_log(seq, Some(seq)).await?;
+            let manifest = handle.fresh_manifest().await?;
+            let entries = handle.read_log(&manifest, seq, Some(seq)).await?;
 
             let entry = entries
                 .into_iter()
@@ -289,11 +291,17 @@ pub async fn materialize_at(
             }
         }
         if at_seq >= manifest.min_seq {
-            found.extend(handle.read_log(manifest.min_seq, Some(at_seq)).await?);
+            found.extend(
+                handle
+                    .read_log(&manifest, manifest.min_seq, Some(at_seq))
+                    .await?,
+            );
         }
         found
     } else {
-        handle.read_log(start_seq + 1, Some(at_seq)).await?
+        handle
+            .read_log(&manifest, start_seq + 1, Some(at_seq))
+            .await?
     };
     info!(entries = entries.len(), "replaying log entries");
     for entry in &entries {

crates/walgit-server/src/bridge.rs

@@ -138,7 +138,7 @@ impl Bridge {
         };
 
         if head > from {
-            let entries = handle.read_log(from + 1, Some(head)).await?;
+            let entries = handle.read_log(&manifest, from + 1, Some(head)).await?;
             let mut batch = Vec::new();
             events::refs_from_entries(id, &entries, &mut batch);
             // A failing sink returns here: cursor untouched, retried on the

crates/walgit-server/src/settings.rs

@@ -108,7 +108,7 @@ pub async fn http_history(
         .map_err(|e| ApiError::Internal(e.to_string()))?;
     let m = h.manifest();
     let entries = h
-        .read_log(m.min_seq.max(1), None)
+        .read_log(&m, m.min_seq.max(1), None)
         .await
         .map_err(|e| ApiError::Internal(e.to_string()))?;
     let items: Vec<serde_json::Value> = entries
@@ -402,7 +402,7 @@ pub async fn http_policy_dry_run(
     };
     let m = h.manifest();
     let entries = h
-        .read_log(m.min_seq.max(1), None)
+        .read_log(&m, m.min_seq.max(1), None)
         .await
         .map_err(|e| ApiError::Internal(e.to_string()))?;
     let pushes: Vec<&walgit_proto::v1::LogEntry> = entries

crates/walgit-server/src/web/ui.rs

@@ -529,13 +529,11 @@ async fn overview(
     let id =
         walgit_git::RepoId::new(&owner, &repo).map_err(|e| ApiError::NotFound(e.to_string()))?;
     let handle = state.registry.open(&id).await.map_err(wal_err)?;
-    // read_log performs its own freshness check; acquire the read guard only
-    // after it has completed because read_log may need the write lock.
-    let entries = handle.read_log(1, None).await.map_err(wal_err)?;
     // Refs-level sync only: the overview must render for repos whose packs do
     // not fit this instance (that is exactly when people look at it).
     let _guard = handle.sync_refs().await.map_err(wal_err)?;
     let manifest = handle.manifest();
+    let entries = handle.read_log(&manifest, 1, None).await.map_err(wal_err)?;
     let version = handle
         .manifest_version()
         .map(|version| version.to_string())

crates/walgit-server/tests/events.rs

@@ -261,3 +261,58 @@ async fn bridge_sink_failure_keeps_the_cursor() -> TestResult {
     assert_eq!(resp.status(), 503, "non-2xx so Pub/Sub redelivers");
     Ok(())
 }
+
+/// A checkpoint that lands while the bridge is catching up must not lose the
+/// entries it folds: the bridge takes `head_seq` from the manifest it holds,
+/// and another instance's checkpoint then trims the store's manifest.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn bridge_publishes_entries_a_concurrent_checkpoint_folds() -> TestResult {
+    let (url, captured) = webhook().await;
+    // The TTL keeps the bridge on its held manifest, so the interleaving
+    // needs no timing.
+    let server = Server::start_with_tweak(|c| {
+        bridge_cfg(&url, Duration::ZERO)(c);
+        c.wal.freshness_ttl = Duration::from_mins(10);
+    })
+    .await?;
+    let bridge = server.state.bridge.clone().expect("bridge enabled");
+    server.put_repo("t", "r").await?;
+    let id = walgit_git::RepoId::new("t", "r")?;
+
+    let src = TestRepo::synthetic(1, 1)?;
+    git_in(&src, &["commit", "--allow-empty", "-m", "a"])?;
+    git_in(&src, &["branch", "-M", "main"])?;
+    git_in(
+        &src,
+        &["remote", "add", "origin", &server.repo_url("t", "r")],
+    )?;
+    git_in(&src, &["push", "-u", "origin", "main"])?;
+    let r = bridge.catch_up(&id).await?;
+    assert_eq!((r.head_seq, r.emitted), (1, 1));
+    wait_for(&captured, 1).await;
+
+    git_in(&src, &["commit", "--allow-empty", "-m", "b"])?;
+    git_in(&src, &["push"])?;
+
+    // Another instance folds seq 2 into a checkpoint.
+    let sibling = server.start_sibling_with(|_| {}).await?;
+    let cp = sibling
+        .state
+        .registry
+        .open(&id)
+        .await?
+        .write_checkpoint()
+        .await?;
+    assert_eq!(cp.seq, 2);
+
+    let r = bridge.catch_up(&id).await?;
+    assert_eq!(
+        (r.from_seq, r.head_seq, r.emitted, r.gap),
+        (1, 2, 1, None),
+        "seq 2 was committed and seen; it must be published, not skipped"
+    );
+    let got = wait_for(&captured, 2).await;
+    assert_eq!(got[1]["_walgit"]["seq"], "2");
+    assert_eq!(cursor_seq(&server, "t", "r").await, Some(2));
+    Ok(())
+}

crates/walgit-server/tests/follow.rs

@@ -88,7 +88,7 @@ async fn follows_upstream_refs_through_the_wal_fast_forward_only() -> anyhow::Re
     );
     let h = step!("open", fo.state.registry.open(&id))?;
     assert_eq!(h.manifest().head_seq, 1);
-    let entries = step!("log", h.read_log(1, None))?;
+    let entries = step!("log", h.read_log(&h.manifest(), 1, None))?;
     assert_eq!(
         entries[0].meta.get("principal").map(String::as_str),
         Some("upstream")

crates/walgit-server/tests/harness.rs

@@ -192,7 +192,9 @@ impl Server {
     ) -> Result<Vec<walgit_proto::v1::LogEntry>> {
         let id = walgit_git::RepoId::new(owner, repo)?;
         let handle = self.registry.open(&id).await?;
-        Ok(handle.read_log(1, None).await?)
+        Ok(handle
+            .read_log(&*handle.fresh_manifest().await?, 1, None)
+            .await?)
     }
 
     /// True when the instance's local copy of the repo has its pack set.

crates/walgit-server/tests/sim.rs

@@ -501,7 +501,7 @@ async fn check_truth(c: &Cluster, pushers: &[Pusher]) -> Result<()> {
         }
     }
     let log = handle
-        .read_log(cp_seq + 1, None)
+        .read_log(&manifest, cp_seq + 1, None)
         .await
         .context("observer read_log")?;
     ensure!(

crates/walgit-wal/src/handle.rs

@@ -1384,7 +1384,7 @@ impl RepoHandle {
         // checkpoint (seq 1, 08-19 21:33Z) was ignored because its manifest ref
         // carries no times; the bundler then cut "08-19 23:00" slots from today's
         // main. The times now come from the checkpoint object when the ref has none.
-        let cp = self.checkpoint_times();
+        let cp = self.checkpoint_times(&self.manifest());
         [
             cp.as_ref().and_then(|c| c.first_state_at),
             cp.as_ref().and_then(|c| c.created_at),
@@ -1395,11 +1395,10 @@ impl RepoHandle {
         .min()
     }
 
-    /// Times of the current checkpoint: from the manifest's `CheckpointRef`, else
-    /// from the checkpoint object read during sync (`checkpoint_times` cache).
+    /// Times of `m`'s checkpoint: from its `CheckpointRef`, else from the
+    /// checkpoint object read during sync (`checkpoint_times` cache).
     /// None = no checkpoint, or its times are not known to this process yet.
-    pub fn checkpoint_times(&self) -> Option<CheckpointTimes> {
-        let m = self.manifest();
+    pub fn checkpoint_times(&self, m: &Manifest) -> Option<CheckpointTimes> {
         let cp = m.checkpoint.as_ref()?;
         let from_ref = CheckpointTimes {
             created_at: cp.created_at.as_ref().map(walgit_proto::time::to_system),
@@ -1418,14 +1417,13 @@ impl RepoHandle {
         }
     }
 
-    /// Read the checkpoint object's times when the manifest ref has none
+    /// Read `m`'s checkpoint object's times when its ref has none
     /// (one 240-byte GET per checkpoint per process; no-op otherwise).
-    pub(crate) async fn learn_checkpoint_times(&self) -> Result<(), WalError> {
+    pub(crate) async fn learn_checkpoint_times(&self, m: &Manifest) -> Result<(), WalError> {
         use walgit_store::ObjectStoreExt;
 
         use prost::Message;
 
-        let m = self.manifest();
         let Some(cp) = m.checkpoint.as_ref() else {
             return Ok(());
         };
@@ -1452,23 +1450,43 @@ impl RepoHandle {
     /// refs-level, pure). See `log_reader::refs_as_of`.
     pub async fn refs_as_of(
         &self,
+        manifest: &Manifest,
         at: std::time::SystemTime,
     ) -> Result<(walgit_proto::v1::RefSnapshot, u64), WalError> {
-        crate::log_reader::refs_as_of(self, at).await
+        crate::log_reader::refs_as_of(self, manifest, at).await
     }
 
     /// Ref state at WAL `seq` exactly (refs-level, pure). See `log_reader::refs_at_seq`.
-    pub async fn refs_at_seq(&self, seq: u64) -> Result<walgit_proto::v1::RefSnapshot, WalError> {
-        crate::log_reader::refs_at_seq(self, seq).await
+    pub async fn refs_at_seq(
+        &self,
+        manifest: &Manifest,
+        seq: u64,
+    ) -> Result<walgit_proto::v1::RefSnapshot, WalError> {
+        crate::log_reader::refs_at_seq(self, manifest, seq).await
+    }
+
+    /// The freshest manifest: the one this handle holds, or the store's if it
+    /// is newer. A lock-free conditional GET, so a caller holding a read guard
+    /// can call it.
+    pub async fn fresh_manifest(&self) -> Result<Arc<Manifest>, WalError> {
+        let (held, known) = self.manifest_pair();
+        Ok(
+            match crate::sync::freshness_check(&self.store, known.as_ref()).await? {
+                crate::sync::SyncOutcome::Unchanged => held,
+                crate::sync::SyncOutcome::Changed { manifest, .. } => manifest,
+            },
+        )
     }
 
-    /// Read log entries [`from_seq`, `to_seq`].
+    /// Read log entries [`from_seq`, `to_seq`] through `manifest`, the one the
+    /// bounds came from: a later checkpoint trims segments only from its own.
     pub async fn read_log(
         &self,
+        manifest: &Manifest,
         from_seq: u64,
         to_seq: Option<u64>,
     ) -> Result<Vec<walgit_proto::v1::LogEntry>, WalError> {
-        crate::log_reader::read_log_impl(self, from_seq, to_seq).await
+        crate::log_reader::read_log_manifest(self, manifest, from_seq, to_seq).await
     }
 
     // ---- internal helpers ----

crates/walgit-wal/src/log_reader.rs

@@ -6,25 +6,8 @@ use walgit_store::{GetOptions, GetResult, ObjectStore};
 use crate::error::WalError;
 use crate::handle::RepoHandle;
 
-/// Read log entries in [`from_seq`, `to_seq`]. If `to_seq` is None, read up to
-/// `manifest.head_seq`.
-pub(crate) async fn read_log_impl(
-    handle: &RepoHandle,
-    from_seq: u64,
-    to_seq: Option<u64>,
-) -> Result<Vec<LogEntry>, WalError> {
-    // Reading the log needs a *fresh manifest*, not a synced local copy: do a
-    // lock-free conditional GET and use whichever manifest is newer. Taking
-    // the repo's write lock here would deadlock callers that hold a read
-    // guard (overview, tests), and freshness_ttl=0 makes that the common case.
-    let (held, known) = handle.manifest_pair();
-    let manifest = match crate::sync::freshness_check(&handle.store, known.as_ref()).await? {
-        crate::sync::SyncOutcome::Unchanged => held,
-        crate::sync::SyncOutcome::Changed { manifest, .. } => manifest,
-    };
-    read_log_manifest(handle, &manifest, from_seq, to_seq).await
-}
-
+/// Read log entries in [`from_seq`, `to_seq`] through `manifest`'s segments. If
+/// `to_seq` is None, read up to `manifest.head_seq`.
 pub(crate) async fn read_log_manifest(
     handle: &RepoHandle,
     manifest: &walgit_proto::v1::Manifest,
@@ -80,9 +63,10 @@ pub(crate) async fn read_log_manifest(
 /// the last entry applied (0 = none).
 pub async fn refs_as_of(
     handle: &super::handle::RepoHandle,
+    manifest: &walgit_proto::v1::Manifest,
     at: std::time::SystemTime,
 ) -> Result<(walgit_proto::v1::RefSnapshot, u64), WalError> {
-    replay_refs(handle, Cut::Time(at)).await
+    replay_refs(handle, manifest, Cut::Time(at)).await
 }
 
 /// Ref state **at** WAL `seq` exactly (the newest checkpoint at or before it + every ref
@@ -91,9 +75,9 @@ pub async fn refs_as_of(
 /// compose, whose header must carry the refs at the base pack's seq whatever moved since.
 pub async fn refs_at_seq(
     handle: &super::handle::RepoHandle,
+    manifest: &walgit_proto::v1::Manifest,
     seq: u64,
 ) -> Result<walgit_proto::v1::RefSnapshot, WalError> {
-    let manifest = handle.manifest();
     if seq > manifest.head_seq {
         return Err(WalError::Corrupt(format!("future refs sequence {seq}")));
     }
@@ -104,7 +88,7 @@ pub async fn refs_at_seq(
             manifest.min_seq
         )));
     }
-    Ok(replay_refs(handle, Cut::Seq(seq)).await?.0)
+    Ok(replay_refs(handle, manifest, Cut::Seq(seq)).await?.0)
 }
 
 enum Cut {
@@ -114,9 +98,9 @@ enum Cut {
 
 async fn replay_refs(
     handle: &super::handle::RepoHandle,
+    manifest: &walgit_proto::v1::Manifest,
     cut: Cut,
 ) -> Result<(walgit_proto::v1::RefSnapshot, u64), WalError> {
-    let manifest = handle.manifest();
     // Start point: checkpoint ≤ cut (checkpoint created_at on the ref; when
     // the ref has no timestamp, fall back to replaying from seq 0).
     let mut snap = walgit_proto::v1::RefSnapshot::default();
@@ -129,8 +113,8 @@ async fn replay_refs(
         let usable = match cut {
             Cut::Seq(seq) => cp.seq <= seq,
             Cut::Time(at) => {
-                handle.learn_checkpoint_times().await?;
-                let times = handle.checkpoint_times();
+                handle.learn_checkpoint_times(manifest).await?;
+                let times = handle.checkpoint_times(manifest);
                 let cp_time = times.and_then(|t| t.as_of.or(t.created_at));
                 cp_time.is_some_and(|t| t <= at)
             }
@@ -151,7 +135,7 @@ async fn replay_refs(
         Cut::Seq(seq) => Some(seq),
         Cut::Time(_) => None,
     };
-    let entries = read_log_manifest(handle, &manifest, from_seq + 1, to_seq).await?;
+    let entries = read_log_manifest(handle, manifest, from_seq + 1, to_seq).await?;
     let mut map: std::collections::BTreeMap<String, walgit_proto::v1::Ref> =
         snap.refs.into_iter().map(|r| (r.name.clone(), r)).collect();
     let mut head_target = snap.head_target;

crates/walgit-wal/src/snapshots.rs

@@ -257,6 +257,6 @@ impl RepoHandle {
         {
             return checkpoint_snapshot(&self.store, cp, &manifest.object_format).await;
         }
-        crate::log_reader::refs_at_seq(self, seq).await
+        crate::log_reader::refs_at_seq(self, &manifest, seq).await
     }
 }

crates/walgit-wal/src/sync.rs

@@ -483,7 +483,7 @@ pub(crate) async fn apply_delta(
 
     // The checkpoint's times feed `first_state_time` / `refs_as_of`; old refs
     // carry none, the object always does.
-    handle.learn_checkpoint_times().await?;
+    handle.learn_checkpoint_times(new_manifest).await?;
     if need_checkpoint_load {
         let cp = new_manifest
             .checkpoint