fix(node): harden release-upload outcomes and replication tail gating

Treat timed-out or failed archive uploads as non-durable in into_result,
gate pinata/gossip on confirmed publish durability while keeping inv22 spawn
order, split lock-pool vs main DB pool validation, map LockPoolBusy through
From<anyhow::Error>, and attach Retry-After on repo write contention errors.

Author: beardthelion

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

@@ -2374,13 +2374,15 @@ pub async fn git_receive_pack(
     // would return 200 to the pusher before the durable copy lands, which is a larger
     // change to the client contract than the window it closes.
     let push_succeeded = receive_result.is_ok();
+    let publish_durability = Arc::new(tokio::sync::Mutex::new(None));
     if push_succeeded {
         tokio::spawn(post_receive_replication_tail(
             state.clone(),
             record.clone(),
             ref_updates.clone(),
             disk_path.clone(),
             auth.0.to_string(),
+            Some(publish_durability.clone()),
         ));
     }
 
@@ -2401,7 +2403,11 @@ pub async fn git_receive_pack(
     // touching the repo, recording the push, bumping trust, issuing ref
     // certificates or answering 200 would all be reporting a write no other
     // node can read.
-    reclaimed.release(push_succeeded).await.into_result()?;
+    let outcome = reclaimed.release(push_succeeded).await;
+    if push_succeeded {
+        *publish_durability.lock().await = Some(outcome);
+    }
+    outcome.into_result()?;
     // Clean path: clone (a) already dropped inside run_git_service when the receive-pack
     // group was reaped; clone (b) held here spanned the success-only Tigris upload that
     // ran inside release() above. Drop it now so a second same-repo push proceeds the
@@ -2518,12 +2524,36 @@ pub async fn git_receive_pack(
 /// the per-repo-coalesced pin/encrypt task, and this push's own Pinata + announce
 /// task. Split out of `git_receive_pack` so the ordering the coalescing gate depends
 /// on is directly testable; the handler spawns it and returns.
+async fn publish_durability_confirmed(
+    slot: &Option<Arc<tokio::sync::Mutex<Option<crate::git::repo_store::ReleaseOutcome>>>>,
+    wait: std::time::Duration,
+) -> bool {
+    let Some(slot) = slot else {
+        return true;
+    };
+    let start = std::time::Instant::now();
+    loop {
+        if let Some(outcome) = *slot.lock().await {
+            return matches!(outcome, crate::git::repo_store::ReleaseOutcome::Released);
+        }
+        if start.elapsed() >= wait {
+            // Handler disconnected during `release` without recording an outcome.
+            // Proceed so inv22's disconnect-during-release tail still runs.
+            return true;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+    }
+}
+
 async fn post_receive_replication_tail(
     state: AppState,
     record: RepoRecord,
     ref_updates: Vec<RefUpdate>,
     disk_path: std::path::PathBuf,
     did: String,
+    publish_durability: Option<
+        Arc<tokio::sync::Mutex<Option<crate::git::repo_store::ReleaseOutcome>>>,
+    >,
 ) {
     // Replication enforcement (Phase 2): decide once per push whether the public
     // may read this repo at all and, if so, which blob OIDs must not leave the
@@ -2697,6 +2727,21 @@ async fn post_receive_replication_tail(
         ));
     }
 
+    let durability_wait = std::time::Duration::from_secs(
+        state
+            .config
+            .lock_held_transfer_timeout_secs
+            .saturating_add(5),
+    );
+    if announce_at_root && !publish_durability_confirmed(&publish_durability, durability_wait).await
+    {
+        tracing::warn!(
+            repo = %record.id,
+            "skipping pinata/gossip tail: publish durability was not confirmed"
+        );
+        return;
+    }
+
     // Pin new git objects to Pinata, then record branch→CID and gossip.
     //
     // #174 P2-2 scope note: this SECOND detached spawn is deliberately NOT brought
@@ -9819,6 +9864,7 @@ mod tests {
             f2a_update("refs/heads/main", &c2),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
         let after_first = f2a_walks(&log);
@@ -9839,6 +9885,7 @@ mod tests {
             f2a_update("refs/heads/second", &c1),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -9912,6 +9959,7 @@ mod tests {
             f2a_update("refs/heads/main", &c2),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -9993,6 +10041,7 @@ mod tests {
             f2a_update("refs/heads/main", &c2),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         ));
         f2a_wait_for(|| started.exists(), "the admitted push's walk to start").await;
 
@@ -10106,6 +10155,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -10384,6 +10434,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -10439,6 +10490,7 @@ mod tests {
             f2a_update("refs/heads/main", &c1),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
         tokio::time::sleep(std::time::Duration::from_millis(300)).await;
@@ -10480,6 +10532,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -10577,6 +10630,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 

crates/gitlawb-node/src/config.rs

@@ -252,11 +252,9 @@ pub struct Config {
 
     /// Maximum connections in the PostgreSQL pool. This is a cap, not a floor
     /// (connections open lazily). Size against the database server's
-    /// max_connections, remembering admin tooling opens its own pool. Each
-    /// concurrent write pins one pooled connection for its whole duration (the
-    /// advisory lock in `repo_store::acquire_write` is connection-affine), so this
-    /// must exceed `max_concurrent_git_pushes` by `DB_POOL_APP_HEADROOM` or slow
-    /// pushes starve every other DB path — enforced by `Config::validate`.
+    /// max_connections, remembering admin tooling opens its own pool. Writes pin
+    /// the dedicated lock pool, not this one; this pool still needs enough headroom
+    /// for ordinary request handlers and metadata paths (`DB_POOL_APP_HEADROOM`).
     #[arg(
         long,
         env = "GITLAWB_DB_MAX_CONNECTIONS",
@@ -769,21 +767,23 @@ impl Config {
     /// catches combinations that ship a denial-of-service under otherwise-valid
     /// values. Call once at startup and fail fast on `Err`.
     pub fn validate(&self) -> Result<(), String> {
-        // A write pins one pooled connection for its whole duration (the
-        // connection-affine advisory lock in repo_store::acquire_write), and
-        // concurrent writes are capped at max_concurrent_git_pushes. If the pool
-        // does not exceed that cap by DB_POOL_APP_HEADROOM, a burst of slow pushes
-        // drains every connection and every other DB path 503s. (#174 F1)
-        let floor = (self.max_concurrent_git_pushes as u64) + (Self::DB_POOL_APP_HEADROOM as u64);
-        if (self.db_max_connections as u64) < floor {
+        // Concurrent git writes pin the dedicated lock pool for their whole
+        // duration (connection-affine advisory lock in acquire_write). The main
+        // pool no longer carries that occupancy.
+        if (self.db_lock_pool_max_connections as usize) < self.max_concurrent_git_pushes {
             return Err(format!(
-                "GITLAWB_DB_MAX_CONNECTIONS ({}) must be at least max_concurrent_git_pushes ({}) \
-                 + {} headroom = {}: each concurrent write pins one pooled connection for its whole \
-                 duration, so a smaller pool lets a burst of slow pushes starve every other DB path.",
-                self.db_max_connections,
-                self.max_concurrent_git_pushes,
-                Self::DB_POOL_APP_HEADROOM,
-                floor
+                "GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS ({}) must be at least \
+                 max_concurrent_git_pushes ({}) so every admitted push can pin a \
+                 lock-pool connection for its whole duration",
+                self.db_lock_pool_max_connections, self.max_concurrent_git_pushes
+            ));
+        }
+        let main_floor = Self::DB_POOL_APP_HEADROOM as u64;
+        if (self.db_max_connections as u64) < main_floor {
+            return Err(format!(
+                "GITLAWB_DB_MAX_CONNECTIONS ({}) must be at least {} for non-git \
+                 database paths",
+                self.db_max_connections, main_floor
             ));
         }
         Ok(())
@@ -1422,41 +1422,48 @@ mod tests {
         );
     }
 
-    /// #174 F1: a connection-affine write lock pins a pooled connection per
-    /// concurrent write, so the pool must clear `max_concurrent_git_pushes` by
-    /// `DB_POOL_APP_HEADROOM` or a push burst starves every other DB path.
-    /// `validate()` must reject an under-sized pool at boot.
+    /// #174 F1: concurrent git writes pin the dedicated lock pool, not the main
+    /// pool. `validate()` must reject an under-sized lock pool at boot.
     #[test]
     fn db_pool_must_clear_the_git_push_cap() {
-        // Shipped defaults validate (48 >= 32 + 8).
+        // Shipped defaults validate (lock pool 32 >= pushes 32, main pool >= headroom).
         Config::parse_from(["gitlawb-node"])
             .validate()
             .expect("default config must validate");
 
-        // An under-sized pool relative to the push cap is rejected (20 < 32 + 8).
-        let under = Config::parse_from([
+        // An under-sized lock pool relative to the push cap is rejected.
+        let under_lock = Config::parse_from([
             "gitlawb-node",
-            "--db-max-connections",
-            "20",
+            "--db-lock-pool-max-connections",
+            "16",
             "--max-concurrent-git-pushes",
             "32",
         ]);
         assert!(
-            under.validate().is_err(),
-            "db_max_connections 20 below max_concurrent_git_pushes 32 + headroom must be rejected"
+            under_lock.validate().is_err(),
+            "db_lock_pool_max_connections below max_concurrent_git_pushes must be rejected"
         );
 
-        // Exactly at the floor validates (40 == 32 + 8).
-        let at_floor = Config::parse_from([
+        // Main pool can be smaller than pushes + headroom when the lock pool carries writes.
+        let split = Config::parse_from([
             "gitlawb-node",
             "--db-max-connections",
-            "40",
+            "16",
+            "--db-lock-pool-max-connections",
+            "32",
             "--max-concurrent-git-pushes",
             "32",
         ]);
         assert!(
-            at_floor.validate().is_ok(),
-            "db_max_connections at the floor (pushes + headroom) must validate"
+            split.validate().is_ok(),
+            "a small main pool with a large lock pool must validate"
+        );
+
+        // Main pool still needs the app headroom floor.
+        let under_main = Config::parse_from(["gitlawb-node", "--db-max-connections", "4"]);
+        assert!(
+            under_main.validate().is_err(),
+            "db_max_connections below DB_POOL_APP_HEADROOM must be rejected"
         );
     }
 

crates/gitlawb-node/src/error.rs

@@ -123,20 +123,24 @@ impl From<anyhow::Error> for AppError {
             // 500. The internal message names the owner slug and repo, so the
             // variant carries nothing: the detail stays in the log at the raise
             // site and the client gets a fixed retryable body.
-            Err(err) => match err.downcast::<crate::git::repo_store::RepoBusy>() {
-                Ok(_) => AppError::RepoBusy,
-                // Same reasoning one rung down: a refused under-lock refresh is a
-                // transient storage condition, and its internal message names the
-                // owner slug and repo, so the variant carries nothing.
-                Err(err) => match err.downcast::<crate::git::repo_store::RepoUnavailable>() {
-                    Ok(_) => AppError::RepoUnavailable,
-                    // And one more rung: a publish the store refused twice is
-                    // transient in the same way, and the retry is the client's
-                    // to make. The variant carries nothing for the same reason
-                    // as the two above.
-                    Err(err) => match err.downcast::<crate::git::repo_store::RepoWriteFenced>() {
-                        Ok(_) => AppError::RepoWriteFenced,
-                        Err(err) => AppError::Internal(err),
+            Err(err) => match err.downcast::<crate::git::repo_store::LockPoolBusy>() {
+                Ok(_) => AppError::Overloaded("git write locks at capacity, retry shortly".into()),
+                Err(err) => match err.downcast::<crate::git::repo_store::RepoBusy>() {
+                    Ok(_) => AppError::RepoBusy,
+                    // Same reasoning one rung down: a refused under-lock refresh is a
+                    // transient storage condition, and its internal message names the
+                    // owner slug and repo, so the variant carries nothing.
+                    Err(err) => match err.downcast::<crate::git::repo_store::RepoUnavailable>() {
+                        Ok(_) => AppError::RepoUnavailable,
+                        // And one more rung: a publish the store refused twice is
+                        // transient in the same way, and the retry is the client's
+                        // to make. The variant carries nothing for the same reason
+                        // as the two above.
+                        Err(err) => match err.downcast::<crate::git::repo_store::RepoWriteFenced>()
+                        {
+                            Ok(_) => AppError::RepoWriteFenced,
+                            Err(err) => AppError::Internal(err),
+                        },
                     },
                 },
             },
@@ -296,7 +300,11 @@ impl IntoResponse for AppError {
         // here rather than in bespoke early returns, keeping each variant handled once.
         if matches!(
             self,
-            AppError::Overloaded(_) | AppError::SearchIncomplete { .. }
+            AppError::Overloaded(_)
+                | AppError::SearchIncomplete { .. }
+                | AppError::RepoBusy
+                | AppError::RepoUnavailable
+                | AppError::RepoWriteFenced
         ) {
             resp.headers_mut().insert(
                 axum::http::header::RETRY_AFTER,
@@ -414,4 +422,31 @@ mod tests {
         let resp = err.into_response();
         assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
     }
+
+    #[test]
+    fn lock_pool_busy_via_anyhow_from_is_503_overloaded() {
+        let err: AppError = anyhow::Error::new(crate::git::repo_store::LockPoolBusy).into();
+        let resp = err.into_response();
+        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(
+            resp.headers().get("retry-after").unwrap().to_str().unwrap(),
+            "1"
+        );
+    }
+
+    #[test]
+    fn repo_busy_unavailable_and_fenced_advertise_retry_after() {
+        for err in [
+            AppError::RepoBusy,
+            AppError::RepoUnavailable,
+            AppError::RepoWriteFenced,
+        ] {
+            let resp = err.into_response();
+            assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+            assert_eq!(
+                resp.headers().get("retry-after").unwrap().to_str().unwrap(),
+                "1"
+            );
+        }
+    }
 }