fix(storage): harden marker machinery — durable pre-ack, locked divergence checks, stable lock hash

Address beardthelion + jatmn rounds on 75060ea/f154768, plus findings
from an internal adversarial review pass:

Reviewer findings:
- Marker persistence is fallible and atomic (tmp+rename); the write-back
  path only acks after the marker is durably on disk, else it falls back
  to strict upload-before-ack. An existing marker's base is preserved on
  re-mark so consecutive upload failures can't corrupt it into a false
  divergence (regression test per the reported repro).
- release_after_write (fork publication) now uploads under the per-repo
  advisory lock, closing the torn-snapshot race against pushes to a
  freshly addressable fork; on divergence it refuses and leaves the
  local copy marked instead of clobbering the concurrent writer.
- Startup sweep (retry_pending_uploads) re-uploads marked repos and
  clears markers; diverged repos stay marked and are counted in a new
  gitlawb_pending_upload_markers gauge.
- try_exists for the marker probe (EIO must not read as "no marker");
  markers skipped entirely in local-only mode; stale markers cleared on
  repo recreation (init) and in fork's undo_claim.
- fs backend: etag sidecar published before the blob rename; sidecar
  content asserted in tests; GITLAWB_ADVISORY_LOCK_POOL_SIZE rejects 0
  at parse; claim-first comments scoped to per-node databases.
- New FlakyStore failing-backend double + sqlx-backed guard tests cover
  the ack window, consecutive failures, failed existence checks, init
  cleanup, and head-failure branches (531 tests).

Internal review findings:
- retry/fork upload decisions now happen UNDER the advisory lock
  (upload_locked_with_marker): an unlocked base check could pass, block
  on a concurrent push's lock, then clobber that push's fresh upload.
- The marker base used for the locked check is read from the marker
  itself, not the versions cache — the cache is invalidated by upload
  failures and lost to restarts.
- Marker cleared before the lock releases, and its base is atomically
  refreshed to the new etag before unlink, so a crash between upload
  and clear self-heals instead of wedging on spurious divergence.
- Lazy migration skips marked repos instead of stranding their markers.
- advisory_lock_key: DefaultHasher (unstable across Rust releases)
  replaced with a SHA-256 prefix — mixed-toolchain rolling deploys can
  no longer silently lose cross-machine mutual exclusion.
- fs put serializes the sidecar+blob publish pair per key so concurrent
  same-key puts can't pair one put's etag with another's content.
- Known limitations documented in place: unlocked read-download swap
  (pre-existing) and the fork creation compound-failure window.

crates/gitlawb-node/src/api/repos.rs

