External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflows): preserve multi-channel listing semantics (#6009)

Browse Source 
**Category:** fix
**User Impact:** Workflow listings reliably include every accessible
channel, including for users with more than 128 memberships and when
connected to older relays.
**Problem:** Multi-value `#h` filters could lose live delivery, apply
channel scoping after SQL limits, mishandle partial authorization or
revocation, and permit unbounded membership work. Desktop also submitted
every channel in one request, exceeding the relay's new 128-value safety
bound.
**Solution:** Preserve NIP-01 OR semantics across relay query, count,
and live-subscription paths while enforcing authorization and bounded
explicit-channel work before database or Redis operations. Desktop keeps
the older-relay-compatible one-channel-per-filter shape, sends filters
in bounded batches, combines responses, and deduplicates signed events
by event ID.

<details>
<summary>File changes</summary>

**crates/buzz-db/src/event.rs**
Distinguishes authorization channel scopes from explicit `#h` scopes in
list and count SQL so requested channels are applied before limits
without implicitly including global rows.

**crates/buzz-relay/src/handlers/req.rs**
Shares explicit-channel scope extraction and limits, preserves valid OR
siblings when malformed branches cannot match, repairs request-local
membership misses, and registers authorized live subscriptions per
channel.

**crates/buzz-relay/src/handlers/count.rs**
Applies the same bounded explicit-channel authorization to COUNT and
preserves channel scope when a multi-channel request narrows to one
authorized channel.

**crates/buzz-relay/src/api/bridge.rs**
Brings HTTP query and count behavior in line with WebSocket semantics
before SQL execution and rejects over-limit explicit-channel requests
before membership I/O.

**crates/buzz-relay/src/subscription.rs**
Indexes multi-channel subscriptions by every authorized channel and
shrinks, rather than destroys, their scope when one channel is revoked.

**crates/buzz-relay/src/handlers/side_effects.rs**
Releases only revoked channel topics and sends terminal closure only
when no authorized channel remains.

**crates/buzz-test-client/tests/e2e_relay.rs**
Adds ignored relay integration coverage for multi-channel delivery and
valid historical/live behavior with malformed or empty OR siblings.

**desktop/src-tauri/src/commands/workflows.rs**
Builds one single-channel filter per membership, submits at most 128 per
relay request, combines batches, and deduplicates by immutable signed
event ID.

**desktop/src-tauri/src/commands/workflows_tests.rs**
Covers filter compatibility, malformed input, 129-channel batching, and
cross-batch event-ID deduplication.

</details>

## Reproduction steps

1. Join multiple channels containing workflows, open **Workflows**, and
confirm workflows from every accessible channel appear.
2. Repeat with more than 128 memberships and confirm the listing remains
complete rather than failing the relay request.
3. Send a multi-value `#h` query/count and confirm only requested
authorized channels affect SQL limits and counts.
4. Subscribe to channels A and B, revoke A, and confirm B continues
delivering live events.
5. Subscribe with a valid channel branch plus a malformed or empty `#h`
sibling and confirm valid history, EOSE, and post-EOSE live delivery
still occur.

## Validation

At pushed head `c419a923f05e483ab26c006a0b3a80cfb3c73844`:

- Relay request tests: 53 passed.
- Desktop full Rust unit suite: 2,468 passed, 17 ignored.
- Relay E2E target compiled with `--no-run`.
- Strict relay clippy passed.
- Desktop Tauri clippy/check passed.
- Pre-push Rust tests and Desktop Tauri checks passed.
- Rust formatting and `git diff --check` passed.

---------

Signed-off-by: Taylor Ho <taylorkmho@gmail.com>
This commit is contained in:
Taylor Ho
2026-08-17 15:16:14 -07:00
committed by GitHub
parent f64899e5d1
commit f7a01bda7b
12 changed files with 1170 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-core/src/filter.rs
+22
View File
@@ -184,6 +184,28 @@ mod tests {
));
}
#[test]
fn h_tag_multi_value_filter_matches_any_channel() {
let channel_a = uuid::Uuid::new_v4();
let channel_b = uuid::Uuid::new_v4();
let stored = stored_with_tag(Tag::parse(["h", &channel_b.to_string()]).unwrap());
let filter = Filter::new().custom_tags(
nostr::SingleLetterTag::lowercase(nostr::Alphabet::H),
[channel_a.to_string(), channel_b.to_string()],
);
assert!(filters_match(&[filter], &stored));
}
#[test]
fn empty_h_tag_filter_matches_nothing() {
let channel_id = uuid::Uuid::new_v4();
let stored = stored_with_tag(Tag::parse(["h", &channel_id.to_string()]).unwrap());
let filter: Filter = serde_json::from_value(serde_json::json!({ "#h": [] })).unwrap();
assert!(!filters_match(&[filter], &stored));
}
#[test]
fn h_tag_fallback_uses_stored_channel_id() {
// Reactions (kind:7) and deletions (kind:5) don't carry h-tags —
crates/buzz-db/src/event.rs
+101 -19
View File
@@ -70,10 +70,17 @@ pub struct EventQuery {
/// Restrict results to events with an `e` tag referencing any of these event IDs (hex).
/// Uses JSONB containment (`tags @> ...`) against the `tags` column.
pub e_tags: Option<Vec<String>>,
/// Restrict results to events in any of these channels, while retaining
/// channel-less global events. Applied before SQL `LIMIT` so access-filtered
/// historical pages have exact exhaustion semantics.
/// Restrict results to events in any of these channels. By default,
/// channel-less global events are retained so this can enforce a viewer's
/// accessible-channel scope without hiding global events. Set
/// [`EventQuery::channel_ids_include_global`] to `false` for an explicit
/// multi-channel `#h` filter, which must match only requested channels.
/// Applied before SQL `LIMIT` so access- and filter-scoped historical pages
/// have exact exhaustion semantics.
pub channel_ids: Option<Vec<uuid::Uuid>>,
/// Whether [`EventQuery::channel_ids`] also retains channel-less global
/// events. Defaults to `true` for access-scope queries.
pub channel_ids_include_global: bool,
/// Override the default page clamp ([`DEFAULT_MAX_PAGE_LIMIT`]). Used by
/// the COUNT fallback path, which needs to fetch all matching events for
/// post-filter counting. When None, the default clamp applies.
@@ -122,6 +129,7 @@ impl EventQuery {
ids: None,
e_tags: None,
channel_ids: None,
channel_ids_include_global: true,
max_limit: None,
shared_gated_reader: None,
}
@@ -404,20 +412,24 @@ pub(crate) async fn query_events_on(
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
}
// Multi-channel IN pushdown: restrict to events in any of these channels
// OR global events (channel_id IS NULL). Used by NIP-45 COUNT to enforce
// channel access at the SQL level without fetching all rows.
// Multi-channel IN pushdown. Access-scope queries retain global events;
// explicit multi-value #h filters do not.
//
// SECURITY: Some(empty vec) means "user has access to NO channels" —
// only global events (channel_id IS NULL) should be returned.
// SECURITY: Some(empty vec) means "match no channels". Access-scope
// queries still retain globals; explicit #h queries match nothing.
if let Some(ref ch_ids) = q.channel_ids {
if ch_ids.is_empty() {
// No channel access — only global (non-channel) events visible.
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
if q.channel_ids_include_global {
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
} else {
qb.push(" AND FALSE");
}
} else {
qb.push(format!(
" AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id IN ("
));
qb.push(" AND (");
if q.channel_ids_include_global {
qb.push(format!("{col_prefix}channel_id IS NULL OR "));
}
qb.push(format!("{col_prefix}channel_id IN ("));
let mut sep = qb.separated(", ");
for ch in ch_ids {
sep.push_bind(*ch);
@@ -670,15 +682,21 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
}
// Multi-channel IN pushdown for COUNT: restrict to accessible channels + global.
// SECURITY: Some(empty vec) = no channel access → global events only.
// Multi-channel IN pushdown for COUNT. Access-scope queries retain global
// events; explicit multi-value #h filters do not.
if let Some(ref ch_ids) = q.channel_ids {
if ch_ids.is_empty() {
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
if q.channel_ids_include_global {
qb.push(format!(" AND {col_prefix}channel_id IS NULL"));
} else {
qb.push(" AND FALSE");
}
} else {
qb.push(format!(
" AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id IN ("
));
qb.push(" AND (");
if q.channel_ids_include_global {
qb.push(format!("{col_prefix}channel_id IS NULL OR "));
}
qb.push(format!("{col_prefix}channel_id IN ("));
let mut sep = qb.separated(", ");
for ch in ch_ids {
sep.push_bind(*ch);
@@ -1878,6 +1896,70 @@ mod tests {
.expect("sign timestamped event")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn explicit_multi_channel_scope_is_applied_before_historical_page_limit() {
let pool = setup_pool().await;
let community_uuid = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_uuid);
let channel_a = make_test_channel(&pool, community_uuid, None).await;
let channel_b = make_test_channel(&pool, community_uuid, None).await;
let unrelated_c = make_test_channel(&pool, community_uuid, None).await;
let base = 1_800_000_000;
let older_a = make_event_at(39_000, "older requested A", base + 1);
insert_event(&pool, community, &older_a, Some(channel_a))
.await
.expect("insert requested A candidate");
let requested_b = make_event_at(39_000, "requested B", base + 2);
insert_event(&pool, community, &requested_b, Some(channel_b))
.await
.expect("insert requested B candidate");
let newer_c = make_event_at(39_000, "newer unrelated C", base + 3);
insert_event(&pool, community, &newer_c, Some(unrelated_c))
.await
.expect("insert unrelated C candidate");
let global = make_event_at(39_000, "global candidate", base + 4);
insert_event(&pool, community, &global, None)
.await
.expect("insert global candidate");
let events = query_events(
&pool,
&EventQuery {
kinds: Some(vec![39_000]),
channel_ids: Some(vec![channel_a, channel_b]),
channel_ids_include_global: false,
limit: Some(1),
..EventQuery::for_community(community)
},
)
.await
.expect("query explicit multi-channel page");
assert_eq!(events.len(), 1);
assert_eq!(
events[0].event.id, requested_b.id,
"newer unrelated channel C must not consume the requested A/B limit"
);
let partial_authorization_count = count_events(
&pool,
&EventQuery {
kinds: Some(vec![39_000]),
channel_ids: Some(vec![channel_a]),
channel_ids_include_global: false,
..EventQuery::for_community(community)
},
)
.await
.expect("count one authorized channel from a multi-channel request");
assert_eq!(
partial_authorization_count, 1,
"partial authorization must exclude requested B, unrelated C, and global rows"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn access_scope_is_applied_before_historical_page_limit() {
crates/buzz-relay/src/api/bridge.rs
+76 -6
View File
@@ -981,6 +981,8 @@ async fn query_events_authed(
.map(|v| serde_json::from_value(v.clone()))
.collect::<Result<_, _>>()
.map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid filters: {e}")))?;
crate::handlers::req::extract_channel_ids_from_filters_limited(&filters)
.map_err(|()| api_error(StatusCode::BAD_REQUEST, "too many explicit channels"))?;
// P-gated kinds (gift wraps, member notifications, observer frames) require
// the caller's own pubkey in the #p tag — same enforcement as WS REQ handler.
@@ -1005,10 +1007,18 @@ async fn query_events_authed(
}
// Get channels this user can access — same enforcement as WS REQ handler.
let accessible_channels = state
let mut accessible_channels = state
.get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes)
.await
.map_err(|e| internal_error(&format!("channel access lookup: {e}")))?;
repair_requested_channel_access(
state,
tenant,
&filters,
&pubkey_bytes,
&mut accessible_channels,
)
.await?;
if filters.iter().any(|f| f.search.is_some()) {
if has_mixed_search_filters(&filters) {
@@ -1234,8 +1244,9 @@ async fn query_events_authed(
tenant.community(),
)
.await;
crate::handlers::req::apply_access_scope_to_query(
crate::handlers::req::apply_channel_scope_to_query(
&mut query,
filter,
extract_channel_from_filter(filter),
&accessible_channels,
);
@@ -1324,6 +1335,39 @@ async fn query_events_authed(
Ok(Json(Value::Array(events)))
}
async fn repair_requested_channel_access(
state: &AppState,
tenant: &TenantContext,
filters: &[nostr::Filter],
pubkey_bytes: &[u8],
accessible_channels: &mut Vec<uuid::Uuid>,
) -> Result<(), (StatusCode, Json<Value>)> {
for filter in filters {
let Some(requested) =
crate::handlers::req::extract_channel_ids_from_filters(std::slice::from_ref(filter))
else {
continue;
};
for channel_id in requested {
if accessible_channels.contains(&channel_id) {
continue;
}
let is_member = state
.db
.is_member(tenant.community(), channel_id, pubkey_bytes)
.await
.map_err(|e| internal_error(&format!("channel membership confirmation: {e}")))?;
crate::handlers::req::resolve_request_local_access(
accessible_channels,
channel_id,
true,
Some(is_member),
);
}
}
Ok(())
}
/// Count events via HTTP bridge (NIP-98 auth). Returns `{"count": N}`.
///
/// Enforces channel access: only counts events in channels the user can access.
@@ -1415,6 +1459,8 @@ async fn count_events_authed(
let filters: Vec<nostr::Filter> = serde_json::from_slice(body)
.map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid filters: {e}")))?;
crate::handlers::req::extract_channel_ids_from_filters_limited(&filters)
.map_err(|()| api_error(StatusCode::BAD_REQUEST, "too many explicit channels"))?;
// P-gated kinds enforcement — same as WS REQ and /query.
let authed_pubkey_hex = pubkey.to_hex();
@@ -1438,10 +1484,18 @@ async fn count_events_authed(
}
// Get channels this user can access.
let accessible_channels = state
let mut accessible_channels = state
.get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes)
.await
.map_err(|e| internal_error(&format!("channel access lookup: {e}")))?;
repair_requested_channel_access(
state,
tenant,
&filters,
&pubkey_bytes,
&mut accessible_channels,
)
.await?;
let mut total: u64 = 0;
for filter in &filters {
@@ -1463,9 +1517,19 @@ async fn count_events_authed(
crate::handlers::req::filter_can_match_shared_gated_kinds(filter);
// If filter targets a specific channel, verify access.
if let Some(ch_id) = extract_channel_from_filter(filter) {
if !accessible_channels.contains(&ch_id) {
continue; // Skip filters targeting inaccessible channels.
if crate::handlers::req::extract_channel_ids_from_filters(std::slice::from_ref(filter))
.is_some()
{
let ch_id = extract_channel_from_filter(filter);
let requested = crate::handlers::req::extract_channel_ids_from_filters(
std::slice::from_ref(filter),
)
.unwrap_or_default();
if !requested
.iter()
.any(|channel_id| accessible_channels.contains(channel_id))
{
continue;
}
// Channel is accessible — count with pushability check.
let mut query = crate::handlers::req::build_event_query_from_filter(
@@ -1475,6 +1539,12 @@ async fn count_events_authed(
tenant.community(),
)
.await;
crate::handlers::req::apply_channel_scope_to_query(
&mut query,
filter,
ch_id,
&accessible_channels,
);
// Shared-gated visibility pushdown: same as REQ and /query paths, so
// the fallback's query_events call doesn't over-fetch private rows.
if needs_shared_gate_filtering {
crates/buzz-relay/src/connection.rs
+12 -12
View File
@@ -24,7 +24,6 @@ use crate::state::{
run_registered_community_connection, AppState, CommunityConnectionControl,
CommunityDisconnectReason,
};
use buzz_pubsub::EventTopic;
/// Maximum time a new socket may hold a connection slot without completing NIP-42 auth.
const AUTH_TIMEOUT: Duration = Duration::from_secs(5);
@@ -287,10 +286,18 @@ async fn handle_active_connection(
let _ = auth_timeout_task.await;
for removed in state.sub_registry.remove_connection(conn.conn_id) {
state
.pubsub
.release_topic(&conn.tenant, topic_for_subscription(removed.channel_id))
.await;
if removed.scope.is_global() {
state
.pubsub
.release_topic(&conn.tenant, buzz_pubsub::EventTopic::Global)
.await;
}
for &channel_id in removed.scope.channel_ids() {
state
.pubsub
.release_topic(&conn.tenant, buzz_pubsub::EventTopic::Channel(channel_id))
.await;
}
}
state.conn_manager.deregister(conn.conn_id);
if let AuthState::Authenticated(ref auth_ctx) = *conn.auth_state.read().await {
@@ -729,13 +736,6 @@ fn send_admission_result(
}
}
fn topic_for_subscription(channel_id: Option<Uuid>) -> EventTopic {
match channel_id {
Some(channel_id) => EventTopic::Channel(channel_id),
None => EventTopic::Global,
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/buzz-relay/src/handlers/close.rs
+12 -12
View File
@@ -5,7 +5,6 @@ use tracing::debug;
use crate::connection::ConnectionState;
use crate::protocol::RelayMessage;
use crate::state::AppState;
use buzz_pubsub::EventTopic;
/// Handle a CLOSE command — remove the subscription and send CLOSED acknowledgement.
pub async fn handle_close(sub_id: String, conn: Arc<ConnectionState>, state: Arc<AppState>) {
@@ -16,20 +15,21 @@ pub async fn handle_close(sub_id: String, conn: Arc<ConnectionState>, state: Arc
// Deregister from the fan-out index before sending CLOSED so no new
// messages are routed to this sub after the client's CLOSE is acknowledged.
if let Some(removed) = state.sub_registry.remove_subscription(conn_id, &sub_id) {
state
.pubsub
.release_topic(&conn.tenant, topic_for_subscription(removed.channel_id))
.await;
if removed.scope.is_global() {
state
.pubsub
.release_topic(&conn.tenant, buzz_pubsub::EventTopic::Global)
.await;
}
for &channel_id in removed.scope.channel_ids() {
state
.pubsub
.release_topic(&conn.tenant, buzz_pubsub::EventTopic::Channel(channel_id))
.await;
}
}
conn.send(RelayMessage::closed(&sub_id, ""));
debug!(conn_id = %conn_id, sub_id = %sub_id, "Subscription closed");
}
fn topic_for_subscription(channel_id: Option<uuid::Uuid>) -> EventTopic {
match channel_id {
Some(channel_id) => EventTopic::Channel(channel_id),
None => EventTopic::Global,
}
}
crates/buzz-relay/src/handlers/count.rs
+66 -43
View File
@@ -13,20 +13,8 @@ use crate::handlers::req::{
use crate::protocol::RelayMessage;
use crate::state::AppState;
/// Extract a channel UUID from a single filter's `#h` tag.
fn extract_channel_from_filter(filter: &Filter) -> Option<uuid::Uuid> {
let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
filter.generic_tags.get(&h_tag).and_then(|vs| {
if vs.len() == 1 {
vs.iter().next()?.parse::<uuid::Uuid>().ok()
} else {
None
}
})
}
/// Handle a COUNT message: require auth, enforce channel access, execute filters,
/// return aggregate count.
/// and return the aggregate count.
pub async fn handle_count(
sub_id: String,
filters: Vec<Filter>,
@@ -75,6 +63,23 @@ pub async fn handle_count(
return;
}
let requested_channel_sets =
match super::req::extract_channel_ids_from_filters_limited(&filters) {
Ok(_) => filters
.iter()
.map(|filter| {
super::req::extract_channel_ids_from_filters(std::slice::from_ref(filter))
})
.collect::<Vec<_>>(),
Err(()) => {
conn.send(RelayMessage::closed(
&sub_id,
"restricted: too many explicit channels",
));
return;
}
};
// Get channels this user can access — same enforcement as WS REQ handler.
let mut accessible_channels = match state
.get_accessible_channel_ids_cached(conn.tenant.community(), &pubkey_bytes)
@@ -98,7 +103,7 @@ pub async fn handle_count(
// For each filter, count matching events with channel access enforcement.
let mut total: u64 = 0;
for filter in &filters {
for (filter, requested_channels) in filters.iter().zip(requested_channel_sets) {
// Determine if this filter can match author-only kinds — if so, the
// fast-path count_events() cannot be used because it doesn't do
// per-event author filtering.
@@ -117,38 +122,50 @@ pub async fn handle_count(
let needs_result_gated_filtering = filter_can_match_result_gated_kinds(filter)
&& !result_gated_count_safe_for_pushdown(filter, &authed_pubkey_hex);
if let Some(ch_id) = extract_channel_from_filter(filter) {
// Filter targets a specific channel — verify access. Mirrors the WS
// REQ handler: a cache-negative may be a stale miss on a non-writer
// pod, so confirm uncached and repair the Vec request-locally via
// `super::req::resolve_request_local_access` (so a just-added channel
// is counted, and any later filter on the same channel sees it too).
let db_is_member = if accessible_channels.contains(&ch_id) {
None
} else {
match state
.db
.is_member(conn.tenant.community(), ch_id, &pubkey_bytes)
.await
{
Ok(member) => Some(member),
Err(e) => {
warn!(sub_id = %sub_id, "Channel membership confirmation failed: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
if let Some(requested_channels) = requested_channels {
for &ch_id in &requested_channels {
if accessible_channels.contains(&ch_id) {
continue;
}
};
if !super::req::resolve_request_local_access(
&mut accessible_channels,
ch_id,
token_channel_ids
let token_allows = token_channel_ids
.as_deref()
.is_none_or(|allowed| allowed.contains(&ch_id)),
db_is_member,
) {
continue; // Skip filters targeting inaccessible channels.
.is_none_or(|allowed| allowed.contains(&ch_id));
let db_is_member = if token_allows {
match state
.db
.is_member(conn.tenant.community(), ch_id, &pubkey_bytes)
.await
{
Ok(member) => Some(member),
Err(e) => {
warn!(sub_id = %sub_id, "Channel membership confirmation failed: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
}
} else {
None
};
super::req::resolve_request_local_access(
&mut accessible_channels,
ch_id,
token_allows,
db_is_member,
);
}
let authorized_requested: Vec<_> = requested_channels
.iter()
.copied()
.filter(|channel_id| accessible_channels.contains(channel_id))
.collect();
if authorized_requested.is_empty() {
continue;
}
// Preserve the original explicit multi-channel shape even when
// authorization narrows it to one channel. The helper must write
// that intersection into `channel_ids`; synthesizing `Some(A)` here
// would leave a query built from multi-#h completely unscoped.
let ch_id = (requested_channels.len() == 1).then_some(authorized_requested[0]);
// Channel is accessible — count with pushability check.
let mut query = super::req::build_event_query_from_filter(
filter,
@@ -157,6 +174,12 @@ pub async fn handle_count(
conn.tenant.community(),
)
.await;
super::req::apply_channel_scope_to_query(
&mut query,
filter,
ch_id,
&accessible_channels,
);
// Shared-gated visibility pushdown: pre-filter the fallback
// query_events candidate page before ORDER/LIMIT.
if needs_shared_gate_filtering {
crates/buzz-relay/src/handlers/req.rs
+379 -97
View File
@@ -33,6 +33,14 @@ const MAX_SUBSCRIPTIONS: usize = 1024;
/// `buffer_unordered`), so dedupe/trace/error semantics are unchanged.
pub(crate) const FILTER_QUERY_CONCURRENCY: usize = 4;
/// Maximum aggregate number of explicit `#h` values accepted in one REQ,
/// COUNT, HTTP `/query`, or HTTP `/count` request.
///
/// Explicit channels may each require an uncached membership lookup and, for a
/// live WS subscription, a registry entry plus Redis topic retain. Bound the
/// values before any of that request-amplified work begins.
pub(crate) const MAX_EXPLICIT_CHANNEL_VALUES: usize = 128;
// Guard: keep the bound a small fraction of any sane Postgres pool size.
// Raising it past this range requires re-running the relay bench and
// reconsidering pool contention (see docs above). Compile-time — violating
@@ -85,6 +93,18 @@ pub async fn handle_req(
}
};
let channel_id = extract_channel_id_from_filters(&filters);
let requested_channel_ids = match extract_channel_ids_from_filters_limited(&filters) {
Ok(ids) => ids,
Err(()) => {
conn.send(RelayMessage::closed(
&sub_id,
"restricted: too many explicit channels",
));
return;
}
};
let mut accessible_channels = if filters_are_nip43_membership_only(&filters) {
metrics::counter!("buzz_req_global_access_resolution_skips_total", "kind" => "13534")
.increment(1);
@@ -106,8 +126,6 @@ pub async fn handle_req(
accessible_channels.retain(|channel_id| allowed.contains(channel_id));
}
let channel_id = extract_channel_id_from_filters(&filters);
// Build the conformance `AbstractState` once at request entry. The
// `Option` only goes `None` on malformed pubkey bytes (already a
// separate failure path elsewhere); on the hot read path this is
@@ -126,50 +144,70 @@ pub async fn handle_req(
// `resolve_request_local_access`). Running this ahead of the search branch
// is what fixes the search false-miss: a `#h=<just-added>` search would
// otherwise be scoped against the stale vector and return empty.
if let Some(ch_id) = channel_id {
let token_allows = token_channel_ids
.as_deref()
.is_none_or(|allowed| allowed.contains(&ch_id));
let db_is_member = if !token_allows || accessible_channels.contains(&ch_id) {
None
} else {
match state
.db
.is_member(conn.tenant.community(), ch_id, &pubkey_bytes)
.await
{
Ok(member) => {
if let Some(state_snap) = trace_state.as_ref() {
crate::conformance::record_req_authcheck(
&state.tracer,
state_snap,
ch_id,
member,
);
if let Some(requested) = requested_channel_ids.as_ref() {
for &ch_id in requested {
let token_allows = token_channel_ids
.as_deref()
.is_none_or(|allowed| allowed.contains(&ch_id));
let db_is_member = if !token_allows || accessible_channels.contains(&ch_id) {
None
} else {
match state
.db
.is_member(conn.tenant.community(), ch_id, &pubkey_bytes)
.await
{
Ok(member) => {
if let Some(state_snap) = trace_state.as_ref() {
crate::conformance::record_req_authcheck(
&state.tracer,
state_snap,
ch_id,
member,
);
}
Some(member)
}
Err(e) => {
warn!(conn_id = %conn_id, "Channel membership confirmation failed: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
Some(member)
}
Err(e) => {
warn!(conn_id = %conn_id, "Channel membership confirmation failed: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
}
};
if !resolve_request_local_access(
&mut accessible_channels,
ch_id,
token_allows,
db_is_member,
) {
conn.send(RelayMessage::closed(
&sub_id,
"restricted: not a channel member",
));
return;
};
// An OR filter may include inaccessible channels; retain every
// authorized requested channel and silently omit the others.
resolve_request_local_access(
&mut accessible_channels,
ch_id,
token_allows,
db_is_member,
);
}
}
let authorized_requested_channels = requested_channel_ids.as_ref().map(|requested| {
requested
.iter()
.copied()
.filter(|channel_id| accessible_channels.contains(channel_id))
.collect::<Vec<_>>()
});
// Partial authorization preserves NIP-01 OR semantics by omitting only
// inaccessible branches. If no valid requested channel survives, retain the
// established single-channel contract: reject instead of registering a
// subscription that can never produce an event or a terminal notice.
if authorized_requested_channels
.as_ref()
.is_some_and(|authorized| authorized.is_empty())
{
conn.send(RelayMessage::closed(
&sub_id,
"restricted: not a channel member",
));
return;
}
// Applied BEFORE the NIP-50 search branch so that an authenticated member
// cannot use `{"search":"...","kinds":[30174]}` (or similar for p-gated
// kinds) to harvest indexed-but-globally-stored sensitive events. Search
@@ -236,23 +274,39 @@ pub async fn handle_req(
subs.insert(sub_id.clone(), filters.clone());
}
let replaced = state.sub_registry.register_scoped(
conn.tenant.community(),
conn_id,
sub_id.clone(),
filters.clone(),
channel_id,
);
let replaced = if let Some(channel_ids) = authorized_requested_channels.as_ref() {
state.sub_registry.register_channels_scoped(
conn.tenant.community(),
conn_id,
sub_id.clone(),
filters.clone(),
channel_ids.clone(),
)
} else {
state.sub_registry.register_scoped(
conn.tenant.community(),
conn_id,
sub_id.clone(),
filters.clone(),
None,
)
};
if let Some(replaced) = replaced {
release_subscription_topics(&state, &conn.tenant, &replaced.scope).await;
}
if let Some(channel_ids) = authorized_requested_channels.as_ref() {
for &channel_id in channel_ids {
state
.pubsub
.retain_topic(&conn.tenant, EventTopic::Channel(channel_id))
.await;
}
} else {
state
.pubsub
.release_topic(&conn.tenant, topic_for_subscription(replaced.channel_id))
.retain_topic(&conn.tenant, EventTopic::Global)
.await;
}
state
.pubsub
.retain_topic(&conn.tenant, topic_for_subscription(channel_id))
.await;
debug!(conn_id = %conn_id, sub_id = %sub_id, "Subscription registered");
@@ -288,7 +342,12 @@ pub async fn handle_req(
};
let mut params =
filter_to_query_params(filter, per_filter_channel, conn.tenant.community());
apply_access_scope_to_query(&mut params, per_filter_channel, &accessible_channels);
apply_channel_scope_to_query(
&mut params,
filter,
per_filter_channel,
&accessible_channels,
);
// Shared-gated visibility pushdown: set reader bytes so query_events
// appends the SQL visibility clause before ORDER/LIMIT, preventing
// newer private events from starving older shared ones off the page.
@@ -785,11 +844,11 @@ pub(crate) fn count_fallback_exceeded(candidate_count: usize) -> bool {
/// an exact count without post-filtering.
///
/// Pushed constraints: kinds, authors (single or multi), ids, since, until,
/// channel_id (#h single), #p (single), #d (single, NIP-33-only kinds), #e (any),
/// channel_ids (injected by caller).
/// authorized channel scope (#h single or multi, injected by caller), #p (single),
/// #d (single, NIP-33-only kinds), #e (any).
///
/// Anything else (multi-#p, #t, #a, search, multi-#h, #d on non-NIP-33)
/// requires post-filtering and cannot use the fast COUNT path.
/// Anything else (multi-#p, #t, #a, search, #d on non-NIP-33) requires
/// post-filtering and cannot use the fast COUNT path.
pub fn filter_fully_pushable(filter: &Filter) -> bool {
// Check if filter exclusively targets NIP-33 kinds (needed for #d pushability).
let is_nip33_only = filter.kinds.as_ref().is_some_and(|ks| {
@@ -803,10 +862,8 @@ pub fn filter_fully_pushable(filter: &Filter) -> bool {
let key = tag_key.to_string();
match key.as_str() {
"h" => {
// Single #h is pushed as channel_id; multi-#h is not.
if tag_values.len() > 1 {
return false;
}
// The caller pushes the complete authorized #h set through
// EventQuery::channel_id/channel_ids before invoking COUNT.
}
"p" => {
// Single #p is pushed via event_mentions join; multi is not.
@@ -854,19 +911,20 @@ fn filters_are_nip43_membership_only(filters: &[Filter]) -> bool {
})
}
/// Extract a channel UUID from a single filter's `#h` tag.
/// Extract the single channel UUID from a filter's `#h` tag.
///
/// A multi-value `#h` filter has NIP-01 OR semantics, so it cannot be reduced
/// to one `EventQuery::channel_id` without dropping matches from the other
/// channels. Return `None` in that case and let the caller apply the accessible
/// channel set in SQL before the full filter is evaluated in Rust.
fn extract_channel_id_from_filter(filter: &Filter) -> Option<uuid::Uuid> {
for (tag_key, tag_values) in filter.generic_tags.iter() {
let key = tag_key.to_string();
if key == "h" {
for val in tag_values {
if let Ok(id) = val.parse::<uuid::Uuid>() {
return Some(id);
}
}
}
let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
let values = filter.generic_tags.get(&h_tag)?;
if values.len() != 1 {
return None;
}
None
values.iter().next()?.parse::<uuid::Uuid>().ok()
}
/// Convert a single NIP-01 filter into an [`EventQuery`] for the database.
@@ -1002,30 +1060,96 @@ fn filter_to_query_params(
}
}
/// Push the caller's authorized channel set into logically global historical
/// queries so SQL `LIMIT` counts visible rows. Channel-less events remain in
/// scope by `EventQuery::channel_ids` contract; an explicit single-channel
/// filter keeps its narrower `channel_id` predicate.
pub(crate) fn apply_access_scope_to_query(
/// Push channel constraints into SQL before `LIMIT`.
///
/// A valid multi-value `#h` is narrowed to the requested channels the reader
/// may access. Invalid values are ignored, and an empty authorized result is an
/// explicit match-nothing scope rather than a global query. Filters without
/// `#h` retain the full accessible-channel scope plus global events.
pub(crate) fn apply_channel_scope_to_query(
query: &mut EventQuery,
filter: &Filter,
channel_id: Option<uuid::Uuid>,
accessible_channels: &[uuid::Uuid],
) {
if channel_id.is_none() {
if channel_id.is_some() {
return;
}
let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
if let Some(values) = filter.generic_tags.get(&h_tag) {
query.channel_ids = Some(
values
.iter()
.filter_map(|value| value.parse::<uuid::Uuid>().ok())
.filter(|requested| accessible_channels.contains(requested))
.collect(),
);
query.channel_ids_include_global = false;
} else {
query.channel_ids = Some(accessible_channels.to_vec());
}
}
/// Extract a single channel UUID from filter generic tags, or `None` if the
/// subscription is logically global.
/// Extract the complete channel set when every filter is explicitly #h-scoped.
/// `None` means at least one filter is community-global.
///
/// Checks the `"h"` tag key — channel-scoped subscriptions use `#h = <uuid>`.
///
/// Returns `None` when:
/// - Any filter has no channel tag (that filter matches all channels → global sub), or
/// - Multiple distinct channel UUIDs appear across filters (can't index under one channel).
///
/// Callers that receive `None` treat the subscription as global (slow-path fan-out).
/// The aggregate value count is checked before UUID parsing or membership I/O;
/// duplicate and malformed values still consume the request budget.
pub(crate) fn extract_channel_ids_from_filters_limited(
filters: &[Filter],
) -> Result<Option<Vec<uuid::Uuid>>, ()> {
let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
let value_count = filters.iter().try_fold(0usize, |count, filter| {
let additional = filter
.generic_tags
.get(&h_tag)
.map_or(0, |values| values.len());
count.checked_add(additional).ok_or(())
})?;
if value_count > MAX_EXPLICIT_CHANNEL_VALUES {
return Err(());
}
Ok(extract_channel_ids_from_filters(filters))
}
/// Extract the complete channel set without applying the aggregate request budget.
/// Callers that can trigger I/O must validate first with
/// [`extract_channel_ids_from_filters_limited`].
pub(crate) fn extract_channel_ids_from_filters(filters: &[Filter]) -> Option<Vec<uuid::Uuid>> {
let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
let mut channel_ids = Vec::new();
for filter in filters {
let values = filter.generic_tags.get(&h_tag)?;
for value in values {
if let Ok(channel_id) = value.parse::<uuid::Uuid>() {
if !channel_ids.contains(&channel_id) {
channel_ids.push(channel_id);
}
}
}
}
Some(channel_ids)
}
async fn release_subscription_topics(
state: &AppState,
tenant: &TenantContext,
scope: &crate::subscription::SubscriptionScope,
) {
if scope.is_global() {
state.pubsub.release_topic(tenant, EventTopic::Global).await;
} else {
for &channel_id in scope.channel_ids() {
state
.pubsub
.release_topic(tenant, EventTopic::Channel(channel_id))
.await;
}
}
}
fn extract_channel_id_from_filters(filters: &[Filter]) -> Option<uuid::Uuid> {
let mut found_id: Option<uuid::Uuid> = None;
for f in filters {
@@ -1289,13 +1413,6 @@ pub(crate) fn author_only_filters_authorized(filters: &[Filter], authed_pubkey_h
})
}
fn topic_for_subscription(channel_id: Option<uuid::Uuid>) -> EventTopic {
match channel_id {
Some(channel_id) => EventTopic::Channel(channel_id),
None => EventTopic::Global,
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -1308,7 +1425,7 @@ mod tests {
uuid::Uuid::new_v4(),
));
apply_access_scope_to_query(&mut query, None, &accessible);
apply_channel_scope_to_query(&mut query, &Filter::new(), None, &accessible);
assert_eq!(query.channel_ids.as_deref(), Some(accessible.as_slice()));
}
@@ -1322,7 +1439,7 @@ mod tests {
));
query.channel_id = Some(channel);
apply_access_scope_to_query(&mut query, Some(channel), &accessible);
apply_channel_scope_to_query(&mut query, &Filter::new(), Some(channel), &accessible);
assert!(query.channel_ids.is_none());
assert_eq!(query.channel_id, Some(channel));
@@ -1551,6 +1668,171 @@ mod tests {
assert_eq!(extract_channel_id_from_filters(&filters), Some(channel_id));
}
#[test]
fn extract_channel_id_from_multi_value_filter_returns_none() {
let channel_a = uuid::Uuid::new_v4();
let channel_b = uuid::Uuid::new_v4();
let filter: Filter = serde_json::from_value(serde_json::json!({
"#h": [channel_a.to_string(), channel_b.to_string()],
}))
.unwrap();
assert_eq!(extract_channel_id_from_filter(&filter), None);
assert_eq!(
filter_to_query_params(
&filter,
extract_channel_id_from_filter(&filter),
buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::nil()),
)
.channel_id,
None,
"multi-channel OR filters must not be narrowed to their first channel",
);
}
#[test]
fn valid_channel_union_survives_malformed_or_empty_explicit_siblings() {
let valid = uuid::Uuid::new_v4();
for sibling in [
serde_json::json!({"#h": ["not-a-uuid"]}),
serde_json::json!({"#h": []}),
] {
let filters = [
filter_with_channel(valid),
serde_json::from_value(sibling).expect("parse sibling filter"),
];
assert_eq!(
extract_channel_ids_from_filters(&filters),
Some(vec![valid]),
);
}
let malformed_only: Filter =
serde_json::from_value(serde_json::json!({"#h": ["not-a-uuid"]}))
.expect("parse malformed filter");
assert_eq!(
extract_channel_ids_from_filters(&[malformed_only]),
Some(Vec::new()),
"malformed-only explicit scope must remain match-nothing, never global",
);
}
#[test]
fn explicit_channel_limit_is_aggregate_and_counts_every_value() {
let channel_values = |count: usize| {
(0..count)
.map(|_| uuid::Uuid::new_v4().to_string())
.collect::<Vec<_>>()
};
let at_limit: Filter = serde_json::from_value(serde_json::json!({
"#h": channel_values(MAX_EXPLICIT_CHANNEL_VALUES),
}))
.unwrap();
assert!(extract_channel_ids_from_filters_limited(&[at_limit]).is_ok());
let first: Filter = serde_json::from_value(serde_json::json!({
"#h": channel_values(MAX_EXPLICIT_CHANNEL_VALUES),
}))
.unwrap();
let duplicate_over_limit: Filter = serde_json::from_value(serde_json::json!({
"#h": [uuid::Uuid::nil().to_string()],
}))
.unwrap();
assert_eq!(
extract_channel_ids_from_filters_limited(&[first, duplicate_over_limit]),
Err(()),
);
let global_then_over_limit = [
Filter::new(),
serde_json::from_value(serde_json::json!({
"#h": channel_values(MAX_EXPLICIT_CHANNEL_VALUES + 1),
}))
.unwrap(),
];
assert_eq!(
extract_channel_ids_from_filters_limited(&global_then_over_limit),
Err(()),
"a global filter must not hide an over-limit explicit filter",
);
}
#[test]
fn multi_value_h_scope_intersects_access_before_limit() {
let channel_a = uuid::Uuid::new_v4();
let channel_b = uuid::Uuid::new_v4();
let unrelated_c = uuid::Uuid::new_v4();
let unauthorized = uuid::Uuid::new_v4();
let filter: Filter = serde_json::from_value(serde_json::json!({
"#h": [
channel_a.to_string(),
channel_b.to_string(),
unauthorized.to_string(),
"not-a-uuid"
],
"limit": 1
}))
.unwrap();
let mut query = filter_to_query_params(
&filter,
extract_channel_id_from_filter(&filter),
buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::nil()),
);
apply_channel_scope_to_query(
&mut query,
&filter,
None,
&[channel_a, channel_b, unrelated_c],
);
let scoped_channels = query.channel_ids.expect("explicit channel scope");
assert_eq!(scoped_channels.len(), 2);
assert!(scoped_channels.contains(&channel_a));
assert!(scoped_channels.contains(&channel_b));
assert!(!query.channel_ids_include_global);
assert_eq!(query.limit, Some(1));
}
#[test]
fn multi_value_h_scope_remains_explicit_when_only_one_channel_is_authorized() {
let authorized = uuid::Uuid::new_v4();
let unauthorized = uuid::Uuid::new_v4();
let filter: Filter = serde_json::from_value(serde_json::json!({
"#h": [authorized.to_string(), unauthorized.to_string()],
}))
.unwrap();
let mut query = filter_to_query_params(
&filter,
extract_channel_id_from_filter(&filter),
buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::nil()),
);
apply_channel_scope_to_query(&mut query, &filter, None, &[authorized]);
assert_eq!(query.channel_id, None);
assert_eq!(query.channel_ids, Some(vec![authorized]));
assert!(!query.channel_ids_include_global);
}
#[test]
fn empty_or_unauthorized_h_scope_matches_nothing() {
for values in [serde_json::json!([]), serde_json::json!(["not-a-uuid"])] {
let filter: Filter =
serde_json::from_value(serde_json::json!({ "#h": values })).unwrap();
let mut query = filter_to_query_params(
&filter,
None,
buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::nil()),
);
apply_channel_scope_to_query(&mut query, &filter, None, &[uuid::Uuid::new_v4()]);
assert_eq!(query.channel_ids, Some(Vec::new()));
assert!(!query.channel_ids_include_global);
}
}
#[test]
fn test_extract_channel_id_mixed_channels_returns_none() {
let channel_a = uuid::Uuid::new_v4();
crates/buzz-relay/src/handlers/side_effects.rs
+12 -15
View File
@@ -116,20 +116,24 @@ async fn evict_conn_channel_subscriptions(
if let Some(subscriptions) = state.conn_manager.subscriptions_for(conn_id) {
let mut conn_subscriptions = subscriptions.lock().await;
for (sub_id, _) in &removed {
conn_subscriptions.remove(sub_id);
for update in &removed {
if update.removed {
conn_subscriptions.remove(&update.sub_id);
}
}
}
for (sub_id, removed_scope) in removed {
for update in removed {
state
.pubsub
.release_topic(tenant, topic_for_subscription(removed_scope.channel_id))
.release_topic(tenant, buzz_pubsub::EventTopic::Channel(channel_id))
.await;
let _ = state.conn_manager.send_to(
conn_id,
RelayMessage::closed(&sub_id, "restricted: channel access revoked"),
);
if update.removed {
let _ = state.conn_manager.send_to(
conn_id,
RelayMessage::closed(&update.sub_id, "restricted: channel access revoked"),
);
}
}
}
@@ -3367,13 +3371,6 @@ pub async fn publish_nipia_unarchived(
.await
}
fn topic_for_subscription(channel_id: Option<Uuid>) -> EventTopic {
match channel_id {
Some(channel_id) => EventTopic::Channel(channel_id),
None => EventTopic::Global,
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/buzz-relay/src/subscription.rs
+279 -84
View File
@@ -13,7 +13,39 @@ pub type ConnId = Uuid;
/// Subscription identifier — the client-supplied string from a REQ message.
pub type SubId = String;
/// Stored subscription entry: filters paired with server-resolved community and optional channel scope.
pub type SubEntry = (Vec<Filter>, CommunityId, Option<Uuid>);
pub type SubEntry = (Vec<Filter>, CommunityId, SubscriptionScope);
/// Server-resolved live-routing scope for a subscription.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SubscriptionScope {
/// Community-global events only.
Global,
/// Events from any of these authorized channels.
Channels(Vec<Uuid>),
}
impl SubscriptionScope {
fn matches_channel(&self, channel_id: Option<Uuid>) -> bool {
match (self, channel_id) {
(Self::Global, None) => true,
(Self::Channels(channels), Some(channel_id)) => channels.contains(&channel_id),
_ => false,
}
}
/// Return the channels retained by this routing scope.
pub fn channel_ids(&self) -> &[Uuid] {
match self {
Self::Global => &[],
Self::Channels(channels) => channels,
}
}
/// Whether this routing scope retains the community-global topic.
pub fn is_global(&self) -> bool {
matches!(self, Self::Global)
}
}
/// Index key combining a channel and event kind for O(1) fan-out lookups.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -32,12 +64,21 @@ struct GlobalPKindIndexKey {
}
/// A removed subscription's server-resolved routing scope.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemovedSubscription {
/// Server-resolved community this subscription belonged to.
pub community_id: CommunityId,
/// Tenant-local channel scope; `None` means the community-global topic.
pub channel_id: Option<Uuid>,
/// Server-resolved topics retained by the removed subscription.
pub scope: SubscriptionScope,
}
/// Result of removing one revoked channel from a live subscription scope.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChannelSubscriptionUpdate {
/// Client-supplied subscription identifier.
pub sub_id: SubId,
/// Whether no authorized channels remain and the subscription was removed.
pub removed: bool,
}
/// Thread-safe registry of active subscriptions with targeted in-memory fan-out indexes.
@@ -73,42 +114,77 @@ impl SubscriptionRegistry {
sub_id: SubId,
filters: Vec<Filter>,
channel_id: Option<Uuid>,
) -> Option<RemovedSubscription> {
let scope = channel_id
.map(|channel_id| SubscriptionScope::Channels(vec![channel_id]))
.unwrap_or(SubscriptionScope::Global);
self.register_with_scope(community_id, conn_id, sub_id, filters, scope)
}
/// Register a subscription under every authorized requested channel.
pub fn register_channels_scoped(
&self,
community_id: CommunityId,
conn_id: ConnId,
sub_id: SubId,
filters: Vec<Filter>,
channel_ids: Vec<Uuid>,
) -> Option<RemovedSubscription> {
self.register_with_scope(
community_id,
conn_id,
sub_id,
filters,
SubscriptionScope::Channels(channel_ids),
)
}
fn register_with_scope(
&self,
community_id: CommunityId,
conn_id: ConnId,
sub_id: SubId,
filters: Vec<Filter>,
scope: SubscriptionScope,
) -> Option<RemovedSubscription> {
let removed = self.remove_subscription(conn_id, &sub_id);
self.subs
.entry(conn_id)
.or_default()
.insert(sub_id.clone(), (filters.clone(), community_id, channel_id));
self.subs.entry(conn_id).or_default().insert(
sub_id.clone(),
(filters.clone(), community_id, scope.clone()),
);
metrics::gauge!("buzz_subscriptions_active").increment(1.0);
if let Some(ch_id) = channel_id {
match extract_kinds_from_filters(&filters) {
None => {
// At least one filter has no `kinds` constraint — wildcard,
// this sub wants all kinds in this channel.
self.channel_wildcard_index
.entry((community_id, ch_id))
.or_default()
.push((conn_id, sub_id.clone()));
}
Some(kinds) if kinds.is_empty() => {
// All filters had explicit empty kinds lists (`kinds: []`).
// Per NIP-01, `kinds: []` means "match no kinds" — this
// subscription will never receive any events. Do not index it
// anywhere; `filters_match` will reject all events at fan-out.
}
Some(kinds) => {
for kind in kinds {
let key = IndexKey {
channel_id: ch_id,
kind,
};
self.channel_kind_index
.entry((community_id, key))
if let SubscriptionScope::Channels(channel_ids) = &scope {
for ch_id in channel_ids {
let ch_id = *ch_id;
match extract_kinds_from_filters(&filters) {
None => {
// At least one filter has no `kinds` constraint — wildcard,
// this sub wants all kinds in this channel.
self.channel_wildcard_index
.entry((community_id, ch_id))
.or_default()
.push((conn_id, sub_id.clone()));
}
Some(kinds) if kinds.is_empty() => {
// All filters had explicit empty kinds lists (`kinds: []`).
// Per NIP-01, `kinds: []` means "match no kinds" — this
// subscription will never receive any events. Do not index it
// anywhere; `filters_match` will reject all events at fan-out.
}
Some(kinds) => {
for kind in kinds {
let key = IndexKey {
channel_id: ch_id,
kind,
};
self.channel_kind_index
.entry((community_id, key))
.or_default()
.push((conn_id, sub_id.clone()));
}
}
}
}
} else {
@@ -177,16 +253,16 @@ impl SubscriptionRegistry {
F: FnOnce(),
{
let mut conn_subs = self.subs.get_mut(&conn_id)?;
let (filters, community_id, channel_id) = conn_subs.remove(sub_id)?;
let (filters, community_id, scope) = conn_subs.remove(sub_id)?;
after_remove();
self.remove_from_index(conn_id, sub_id, &filters, community_id, channel_id);
self.remove_from_index(conn_id, sub_id, &filters, community_id, &scope);
drop(conn_subs);
metrics::gauge!("buzz_subscriptions_active").decrement(1.0);
Some(RemovedSubscription {
community_id,
channel_id,
scope,
})
}
@@ -195,11 +271,11 @@ impl SubscriptionRegistry {
let mut removed = Vec::new();
if let Some((_, conn_subs)) = self.subs.remove(&conn_id) {
let count = conn_subs.len();
for (sub_id, (filters, community_id, channel_id)) in &conn_subs {
self.remove_from_index(conn_id, sub_id, filters, *community_id, *channel_id);
for (sub_id, (filters, community_id, scope)) in &conn_subs {
self.remove_from_index(conn_id, sub_id, filters, *community_id, scope);
removed.push(RemovedSubscription {
community_id: *community_id,
channel_id: *channel_id,
scope: scope.clone(),
});
}
metrics::gauge!("buzz_subscriptions_active").decrement(count as f64);
@@ -207,34 +283,58 @@ impl SubscriptionRegistry {
removed
}
/// Remove all subscriptions on `conn_id` scoped to `channel_id` in one community.
/// Remove one revoked channel from every matching subscription in a community.
/// Multi-channel subscriptions are re-indexed with their remaining scope;
/// subscriptions with no channels left are removed entirely.
pub fn remove_channel_subscriptions_scoped(
&self,
community_id: CommunityId,
conn_id: ConnId,
channel_id: Uuid,
) -> Vec<(SubId, RemovedSubscription)> {
) -> Vec<ChannelSubscriptionUpdate> {
let sub_ids: Vec<SubId> = self
.subs
.get(&conn_id)
.map(|conn_subs| {
conn_subs
.iter()
.filter_map(|(sub_id, (_, sub_community_id, sub_channel_id))| {
(*sub_community_id == community_id && *sub_channel_id == Some(channel_id))
.then_some(sub_id.clone())
.filter_map(|(sub_id, (_, sub_community_id, scope))| {
(*sub_community_id == community_id
&& scope.channel_ids().contains(&channel_id))
.then_some(sub_id.clone())
})
.collect()
})
.unwrap_or_default();
sub_ids
.into_iter()
.filter_map(|sub_id| {
let removed = self.remove_subscription(conn_id, &sub_id)?;
Some((sub_id, removed))
})
.collect()
let mut updates = Vec::with_capacity(sub_ids.len());
for sub_id in sub_ids {
let Some(mut conn_subs) = self.subs.get_mut(&conn_id) else {
break;
};
let Some((filters, _, scope)) = conn_subs.get_mut(&sub_id) else {
continue;
};
let filters = filters.clone();
let SubscriptionScope::Channels(channel_ids) = scope else {
continue;
};
channel_ids.retain(|candidate| *candidate != channel_id);
let removed = channel_ids.is_empty();
self.remove_from_index(
conn_id,
&sub_id,
&filters,
community_id,
&SubscriptionScope::Channels(vec![channel_id]),
);
if removed {
conn_subs.remove(&sub_id);
metrics::gauge!("buzz_subscriptions_active").decrement(1.0);
}
updates.push(ChannelSubscriptionUpdate { sub_id, removed });
}
updates
}
/// Test-only convenience wrapper preserving the original single-tenant test API.
@@ -242,7 +342,8 @@ impl SubscriptionRegistry {
pub fn remove_channel_subscriptions(&self, conn_id: ConnId, channel_id: Uuid) -> Vec<SubId> {
self.remove_channel_subscriptions_scoped(test_community(), conn_id, channel_id)
.into_iter()
.map(|(sub_id, _)| sub_id)
.filter(|update| update.removed)
.map(|update| update.sub_id)
.collect()
}
@@ -441,12 +542,12 @@ impl SubscriptionRegistry {
seen: &mut HashSet<(ConnId, SubId)>,
) {
if let Some(conn_subs) = self.subs.get(&conn_id) {
if let Some((filters, sub_community_id, sub_channel_id)) = conn_subs.get(sub_id) {
if let Some((filters, sub_community_id, scope)) = conn_subs.get(sub_id) {
// Candidate snapshots can become stale while a same-ID replacement
// moves the subscription. Re-check its authoritative scope before
// matching so an old index entry cannot deliver across scopes.
if *sub_community_id == community_id
&& *sub_channel_id == event.channel_id
&& scope.matches_channel(event.channel_id)
&& filters_match(filters, event)
{
let entry = (conn_id, sub_id.to_string());
@@ -466,42 +567,45 @@ impl SubscriptionRegistry {
sub_id: &str,
filters: &[Filter],
community_id: CommunityId,
channel_id: Option<Uuid>,
scope: &SubscriptionScope,
) {
if let Some(ch_id) = channel_id {
match extract_kinds_from_filters(filters) {
// None = wildcard (at least one filter had no kinds constraint).
None => {
// Was in wildcard index.
if let Some(mut entries) =
self.channel_wildcard_index.get_mut(&(community_id, ch_id))
{
entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
if entries.is_empty() {
drop(entries);
self.channel_wildcard_index.remove(&(community_id, ch_id));
}
}
}
Some(kinds) if kinds.is_empty() => {
// `kinds: []` subscriptions are never indexed (they match nothing),
// so there is nothing to remove here.
}
Some(kinds) => {
// Was in kind-specific index.
for kind in kinds {
let key = IndexKey {
channel_id: ch_id,
kind,
};
if let Some(mut entries) = self
.channel_kind_index
.get_mut(&(community_id, key.clone()))
if let SubscriptionScope::Channels(channel_ids) = scope {
for ch_id in channel_ids {
let ch_id = *ch_id;
match extract_kinds_from_filters(filters) {
// None = wildcard (at least one filter had no kinds constraint).
None => {
// Was in wildcard index.
if let Some(mut entries) =
self.channel_wildcard_index.get_mut(&(community_id, ch_id))
{
entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
if entries.is_empty() {
drop(entries);
self.channel_kind_index.remove(&(community_id, key));
self.channel_wildcard_index.remove(&(community_id, ch_id));
}
}
}
Some(kinds) if kinds.is_empty() => {
// `kinds: []` subscriptions are never indexed (they match nothing),
// so there is nothing to remove here.
}
Some(kinds) => {
// Was in kind-specific index.
for kind in kinds {
let key = IndexKey {
channel_id: ch_id,
kind,
};
if let Some(mut entries) = self
.channel_kind_index
.get_mut(&(community_id, key.clone()))
{
entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
if entries.is_empty() {
drop(entries);
self.channel_kind_index.remove(&(community_id, key));
}
}
}
}
@@ -686,6 +790,49 @@ mod tests {
assert_eq!(matches[0].1, sub_id);
}
#[test]
fn multi_channel_subscription_fans_out_only_requested_channels() {
let registry = SubscriptionRegistry::new();
let conn_id = Uuid::new_v4();
let channel_a = Uuid::new_v4();
let channel_b = Uuid::new_v4();
let unrelated = Uuid::new_v4();
let sub_id = "multi-channel".to_string();
let filters = vec![Filter::new()
.kind(Kind::TextNote)
.custom_tag(
SingleLetterTag::lowercase(Alphabet::H),
channel_a.to_string(),
)
.custom_tag(
SingleLetterTag::lowercase(Alphabet::H),
channel_b.to_string(),
)];
registry.register_channels_scoped(
test_community(),
conn_id,
sub_id.clone(),
filters,
vec![channel_a, channel_b],
);
assert_eq!(
registry.fan_out(&make_stored_event(Kind::TextNote, Some(channel_a))),
vec![(conn_id, sub_id.clone())]
);
assert_eq!(
registry.fan_out(&make_stored_event(Kind::TextNote, Some(channel_b))),
vec![(conn_id, sub_id)]
);
assert!(registry
.fan_out(&make_stored_event(Kind::TextNote, Some(unrelated)))
.is_empty());
assert!(registry
.fan_out(&make_stored_event(Kind::TextNote, None))
.is_empty());
}
#[test]
fn test_subscription_registry_remove() {
let registry = SubscriptionRegistry::new();
@@ -1688,6 +1835,54 @@ mod tests {
);
}
#[test]
fn revoking_one_channel_keeps_multi_channel_subscription_live() {
let registry = SubscriptionRegistry::new();
let community = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
let conn = Uuid::new_v4();
let channel_a = Uuid::new_v4();
let channel_b = Uuid::new_v4();
let filters = vec![Filter::new().kind(Kind::TextNote)];
registry.register_channels_scoped(
community,
conn,
"multi".to_string(),
filters,
vec![channel_a, channel_b],
);
let updates = registry.remove_channel_subscriptions_scoped(community, conn, channel_a);
assert_eq!(
updates,
vec![ChannelSubscriptionUpdate {
sub_id: "multi".to_string(),
removed: false,
}]
);
assert!(registry
.fan_out_scoped(
community,
&make_stored_event(Kind::TextNote, Some(channel_a))
)
.is_empty());
assert_eq!(
registry.fan_out_scoped(
community,
&make_stored_event(Kind::TextNote, Some(channel_b))
),
vec![(conn, "multi".to_string())]
);
let updates = registry.remove_channel_subscriptions_scoped(community, conn, channel_b);
assert_eq!(
updates,
vec![ChannelSubscriptionUpdate {
sub_id: "multi".to_string(),
removed: true,
}]
);
}
#[test]
fn per_community_subscriptions_snapshot_is_correctly_scoped() {
// Verify that per_community_subscriptions() returns the correct
crates/buzz-test-client/tests/e2e_relay.rs
+78 -2
View File
@@ -72,8 +72,9 @@ fn nip98_post_header(keys: &Keys, url: &str, body: &str) -> String {
}
async fn e2e_db_pool() -> sqlx::Pool<sqlx::Postgres> {
let database_url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string());
let database_url = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
"postgres://buzz:buzz_dev@localhost:5432/buzz".to_string() // sadscan:disable np.postgres.1
});
sqlx::postgres::PgPoolOptions::new()
.max_connections(1)
.connect(&database_url)
@@ -721,6 +722,81 @@ async fn test_stored_events_returned_before_eose() {
client.disconnect().await.expect("disconnect");
}
/// An explicit `#h` branch that cannot match must not cancel a valid OR sibling.
/// The valid channel remains usable for historical delivery and live fan-out;
/// malformed-only requests still close because no authorized UUID survives.
#[tokio::test]
#[ignore]
async fn test_valid_channel_survives_malformed_or_empty_h_sibling() {
let url = relay_url();
let kind: u16 = 9;
let keys = Keys::generate();
let channel = create_test_channel(&keys).await;
let mut client = BuzzTestClient::connect(&url, &keys).await.expect("connect");
for (label, sibling) in [
(
"malformed",
serde_json::json!({"kinds": [kind], "#h": ["not-a-uuid"]}),
),
("empty", serde_json::json!({"kinds": [kind], "#h": []})),
] {
let historical = format!("{label}-historical-{}", Uuid::new_v4());
let ok = client
.send_text_message(&keys, &channel, &historical, kind)
.await
.expect("send historical event");
assert!(ok.accepted, "historical event rejected: {}", ok.message);
let valid = Filter::new()
.kind(Kind::Custom(kind))
.custom_tags(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
let sibling: Filter = serde_json::from_value(sibling).expect("parse sibling filter");
let sid = sub_id(label);
client
.subscribe(&sid, vec![valid, sibling])
.await
.expect("subscribe");
let events = client
.collect_until_eose(&sid, Duration::from_secs(5))
.await
.expect("valid sibling history followed by EOSE");
assert!(
events.iter().any(|event| event.content == historical),
"valid sibling history missing for {label} #h branch: {events:?}",
);
let live = format!("{label}-live-{}", Uuid::new_v4());
let ok = client
.send_text_message(&keys, &channel, &live, kind)
.await
.expect("send live event");
assert!(ok.accepted, "live event rejected: {}", ok.message);
let message = client
.recv_event(Duration::from_secs(5))
.await
.expect("receive post-EOSE live event");
match message {
RelayMessage::Event {
subscription_id,
event,
} => {
assert_eq!(subscription_id, sid);
assert_eq!(event.content, live);
}
other => panic!("expected live EVENT for {label} sibling, got {other:?}"),
}
client
.close_subscription(&sid)
.await
.expect("close subscription");
}
client.disconnect().await.expect("disconnect");
}
/// Ephemeral events (kind 20000–29999) must be accepted but not persisted.
#[tokio::test]
#[ignore]
desktop/src-tauri/src/commands/workflows.rs
+58 -17
View File
@@ -1,3 +1,5 @@
use std::collections::HashSet;
use serde::Serialize;
use serde_json::Value;
use tauri::State;
@@ -103,34 +105,73 @@ pub async fn get_channel_workflows(
Ok(events.iter().map(workflow_from_event).collect())
}
/// Fetch workflows across many channels in a single relay round-trip.
// Keep this aligned with the relay's aggregate explicit-`#h` request bound.
// Each filter below carries exactly one explicit value so old relays retain the
// known-compatible shape while current relays cannot reject large memberships.
const WORKFLOW_QUERY_CHANNEL_BATCH_SIZE: usize = 128;
/// Fetch workflows across many channels using bounded relay round-trips.
///
/// The Workflows overview screen previously issued one `get_channel_workflows`
/// query per member channel (`Promise.all` fanout in `WorkflowsView`), i.e. N
/// relay POSTs. A nostr `#h` filter matches ANY of its listed values, so one
/// query with all channel ids returns the same set. Each `WorkflowWire` carries
/// its own `channel_id` (from the event's `h` tag), so the frontend can still
/// group results by channel. Neither this nor the per-channel command sets a
/// `limit`, so batching does not change result completeness.
/// relay POSTs. This sends one single-channel filter per channel, in requests of
/// at most 128 filters. Using one multi-value `#h` filter is equivalent under
/// NIP-01, but older relays incorrectly narrowed that shape to its first
/// channel. Each `WorkflowWire` carries its own `channel_id` (from the event's
/// `h` tag), so the frontend can still group results by channel. Neither this
/// nor the per-channel command sets a `limit`, so batching does not change
/// result completeness. Results are deduplicated by signed event ID in case a
/// caller supplies duplicate channel IDs.
#[tauri::command]
pub async fn get_channels_workflows(
channel_ids: Vec<String>,
state: State<'_, AppState>,
) -> Result<Vec<WorkflowWire>, String> {
if channel_ids.is_empty() {
return Ok(Vec::new());
let filter_batches = channel_workflow_filter_batches(channel_ids)?;
let mut seen_event_ids = HashSet::new();
let mut workflows = Vec::new();
for filters in filter_batches {
let events = query_relay(&state, &filters).await?;
append_unique_workflows(&mut workflows, &mut seen_event_ids, &events);
}
let events = query_relay(
&state,
&[serde_json::json!({
"kinds": [30620],
"#h": channel_ids,
})],
)
.await?;
Ok(workflows)
}
Ok(events.iter().map(workflow_from_event).collect())
fn append_unique_workflows(
workflows: &mut Vec<WorkflowWire>,
seen_event_ids: &mut HashSet<nostr::EventId>,
events: &[nostr::Event],
) {
workflows.extend(
events
.iter()
.filter(|event| seen_event_ids.insert(event.id))
.map(workflow_from_event),
);
}
fn channel_workflow_filter_batches(channel_ids: Vec<String>) -> Result<Vec<Vec<Value>>, String> {
let filters = channel_workflow_filters(channel_ids)?;
Ok(filters
.chunks(WORKFLOW_QUERY_CHANNEL_BATCH_SIZE)
.map(<[Value]>::to_vec)
.collect())
}
fn channel_workflow_filters(channel_ids: Vec<String>) -> Result<Vec<Value>, String> {
channel_ids
.into_iter()
.map(|channel_id| {
let channel_id = uuid::Uuid::parse_str(channel_id.trim())
.map_err(|_| "invalid channel id".to_string())?;
Ok(serde_json::json!({
"kinds": [30620],
"#h": [channel_id.to_string()],
}))
})
.collect()
}
#[tauri::command]
desktop/src-tauri/src/commands/workflows_tests.rs
+75
View File
@@ -192,6 +192,81 @@ fn workflow_wire_serializes_with_snake_case_keys() {
}
}
#[test]
fn multi_channel_workflow_query_uses_one_filter_per_channel() {
let other_channel = "33333333-3333-3333-3333-333333333333";
let filters = channel_workflow_filters(vec![CHAN.to_string(), other_channel.to_string()])
.expect("valid channels");
assert_eq!(filters.len(), 2);
assert_eq!(
filters[0],
serde_json::json!({
"kinds": [30620],
"#h": [CHAN],
})
);
assert_eq!(
filters[1],
serde_json::json!({
"kinds": [30620],
"#h": [other_channel],
})
);
}
#[test]
fn workflow_queries_batch_above_relay_explicit_channel_limit() {
let channel_ids = (0..WORKFLOW_QUERY_CHANNEL_BATCH_SIZE + 1)
.map(|index| uuid::Uuid::from_u128(index as u128 + 1).to_string())
.collect();
let batches = channel_workflow_filter_batches(channel_ids).expect("valid channels");
assert_eq!(batches.len(), 2);
assert_eq!(batches[0].len(), WORKFLOW_QUERY_CHANNEL_BATCH_SIZE);
assert_eq!(batches[1].len(), 1);
assert!(batches.iter().flatten().all(|filter| filter["#h"]
.as_array()
.is_some_and(|values| values.len() == 1)));
}
#[test]
fn workflow_query_results_are_deduplicated_by_event_id() {
let first = wf_event(WF, CHAN, YAML);
let second_workflow = "33333333-3333-3333-3333-333333333333";
let second = wf_event(second_workflow, CHAN, YAML);
let mut workflows = Vec::new();
let mut seen_event_ids = HashSet::new();
append_unique_workflows(
&mut workflows,
&mut seen_event_ids,
&[first.clone(), second.clone()],
);
append_unique_workflows(&mut workflows, &mut seen_event_ids, &[first, second]);
assert_eq!(workflows.len(), 2);
assert_eq!(workflows[0].id, WF);
assert_eq!(workflows[1].id, second_workflow);
}
#[test]
fn channel_workflow_filters_reject_malformed_or_blank_channel_ids() {
for channel_id in ["not-a-uuid", "", " "] {
let error = channel_workflow_filters(vec![channel_id.to_string()])
.expect_err("malformed channel id must fail before querying the relay");
assert_eq!(error, "invalid channel id");
}
}
#[test]
fn channel_workflow_filters_accepts_empty_input() {
assert_eq!(
channel_workflow_filters(Vec::new()).expect("empty input is valid"),
Vec::<Value>::new()
);
}
#[test]
fn trigger_response_uses_persisted_run_id_contract() {
let wire = trigger_wire_from_message(