From b730780fe2a86106c2b3b1a768bf46283688540f Mon Sep 17 00:00:00 2001 From: michaelzeyuchen Date: Sat, 22 Aug 2026 00:18:38 +1000 Subject: [PATCH 1/4] fix: close reviewed relay and deployment gaps [skip-ultra-ship] Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: michaelzeyuchen --- Cargo.lock | 1 + Dockerfile | 1 - .../buzz-backend-kubernetes/src/reconcile.rs | 90 ++++++- crates/buzz-relay/Cargo.toml | 1 + crates/buzz-relay/src/api/git/hook.rs | 48 ++-- crates/buzz-relay/src/main.rs | 170 ++++++++++++- crates/buzz-relay/src/push_runtime.rs | 229 ++++++++++++++++-- crates/buzz-relay/tests/hook_hmac_cli.rs | 36 +++ deploy/charts/buzz/README.md | 7 +- .../charts/buzz/examples/secret-sample.yaml | 2 +- deploy/charts/buzz/templates/_validate.tpl | 13 +- deploy/charts/buzz/templates/deployment.yaml | 1 - deploy/charts/buzz/tests/networking_test.yaml | 4 + deploy/charts/buzz/tests/secrets_test.yaml | 28 +++ deploy/charts/buzz/tests/validation_test.yaml | 23 +- deploy/charts/buzz/values.schema.json | 2 +- deploy/charts/buzz/values.yaml | 7 +- .../src/commands/personas/snapshot/tests.rs | 4 + .../agents/ui/AgentSnapshotImportDialog.tsx | 2 +- .../ui/agentSnapshotImportDialog.test.mjs | 23 ++ .../api/tauriPersonas.snapshotImport.test.mjs | 2 +- desktop/src/shared/api/tauriPersonas.ts | 2 +- desktop/src/testing/e2eBridge.ts | 2 +- 23 files changed, 638 insertions(+), 60 deletions(-) create mode 100644 crates/buzz-relay/tests/hook_hmac_cli.rs diff --git a/Cargo.lock b/Cargo.lock index 18c53c18ca0..1ae652b0544 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1317,6 +1317,7 @@ dependencies = [ "tracing-subscriber", "url", "uuid", + "zeroize", ] [[package]] diff --git a/Dockerfile b/Dockerfile index d883ac6b015..af4d2dc5c2f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -136,7 +136,6 @@ RUN apt-get update \ ca-certificates \ curl \ git \ - openssl \ && rm -rf /var/lib/apt/lists/* \ && groupadd --system --gid 1000 buzz \ && useradd --system --uid 1000 --gid 1000 --home-dir /var/lib/buzz \ diff --git a/crates/buzz-backend-kubernetes/src/reconcile.rs b/crates/buzz-backend-kubernetes/src/reconcile.rs index df2f99789ff..d1e819e1e89 100644 --- a/crates/buzz-backend-kubernetes/src/reconcile.rs +++ b/crates/buzz-backend-kubernetes/src/reconcile.rs @@ -557,6 +557,23 @@ mod tests { } } + /// The fake mirrors the equality-based selector emitted by + /// `AgentIdentity::selector`, rather than returning every object in its + /// namespace. This keeps tests honest about the isolation supplied by the + /// apiserver before reconciliation and GC inspect candidates. + fn selector_matches(selector: &str, labels: Option<&BTreeMap>) -> bool { + let Some(labels) = labels else { + return false; + }; + + selector.split(',').all(|requirement| { + let Some((key, value)) = requirement.split_once('=') else { + panic!("fake does not support label selector requirement {requirement:?}"); + }; + labels.get(key).map(String::as_str) == Some(value) + }) + } + impl Substrate for Fake { async fn ensure_namespace(&self, namespace: &str) -> Result<(), String> { match &self.namespace_error { @@ -570,16 +587,27 @@ mod tests { async fn list_pods( &self, - _selector: &str, + selector: &str, ) -> Result<(Vec, Option>), String> { Ok(( - self.pods.borrow().values().cloned().collect(), + self.pods + .borrow() + .values() + .filter(|pod| selector_matches(selector, pod.metadata.labels.as_ref())) + .cloned() + .collect(), self.server_now, )) } - async fn list_secrets(&self, _selector: &str) -> Result, String> { - Ok(self.secrets.borrow().clone()) + async fn list_secrets(&self, selector: &str) -> Result, String> { + Ok(self + .secrets + .borrow() + .iter() + .filter(|secret| selector_matches(selector, secret.metadata.labels.as_ref())) + .cloned() + .collect()) } async fn secret_exists(&self, name: &str) -> Result { @@ -806,6 +834,60 @@ mod tests { ); } + #[test] + fn fake_lists_only_objects_matching_the_requested_identity() { + let id = identity(); + let other = identity(); + let cfg = config(); + let ours = our_pod(&id, &cfg, Some(running())); + let theirs = our_pod(&other, &cfg, Some(running())); + let our_secret = crate::pod::build_secret(&id, &cfg.namespace, "gen-ours", env()); + let their_secret = crate::pod::build_secret(&other, &cfg.namespace, "gen-theirs", env()); + let fake = Fake::default().with_pod(ours).with_pod(theirs); + fake.secrets.borrow_mut().extend([our_secret, their_secret]); + + let (pods, _) = block_on(fake.list_pods(&id.selector())).unwrap(); + let secrets = block_on(fake.list_secrets(&id.selector())).unwrap(); + + assert_eq!(pods.len(), 1); + assert_eq!( + pods[0].metadata.name.as_deref(), + Some(id.pod_name().as_str()) + ); + assert_eq!(secrets.len(), 1); + assert_eq!( + secrets[0].metadata.name.as_deref(), + Some(id.secret_name("gen-ours").as_str()) + ); + } + + #[test] + fn unrelated_tenant_objects_do_not_enter_reconciliation_or_gc() { + let id = identity(); + let other = identity(); + let cfg = config(); + let ours = our_pod(&id, &cfg, Some(running())); + let theirs = our_pod(&other, &cfg, Some(terminated())); + let their_pod_name = theirs.metadata.name.clone().unwrap(); + let their_secret = crate::pod::build_secret(&other, &cfg.namespace, "gen-existing", env()); + let their_secret_name = their_secret.metadata.name.clone().unwrap(); + let fake = Fake::default().with_pod(ours).with_pod(theirs); + fake.secrets.borrow_mut().push(their_secret); + + assert_eq!(run(&fake, &id, &cfg).unwrap(), id.pod_name()); + assert_eq!( + fake.mutations(), + [format!("ensure_namespace {}", cfg.namespace)], + "another tenant affected this deploy" + ); + assert!(fake.pods.borrow().contains_key(&their_pod_name)); + assert!(fake + .secrets + .borrow() + .iter() + .any(|secret| secret.metadata.name.as_deref() == Some(&their_secret_name))); + } + /// The strict no-op row: a started pod returns its id having mutated /// nothing at all. Asserted on the *call log*, not on final state — a /// delete-then-recreate would leave identical final state. diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index deb2e7e16a5..b083d8b356b 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -72,6 +72,7 @@ serde_yaml = { workspace = true } sha2 = { workspace = true } hmac = { workspace = true } subtle = { workspace = true } +zeroize = { workspace = true } rand = { workspace = true } hex = { workspace = true } url = { workspace = true } diff --git a/crates/buzz-relay/src/api/git/hook.rs b/crates/buzz-relay/src/api/git/hook.rs index e8e2c4d342e..4ca11855e95 100644 --- a/crates/buzz-relay/src/api/git/hook.rs +++ b/crates/buzz-relay/src/api/git/hook.rs @@ -11,6 +11,9 @@ //! - Fail-closed: curl failure, timeout, non-200 → exit 1 //! - Quarantine vars inherited for ancestry checks //! - HMAC binds callback to specific push operation +//! - The HMAC key never appears in a child process's argv: it is handed to +//! `buzz-relay hook-hmac` on file descriptor 3, so a same-UID process +//! reading `/proc//cmdline` or `ps` cannot recover it use std::path::Path; @@ -21,7 +24,9 @@ use tracing::{error, info}; /// /// Environment variables set by the relay before spawning git receive-pack: /// - `BUZZ_HOOK_URL` — internal policy endpoint (http://127.0.0.1:{port}/internal/git/policy) -/// - `BUZZ_HOOK_SECRET` — per-push HMAC secret +/// - `BUZZ_HOOK_SECRET` — deployment-wide HMAC secret (`git_hook_hmac_secret`, +/// injected by `transport.rs`; shared by every push and every replica, so it +/// must never reach a child process's argv — see the signing step below) /// - `BUZZ_REPO_ID` — repo identifier (d-tag) /// - `BUZZ_COMMUNITY_ID` — server-resolved community UUID for the git HTTP request /// - `BUZZ_PUSHER_PUBKEY` — authenticated pusher's hex pubkey @@ -112,7 +117,10 @@ if [ -f "$HMAC_FILE" ]; then fi HMAC_INPUT="${HMAC_INPUT}|${TIMESTAMP}" -SIGNATURE=$(printf '%s' "$HMAC_INPUT" | openssl dgst -sha256 -hmac "$BUZZ_HOOK_SECRET" -hex 2>/dev/null | sed 's/.*= //') +# Feed the secret over fd 3, never as an openssl command-line argument. The +# helper is the same Rust binary as the relay, selected by argv[1]; using a +# dedicated fd leaves stdin available for the canonical payload. +SIGNATURE=$(printf '%s' "$HMAC_INPUT" | /usr/local/bin/buzz-relay hook-hmac 3<<<"$BUZZ_HOOK_SECRET" 2>/dev/null) if [ -z "$SIGNATURE" ]; then echo "error: failed to compute HMAC signature" >&2 exit 1 @@ -182,26 +190,34 @@ mod tests { use super::PRE_RECEIVE_HOOK; #[test] - fn runtime_image_installs_pre_receive_hook_tools() { + fn hook_uses_in_image_tools_and_keeps_hmac_secret_off_argv() { let dockerfile = include_str!("../../../../../Dockerfile"); let runtime_stage = dockerfile - .split("FROM debian:${DEBIAN_VERSION}-slim AS runtime") + .split("FROM debian:${DEBIAN_VERSION}-slim AS runtime-base") .nth(1) - .expect("Dockerfile should have a runtime stage"); + .expect("Dockerfile should have a runtime-base stage"); let runtime_setup = runtime_stage - .split("COPY --from=builder") + .split("COPY --from=web-builder") .next() .expect("runtime stage should copy built artifacts after package setup"); - for tool in ["curl", "openssl"] { - assert!( - PRE_RECEIVE_HOOK.contains(tool), - "test setup expected the pre-receive hook to invoke {tool}" - ); - assert!( - runtime_setup.contains(&format!("\n {tool} \\")), - "relay runtime image must install {tool}; the git pre-receive hook uses it and fails closed without it" - ); - } + assert!(PRE_RECEIVE_HOOK.contains("curl")); + assert!( + runtime_setup.contains("\n curl \\"), + "relay runtime image must install curl; the git hook fails closed without it" + ); + assert!( + PRE_RECEIVE_HOOK + .contains("/usr/local/bin/buzz-relay hook-hmac 3<<<\"$BUZZ_HOOK_SECRET\""), + "the hook must pass its HMAC secret over a dedicated fd" + ); + assert!( + !PRE_RECEIVE_HOOK.contains("-hmac \"$BUZZ_HOOK_SECRET\""), + "the hook secret must not be exposed in process argv" + ); + assert!( + !runtime_setup.contains("\n openssl \\"), + "the hook no longer needs the openssl CLI in the runtime image" + ); } } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 566b684f830..780e968f85e 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1,9 +1,13 @@ use std::collections::{HashMap, HashSet}; +use std::io::{self, Read, Write}; use std::sync::atomic::Ordering; use std::sync::Arc; +use hmac::{Hmac, KeyInit, Mac}; +use sha2::Sha256; use tracing::{error, info, warn}; use tracing_subscriber::{fmt, prelude::*, EnvFilter}; +use zeroize::Zeroizing; fn log_env_filter(rust_log: Option<&str>) -> EnvFilter { EnvFilter::new(rust_log.unwrap_or("buzz_relay=info")) @@ -83,8 +87,74 @@ impl EmissionScope { const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452; +const HOOK_HMAC_COMMAND: &str = "hook-hmac"; + +struct MacWriter<'a>(&'a mut Hmac); + +impl Write for MacWriter<'_> { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.update(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +fn run_hook_hmac( + mut payload: R, + mut secret: K, + mut output: W, +) -> anyhow::Result<()> { + let mut secret_bytes = Zeroizing::new(Vec::new()); + secret + .read_to_end(&mut secret_bytes) + .map_err(|error| anyhow::anyhow!("failed to read hook HMAC secret: {error}"))?; + + while secret_bytes + .last() + .is_some_and(|byte| matches!(byte, b'\n' | b'\r')) + { + secret_bytes.pop(); + } + if secret_bytes.is_empty() { + anyhow::bail!("hook HMAC secret is empty"); + } + + let mut mac = as KeyInit>::new_from_slice(&secret_bytes) + .expect("HMAC accepts keys of any size"); + io::copy(&mut payload, &mut MacWriter(&mut mac)) + .map_err(|error| anyhow::anyhow!("failed to read hook HMAC payload: {error}"))?; + writeln!(output, "{}", hex::encode(mac.finalize().into_bytes())) + .map_err(|error| anyhow::anyhow!("failed to write hook HMAC: {error}"))?; + Ok(()) +} + +#[cfg(unix)] +fn run_hook_hmac_from_fd() -> anyhow::Result<()> { + let secret = std::fs::File::open("/dev/fd/3") + .map_err(|error| anyhow::anyhow!("failed to open hook HMAC secret fd 3: {error}"))?; + run_hook_hmac(io::stdin().lock(), secret, io::stdout().lock()) +} + +#[cfg(not(unix))] +fn run_hook_hmac_from_fd() -> anyhow::Result<()> { + anyhow::bail!("hook-hmac is supported only on Unix") +} + +fn is_hook_hmac_invocation() -> bool { + let mut args = std::env::args_os(); + let _executable = args.next(); + args.next().is_some_and(|arg| arg == HOOK_HMAC_COMMAND) && args.next().is_none() +} + #[tokio::main] async fn main() -> anyhow::Result<()> { + if is_hook_hmac_invocation() { + return run_hook_hmac_from_fd(); + } + // Install the ring CryptoProvider for rustls. Required before any rustls // TLS connection (rediss:// to ElastiCache, wss://, S3 over TLS): both // aws-lc-rs and ring are compiled in transitively, so rustls can't @@ -1521,6 +1591,20 @@ fn refresh_legacy_active_gauge_recency() { metrics::gauge!("buzz_subscriptions_active").increment(0.0); } +/// Refresh the exporter recency for startup-only configuration gauges. +/// +/// These are written once during boot, so without a refresh the idle-timeout +/// pruner drops them mid-run and the scrape silently loses the relay's own +/// configuration. `increment(0.0)` advances the recorder generation the +/// recency policy reads without disturbing the value, exactly as +/// [`refresh_legacy_active_gauge_recency`] does for lifecycle gauges. +/// +/// Dynamic per-community gauges are deliberately excluded: their pruning is +/// what keeps departed communities out of the scrape. +fn refresh_static_config_gauge_recency() { + metrics::gauge!("buzz_audit_enabled").increment(0.0); +} + /// Emit pod-local gauges and zero only label keys that disappeared since the /// preceding tick. The key stores the resolved host label so a removed or /// renamed community can still receive its final zero. @@ -1540,6 +1624,7 @@ fn emit_in_memory_usage_metrics( metrics::gauge!("buzz_total_users_online_pod").set(users_online.values().sum::() as f64); metrics::gauge!("buzz_total_subscriptions").set(total_subscriptions as f64); refresh_legacy_active_gauge_recency(); + refresh_static_config_gauge_recency(); let Some(host_map) = host_map else { return; @@ -2037,8 +2122,8 @@ mod tests { use super::{ buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs, - refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, EmissionScope, - InMemoryMetricKey, + refresh_legacy_active_gauge_recency, refresh_static_config_gauge_recency, run_hook_hmac, + run_periodic_until_cancelled, EmissionScope, InMemoryMetricKey, }; use metrics::GaugeFn; use metrics_util::{ @@ -2046,6 +2131,29 @@ mod tests { registry::{GenerationalAtomicStorage, Registry}, }; + #[test] + fn hook_hmac_reads_secret_separately_from_payload() { + let mut output = Vec::new(); + run_hook_hmac( + b"hello payload".as_slice(), + b"cross-boundary-test-secret-key-1234\n".as_slice(), + &mut output, + ) + .expect("hook HMAC helper"); + + assert_eq!( + String::from_utf8(output).expect("hex output"), + "009763fb8bdc5cb2a31b937a854a139d87fcfd5ec56acfd6523b0266e8024a1c\n" + ); + } + + #[test] + fn hook_hmac_rejects_an_empty_secret() { + let error = run_hook_hmac(b"payload".as_slice(), b"\n".as_slice(), Vec::new()) + .expect_err("empty secret must fail closed"); + assert_eq!(error.to_string(), "hook HMAC secret is empty"); + } + #[tokio::test(start_paused = true)] async fn periodic_loop_exits_immediately_on_cancellation() { let cancel = CancellationToken::new(); @@ -2153,6 +2261,64 @@ mod tests { assert!(gauge.get_generation() > generation_before); } + #[test] + fn test_static_config_gauge_recency_refresh_preserves_value() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + metrics::gauge!("buzz_audit_enabled").set(1.0); + refresh_static_config_gauge_recency(); + }); + + let values = snapshotter + .snapshot() + .into_vec() + .into_iter() + .map(|(key, _, _, value)| { + let DebugValue::Gauge(value) = value else { + panic!("{} must be a gauge", key.key().name()); + }; + (key.key().name().to_owned(), value.into_inner()) + }) + .collect::>(); + + assert_eq!(values.get("buzz_audit_enabled"), Some(&1.0)); + } + + #[test] + fn test_static_config_refresh_retains_only_the_static_gauge() { + use metrics_exporter_prometheus::PrometheusBuilder; + use metrics_util::MetricKindMask; + + let recorder = PrometheusBuilder::new() + .idle_timeout(MetricKindMask::GAUGE, Some(Duration::from_millis(40))) + .build_recorder(); + let handle = recorder.handle(); + + metrics::with_local_recorder(&recorder, || { + metrics::gauge!("buzz_audit_enabled").set(1.0); + metrics::gauge!("buzz_community_ws_connections", "community" => "gone.example") + .set(3.0); + }); + let initial = handle.render(); + assert!(initial.contains("buzz_audit_enabled 1")); + assert!(initial.contains("buzz_community_ws_connections")); + + std::thread::sleep(Duration::from_millis(60)); + metrics::with_local_recorder(&recorder, refresh_static_config_gauge_recency); + let after_timeout = handle.render(); + + assert!( + after_timeout.contains("buzz_audit_enabled 1"), + "startup gauge must survive the exporter idle timeout: {after_timeout}" + ); + assert!( + !after_timeout.contains("buzz_community_ws_connections"), + "unrefreshed dynamic gauges must still be pruned: {after_timeout}" + ); + } + #[test] fn test_idle_timeout_is_at_least_three_usage_intervals() { assert_eq!(idle_timeout_secs(None, 300), 900); diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 4946b248c65..80d84748ef8 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -1,6 +1,9 @@ //! Durable NIP-PL event matcher and gateway delivery worker. -use std::{sync::Arc, time::Duration}; +use std::{ + sync::{Arc, OnceLock}, + time::Duration, +}; use base64::Engine as _; use buzz_core::filter::{filters_match, reader_authorized_for_event}; @@ -88,10 +91,78 @@ pub async fn run_matcher(state: Arc) { } } +/// One lease plus the lazily prepared form of its immutable subscription JSON. +/// +/// The cache lives in [`MatchContext`], so every event in one claimed batch +/// shares it, while the next batch still reloads the current lease snapshot. +struct PreparedMatchLease { + lease: buzz_db::push::MatchLease, + prepared: OnceLock, String>>, + #[cfg(test)] + preparation_count: std::sync::atomic::AtomicUsize, +} + +struct PreparedSubscription { + filter: Filter, + class: String, + ignore: Vec, + suppress: Option, +} + +impl PreparedMatchLease { + fn new(lease: buzz_db::push::MatchLease) -> Self { + Self { + lease, + prepared: OnceLock::new(), + #[cfg(test)] + preparation_count: std::sync::atomic::AtomicUsize::new(0), + } + } + + fn subscriptions(&self) -> anyhow::Result<&[PreparedSubscription]> { + self.prepared + .get_or_init(|| { + #[cfg(test)] + self.preparation_count + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + prepare_subscriptions(&self.lease.subscriptions) + }) + .as_deref() + .map_err(|message| anyhow::anyhow!(message.clone())) + } +} + +fn prepare_subscriptions( + raw_subscriptions: &serde_json::Value, +) -> Result, String> { + let subscriptions: Vec = + serde_json::from_value(raw_subscriptions.clone()).map_err(|error| error.to_string())?; + subscriptions + .into_iter() + .map(|subscription| { + let filter = serde_json::from_value(serde_json::Value::Object(subscription.filter)) + .map_err(|error| error.to_string())?; + let ignore = subscription + .ignore + .into_iter() + // Preserve the previous matcher semantics: malformed ignore + // filters are non-matches rather than poison lease data. + .filter_map(|raw| serde_json::from_value(serde_json::Value::Object(raw)).ok()) + .collect(); + Ok(PreparedSubscription { + filter, + class: subscription.class, + ignore, + suppress: subscription.suppress, + }) + }) + .collect() +} + /// Per-batch state shared by every job: the community's active leases and /// the exact (channel, lease author) membership pairs the jobs can consult. struct MatchContext { - leases: Vec, + leases: Vec, memberships: std::collections::HashSet<(uuid::Uuid, Vec)>, } @@ -117,7 +188,7 @@ async fn load_match_context( .into_iter() .collect(); Ok(MatchContext { - leases, + leases: leases.into_iter().map(PreparedMatchLease::new).collect(), memberships, }) } @@ -222,7 +293,8 @@ fn match_job( context: &MatchContext, ) -> anyhow::Result> { let mut wakes = Vec::new(); - for lease in &context.leases { + for prepared_lease in &context.leases { + let lease = &prepared_lease.lease; let author_hex = hex::encode(&lease.author); if !reader_authorized_for_event(&job.event.event, &author_hex) { continue; @@ -235,20 +307,17 @@ fn match_job( continue; } } - let subscriptions: Vec = serde_json::from_value(lease.subscriptions.clone())?; let mut class: Option<&str> = None; - for sub in &subscriptions { - let filter: Filter = - serde_json::from_value(serde_json::Value::Object(sub.filter.clone()))?; - if !push_filter_authorized_for_event(&filter, &job.event.event, &author_hex) - || !filters_match(std::slice::from_ref(&filter), &job.event) + for sub in prepared_lease.subscriptions()? { + if !push_filter_authorized_for_event(&sub.filter, &job.event.event, &author_hex) + || !filters_match(std::slice::from_ref(&sub.filter), &job.event) { continue; } - let ignored = sub.ignore.iter().any(|raw| { - serde_json::from_value::(serde_json::Value::Object(raw.clone())) - .is_ok_and(|f| filters_match(&[f], &job.event)) - }); + let ignored = sub + .ignore + .iter() + .any(|filter| filters_match(std::slice::from_ref(filter), &job.event)); let p_count = job .event .event @@ -615,6 +684,138 @@ mod tests { use std::{future::IntoFuture, sync::Arc}; use tokio::sync::Mutex; + fn match_lease(author: &nostr::Keys, subscriptions: serde_json::Value) -> PreparedMatchLease { + PreparedMatchLease::new(buzz_db::push::MatchLease { + author: author.public_key().to_bytes().to_vec(), + installation_id: "device-1".to_owned(), + generation: 7, + subscriptions, + expires_at: Utc::now().timestamp() + 600, + }) + } + + fn match_job_for( + author: &nostr::Keys, + channel_id: Option, + ) -> buzz_db::push::BatchedMatch { + let event = EventBuilder::new(Kind::TextNote, "hello") + .sign_with_keys(author) + .unwrap(); + buzz_db::push::BatchedMatch { + event: buzz_core::StoredEvent::new(event, channel_id), + attempt: 1, + } + } + + fn valid_subscriptions(ignore: serde_json::Value) -> serde_json::Value { + serde_json::json!([{ + "filter": {"kinds": [1]}, + "class": "urgent", + "ignore": [ignore], + "suppress": null + }]) + } + + #[test] + fn match_context_prepares_a_reached_lease_once_for_multiple_jobs() { + let author = nostr::Keys::generate(); + let context = MatchContext { + leases: vec![match_lease( + &author, + serde_json::json!([{ + "filter": {"kinds": [1]}, + "class": "urgent", + "ignore": [], + "suppress": null + }]), + )], + memberships: Default::default(), + }; + + let first = match_job(&match_job_for(&author, None), &context).unwrap(); + let second = match_job(&match_job_for(&author, None), &context).unwrap(); + + assert_eq!(first.len(), 1); + assert_eq!(second.len(), 1); + assert_eq!(first[0].class, "urgent"); + assert_eq!(second[0].class, "urgent"); + assert_eq!( + context.leases[0] + .preparation_count + .load(std::sync::atomic::Ordering::Relaxed), + 1 + ); + } + + #[test] + fn malformed_lease_is_parsed_only_after_authorization_and_membership() { + let owner = nostr::Keys::generate(); + let other = nostr::Keys::generate(); + let private_event = + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_DM_VISIBILITY as u16), "") + .tag(Tag::public_key(other.public_key())) + .sign_with_keys(&other) + .unwrap(); + let unauthorized = buzz_db::push::BatchedMatch { + event: buzz_core::StoredEvent::new(private_event, None), + attempt: 1, + }; + let channel = uuid::Uuid::new_v4(); + let context = MatchContext { + leases: vec![match_lease(&owner, serde_json::json!({"malformed": true}))], + memberships: Default::default(), + }; + + assert!(match_job(&unauthorized, &context).unwrap().is_empty()); + assert_eq!( + context.leases[0] + .preparation_count + .load(std::sync::atomic::Ordering::Relaxed), + 0 + ); + assert!(match_job(&match_job_for(&owner, Some(channel)), &context) + .unwrap() + .is_empty()); + assert_eq!( + context.leases[0] + .preparation_count + .load(std::sync::atomic::Ordering::Relaxed), + 0 + ); + assert!(match_job(&match_job_for(&owner, None), &context).is_err()); + assert_eq!( + context.leases[0] + .preparation_count + .load(std::sync::atomic::Ordering::Relaxed), + 1 + ); + assert!(match_job(&match_job_for(&owner, None), &context).is_err()); + assert_eq!( + context.leases[0] + .preparation_count + .load(std::sync::atomic::Ordering::Relaxed), + 1, + "preparation failures must be cached too" + ); + } + + #[test] + fn malformed_ignore_filter_remains_a_non_match() { + let owner = nostr::Keys::generate(); + let context = MatchContext { + leases: vec![match_lease( + &owner, + valid_subscriptions(serde_json::json!({"kinds": "not-an-array"})), + )], + memberships: Default::default(), + }; + + let wakes = match_job(&match_job_for(&owner, None), &context).unwrap(); + + assert_eq!(wakes.len(), 1); + assert_eq!(wakes[0].class, "urgent"); + } + #[test] fn gift_wrap_match_requires_self_p_filter_and_recipient() { let recipient = nostr::Keys::generate(); diff --git a/crates/buzz-relay/tests/hook_hmac_cli.rs b/crates/buzz-relay/tests/hook_hmac_cli.rs new file mode 100644 index 00000000000..6497c02518e --- /dev/null +++ b/crates/buzz-relay/tests/hook_hmac_cli.rs @@ -0,0 +1,36 @@ +//! End-to-end fail-closed check for the `buzz-relay hook-hmac` helper the +//! generated pre-receive hook shells out to. +//! +//! The unit tests in `main.rs` cover signing with separate in-memory readers. +//! This exercises the real binary and pins the remaining process boundary: a +//! missing secret descriptor must not yield anything the hook could mistake +//! for a signature. + +use std::process::{Command, Stdio}; + +const HELPER: &str = env!("CARGO_BIN_EXE_buzz-relay"); + +#[test] +fn helper_fails_closed_without_secret_fd() { + let output = Command::new(HELPER) + .arg("hook-hmac") + .stdin(Stdio::null()) + .output() + .expect("run hook-hmac helper"); + + assert!( + !output.status.success(), + "helper must fail when fd 3 was never opened, got {:?}", + output.status + ); + assert!( + output.stdout.is_empty(), + "a failed signing attempt must not print a signature: {:?}", + String::from_utf8_lossy(&output.stdout) + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("hook HMAC secret fd 3"), + "error must name the missing secret descriptor, got {stderr:?}" + ); +} diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 30cee4f4063..67e301569fa 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -169,13 +169,14 @@ an Ingress or HTTPRoute for the pairing Service; route the public hostname to ## HA (production) -`replicaCount > 1` hard-requires Redis: +Redis is required for every relay process: -- Redis (`redis.enabled=true`, `externalRedis.url`, or `REDIS_URL` in `existingSecret`) — for `buzz-pubsub` fan-out +- configure `redis.enabled=true`, `externalRedis.url`, or `REDIS_URL` in `existingSecret`, including for a single replica +- in HA, Redis additionally provides cross-pod `buzz-pubsub` fan-out It does **not** require ReadWriteMany git storage. Git ref/object state is object-store-backed (each request hydrates an ephemeral repo from S3-compatible storage; writer serialization is the object-store pointer CAS — see `docs/git-on-object-storage.md`), and repo-name uniqueness lives in Postgres. Each replica can use its own `ReadWriteOnce` volume; no shared filesystem is needed. -The chart **template-fails** if the Redis invariant is broken at `replicaCount > 1`. No silent degradation. +The chart **template-fails** if the Redis invariant is broken. No silent degradation. ### Relay autoscaling diff --git a/deploy/charts/buzz/examples/secret-sample.yaml b/deploy/charts/buzz/examples/secret-sample.yaml index 42d3254d486..46360ef441d 100644 --- a/deploy/charts/buzz/examples/secret-sample.yaml +++ b/deploy/charts/buzz/examples/secret-sample.yaml @@ -9,7 +9,7 @@ # BUZZ_GIT_HOOK_HMAC_SECRET — 32+ chars; required when replicaCount > 1 # DATABASE_URL — postgres://... # READ_DATABASE_URL — postgres://... (optional read-replica; omit to disable read routing) -# REDIS_URL — redis://... (required when replicaCount > 1) +# REDIS_URL — redis://... (required for every relay) # BUZZ_S3_ACCESS_KEY # BUZZ_S3_SECRET_KEY apiVersion: v1 diff --git a/deploy/charts/buzz/templates/_validate.tpl b/deploy/charts/buzz/templates/_validate.tpl index aa7f7ac13cf..e25f7219543 100644 --- a/deploy/charts/buzz/templates/_validate.tpl +++ b/deploy/charts/buzz/templates/_validate.tpl @@ -10,12 +10,9 @@ surface at template time regardless of which manifest helm renders first. {{- fail "relayUrl is required: set --set relayUrl=wss://your.domain" -}} {{- end -}} -{{/* Multiple replicas require Redis, whether fixed or autoscaled. */}} -{{- $minimumReplicas := include "buzz.minimumReplicas" . | int -}} -{{- if gt $minimumReplicas 1 -}} - {{- if and (not .Values.redis.enabled) (not .Values.externalRedis.url) (not .Values.secrets.existingSecret) -}} - {{- fail (printf "minimum replica count %d requires Redis for buzz-pubsub. Enable redis.enabled=true, set externalRedis.url, or provide secrets.existingSecret with key REDIS_URL." $minimumReplicas) -}} - {{- end -}} +{{/* Redis is required for buzz-pubsub, even at a single replica. */}} +{{- if not (or .Values.redis.enabled .Values.externalRedis.url .Values.secrets.existingSecret) -}} + {{- fail "Redis source missing: enable redis.enabled=true, set externalRedis.url, or provide secrets.existingSecret with key REDIS_URL." -}} {{- end -}} {{/* Multiple replicas do NOT require ReadWriteMany git storage. @@ -30,8 +27,8 @@ surface at template time regardless of which manifest helm renders first. The prior hard-fail requiring persistence.git.accessMode=ReadWriteMany was removed here: its stated reason ("git on-disk state must be shared across - replicas") is no longer true. Redis (validated above) remains the real - multi-pod requirement for buzz-pubsub. */}} + replicas") is no longer true. Redis is required for every relay process; + in HA it additionally supplies cross-pod buzz-pubsub fan-out. */}} {{/* Autoscaling bounds must be coherent. */}} {{- if .Values.autoscaling.enabled -}} diff --git a/deploy/charts/buzz/templates/deployment.yaml b/deploy/charts/buzz/templates/deployment.yaml index 451ebb1cded..42b81df7201 100644 --- a/deploy/charts/buzz/templates/deployment.yaml +++ b/deploy/charts/buzz/templates/deployment.yaml @@ -202,7 +202,6 @@ spec: secretKeyRef: name: {{ include "buzz.envSecretName" . }} key: REDIS_URL - optional: {{ and (eq (include "buzz.minimumReplicas" . | int) 1) (not .Values.redis.enabled) (not .Values.externalRedis.url) }} - name: BUZZ_S3_ACCESS_KEY valueFrom: secretKeyRef: diff --git a/deploy/charts/buzz/tests/networking_test.yaml b/deploy/charts/buzz/tests/networking_test.yaml index 882679f2e6d..18266960583 100644 --- a/deploy/charts/buzz/tests/networking_test.yaml +++ b/deploy/charts/buzz/tests/networking_test.yaml @@ -9,6 +9,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -27,6 +28,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -43,6 +45,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -62,6 +65,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s diff --git a/deploy/charts/buzz/tests/secrets_test.yaml b/deploy/charts/buzz/tests/secrets_test.yaml index dca83ce27ff..0aada8683d8 100644 --- a/deploy/charts/buzz/tests/secrets_test.yaml +++ b/deploy/charts/buzz/tests/secrets_test.yaml @@ -8,6 +8,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -29,6 +30,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -43,6 +45,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -59,11 +62,33 @@ tests: optional: true template: templates/deployment.yaml + - it: Deployment env requires REDIS_URL from existingSecret + set: + relayUrl: wss://buzz.example.com + ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" + externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 + s3.endpoint: http://minio:9000 + s3.accessKey: a + s3.secretKey: s + secrets.existingSecret: "buzz-secrets" + asserts: + - contains: + path: spec.template.spec.containers[0].env + content: + name: REDIS_URL + valueFrom: + secretKeyRef: + name: buzz-secrets + key: REDIS_URL + template: templates/deployment.yaml + - it: Deployment env points READ_DATABASE_URL at existingSecret as optional set: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -85,6 +110,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -105,6 +131,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s @@ -126,6 +153,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.accessKey: a s3.secretKey: s diff --git a/deploy/charts/buzz/tests/validation_test.yaml b/deploy/charts/buzz/tests/validation_test.yaml index a5a0050a866..be6dfb99998 100644 --- a/deploy/charts/buzz/tests/validation_test.yaml +++ b/deploy/charts/buzz/tests/validation_test.yaml @@ -16,6 +16,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 asserts: - failedTemplate: errorPattern: "ownerPubkey is required when relay.requireRelayMembership=true" @@ -25,11 +26,12 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "NOTAHEX" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 asserts: - failedTemplate: errorPattern: "ownerPubkey: Does not match pattern" - - it: fails when replicaCount>1 without Redis + - it: fails when Redis source is missing for multiple replicas set: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" @@ -37,13 +39,27 @@ tests: replicaCount: 3 asserts: - failedTemplate: - errorPattern: "minimum replica count 3 requires Redis" + errorPattern: "Redis source missing" + + - it: fails when Redis source is missing for a single replica + set: + relayUrl: wss://buzz.example.com + ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" + externalPostgresql.url: postgres://u:p@h:5432/d + s3.endpoint: http://minio:9000 + s3.accessKey: a + s3.secretKey: s + replicaCount: 1 + asserts: + - failedTemplate: + errorPattern: "Redis source missing" - it: fails when ingress and httproute both enabled set: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 ingress.enabled: true httproute.enabled: true asserts: @@ -54,6 +70,7 @@ tests: set: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" + externalRedis.url: redis://h:6379 asserts: - failedTemplate: errorPattern: "Postgres source missing" @@ -63,6 +80,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 s3.endpoint: http://minio:9000 s3.addressingStyle: auto asserts: @@ -74,6 +92,7 @@ tests: relayUrl: wss://buzz.example.com ownerPubkey: "0000000000000000000000000000000000000000000000000000000000000000" externalPostgresql.url: postgres://u:p@h:5432/d + externalRedis.url: redis://h:6379 asserts: - failedTemplate: errorPattern: "S3/object-storage source missing" diff --git a/deploy/charts/buzz/values.schema.json b/deploy/charts/buzz/values.schema.json index 94d369c8903..29a4aee7025 100644 --- a/deploy/charts/buzz/values.schema.json +++ b/deploy/charts/buzz/values.schema.json @@ -26,7 +26,7 @@ "replicaCount": { "type": "integer", "minimum": 1, - "description": "Replica count for the relay Deployment. replicaCount > 1 requires Redis (for buzz-pubsub) — enforced by _validate.tpl. Git storage does NOT need ReadWriteMany: git state is object-store-backed and repo names live in Postgres, so ReadWriteOnce is fine per replica." + "description": "Replica count for the relay Deployment. Every relay requires Redis for buzz-pubsub; with multiple replicas it additionally provides cross-pod fan-out. This is enforced by _validate.tpl. Git storage does NOT need ReadWriteMany: git state is object-store-backed and repo names live in Postgres, so ReadWriteOnce is fine per replica." }, "relayUrl": { "type": "string", diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index ca3403a633f..c92e3b851d7 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -29,8 +29,9 @@ image: pullSecrets: [] # ── Topology ──────────────────────────────────────────────────────────────── -# replicaCount > 1 hard-requires Redis for buzz-pubsub (in-cluster or external). -# It does NOT require ReadWriteMany git storage: git ref/object state is +# Every relay requires Redis for buzz-pubsub (in-cluster or external); with +# multiple replicas it additionally provides cross-pod fan-out. +# Multiple replicas do NOT require ReadWriteMany git storage: git ref/object state is # object-store-backed (each request hydrates an ephemeral repo from S3; writer # serialization is the object-store pointer CAS), and repo-name uniqueness lives # in Postgres. Each replica can use its own ReadWriteOnce volume (or none). @@ -90,7 +91,7 @@ ownerPubkey: "" # BUZZ_GIT_HOOK_HMAC_SECRET — 32+ chars; required when replicaCount > 1 # DATABASE_URL — full Postgres URL (preferred over externalPostgresql.url) # READ_DATABASE_URL — optional Postgres read-replica URL; omit to keep all reads on the writer -# REDIS_URL — full Redis URL with auth +# REDIS_URL — full Redis URL with auth; required # BUZZ_S3_ACCESS_KEY — S3 access key # BUZZ_S3_SECRET_KEY — S3 secret key secrets: diff --git a/desktop/src-tauri/src/commands/personas/snapshot/tests.rs b/desktop/src-tauri/src/commands/personas/snapshot/tests.rs index fedb0e60585..5f78c118917 100644 --- a/desktop/src-tauri/src/commands/personas/snapshot/tests.rs +++ b/desktop/src-tauri/src/commands/personas/snapshot/tests.rs @@ -571,6 +571,10 @@ fn import_preview_includes_exported_definition_metadata() { assert!(preview.is_builtin); assert_eq!(preview.model.as_deref(), Some("claude-opus-4-5")); assert_eq!(preview.runtime.as_deref(), Some("goose")); + + let wire = serde_json::to_value(&preview).unwrap(); + assert_eq!(wire.get("isBuiltin"), Some(&serde_json::Value::Bool(true))); + assert!(wire.get("isBuiltIn").is_none()); } // ── Import: resolve_snapshot_import_behavior — the production selection path diff --git a/desktop/src/features/agents/ui/AgentSnapshotImportDialog.tsx b/desktop/src/features/agents/ui/AgentSnapshotImportDialog.tsx index 0527565763a..0bd72377460 100644 --- a/desktop/src/features/agents/ui/AgentSnapshotImportDialog.tsx +++ b/desktop/src/features/agents/ui/AgentSnapshotImportDialog.tsx @@ -178,7 +178,7 @@ export function PreviewBody({ ) : null} diff --git a/desktop/src/features/agents/ui/agentSnapshotImportDialog.test.mjs b/desktop/src/features/agents/ui/agentSnapshotImportDialog.test.mjs index a02bb5a825d..f552a06625e 100644 --- a/desktop/src/features/agents/ui/agentSnapshotImportDialog.test.mjs +++ b/desktop/src/features/agents/ui/agentSnapshotImportDialog.test.mjs @@ -74,6 +74,9 @@ function makeResult(overrides = {}) { function makePreview(overrides = {}) { return { displayName: "TestBot", + isBuiltin: true, + model: "claude-opus-4-5", + runtime: "goose", systemPrompt: "Inspect every boundary before changing code.", avatarUrl: null, memoryLevel: "none", @@ -114,6 +117,26 @@ test("preview_body_discloses_prompt_allowlist_and_full_manifest", () => { ); }); +test("preview_body_passes_serialized_builtin_metadata", () => { + const preview = makePreview(); + const element = PreviewBody({ + preview, + hasMemory: false, + memoryLevelLabel: "none", + keepAllowlist: false, + onKeepAllowlistChange: () => {}, + }); + const [metadata] = findAll( + element, + (node) => + node.props?.model === preview.model && + node.props?.runtime === preview.runtime, + ); + + assert.ok(metadata); + assert.equal(metadata.props.isBuiltIn, true); +}); + // ── locked-card provenance notice ───────────────────────────────────────────── test("preview_body_shows_locked_notice_only_for_locked_cards", () => { diff --git a/desktop/src/shared/api/tauriPersonas.snapshotImport.test.mjs b/desktop/src/shared/api/tauriPersonas.snapshotImport.test.mjs index d4e0b1d5f3e..70ecfa15bf5 100644 --- a/desktop/src/shared/api/tauriPersonas.snapshotImport.test.mjs +++ b/desktop/src/shared/api/tauriPersonas.snapshotImport.test.mjs @@ -12,7 +12,7 @@ import test from "node:test"; function makePreview(overrides = {}) { return { displayName: "Test Agent", - isBuiltIn: false, + isBuiltin: false, model: null, runtime: null, systemPrompt: "You are helpful.", diff --git a/desktop/src/shared/api/tauriPersonas.ts b/desktop/src/shared/api/tauriPersonas.ts index 3cd9734ae26..4aa8c12f9e9 100644 --- a/desktop/src/shared/api/tauriPersonas.ts +++ b/desktop/src/shared/api/tauriPersonas.ts @@ -369,7 +369,7 @@ export async function loadAgentCard(storedFileName: string): Promise { export type AgentSnapshotImportPreview = { displayName: string; /** Source classification shown in the preview; imports remain custom. */ - isBuiltIn: boolean; + isBuiltin: boolean; model: string | null; runtime: string | null; systemPrompt: string | null; diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index e4028f01716..de75cf5683d 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -12940,7 +12940,7 @@ export function maybeInstallE2eTauriMocks() { // Return a minimal preview — no writes performed. return { displayName: "Imported Agent", - isBuiltIn: true, + isBuiltin: true, model: "claude-opus-4-5", runtime: "goose", systemPrompt: null, From 7ad9d1faff2af9a814026e70646249da7132ba93 Mon Sep 17 00:00:00 2001 From: michaelzeyuchen Date: Tue, 18 Aug 2026 22:25:01 +1000 Subject: [PATCH 2/4] fix(relay,search): close R3-F1..F7,F9..F13 (12 of 13); F8 deferred R3-F1..F13 round-2 followup hardens and de-duplicates work that didn't fit in the R2 close on this branch. 12 of 13 findings land with code+test evidence; R3-F8 (serialize-once across the fan-out path) is recorded in the CHECKPOINT receipt as deferred-to-followup because it requires a structural plumbing change through 5+ callsites of fan_out_event_to_local_subscribers. Closed: - F1: per-IP connection admission wired through check_ip_connection - F2: agent_elevated_messages_per_min honored in WS tier selector - F3: scripts/run-tests.sh unit + just test-unit now exercise buzz-relay --lib - F4/F5: redis/transport-dependent mesh_demo tests #[ignore]d with explicit precondition; transport_routes_to_owner_runtime + mesh_demo_no_redis_branch tests - F6: filter inner loop uses TagKind::as_str instead of to_string - F7: per-recipient membership DB calls batched into membership_pairs_cached - F9: DashMap Ref released before iteration in fan_out_scoped (deadlock fix) - F10: handle_text_message takes &str instead of String - F11: HashSet for accessible_channels in req handler inner loop - F12: FTS ORDER BY ts_rank_cd(numeric) instead of @@ boolean - F13: keyset cursor pagination; existing page-based path preserved Receipt: status CHECKPOINT, exit_round 8, covered_tree_sha d49c926b17a034503b4e72209ae852e46740fbc739f1de87e3754190a1ba3f4c, verification_evidence from bash scripts/run-tests.sh unit (sha256 50a44bf5c63de5365ce90d26e779598e3223abfa206bacb8646ea6824f75a119). [skip-ultra-ship] Signed-off-by: Michael Ze Yu Chen Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: michaelzeyuchen --- Justfile | 3 + crates/buzz-core/src/filter.rs | 6 +- crates/buzz-relay/src/api/bridge.rs | 1 + crates/buzz-relay/src/api/mesh_demo.rs | 102 +++++++++++++++++--- crates/buzz-relay/src/connection.rs | 58 +++++++++-- crates/buzz-relay/src/handlers/event.rs | 36 ++++--- crates/buzz-relay/src/handlers/req.rs | 9 +- crates/buzz-relay/src/state.rs | 50 ++++++++++ crates/buzz-relay/src/subscription.rs | 4 + crates/buzz-search/src/lib.rs | 4 +- crates/buzz-search/src/query.rs | 74 ++++++++++++-- crates/buzz-search/tests/fts_integration.rs | 27 ++++++ scripts/run-tests.sh | 5 + 13 files changed, 335 insertions(+), 44 deletions(-) diff --git a/Justfile b/Justfile index fe5d7bf2858..b84b29e3a31 100644 --- a/Justfile +++ b/Justfile @@ -340,6 +340,9 @@ test-unit: # `cargo test --workspace`; without this step a manifest edit that # diverges Rust from the corpus ships green. cargo nextest run -p buzz-agent --lib + # buzz-relay --lib: 906 unit tests; the Postgres/Redis-backed paths + # are #[ignore]d or runtime-skipped, so this stays infra-free. + cargo nextest run -p buzz-relay --lib else ./scripts/run-tests.sh unit fi diff --git a/crates/buzz-core/src/filter.rs b/crates/buzz-core/src/filter.rs index 32e3a7ad16b..0a8a691a9de 100644 --- a/crates/buzz-core/src/filter.rs +++ b/crates/buzz-core/src/filter.rs @@ -66,12 +66,12 @@ fn filter_match_one(f: &Filter, ev: &StoredEvent) -> bool { } for (tag_key, tag_values) in f.generic_tags.iter() { - let tag_key_str = tag_key.to_string(); + let tag_key_str = tag_key.as_str(); let has_match = tag_values.iter().any(|filter_val| { ev.event .tags .iter() - .filter(|t| t.kind().to_string() == tag_key_str) + .filter(|t| t.kind().as_str() == tag_key_str) .filter_map(|t| t.content()) .any(|event_val| event_val == filter_val.as_str()) }); @@ -81,7 +81,7 @@ fn filter_match_one(f: &Filter, ev: &StoredEvent) -> bool { // fallback ONLY when the event has no h-tags at all — if the event // has explicit h-tags, those are authoritative and must match. if !has_match && tag_key_str == "h" { - let event_has_h_tags = ev.event.tags.iter().any(|t| t.kind().to_string() == "h"); + let event_has_h_tags = ev.event.tags.iter().any(|t| t.kind().as_str() == "h"); if !event_has_h_tags { if let Some(ch_id) = ev.channel_id { let ch_str = ch_id.to_string(); diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 8fdea4b3c02..804fe8281bd 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1799,6 +1799,7 @@ async fn handle_bridge_search( page: search_page, per_page: limit, mode: search_mode, + cursor: None, }; let search_result = state diff --git a/crates/buzz-relay/src/api/mesh_demo.rs b/crates/buzz-relay/src/api/mesh_demo.rs index 8649b97671a..bff44e28310 100644 --- a/crates/buzz-relay/src/api/mesh_demo.rs +++ b/crates/buzz-relay/src/api/mesh_demo.rs @@ -166,8 +166,15 @@ mod tests { .expect("create redis pool") } - async fn redis_directory_if_available() -> Option { - let pool = pool(); + pub(crate) async fn redis_directory_if_available( + url_override: Option<&str>, + ) -> Option { + let pool = match url_override { + Some(url) => deadpool_redis::Config::from_url(url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .ok()?, + None => pool(), + }; let mut conn = pool.get().await.ok()?; redis::cmd("PING") .query_async::(&mut *conn) @@ -179,7 +186,7 @@ mod tests { )) } - struct NoopTransport; + pub(crate) struct NoopTransport; impl RelayPeerTransport for NoopTransport { fn send_datagram(&self, _to: RuntimeId, _dgram: MeshDatagram) -> Result<(), MeshError> { @@ -199,13 +206,31 @@ mod tests { fn set_inbound(&self, _handler: Box) {} } - struct DirectTransport { + /// Recording variant: `send_datagram` captures the destination runtime id + /// so tests can assert which peer the forwarded arm routed to. Opens + /// streams through the peer like the prior `DirectTransport`. + pub(crate) struct DirectTransport { peer: buzz_relay_mesh::peer::MeshPeer, + recorded_to: std::sync::Mutex>, + } + + impl DirectTransport { + pub(crate) fn new(peer: buzz_relay_mesh::peer::MeshPeer) -> Self { + Self { + peer, + recorded_to: std::sync::Mutex::new(None), + } + } + + pub(crate) fn last_recorded_to(&self) -> Option { + *self.recorded_to.lock().unwrap() + } } impl RelayPeerTransport for DirectTransport { - fn send_datagram(&self, _to: RuntimeId, _dgram: MeshDatagram) -> Result<(), MeshError> { - unreachable!("demo forwarded-arm test never sends datagrams") + fn send_datagram(&self, to: RuntimeId, _dgram: MeshDatagram) -> Result<(), MeshError> { + *self.recorded_to.lock().unwrap() = Some(to); + Ok(()) } fn open_session_stream( @@ -215,8 +240,9 @@ mod tests { ) -> std::pin::Pin< Box> + Send + '_>, > { + let peer = self.peer.clone(); Box::pin(async move { - let mut stream = self.peer.open_bi().await?; + let mut stream = peer.open_bi().await?; stream.send_frame(MeshStreamFrame::Hello(hello)).await?; Ok(stream) }) @@ -225,15 +251,16 @@ mod tests { fn set_inbound(&self, _handler: Box) {} } - async fn body_json(resp: Response) -> serde_json::Value { + pub(crate) async fn body_json(resp: Response) -> serde_json::Value { let bytes = to_bytes(resp.into_body(), 1024 * 1024).await.unwrap(); serde_json::from_slice(&bytes).unwrap() } /// First post for a session acquires the fenced lease and reports `owned`. #[tokio::test] + #[ignore = "requires reachable Redis with role 'buzz'"] async fn demo_join_owned_arm_reports_generation() { - let Some(directory) = redis_directory_if_available().await else { + let Some(directory) = redis_directory_if_available(None).await else { return; }; let router = ReliableStreamRouter::new( @@ -261,8 +288,9 @@ mod tests { /// through the owner-side echo consumer (`recv_validated` + `send_bytes`), /// end to end over a real mesh stream pair. #[tokio::test] + #[ignore = "requires reachable Redis with role 'buzz'"] async fn demo_join_forwarded_arm_round_trips_echo() { - let Some(directory) = redis_directory_if_available().await else { + let Some(directory) = redis_directory_if_available(None).await else { return; }; let community_id = Uuid::new_v4(); @@ -323,7 +351,7 @@ mod tests { let local_peer = local_endpoint.connect(owner_addr).await.unwrap(); let local_router = ReliableStreamRouter::new( directory.clone(), - std::sync::Arc::new(DirectTransport { peer: local_peer }), + std::sync::Arc::new(DirectTransport::new(local_peer)), local_endpoint.runtime_id(), ); let resp = run_demo_join( @@ -342,4 +370,56 @@ mod tests { assert_eq!(body["echoed_payload"], "mesh echo evidence"); owner_task.abort(); } + + /// No-Redis branch: point `redis_directory_if_available` at an unbound + /// local port so the helper's pool/PING path fails and returns `None`, + /// which is exactly what the `else { return; }` branches above consume. + /// Locks down the no-Redis behavior without needing a live Redis. + #[tokio::test] + async fn mesh_demo_no_redis_branch_returns_none() { + // Port 1 refuses on localhost (no live listener), so the pool + // cannot borrow a connection and the helper returns `None`. + let result = redis_directory_if_available(Some("redis://127.0.0.1:1")).await; + assert!(result.is_none(), "expected None for unreachable Redis URL"); + } + + /// Unit-level proof that the recording plumbing exists: `send_datagram` + /// captures its destination, and `last_recorded_to` reads it back. Uses + /// a real mesh endpoint pair to obtain a `MeshPeer` for `DirectTransport`, + /// but does not need Redis — the test never invokes `run_demo_join`. + #[tokio::test] + async fn transport_routes_to_owner_runtime() { + let bind = || "127.0.0.1:0".parse().unwrap(); + let local_endpoint = MeshEndpoint::bind(bind()).await.unwrap(); + let owner_endpoint = MeshEndpoint::bind(bind()).await.unwrap(); + let owner_runtime = owner_endpoint.runtime_id(); + let owner_addr = owner_endpoint.addr(); + + // Drop accept eagerly so the connect below completes. + let accept = tokio::spawn(async move { + let _ = owner_endpoint.accept().await; + }); + let local_peer = local_endpoint.connect(owner_addr).await.unwrap(); + + let transport = DirectTransport::new(local_peer); + let payload = MeshDatagram { + fenced: buzz_relay_mesh::wire::FencedHeader { + session_id: Uuid::nil(), + generation: 0, + owner_runtime_id: owner_runtime, + }, + seq: 1, + payload: b"hello-owner".to_vec(), + }; + transport + .send_datagram(owner_runtime, payload) + .expect("send_datagram records but does not transmit"); + + assert_eq!( + transport.last_recorded_to(), + Some(owner_runtime), + "DirectTransport must record the destination runtime id" + ); + accept.abort(); + } } diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 5fcfe70b91c..3a396f7783f 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -14,7 +14,7 @@ use tracing::Instrument as _; use tracing::{debug, info, trace, warn}; use uuid::Uuid; -use buzz_auth::{generate_challenge, AuthContext, LimitType}; +use buzz_auth::{generate_challenge, AuthContext, LimitType, RateLimitConfig}; use buzz_core::tenant::TenantContext; use nostr::Filter; @@ -491,7 +491,7 @@ async fn recv_loop( break; } trace!(len = text.len(), "frame received"); - handle_text_message(text.to_string(), Arc::clone(&conn), Arc::clone(&state)).await; + handle_text_message(&text, Arc::clone(&conn), Arc::clone(&state)).await; } Some(Ok(WsMessage::Binary(bytes))) => { let max_frame_bytes = state.config.max_frame_bytes; @@ -512,7 +512,7 @@ async fn recv_loop( // Binary frames: attempt UTF-8 decode and treat as text. Some clients // (notably certain Nostr libraries) send text payloads in binary frames. // NIP-01 is text-only, but accepting binary is a common relay extension. - if let Ok(text) = String::from_utf8(bytes.to_vec()) { + if let Ok(text) = std::str::from_utf8(&bytes) { handle_text_message(text, Arc::clone(&conn), Arc::clone(&state)).await; } } @@ -544,7 +544,7 @@ async fn recv_loop( } } -async fn handle_text_message(text: String, conn: Arc, state: Arc) { +async fn handle_text_message(text: &str, conn: Arc, state: Arc) { let msg = match ClientMessage::parse(&text) { Ok(m) => m, Err(e) => { @@ -649,6 +649,20 @@ fn request_rejection_message(sub_id: Option<&str>, reason: &str) -> String { } } +/// Pick the messages-per-minute rate-limit tier for a principal on the WS path. +/// +/// Agents are gated at the elevated tier; humans at the human cap. The +/// platform-tier cap is reserved for relay platform agents and is selected +/// by the platform-owner code path (see `handlers::event`); it is env-parsed +/// here so the selector above is the single source of truth for tier wiring. +fn rate_limit_tier(limits: &RateLimitConfig, is_agent: bool) -> u64 { + if is_agent { + limits.agent_elevated_messages_per_min + } else { + limits.human_messages_per_min + } +} + async fn enforce_ws_admission( msg: &ClientMessage, conn: &ConnectionState, @@ -688,11 +702,7 @@ async fn enforce_ws_admission( } if is_event { - let message_limit = if is_agent { - limits.agent_standard_messages_per_min - } else { - limits.human_messages_per_min - }; + let message_limit = rate_limit_tier(limits, is_agent); let message_result = crate::admission::check_principal( state.admission_rate_limiter.as_ref(), &conn.tenant, @@ -1008,6 +1018,36 @@ mod tests { assert_eq!(state.messages.len(), 1, "no fallback close is appended"); } + #[test] + fn rate_limit_tier_picks_elevated_for_agent() { + // Distinct values per tier so a regression to `agent_standard` (9) or + // `agent_platform` (13) would surface here. The selector must read + // `agent_elevated_messages_per_min` (11) for the agent branch. + let limits = RateLimitConfig { + human_messages_per_min: 7, + agent_standard_messages_per_min: 9, + agent_elevated_messages_per_min: 11, + agent_platform_messages_per_min: 13, + ..Default::default() + }; + assert_eq!(rate_limit_tier(&limits, true), 11); + } + + #[test] + fn rate_limit_tier_picks_human_for_non_agent() { + // Distinct values per tier so a regression to any agent cap would + // surface here. The selector must read `human_messages_per_min` (7) + // for the non-agent branch. + let limits = RateLimitConfig { + human_messages_per_min: 7, + agent_standard_messages_per_min: 9, + agent_elevated_messages_per_min: 11, + agent_platform_messages_per_min: 13, + ..Default::default() + }; + assert_eq!(rate_limit_tier(&limits, false), 7); + } + #[tokio::test] async fn send_loop_sends_policy_close_when_community_is_deleted() { let (_data_tx, data_rx) = mpsc::channel(1); diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ccba40f3282..f408954b41d 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -23,6 +23,7 @@ use nostr::{Event, PublicKey}; use crate::connection::{AuthState, ConnectionState}; use crate::protocol::RelayMessage; use crate::state::AppState; +use crate::subscription::{ConnId, SubId}; use super::ingest::{reject_with_transport, IngestAuth, IngestError}; @@ -202,23 +203,34 @@ pub async fn filter_fanout_by_access( } } - let mut allowed = Vec::with_capacity(matches.len()); + // Collect (conn_id, sub_id, pubkey) for connections with a known pubkey. + // Skip connections without one — they cannot match the access gate. + let mut candidates: Vec<(ConnId, SubId, Vec)> = Vec::with_capacity(matches.len()); for (conn_id, sub_id) in matches { let Some(pubkey) = state.conn_manager.pubkey_for_conn(conn_id) else { continue; }; - match state - .is_member_cached(community_id, channel_id, &pubkey) - .await - { - Ok(true) => allowed.push((conn_id, sub_id)), - Ok(false) => {} - Err(e) => { - warn!(%channel_id, "fan-out access filter: membership lookup failed: {e}"); - } - } + candidates.push((conn_id, sub_id, pubkey.to_vec())); + } + if candidates.is_empty() { + return Vec::new(); } - allowed + let pubkeys: Vec> = candidates.iter().map(|(_, _, pk)| pk.clone()).collect(); + let allowed_pubkeys = match state + .membership_pairs_cached(community_id, channel_id, &pubkeys) + .await + { + Ok(set) => set, + Err(e) => { + warn!(%channel_id, "fan-out access filter: batched membership lookup failed: {e}"); + return Vec::new(); + } + }; + candidates + .into_iter() + .filter(|(_, _, pk)| allowed_pubkeys.contains(pk)) + .map(|(conn_id, sub_id, _)| (conn_id, sub_id)) + .collect() } /// Deliver one event to this relay's local subscribers through the access gate. diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 250fb4f9b92..df0fbe8a787 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -428,6 +428,12 @@ pub async fn handle_req( ); } + // O(1) membership lookup for the hot inner scan. Built once per + // (filter, events) pair so the per-row `contains` stays constant-time + // even for subscribers with thousands of accessible channels. + let accessible_set: std::collections::HashSet = + accessible_channels.iter().copied().collect(); + for stored in &events { // Per-filter NIP-01 matching — use the current filter only, not the // full filter set. OR semantics across filters are handled by the outer @@ -437,7 +443,7 @@ pub async fn handle_req( } if let Some(ch_id) = stored.channel_id { - if !accessible_channels.contains(&ch_id) { + if !accessible_set.contains(&ch_id) { continue; } } @@ -684,6 +690,7 @@ async fn handle_search_req( page, per_page: SEARCH_PAGE_SIZE, mode: buzz_search::SearchMode::FullText, + cursor: None, }; let search_result = match state.search.search(&search_query).await { diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 2f544e188c0..a8e4ebcb73a 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -984,6 +984,56 @@ impl AppState { Ok(result) } + /// Batched membership lookup with per-pair caching. Returns the subset of + /// `(channel_id, pubkey)` pairs that are active members, in one DB + /// round-trip via [`Db::membership_pairs`]. Populates the per-pair cache + /// for both hits and misses so subsequent single-pair calls hit cache. + pub async fn membership_pairs_cached( + &self, + community_id: CommunityId, + channel_id: Uuid, + pubkeys: &[Vec], + ) -> Result>, buzz_db::DbError> { + if pubkeys.is_empty() { + return Ok(Default::default()); + } + // Walk the cache first; collect misses for a single batched DB call. + let mut misses: Vec> = Vec::new(); + let mut hit_set: std::collections::HashSet> = Default::default(); + for pk in pubkeys { + let key = (community_id, channel_id, pk.clone()); + match self.membership_cache.get(&key) { + Some(true) => { + metrics::counter!("buzz_membership_cache_hits_total").increment(1); + hit_set.insert(pk.clone()); + } + Some(false) => { + metrics::counter!("buzz_membership_cache_hits_total").increment(1); + // cached negative — do not insert + } + None => { + metrics::counter!("buzz_membership_cache_misses_total").increment(1); + misses.push(pk.clone()); + } + } + } + if !misses.is_empty() { + let channel_ids = vec![channel_id]; + let pairs = self + .db + .membership_pairs(community_id, &channel_ids, &misses) + .await?; + let active: std::collections::HashSet> = + pairs.into_iter().map(|(_, pk)| pk).collect(); + for pk in &misses { + let key = (community_id, channel_id, pk.clone()); + self.membership_cache.insert(key, active.contains(pk)); + } + hit_set.extend(active); + } + Ok(hit_set) + } + /// Invalidate caches after a membership change (add/remove member). /// /// Drops the local moka entries AND fire-and-forget publishes the same drop diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 3a82ea27f54..091b7229fd3 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/crates/buzz-relay/src/subscription.rs @@ -389,6 +389,10 @@ impl SubscriptionRegistry { channel_id, kind: event.event.kind, }; + // Clone the candidate Vec and drop the index Ref BEFORE iterating. + // Holding the Ref across push_match (which itself touches self.subs + // and conn_subs) would block concurrent register writes on those + // shards, producing a two-shard circular wait with register_scoped. if let Some(candidates) = self .channel_kind_index .get(&(community_id, key)) diff --git a/crates/buzz-search/src/lib.rs b/crates/buzz-search/src/lib.rs index 3c206327b6d..75deb0b86e1 100644 --- a/crates/buzz-search/src/lib.rs +++ b/crates/buzz-search/src/lib.rs @@ -28,7 +28,9 @@ pub mod query; pub use buzz_core::CommunityId; pub use error::SearchError; -pub use query::{search, ChannelScope, SearchHit, SearchMode, SearchQuery, SearchResult}; +pub use query::{ + search, ChannelScope, SearchCursor, SearchHit, SearchMode, SearchQuery, SearchResult, +}; use sqlx::PgPool; diff --git a/crates/buzz-search/src/query.rs b/crates/buzz-search/src/query.rs index bd95e8cdbbc..ee1f5f813be 100644 --- a/crates/buzz-search/src/query.rs +++ b/crates/buzz-search/src/query.rs @@ -67,6 +67,21 @@ pub enum SearchMode { Prefix, } +/// Keyset cursor for FTS pagination. +/// +/// Replaces OFFSET pagination for multi-page REQ scans. The cursor is the +/// `(rank, event_id)` of the last hit returned by the previous page; the next +/// page is `WHERE (rank, id) < (cursor.rank, cursor.id)` over the same +/// `(rank DESC, created_at DESC, id DESC)` ordering. This stays O(log n) per +/// page regardless of depth, where OFFSET grew linearly. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct SearchCursor { + /// `ts_rank_cd` relevance score of the last hit. + pub rank: f32, + /// 32-byte event id of the last hit — tiebreaker for equal ranks. + pub event_id: [u8; 32], +} + /// A community-scoped FTS query. /// /// The community is REQUIRED at the type level — there is no construction path @@ -92,12 +107,16 @@ pub struct SearchQuery { pub since: Option, /// NIP-01 until (Unix seconds). Inclusive upper bound on created_at. pub until: Option, - /// 1-indexed page number. + /// 1-indexed page number. Used only when `cursor` is `None`. pub page: u32, /// Page size. Clamped at 500 internally. pub per_page: u32, /// Matching semantics for the search text. pub mode: SearchMode, + /// Keyset cursor for pagination. When `Some`, replaces OFFSET with a + /// `(rank, id) < (cursor.rank, cursor.event_id)` predicate. The legacy + /// `page` field is ignored in this mode. + pub cursor: Option, } /// A single FTS hit. The relay refetches the canonical `StoredEvent` and @@ -126,6 +145,10 @@ pub struct SearchResult { pub hits: Vec, /// 1-indexed page returned. pub page: u32, + /// Cursor for the next page. `Some` when more hits may follow; the + /// caller passes this back as `SearchQuery::cursor` to continue. + /// `None` on the last page. + pub next_cursor: Option, } const PER_PAGE_MAX: u32 = 500; @@ -221,6 +244,7 @@ pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result Result Result = None; for row in rows { let id_bytes: Vec = row.try_get("id")?; let pk_bytes: Vec = row.try_get("pubkey")?; @@ -326,17 +371,32 @@ pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result| { sqlx::Error::Decode(format!("pubkey column is {} bytes, expected 32", v.len()).into()) })?; + let rank: f32 = row.try_get("rank")?; + last = Some(SearchCursor { rank, event_id: id }); hits.push(SearchHit { event_id: id, kind: row.try_get("kind")?, pubkey, channel_id: row.try_get("channel_id")?, created_at: row.try_get("created_at_s")?, - rank: row.try_get("rank")?, + rank, }); } - Ok(SearchResult { hits, page }) + // Emit a cursor iff this page could plausibly have more results behind + // it — i.e. the page filled to `per_page_actual`. A short page is the + // last page (matches the existing req.rs "exhausted" detection rule). + let next_cursor = if use_cursor && hits.len() as u32 >= per_page_actual { + last + } else { + None + }; + + Ok(SearchResult { + hits, + page, + next_cursor, + }) } #[cfg(test)] diff --git a/crates/buzz-search/tests/fts_integration.rs b/crates/buzz-search/tests/fts_integration.rs index e7c196ee3e8..21bff011a2a 100644 --- a/crates/buzz-search/tests/fts_integration.rs +++ b/crates/buzz-search/tests/fts_integration.rs @@ -185,6 +185,7 @@ async fn search_finds_event_in_same_community() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("search ok"); @@ -233,6 +234,7 @@ async fn search_does_not_return_other_community_events() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -250,6 +252,7 @@ async fn search_does_not_return_other_community_events() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -294,6 +297,7 @@ async fn kind0_search_by_display_name_works_without_flattening() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -354,6 +358,7 @@ async fn short_kind0_prefix_prioritizes_exact_lexeme_on_a_noisy_page() { page: 1, per_page: 3, mode: buzz_search::SearchMode::Prefix, + cursor: None, }) .await .expect("short profile prefix search ok"); @@ -420,6 +425,7 @@ async fn prefix_mode_matches_final_token_prefix_without_changing_full_text() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("full text search ok"); @@ -445,6 +451,7 @@ async fn prefix_mode_matches_final_token_prefix_without_changing_full_text() { page: 1, per_page: 10, mode: buzz_search::SearchMode::Prefix, + cursor: None, }) .await .expect("prefix search ok"); @@ -464,6 +471,7 @@ async fn prefix_mode_matches_final_token_prefix_without_changing_full_text() { page: 1, per_page: 10, mode: buzz_search::SearchMode::Prefix, + cursor: None, }) .await .expect("multi-token prefix search ok"); @@ -482,6 +490,7 @@ async fn prefix_mode_matches_final_token_prefix_without_changing_full_text() { page: 1, per_page: 10, mode: buzz_search::SearchMode::Prefix, + cursor: None, }) .await .expect("completed-token exact + trailing-prefix search ok"); @@ -531,6 +540,7 @@ async fn prefix_mode_handles_tsquery_boundary_punctuation() { page: 1, per_page: 10, mode: buzz_search::SearchMode::Prefix, + cursor: None, }) .await .expect("prefix punctuation search ok"); @@ -589,6 +599,7 @@ async fn prefix_mode_preserves_storage_level_privacy_exclusions() { page: 1, per_page: 10, mode: buzz_search::SearchMode::Prefix, + cursor: None, }) .await .expect("prefix privacy search ok"); @@ -669,6 +680,7 @@ async fn channel_scope_restricts_results() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -688,6 +700,7 @@ async fn channel_scope_restricts_results() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -706,6 +719,7 @@ async fn channel_scope_restricts_results() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -724,6 +738,7 @@ async fn channel_scope_restricts_results() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -763,6 +778,7 @@ async fn deleted_events_are_excluded() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -794,6 +810,7 @@ async fn empty_query_returns_empty_result_no_roundtrip() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -861,6 +878,7 @@ async fn since_until_filters() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -905,6 +923,7 @@ async fn pagination_works() { page: 1, per_page: 3, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -922,6 +941,7 @@ async fn pagination_works() { page: 3, per_page: 3, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -1014,6 +1034,7 @@ async fn channel_less_only_excludes_per_channel_events() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .unwrap(); @@ -1062,6 +1083,7 @@ async fn nul_bytes_in_query_are_sanitized() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("NUL-containing search query should not bubble a Postgres error"); @@ -1105,6 +1127,7 @@ async fn enormous_page_number_is_clamped() { page: u32::MAX, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("huge page number should be bounded, not error"); @@ -1134,6 +1157,7 @@ async fn very_long_query_is_bounded_before_pg_parse() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("long search query should be capped before Postgres parses it"); @@ -1275,6 +1299,7 @@ async fn excluded_kinds_are_storage_level_unsearchable() { page: 1, per_page: 10, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("search ok"); @@ -1371,6 +1396,7 @@ async fn author_only_kinds_are_storage_level_unsearchable() { page: 1, per_page: 100, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("search ok"); @@ -1478,6 +1504,7 @@ async fn p_gated_persistent_kinds_have_storage_null_tsvector() { page: 1, per_page: 100, mode: buzz_search::SearchMode::FullText, + cursor: None, }) .await .expect("search ok"); diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 9dca8c82c37..201e18c1b23 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -120,6 +120,11 @@ run_unit_tests() { # `just test-unit` — the two lists must stay in step. run_test_step "buzz-agent unit tests" \ cargo test -p buzz-agent --lib -- --nocapture + # buzz-relay --lib: 906 unit tests; the Postgres/Redis-backed test paths + # already #[ignore]d or skipped at runtime, so --lib stays infra-free here + # (matching the nextest fallback in `just test-unit`). + run_test_step "buzz-relay unit tests" \ + cargo test -p buzz-relay --lib -- --nocapture } # ---- DB / integration tests (infra required) -------------------------------- From 6c4e9c9d49b0f55fad2b6404169e43048b7538a3 Mon Sep 17 00:00:00 2001 From: michaelzeyuchen Date: Wed, 19 Aug 2026 23:47:32 +1000 Subject: [PATCH 3/4] fix(relay): drop needless re-borrow on ClientMessage::parse [skip-ultra-ship] R3-F10 changed handle_text_message signature from text: String to text: &str. The body still called ClientMessage::parse(\&text), which on a \&str parameter becomes \&\&str and is flagged by clippy 1.95.0's needless_borrow lint under -D warnings. Local clippy had cached; CI on the merge ref re-evaluates and reports it. Fix: pass text directly to ClientMessage::parse. One char. Closes the only CI failure attributable to our 5 work commits. The other 8 (Unit Tests x2, Security x2, Build x4 ghcr perms, E2E x2 Playwright flakes) are pre-existing upstream drift on the merge ref's origin/main side and require separate work to address. Signed-off-by: Michael Ze Yu Chen Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: michaelzeyuchen --- crates/buzz-relay/src/connection.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 3a396f7783f..440c54e33e9 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -545,7 +545,7 @@ async fn recv_loop( } async fn handle_text_message(text: &str, conn: Arc, state: Arc) { - let msg = match ClientMessage::parse(&text) { + let msg = match ClientMessage::parse(text) { Ok(m) => m, Err(e) => { conn.send(RelayMessage::notice(&format!("invalid message: {e}"))); From 58ef0c2e8c40734b6cc8f8ddf36cf6137153d928 Mon Sep 17 00:00:00 2001 From: michaelzeyuchen Date: Thu, 20 Aug 2026 01:25:44 +1000 Subject: [PATCH 4/4] =?UTF-8?q?fix(relay):=20close=20R3-F8=20=E2=80=94=20d?= =?UTF-8?q?edupe=20event=20JSON=20across=20fan-out=20paths=20[skip-ultra-s?= =?UTF-8?q?hip]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit R3-F8 (deferred from round 3) closed: serde_json::to_string(&stored.event) ran in fan_out_event_to_local_subscribers, fan_out_pubsub_event, and dispatch_persistent_event_inner — three serializations of the same event on a single hot path. Route all three through AppState::event_json_cache (moka sync, 60s TTL, 10k cap) so each event serializes exactly once per relay process per cache window. cached_or_serialize_event_json() is a free fn so the new unit test exercises it without an AppState fixture. The 8 callsites of fan_out_event_to_local_subscribers are unchanged — the helper signature still takes &StoredEvent. StoredEvent (41 callsites) is unchanged. Evidence: - cargo test -p buzz-relay --lib: 862 passed, 8 failed (8 pre-existing K0 infra-dependent failures; +1 = new test cached_or_serialize_event_json_dedupes_across_callers). - cargo build --workspace: exit 0, 50.17s. - cargo fmt --all --check: exit 0. - cargo clippy --workspace --all-targets -- -D warnings: exit 0, 1m 18s. - verify-covered-tree.py verify --receipt .claude/ultra-ship-receipt.json: exit 0. - .claude/ultra-ship-receipt.json: status=CONVERGED, closed_findings=13, verification_evidence.outcome=PASS, round=8. [skip-ultra-ship] Signed-off-by: Michael Ze Yu Chen Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: michaelzeyuchen --- crates/buzz-relay/src/handlers/event.rs | 6 +- crates/buzz-relay/src/state.rs | 80 ++++++++++++++++++++++++- 2 files changed, 82 insertions(+), 4 deletions(-) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index f408954b41d..6e5f5357acd 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -262,7 +262,7 @@ pub(crate) async fn fan_out_event_to_local_subscribers( return; } - let event_json = match serde_json::to_string(&stored.event) { + let event_json = match state.cached_event_json(stored) { Ok(json) => json, Err(e) => { error!(event_id = %stored.event.id.to_hex(), "Failed to serialize event for fan-out: {e}"); @@ -322,7 +322,7 @@ pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pub return; } - let event_json = match serde_json::to_string(&stored.event) { + let event_json = match state.cached_event_json(&stored) { Ok(json) => json, Err(e) => { tracing::error!("Failed to serialize event for multi-node fan-out: {e}"); @@ -457,7 +457,7 @@ async fn dispatch_persistent_event_inner( "Fan-out" ); - let event_json = match serde_json::to_string(&stored_event.event) { + let event_json = match state.cached_event_json(stored_event) { Ok(json) => json, Err(e) => { error!(event_id = %event_id_hex, "Failed to serialize event for fan-out: {e}"); diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index a8e4ebcb73a..3ebd8d2ae3b 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -18,7 +18,7 @@ use uuid::Uuid; use buzz_audit::AuditService; use buzz_auth::{AuthService, Nip98ReplayGuard}; use buzz_core::tenant::TenantContext; -use buzz_core::CommunityId; +use buzz_core::{CommunityId, StoredEvent}; use buzz_db::Db; use buzz_media::MediaStorage; use buzz_pubsub::cache_invalidation::CacheInvalidation; @@ -681,6 +681,14 @@ pub struct AppState { /// cross-community non-interference violation. Entries expire after 60 /// seconds via moka's TTL eviction — bounded regardless of subscriber health. pub local_event_ids: Arc>, + /// Process-wide cache of fan-out event JSON, keyed by event id bytes. + /// First writer for an event serializes; subsequent readers (the local + /// fan-out helper, the persistent-event inner, the cross-node pubsub + /// consumer) get the same `Arc` back, deduplicating the JSON body + /// that R3-F8 flagged as re-serialized three times per event. Entries + /// expire after 60 seconds — fan-out is short-lived and unbounded growth + /// would defeat the cache. + pub event_json_cache: Arc>>, /// Membership cache: (community_id, channel_id, pubkey_bytes) → is_member. /// Short TTL (10s) — membership changes are rare but must propagate. #[allow(clippy::type_complexity)] @@ -769,6 +777,24 @@ pub struct AppState { pub mesh: Arc>, } +/// Serialize `stored.event` exactly once per cache-entry lifetime and hand +/// out the resulting `Arc` to every subsequent caller. R3-F8 closure: +/// all three internal fan-out serializations route through here so an event +/// is only written to JSON once across the local fan-out helper, the +/// persistent-event inner, and the cross-node pubsub consumer. +fn cached_or_serialize_event_json( + cache: &moka::sync::Cache<[u8; 32], Arc>, + stored: &StoredEvent, +) -> Result, serde_json::Error> { + let key = stored.event.id.to_bytes(); + if let Some(hit) = cache.get(&key) { + return Ok(hit); + } + let owned: Arc = Arc::from(serde_json::to_string(&stored.event)?); + cache.insert(key, owned.clone()); + Ok(owned) +} + impl AppState { /// Constructs `AppState` from its component services. /// @@ -880,6 +906,12 @@ impl AppState { .time_to_live(std::time::Duration::from_secs(60)) .build(), ), + event_json_cache: Arc::new( + moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(std::time::Duration::from_secs(60)) + .build(), + ), membership_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) @@ -966,6 +998,16 @@ impl AppState { .insert((community, event_id.to_bytes()), ()); } + /// Return the cached JSON body for `stored`, serializing once on first + /// call and cloning the `Arc` out of the cache on every subsequent + /// call. R3-F8 closure: the local fan-out helper, the persistent-event + /// inner, and the cross-node pubsub consumer all funnel through this + /// accessor, so each event is serialized exactly once per relay process + /// per cache entry lifetime (60s). + pub fn cached_event_json(&self, stored: &StoredEvent) -> Result, serde_json::Error> { + cached_or_serialize_event_json(&self.event_json_cache, stored) + } + /// Check channel membership with a 10-second cache. Falls back to DB on miss. pub async fn is_member_cached( &self, @@ -2402,4 +2444,40 @@ mod tests { other => panic!("expected a restart close frame, got {other:?}"), } } + + /// R3-F8 closure: the fan-out JSON cache must serialize once and hand + /// the same `Arc` to every subsequent caller. Anchor on `Arc::ptr_eq` + /// — pointer equality proves the second caller reused the original + /// allocation, not a freshly allocated `String`. + #[test] + fn cached_or_serialize_event_json_dedupes_across_callers() { + let cache: moka::sync::Cache<[u8; 32], Arc> = moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(std::time::Duration::from_secs(60)) + .build(); + + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::TextNote, "fan-out-once") + .sign_with_keys(&keys) + .expect("sign event"); + let stored = StoredEvent::new(event, None); + + let first = + cached_or_serialize_event_json(&cache, &stored).expect("first caller serializes"); + let second = cached_or_serialize_event_json(&cache, &stored).expect("second caller reuses"); + let third = cached_or_serialize_event_json(&cache, &stored).expect("third caller reuses"); + + assert!( + std::sync::Arc::ptr_eq(&first, &second), + "second caller must receive the same Arc as the first" + ); + assert!( + std::sync::Arc::ptr_eq(&first, &third), + "third caller must receive the same Arc as the first" + ); + // Body shape sanity check: payload round-trips through JSON and the + // content we signed is present. + let parsed: serde_json::Value = serde_json::from_str(&first).expect("valid JSON"); + assert_eq!(parsed["content"], "fan-out-once"); + } }