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
This commit is contained in:
beardthelion committed 2026-11-11 14:45:34 -06:00
1 parent edfa6bb7b4
commit 96ed2b24ff
9 files changed
+102 -2

No files matched your search

+23
View File
@@ -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");
+7
View File
@@ -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),
+24 -1
View File
@@ -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<bool> {
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<String> =
@@ -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<PathBuf> {
let (path, upload) = self.init_handle(owner_did, repo_name).await?;
drop(upload);
+28
View File
@@ -14,6 +14,11 @@ use subscription::SubscriptionRoot;
pub type GitlawbSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;
/// 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<QueryRoot, async_graphql::EmptyMutation, SubscriptionRoot>;
// 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<Db>,
ref_update_tx: tokio::sync::broadcast::Sender<RefUpdateBroadcast>,
task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
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::*;
+7
View File
@@ -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,
+1 -1
View File
@@ -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| {
+1
View File
@@ -72,6 +72,7 @@ pub struct AppState {
pub task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
/// GraphQL schema (queries + mutations + subscriptions)
pub graphql_schema: Arc<crate::graphql::GitlawbSchema>,
pub graphql_ws_schema: Arc<crate::graphql::GitlawbWsSchema>,
/// 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.
+8
View File
@@ -94,6 +94,13 @@ fn build_state(db: Arc<crate::db::Db>, 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<crate::db::Db>, 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),