diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 12b673df36..db1315ab40 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -2224,6 +2224,95 @@ pub(crate) mod tests { } } + /// An ordinary REQ that `handle_req` itself refuses, before it claims its + /// ID, still replaces the old holder (NIP-01): a live subscription or an + /// artifact sync must not keep using an ID the client was told is closed. + /// Covers the auth, scope, channel-count and database-error refusals. + /// Mutation: send any of them as a bare `CLOSED` → RED. + #[tokio::test] + async fn in_handler_req_refusal_retires_the_old_holder() { + use buzz_pubsub::EventTopic; + let mut scoped = auth_context(); + scoped.scopes = vec![buzz_auth::Scope::ChannelsRead]; + let channel = uuid::Uuid::new_v4(); + let too_many: Vec = (0..=handlers::req::MAX_EXPLICIT_CHANNEL_VALUES) + .map(|_| uuid::Uuid::new_v4().to_string()) + .collect(); + let cases = [ + ( + AuthState::Failed, + serde_json::json!({"kinds": [1]}), + "auth-required: not authenticated", + ), + ( + AuthState::Authenticated(scoped), + serde_json::json!({"kinds": [1]}), + "restricted: insufficient scope", + ), + ( + authenticated_state(), + serde_json::json!({"kinds": [1], "#h": too_many}), + "restricted: too many explicit channels", + ), + ( + authenticated_state(), + serde_json::json!({"search": "x"}), + "error: database error", + ), + ]; + for ((auth, filter, reason), sync_holder) in cases + .into_iter() + .flat_map(|case| [(case.clone(), false), (case, true)]) + { + let case = format!("{reason}, sync holder: {sync_holder}"); + let state = + crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none") + .await; + let (conn, mut rx) = test_conn_with_auth(auth); + let topic = EventTopic::Channel(channel); + handlers::req::claim_live_subscription( + "s", + &[nostr::Filter::new().kind(Kind::TextNote)], + Some(&[channel]), + &conn, + &state, + ) + .await + .expect("claim the old holder"); + if sync_holder { + state.artifact_syncs.insert_idle(conn.conn_id, "s"); + } + + let filter = serde_json::from_value(filter).expect("filter"); + handlers::req::handle_req( + "s".into(), + vec![filter], + vec![None], + Arc::clone(&conn), + Arc::clone(&state), + ) + .await; + + assert!(conn.subscriptions.lock().await.is_empty(), "{case}"); + assert!( + state.sub_registry.get_filters(conn.conn_id, "s").is_none(), + "{case}" + ); + assert!(!state.artifact_syncs.contains(conn.conn_id, "s"), "{case}"); + assert_eq!( + state.pubsub.topic_refcount(&conn.tenant, topic).await, + 0, + "{case}" + ); + let mut frame = read_frame(&mut rx); + if frame[0] == "NOTICE" { + frame = read_frame(&mut rx); + } + assert_eq!(frame, serde_json::json!(["CLOSED", "s", reason]), "{case}"); + assert!(rx.try_recv().is_err(), "{case}: nothing follows CLOSED"); + } + } + /// A pre-handler refusal whose `CLOSED` cannot be queued has already /// retired the old holder, so it cancels the connection: the client /// learns the ID is gone from the disconnect. The queue is full but diff --git a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs index f33b709e56..347585fd63 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs @@ -521,3 +521,93 @@ async fn close_during_a_live_head_read_sends_nothing_more() { .unwrap(); no_frame(&mut rx).await; } + +/// Run an ordinary `REQ "s"` through the real handler. +async fn ordinary_req(f: &Fixture, conn: &Arc, filters: Vec) { + let filters: Vec = filters + .into_iter() + .map(|filter| serde_json::from_value(filter).expect("filter")) + .collect(); + let before_ids = vec![None; filters.len()]; + super::req::handle_req( + "s".into(), + filters, + before_ids, + Arc::clone(conn), + Arc::clone(&f.state), + ) + .await; +} + +/// An ordinary `REQ` that replaces a live sync or a live subscription with +/// the same ID retires it, also when `handle_req` refuses the new request +/// (mixed search) or answers it once (huddle liveness) without registering +/// it. A later change in the channel then sends nothing under that ID. +/// Mutation: refuse mixed search with a bare `CLOSED`, or answer liveness +/// without claiming the ID → RED. +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn refused_or_one_shot_replacement_retires_the_old_holder() { + let f = fixture().await; + let owner = Keys::generate(); + let home = create_channel(&f.state, &f.tenant, &owner, &[]).await; + let mixed = vec![ + json!({"search": "x", "#h": [home]}), + json!({"kinds": [45010], "#h": [home]}), + ]; + let liveness = vec![json!({"kinds": [buzz_core::kind::KIND_HUDDLE_LIVENESS], "#h": [home]})]; + for (sync_holder, replacement, terminal) in [ + ( + true, + &mixed, + json!([ + "CLOSED", + "s", + "error: mixed search and non-search filters not supported" + ]), + ), + ( + false, + &mixed, + json!([ + "CLOSED", + "s", + "error: mixed search and non-search filters not supported" + ]), + ), + (true, &liveness, json!(["EOSE", "s"])), + (false, &liveness, json!(["EOSE", "s"])), + ] { + let case = format!("sync holder: {sync_holder}, replacement: {terminal}"); + let (conn, mut rx) = connect(&f, &owner); + if sync_holder { + start_sync(&f, &conn, sync_filter(&[home])).await.unwrap(); + } else { + ordinary_req( + &f, + &conn, + vec![json!({"kinds": [45010], "#h": [home], "limit": 0})], + ) + .await; + } + assert!(initial_set(&mut rx).await.is_empty(), "{case}"); + + ordinary_req(&f, &conn, replacement.clone()).await; + assert_eq!(frame(&mut rx).await, terminal, "{case}"); + assert!(conn.subscriptions.lock().await.is_empty(), "{case}"); + assert!( + !f.state.artifact_syncs.contains(conn.conn_id, "s"), + "{case}" + ); + assert!( + f.state + .sub_registry + .get_filters(conn.conn_id, "s") + .is_none(), + "{case}" + ); + + revision(&f, &owner, home, Uuid::new_v4(), "create", None).await; + no_frame(&mut rx).await; + } +} diff --git a/crates/buzz-relay/src/handlers/close.rs b/crates/buzz-relay/src/handlers/close.rs index 8593b24f96..053442e08c 100644 --- a/crates/buzz-relay/src/handlers/close.rs +++ b/crates/buzz-relay/src/handlers/close.rs @@ -28,7 +28,7 @@ pub async fn handle_close(sub_id: String, conn: Arc, state: Arc /// Refuse a `REQ` or sync `REQ` with `CLOSED reason`. A `REQ` replaces /// whatever held its ID (NIP-01), and `CLOSED` tells the client the ID is /// closed, so the old holder is retired before the frame, under the lifecycle -/// lock. Every refusal before a request's handler runs goes through here. If +/// lock. Every refusal before a request claims its ID goes through here. If /// the frame cannot be queued, the connection is cancelled: the client would /// otherwise never learn that the old holder is gone. pub(crate) async fn refuse(sub_id: &str, reason: &str, conn: &ConnectionState, state: &AppState) { diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 9b2484324c..9b5785f13b 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -65,10 +65,8 @@ pub async fn handle_req( AuthState::Authenticated(ctx) => { if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) { conn.send(RelayMessage::notice("restricted: insufficient scope")); - conn.send(RelayMessage::closed( - &sub_id, - "restricted: insufficient scope", - )); + super::close::refuse(&sub_id, "restricted: insufficient scope", &conn, &state) + .await; return; } @@ -89,10 +87,8 @@ pub async fn handle_req( conn.send(RelayMessage::notice( "auth-required: authenticate before subscribing", )); - conn.send(RelayMessage::closed( - &sub_id, - "auth-required: not authenticated", - )); + super::close::refuse(&sub_id, "auth-required: not authenticated", &conn, &state) + .await; return; } } @@ -102,10 +98,13 @@ pub async fn handle_req( let requested_channel_ids = match extract_channel_ids_from_filters_limited(&filters) { Ok(ids) => ids, Err(()) => { - conn.send(RelayMessage::closed( + super::close::refuse( &sub_id, "restricted: too many explicit channels", - )); + &conn, + &state, + ) + .await; return; } }; @@ -122,7 +121,7 @@ pub async fn handle_req( Ok(ids) => ids, Err(e) => { warn!(conn_id = %conn_id, "Failed to get accessible channels: {e}"); - conn.send(RelayMessage::closed(&sub_id, "error: database error")); + super::close::refuse(&sub_id, "error: database error", &conn, &state).await; return; } } @@ -146,7 +145,7 @@ pub async fn handle_req( Ok(ids) => ids, Err(e) => { warn!(conn_id = %conn_id, "Failed to get readable named-only channels: {e}"); - conn.send(RelayMessage::closed(&sub_id, "error: database error")); + super::close::refuse(&sub_id, "error: database error", &conn, &state).await; return; } }; @@ -185,7 +184,7 @@ pub async fn handle_req( Ok(confirmed) => confirmed, Err(e) => { warn!(conn_id = %conn_id, "Channel membership confirmation failed: {e}"); - conn.send(RelayMessage::closed(&sub_id, "error: database error")); + super::close::refuse(&sub_id, "error: database error", &conn, &state).await; return; } }; @@ -204,7 +203,7 @@ pub async fn handle_req( .await { warn!(conn_id = %conn_id, "Failed to classify repaired channels: {e}"); - conn.send(RelayMessage::closed(&sub_id, "error: database error")); + super::close::refuse(&sub_id, "error: database error", &conn, &state).await; return; } } @@ -229,10 +228,7 @@ pub async fn handle_req( .as_ref() .is_some_and(|authorized| authorized.is_empty()) { - conn.send(RelayMessage::closed( - &sub_id, - "restricted: not a channel member", - )); + super::close::refuse(&sub_id, "restricted: not a channel member", &conn, &state).await; return; } @@ -251,17 +247,20 @@ pub async fn handle_req( Err(crate::nip_fi_gate::SessionExpired) => { // Fix 4: [FI-TRACE-DENIAL-ORACLE] gate is off_mode when no assertion // exists, so SessionExpired here always implies an active FI session. - conn.send(RelayMessage::closed( - &sub_id, - "restricted: authorization denied", - )); + super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state) + .await; return; } }; + // One-shot like search: accepting it replaces the ID's old holder. + let Some(owner) = claim_search_subscription(&sub_id, &conn, &state).await else { + return; + }; unless_cancelled( &conn, handle_huddle_liveness_req( &sub_id, + owner, &filters, authorized_requested_channels.as_deref().unwrap_or_default(), &conn, @@ -269,6 +268,7 @@ pub async fn handle_req( ), ) .await; + super::close::close_if_owner(&sub_id, owner, None, &conn, &state).await; return; } @@ -283,24 +283,33 @@ pub async fn handle_req( if channel_id.is_none() { let authed_pubkey_hex = hex::encode(&pubkey_bytes); if !p_gated_filters_authorized(&filters, &authed_pubkey_hex) { - conn.send(RelayMessage::closed( + super::close::refuse( &sub_id, "restricted: p-gated events require #p matching your pubkey", - )); + &conn, + &state, + ) + .await; return; } if !engram_filters_authorized(&filters, &authed_pubkey_hex) { - conn.send(RelayMessage::closed( + super::close::refuse( &sub_id, "restricted: agent-engram reads require authors=[self] or #p=[self]", - )); + &conn, + &state, + ) + .await; return; } if !author_only_filters_authorized(&filters, &authed_pubkey_hex) { - conn.send(RelayMessage::closed( + super::close::refuse( &sub_id, "restricted: author-only kinds require authors=[self]", - )); + &conn, + &state, + ) + .await; return; } } @@ -312,10 +321,13 @@ pub async fn handle_req( let has_search = filters.iter().any(|f| f.search.is_some()); if has_search { if filters.iter().any(|f| f.search.is_none()) { - conn.send(RelayMessage::closed( + super::close::refuse( &sub_id, "error: mixed search and non-search filters not supported", - )); + &conn, + &state, + ) + .await; return; } // IMPORTANT 6: acquire a REQ effect permit before the search query and @@ -327,10 +339,8 @@ pub async fn handle_req( Ok(permit) => permit, Err(crate::nip_fi_gate::SessionExpired) => { // Fix 4: [FI-TRACE-DENIAL-ORACLE] - conn.send(RelayMessage::closed( - &sub_id, - "restricted: authorization denied", - )); + super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state) + .await; return; } }; @@ -376,10 +386,7 @@ pub async fn handle_req( Ok(permit) => permit, Err(crate::nip_fi_gate::SessionExpired) => { // Fix 4: [FI-TRACE-DENIAL-ORACLE] - conn.send(RelayMessage::closed( - &sub_id, - "restricted: authorization denied", - )); + super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state).await; return; } }; @@ -1413,16 +1420,21 @@ fn huddle_liveness_session_ids(filters: &[Filter]) -> Vec { async fn handle_huddle_liveness_req( sub_id: &str, + owner: u64, filters: &[Filter], parent_channel_ids: &[uuid::Uuid], conn: &ConnectionState, state: &AppState, ) { if parent_channel_ids.is_empty() { - conn.send(RelayMessage::closed( + super::close::close_if_owner( sub_id, - "restricted: huddle liveness requires an authorized #h channel", - )); + owner, + Some("restricted: huddle liveness requires an authorized #h channel"), + conn, + state, + ) + .await; return; } @@ -1439,7 +1451,8 @@ async fn handle_huddle_liveness_req( Ok(links) => links, Err(error) => { warn!("Huddle liveness linkage batch failed: {error}"); - conn.send(RelayMessage::closed(sub_id, "error: database error")); + super::close::close_if_owner(sub_id, owner, Some("error: database error"), conn, state) + .await; return; } }; @@ -1457,7 +1470,14 @@ async fn handle_huddle_liveness_req( Ok(_) => continue, Err(error) => { warn!(session_id = %session_id, "Huddle liveness lease lookup failed: {error}"); - conn.send(RelayMessage::closed(sub_id, "error: liveness unavailable")); + super::close::close_if_owner( + sub_id, + owner, + Some("error: liveness unavailable"), + conn, + state, + ) + .await; return; } } @@ -1495,7 +1515,14 @@ async fn handle_huddle_liveness_req( Ok(event) => event, Err(error) => { warn!(session_id = %session_id, "Huddle liveness signing failed: {error}"); - conn.send(RelayMessage::closed(sub_id, "error: signing failed")); + super::close::close_if_owner( + sub_id, + owner, + Some("error: signing failed"), + conn, + state, + ) + .await; return; } }; @@ -1566,10 +1593,10 @@ pub(crate) async fn claim_live_subscription( Some(owner) } -/// Claim `sub_id` for a one-shot search REQ under the lifecycle lock. -/// Accepting it retires whatever holds the ID (NIP-01 replacement), even -/// though search never registers for fan-out. `None` once the connection is -/// closing. +/// Claim `sub_id` for a one-shot search or huddle-liveness REQ under the +/// lifecycle lock. Accepting it retires whatever holds the ID (NIP-01 +/// replacement), even though neither registers for fan-out. `None` once the +/// connection is closing. async fn claim_search_subscription( sub_id: &str, conn: &ConnectionState,