fix(storage): keep the marker gauge live; stripe the fs publish lock

CodeRabbit round on 5605b9c:

- gitlawb_pending_upload_markers is now delta-maintained by the marker
  lifecycle itself (+1 on create, -1 on remove), with the startup sweep
  seeding the absolute baseline — the gauge no longer drifts between
  sweeps.
- FsBlobStore's put publish lock uses a fixed 64-stripe array instead of
  a per-key map that grew for the life of the process.

Skipped with reason: restoring list() on the BlobStore trait — it was
removed deliberately at maintainer request until a production consumer
(GC/admin/migration) exists; the fs test helper covers the tests that
need it. Whole-object buffering remains the tracked streaming follow-up.

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

@@ -467,6 +467,11 @@ impl RepoStore {
             }
         }
 
+        // Seed the gauge with the surviving-marker count before processing;
+        // the clears below (and all runtime marker churn) then keep it
+        // current via deltas.
+        crate::metrics::set_pending_upload_markers(markers.len() as i64);
+
         for (slug, repo_name, local_path) in markers {
             if !local_path.exists() {
                 // Stale litter (repo dir gone) — storage is the best remaining
@@ -503,7 +508,6 @@ impl RepoStore {
                 }
             }
         }
-        crate::metrics::set_pending_upload_markers(still_pending as i64);
         (reuploaded, still_pending)
     }
 
@@ -992,11 +996,15 @@ fn mark_pending_upload(local_path: &Path, base_etag: Option<&str>) -> Result<()>
         .inspect_err(|_| {
             let _ = std::fs::remove_file(&tmp);
         })
-        .context("publishing pending-upload marker")
+        .context("publishing pending-upload marker")?;
+    crate::metrics::add_pending_upload_markers(1);
+    Ok(())
 }
 
 pub(crate) fn clear_pending_upload(local_path: &Path) {
-    let _ = std::fs::remove_file(pending_upload_marker(local_path));
+    if std::fs::remove_file(pending_upload_marker(local_path)).is_ok() {
+        crate::metrics::add_pending_upload_markers(-1);
+    }
 }
 
 /// Remove the marker after a *successful* upload, first atomically rewriting
@@ -1018,7 +1026,9 @@ fn clear_pending_upload_after_success(local_path: &Path, new_etag: Option<&str>)
             }
         }
     }
-    let _ = std::fs::remove_file(&marker);
+    if std::fs::remove_file(&marker).is_ok() {
+        crate::metrics::add_pending_upload_markers(-1);
+    }
 }
 
 /// Outcome of a marker-protected upload attempt.

crates/gitlawb-node/src/metrics.rs

@@ -287,6 +287,16 @@ pub fn set_pending_upload_markers(count: i64) {
     }
 }
 
+/// Adjust the pending-upload marker gauge by a delta (marker created = +1,
+/// marker removed = -1). The startup sweep seeds the absolute count via
+/// [`set_pending_upload_markers`]; runtime marker churn keeps it current
+/// through this.
+pub fn add_pending_upload_markers(delta: i64) {
+    if let Some(g) = PENDING_UPLOAD_MARKERS.get() {
+        g.add(delta);
+    }
+}
+
 /// 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,21 +49,23 @@ impl FsBlobStore {
         PathBuf::from(os)
     }
 
-    /// Per-final-path lock serializing the sidecar+blob publish step of `put`.
+    /// Striped 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()
+    /// serves stale data as current. A fixed stripe array (same-key puts
+    /// always hash to the same stripe) bounds memory regardless of how many
+    /// distinct keys the process ever writes; cross-key collisions only cost
+    /// a moment of extra serialization on the cheap rename pair.
+    fn put_publish_lock(path: &Path) -> &'static std::sync::Mutex<()> {
+        use std::hash::{Hash, Hasher};
+        use std::sync::{Mutex, OnceLock};
+        const STRIPES: usize = 64;
+        static LOCKS: OnceLock<Vec<Mutex<()>>> = OnceLock::new();
+        let locks = LOCKS.get_or_init(|| (0..STRIPES).map(|_| Mutex::new(())).collect());
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        path.hash(&mut hasher);
+        &locks[(hasher.finish() as usize) % STRIPES]
     }
 
     /// Fallback fingerprint for objects written before the sidecar existed.
@@ -122,8 +124,9 @@ impl BlobStore for FsBlobStore {
                 // 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");
+                let _guard = Self::put_publish_lock(&path2)
+                    .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")?;