diff --git a/Cargo.lock b/Cargo.lock index ebbe9def5..5aa78909a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1147,6 +1147,7 @@ dependencies = [ "dashmap", "evalexpr", "hex", + "moka", "nostr", "reqwest 0.13.3", "serde", diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index eafd376eb..69150aaad 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1758,12 +1758,13 @@ impl Db { } /// Delete a workflow only when it belongs to the provided owner. + /// Returns the deleted workflow's `channel_id`. pub async fn delete_workflow_for_owner( &self, community_id: CommunityId, id: Uuid, owner_pubkey: &[u8], - ) -> Result<()> { + ) -> Result> { workflow::delete_workflow_for_owner(&self.pool, community_id, id, owner_pubkey).await } diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index d99ac2243..2ff90280a 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -270,6 +270,9 @@ pub struct ApprovalRecord { /// Insert a new workflow record. Returns the new workflow's UUID. /// New workflows start as `active` and `enabled = TRUE`. +/// +/// NOTE: see the cache-invalidation note on [`update_workflow`]. The relay's +/// creation path is [`upsert_workflow`] via event ingest. (No current callers.) pub async fn create_workflow( pool: &PgPool, community_id: CommunityId, @@ -607,6 +610,11 @@ pub async fn prune_scheduled_workflow_fires_before( } /// Update a workflow's name, definition, and definition_hash. +/// +/// NOTE: the relay's `WorkflowEngine` caches enabled workflows per +/// `(community_id, channel_id)`; a caller mutating trigger behavior must +/// invalidate via `WorkflowEngine::invalidate_channel_workflows` or trigger +/// matching lags the change by up to the cache TTL. (No current callers.) pub async fn update_workflow( pool: &PgPool, community_id: CommunityId, @@ -638,6 +646,9 @@ pub async fn update_workflow( } /// Update a workflow's status (active -> disabled -> archived). +/// +/// NOTE: status gates trigger eligibility; see the cache-invalidation note on +/// [`update_workflow`]. (No current callers.) pub async fn update_workflow_status( pool: &PgPool, community_id: CommunityId, @@ -665,6 +676,9 @@ pub async fn update_workflow_status( } /// Enable or disable a workflow without changing its status. +/// +/// NOTE: `enabled` gates trigger eligibility; see the cache-invalidation note +/// on [`update_workflow`]. (No current callers.) pub async fn set_workflow_enabled( pool: &PgPool, community_id: CommunityId, @@ -692,6 +706,10 @@ pub async fn set_workflow_enabled( } /// Delete a workflow and all its runs/approvals (CASCADE). +/// +/// NOTE: see the cache-invalidation note on [`update_workflow`]. The relay's +/// deletion path uses [`delete_workflow_for_owner`], which returns the +/// `channel_id` needed for invalidation. (No current callers.) pub async fn delete_workflow(pool: &PgPool, community_id: CommunityId, id: Uuid) -> Result<()> { let affected = sqlx::query("DELETE FROM workflows WHERE community_id = $1 AND id = $2") .bind(community_id.as_uuid()) @@ -712,26 +730,29 @@ pub async fn delete_workflow(pool: &PgPool, community_id: CommunityId, id: Uuid) /// controlled. Keeping the owner predicate in the DELETE statement avoids a /// check-then-delete race and ensures a caller cannot delete another user's /// workflow just by learning its UUID. +/// +/// Returns the deleted workflow's `channel_id` so the caller can invalidate +/// the per-channel trigger cache without a separate lookup. pub async fn delete_workflow_for_owner( pool: &PgPool, community_id: CommunityId, id: Uuid, owner_pubkey: &[u8], -) -> Result<()> { - let affected = sqlx::query( - "DELETE FROM workflows WHERE community_id = $1 AND id = $2 AND owner_pubkey = $3", +) -> Result> { + let row = sqlx::query( + "DELETE FROM workflows WHERE community_id = $1 AND id = $2 AND owner_pubkey = $3 \ + RETURNING channel_id", ) .bind(community_id.as_uuid()) .bind(id) .bind(owner_pubkey) - .execute(pool) - .await? - .rows_affected(); + .fetch_optional(pool) + .await?; - if affected == 0 { - return Err(DbError::NotFound(format!("workflow {id}"))); + match row { + Some(row) => Ok(row.try_get("channel_id")?), + None => Err(DbError::NotFound(format!("workflow {id}"))), } - Ok(()) } // -- Workflow Run CRUD -------------------------------------------------------- diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 4ba2611b0..e3b19770b 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -756,6 +756,12 @@ async fn handle_workflow_def( other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), })?; + // Drop the trigger-path cache entry so the new/updated definition fires on + // the next matching event instead of after the cache TTL. + state + .workflow_engine + .invalidate_channel_workflows(community_id, channel_id); + // Commit the event transaction after the idempotent workflow upsert succeeds. tx.commit() .await diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 750018a5c..a58f10e2e 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -59,11 +59,19 @@ fn bounded_kind_label(kind: u32) -> String { /// fail closed. This is the cluster-wide backstop: even if a stale subscription /// survives on another node after an open->private flip, its events are not /// delivered here. +/// +/// `threaded` is an optional visibility read resolved earlier in the same +/// request (E1 phase-2, §4.8 phase-2 addendum). It is consulted only when its +/// `(community_id, channel_id)` exactly match this fan-out's — a mismatched or +/// absent bundle falls back to the fresh fail-closed lookup below, never to +/// "assume open". Membership checks stay fresh either way; the threaded value +/// only replaces the visibility SELECT. pub async fn filter_fanout_by_access( state: &AppState, community_id: CommunityId, stored_event: &StoredEvent, matches: Vec<(crate::subscription::ConnId, crate::subscription::SubId)>, + threaded: Option<&crate::state::ThreadedChannelVisibility>, ) -> Vec<(crate::subscription::ConnId, crate::subscription::SubId)> { // First enforce the receiver-side tenant label. Subscription indexes are // community-scoped, but stale/injected matches and future fan-out helpers @@ -100,10 +108,21 @@ pub async fn filter_fanout_by_access( let Some(channel_id) = stored_event.channel_id else { return matches; }; - match state - .channel_visibility_cached(community_id, channel_id) - .await - { + // Fence 3 (§4.8 phase-2): the threaded value is used only when it was + // resolved under exactly this (community_id, channel_id); anything else + // falls through to the fresh lookup. Fence 1: absence of a usable threaded + // value is never "open" — it is the same fail-closed path as before. + let visibility = match threaded { + Some(t) if t.community_id == community_id && t.channel_id == channel_id => { + Ok(t.visibility.clone()) + } + _ => { + state + .channel_visibility_cached(community_id, channel_id, None) + .await + } + }; + match visibility { Ok(v) if v != "private" => return matches, Ok(_) => {} Err(e) => { @@ -156,7 +175,7 @@ pub(crate) async fn fan_out_event_to_local_subscribers( stored: &StoredEvent, ) { let matches = state.sub_registry.fan_out_scoped(community_id, stored); - let matches = filter_fanout_by_access(state, community_id, stored, matches).await; + let matches = filter_fanout_by_access(state, community_id, stored, matches, None).await; metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64); if matches.is_empty() { return; @@ -212,7 +231,7 @@ pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pub } let matches = state.sub_registry.fan_out_scoped(community_id, &stored); - let matches = filter_fanout_by_access(state, community_id, &stored, matches).await; + let matches = filter_fanout_by_access(state, community_id, &stored, matches, None).await; metrics::counter!("buzz_multinode_fanout_total").increment(1); if matches.is_empty() { return; @@ -256,6 +275,7 @@ pub(crate) async fn dispatch_persistent_event( stored_event: &StoredEvent, kind_u32: u32, actor_pubkey_hex: &str, + threaded_visibility: Option, ) -> usize { let event_id_hex = stored_event.event.id.to_hex(); enqueue_event_created_audit( @@ -282,6 +302,7 @@ pub(crate) async fn dispatch_persistent_event( kind_u32, &actor_pubkey_hex, false, + threaded_visibility, ) .await; debug!( @@ -302,6 +323,7 @@ async fn dispatch_persistent_event_inner( kind_u32: u32, actor_pubkey_hex: &str, enqueue_audit: bool, + threaded_visibility: Option, ) -> usize { // No `crate::conformance` emit here — the spec doesn't have a // separate fan-out action. Acceptance was already recorded at the @@ -330,7 +352,14 @@ async fn dispatch_persistent_event_inner( let matches = state .sub_registry .fan_out_scoped(tenant.community(), stored_event); - let matches = filter_fanout_by_access(state, tenant.community(), stored_event, matches).await; + let matches = filter_fanout_by_access( + state, + tenant.community(), + stored_event, + matches, + threaded_visibility.as_ref(), + ) + .await; metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64); debug!( event_id = %event_id_hex, @@ -756,9 +785,14 @@ async fn handle_ephemeral_event( // Check channel membership before publishing other ephemeral events. if let Some(ch_id) = super::ingest::extract_channel_id(&event) { - if let Err(msg) = - super::ingest::check_channel_membership(&conn.tenant, &state, ch_id, &pubkey_bytes) - .await + if let Err(msg) = super::ingest::check_channel_membership( + &conn.tenant, + &state, + ch_id, + &pubkey_bytes, + None, + ) + .await { conn.send(RelayMessage::ok(event_id_hex, false, &msg)); return; @@ -1548,6 +1582,7 @@ mod tests { KIND_PRESENCE_UPDATE, &actor_hex, true, + None, ) .await; @@ -1649,6 +1684,7 @@ mod tests { KIND_PRESENCE_UPDATE, &actor_hex, true, + None, ) .await; super::super::dispatch_persistent_event_inner( @@ -1658,6 +1694,7 @@ mod tests { KIND_PRESENCE_UPDATE, &actor_hex, true, + None, ) .await; @@ -1867,6 +1904,7 @@ mod tests { buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()), &channel_event(None), matches.clone(), + None, ) .await; assert_eq!(out, matches); @@ -1889,6 +1927,7 @@ mod tests { community_id, &channel_event(Some(channel_id)), matches.clone(), + None, ) .await; assert_eq!(out, matches); @@ -1926,6 +1965,7 @@ mod tests { community_id, &channel_event(Some(channel_id)), matches, + None, ) .await; assert_eq!(out, vec![(member, "m".to_string())]); @@ -1964,6 +2004,7 @@ mod tests { buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()), &stored, matches, + None, ) .await; @@ -1971,6 +2012,112 @@ mod tests { // unauthenticated connection are both dropped. assert_eq!(out, vec![(author_conn, "a".to_string())]); } + + // -- E1 phase-2: threaded-visibility fences (§4.8 phase-2 addendum) -- + + /// Fence 3: a threaded visibility resolved under a different + /// (community, channel) must be ignored — the filter falls back to + /// the fresh fail-closed lookup, not the threaded value. + #[tokio::test] + async fn threaded_visibility_mismatch_falls_back_to_fresh_lookup() { + let state = test_state().await; + let channel_id = Uuid::new_v4(); + let community_id = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()); + // Fresh lookup path resolves private via the fail-safe cache. + state + .channel_visibility_cache + .insert((community_id, channel_id), "private".to_string()); + + // Threaded value says "open" but for a DIFFERENT channel. + let threaded = crate::state::ThreadedChannelVisibility { + community_id, + channel_id: Uuid::new_v4(), + visibility: "open".to_string(), + }; + + // Unauthenticated conn: kept on open, dropped on private. + let conn = register_conn(&state, None); + let matches = vec![(conn, "s".to_string())]; + let out = filter_fanout_by_access( + &state, + community_id, + &channel_event(Some(channel_id)), + matches, + Some(&threaded), + ) + .await; + assert!( + out.is_empty(), + "mismatched threaded visibility must not be consulted; \ + fresh lookup says private → unauthenticated conn dropped" + ); + } + + /// Matching threaded `private` gates recipients without a DB read, + /// identically to the fresh-lookup private path. + #[tokio::test] + async fn threaded_visibility_private_filters_members_only() { + let state = test_state().await; + let channel_id = Uuid::new_v4(); + let community_id = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()); + + let member_pk = vec![1u8; 32]; + let non_member_pk = vec![2u8; 32]; + state + .membership_cache + .insert((community_id, channel_id, member_pk.clone()), true); + state + .membership_cache + .insert((community_id, channel_id, non_member_pk.clone()), false); + + let threaded = crate::state::ThreadedChannelVisibility { + community_id, + channel_id, + visibility: "private".to_string(), + }; + + let member = register_conn(&state, Some(member_pk)); + let non_member = register_conn(&state, Some(non_member_pk)); + let matches = vec![(member, "m".to_string()), (non_member, "n".to_string())]; + let out = filter_fanout_by_access( + &state, + community_id, + &channel_event(Some(channel_id)), + matches, + Some(&threaded), + ) + .await; + assert_eq!(out, vec![(member, "m".to_string())]); + } + + /// Matching threaded `open` passes recipients through with no + /// visibility SELECT (no visibility cache entry exists and the lazy + /// PG pool in `test_state` would error a fresh lookup → fail closed; + /// passing through proves the threaded value was used). + #[tokio::test] + async fn threaded_visibility_open_passes_through() { + let state = test_state().await; + let channel_id = Uuid::new_v4(); + let community_id = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()); + + let threaded = crate::state::ThreadedChannelVisibility { + community_id, + channel_id, + visibility: "open".to_string(), + }; + + let conn = register_conn(&state, None); + let matches = vec![(conn, "s".to_string())]; + let out = filter_fanout_by_access( + &state, + community_id, + &channel_event(Some(channel_id)), + matches.clone(), + Some(&threaded), + ) + .await; + assert_eq!(out, matches); + } } // --------------------------------------------------------------------- @@ -2099,7 +2246,7 @@ mod tests { let stored = StoredEvent::new(presence, None); let matches = vec![(a_socket, "presence".to_string())]; - let out = filter_fanout_by_access(&state, community_b, &stored, matches).await; + let out = filter_fanout_by_access(&state, community_b, &stored, matches, None).await; // Correct behavior: A-socket dropped because its tenant != B. assert!( diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 8412eec6f..5c0c188a7 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -413,12 +413,17 @@ pub(crate) fn requires_h_channel_scope(kind: u32) -> bool { /// Check channel membership: member OR open-visibility channel. /// +/// `channel` is the request's already-fetched channel row, when the caller has +/// one (E1 within-request threading; correctness ruling §4.8). Callers without +/// a row pass `None` and the open-visibility fallback reads the DB directly. +/// /// Returns `Ok(())` if allowed, `Err(reason)` if denied. pub(crate) async fn check_channel_membership( tenant: &TenantContext, state: &AppState, ch_id: Uuid, pubkey_bytes: &[u8], + channel: Option<&buzz_db::channel::ChannelRecord>, ) -> Result<(), String> { match state .is_member_cached(tenant.community(), ch_id, pubkey_bytes) @@ -429,12 +434,15 @@ pub(crate) async fn check_channel_membership( Err(e) => return Err(format!("error: database error: {e}")), } // Not a member — check if channel is open. - let is_open = state - .db - .get_channel(tenant.community(), ch_id) - .await - .map(|ch| ch.visibility == "open") - .unwrap_or(false); + let is_open = match channel { + Some(ch) => ch.visibility == "open", + None => state + .db + .get_channel(tenant.community(), ch_id) + .await + .map(|ch| ch.visibility == "open") + .unwrap_or(false), + }; if is_open { Ok(()) } else { @@ -1411,6 +1419,36 @@ async fn ingest_event_inner( } let pubkey_bytes = auth.pubkey().to_bytes().to_vec(); + // E1 (§4.8): fetch the community-scoped channel row once per request and + // thread it through the gates below (membership open-fallback, archived + // check, join visibility) instead of re-SELECTing it at each. `None` when + // the event is global or the channel doesn't exist yet (kind:9007 creates + // it later in this request); each gate keeps its existing missing-row + // behavior. + let channel_row = match channel_id { + Some(ch_id) => state.db.get_channel(tenant.community(), ch_id).await.ok(), + None => None, + }; + // E1 phase-2 (§4.8 phase-2 addendum): resolve the fan-out visibility once, + // here, through the same `channel_visibility_cached` gate fan-out uses + // (fence 2: cached `private` wins over the prefetched row; a `private` + // read still populates the cache). The value travels to fan-out bundled + // with the (community, channel) it was resolved under (fence 3). When the + // row is missing (global event, kind:9007 pre-create) this is `None` and + // fan-out performs its own fresh fail-closed lookup — `None` is never + // "assume open" (fence 1). + let threaded_visibility = match (channel_id, &channel_row) { + (Some(ch_id), Some(row)) => state + .channel_visibility_cached(tenant.community(), ch_id, Some(row)) + .await + .ok() + .map(|visibility| crate::state::ThreadedChannelVisibility { + community_id: tenant.community(), + channel_id: ch_id, + visibility, + }), + _ => None, + }; if let Some(ch_id) = channel_id { // kind:9021 (join) doesn't require prior membership. // kind:9007 (create) — channel doesn't exist yet; creator becomes owner in step 16. @@ -1434,7 +1472,9 @@ async fn ingest_event_inner( // at `check_channel_membership`'s `is_member_cached(tenant // .community(), …)` call (see crates/buzz-relay/src/handlers // /ingest.rs:424). - let auth_result = check_channel_membership(tenant, state, ch_id, &pubkey_bytes).await; + let auth_result = + check_channel_membership(tenant, state, ch_id, &pubkey_bytes, channel_row.as_ref()) + .await; let claimed = claimed_community_from_event(&event); let verdict = if auth_result.is_ok() { Verdict::Allow @@ -1574,7 +1614,7 @@ async fn ingest_event_inner( .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; } - if let Some(ch_id) = channel_id { + if channel_id.is_some() { // Allow kind:9002 with archived=false (unarchive operation) let is_unarchive = kind_u32 == KIND_NIP29_EDIT_METADATA && event.tags.iter().any(|t| { @@ -1583,7 +1623,7 @@ async fn ingest_event_inner( }); if !is_unarchive { - if let Ok(channel) = state.db.get_channel(tenant.community(), ch_id).await { + if let Some(channel) = &channel_row { if channel.archived_at.is_some() { return Err(IngestError::Rejected("invalid: channel is archived".into())); } @@ -1740,14 +1780,14 @@ async fn ingest_event_inner( "invalid: join request must include an h tag".into(), )); } - if let Some(ch_id) = channel_id { - match state.db.get_channel(tenant.community(), ch_id).await { - Ok(ch) if ch.visibility == "private" => { + if channel_id.is_some() { + match &channel_row { + Some(ch) if ch.visibility == "private" => { return Err(IngestError::Rejected( "restricted: channel is private".into(), )); } - Err(_) => { + None => { return Err(IngestError::Rejected("invalid: channel not found".into())); } _ => {} // open — OK @@ -1951,7 +1991,15 @@ async fn ingest_event_inner( } }; emit(tracer, action, state_for_request(tenant, auth.pubkey())); - dispatch_persistent_event(tenant, state, &stored_event, kind_u32, &pubkey_hex).await; + dispatch_persistent_event( + tenant, + state, + &stored_event, + kind_u32, + &pubkey_hex, + threaded_visibility.clone(), + ) + .await; info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline"); return Ok(IngestResult { @@ -2077,7 +2125,15 @@ async fn ingest_event_inner( }; emit(tracer, action, state_for_request(tenant, auth.pubkey())); } - dispatch_persistent_event(tenant, state, &stored_event, kind_u32, &pubkey_hex).await; + dispatch_persistent_event( + tenant, + state, + &stored_event, + kind_u32, + &pubkey_hex, + threaded_visibility.clone(), + ) + .await; info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline"); diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 550ede9a2..c5485721b 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -810,7 +810,7 @@ async fn emit_addressable_discovery_event( .await?; if was_inserted { let kind_u32 = event_kind_u32(&stored.event); - dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex).await; + dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await; } Ok(()) } @@ -1826,11 +1826,16 @@ async fn handle_a_tag_deletion( buzz_core::kind::KIND_WORKFLOW_DEF => { // Try UUID first (workflow_id); fall back to name-based lookup. if let Ok(wf_id) = uuid::Uuid::parse_str(d_tag) { - state + let channel_id = state .db .delete_workflow_for_owner(tenant.community(), wf_id, &actor_bytes) .await .map_err(|e| anyhow::anyhow!("failed to delete workflow {wf_id}: {e}"))?; + if let Some(channel_id) = channel_id { + state + .workflow_engine + .invalidate_channel_workflows(tenant.community(), channel_id); + } tracing::info!(workflow_id = %wf_id, "Workflow deleted via NIP-09 a-tag (UUID)"); } else { // Name-based lookup @@ -1840,13 +1845,18 @@ async fn handle_a_tag_deletion( .await { Ok(Some(wf)) => { - state + let channel_id = state .db .delete_workflow_for_owner(tenant.community(), wf.id, &actor_bytes) .await .map_err(|e| { anyhow::anyhow!("failed to delete workflow {}: {e}", wf.id) })?; + if let Some(channel_id) = channel_id { + state + .workflow_engine + .invalidate_channel_workflows(tenant.community(), channel_id); + } tracing::info!(workflow_id = %wf.id, name = d_tag, "Workflow deleted via NIP-09 a-tag (name)"); } Ok(None) => { @@ -2552,6 +2562,7 @@ pub async fn publish_nip43_membership_list( &stored, KIND_NIP43_MEMBERSHIP_LIST, &relay_pubkey_hex, + None, ) .await; } @@ -2727,6 +2738,7 @@ pub async fn publish_nipia_archival_list( &stored, KIND_IA_ARCHIVED_LIST, &relay_pubkey_hex, + None, ) .await; } @@ -2813,6 +2825,7 @@ pub async fn publish_dm_visibility_snapshot( &stored, KIND_DM_VISIBILITY, &relay_pubkey_hex, + None, ) .await; } @@ -2876,7 +2889,7 @@ async fn publish_nipia_delta( return Ok(()); } - dispatch_persistent_event(tenant, state, &stored, kind, &relay_pubkey_hex).await; + dispatch_persistent_event(tenant, state, &stored, kind, &relay_pubkey_hex, None).await; info!( target = %target_pubkey_hex, diff --git a/crates/buzz-relay/src/mesh_status_publisher.rs b/crates/buzz-relay/src/mesh_status_publisher.rs index 1194777b5..4285e7221 100644 --- a/crates/buzz-relay/src/mesh_status_publisher.rs +++ b/crates/buzz-relay/src/mesh_status_publisher.rs @@ -235,6 +235,7 @@ pub async fn publish_mesh_status( &stored, KIND_MESH_LLM_RELAY_STATUS, &relay_pubkey_hex, + None, ) .await; } diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 05dae1d39..69d1dc7cb 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -620,10 +620,18 @@ impl AppState { /// only `private` keeps the cache fail-safe: the worst stale entry is an /// over-restrictive `private` (drops non-members on a now-open channel for /// <=10s), never a leak. + /// + /// `prefetched` lets a caller that already holds the channel row for this + /// request (ingest's once-per-request fetch, E1 §4.8) reuse it instead of + /// re-SELECTing. The gate is unchanged: a cached `private` still wins over + /// the prefetched row (the cache is fail-safe by design), and a `private` + /// read from the row still populates the cache. With `Some(row)` this + /// method performs no DB I/O and cannot error. pub async fn channel_visibility_cached( &self, community_id: CommunityId, channel_id: Uuid, + prefetched: Option<&buzz_db::channel::ChannelRecord>, ) -> Result { if let Some(cached) = self .channel_visibility_cache @@ -631,11 +639,15 @@ impl AppState { { return Ok(cached); } - let visibility = self - .db - .get_channel(community_id, channel_id) - .await? - .visibility; + let visibility = match prefetched { + Some(row) => row.visibility.clone(), + None => { + self.db + .get_channel(community_id, channel_id) + .await? + .visibility + } + }; if visibility == "private" { self.channel_visibility_cache .insert((community_id, channel_id), visibility.clone()); @@ -644,6 +656,25 @@ impl AppState { } } +/// A channel-visibility read resolved at ingest and threaded through to +/// fan-out within the same request (E1 phase-2, §4.8 phase-2 addendum). +/// +/// The community and channel ids the visibility was resolved under travel +/// with the value so it can never be consulted for a different channel or +/// community's fan-out (channel UUIDs collide across communities — +/// `Inv_LabelPropagation`). Consumers must treat a missing/mismatched bundle +/// as "no threaded visibility" and fall back to a fresh fail-closed lookup — +/// never as "assume open". +#[derive(Debug, Clone)] +pub struct ThreadedChannelVisibility { + /// Community the visibility was resolved under (server-resolved tenant). + pub community_id: CommunityId, + /// Channel the visibility was resolved for. + pub channel_id: Uuid, + /// The visibility string read at ingest (`"open"` / `"private"` / ...). + pub visibility: String, +} + /// Handle for graceful audit worker shutdown. /// /// Signals the worker to stop accepting new entries, drain its buffer, diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 4fc5d3afd..857c14626 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -191,6 +191,7 @@ impl ActionSink for RelayActionSink { &stored_event, kind_u32, &author_pubkey_hex, + None, ) .await; } diff --git a/crates/buzz-workflow/Cargo.toml b/crates/buzz-workflow/Cargo.toml index 4e985f9f2..6e213c505 100644 --- a/crates/buzz-workflow/Cargo.toml +++ b/crates/buzz-workflow/Cargo.toml @@ -15,6 +15,7 @@ serde = { workspace = true } serde_json = { workspace = true } serde_yaml = { workspace = true } dashmap = { workspace = true } +moka = { workspace = true } evalexpr = "11" cron = "0.16" uuid = { workspace = true } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 2befec624..93581225e 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -87,6 +87,21 @@ pub struct WorkflowEngine { /// Action sink for executing side-effects (SendMessage, etc.). /// Late-initialized via [`set_action_sink`] after `AppState` construction. pub(crate) action_sink: OnceLock>, + /// Short-TTL cache for the per-event enabled-workflow lookup, keyed + /// `(community_id, channel_id)`. Most channels have no workflows, so this + /// removes one SELECT from nearly every ingested event. + /// + /// Consistency: the relay invalidates this cache on its own pod at the two + /// workflow mutation sites (command upsert, NIP-09 deletion). There is + /// deliberately no cross-pod invalidation — workflow triggering is not an + /// access-control fence, so the worst case on another pod is a just-deleted + /// workflow firing (or a just-created one missing events) for up to the TTL. + /// The same TTL also bounds the same-pod look-aside race (a stale fill + /// landing just after an invalidation). Workflow mutations are rare; the + /// 10s window matches the relay's other moka caches (see `AppState` in + /// `buzz-relay`). + pub(crate) workflow_cache: + moka::sync::Cache<(CommunityId, Uuid), Arc>>, } impl WorkflowEngine { @@ -100,9 +115,23 @@ impl WorkflowEngine { run_semaphore, last_fired: DashMap::new(), action_sink: OnceLock::new(), + workflow_cache: moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(std::time::Duration::from_secs(10)) + .build(), } } + /// Drop the cached enabled-workflow list for a channel. + /// + /// Must be called after any write to a workflow's trigger eligibility or + /// channel binding (currently the relay's command upsert and NIP-09 + /// deletion paths) so same-pod trigger matching sees the change + /// immediately instead of after the cache TTL. + pub fn invalidate_channel_workflows(&self, community_id: CommunityId, channel_id: Uuid) { + self.workflow_cache.invalidate(&(community_id, channel_id)); + } + /// Set the action sink. Called once after `AppState` construction. /// /// # Panics @@ -265,11 +294,20 @@ impl WorkflowEngine { return Ok(()); } - let workflows = self - .db - .list_enabled_channel_workflows(community_id, channel_id) - .await - .map_err(WorkflowError::from)?; + let cache_key = (community_id, channel_id); + let workflows = match self.workflow_cache.get(&cache_key) { + Some(cached) => cached, + None => { + let fresh = Arc::new( + self.db + .list_enabled_channel_workflows(community_id, channel_id) + .await + .map_err(WorkflowError::from)?, + ); + self.workflow_cache.insert(cache_key, Arc::clone(&fresh)); + fresh + } + }; if workflows.is_empty() { return Ok(()); @@ -285,7 +323,7 @@ impl WorkflowEngine { } }; - for workflow in &workflows { + for workflow in workflows.iter() { let def: WorkflowDef = match serde_json::from_value(workflow.definition.clone()) { Ok(d) => d, Err(e) => { diff --git a/deny.toml b/deny.toml index 149b39bdf..2f54aa184 100644 --- a/deny.toml +++ b/deny.toml @@ -6,6 +6,15 @@ ignore = [ # paste 1.0.15 — unmaintained. Transitive dep: mesh-llm → iroh → netlink-* → paste. # No safe upgrade available; tracked for upstream (iroh/netlink) replacement. { id = "RUSTSEC-2024-0436", reason = "transitive dep via mesh-llm → iroh → netlink; no upstream fix available" }, + # quick-xml < 0.41: quadratic runtime on duplicate-attribute check (0194) and + # unbounded namespace-declaration allocation in NsReader (0195). Both DoS-class, + # requiring attacker-controlled XML. Our two locked versions only parse trusted + # input: 0.38.4 (rust-s3/aws-creds — responses from our own S3/MinIO endpoint) + # and 0.39.4 (mesh-llm → iroh → netdev → plist — local macOS system plists). + # Patched release (>= 0.41.0) is unreachable until rust-s3 and plist/netdev bump; + # remove these when upstream catches up. + { id = "RUSTSEC-2026-0194", reason = "transitive via rust-s3 and mesh-llm→plist; trusted-input XML only; no upstream fix available yet" }, + { id = "RUSTSEC-2026-0195", reason = "transitive via rust-s3 and mesh-llm→plist; trusted-input XML only; no upstream fix available yet" }, ] [licenses]