From bbad34ab261dfd94d17e77801b857dff8d511522 Mon Sep 17 00:00:00 2001 From: Isaac Corbrey Date: Thu, 24 Sep 2026 00:16:01 -0400 Subject: [PATCH] jetstream: tolerate joins referencing deleted games MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handle_action getRecords each joins game; deleted games 400 and blow up the whole event. A stale join is expected noise in replayed streams — skip it quietly while every other failure stays loud. --- src/jetstream.rs | 15 ++++++++++++-- src/main.rs | 51 ++++++++++++++++++++++++++++++++++++------------ 2 files changed, 51 insertions(+), 15 deletions(-) diff --git a/src/jetstream.rs b/src/jetstream.rs index b4231e6..fc49fda 100644 --- a/src/jetstream.rs +++ b/src/jetstream.rs @@ -444,7 +444,10 @@ async fn handle_join(app: &App, record: &serde_json::Value) -> Result<()> { } } - let current = app + // A join can reference a game deleted since it was posted — stale noise + // in replayed streams, and expected often enough that it must not fail + // the event. Every other failure still takes the loud error path. + let current = match app .agent .api .com @@ -459,7 +462,15 @@ async fn handle_join(app: &App, record: &serde_json::Value) -> Result<()> { } .into(), ) - .await?; + .await + { + Ok(current) => current, + Err(e) if crate::record_not_found(&e) => { + debug!(game = %game_uri, "join references a deleted game — skipping"); + return Ok(()); + } + Err(e) => return Err(e.into()), + }; let current_val = unknown_to_value(¤t.value); let current_obj = match current_val.as_object() { diff --git a/src/main.rs b/src/main.rs index 672e840..a350512 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,6 +6,8 @@ mod state; use anyhow::{Context, Result}; use atrium_api::agent::atp_agent::{store::MemorySessionStore, AtpAgent}; +use atrium_api::com::atproto::repo::get_record::Error as GetRecordError; +use atrium_api::xrpc::error::{Error as XrpcError, XrpcErrorKind}; use atrium_xrpc_client::reqwest::ReqwestClient; use sqlx::sqlite::SqlitePoolOptions; use std::collections::HashMap; @@ -287,23 +289,21 @@ async fn list_repo_records(app: &App, repo: &str, collection: &str) -> Result, -) -> bool { - use atrium_api::com::atproto::repo::get_record::Error as GetRecordError; - use atrium_api::xrpc::error::{Error as XrpcError, XrpcErrorKind}; +fn xrpc_kind_is_not_found(kind: &XrpcErrorKind) -> bool { + match kind { + XrpcErrorKind::Custom(GetRecordError::RecordNotFound(_)) => true, + XrpcErrorKind::Undefined(body) => body.error.as_deref() == Some("RecordNotFound"), + _ => false, + } +} +/// Whether a get_record failure means "no such record". +pub(crate) fn record_not_found(err: &XrpcError) -> bool { match err { - XrpcError::XrpcResponse(xe) => match &xe.error { - Some(XrpcErrorKind::Custom(GetRecordError::RecordNotFound(_))) => true, - Some(XrpcErrorKind::Undefined(body)) => { - body.error.as_deref() == Some("RecordNotFound") - } - _ => false, - }, + XrpcError::XrpcResponse(xe) => xe.error.as_ref().is_some_and(xrpc_kind_is_not_found), _ => false, } } @@ -357,6 +357,31 @@ async fn fetch_game_record(app: &App, record_uri: &str) -> Result = + XrpcErrorKind::Custom(GetRecordError::RecordNotFound(None)); + assert!(xrpc_kind_is_not_found(&missing)); + + let undefined_name: XrpcErrorKind = XrpcErrorKind::Undefined(ErrorResponseBody { + error: Some("RecordNotFound".into()), + message: Some("Could not find record".into()), + }); + assert!(xrpc_kind_is_not_found(&undefined_name)); + + let other: XrpcErrorKind = XrpcErrorKind::Undefined(ErrorResponseBody { + error: Some("InvalidRequest".into()), + message: None, + }); + assert!(!xrpc_kind_is_not_found(&other), "real failures must stay loud"); + } +} + /// Fetch a single game's moves and passes posted into `repo` (one player's /// own repo), filtered by exact `game` URI match. Both share one /// moveNumber sequence (per the boo.sky.go.pass lexicon), so they're -- knotmirror 0.0.1