From 00209076c7a10d9e4a475466c313e8ebecf041f5 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Thu, 10 Sep 2026 14:20:56 -0400 Subject: [PATCH] fix(relay): reject presence updates when Redis storage fails (#7532) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary - Reject kind:20001 presence events with `OK false` / `error: presence storage unavailable` when Redis SET or DEL fails, before publishing, local fan-out, or local-event marking. - Preserve the producer contract needed by snapshot-confirming consumers: delivered live presence must follow successful mutation of the Redis state read by snapshots. - Classify those backend rejections with the existing `IngestError` taxonomy so a presence storage outage counts as `buzz_events_rejected_total{transport="ws",reason="error"}`, not client `reason="invalid"`; genuine client-input refusals (verification failure, membership gates) stay `invalid`, and every wire message is an unchanged fixed sanitized string (review follow-up, no protocol wording change). - Add actual `handle_event` integration coverage for rejected online/offline transitions, healthy online→offline accepted/stored/fanned-out behavior, and the rejection-counter routing on storage failure with an invalid-signature control. This is standalone on main; it does not depend on the mobile implementation. Deploy this relay prerequisite before relying on #7526's snapshot-confirmation policy. Existing pubsub-failure-after-successful-storage behavior and disconnect TTL cleanup are deliberately unchanged. A storage error may be an ambiguous write outcome, not a rollback guarantee; the rejected event is not published by this handler. Clients may retry the generic `error:` rejection. Desktop's 60s heartbeat retries non-offline presence, not every explicit offline transition. ### Related issue Addresses the relay prerequisite identified in [#7526 review 5157607827](https://github.com/block/buzz/pull/7526#pullrequestreview-5157607827). Searched open presence/storage PRs; no duplicate relay storage-error rejection fix found. #7382/#7383/#7526 heads and bases are unchanged. ### Testing Exact head: `389174df29cc02d0f885c03209eff661d8bb2ec0` (+380/-13; 393 total), one commit `389174df2` on top of the reviewed `c031d6eb1` (DCO-signed; base `bfc384855889432df4a333a0edf3080f332ee169` unchanged). - PASS: `cargo fmt --all -- --check`, `cargo clippy -p buzz-relay --all-targets -- -D warnings`, `git diff --check`, `just file-size-check`, PostgreSQL discovery validation — all run at the exact final head with a clean tree before and after. - PASS: documented native `scripts/postgres-test-run.sh -p buzz-relay --lib --tests`: **89/89** actual integration tests, including the four presence cases (online/offline storage rejection, healthy online→offline, and the new rejection-classification case). Owned PostgreSQL 17/Redis on isolated loopback ports, schema plus reconciliation applied; no shared development database. - PASS: explicit `cargo test -p buzz-relay presence_storage -- --ignored --nocapture`: **4/4**, not skipped. - Full isolated relay crate suite at the final head (`cargo nextest run -p buzz-relay --lib --tests`): **1062 run: 1062 passed, 94 skipped**. The previously failing `api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo` passed in this run (1.5s); it is a known timing-sensitive main baseline failure tracked open in #7140 and untouched by this PR, so this single passing run is reported as-is and does not claim environmental clearance or close #7140. No full-suite-green claim is made beyond this run. - Mobile is untouched; #7526's existing 2090-test/format/analyze evidence remains scoped to its unchanged head. Its separate Desktop Smoke E2E (2) failure remains red; no CI retries requested. [Production-seam regression coverage](https://github.com/block/buzz/blob/389174df29cc02d0f885c03209eff661d8bb2ec0/crates/buzz-relay/src/handlers/event.rs#L1491-L1803): the metric case drives real `handle_event` traffic against a genuinely dead Redis endpoint with a seeded active PostgreSQL community and a registered presence watcher, asserts the storage rejection counts `reason="error"` while a tampered-signature control through the same dispatcher arm stays `reason="invalid"`, and re-asserts the rejected ACK, no fan-out, and no local-event marker. Counter assertions use a thread-local recorder guard held across `.await` points (the buzz-db counter-test convention) inside the per-process nextest postgres-ci lane, so no parallel test can race the counter snapshot. No UI change or screenshot. Local logs and reproducible service/gate scripts are retained under `WORK_LOGS/MOBILE_FEEDBACK_PRESENCE_20260909/relay_prerequisite/metric_correction/` in the engineering workspace. This PR is a review candidate, not merge clearance. Causal checks: restoring only the pre-fix production mutation block makes both original rejection tests fail (`OK true` instead of `false`); healthy success still passes. Reverting only the typed classification (mapping the ephemeral `Internal` arm back to `invalid`) makes the new metric regression fail with the outage counted as `[("ws","invalid",2)]` instead of `[("ws","error",1),("ws","invalid",1)]`. The unchanged mesh echo case also failed 504/200 with the main-production block restored in the prior run, supporting its separation from this change without claiming environmental clearance. Candidate source restored byte-for-byte after each mutation. Repository-wide `just ci` was not rerun; the scoped relay gates above are the new evidence. --------- Signed-off-by: Logan Johnson --- crates/buzz-relay/src/handlers/event.rs | 393 +++++++++++++++++++++++- 1 file changed, 380 insertions(+), 13 deletions(-) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 66a8ff9e7c..082522b123 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -743,8 +743,21 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc { conn.send(RelayMessage::ok(&event_id_hex, true, "")); } - Err(message) => { - reject("invalid"); + Err(e) => { + // Rejections carry the ingest taxonomy so backend failures + // count as `error` — a Redis presence-storage outage is a + // server fault, not client misbehavior — while genuine + // client-input rejections (bad signature, non-member + // sender) stay `invalid`. The ephemeral handler only emits + // fixed, sanitized message strings, so unlike the + // persistent arm below, `Internal` text is safe to forward + // verbatim. + let (message, reason) = match e { + IngestError::Rejected(message) => (message, "invalid"), + IngestError::AuthFailed(message) => (message, "auth"), + IngestError::Internal(message) => (message, "error"), + }; + reject(reason); conn.send(RelayMessage::ok(&event_id_hex, false, &message)); } } @@ -792,6 +805,12 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, state: Arc, -) -> Result<(), String> { +) -> Result<(), IngestError> { let event_clone = event.clone(); let event_id = event.id.to_hex(); let verify_result = tokio::task::spawn_blocking(move || verify_event(&event_clone)).await; match verify_result { Ok(Ok(())) => {} - Ok(Err(e)) => return Err(format!("invalid: {e}")), - Err(_) => return Err("error: internal error".to_string()), + Ok(Err(e)) => return Err(IngestError::Rejected(format!("invalid: {e}"))), + Err(_) => return Err(IngestError::Internal("error: internal error".to_string())), } // Special handling for presence events (kind:20001). @@ -829,16 +848,45 @@ async fn handle_ephemeral_event( raw }; + // Presence mutation is the inclusion contract for the live fan-out + // below: a client that observes the fanned-out event treats a later + // snapshot as reflecting it (see `synthesize_presence` in + // `api/bridge.rs`, which reads Redis). If the mutation failed we + // published nothing — so we must also fan out nothing and reject the + // ACK, or a snapshot later "confirms" stale storage over a live event + // the sender believes was delivered. if status == "offline" { - let _ = state + if let Err(e) = state .pubsub .clear_presence(&conn.tenant, &auth_pubkey) - .await; - } else { - let _ = state - .pubsub - .set_presence(&conn.tenant, &auth_pubkey, &status) - .await; + .await + { + warn!( + conn_id = %conn_id, + event_id = %event_id, + "Presence clear failed, refusing publish and fan-out: {e}" + ); + // Internal, not Rejected: a storage outage is a server + // failure, so the dispatcher must count it under the + // `error` reason, not as client-invalid input. + return Err(IngestError::Internal( + "error: presence storage unavailable".to_string(), + )); + } + } else if let Err(e) = state + .pubsub + .set_presence(&conn.tenant, &auth_pubkey, &status) + .await + { + warn!( + conn_id = %conn_id, + event_id = %event_id, + "Presence set failed, refusing publish and fan-out: {e}" + ); + // Internal for the same reason as the clear arm above. + return Err(IngestError::Internal( + "error: presence storage unavailable".to_string(), + )); } // Presence is a channel-less ephemeral event. After updating Redis @@ -848,8 +896,12 @@ async fn handle_ephemeral_event( // Check channel membership before publishing other ephemeral events. if let Some(ch_id) = super::ingest::extract_channel_id(&event) { + // Membership refusals are client-input rejections, and the shared + // gate's message text is surfaced verbatim exactly as before this + // typed classification; no behavior change on this path. super::ingest::check_channel_membership(&conn.tenant, &state, ch_id, &pubkey_bytes, None) - .await?; + .await + .map_err(IngestError::Rejected)?; // Mark as local before Redis publish to prevent double-delivery when // the event comes back through the Redis subscriber loop. @@ -1435,6 +1487,321 @@ mod tests { ); } + // PostgreSQL/Redis tests are discovered by the isolated postgres-ci lane. + mod presence_storage_postgres_tests { + use super::*; + use buzz_core::{CommunityId, TenantContext}; + use nostr::Filter; + + async fn exercise(storage_available: bool, statuses: &[&str]) { + let redis_url = std::env::var("REDIS_URL").expect("test Redis URL required"); + // Pick an unused local endpoint for a real connection failure. + let dead_socket = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let dead_port = dead_socket.local_addr().unwrap().port(); + drop(dead_socket); + let dead_url = format!("redis://127.0.0.1:{dead_port}"); + let state = fanout_access::test_state_with_redis_url(if storage_available { + &redis_url + } else { + &dead_url + }) + .await; + let pool = sqlx::PgPool::connect(&state.config.database_url) + .await + .unwrap(); + let community_uuid = Uuid::new_v4(); + let host = format!("presence-storage-{}.example", community_uuid.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(&host) + .execute(&pool) + .await + .expect("seed active community"); + let tenant = TenantContext::resolved(CommunityId::from_uuid(community_uuid), host); + let keys = Keys::generate(); + let (send_tx, mut send_rx) = mpsc::channel(10); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(10); + let conn = Arc::new(crate::connection::ConnectionState { + conn_id: Uuid::new_v4(), + tenant: tenant.clone(), + remote_addr: "127.0.0.1:1234".parse().unwrap(), + auth_state: RwLock::new(crate::connection::AuthState::Authenticated( + buzz_auth::AuthContext { + pubkey: keys.public_key(), + scopes: vec![], + channel_ids: None, + auth_method: buzz_auth::AuthMethod::Nip42, + agent_owner_pubkey: None, + }, + )), + subscriptions: Arc::new(Mutex::new(HashMap::new())), + send_tx, + ctrl_tx, + cancel: CancellationToken::new(), + backpressure_count: Arc::new(AtomicU8::new(0)), + grace_limit: 3, + }); + let watcher = Uuid::new_v4(); + let (tx, mut rx) = mpsc::channel(10); + let (ctrl, _ctrl_rx) = mpsc::channel(10); + state.conn_manager.register( + watcher, + tx, + ctrl, + None, + CancellationToken::new(), + tenant.community(), + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + ); + state.sub_registry.register_scoped( + tenant.community(), + watcher, + "presence".into(), + vec![Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16))], + None, + ); + // Online followed by offline also proves DEL removes an existing value. + for &status in statuses { + let event = EventBuilder::new(Kind::Custom(KIND_PRESENCE_UPDATE as u16), status) + .sign_with_keys(&keys) + .unwrap(); + super::super::handle_event(event.clone(), conn.clone(), state.clone()).await; + let axum::extract::ws::Message::Text(text) = send_rx.try_recv().expect("ACK") + else { + panic!("expected text ACK"); + }; + let ack: serde_json::Value = serde_json::from_str(&text).unwrap(); + assert_eq!(ack[0], "OK"); + assert_eq!(ack[1], event.id.to_hex()); + assert_eq!(ack[2], storage_available); + if storage_available { + assert_eq!(ack[3], ""); + let stored = state + .pubsub + .get_presence(&tenant, &keys.public_key()) + .await + .unwrap(); + assert_eq!(stored.as_deref(), (status != "offline").then_some(status)); + let axum::extract::ws::Message::Text(text) = + rx.try_recv().expect("live fanout") + else { + panic!("expected text event"); + }; + let frame: serde_json::Value = serde_json::from_str(&text).unwrap(); + assert_eq!(frame[0], "EVENT"); + assert_eq!(frame[2]["id"], event.id.to_hex()); + } else { + assert_eq!(ack[3], "error: presence storage unavailable"); + assert!(state + .local_event_ids + .get(&(tenant.community(), event.id.to_bytes())) + .is_none()); + } + assert!(rx.try_recv().is_err(), "no extra or rejected-event fanout"); + assert!(send_rx.try_recv().is_err(), "exactly one ACK"); + } + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_uuid) + .execute(&pool) + .await + .unwrap(); + } + + #[tokio::test] + #[ignore = "requires PostgreSQL and Redis"] + async fn rejects_online_when_presence_storage_fails() { + exercise(false, &["online"]).await; + } + + #[tokio::test] + #[ignore = "requires PostgreSQL and Redis"] + async fn rejects_offline_when_presence_storage_fails() { + exercise(false, &["offline"]).await; + } + + #[tokio::test] + #[ignore = "requires PostgreSQL and Redis"] + async fn accepts_stores_and_fans_out_online_and_offline() { + exercise(true, &["online", "offline"]).await; + } + + /// Production-seam regression for the rejection classification: a + /// Redis presence-storage outage must count as a server `error`, + /// never as client `invalid`, while a genuinely invalid event + /// through the same seam stays `invalid`. Both rejections still ACK + /// `false` with no fan-out and no local-event marker (the shared + /// storage guard is unchanged); only the counter routing is under + /// test here. + /// + /// The recorder guard form (not `metrics::with_local_recorder`, + /// whose closure cannot host an await) keeps the thread-local + /// recorder installed across the `.await` points of this + /// single-threaded test runtime — the same convention as the + /// buzz-db route-decision counter tests. The isolated postgres-ci + /// lane runs each test in its own nextest process, so no parallel + /// test can race this counter snapshot. + #[tokio::test] + #[ignore = "requires PostgreSQL"] + async fn counts_presence_storage_failure_as_error_not_invalid() { + // Pick an unused local endpoint for a real connection failure. + let dead_socket = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let dead_port = dead_socket.local_addr().unwrap().port(); + drop(dead_socket); + let dead_url = format!("redis://127.0.0.1:{dead_port}"); + let state = fanout_access::test_state_with_redis_url(&dead_url).await; + let pool = sqlx::PgPool::connect(&state.config.database_url) + .await + .unwrap(); + let community_uuid = Uuid::new_v4(); + let host = format!("presence-metrics-{}.example", community_uuid.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(&host) + .execute(&pool) + .await + .expect("seed active community"); + let tenant = TenantContext::resolved(CommunityId::from_uuid(community_uuid), host); + let keys = Keys::generate(); + let (send_tx, mut send_rx) = mpsc::channel(10); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(10); + let conn = Arc::new(crate::connection::ConnectionState { + conn_id: Uuid::new_v4(), + tenant: tenant.clone(), + remote_addr: "127.0.0.1:1234".parse().unwrap(), + auth_state: RwLock::new(crate::connection::AuthState::Authenticated( + buzz_auth::AuthContext { + pubkey: keys.public_key(), + scopes: vec![], + channel_ids: None, + auth_method: buzz_auth::AuthMethod::Nip42, + agent_owner_pubkey: None, + }, + )), + subscriptions: Arc::new(Mutex::new(HashMap::new())), + send_tx, + ctrl_tx, + cancel: CancellationToken::new(), + backpressure_count: Arc::new(AtomicU8::new(0)), + grace_limit: 3, + }); + // Same watcher registration as the ACK/fan-out cases, proving + // the storage failure still reaches no subscriber while its + // rejection is counted under `error`. + let watcher = Uuid::new_v4(); + let (tx, mut rx) = mpsc::channel(10); + let (ctrl, _ctrl_rx) = mpsc::channel(10); + state.conn_manager.register( + watcher, + tx, + ctrl, + None, + CancellationToken::new(), + tenant.community(), + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + ); + state.sub_registry.register_scoped( + tenant.community(), + watcher, + "presence".into(), + vec![Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16))], + None, + ); + + let valid = EventBuilder::new(Kind::Custom(KIND_PRESENCE_UPDATE as u16), "online") + .sign_with_keys(&keys) + .unwrap(); + // Invalid control through the same production seam: identical + // event, signature no longer verifies. The id is unchanged (it + // does not cover the signature), so ACKs are told apart by + // order and message text below. + use nostr::JsonUtil as _; + let mut json: serde_json::Value = + serde_json::from_str(&valid.as_json()).expect("parse event json"); + json["sig"] = serde_json::Value::String("0".repeat(128)); + let tampered = nostr::Event::from_json(json.to_string()).expect("parse tampered"); + + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let acks: Vec = { + let _guard = metrics::set_default_local_recorder(&recorder); + super::super::handle_event(valid.clone(), conn.clone(), state.clone()).await; + super::super::handle_event(tampered, conn.clone(), state.clone()).await; + (0..2) + .map(|_| { + let axum::extract::ws::Message::Text(text) = + send_rx.try_recv().expect("ACK") + else { + panic!("expected text ACK"); + }; + serde_json::from_str(&text).unwrap() + }) + .collect() + }; + + // Storage failure: rejected ACK, no fan-out frame, no marker. + assert_eq!(acks[0][0], "OK"); + assert_eq!(acks[0][1], valid.id.to_hex()); + assert_eq!(acks[0][2], false); + assert_eq!(acks[0][3], "error: presence storage unavailable"); + // Invalid control through the same dispatcher arm. + assert_eq!(acks[1][0], "OK"); + assert_eq!(acks[1][1], valid.id.to_hex()); + assert_eq!(acks[1][2], false); + assert!( + acks[1][3].as_str().unwrap().starts_with("invalid:"), + "tampered control must stay an invalid rejection, got: {}", + acks[1][3] + ); + assert!(rx.try_recv().is_err(), "no fan-out for either rejection"); + assert!(send_rx.try_recv().is_err(), "exactly two ACKs"); + assert!(state + .local_event_ids + .get(&(tenant.community(), valid.id.to_bytes())) + .is_none()); + + let mut rejections: Vec<(String, String, u64)> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_events_rejected_total") + .map(|(key, _, _, value)| { + let metrics_util::debugging::DebugValue::Counter(count) = value else { + panic!("buzz_events_rejected_total must be a counter"); + }; + let labels: Vec<_> = key.key().labels().collect(); + let label = |name: &str| { + labels + .iter() + .find(|l| l.key() == name) + .map(|l| l.value().to_owned()) + .unwrap_or_default() + }; + (label("transport"), label("reason"), count) + }) + .collect(); + rejections.sort(); + assert_eq!( + rejections, + vec![ + ("ws".to_owned(), "error".to_owned(), 1), + ("ws".to_owned(), "invalid".to_owned(), 1), + ], + "storage outage must count as reason=\"error\" and the tampered \ + control as reason=\"invalid\"; got {rejections:?}" + ); + + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_uuid) + .execute(&pool) + .await + .unwrap(); + } + } + mod pubsub_fanout { use std::collections::HashMap; use std::sync::atomic::AtomicU8;