mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
fix(cli): enrich template cardinality error with per-candidate presence and profile hints (#4825)
## Problem
`buzz channels create --template` fails with a bare-pubkey
duplicate-instance error when a persona has more than one live instance.
When duplicate instances share the same display name and avatar, the
error gives the operator no signal to identify which instance is stale.
## What changed
**`crates/buzz-cli/src/commands/channels.rs`** (boundary) + one
visibility line in **`crates/buzz-cli/src/commands/users.rs`** (`fn
presence_subject` → `pub(crate)`), plus a relay behavior change in
**`crates/buzz-relay/src/api/bridge.rs`** and a supporting test in
**`crates/buzz-pubsub/src/presence.rs`**.
### Relay: presence-lookup failure is now surfaced as a failure
`synthesize_presence` previously collapsed a Redis `get_presence_bulk`
error into `unwrap_or_default()` → HTTP 200 `[]`, making a backend
outage indistinguishable from an authoritative all-offline snapshot. It
now returns `Option<Result<Vec<Value>, (StatusCode, Json<Value>)>>`:
`None` means the request is not a presence query (fall through),
`Some(Ok)` is a real snapshot (an empty vec is an authoritative
all-offline result), and `Some(Err)` is a backend failure surfaced as a
non-2xx response. The event tag-parse and sign paths were converted from
`.ok()?` fall-throughs to explicit `Some(Err(...))` for the same reason
— a silent fall-through would have reintroduced the fake-empty-success
anti-pattern. The function takes `pubsub` and `relay_keypair` directly
rather than the whole `AppState`, which keeps the Redis-error seam
unit-testable without live infrastructure; its single production caller
was updated accordingly, and no other caller consumes
`synthesize_presence`.
### Hint fetch architecture
**`assemble_roster_resolution<F, Fut>`** — post-fetch stage extracted
from `build_roster_resolution` as an injectable-fetcher async generic.
It contains: archive-filter-based duplicate detection, the conditional
`fetch_hints` call, and `finalize_roster_resolution` delegation.
Accepting the fetcher as a closure makes the wiring directly testable
without a relay.
`build_roster_resolution` reduces to: gather slugs → `tokio::join!(scan,
archive)` → delegate with the real `fetch_candidate_hints` closure.
**Zero hint queries on the happy path**: `assemble_roster_resolution`
only invokes the fetcher when duplicate live instances exist after
archive filtering. Untrusted archive snapshot (`Err`) conservatively
treats all found instances as live.
When duplicates exist, `fetch_candidate_hints` runs the presence
(kind:40902) and profile (kind:0) queries concurrently, bounding **each
independently** through `join_bounded_queries`: a per-query 3-second
timeout maps to `Err`, so a lookup that completes is never discarded
because its sibling hung. On total failure or timeout, bare pubkeys are
printed promptly.
### Trusted-snapshot boundary — `hints_from_results`
The relay drops the Redis presence key when an identity goes offline, so
the bulk presence snapshot never returns an event for an offline
instance — precisely the stale duplicate an operator needs flagged.
`hints_from_results` therefore treats a *trusted* presence snapshot as a
**complete snapshot** for the requested pubkeys: it seeds all of them
`offline`, then overlays returned statuses.
A response is trusted only when `trusted_presence_snapshot` validates
**every** array element as the actual snapshot contract: each must parse
as a complete signed `nostr::Event` of the presence-update kind carrying
exactly one `p` tag whose subject is one of the requested pubkeys.
Requiring the *sole* `p`-tag subject — the exact value `build_hint_map`
later reads via `presence_subject` — is what stops a mixed-tag event
(`[["p","<unrequested>"],["p","<requested>"]]`) from passing the gate
yet overlaying a different subject downstream. Any element that is not
such an event — a vacuous object like
`{"pubkey":"…","content":"online"}`, `[{}]`, `[null]`, an event of the
wrong kind, one for an unrequested subject, or one carrying more than
one `p` tag — makes presence enrichment untrusted: nothing is seeded
`offline`, so absence is never falsely inferred as offline. A failed,
timed-out, or non-array presence response is likewise untrusted. In
every untrusted case the successful profile sibling still contributes
its hints.
### Response-to-map conversion — `build_hint_map`
Sync production function taking the offline seed plus raw
presence/profile event slices. Relay-signed presence events carry the
agent pubkey in the `p` tag (not the event author); `build_hint_map`
calls `presence_subject` from `users.rs` (now `pub(crate)`) to resolve
the correct key.
### Hint display — `format_candidate`
Appends `[online/offline, profile updated YYYY-MM-DD]` (or subset) to
each candidate pubkey in the error. `profile_updated_at` names the field
correctly: kind:0 is replaceable state that desktop republishes on
rename and profile reconciliation, so `created_at` reflects the last
profile update, not provisioning time. Date formatted via
`chrono::DateTime::from_timestamp`; out-of-range timestamps omit the
date. Missing entries fall back to bare pubkeys.
### Cardinality rule
`apply_cardinality_rule` remains pure and relay-free; zero/one/many
semantics unchanged.
---------
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Carl <9d00794d3df50972eb8b615511783cab12a77a8fd5dd5edd58073ec73b54bd8b@buzz.block.builderlab.xyz>
Signed-off-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
Co-authored-by: npub1g8493u0xfsjrvflg4n08ezd7vec99mnwzlv0qgwpr9d7gvjwhuzqx59rhw <41ea58f1e64c243627e8acde7c89be667052ee6e17d8f021c1195be4324ebf04@buzz.block.builderlab.xyz>
Co-authored-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -91,3 +91,5 @@ rand = { workspace = true }
|
||||
tempfile = "3"
|
||||
# Minimal HTTP test server for retry/policy integration tests
|
||||
axum = { workspace = true }
|
||||
# `test-util` enables paused-time control for deterministic timeout tests
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] }
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -485,7 +485,7 @@ pub async fn cmd_get_presence(client: &BuzzClient, pubkeys_csv: &str) -> Result<
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn presence_subject(event: &serde_json::Value) -> &str {
|
||||
pub(crate) fn presence_subject(event: &serde_json::Value) -> &str {
|
||||
event
|
||||
.get("tags")
|
||||
.and_then(|tags| tags.as_array())
|
||||
|
||||
@@ -109,6 +109,26 @@ mod tests {
|
||||
TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(id)), host)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_presence_bulk_surfaces_connection_failure_as_error() {
|
||||
// A backend outage must surface as `Err`, not a silently-empty `Ok`.
|
||||
// `synthesize_presence` relies on this to return an error response
|
||||
// rather than a fake-empty "all offline" snapshot on a Redis failure.
|
||||
// Pool points at a closed port so the connection attempt fails.
|
||||
let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1")
|
||||
.create_pool(Some(deadpool_redis::Runtime::Tokio1))
|
||||
.expect("pool builds lazily");
|
||||
let ctx = ctx(0xaaaa, "a.example");
|
||||
let pubkey = make_pubkey();
|
||||
|
||||
let result = get_presence_bulk(&pool, &ctx, &[pubkey]).await;
|
||||
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"a connection failure must surface as Err, got {result:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn presence_ttl_is_three_one_minute_heartbeat_windows() {
|
||||
assert_eq!(PRESENCE_TTL_SECS, 180);
|
||||
|
||||
@@ -1125,8 +1125,10 @@ async fn query_events_authed(
|
||||
.await;
|
||||
}
|
||||
|
||||
if let Some(presence_events) = synthesize_presence(state, tenant, &filters).await {
|
||||
return Ok(Json(Value::Array(presence_events)));
|
||||
if let Some(presence_result) =
|
||||
synthesize_presence(&state.pubsub, &state.relay_keypair, tenant, &filters).await
|
||||
{
|
||||
return presence_result.map(|events| Json(Value::Array(events)));
|
||||
}
|
||||
|
||||
let mut events: Vec<Value> = Vec::new();
|
||||
@@ -2176,12 +2178,19 @@ pub async fn workflow_webhook(
|
||||
/// presence from Redis instead of querying the DB (ephemeral events are never
|
||||
/// stored, and kind:40902 snapshots are relay-generated on demand).
|
||||
///
|
||||
/// Returns `Some(events)` if handled, `None` to fall through to normal query.
|
||||
/// Returns `None` when the filters are not a presence query (fall through to
|
||||
/// the normal query path). Returns `Some(Ok(events))` when a presence snapshot
|
||||
/// was produced — an empty vec is an authoritative "all offline" answer.
|
||||
/// Returns `Some(Err(_))` when the backing Redis lookup failed: callers must
|
||||
/// propagate that as an error response rather than a fake-empty success, so a
|
||||
/// consumer cannot mistake a backend outage for an authoritative snapshot.
|
||||
#[allow(clippy::type_complexity)]
|
||||
async fn synthesize_presence(
|
||||
state: &AppState,
|
||||
pubsub: &buzz_pubsub::PubSubManager,
|
||||
relay_keypair: &nostr::Keys,
|
||||
tenant: &buzz_core::tenant::TenantContext,
|
||||
filters: &[nostr::Filter],
|
||||
) -> Option<Vec<Value>> {
|
||||
) -> Option<Result<Vec<Value>, (StatusCode, Json<Value>)>> {
|
||||
use buzz_core::kind::{KIND_PRESENCE_SNAPSHOT, KIND_PRESENCE_UPDATE};
|
||||
|
||||
// Only intercept if every filter targets kind:20001 or 40902 with authors.
|
||||
@@ -2201,22 +2210,23 @@ async fn synthesize_presence(
|
||||
}
|
||||
|
||||
if all_pubkeys.is_empty() {
|
||||
return Some(Vec::new());
|
||||
return Some(Ok(Vec::new()));
|
||||
}
|
||||
|
||||
// Dedup pubkeys.
|
||||
all_pubkeys.sort_by_key(|pk| pk.to_hex());
|
||||
all_pubkeys.dedup();
|
||||
|
||||
// Look up Redis.
|
||||
let presence_map = state
|
||||
.pubsub
|
||||
.get_presence_bulk(tenant, &all_pubkeys)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
// Look up Redis. A lookup failure must surface as an error, not a
|
||||
// fake-empty success — otherwise a Redis outage is indistinguishable from
|
||||
// an authoritative all-offline snapshot to the consumer.
|
||||
let presence_map = match pubsub.get_presence_bulk(tenant, &all_pubkeys).await {
|
||||
Ok(map) => map,
|
||||
Err(e) => return Some(Err(internal_error(&format!("presence lookup: {e}")))),
|
||||
};
|
||||
|
||||
if presence_map.is_empty() {
|
||||
return Some(Vec::new());
|
||||
return Some(Ok(Vec::new()));
|
||||
}
|
||||
|
||||
// Synthesize kind:20001 events signed by the relay.
|
||||
@@ -2228,20 +2238,30 @@ async fn synthesize_presence(
|
||||
let mut events = Vec::with_capacity(presence_map.len());
|
||||
for (pubkey_hex, status) in &presence_map {
|
||||
// Build a synthetic event: relay-signed, content = status, p-tag = subject.
|
||||
let tags = vec![nostr::Tag::parse(["p", pubkey_hex]).ok()?];
|
||||
let event =
|
||||
nostr::EventBuilder::new(nostr::Kind::Custom(KIND_PRESENCE_UPDATE as u16), status)
|
||||
.tags(tags)
|
||||
.custom_created_at(nostr::Timestamp::from(now))
|
||||
.sign_with_keys(&state.relay_keypair)
|
||||
.ok()?;
|
||||
// A build/sign failure here is an internal fault, not a "not a presence
|
||||
// query" signal, so surface it as an error rather than falling through.
|
||||
let tags = match nostr::Tag::parse(["p", pubkey_hex]) {
|
||||
Ok(tag) => vec![tag],
|
||||
Err(e) => return Some(Err(internal_error(&format!("presence tag: {e}")))),
|
||||
};
|
||||
let event = match nostr::EventBuilder::new(
|
||||
nostr::Kind::Custom(KIND_PRESENCE_UPDATE as u16),
|
||||
status,
|
||||
)
|
||||
.tags(tags)
|
||||
.custom_created_at(nostr::Timestamp::from(now))
|
||||
.sign_with_keys(relay_keypair)
|
||||
{
|
||||
Ok(event) => event,
|
||||
Err(e) => return Some(Err(internal_error(&format!("presence sign: {e}")))),
|
||||
};
|
||||
|
||||
if let Ok(v) = serde_json::to_value(&event) {
|
||||
events.push(v);
|
||||
}
|
||||
}
|
||||
|
||||
Some(events)
|
||||
Some(Ok(events))
|
||||
}
|
||||
|
||||
// ── Moderation queue reads (L6 — Quinn) ───────────────────────────────────────
|
||||
@@ -2499,6 +2519,48 @@ mod tests {
|
||||
assert!(!has_mixed_search_filters(&filters));
|
||||
}
|
||||
|
||||
/// Production-wiring seam for the Redis-outage boundary. Drives the real
|
||||
/// `synthesize_presence` with a `PubSubManager` whose pool points at a
|
||||
/// closed port, so the `get_presence_bulk` lookup fails. A presence-snapshot
|
||||
/// filter must yield `Some(Err(500))` — never `Some(Ok([]))`, which would
|
||||
/// let a consumer mistake a backend outage for an authoritative all-offline
|
||||
/// snapshot. Restoring `unwrap_or_default()` inside `synthesize_presence`
|
||||
/// turns this red (it would return `Some(Ok([]))`), which is what protects
|
||||
/// the error-mapping seam Thufir found otherwise mutation-unprotected.
|
||||
#[tokio::test]
|
||||
async fn synthesize_presence_surfaces_redis_failure_as_error_response() {
|
||||
use buzz_core::kind::KIND_PRESENCE_SNAPSHOT;
|
||||
|
||||
// Pool at a closed port: get_presence_bulk's connection attempt fails.
|
||||
let dead_pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1")
|
||||
.create_pool(Some(deadpool_redis::Runtime::Tokio1))
|
||||
.expect("pool builds lazily");
|
||||
let pubsub = buzz_pubsub::PubSubManager::new("redis://127.0.0.1:1", dead_pool)
|
||||
.await
|
||||
.expect("PubSubManager::new performs no IO");
|
||||
let relay_keypair = Keys::generate();
|
||||
let tenant = fresh_tenant("relay.example");
|
||||
|
||||
// A presence-snapshot query for a concrete author reaches the Redis
|
||||
// lookup (an empty author set would short-circuit to an empty snapshot).
|
||||
let filters = vec![nostr::Filter::new()
|
||||
.kind(Kind::Custom(KIND_PRESENCE_SNAPSHOT as u16))
|
||||
.author(Keys::generate().public_key())];
|
||||
|
||||
let result = synthesize_presence(&pubsub, &relay_keypair, &tenant, &filters).await;
|
||||
|
||||
match result {
|
||||
Some(Err((status, _))) => assert_eq!(
|
||||
status,
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"a Redis lookup failure must surface as HTTP 500"
|
||||
),
|
||||
other => panic!(
|
||||
"a Redis outage must yield Some(Err(500)), not a fake-empty success: {other:?}"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn thread_aux_query_targets_root_and_replies() {
|
||||
let tenant = fresh_tenant("relay.example");
|
||||
|
||||
Reference in New Issue
Block a user