mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
fix(relay): confirm requested channels in one membership query (#8378)
## What A request that names channels with `#h` used to check each channel missing from the cached access set with its own `is_member` query. These checks ran one after another, so the database cost of a request grew with the number of channels it named. That cost is why the relay capped explicit `#h` values at 128, and why mobile splits large communities into many requests ([#8342](https://github.com/block/buzz/pull/8342)). This PR makes the check one query, and then raises the cap. - **One membership query.** WS REQ, WS COUNT, and HTTP `/query` and `/count` now share `confirm_requested_channel_access`. It checks every unconfirmed channel with one `membership_pairs` statement (`channel_id = ANY($2)`), so one request costs one query, whatever number of channels it names. `membership_pairs` uses the same SQL rules as `is_member`: deleted channels and removed members don't count, and it runs on the writer pool. - **Hash-set lookups.** Comparing the requested channels with the accessible set no longer costs requested × accessible comparisons. This covers `apply_channel_scope_to_query`, the system-channel scope helpers, the COUNT and bridge loops, and `#h` dedupe. - **`MAX_EXPLICIT_CHANNEL_VALUES` goes from 128 to 2048.** Huddle liveness used the same constant to limit session IDs. It now has its own constant, `MAX_HUDDLE_LIVENESS_SESSIONS`, which stays at 128, because each linked session still costs a separate mesh-directory lookup. Authorization rules are unchanged: - A scoped token still sets the upper limit. A channel outside the token is never checked. - Channels already cached and readable system channels are skipped. - Repaired channels are added in request order, each once, so `keep_repaired_system_channels_named` still sees the same repaired tail. - A database error still closes the request (or returns 500 over HTTP). One difference: the old loop could leave a partial repair if a later query failed, and that can no longer happen. ## Not in this PR - **Client chunk sizes.** `desktop/src/shared/api/relayClientShared.ts:39` and `mobile/lib/shared/relay/nostr_filters.dart:8` stay at 128. They can go up only after every relay they talk to runs this change. - **Per-connection limit on live channel scope.** Each authorized channel in a live WS subscription still gets a fan-out index entry for each kind, plus a Redis topic retain. Real membership limits this, not the cap. Still, an account in at least 2048 channels could now register up to 1024 × 2048 channel scopes on one connection; before, the most was 1024 × 128. If we want to keep the old worst case, the fix is a limit on the total channel scope per connection, not per REQ. ## Testing - `cargo clippy -p buzz-relay --tests` and `cargo fmt` are clean. - `cargo test -p buzz-relay --lib`: 1449 passed, 6 failed. All 6 are in `api::media::tests` and fail because the local Postgres on :5432 has no `buzz` role. Nothing in this change touches them. - I ran the Postgres lane (`scripts/postgres-test-run.sh`) for the bridge, req, count, group-state, and system-channel tests on local PG14: 55 passed. That includes the new `named_channels_missing_from_the_cache_are_confirmed_together`, which checks that the reader's channels missing from the cache are added in request order, each once, and that a stranger's channel stays out. - Not done yet: a live run against a local relay sending a REQ with more than 128 `#h` values. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Signed-off-by: Bradley Axen <baxen@squareup.com> Co-authored-by: murderbot <3754f8729004d95654c46dbab3129e4ab9ef05cc2534e2a3fbfc155983bd637b@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1965,30 +1965,24 @@ pub(crate) async fn repair_requested_channel_access(
|
||||
readable_system: &mut Vec<uuid::Uuid>,
|
||||
) -> Result<(), (StatusCode, Json<Value>)> {
|
||||
let repaired_from = accessible_channels.len();
|
||||
for filter in filters {
|
||||
let Some(requested) =
|
||||
let requested: Vec<uuid::Uuid> = filters
|
||||
.iter()
|
||||
.filter_map(|filter| {
|
||||
crate::handlers::req::extract_channel_ids_from_filters(std::slice::from_ref(filter))
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
for channel_id in requested {
|
||||
// A readable system channel stays out of the request-wide set.
|
||||
if accessible_channels.contains(&channel_id) || readable_system.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),
|
||||
);
|
||||
}
|
||||
}
|
||||
})
|
||||
.flatten()
|
||||
.collect();
|
||||
crate::handlers::req::confirm_requested_channel_access(
|
||||
state,
|
||||
tenant.community(),
|
||||
pubkey_bytes,
|
||||
&requested,
|
||||
None,
|
||||
accessible_channels,
|
||||
readable_system,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| internal_error(&format!("channel membership confirmation: {e}")))?;
|
||||
crate::handlers::system_channels::keep_repaired_system_channels_named(
|
||||
state,
|
||||
tenant.community(),
|
||||
@@ -2209,9 +2203,10 @@ async fn count_events_authed(
|
||||
std::slice::from_ref(filter),
|
||||
)
|
||||
.unwrap_or_default();
|
||||
let accessible: std::collections::HashSet<_> = accessible_channels.iter().collect();
|
||||
if !requested
|
||||
.iter()
|
||||
.any(|channel_id| accessible_channels.contains(channel_id))
|
||||
.any(|channel_id| accessible.contains(channel_id))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
//! NIP-45 COUNT handler — aggregate queries with channel access enforcement.
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
|
||||
use nostr::Filter;
|
||||
@@ -95,7 +96,7 @@ pub async fn handle_count(
|
||||
// this, a scoped token would COUNT events in channels outside its scope via
|
||||
// the no-channel-filter SQL pushdown below (which counts every accessible
|
||||
// channel). The per-filter targeted-channel repair is bounded by the same
|
||||
// scope through `resolve_request_local_access`'s `token_allows` argument.
|
||||
// scope through `confirm_requested_channel_access`'s `token_channel_ids` argument.
|
||||
if let Some(allowed) = token_channel_ids.as_deref() {
|
||||
accessible_channels.retain(|channel_id| allowed.contains(channel_id));
|
||||
}
|
||||
@@ -164,35 +165,20 @@ pub async fn handle_count(
|
||||
|
||||
if let Some(requested_channels) = requested_channels {
|
||||
let repaired_from = accessible_channels.len();
|
||||
for &ch_id in &requested_channels {
|
||||
if accessible_channels.contains(&ch_id) || readable_system.contains(&ch_id) {
|
||||
continue;
|
||||
}
|
||||
let token_allows = token_channel_ids
|
||||
.as_deref()
|
||||
.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,
|
||||
);
|
||||
if let Err(e) = super::req::confirm_requested_channel_access(
|
||||
&state,
|
||||
conn.tenant.community(),
|
||||
&pubkey_bytes,
|
||||
&requested_channels,
|
||||
token_channel_ids.as_deref(),
|
||||
&mut accessible_channels,
|
||||
&readable_system,
|
||||
)
|
||||
.await
|
||||
{
|
||||
warn!(sub_id = %sub_id, "Channel membership confirmation failed: {e}");
|
||||
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
|
||||
return;
|
||||
}
|
||||
if let Err(e) = super::system_channels::keep_repaired_system_channels_named(
|
||||
&state,
|
||||
@@ -212,10 +198,11 @@ pub async fn handle_count(
|
||||
&accessible_channels,
|
||||
&readable_system,
|
||||
);
|
||||
let scope: HashSet<_> = filter_scope.iter().collect();
|
||||
let authorized_requested: Vec<_> = requested_channels
|
||||
.iter()
|
||||
.copied()
|
||||
.filter(|channel_id| filter_scope.contains(channel_id))
|
||||
.filter(|channel_id| scope.contains(channel_id))
|
||||
.collect();
|
||||
if authorized_requested.is_empty() {
|
||||
continue;
|
||||
|
||||
@@ -25,6 +25,19 @@ use crate::state::AppState;
|
||||
|
||||
const MAX_SUBSCRIPTIONS: usize = 1024;
|
||||
|
||||
/// Maximum aggregate number of explicit `#h` values accepted in one REQ,
|
||||
/// COUNT, HTTP `/query`, or HTTP `/count` request.
|
||||
///
|
||||
/// Cache-missed channels are confirmed in one membership query, so the DB cost
|
||||
/// does not grow with the count. A live WS subscription still gives each
|
||||
/// authorized channel a fan-out index entry and a Redis topic retain; the
|
||||
/// reader's real membership bounds that part.
|
||||
pub(crate) const MAX_EXPLICIT_CHANNEL_VALUES: usize = 2048;
|
||||
|
||||
/// Maximum huddle sessions one liveness REQ looks up. Each linked session
|
||||
/// costs a sequential mesh-directory lookup.
|
||||
const MAX_HUDDLE_LIVENESS_SESSIONS: usize = 128;
|
||||
|
||||
/// Maximum `query_events` calls in flight per multi-filter REQ / bridge query.
|
||||
///
|
||||
/// NIP-01 gives each filter its own DB query (OR semantics — see the comment at
|
||||
@@ -34,14 +47,6 @@ 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
|
||||
@@ -161,54 +166,33 @@ pub async fn handle_req(
|
||||
// on the pod that processed the write, before the 10s TTL expires or the
|
||||
// cross-pod invalidation lands), so on a miss we confirm uncached against
|
||||
// the DB; a verified positive repairs the vector request-locally (see
|
||||
// `resolve_request_local_access`). Running this ahead of the search branch
|
||||
// `confirm_requested_channel_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(requested) = requested_channel_ids.as_ref() {
|
||||
let repaired_from = accessible_channels.len();
|
||||
for &ch_id in requested {
|
||||
let token_allows = token_channel_ids
|
||||
.as_deref()
|
||||
.is_none_or(|allowed| allowed.contains(&ch_id));
|
||||
// A readable system channel is authorized for the filters that
|
||||
// name it; it must not enter the request-wide vector.
|
||||
let db_is_member = if !token_allows
|
||||
|| accessible_channels.contains(&ch_id)
|
||||
|| readable_system.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;
|
||||
}
|
||||
}
|
||||
};
|
||||
// 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 confirmed = match confirm_requested_channel_access(
|
||||
&state,
|
||||
conn.tenant.community(),
|
||||
&pubkey_bytes,
|
||||
requested,
|
||||
token_channel_ids.as_deref(),
|
||||
&mut accessible_channels,
|
||||
&readable_system,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(confirmed) => confirmed,
|
||||
Err(e) => {
|
||||
warn!(conn_id = %conn_id, "Channel membership confirmation failed: {e}");
|
||||
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
|
||||
return;
|
||||
}
|
||||
};
|
||||
if let Some(state_snap) = trace_state.as_ref() {
|
||||
for (ch_id, member) in confirmed {
|
||||
crate::conformance::record_req_authcheck(&state.tracer, state_snap, ch_id, member);
|
||||
}
|
||||
}
|
||||
if let Err(e) = super::system_channels::keep_repaired_system_channels_named(
|
||||
&state,
|
||||
@@ -226,12 +210,15 @@ pub async fn handle_req(
|
||||
}
|
||||
|
||||
let authorized_requested_channels = requested_channel_ids.as_ref().map(|requested| {
|
||||
let authorized: HashSet<_> = accessible_channels
|
||||
.iter()
|
||||
.chain(&readable_system)
|
||||
.copied()
|
||||
.collect();
|
||||
requested
|
||||
.iter()
|
||||
.copied()
|
||||
.filter(|channel_id| {
|
||||
accessible_channels.contains(channel_id) || readable_system.contains(channel_id)
|
||||
})
|
||||
.filter(|channel_id| authorized.contains(channel_id))
|
||||
.collect::<Vec<_>>()
|
||||
});
|
||||
// Partial authorization preserves NIP-01 OR semantics by omitting only
|
||||
@@ -647,50 +634,91 @@ const SEARCH_PAGE_SIZE: u32 = 100;
|
||||
/// the response.
|
||||
const MAX_SEARCH_PAGES: u32 = (buzz_db::DEFAULT_MAX_PAGE_LIMIT as u32).div_ceil(SEARCH_PAGE_SIZE);
|
||||
|
||||
/// Resolve request-local channel access, repairing a stale cache-negative.
|
||||
/// Requested channels whose access needs a DB membership check: not already
|
||||
/// in the cached `accessible_channels` or `readable_system` sets, and inside
|
||||
/// the scoped-auth `token_channel_ids` bound when one is present. Deduplicated,
|
||||
/// in request order.
|
||||
///
|
||||
/// The token bound is an upper bound on the repair: a channel outside it is
|
||||
/// never checked, so a DB positive cannot push it back in. Otherwise a token
|
||||
/// scoped to channel A could reach channel B merely because the user is a DB
|
||||
/// member of B.
|
||||
pub(crate) fn channels_needing_membership_check(
|
||||
requested: &[uuid::Uuid],
|
||||
accessible_channels: &[uuid::Uuid],
|
||||
readable_system: &[uuid::Uuid],
|
||||
token_channel_ids: Option<&[uuid::Uuid]>,
|
||||
) -> Vec<uuid::Uuid> {
|
||||
let mut known: HashSet<uuid::Uuid> = accessible_channels
|
||||
.iter()
|
||||
.chain(readable_system)
|
||||
.copied()
|
||||
.collect();
|
||||
let token: Option<HashSet<uuid::Uuid>> =
|
||||
token_channel_ids.map(|allowed| allowed.iter().copied().collect());
|
||||
requested
|
||||
.iter()
|
||||
.copied()
|
||||
.filter(|channel_id| {
|
||||
token
|
||||
.as_ref()
|
||||
.is_none_or(|allowed| allowed.contains(channel_id))
|
||||
})
|
||||
.filter(|channel_id| known.insert(*channel_id))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Confirm requested channels against the DB, repairing stale cache-negatives
|
||||
/// request-locally. Returns each checked channel with its membership answer.
|
||||
///
|
||||
/// `accessible_channels` is the per-request membership vector — built once from
|
||||
/// the 10s cache (and already narrowed by any scoped-auth `token_channel_ids`
|
||||
/// via `retain`) and reused for subscription registration, historical delivery,
|
||||
/// search scope, and COUNT. On a multi-pod relay it can be stale on a non-writer
|
||||
/// pod (a member just added on another pod, before the TTL expires or the
|
||||
/// cross-pod invalidation lands), so the cache-negative branch confirms against
|
||||
/// the DB uncached and passes the result here.
|
||||
/// cross-pod invalidation lands), so every requested channel the cache does not
|
||||
/// grant is confirmed uncached. All of them go to the DB in one statement, so a
|
||||
/// request costs one round trip however many channels it names.
|
||||
///
|
||||
/// `token_allows` is the scoped-auth upper bound: `false` when a scoped token is
|
||||
/// present and does NOT cover `ch_id`. The DB-positive repair must never push a
|
||||
/// channel back in past that bound, or a token scoped to channel A could reach
|
||||
/// channel B merely because the user is a DB member of B.
|
||||
///
|
||||
/// Truth table:
|
||||
/// - token denies `ch_id` → denied, no DB needed, no repair
|
||||
/// - cached contains `ch_id` → allowed, no repair, no DB needed
|
||||
/// - cache-miss + DB says member → allowed, `ch_id` pushed once (repair)
|
||||
/// - cache-miss + DB says not a member → denied, vector unchanged
|
||||
///
|
||||
/// The push is what makes the confirmation request-local-authoritative: every
|
||||
/// downstream consumer reads the same repaired vector, so a stale negative
|
||||
/// cannot stay sticky for the rest of the request. `db_is_member` is `None` when
|
||||
/// the cache hit or the token bound denied (DB was never consulted).
|
||||
pub(crate) fn resolve_request_local_access(
|
||||
/// A confirmed member is pushed onto `accessible_channels`. The push is what
|
||||
/// makes the confirmation request-local-authoritative: every downstream
|
||||
/// consumer reads the same repaired vector, so a stale negative cannot stay
|
||||
/// sticky for the rest of the request.
|
||||
pub(crate) async fn confirm_requested_channel_access(
|
||||
state: &AppState,
|
||||
community: buzz_core::tenant::CommunityId,
|
||||
pubkey: &[u8],
|
||||
requested: &[uuid::Uuid],
|
||||
token_channel_ids: Option<&[uuid::Uuid]>,
|
||||
accessible_channels: &mut Vec<uuid::Uuid>,
|
||||
ch_id: uuid::Uuid,
|
||||
token_allows: bool,
|
||||
db_is_member: Option<bool>,
|
||||
) -> bool {
|
||||
if !token_allows {
|
||||
return false;
|
||||
}
|
||||
if accessible_channels.contains(&ch_id) {
|
||||
return true;
|
||||
}
|
||||
match db_is_member {
|
||||
Some(true) => {
|
||||
accessible_channels.push(ch_id);
|
||||
true
|
||||
}
|
||||
_ => false,
|
||||
readable_system: &[uuid::Uuid],
|
||||
) -> buzz_db::Result<Vec<(uuid::Uuid, bool)>> {
|
||||
let unconfirmed = channels_needing_membership_check(
|
||||
requested,
|
||||
accessible_channels,
|
||||
readable_system,
|
||||
token_channel_ids,
|
||||
);
|
||||
if unconfirmed.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let members: HashSet<uuid::Uuid> = state
|
||||
.db
|
||||
.membership_pairs(community, &unconfirmed, &[pubkey.to_vec()])
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|(channel_id, _)| channel_id)
|
||||
.collect();
|
||||
Ok(unconfirmed
|
||||
.into_iter()
|
||||
.map(|channel_id| {
|
||||
let member = members.contains(&channel_id);
|
||||
if member {
|
||||
accessible_channels.push(channel_id);
|
||||
}
|
||||
(channel_id, member)
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Map the legacy `(accessible_channels, include_global)` pair onto the
|
||||
@@ -1295,11 +1323,12 @@ pub(crate) fn apply_channel_scope_to_query(
|
||||
|
||||
let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H);
|
||||
if let Some(values) = filter.generic_tags.get(&h_tag) {
|
||||
let accessible: HashSet<_> = accessible_channels.iter().collect();
|
||||
query.channel_ids = Some(
|
||||
values
|
||||
.iter()
|
||||
.filter_map(|value| value.parse::<uuid::Uuid>().ok())
|
||||
.filter(|requested| accessible_channels.contains(requested))
|
||||
.filter(|requested| accessible.contains(requested))
|
||||
.collect(),
|
||||
);
|
||||
query.channel_ids_include_global = false;
|
||||
@@ -1336,12 +1365,13 @@ pub(crate) fn extract_channel_ids_from_filters_limited(
|
||||
/// [`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 seen = HashSet::new();
|
||||
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) {
|
||||
if seen.insert(channel_id) {
|
||||
channel_ids.push(channel_id);
|
||||
}
|
||||
}
|
||||
@@ -1376,7 +1406,7 @@ fn huddle_liveness_session_ids(filters: &[Filter]) -> Vec<uuid::Uuid> {
|
||||
}
|
||||
}
|
||||
}
|
||||
session_ids.truncate(MAX_EXPLICIT_CHANNEL_VALUES);
|
||||
session_ids.truncate(MAX_HUDDLE_LIVENESS_SESSIONS);
|
||||
session_ids
|
||||
}
|
||||
|
||||
@@ -2373,7 +2403,7 @@ mod tests {
|
||||
#[test]
|
||||
fn huddle_liveness_session_ids_are_deduplicated_and_bounded() {
|
||||
let d_tag = SingleLetterTag::lowercase(Alphabet::D);
|
||||
let input = (0..MAX_EXPLICIT_CHANNEL_VALUES + 16)
|
||||
let input = (0..MAX_HUDDLE_LIVENESS_SESSIONS + 16)
|
||||
.map(|_| uuid::Uuid::new_v4())
|
||||
.collect::<Vec<_>>();
|
||||
let first = input.iter().fold(Filter::new(), |filter, session_id| {
|
||||
@@ -2387,7 +2417,7 @@ mod tests {
|
||||
let extracted_set = extracted.iter().copied().collect::<HashSet<_>>();
|
||||
let input_set = input.iter().copied().collect::<HashSet<_>>();
|
||||
|
||||
assert_eq!(extracted.len(), MAX_EXPLICIT_CHANNEL_VALUES);
|
||||
assert_eq!(extracted.len(), MAX_HUDDLE_LIVENESS_SESSIONS);
|
||||
assert_eq!(extracted_set.len(), extracted.len());
|
||||
assert!(extracted_set.is_subset(&input_set));
|
||||
}
|
||||
@@ -2454,71 +2484,46 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_local_access_cache_positive_no_db_no_repair() {
|
||||
let ch = uuid::Uuid::new_v4();
|
||||
let mut accessible = vec![ch];
|
||||
// Cache hit: DB was never consulted (None), allowed, vector unchanged.
|
||||
assert!(resolve_request_local_access(
|
||||
&mut accessible,
|
||||
ch,
|
||||
true,
|
||||
None
|
||||
));
|
||||
assert_eq!(accessible, vec![ch], "no repair, no duplicate on cache hit");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_local_access_cache_negative_db_member_repairs() {
|
||||
let ch = uuid::Uuid::new_v4();
|
||||
let mut accessible: Vec<uuid::Uuid> = vec![];
|
||||
// Stale cache-miss but DB confirms membership: allowed AND repaired.
|
||||
assert!(resolve_request_local_access(
|
||||
&mut accessible,
|
||||
ch,
|
||||
true,
|
||||
Some(true)
|
||||
));
|
||||
assert!(
|
||||
accessible.contains(&ch),
|
||||
"verified positive must push ch_id so the rest of the request sees it"
|
||||
fn membership_check_skips_cached_and_readable_system_channels() {
|
||||
let cached = uuid::Uuid::new_v4();
|
||||
let system = uuid::Uuid::new_v4();
|
||||
let missed = uuid::Uuid::new_v4();
|
||||
assert_eq!(
|
||||
channels_needing_membership_check(
|
||||
&[cached, system, missed],
|
||||
&[cached],
|
||||
&[system],
|
||||
None
|
||||
),
|
||||
vec![missed],
|
||||
"only the cache-miss needs the DB"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_local_access_cache_negative_db_nonmember_denied() {
|
||||
let ch = uuid::Uuid::new_v4();
|
||||
let mut accessible: Vec<uuid::Uuid> = vec![];
|
||||
// Cache-miss and DB confirms non-membership: denied, vector unchanged.
|
||||
assert!(!resolve_request_local_access(
|
||||
&mut accessible,
|
||||
ch,
|
||||
true,
|
||||
Some(false)
|
||||
));
|
||||
assert!(
|
||||
accessible.is_empty(),
|
||||
"denied access must not mutate the request-local vector"
|
||||
fn membership_check_asks_once_per_channel_in_request_order() {
|
||||
let a = uuid::Uuid::new_v4();
|
||||
let b = uuid::Uuid::new_v4();
|
||||
assert_eq!(
|
||||
channels_needing_membership_check(&[b, a, b, a], &[], &[], None),
|
||||
vec![b, a]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_local_access_token_denies_never_repairs() {
|
||||
let ch = uuid::Uuid::new_v4();
|
||||
let mut accessible: Vec<uuid::Uuid> = vec![];
|
||||
// Scoped token does NOT cover ch_id: denied even though the DB confirms
|
||||
// membership. The token scope is an upper bound on the repair — a DB
|
||||
// positive must never push a channel back in past a narrower token, or
|
||||
// a token scoped to channel A could reach channel B merely because the
|
||||
// user is a DB member of B.
|
||||
assert!(!resolve_request_local_access(
|
||||
&mut accessible,
|
||||
ch,
|
||||
false,
|
||||
Some(true)
|
||||
));
|
||||
assert!(
|
||||
accessible.is_empty(),
|
||||
"token-denied access must not be repaired into the vector"
|
||||
fn membership_check_never_asks_past_the_token_bound() {
|
||||
let in_scope = uuid::Uuid::new_v4();
|
||||
let out_of_scope = uuid::Uuid::new_v4();
|
||||
// A scoped token is an upper bound on the repair: a channel outside it
|
||||
// is never checked, so a DB positive cannot push it back in.
|
||||
assert_eq!(
|
||||
channels_needing_membership_check(
|
||||
&[in_scope, out_of_scope],
|
||||
&[],
|
||||
&[],
|
||||
Some(&[in_scope]),
|
||||
),
|
||||
vec![in_scope]
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -388,6 +388,46 @@ async fn a_system_channel_joined_mid_request_stays_in_its_named_filter() {
|
||||
assert_eq!(scope(&filters[1]), [ordinary], "sibling filter");
|
||||
}
|
||||
|
||||
/// A request that names channels the cached access set lacks confirms them
|
||||
/// all in one membership query. Members join the request-local set in request
|
||||
/// order, each once; a channel the reader is not in stays out.
|
||||
/// Mutation: push every unconfirmed channel in
|
||||
/// `confirm_requested_channel_access` → RED (the stranger's channel gets in);
|
||||
/// skip the push → RED (the reader's channels stay out).
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres and Redis"]
|
||||
async fn named_channels_missing_from_the_cache_are_confirmed_together() {
|
||||
let f = fixture().await;
|
||||
let reader = Keys::generate();
|
||||
let stranger = Keys::generate();
|
||||
let first = create_channel(&f.state, &f.tenant, &reader, &[]).await;
|
||||
let second = create_channel(&f.state, &f.tenant, &reader, &[]).await;
|
||||
let foreign = create_channel(&f.state, &f.tenant, &stranger, &[]).await;
|
||||
|
||||
let filters: Vec<nostr::Filter> = serde_json::from_value(json!([
|
||||
{"kinds": [9], "#h": [second.to_string(), foreign.to_string()]},
|
||||
{"kinds": [9], "#h": [first.to_string(), second.to_string()]},
|
||||
]))
|
||||
.expect("filters");
|
||||
// The request's cached lookup ran before the reader joined either channel.
|
||||
let mut implicit = Vec::new();
|
||||
let mut readable_system = Vec::new();
|
||||
|
||||
crate::api::bridge::repair_requested_channel_access(
|
||||
&f.state,
|
||||
&f.tenant,
|
||||
&filters,
|
||||
&reader.public_key().to_bytes(),
|
||||
&mut implicit,
|
||||
&mut readable_system,
|
||||
)
|
||||
.await
|
||||
.expect("repair");
|
||||
|
||||
assert_eq!(implicit, [second, first]);
|
||||
assert!(readable_system.is_empty());
|
||||
}
|
||||
|
||||
/// Adding a member to a system channel sends no kind:44100 notification,
|
||||
/// which an existing client would show as a new channel. An ordinary
|
||||
/// channel still sends one, so the check is not vacuous.
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
//! implicit set, so naming one system channel cannot widen a sibling filter.
|
||||
|
||||
use std::borrow::Cow;
|
||||
use std::collections::HashSet;
|
||||
|
||||
use nostr::{Alphabet, Filter, SingleLetterTag};
|
||||
use uuid::Uuid;
|
||||
@@ -81,7 +82,11 @@ fn filter_opts_in(filter: &Filter) -> bool {
|
||||
|
||||
/// Whether `filter` could need a system channel added to its scope.
|
||||
fn filter_needs_system_channels(filter: &Filter, implicit: &[Uuid]) -> bool {
|
||||
filter_opts_in(filter) || named_channels(filter).any(|id| !implicit.contains(&id))
|
||||
if filter_opts_in(filter) {
|
||||
return true;
|
||||
}
|
||||
let implicit: HashSet<Uuid> = implicit.iter().copied().collect();
|
||||
named_channels(filter).any(|id| !implicit.contains(&id))
|
||||
}
|
||||
|
||||
/// The channels `filter` may read: the implicit set, plus the readable
|
||||
@@ -101,8 +106,10 @@ pub(crate) fn filter_channel_scope<'a>(
|
||||
} else {
|
||||
named_channels(filter).collect()
|
||||
};
|
||||
let readable: HashSet<Uuid> = readable_system.iter().copied().collect();
|
||||
let mut seen: HashSet<Uuid> = implicit.iter().copied().collect();
|
||||
for id in candidates {
|
||||
if readable_system.contains(&id) && !implicit.contains(&id) && !extra.contains(&id) {
|
||||
if readable.contains(&id) && seen.insert(id) {
|
||||
extra.push(id);
|
||||
}
|
||||
}
|
||||
@@ -145,7 +152,7 @@ pub(crate) async fn readable_system_channels_for_filters(
|
||||
/// Move system channels out of the repaired tail of the implicit set.
|
||||
///
|
||||
/// A request re-checks membership for named channels missing from its first
|
||||
/// lookups, and [`super::req::resolve_request_local_access`] appends each
|
||||
/// lookups, and [`super::req::confirm_requested_channel_access`] appends each
|
||||
/// confirmed one to the implicit set. A system channel lands there when the
|
||||
/// reader joins it between the lookups and the re-check. This moves it to
|
||||
/// `readable_system`, so only the filters that name it can read it.
|
||||
|
||||
Reference in New Issue
Block a user