fix(repo_store): dedicate lock pool, acquire inside retry loop, close conn on unlock cancellation

Author: Gravirei

CHANGELOG.md

@@ -1,11 +1,5 @@
 # Changelog
 
-## Unreleased
-
-### Known caveats
-
-* **repo_store:** the advisory-lock key for cross-machine write exclusion is now derived via SHA-256 of `(owner_slug, repo_name)` instead of `std::collections::DefaultHasher` (the previous algorithm was not frozen by the Rust standard). During a shared-Postgres rolling upgrade, old-binary nodes holding the legacy `DefaultHasher` key and new-binary nodes holding the SHA-256 key for the same repo compute *different* i64 keys, so PostgreSQL treats them as independent locks and cross-machine write-exclusion is lost for the deploy window. Operators should drain in-flight writes or cut over through a single node before bringing new-binary nodes online. A future transition release may acquire both keys for one cycle. See `RepoStore::acquire_write` and issue #210.
-
 ## [0.5.1](https://github.com/Gitlawb/node/compare/v0.5.0...v0.5.1) (2026-07-10)
 
 

crates/gitlawb-node/src/auth/mod.rs

@@ -495,6 +495,10 @@ mod tests {
         let pool = sqlx::postgres::PgPoolOptions::new()
             .connect_lazy("postgres://localhost/gitlawb_test_placeholder")
             .expect("lazy pool creation should not fail");
+        let lock_pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(2)
+            .connect_lazy("postgres://localhost/gitlawb_test_placeholder")
+            .expect("lazy lock pool creation should not fail");
         let db = Arc::new(crate::db::Db::for_testing(pool.clone()));
         let schema = Arc::new(graphql::build_schema(
             db.clone(),
@@ -512,7 +516,11 @@ mod tests {
             task_event_tx: task_tx,
             graphql_schema: schema,
             machine_id: None,
-            repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
+            repo_store: crate::git::repo_store::RepoStore::for_testing(
+                PathBuf::from("/tmp"),
+                pool,
+                lock_pool,
+            ),
             rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
             create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
             push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),

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

@@ -26,29 +26,42 @@ use super::tigris::TigrisClient;
 pub struct RepoStore {
     repos_dir: PathBuf,
     tigris: Option<TigrisClient>,
-    /// Shared Postgres pool for advisory locks.
+    /// Shared Postgres pool for application queries.
+    /// NOTE: advisory-lock connections use `lock_pool` instead.
+    #[allow(dead_code)]
     pool: PgPool,
+    /// Dedicated pool for advisory-lock sessions. Kept separate from
+    /// `pool` so lock-holding connections (which may be held for the full
+    /// duration of a write) never compete with application queries.
+    lock_pool: PgPool,
     /// Tracks repos already confirmed to exist in Tigris — avoids redundant
     /// HEAD checks and background uploads for repos we've already migrated.
     migrated: Arc<Mutex<HashSet<String>>>,
 }
 
 impl RepoStore {
     #[cfg(test)]
-    pub fn for_testing(repos_dir: PathBuf, pool: PgPool) -> Self {
+    pub fn for_testing(repos_dir: PathBuf, pool: PgPool, lock_pool: PgPool) -> Self {
         Self {
             repos_dir,
             tigris: None,
             pool,
+            lock_pool,
             migrated: Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())),
         }
     }
 
-    pub fn new(repos_dir: PathBuf, tigris: Option<TigrisClient>, pool: PgPool) -> Self {
+    pub fn new(
+        repos_dir: PathBuf,
+        tigris: Option<TigrisClient>,
+        pool: PgPool,
+        lock_pool: PgPool,
+    ) -> Self {
         Self {
             repos_dir,
             tigris,
             pool,
+            lock_pool,
             migrated: Arc::new(Mutex::new(HashSet::new())),
         }
     }
@@ -204,40 +217,47 @@ impl RepoStore {
         // writer cannot be handed this connection and reentrantly grab the
         // same lock, and the unlock on release() is guaranteed to run on the
         // owning session.
-        let mut conn: PoolConnection<sqlx::Postgres> = self
-            .pool
-            .acquire()
-            .await
-            .context("acquiring connection for advisory lock")?;
-        let mut acquired = false;
+        //
+        // Each probe checks out and releases a connection from `lock_pool`
+        // (not the shared `pool`), so a writer that waits for a contended repo
+        // never occupies a slot that an unrelated query needs.
+        let mut guard: Option<RepoWriteGuard> = None;
         for attempt in 0..60 {
+            let mut conn: PoolConnection<sqlx::Postgres> = self
+                .lock_pool
+                .acquire()
+                .await
+                .context("acquiring connection for advisory lock")?;
+
             let row: (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock($1)")
                 .bind(lock_key)
                 .fetch_one(&mut *conn)
                 .await
                 .context("trying advisory lock")?;
+
             if row.0 {
-                acquired = true;
+                // Winner — wrap immediately so Drop protects cancellation.
+                guard = Some(RepoWriteGuard {
+                    owner_slug: owner_slug.clone(),
+                    repo_name: repo_name.to_string(),
+                    local_path,
+                    lock_key,
+                    conn: Some(conn),
+                    tigris: self.tigris.clone(),
+                });
                 break;
             }
+            // Not acquired — conn drops back to lock_pool (no lock held).
             if attempt < 59 {
                 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
             }
         }
-        if !acquired {
-            anyhow::bail!("could not acquire advisory lock after 60s — possible stale lock for {owner_slug}/{repo_name}");
-        }
 
-        // Construct the guard immediately so its Drop implementation protects
-        // subsequent cancellation points (tigris download below).
-        let mut guard = RepoWriteGuard {
-            owner_slug,
-            repo_name: repo_name.to_string(),
-            local_path,
-            lock_key,
-            conn: Some(conn),
-            tigris: self.tigris.clone(),
-        };
+        let mut guard = guard.ok_or_else(|| {
+            anyhow::anyhow!(
+                "could not acquire advisory lock after 60s — possible stale lock for {owner_slug}/{repo_name}"
+            )
+        })?;
 
         // Always download the latest from Tigris before writing.
         // Local disk may be stale if another machine pushed since our last access.
@@ -260,16 +280,23 @@ impl RepoStore {
                             "write acquire: tigris download failed — falling back to local copy");
                     } else {
                         // Drop the lock on the error path before returning.
-                        // Unlock through the guard's connection, then clear conn
-                        // so Drop returns it to the pool (lock-free) instead of
-                        // closing it.
+                        // Unlock through the guard's connection (borrowed, not
+                        // taken), then clear conn so Drop returns the unlocked
+                        // connection to the pool. If the unlock query itself is
+                        // cancelled, Drop's close_on_drop handles it.
                         if let Some(ref mut gconn) = guard.conn {
-                            let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
+                            let unlocked = sqlx::query("SELECT pg_advisory_unlock($1)")
                                 .bind(guard.lock_key)
                                 .execute(&mut **gconn)
                                 .await;
+                            if let Err(e) = unlocked {
+                                warn!(repo = %repo_name, err = %e,
+                                    "advisory unlock failed on error path — connection will be closed");
+                                // Leave conn in guard so Drop closes it.
+                            } else {
+                                guard.conn = None;
+                            }
                         }
-                        guard.conn = None;
                         return Err(e).context("downloading repo from tigris for write");
                     }
                 }
@@ -465,22 +492,50 @@ impl RepoWriteGuard {
         }
 
         // Release advisory lock on the owning session, then return connection
-        // to the pool. Take the connection so Drop sees None and does nothing.
-        if let Some(mut conn) = self.conn.take() {
-            let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
+        // to the pool.
+        //
+        // Crucially we keep the connection inside `self.conn` across the
+        // `.await` so that if the future is cancelled here, the guard's `Drop`
+        // (which calls `close_on_drop`) will close the session and the lock
+        // is released by Postgres when the backend dies. Only after a
+        // *successful* unlock is the connection taken out and returned to the
+        // pool normally.
+        //
+        // See also `Drop` below.
+        if let Some(ref mut conn) = self.conn {
+            match sqlx::query("SELECT pg_advisory_unlock($1)")
                 .bind(self.lock_key)
-                .execute(&mut *conn)
-                .await;
-            // conn drops here, returned to the pool
+                .execute(&mut **conn)
+                .await
+            {
+                Ok(_) => {
+                    // Unlock confirmed — safe to return connection to the pool.
+                    self.conn.take();
+                }
+                Err(e) => {
+                    // Unlock failed (e.g. connection broken). Leave conn in
+                    // the guard so Drop closes the session — the lock will be
+                    // released when Postgres sees the backend die.
+                    warn!(
+                        repo = %self.repo_name,
+                        err = %e,
+                        "advisory unlock query failed — closing session to release lock"
+                    );
+                }
+            }
         }
     }
 }
 
 impl Drop for RepoWriteGuard {
     fn drop(&mut self) {
-        // If conn is still Some, release() was never called (e.g. cancellation).
-        // Close the connection instead of returning it to the pool, so the
-        // Postgres backend session ends and the advisory lock is released.
+        // If conn is still Some, either:
+        //   * release() was never called (cancellation), or
+        //   * the unlock query itself failed / was cancelled.
+        //
+        // In every case, the lock is (or may be) held on this session.
+        // Close the connection instead of returning it to the pool, so
+        // Postgres ends the backend and releases the lock.
         if let Some(ref mut conn) = self.conn {
             conn.close_on_drop();
         }
@@ -623,7 +678,13 @@ mod tests {
         // the pool or the network. Fabricate a pool reference via PgPool::connect_lazy
         // so we don't need a live DB.
         let pool = sqlx::PgPool::connect_lazy("postgres://invalid").unwrap();
-        RepoStore::new(PathBuf::from("/var/lib/gitlawb/repos"), None, pool)
+        let lock_pool = sqlx::PgPool::connect_lazy("postgres://invalid").unwrap();
+        RepoStore::new(
+            PathBuf::from("/var/lib/gitlawb/repos"),
+            None,
+            pool,
+            lock_pool,
+        )
     }
 
     #[tokio::test]
@@ -679,6 +740,96 @@ mod tests {
         assert_eq!(key, -6680856138670956537_i64);
     }
 
+    // ── integration: advisory lock exclusion ────────────────────────────────
+    //
+    // These tests exercise the full acquire/release/cancel lifecycle against a
+    // real Postgres and are marked #[sqlx::test] so the framework provisions a
+    // fresh database for each.
+
+    fn make_store_for_live(pool: PgPool) -> RepoStore {
+        let lock_pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(2)
+            .connect_lazy("postgres://invalid")
+            .expect("lazy lock pool for testing");
+        RepoStore::new(PathBuf::from("/tmp"), None, pool, lock_pool)
+    }
+
+    #[sqlx::test]
+    #[ignore = "requires DATABASE_URL — run manually against a real Postgres"]
+    async fn competing_session_is_excluded_while_guard_held(pool: PgPool) {
+        let store = make_store_for_live(pool.clone());
+
+        // Hold the lock in a guard.
+        let guard = store
+            .acquire_write("did:key:z6MkAlice", "excl-test")
+            .await
+            .unwrap();
+
+        // A second session on the same pool cannot acquire the same lock key.
+        let key = advisory_lock_key("did_key_z6MkAlice", "excl-test");
+        let row: (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock($1)")
+            .bind(key)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert!(!row.0, "competing session must be excluded");
+
+        // Release.
+        guard.release(true).await;
+
+        // Now the same session can acquire the lock.
+        let row: (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock($1)")
+            .bind(key)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        sqlx::query("SELECT pg_advisory_unlock($1)")
+            .bind(key)
+            .execute(&pool)
+            .await
+            .unwrap();
+        assert!(row.0, "lock must be released after guard.release()");
+    }
+
+    #[sqlx::test]
+    #[ignore = "requires DATABASE_URL — run manually against a real Postgres"]
+    async fn cancelled_guard_releases_lock_when_dropped(pool: PgPool) {
+        // Spawn a task that acquires the lock and then drops the guard without
+        // calling release() — simulating cancellation.
+        let lock_pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(2)
+            .connect_lazy("postgres://invalid")
+            .expect("lazy lock pool");
+        let inner_store = RepoStore::new(PathBuf::from("/tmp"), None, pool.clone(), lock_pool);
+        let handle = tokio::spawn(async move {
+            let guard = inner_store
+                .acquire_write("did:key:z6MkBob", "cancel-test")
+                .await
+                .unwrap();
+            // Drop guard without calling release().
+            drop(guard);
+        });
+        handle.await.unwrap();
+
+        // After the task completes, the lock must be released (the guard's Drop
+        // should have closed the connection, ending the session).
+        let key = advisory_lock_key("did_key_z6MkBob", "cancel-test");
+        let row: (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock($1)")
+            .bind(key)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        sqlx::query("SELECT pg_advisory_unlock($1)")
+            .bind(key)
+            .execute(&pool)
+            .await
+            .unwrap();
+        assert!(
+            row.0,
+            "lock must be released after guard is dropped without release()"
+        );
+    }
+
     #[test]
     fn advisory_lock_key_differs_for_different_inputs() {
         // Vary one axis at a time so a regression that drops either parameter

crates/gitlawb-node/src/main.rs

@@ -279,8 +279,20 @@ async fn main() -> Result<()> {
         None
     };
 
-    let repo_store =
-        git::repo_store::RepoStore::new(config.repos_dir.clone(), tigris, db.pool().clone());
+    let lock_pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(2)
+        .acquire_timeout(std::time::Duration::from_secs(
+            config.db_acquire_timeout_secs,
+        ))
+        .connect(&config.database_url)
+        .await
+        .context("creating advisory-lock pool")?;
+    let repo_store = git::repo_store::RepoStore::new(
+        config.repos_dir.clone(),
+        tigris,
+        db.pool().clone(),
+        lock_pool,
+    );
 
     // Per-DID limiter for the creation endpoints. Keyed on the authenticated
     // DID (attacker-varied), so bound its key set to cap memory.

crates/gitlawb-node/src/test_support.rs

@@ -74,7 +74,14 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
         task_event_tx: task_tx,
         graphql_schema: schema,
         machine_id: None,
-        repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
+        repo_store: crate::git::repo_store::RepoStore::for_testing(
+            PathBuf::from("/tmp"),
+            pool,
+            sqlx::postgres::PgPoolOptions::new()
+                .max_connections(2)
+                .connect_lazy("postgres://localhost/gitlawb_test_placeholder")
+                .expect("lazy lock pool creation should not fail"),
+        ),
         rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
         create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
         push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),

docs/RUN-A-NODE.md

@@ -178,6 +178,7 @@ GITLAWB_ENFORCE_OWNER_PUSH=true
 | Operator key | Dedicated wallet, small ETH balance, not your main treasury |
 | Monitoring | Watch `lastHeartbeat` on-chain; alert if > 22h since last beat |
 | Public URL | Must resolve and serve `/health` — peers will ping it |
+| Rolling upgrade (advisory-lock algorithm) | v0.7.0 re-keys every repo's advisory lock (SHA-256 replaces `DefaultHasher`). During a shared-Postgres rolling upgrade, old nodes compute a different `i64` key than new nodes for the same repo, so PostgreSQL treats them as independent locks and cross-machine write-exclusion is lost. **Drain in-flight writes** or cut over through a **single node** before bringing new-binary nodes online. |
 
 ---