@@ -208,10 +208,12 @@ pub async fn create_repo(
     let verified_proof = proof.consume(&state.db).await?;
 
     // Claim-first ordering: insert the DB row before creating anything durable.
-    // The row is the claim on (owner, name) — a concurrent same-name create
-    // loses at the insert with nothing on disk or in storage yet, so failure
-    // compensation below only ever touches state THIS attempt created and can
-    // never delete a racing winner's repo or archive.
+    // Within one node's database the row is the claim on (owner, name) — a
+    // concurrent same-name create loses at the insert with nothing on disk or
+    // in storage yet. Across nodes (each fly app has its own Postgres) the
+    // insert arbitrates nothing; the cross-node safety property is that
+    // failure compensation below only ever touches state THIS attempt created
+    // (its own row by id, its own local dir) and never deletes a storage key.
     let disk_path = store::repo_disk_path(&state.config.repos_dir, &owner_did, &req.name);
     let now = Utc::now();
     let record = crate::db::RepoRecord {
@@ -955,7 +957,11 @@ pub async fn git_receive_pack(
     // from blocking subsequent pushes. Only upload to storage when the push
     // succeeded; uploading a half-applied repo would propagate corruption.
     let push_ok = receive_result.is_ok();
+    // `Some` = the guard still needs a synchronous (strict) release; taken by
+    // the write-back path only once its intent marker is durably on disk.
+    let mut strict_guard = Some(guard);
     if push_ok && state.config.async_upload {
+        let guard = strict_guard.take().expect("guard present before release");
         // Write-back: ack the client now; the durable upload to object storage
         // and the advisory-lock release run in the background. The lock is held
         // until the upload finishes, so a concurrent writer on another machine
@@ -972,17 +978,29 @@ pub async fn git_receive_pack(
         // The intent marker must be on disk BEFORE the ack: the spawned task
         // may never be polled if the process stops right after the response,
         // and without the marker a restart would treat the stale storage
-        // archive as newer and roll the acked push back.
-        guard.mark_pending().await;
-        let repo_label = name.to_string();
-        tokio::spawn(async move {
-            if let Err(e) = guard.release(true).await {
-                tracing::error!(repo = %repo_label, err = %e,
-                    "write-back durable upload failed after push was acked");
+        // archive as newer and roll the acked push back. If the marker itself
+        // cannot be persisted, do NOT ack early — fall back to the strict
+        // upload-before-ack path below.
+        match guard.mark_pending().await {
+            Ok(()) => {
+                let repo_label = name.to_string();
+                tokio::spawn(async move {
+                    if let Err(e) = guard.release(true).await {
+                        tracing::error!(repo = %repo_label, err = %e,
+                            "write-back durable upload failed after push was acked");
+                    }
+                });
             }
-        });
-    } else {
-        // Strict path (or failed push): upload-before-ack.
+            Err(e) => {
+                tracing::warn!(repo = %name, err = %e,
+                    "pending-upload marker write failed — falling back to strict upload-before-ack");
+                strict_guard = Some(guard);
+            }
+        }
+    }
+    if let Some(guard) = strict_guard {
+        // Strict path (failed push, async_upload off, or marker write failure):
+        // upload-before-ack.
         if let Err(e) = guard.release(push_ok).await {
             if push_ok {
                 // A successful push whose durable upload then failed — the
@@ -1622,11 +1640,13 @@ pub async fn fork_repo(
     let disk_path = store::repo_disk_path(&state.config.repos_dir, &forker_did, &fork_name);
 
     // Claim-first ordering: insert the DB row before cloning or uploading
-    // anything. The row is the claim on (owner, name) — a concurrent same-name
-    // fork loses at the insert with nothing on disk or in storage, so the
-    // failure compensation below only ever removes state THIS attempt created
-    // (its own row by id, its own clone dir) and never touches a storage
-    // archive that may belong to a racing winner's live repo.
+    // anything. Within one node's database the row is the claim on (owner,
+    // name) — a concurrent same-name fork loses at the insert with nothing on
+    // disk or in storage. Across nodes (per-node Postgres) the insert
+    // arbitrates nothing; the cross-node safety property is that the failure
+    // compensation below only ever removes state THIS attempt created (its own
+    // row by id, its own clone dir) and never touches a storage archive that
+    // may belong to a racing winner's live repo.
     let now = Utc::now();
     let record = crate::db::RepoRecord {
         id: Uuid::new_v4().to_string(),
@@ -1647,6 +1667,13 @@ pub async fn fork_repo(
     // our generated id) and whatever `git clone` left at `disk_path` — clone
     // can create the destination before failing, and an orphaned dir makes a
     // retry fail on an existing destination.
+    //
+    // KNOWN COMPOUND-FAILURE WINDOW: the row makes the fork pushable before
+    // the clone+upload complete, and this flow holds no advisory lock across
+    // that phase. A push acked in the window, whose own upload also failed,
+    // lives only in this dir — undoing the claim then discards it. Requires
+    // fork-upload failure + an interleaved push + that push's upload failure;
+    // the follow-up is to hold the advisory lock across clone+publication.
     let undo_claim = |state: AppState, record_id: String, disk_path: std::path::PathBuf| async move {
         match tokio::fs::remove_dir_all(&disk_path).await {
             Ok(()) => {}
@@ -1656,6 +1683,9 @@ pub async fn fork_repo(
                 "failed to clean up local fork dir after fork error"
             ),
         }
+        // Also drop the pending-upload marker a failed upload just wrote: left
+        // behind, it would read as divergence and wedge a same-name recreate.
+        crate::git::repo_store::clear_pending_upload(&disk_path);
         if let Err(e) = state.db.delete_repo_by_id(&record_id).await {
             tracing::warn!(record_id = %record_id, err = %e,
                 "failed to remove fork row after fork error");

crates/gitlawb-node/src/config.rs

@@ -174,8 +174,14 @@ pub struct Config {
     /// Max connections in the dedicated advisory-lock pool. A push pins one
     /// connection for its whole lifetime (lock held across receive-pack and
     /// the storage upload), so this bounds per-node push concurrency. Counts
-    /// against Postgres `max_connections` on top of the handler pool.
-    #[arg(long, env = "GITLAWB_ADVISORY_LOCK_POOL_SIZE", default_value_t = 16)]
+    /// against Postgres `max_connections` on top of the handler pool. Must be
+    /// at least 1 — zero would let the node start but time out every write.
+    #[arg(
+        long,
+        env = "GITLAWB_ADVISORY_LOCK_POOL_SIZE",
+        default_value_t = 16,
+        value_parser = clap::value_parser!(u32).range(1..)
+    )]
     pub advisory_lock_pool_size: u32,
 
     /// Maximum pack body size for git-receive-pack and git-upload-pack, in bytes.
@@ -316,4 +322,21 @@ mod tests {
             Config::try_parse_from(["gitlawb-node", "--git-service-timeout-secs", "0"]).is_err()
         );
     }
+
+    #[test]
+    fn advisory_lock_pool_size_defaults_to_16_and_rejects_zero() {
+        assert_eq!(
+            Config::parse_from(["gitlawb-node"]).advisory_lock_pool_size,
+            16
+        );
+        assert_eq!(
+            Config::parse_from(["gitlawb-node", "--advisory-lock-pool-size", "1"])
+                .advisory_lock_pool_size,
+            1
+        );
+        // 0 would let the node start but time out every write on pool acquire.
+        assert!(
+            Config::try_parse_from(["gitlawb-node", "--advisory-lock-pool-size", "0"]).is_err()
+        );
+    }
 }

crates/gitlawb-node/src/git/repo_store.rs

@@ -94,7 +94,21 @@ impl RepoStore {
         };
 
         let marker = pending_upload_marker(local_path);
-        if marker.exists() {
+        // try_exists, not exists(): a transient EACCES/EIO must not read as
+        // "no marker" — that path downloads and can roll back a pending local
+        // write. Fail the write path closed; treat as present on the read path.
+        let marker_present = match marker.try_exists() {
+            Ok(present) => present,
+            Err(e) => {
+                if require_fresh {
+                    return Err(e).context("probing pending-upload marker");
+                }
+                warn!(repo = %repo_name, err = %e,
+                    "pending-upload marker probe failed — treating as present");
+                true
+            }
+        };
+        if marker_present {
             if local_path.exists() {
                 // The local copy has a write that storage never received (its
                 // upload failed, or the node stopped first). The marker records
@@ -172,6 +186,12 @@ impl RepoStore {
             }
         }
 
+        // KNOWN LIMITATION (pre-dates this layer): read-path downloads and
+        // their swap-into-place are not serialized against the advisory write
+        // lock, so a slow in-flight download decided before a push began can
+        // swap a stale tree under a running receive-pack on the same node.
+        // Requires a cache-miss/stale read racing a same-repo write; the
+        // follow-up is to serialize download+swap with writers.
         match archive.download(owner_slug, repo_name, local_path).await {
             Ok(()) => {
                 self.versions.lock().await.insert(key, remote_etag);
@@ -207,7 +227,14 @@ impl RepoStore {
             if self.archive.is_some() {
                 let key = format!("{owner_slug}/{repo_name}");
                 let already_migrated = self.migrated.lock().await.contains(&key);
-                if !already_migrated {
+                // A pending-upload marker means the marker machinery already
+                // owns this repo's next upload (next write, or the startup
+                // retry). Migration must not steal it: `upload_under_lock`
+                // knows nothing about markers, so its upload would strand the
+                // marker with a base that no longer matches storage, wedging
+                // the repo's writes on a spurious divergence.
+                let marker_pending = pending_upload_marker(&local_path).exists();
+                if !already_migrated && !marker_pending {
                     let this = self.clone();
                     let slug = owner_slug.clone();
                     let name = repo_name.to_string();
@@ -311,6 +338,11 @@ impl RepoStore {
             let _ = std::fs::remove_dir_all(&local_path);
             return Err(e).context("initializing bare repo");
         }
+        // A marker left by a previous same-name repo (failed creation, deleted
+        // repo) describes THAT repo's history, not this fresh one — once this
+        // repo's archive exists, a stale empty-base marker would read as
+        // divergence and wedge its writes.
+        clear_pending_upload(&local_path);
 
         // Upload the new repo synchronously under the advisory lock: a background
         // upload could land the empty repo *after* a racing first push and clobber
@@ -336,25 +368,45 @@ impl RepoStore {
     /// Call this after any operation that modifies the git repo on disk. Returns
     /// `Err` if the durable upload fails so the caller can surface it rather than
     /// acking a write that never reached storage.
+    ///
+    /// The upload runs under the per-repo advisory lock: claim-first creation
+    /// makes a fork addressable (and pushable) before this upload finishes, so
+    /// an unlocked PUT could compress a pre-push snapshot and land it AFTER a
+    /// concurrent locked push's upload — which that push's cleared marker no
+    /// longer protects against.
     pub async fn release_after_write(&self, owner_did: &str, repo_name: &str) -> Result<()> {
-        let Some(ref archive) = self.archive else {
+        if self.archive.is_none() {
             return Ok(());
-        };
+        }
         let (owner_slug, local_path) = self
             .local_path(owner_did, repo_name)
             .context("rejected unsafe path in release_after_write")?;
         let key = format!("{owner_slug}/{repo_name}");
         let base = self.versions.lock().await.get(&key).cloned();
-        mark_pending_upload(&local_path, repo_name, base.as_deref());
-        match archive.upload(&owner_slug, repo_name, &local_path).await {
-            Ok(Some(etag)) => {
-                self.versions.lock().await.insert(key, etag);
-                clear_pending_upload(&local_path);
-                Ok(())
-            }
-            Ok(None) => {
-                clear_pending_upload(&local_path);
-                Ok(())
+        mark_pending_upload(&local_path, base.as_deref())
+            .context("persisting pending-upload marker")?;
+        // The marker's recorded base is authoritative, not the cache: if a
+        // marker already existed (earlier failed upload), mark_pending_upload
+        // preserved its ORIGINAL base while the versions cache was invalidated
+        // by that failure — or emptied entirely by a restart. Comparing
+        // against the cache value would misread that state as divergence.
+        let base = std::fs::read_to_string(pending_upload_marker(&local_path))
+            .map(|s| s.trim().to_string())
+            .unwrap_or_else(|_| base.unwrap_or_default());
+        match self
+            .upload_locked_with_marker(&owner_slug, repo_name, &local_path, &base)
+            .await
+        {
+            Ok(PendingUploadOutcome::Uploaded) => Ok(()),
+            Ok(PendingUploadOutcome::Diverged) => {
+                // Another writer advanced storage between our sync and this
+                // upload. Refusing (rather than uploading) protects their
+                // acked write; ours stays local behind the marker.
+                self.versions.lock().await.remove(&key);
+                anyhow::bail!(
+                    "storage for {key} advanced during the write — upload aborted to \
+                     avoid clobbering the concurrent writer; local copy left marked"
+                )
             }
             Err(e) => {
                 // Storage is now behind local. Drop the cached etag, and leave
@@ -367,6 +419,148 @@ impl RepoStore {
         }
     }
 
+    /// Startup sweep re-attempting the durable upload for every repo whose
+    /// pending-upload marker survived a crash or a failed upload. Without this,
+    /// a repo that receives no further writes stays divergent from storage
+    /// indefinitely, visible only as one log line at failure time.
+    ///
+    /// Applies the same base-etag rule as `sync_down_if_stale`: a repo whose
+    /// storage advanced past the marker's base is left marked (its writes stay
+    /// wedged pending manual reconciliation) and only logged. Returns
+    /// `(reuploaded, still_pending)`.
+    pub async fn retry_pending_uploads(&self) -> (usize, usize) {
+        if self.archive.is_none() {
+            return (0, 0);
+        }
+        let mut reuploaded = 0usize;
+        let mut still_pending = 0usize;
+
+        let mut markers: Vec<(String, String, PathBuf)> = Vec::new(); // (slug, repo, local)
+        let Ok(owners) = std::fs::read_dir(&self.repos_dir) else {
+            return (0, 0);
+        };
+        for owner in owners.flatten() {
+            if !owner.path().is_dir() {
+                continue;
+            }
+            let slug = owner.file_name().to_string_lossy().into_owned();
+            let Ok(entries) = std::fs::read_dir(owner.path()) else {
+                continue;
+            };
+            for entry in entries.flatten() {
+                let name = entry.file_name().to_string_lossy().into_owned();
+                // Marker layout: `.{repo}.git.pending-upload`
+                let Some(repo_dir) = name
+                    .strip_prefix('.')
+                    .and_then(|n| n.strip_suffix(".pending-upload"))
+                else {
+                    continue;
+                };
+                let Some(repo_name) = repo_dir.strip_suffix(".git") else {
+                    continue;
+                };
+                markers.push((
+                    slug.clone(),
+                    repo_name.to_string(),
+                    owner.path().join(repo_dir),
+                ));
+            }
+        }
+
+        for (slug, repo_name, local_path) in markers {
+            if !local_path.exists() {
+                // Stale litter (repo dir gone) — storage is the best remaining
+                // state; drop the marker.
+                clear_pending_upload(&local_path);
+                continue;
+            }
+            let base =
+                std::fs::read_to_string(pending_upload_marker(&local_path)).unwrap_or_default();
+            let base = base.trim().to_string();
+            // The base-vs-remote decision and the upload both happen inside
+            // `upload_locked_with_marker`, UNDER the advisory lock: an
+            // unlocked pre-check here could pass, then block on a concurrent
+            // push's lock for that push's whole duration, and the stale
+            // decision would clobber the push's freshly-uploaded archive.
+            match self
+                .upload_locked_with_marker(&slug, &repo_name, &local_path, &base)
+                .await
+            {
+                Ok(PendingUploadOutcome::Uploaded) => {
+                    debug!(repo = %repo_name, "pending-upload retry: re-synced storage");
+                    reuploaded += 1;
+                }
+                Ok(PendingUploadOutcome::Diverged) => {
+                    warn!(repo = %repo_name,
+                        "pending-upload retry: storage diverged from marker base — \
+                         leaving marked; writes stay blocked pending manual reconciliation");
+                    still_pending += 1;
+                }
+                Err(e) => {
+                    warn!(repo = %repo_name, err = %e,
+                        "pending-upload retry: upload failed — will retry on next write");
+                    still_pending += 1;
+                }
+            }
+        }
+        crate::metrics::set_pending_upload_markers(still_pending as i64);
+        (reuploaded, still_pending)
+    }
+
+    /// Marker-protected upload: takes the per-repo advisory lock, re-checks
+    /// that storage still matches `base` UNDER the lock, and only then uploads,
+    /// updates the versions cache, and clears the marker — all before the lock
+    /// is released.
+    ///
+    /// Both halves of that ordering are load-bearing:
+    /// - The divergence check must run under the lock. An unlocked check can
+    ///   pass just before a concurrent locked push advances storage (the check
+    ///   then blocks on that push's lock), and blindly uploading afterwards
+    ///   would clobber the acked push it lost the race to.
+    /// - The marker must be cleared before the lock is released, or a writer
+    ///   queued on the lock could observe marker + fresh etag and fail with a
+    ///   spurious "diverged — reconcile manually" on a consistent repo.
+    async fn upload_locked_with_marker(
+        &self,
+        owner_slug: &str,
+        repo_name: &str,
+        local_path: &Path,
+        base: &str,
+    ) -> Result<PendingUploadOutcome> {
+        let Some(ref archive) = self.archive else {
+            anyhow::bail!("upload_locked_with_marker called without a storage backend");
+        };
+        let lock_key = advisory_lock_key(owner_slug, repo_name);
+        let label = format!("{owner_slug}/{repo_name}");
+        let lock = LockedConn::acquire(&self.lock_pool, lock_key, &label).await?;
+
+        let outcome: Result<PendingUploadOutcome> = async {
+            let remote = archive
+                .head_etag(owner_slug, repo_name)
+                .await
+                .context("storage head under lock before pending upload")?;
+            if remote.as_deref().unwrap_or("") != base {
+                return Ok(PendingUploadOutcome::Diverged);
+            }
+            let etag = archive
+                .upload(owner_slug, repo_name, local_path)
+                .await
+                .context("uploading repo to storage under lock")?;
+            if let Some(ref etag) = etag {
+                self.versions
+                    .lock()
+                    .await
+                    .insert(label.clone(), etag.clone());
+            }
+            clear_pending_upload_after_success(local_path, etag.as_deref());
+            Ok(PendingUploadOutcome::Uploaded)
+        }
+        .await;
+
+        lock.unlock().await;
+        outcome
+    }
+
     /// Upload `local_path` to storage while holding the per-repo advisory lock,
     /// so a background or init-time upload can't clobber a concurrent locked
     /// write by landing an older snapshot after it. With `skip_if_exists`, skips
@@ -665,15 +859,21 @@ impl RepoWriteGuard {
     }
 
     /// Durably record intent-to-upload NOW, before the caller acks the client.
-    /// Write-back callers must call this before spawning `release()`: the
+    /// Write-back callers must call this before spawning `release()` — the
     /// spawned task may never be polled if the process stops right after the
     /// ack, and without the marker already on disk a restart would treat the
-    /// stale storage archive as newer and roll the acked write back.
-    /// Idempotent with the marker `release()` writes itself.
-    pub async fn mark_pending(&self) {
+    /// stale storage archive as newer and roll the acked write back. On `Err`
+    /// the caller must NOT ack early; fall back to strict upload-before-ack.
+    /// Idempotent with the marker `release()` writes itself. No-op without a
+    /// storage backend (markers would be inert until a backend appears, then
+    /// wedge repos whose archives predate them).
+    pub async fn mark_pending(&self) -> Result<()> {
+        if self.archive.is_none() {
+            return Ok(());
+        }
         let key = format!("{}/{}", self.owner_slug, self.repo_name);
         let base = self.versions.lock().await.get(&key).cloned();
-        mark_pending_upload(&self.local_path, &self.repo_name, base.as_deref());
+        mark_pending_upload(&self.local_path, base.as_deref())
     }
 
     /// Upload to storage (only when the write succeeded) and release the advisory
@@ -696,14 +896,20 @@ impl RepoWriteGuard {
         let upload_result: Result<()> = if success {
             if let Some(ref archive) = self.archive {
                 let base = self.versions.lock().await.get(&key).cloned();
-                mark_pending_upload(&self.local_path, &self.repo_name, base.as_deref());
+                if let Err(e) = mark_pending_upload(&self.local_path, base.as_deref()) {
+                    // Proceed with the upload anyway: if it succeeds, no marker
+                    // is needed; if both fail, the error below reaches the
+                    // caller (double-failure corner, same exposure as
+                    // pre-marker behavior).
+                    warn!(repo = %self.repo_name, err = %e, "failed to write pending-upload marker");
+                }
                 match archive
                     .upload(&self.owner_slug, &self.repo_name, &self.local_path)
                     .await
                 {
                     Ok(Some(etag)) => {
-                        self.versions.lock().await.insert(key.clone(), etag);
-                        clear_pending_upload(&self.local_path);
+                        self.versions.lock().await.insert(key.clone(), etag.clone());
+                        clear_pending_upload_after_success(&self.local_path, Some(&etag));
                         Ok(())
                     }
                     Ok(None) => {
@@ -757,33 +963,86 @@ fn pending_upload_marker(local_path: &Path) -> PathBuf {
     local_path.with_file_name(format!(".{name}.pending-upload"))
 }
 
+/// Persist the intent-to-upload marker. Fallible (write-back callers must NOT
+/// ack the client if this fails) and atomic (tmp + rename, so a crash cannot
+/// leave a torn marker).
+///
 /// `base_etag` is the storage etag the local write was built on (empty when
 /// storage held nothing). `sync_down_if_stale` compares it against the current
 /// remote etag to distinguish "local strictly ahead" from cross-node
 /// divergence.
-fn mark_pending_upload(local_path: &Path, repo_name: &str, base_etag: Option<&str>) {
-    if let Err(e) = std::fs::write(
-        pending_upload_marker(local_path),
-        base_etag.unwrap_or_default(),
-    ) {
-        // Best-effort: without the marker we fall back to the pre-marker
-        // behavior (a failed upload risks rollback), but the upload itself
-        // must still proceed.
-        warn!(repo = %repo_name, err = %e, "failed to write pending-upload marker");
+///
+/// An existing marker is preserved untouched: its base is the last storage
+/// etag this node confirmed, which stays correct for every further write
+/// stacked on the same undiverged local copy. Re-marking would record the
+/// CURRENT cache — emptied by the preceding upload failure — and a corrupted
+/// (empty) base makes the next sync read unchanged storage as divergence,
+/// wedging the repo's whole write surface after two consecutive upload
+/// failures.
+fn mark_pending_upload(local_path: &Path, base_etag: Option<&str>) -> Result<()> {
+    let marker = pending_upload_marker(local_path);
+    match marker.try_exists() {
+        Ok(true) => return Ok(()), // keep the original base
+        Ok(false) => {}
+        Err(e) => return Err(e).context("probing pending-upload marker"),
     }
+    let tmp = marker.with_file_name(format!(".pending-upload.tmp-{}", uuid::Uuid::new_v4()));
+    std::fs::write(&tmp, base_etag.unwrap_or_default()).context("writing pending-upload marker")?;
+    std::fs::rename(&tmp, &marker)
+        .inspect_err(|_| {
+            let _ = std::fs::remove_file(&tmp);
+        })
+        .context("publishing pending-upload marker")
 }
 
-fn clear_pending_upload(local_path: &Path) {
+pub(crate) fn clear_pending_upload(local_path: &Path) {
     let _ = std::fs::remove_file(pending_upload_marker(local_path));
 }
 
+/// Remove the marker after a *successful* upload, first atomically rewriting
+/// its base to the just-uploaded etag. A crash between the rewrite and the
+/// unlink then reads as "local ahead, base matches" — which self-heals on the
+/// next write or startup retry — instead of "base predates storage", which
+/// would wedge the repo behind a spurious permanent divergence even though
+/// local and storage are identical.
+fn clear_pending_upload_after_success(local_path: &Path, new_etag: Option<&str>) {
+    let marker = pending_upload_marker(local_path);
+    if let Some(etag) = new_etag {
+        if marker.exists() {
+            let tmp =
+                marker.with_file_name(format!(".pending-upload.tmp-{}", uuid::Uuid::new_v4()));
+            if std::fs::write(&tmp, etag).is_ok() {
+                let _ = std::fs::rename(&tmp, &marker);
+            } else {
+                let _ = std::fs::remove_file(&tmp);
+            }
+        }
+    }
+    let _ = std::fs::remove_file(&marker);
+}
+
+/// Outcome of a marker-protected upload attempt.
+enum PendingUploadOutcome {
+    /// Uploaded, versions cache updated, marker cleared — all under the lock.
+    Uploaded,
+    /// Storage no longer matches the marker's base: another writer advanced it.
+    /// Nothing was uploaded and the marker was left in place.
+    Diverged,
+}
+
 /// Compute a stable i64 hash for a Postgres advisory lock key.
+///
+/// SHA-256 prefix, NOT `DefaultHasher`: the default hasher's algorithm is
+/// explicitly unspecified across Rust releases, and this lock is the sole
+/// cross-machine write serializer — two machines built with different
+/// toolchains hashing the same repo to different keys would silently stop
+/// excluding each other mid rolling deploy. (Changing the scheme is itself a
+/// one-deploy exclusion gap between old and new binaries; accepted once,
+/// here, to get onto a stable function.)
 fn advisory_lock_key(owner_slug: &str, repo_name: &str) -> i64 {
-    use std::hash::{Hash, Hasher};
-    let mut hasher = std::collections::hash_map::DefaultHasher::new();
-    owner_slug.hash(&mut hasher);
-    repo_name.hash(&mut hasher);
-    hasher.finish() as i64
+    use sha2::{Digest, Sha256};
+    let digest = Sha256::digest(format!("{owner_slug}/{repo_name}").as_bytes());
+    i64::from_be_bytes(digest[..8].try_into().expect("digest has at least 8 bytes"))
 }
 
 #[cfg(test)]
@@ -1000,11 +1259,18 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn pending_marker_prevents_rollback_and_clears_on_next_upload() {
+    // Needs a real pool: `release_after_write` uploads under the advisory lock.
+    #[sqlx::test]
+    async fn pending_marker_prevents_rollback_and_clears_on_next_upload(pool: PgPool) {
         let store_root = tempfile::tempdir().unwrap();
         let repos_dir = tempfile::tempdir().unwrap();
-        let store = store_with_fs_archive(repos_dir.path().to_path_buf(), store_root.path());
+        let blob: Arc<dyn crate::storage::BlobStore> =
+            Arc::new(crate::storage::fs::FsBlobStore::new(store_root.path()).unwrap());
+        let store = RepoStore::new(
+            repos_dir.path().to_path_buf(),
+            Some(crate::storage::archive::RepoArchive::new(blob)),
+            pool,
+        );
 
         // Storage holds v1; local downloads it.
         let seed = tempfile::tempdir().unwrap();
@@ -1034,7 +1300,7 @@ mod tests {
             .unwrap()
             .unwrap();
         std::fs::write(local.join("HEAD"), b"ACKED-WRITE\n").unwrap();
-        mark_pending_upload(&local, "repo", Some(&base));
+        mark_pending_upload(&local, Some(&base)).unwrap();
         store.versions.lock().await.clear();
 
         // Both the read and the write path must serve local, not roll it back.
@@ -1090,7 +1356,7 @@ mod tests {
             .unwrap();
         let base = archive.head_etag("owner", "repo").await.unwrap().unwrap();
         std::fs::write(local.join("HEAD"), b"LOCAL-AHEAD\n").unwrap();
-        mark_pending_upload(&local, "repo", Some(&base));
+        mark_pending_upload(&local, Some(&base)).unwrap();
         store.versions.lock().await.clear();
 
         // Another node advances storage past our base.
@@ -1135,7 +1401,7 @@ mod tests {
         // the marker is stale — drop it and download normally.
         let local = repos_dir.path().join("owner").join("repo.git");
         std::fs::create_dir_all(local.parent().unwrap()).unwrap();
-        mark_pending_upload(&local, "repo", Some("whatever"));
+        mark_pending_upload(&local, Some("whatever")).unwrap();
         store
             .sync_down_if_stale("owner", "repo", &local, true)
             .await
@@ -1197,4 +1463,291 @@ mod tests {
             .expect("require_fresh=false must fall back to the local copy");
         assert_eq!(std::fs::read(local.join("HEAD")).unwrap(), b"v1\n");
     }
+
+    // ── failing-store double: exercises error branches no real backend can ──
+
+    /// BlobStore wrapper whose `put`/`head` can be flipped to fail, unlocking
+    /// deterministic coverage of the upload-failure and head-failure branches.
+    struct FlakyStore {
+        inner: crate::storage::fs::FsBlobStore,
+        fail_put: std::sync::atomic::AtomicBool,
+        fail_head: std::sync::atomic::AtomicBool,
+    }
+
+    impl FlakyStore {
+        fn new(root: &Path) -> Arc<Self> {
+            Arc::new(Self {
+                inner: crate::storage::fs::FsBlobStore::new(root).unwrap(),
+                fail_put: std::sync::atomic::AtomicBool::new(false),
+                fail_head: std::sync::atomic::AtomicBool::new(false),
+            })
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage::BlobStore for FlakyStore {
+        fn backend_name(&self) -> &'static str {
+            "flaky"
+        }
+        async fn get(&self, key: &str) -> Result<Option<bytes::Bytes>> {
+            self.inner.get(key).await
+        }
+        async fn put(&self, key: &str, body: bytes::Bytes) -> Result<crate::storage::ObjectMeta> {
+            if self.fail_put.load(std::sync::atomic::Ordering::Relaxed) {
+                anyhow::bail!("injected put failure");
+            }
+            self.inner.put(key, body).await
+        }
+        async fn head(&self, key: &str) -> Result<Option<crate::storage::ObjectMeta>> {
+            if self.fail_head.load(std::sync::atomic::Ordering::Relaxed) {
+                anyhow::bail!("injected head failure");
+            }
+            self.inner.head(key).await
+        }
+        async fn delete(&self, key: &str) -> Result<()> {
+            self.inner.delete(key).await
+        }
+    }
+
+    fn store_with_flaky(repos_dir: PathBuf, flaky: Arc<FlakyStore>, pool: PgPool) -> RepoStore {
+        let blob: Arc<dyn crate::storage::BlobStore> = flaky;
+        let archive = crate::storage::archive::RepoArchive::new(blob);
+        RepoStore::new(repos_dir, Some(archive), pool)
+    }
+
+    /// beardthelion P1 regression: two consecutive failed uploads must not
+    /// corrupt the marker's base — the second `release` re-marks while the
+    /// versions cache is empty, and overwriting the base with "" would make
+    /// the third write read unchanged storage as divergence and wedge the
+    /// repo's entire write surface.
+    #[sqlx::test]
+    async fn two_consecutive_failed_uploads_preserve_marker_base(pool: PgPool) {
+        let store_root = tempfile::tempdir().unwrap();
+        let repos_dir = tempfile::tempdir().unwrap();
+        let flaky = FlakyStore::new(store_root.path());
+        let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool);
+
+        // Storage v1, synced down.
+        let seed = tempfile::tempdir().unwrap();
+        std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap();
+        store
+            .archive
+            .as_ref()
+            .unwrap()
+            .upload("owner", "repo", seed.path())
+            .await
+            .unwrap();
+        let local = repos_dir.path().join("owner").join("repo.git");
+        store
+            .sync_down_if_stale("owner", "repo", &local, true)
+            .await
+            .unwrap();
+
+        // Write 1: mutate, upload fails.
+        flaky
+            .fail_put
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+        std::fs::write(local.join("HEAD"), b"write-1\n").unwrap();
+        let guard = store.acquire_write("owner", "repo").await.unwrap();
+        assert!(guard.release(true).await.is_err(), "injected put must fail");
+        let base_after_first = std::fs::read_to_string(pending_upload_marker(&local)).unwrap();
+        assert!(!base_after_first.trim().is_empty(), "base must be recorded");
+
+        // Write 2: acquire must succeed (storage unchanged == local-ahead),
+        // and the second failed release must NOT re-mark with an empty base.
+        let guard = store.acquire_write("owner", "repo").await.unwrap();
+        std::fs::write(local.join("HEAD"), b"write-2\n").unwrap();
+        assert!(guard.release(true).await.is_err());
+        assert_eq!(
+            std::fs::read_to_string(pending_upload_marker(&local)).unwrap(),
+            base_after_first,
+            "an existing marker's base must be preserved on re-mark"
+        );
+
+        // Write 3: still not wedged — and once the store heals, everything
+        // re-syncs and the marker clears.
+        let guard = store
+            .acquire_write("owner", "repo")
+            .await
+            .expect("repeated upload failures must not wedge the write surface");
+        flaky
+            .fail_put
+            .store(false, std::sync::atomic::Ordering::Relaxed);
+        guard.release(true).await.unwrap();
+        assert!(!pending_upload_marker(&local).exists());
+    }
+
+    /// jatmn P1 regression: the write-back ack window. `mark_pending` runs
+    /// before the ack; if the process dies before the spawned release is ever
+    /// polled (simulated by dropping the guard), the marker alone must keep
+    /// the next sync from rolling the acked write back.
+    #[sqlx::test]
+    async fn mark_pending_alone_protects_the_ack_window(pool: PgPool) {
+        let store_root = tempfile::tempdir().unwrap();
+        let repos_dir = tempfile::tempdir().unwrap();
+        let flaky = FlakyStore::new(store_root.path());
+        let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool);
+
+        let seed = tempfile::tempdir().unwrap();
+        std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap();
+        store
+            .archive
+            .as_ref()
+            .unwrap()
+            .upload("owner", "repo", seed.path())
+            .await
+            .unwrap();
+        let local = repos_dir.path().join("owner").join("repo.git");
+
+        let guard = store.acquire_write("owner", "repo").await.unwrap();
+        std::fs::write(local.join("HEAD"), b"ACKED\n").unwrap();
+        guard.mark_pending().await.unwrap();
+        drop(guard); // crash before release() is ever polled
+
+        store
+            .sync_down_if_stale("owner", "repo", &local, true)
+            .await
+            .unwrap();
+        assert_eq!(
+            std::fs::read(local.join("HEAD")).unwrap(),
+            b"ACKED\n",
+            "the pre-ack marker alone must prevent rollback"
+        );
+    }
+
+    /// The lazy-migration existence check must propagate failure instead of
+    /// reading it as "absent" and uploading over a possibly-newer archive.
+    #[sqlx::test]
+    async fn upload_under_lock_propagates_failed_existence_check(pool: PgPool) {
+        let store_root = tempfile::tempdir().unwrap();
+        let repos_dir = tempfile::tempdir().unwrap();
+        let flaky = FlakyStore::new(store_root.path());
+        let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool);
+
+        let local = repos_dir.path().join("owner").join("repo.git");
+        std::fs::create_dir_all(&local).unwrap();
+        std::fs::write(local.join("HEAD"), b"local\n").unwrap();
+
+        flaky
+            .fail_head
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+        assert!(
+            store
+                .upload_under_lock("owner", "repo", &local, true)
+                .await
+                .is_err(),
+            "a failed existence check must not read as absent"
+        );
+        assert!(
+            store
+                .archive
+                .as_ref()
+                .unwrap()
+                .head_etag("owner", "repo")
+                .await
+                .is_err(),
+            "sanity: head still failing"
+        );
+    }
+
+    /// init() must remove its local dir when the initial upload fails, so a
+    /// retry of the same name doesn't hit an existing destination.
+    #[sqlx::test]
+    async fn init_removes_local_dir_when_upload_fails(pool: PgPool) {
+        let store_root = tempfile::tempdir().unwrap();
+        let repos_dir = tempfile::tempdir().unwrap();
+        let flaky = FlakyStore::new(store_root.path());
+        let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool);
+
+        flaky
+            .fail_put
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+        assert!(store.init("did:key:z6MkOwner", "newrepo").await.is_err());
+        let local = repos_dir
+            .path()
+            .join("did_key_z6MkOwner")
+            .join("newrepo.git");
+        assert!(
+            !local.exists(),
+            "failed init must not leave a local dir behind"
+        );
+    }
+
+    /// Marker + head failure: the write path fails closed, the read path
+    /// serves the local copy.
+    #[tokio::test]
+    async fn marker_with_failing_head_fails_write_closed_serves_read() {
+        let store_root = tempfile::tempdir().unwrap();
+        let repos_dir = tempfile::tempdir().unwrap();
+        let flaky = FlakyStore::new(store_root.path());
+        // sync_down never touches the pool — lazy is fine here.
+        let pool = sqlx::PgPool::connect_lazy("postgres://invalid").unwrap();
+        let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool);
+
+        let local = repos_dir.path().join("owner").join("repo.git");
+        std::fs::create_dir_all(&local).unwrap();
+        std::fs::write(local.join("HEAD"), b"pending\n").unwrap();
+        mark_pending_upload(&local, Some("base-etag")).unwrap();
+
+        flaky
+            .fail_head
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+        assert!(
+            store
+                .sync_down_if_stale("owner", "repo", &local, true)
+                .await
+                .is_err(),
+            "write path must fail closed when freshness is unknowable"
+        );
+        store
+            .sync_down_if_stale("owner", "repo", &local, false)
+            .await
+            .expect("read path serves the local copy");
+        assert_eq!(std::fs::read(local.join("HEAD")).unwrap(), b"pending\n");
+    }
+
+    /// Startup sweep: re-uploads marked repos whose storage didn't move, and
+    /// leaves diverged ones marked.
+    #[sqlx::test]
+    async fn retry_pending_uploads_heals_and_respects_divergence(pool: PgPool) {
+        let store_root = tempfile::tempdir().unwrap();
+        let repos_dir = tempfile::tempdir().unwrap();
+        let flaky = FlakyStore::new(store_root.path());
+        let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool);
+        let archive = store.archive.as_ref().unwrap();
+
+        // Repo A: storage v1, local ahead with matching base — heals.
+        let seed = tempfile::tempdir().unwrap();
+        std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap();
+        archive.upload("owner", "heals", seed.path()).await.unwrap();
+        let base_a = archive.head_etag("owner", "heals").await.unwrap().unwrap();
+        let local_a = repos_dir.path().join("owner").join("heals.git");
+        std::fs::create_dir_all(&local_a).unwrap();
+        std::fs::write(local_a.join("HEAD"), b"local-ahead\n").unwrap();
+        mark_pending_upload(&local_a, Some(&base_a)).unwrap();
+
+        // Repo B: marker base predates current storage — stays marked.
+        archive
+            .upload("owner", "diverged", seed.path())
+            .await
+            .unwrap();
+        let local_b = repos_dir.path().join("owner").join("diverged.git");
+        std::fs::create_dir_all(&local_b).unwrap();
+        std::fs::write(local_b.join("HEAD"), b"local-b\n").unwrap();
+        mark_pending_upload(&local_b, Some("stale-base")).unwrap();
+
+        let (reuploaded, still_pending) = store.retry_pending_uploads().await;
+        assert_eq!((reuploaded, still_pending), (1, 1));
+        assert!(!pending_upload_marker(&local_a).exists());
+        assert!(pending_upload_marker(&local_b).exists());
+
+        // A's local content is now durably in storage.
+        let out = tempfile::tempdir().unwrap();
+        let restored = out.path().join("restored.git");
+        archive.download("owner", "heals", &restored).await.unwrap();
+        assert_eq!(
+            std::fs::read(restored.join("HEAD")).unwrap(),
+            b"local-ahead\n"
+        );
+    }
 }

crates/gitlawb-node/src/main.rs

@@ -303,6 +303,20 @@ async fn main() -> Result<()> {
 
     let repo_store = git::repo_store::RepoStore::new(config.repos_dir.clone(), archive, lock_pool);
 
+    // Re-attempt uploads for repos whose pending-upload marker survived a
+    // crash or failed upload — otherwise a repo with no further writes stays
+    // divergent from storage indefinitely. Background task: it takes the
+    // per-repo advisory locks, so it serializes correctly with live pushes.
+    {
+        let store = repo_store.clone();
+        tokio::spawn(async move {
+            let (reuploaded, still_pending) = store.retry_pending_uploads().await;
+            if reuploaded > 0 || still_pending > 0 {
+                info!(reuploaded, still_pending, "pending-upload marker sweep");
+            }
+        });
+    }
+
     // Per-DID limiter for the creation endpoints. Keyed on the authenticated
     // DID (attacker-varied), so bound its key set to cap memory.
     let rate_limiter =

crates/gitlawb-node/src/metrics.rs

@@ -15,6 +15,8 @@
 //!     `gitlawb_pack_size_bytes`
 //!   * a single `gitlawb_info{version, did}` gauge = 1, for joins/dashboards
 //!   * currently-connected peer count — `gitlawb_peers_connected`
+//!   * repos whose local copy is ahead of durable storage —
+//!     `gitlawb_pending_upload_markers`
 //!
 //! All metrics live in a single process-wide registry initialized by
 //! [`init`]. Increment helpers (`record_push`, `record_auth_failure`, ...)
@@ -51,6 +53,7 @@ static SYNC_PROCESSED: OnceLock<IntCounterVec> = OnceLock::new();
 static WEBHOOK_DELIVERIES: OnceLock<IntCounterVec> = OnceLock::new();
 static PACK_SIZE: OnceLock<Histogram> = OnceLock::new();
 static PEERS_CONNECTED: OnceLock<IntGauge> = OnceLock::new();
+static PENDING_UPLOAD_MARKERS: OnceLock<IntGauge> = OnceLock::new();
 
 /// One-time initializer. Builds the registry, registers every metric,
 /// and sets the constant `gitlawb_info` gauge. Idempotent — calling
@@ -197,6 +200,18 @@ pub fn init(version: &str, node_did: &str) {
         .set(peers_connected)
         .expect("set PEERS_CONNECTED once");
 
+    let pending_markers = IntGauge::with_opts(Opts::new(
+        "gitlawb_pending_upload_markers",
+        "Repos whose local copy is ahead of durable storage (pending-upload markers outstanding)",
+    ))
+    .expect("gitlawb_pending_upload_markers definition");
+    registry
+        .register(Box::new(pending_markers.clone()))
+        .expect("register gitlawb_pending_upload_markers");
+    PENDING_UPLOAD_MARKERS
+        .set(pending_markers)
+        .expect("set PENDING_UPLOAD_MARKERS once");
+
     REGISTRY
         .set(registry)
         .expect("set REGISTRY once (init must be called exactly once)");
@@ -264,6 +279,14 @@ pub fn set_peers_connected(count: i64) {
     }
 }
 
+/// Update the outstanding pending-upload marker gauge (repos whose local copy
+/// is ahead of durable storage).
+pub fn set_pending_upload_markers(count: i64) {
+    if let Some(g) = PENDING_UPLOAD_MARKERS.get() {
+        g.set(count);
+    }
+}
+
 /// Encode the registry as the standard Prometheus text exposition format.
 /// Returns an error if `init` was never called.
 pub fn encode() -> Result<String, prometheus::Error> {

crates/gitlawb-node/src/storage/fs.rs

@@ -49,6 +49,23 @@ impl FsBlobStore {
         PathBuf::from(os)
     }
 
+    /// Per-final-path lock serializing the sidecar+blob publish step of `put`.
+    /// The two renames are only pairwise-consistent when same-key puts don't
+    /// interleave: unserialized, put A's content can land under put B's etag,
+    /// and "etag unchanged ⇒ content unchanged" breaks in the direction that
+    /// serves stale data as current. (Same never-evicted-map tradeoff as
+    /// `archive::publish_lock` — bounded by distinct keys per process life.)
+    fn put_publish_lock(path: &Path) -> std::sync::Arc<std::sync::Mutex<()>> {
+        use std::collections::HashMap;
+        use std::sync::{Arc, Mutex, OnceLock};
+        static LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> = OnceLock::new();
+        let locks = LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
+        let mut map = locks.lock().expect("put publish lock map poisoned");
+        map.entry(path.to_path_buf())
+            .or_insert_with(|| Arc::new(Mutex::new(())))
+            .clone()
+    }
+
     /// Fallback fingerprint for objects written before the sidecar existed.
     fn legacy_etag(md: &std::fs::Metadata) -> String {
         let mtime = md
@@ -92,20 +109,32 @@ impl BlobStore for FsBlobStore {
         // no synchronous fs call ever touches the async runtime.
         tokio::task::spawn_blocking(move || -> Result<ObjectMeta> {
             std::fs::create_dir_all(&parent).context("creating blob parent dir")?;
+            let etag = uuid::Uuid::new_v4().to_string();
+            let sidecar = Self::sidecar_of(&path2);
+            let sidecar_tmp = sidecar.with_extension(format!("{}.tmp-put", uuid::Uuid::new_v4()));
             let write_and_swap = (|| -> Result<()> {
                 std::fs::write(&tmp, &body).context("writing temp blob")?;
+                // Publish the new etag BEFORE the blob (each via its own
+                // tmp+rename): a crash between the two renames then yields
+                // new-etag/old-content — a redundant download, or at worst a
+                // loud-and-safe pending-marker divergence — instead of
+                // old-etag/new-content, which etag-equality consumers would
+                // silently serve as current while stale. The publish pair is
+                // serialized per key so concurrent same-key puts can't
+                // interleave one put's etag with another's content.
+                let publish = Self::put_publish_lock(&path2);
+                let _guard = publish.lock().expect("put publish lock poisoned");
+                std::fs::write(&sidecar_tmp, &etag).context("writing etag sidecar")?;
+                std::fs::rename(&sidecar_tmp, &sidecar).context("publishing etag sidecar")?;
                 std::fs::rename(&tmp, &path2).context("renaming blob into place")?;
                 Ok(())
             })();
             if let Err(e) = write_and_swap {
                 let _ = std::fs::remove_file(&tmp);
+                let _ = std::fs::remove_file(&sidecar_tmp);
                 return Err(e);
             }
             let md = std::fs::metadata(&path2).context("stat blob after write")?;
-            // Persist a fresh per-write etag; see `sidecar_of` for why
-            // size-mtime is not collision-resistant enough here.
-            let etag = uuid::Uuid::new_v4().to_string();
-            std::fs::write(Self::sidecar_of(&path2), &etag).context("writing etag sidecar")?;
             Ok(ObjectMeta {
                 size: md.len(),
                 etag: Some(etag),
@@ -250,9 +279,18 @@ mod tests {
         let m2 = store.put(key, body).await.unwrap();
         assert_ne!(m1.etag, m2.etag, "every put must produce a new etag");
 
-        // head() reports the latest persisted etag.
+        // head() reports the latest persisted etag, and it really is the
+        // sidecar's content — not a size-mtime fingerprint that would also
+        // pass the assert_ne above when consecutive writes get distinct
+        // mtimes.
         let h = store.head(key).await.unwrap().unwrap();
         assert_eq!(h.etag, m2.etag);
+        let sidecar = dir.path().join(format!("{key}.etag"));
+        assert_eq!(
+            std::fs::read_to_string(&sidecar).unwrap(),
+            m2.etag.clone().unwrap(),
+            "etag must come from the persisted sidecar"
+        );
 
         // delete() removes the sidecar with the object.
         store.delete(key).await.unwrap();