E1+E3: reduce relay ingest/fan-out DB round trips; ack p99 −7–16%, fd p99 −6–28%, p999 tails −29–53% vs PR #1453 tip (#1454)

Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
Co-authored-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: Tyler Longwell <tlongwell@block.xyz>

Cargo.lock

@@ -1147,6 +1147,7 @@ dependencies = [
  "dashmap",
  "evalexpr",
  "hex",
+ "moka",
  "nostr",
  "reqwest 0.13.3",
  "serde",

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<Option<Uuid>> {
         workflow::delete_workflow_for_owner(&self.pool, community_id, id, owner_pubkey).await
     }
 

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<Option<Uuid>> {
+    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 --------------------------------------------------------

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

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<AppState>, 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<crate::state::ThreadedChannelVisibility>,
 ) -> 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<crate::state::ThreadedChannelVisibility>,
 ) -> 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!(

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");
 

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,

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;
     }

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<String, buzz_db::DbError> {
         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,

crates/buzz-relay/src/workflow_sink.rs

@@ -191,6 +191,7 @@ impl ActionSink for RelayActionSink {
                     &stored_event,
                     kind_u32,
                     &author_pubkey_hex,
+                    None,
                 )
                 .await;
             }

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 }

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<Arc<dyn ActionSink>>,
+    /// 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<Vec<buzz_db::workflow::WorkflowRecord>>>,
 }
 
 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) => {

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]