From 96ed2b24ff4fe04679bccde2bd4d7dbf5fca020a Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 14:38:36 -0600 Subject: [PATCH] fix(node): subscriptions-only ws schema and unsettled-work purge guard - /graphql/ws serves a schema with no mutation root, so the socket can no longer execute task mutations outside the write-brake bucket; queries and subscriptions still run, and the schema carries the same Db, broadcast, and cursor-key data the resolvers consume - purge-spam skips a repo that still has unsettled receive-pack requests or pending ref transitions (checked under the repo lock, fail-closed) instead of stranding that work mid-effect --- crates/gitlawb-node/src/admin.rs | 23 +++++++++++++++++++ crates/gitlawb-node/src/auth/mod.rs | 7 ++++++ crates/gitlawb-node/src/db/mod.rs | 25 +++++++++++++++++++- crates/gitlawb-node/src/git/repo_store.rs | 3 +++ crates/gitlawb-node/src/graphql/mod.rs | 28 +++++++++++++++++++++++ crates/gitlawb-node/src/main.rs | 7 ++++++ crates/gitlawb-node/src/server.rs | 2 +- crates/gitlawb-node/src/state.rs | 1 + crates/gitlawb-node/src/test_support.rs | 8 +++++++ 9 files changed, 102 insertions(+), 2 deletions(-) diff --git a/crates/gitlawb-node/src/admin.rs b/crates/gitlawb-node/src/admin.rs index 4b7f2483..d4e3381c 100644 --- a/crates/gitlawb-node/src/admin.rs +++ b/crates/gitlawb-node/src/admin.rs @@ -47,6 +47,10 @@ pub struct PurgeSummary { /// authoritative emptiness recheck (fail-closed: never delete on a store /// error rather than risk deleting a repo with live remote refs). pub skipped_store_error: usize, + /// Candidates skipped because receive-pack requests or pending ref + /// transitions still have in-flight work on the repo; deleting its + /// storage would strand that work mid-effect. + pub skipped_unsettled: usize, /// Rows deleted whose on-disk dir removal FAILED (or was refused by the /// containment guard) — DB/disk drift the operator must reconcile. pub disk_failed: usize, @@ -466,6 +470,25 @@ pub async fn run_purge_spam( guard.release().await; continue; } + // Never delete storage under work still settling: receive-pack requests + // and pending ref transitions keyed by this repo would otherwise be + // stranded mid-effect. Fail closed on a check error. + let unsettled = match db.has_unsettled_repo_work(&c.id).await { + Ok(u) => u, + Err(e) => { + warn!(repo = %c.id, err = %e, + "purge-spam: could not check for unsettled work — skipped (fail-closed)"); + guard.release().await; + continue; + } + }; + if unsettled { + warn!(repo = %c.id, + "purge-spam: repo has unsettled receive-pack or transition work — skipped"); + summary.skipped_unsettled += 1; + guard.release().await; + continue; + } match db.delete_repo_by_id(&c.id).await { Ok(0) => { warn!(repo = %c.id, "purge-spam: repo row already gone — nothing to delete"); diff --git a/crates/gitlawb-node/src/auth/mod.rs b/crates/gitlawb-node/src/auth/mod.rs index ae83dc08..4ea3a132 100644 --- a/crates/gitlawb-node/src/auth/mod.rs +++ b/crates/gitlawb-node/src/auth/mod.rs @@ -738,6 +738,12 @@ mod tests { task_tx.clone(), task_cursor_key.clone(), )); + let ws_schema = Arc::new(graphql::build_ws_schema( + db.clone(), + ref_tx.clone(), + task_tx.clone(), + task_cursor_key.clone(), + )); crate::state::AppState { config: Arc::new(Config::parse_from(["gitlawb-node"])), db, @@ -748,6 +754,7 @@ mod tests { ref_update_tx: ref_tx, task_event_tx: task_tx, graphql_schema: schema, + graphql_ws_schema: ws_schema, task_cursor_key, machine_id: None, repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool), diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 4c92383e..47a0a9be 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -933,7 +933,7 @@ impl Db { .await .context("acquiring migration advisory lock")?; - let result = self.run_pending_migrations(&mut *lock_conn).await; + let result = self.run_pending_migrations(&mut lock_conn).await; let _ = sqlx::query("SELECT pg_advisory_unlock($1)") .bind(MIGRATION_ADVISORY_LOCK) @@ -947,6 +947,29 @@ impl Db { /// Must be called while holding the migration advisory lock. /// Runs through the connection already holding the migration advisory /// lock, so a pool of size 1 cannot deadlock against itself. + /// True while receive-pack requests or pending ref transitions still have + /// in-flight work on this repo. Purge checks this under the repo lock + /// before deleting storage: a row whose effects have not settled must not + /// be stranded against a repo that no longer exists. Settled receive-pack + /// states are 'complete' and 'rejected_at_git'; a 'quarantined' row stays + /// unsettled because the operator resolves it against the live repo. A + /// transition counts settled only once 'applied'. + pub(crate) async fn has_unsettled_repo_work(&self, repo_id: &str) -> Result { + let unsettled: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM receive_pack_requests + WHERE repo_id = $1 AND state NOT IN ('complete', 'rejected_at_git') + UNION ALL + SELECT 1 FROM pending_ref_transitions + WHERE repo_id = $1 AND state <> 'applied' + )", + ) + .bind(repo_id) + .fetch_one(&self.pool) + .await?; + Ok(unsettled) + } + async fn run_pending_migrations(&self, conn: &mut sqlx::PgConnection) -> Result<()> { for m in MIGRATIONS { let existing: Option = diff --git a/crates/gitlawb-node/src/git/repo_store.rs b/crates/gitlawb-node/src/git/repo_store.rs index c22a3a6f..30382437 100644 --- a/crates/gitlawb-node/src/git/repo_store.rs +++ b/crates/gitlawb-node/src/git/repo_store.rs @@ -811,6 +811,9 @@ impl RepoStore { } /// Initialize a new bare repo on local disk and upload to Tigris. + /// Test-only convenience over `init_handle`; every + /// production caller keeps the upload inside its lock. + #[cfg(test)] pub async fn init(&self, owner_did: &str, repo_name: &str) -> Result { let (path, upload) = self.init_handle(owner_did, repo_name).await?; drop(upload); diff --git a/crates/gitlawb-node/src/graphql/mod.rs b/crates/gitlawb-node/src/graphql/mod.rs index 70f8b92a..1097d470 100644 --- a/crates/gitlawb-node/src/graphql/mod.rs +++ b/crates/gitlawb-node/src/graphql/mod.rs @@ -14,6 +14,11 @@ use subscription::SubscriptionRoot; pub type GitlawbSchema = Schema; +/// No mutations: the /graphql/ws transport keeps queries and subscriptions but +/// has no mutation root, so the socket can never be used to run a write and +/// sidestep the write-brake bucket the HTTP route pays. +pub type GitlawbWsSchema = Schema; + // Keep ordinary schema discovery and application queries usable while // bounding validation work and the number of resolver selections per request. const GRAPHQL_MAX_COMPLEXITY: usize = 400; @@ -171,6 +176,29 @@ pub fn build_schema( .finish() } +/// Build the subscriptions-only schema served over /graphql/ws. Carries the +/// same broadcast senders the subscription resolvers consume; no query or +/// mutation root exists on this type, so the transport cannot be used to +/// execute either kind of operation. +pub fn build_ws_schema( + db: Arc, + ref_update_tx: tokio::sync::broadcast::Sender, + task_event_tx: tokio::sync::broadcast::Sender, + task_cursor_key: crate::api::task_cursor::TaskCursorKey, +) -> GitlawbWsSchema { + apply_query_limits(Schema::build( + QueryRoot, + async_graphql::EmptyMutation, + SubscriptionRoot, + )) + .data(db) + .data(ref_update_tx) + .data(task_event_tx) + .data(task_cursor_key) + .extension(TaskReadBrakeExtension) + .finish() +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/gitlawb-node/src/main.rs b/crates/gitlawb-node/src/main.rs index 7826331b..7b3bd7d6 100644 --- a/crates/gitlawb-node/src/main.rs +++ b/crates/gitlawb-node/src/main.rs @@ -328,6 +328,12 @@ async fn main() -> Result<()> { task_event_tx.clone(), task_cursor_key.clone(), )); + let graphql_ws_schema = Arc::new(graphql::build_ws_schema( + Arc::clone(&db), + ref_update_tx.clone(), + task_event_tx.clone(), + task_cursor_key.clone(), + )); let machine_id = std::env::var("FLY_MACHINE_ID").ok(); if let Some(ref mid) = machine_id { @@ -466,6 +472,7 @@ async fn main() -> Result<()> { ref_update_tx, task_event_tx, graphql_schema, + graphql_ws_schema, machine_id, repo_store, rate_limiter, diff --git a/crates/gitlawb-node/src/server.rs b/crates/gitlawb-node/src/server.rs index 1283d011..bc68e271 100644 --- a/crates/gitlawb-node/src/server.rs +++ b/crates/gitlawb-node/src/server.rs @@ -67,7 +67,7 @@ async fn graphql_ws_handler( key: rate_limit::client_key(&headers, peer, state.push_limiter_trust), request_count: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)), }); - let schema = state.graphql_schema.as_ref().clone(); + let schema = state.graphql_ws_schema.as_ref().clone(); upgrade .protocols(ALL_WEBSOCKET_PROTOCOLS) .on_upgrade(move |stream| { diff --git a/crates/gitlawb-node/src/state.rs b/crates/gitlawb-node/src/state.rs index 5705e076..a15bb35a 100644 --- a/crates/gitlawb-node/src/state.rs +++ b/crates/gitlawb-node/src/state.rs @@ -72,6 +72,7 @@ pub struct AppState { pub task_event_tx: tokio::sync::broadcast::Sender, /// GraphQL schema (queries + mutations + subscriptions) pub graphql_schema: Arc, + pub graphql_ws_schema: Arc, /// MAC key for task-list continuation tokens, derived from the node /// keypair seed. Held here (and handed to the GraphQL schema) so REST and /// GraphQL mint and accept the same tokens. diff --git a/crates/gitlawb-node/src/test_support.rs b/crates/gitlawb-node/src/test_support.rs index d36b0b7c..1671ac96 100644 --- a/crates/gitlawb-node/src/test_support.rs +++ b/crates/gitlawb-node/src/test_support.rs @@ -94,6 +94,13 @@ fn build_state(db: Arc, pool: PgPool) -> AppState { task_tx.clone(), task_cursor_key.clone(), )); + + let ws_schema = Arc::new(graphql::build_ws_schema( + db.clone(), + ref_tx.clone(), + task_tx.clone(), + task_cursor_key.clone(), + )); AppState { config: Arc::new(Config::parse_from(["gitlawb-node"])), db, @@ -104,6 +111,7 @@ fn build_state(db: Arc, pool: PgPool) -> AppState { ref_update_tx: ref_tx, task_event_tx: task_tx, graphql_schema: schema, + graphql_ws_schema: ws_schema, task_cursor_key, machine_id: None, repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),