jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(buzz-acp): bound sibling-gate lookup failures instead of caching them (#7216)

Browse Source 
## Summary

Fixes #5450. Supersedes #5459 (see "Related issue").

In `owner-only` / `allowlist` respond modes, the inbound author gate
admits a sibling agent (same owner, proven by the NIP-OA auth tag on its
kind:0) only after `check_sibling_via_profile` fetches that profile over
REST (`POST /query`, 2 s timeout). The boolean result of that lookup was
cached in `OwnerCache` **for the process lifetime, unconditionally** —
so a timeout, a network error, or a non-2xx response was cached as "not
a sibling". One relay blip at first contact and the agent silently
ignores that teammate bot until restart. The owner never hits this path
(`author == owner` short-circuits with no network call), which is
exactly the observed shape: agents keep answering their owner while
dropping every agent-to-agent mention.

Two things kept it invisible: the author-gate drop was logged at `debug`
(hidden by the default filter), and `RestClient::request_with_retry`
returned non-retriable HTTP statuses (401/403/500) with no log line at
all.

### What changed (`crates/buzz-acp`)

- **Tri-state verdict.** `check_sibling_via_profile` returns
`SiblingVerdict::{Sibling, NotSibling, Indeterminate}`. Only a *fetched*
profile yields a cacheable verdict. A failed query, a malformed body, or
a missing kind:0 (not published, or not yet replicated to this relay) is
`Indeterminate` — nothing was attested.
- **Bounded per-author retry instead of retry-per-event.** An
indeterminate lookup fails closed for the current event and arms a retry
deadline: 5 s, doubling per consecutive failure to a 60 s cap
(`SIBLING_RETRY_BASE` / `SIBLING_RETRY_MAX`). Events inside the deadline
are denied with no relay call; expiry re-arms exactly one lookup; a
proven verdict replaces the deadline. This matters because the lookup
awaits inline in the sole inbound-event loop — an unthrottled retry
under a persistent `/query` outage lets one chatty sibling stall ingress
for the owner too. With the backoff, a dead relay costs at most one
inline lookup per author per minute.
- **No wholesale cache clear at capacity.** Replacing an author already
present never evicts. A new author at the 256-entry cap evicts exactly
one entry — the longest-expired retry record if any, else an arbitrary
entry — so proven verdicts and the other authors' backoff tiers (their
`failures` counters) survive saturation.
- **Diagnosability.** The author-gate drop logs at INFO;
`request_with_retry` WARNs on non-retriable HTTP statuses.

### Reviewer notes

- Fail-closed security posture is unchanged: an unverifiable author is
still dropped. Only what is *remembered* about that author changes.
- A fetched profile with no verifying owner attestation is still a
cached `NotSibling`, so a stranger spamming an owner-only channel
doesn't cost a REST round-trip per message. A *missing* profile is
retried under the backoff, which removes the need to restart a harness
after a sibling's profile is published late.
- When no expired retry record exists, the single eviction victim is an
arbitrary map entry rather than LRU; per-entry access timestamps didn't
seem worth it for a cap the fleet sits far below. Happy to change if
you'd prefer it prefer throttled entries over proven ones.

### Related issue

- Fixes #5450.
- #5459 by @whyihaveyou introduced the tri-state verdict and the
no-cache-on-indeterminate rule for the same bug. It has been open
without review since 2026-08-10 and is now conflicting with `main`; it
also retries the inline lookup on every event and leaves the wholesale
`map.clear()` in place. This PR carries the same first half and adds the
bounded retry and eviction behaviour that came out of review on our fork
(yjc801/buzz#112). Credit for the diagnosis and the verdict design goes
to that PR.

### Testing

Nine regression tests in `author_gate_tests` bind the production seam
(`is_owner_or_sibling` → `check_sibling_via_profile`) against loopback
`/query` stubs:

- unreachable relay ⇒ denied, **not** cached as a verdict
- missing profile (`[]`) ⇒ denied, not cached, throttled under the
backoff
- fetched profile without a verifying tag ⇒ denied and cached `false`
- kind:0 carrying a real `compute_auth_tag` attestation ⇒ admitted and
cached `true`
- flaky relay (500 until healed): a burst inside the deadline makes
**no** relay calls, expiry re-queries once, recovery caches the proven
verdict and ends throttling
- backoff sequence 5 → 10 → cap, tracked per author
- re-arming one author's deadline in a full cache keeps the other 255
proven verdicts
- a new author at capacity displaces an expired record first, then
exactly one entry
- with two expired records at capacity, only the longest-expired is
evicted and the survivor keeps its backoff tier

`cargo test -p buzz-acp`: all unit and lifecycle tests pass on this
branch; `cargo clippy -p buzz-acp --all-targets` and `cargo fmt --check`
clean. Commit is DCO-signed.

Live context that led here: on a hosted relay whose REST endpoint
flapped for a day, every local agent that restarted with an empty cache
stopped answering its teammates within hours while still answering the
owner; two agents sat online with zero turns. Not live-verified post-fix
upstream — verified by the tests above and by the fork's review rounds.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Junchao Yan <yjc801@gmail.com>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Junchao Yan
2026-10-06 14:14:04 -07:00
committed by GitHub
parent 1a2ddf9eaa
commit 003770cb79
2 changed files with 624 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

648
crates/buzz-acp/src/lib.rs
View File

@@ -170,24 +170,80 @@ fn resolve_agent_owner(config: &Config) -> Option<String> {
config.agent_owner.clone()
}
/// Base delay before re-consulting the relay about an author whose last
/// sibling lookup was [`SiblingVerdict::Indeterminate`]. Doubles per
/// consecutive indeterminate result up to [`SIBLING_RETRY_MAX`].
///
/// The lookup awaits inline in the sole inbound-event loop (2s timeout), so an
/// unthrottled retry-per-event under a persistent `/query` outage would let a
/// chatty sibling stall ingress for the owner too. With this backoff a dead
/// relay costs at most one inline lookup per author per minute.
const SIBLING_RETRY_BASE: Duration = Duration::from_secs(5);
/// Ceiling for the indeterminate-lookup retry backoff.
const SIBLING_RETRY_MAX: Duration = Duration::from_secs(60);
/// One author's entry in the sibling cache.
#[derive(Debug, Clone, Copy)]
enum SiblingEntry {
/// Relay-proven verdict; immutable for the process lifetime.
Proven(bool),
/// Last lookup was indeterminate: denied without a relay call until
/// `retry_at`. `failures` counts consecutive indeterminate results and
/// drives the exponential backoff.
Indeterminate {
retry_at: std::time::Instant,
failures: u32,
},
}
/// What the cache can say about an author before any relay call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SiblingCacheState {
/// Relay-proven verdict.
Proven(bool),
/// An indeterminate verdict is still inside its retry deadline: deny
/// this event without consulting the relay.
Throttled,
/// Never seen, or the retry deadline has passed: consult the relay.
Unknown,
}
/// Cache for the agent's owner pubkey.
///
/// Owner is now provided via `--agent-owner` config flag (no REST lookup).
/// Cache for the agent's owner pubkey + sibling lookups.
///
/// Siblings are other agents whose NIP-OA auth tag proves the same owner.
/// Lookup results are cached for the process lifetime (attestations are immutable).
/// Only *proven* verdicts are cached for the process lifetime (attestations
/// are immutable). A lookup that failed because the relay was unreachable is
/// [`SiblingVerdict::Indeterminate`]: it is remembered only as a bounded retry
/// deadline ([`SiblingEntry::Indeterminate`]) — never as a verdict — so one
/// relay blip cannot make this agent ignore a same-owner sibling until
/// restart, and a persistent outage cannot turn every sibling event into an
/// inline relay round-trip.
struct OwnerCache {
pubkey: Option<String>,
/// author_hex → is_sibling (true = same owner, false = not)
siblings: std::sync::Mutex<HashMap<String, bool>>,
siblings: std::sync::Mutex<HashMap<String, SiblingEntry>>,
retry_base: Duration,
retry_max: Duration,
}
impl OwnerCache {
fn new(initial: Option<String>) -> Self {
Self::with_retry_backoff(initial, SIBLING_RETRY_BASE, SIBLING_RETRY_MAX)
}
/// Construct with an explicit indeterminate-retry backoff (tests shorten it).
fn with_retry_backoff(
initial: Option<String>,
retry_base: Duration,
retry_max: Duration,
) -> Self {
Self {
pubkey: initial,
siblings: std::sync::Mutex::new(HashMap::new()),
retry_base,
retry_max,
}
}
@@ -196,27 +252,134 @@ impl OwnerCache {
self.pubkey.as_deref()
}
/// Check if author is a known sibling (cached result).
fn is_known_sibling(&self, author: &str) -> Option<bool> {
self.siblings.lock().ok()?.get(author).copied()
}
/// Cache a sibling lookup result.
fn cache_sibling(&self, author: String, is_sibling: bool) {
if let Ok(mut map) = self.siblings.lock() {
// Cap at 256 entries to prevent unbounded growth.
if map.len() >= 256 {
map.clear();
/// What is known about `author` right now, without touching the relay.
fn sibling_state(&self, author: &str) -> SiblingCacheState {
let Ok(map) = self.siblings.lock() else {
return SiblingCacheState::Unknown;
};
match map.get(author) {
None => SiblingCacheState::Unknown,
Some(SiblingEntry::Proven(is_sibling)) => SiblingCacheState::Proven(*is_sibling),
Some(SiblingEntry::Indeterminate { retry_at, .. }) => {
if std::time::Instant::now() < *retry_at {
SiblingCacheState::Throttled
} else {
SiblingCacheState::Unknown
}
}
map.insert(author, is_sibling);
}
}
/// The relay-proven verdict for `author`, if one has been cached.
#[cfg(test)]
fn proven_verdict(&self, author: &str) -> Option<bool> {
match self.sibling_state(author) {
SiblingCacheState::Proven(is_sibling) => Some(is_sibling),
_ => None,
}
}
/// Cache a proven sibling verdict.
fn cache_sibling(&self, author: String, is_sibling: bool) {
self.insert(author, SiblingEntry::Proven(is_sibling));
}
/// Record an indeterminate lookup for `author` and return the backoff
/// during which further events from them are denied without a relay call.
fn note_indeterminate(&self, author: &str) -> Duration {
let failures = match self
.siblings
.lock()
.ok()
.and_then(|map| map.get(author).copied())
{
Some(SiblingEntry::Indeterminate { failures, .. }) => failures.saturating_add(1),
_ => 1,
};
let backoff = self
.retry_base
.saturating_mul(1u32 << (failures - 1).min(16))
.min(self.retry_max);
self.insert(
author.to_string(),
SiblingEntry::Indeterminate {
retry_at: std::time::Instant::now() + backoff,
failures,
},
);
backoff
}
/// Number of cached authors at the eviction threshold.
const SIBLING_CACHE_CAP: usize = 256;
/// Insert or replace `author`'s entry under the bounded-size policy.
///
/// Replacing an author already present never evicts anything — a retry
/// deadline being re-armed at capacity must not wipe the other entries,
/// or a saturated cache under a `/query` outage would collapse every
/// sibling back to a serial inline lookup per event. A previously absent
/// author at capacity evicts exactly one entry — the space the newcomer
/// needs: the longest-expired retry record if there is one, else an
/// arbitrary entry. Expired records are not purged wholesale because their
/// `failures` counter is the author's backoff tier; dropping it would
/// restart every affected author at the base delay after its next failed
/// lookup and multiply serial relay calls during the same outage.
fn insert(&self, author: String, entry: SiblingEntry) {
let Ok(mut map) = self.siblings.lock() else {
return;
};
if !map.contains_key(&author) && map.len() >= Self::SIBLING_CACHE_CAP {
let now = std::time::Instant::now();
let victim = map
.iter()
.filter_map(|(cached, existing)| match existing {
SiblingEntry::Indeterminate { retry_at, .. } if *retry_at <= now => {
Some((cached, *retry_at))
}
_ => None,
})
.min_by_key(|(_, retry_at)| *retry_at)
.map(|(cached, _)| cached.clone())
.or_else(|| map.keys().next().cloned());
if let Some(victim) = victim {
map.remove(&victim);
}
}
map.insert(author, entry);
}
/// Number of authors currently cached (proven or throttled).
#[cfg(test)]
fn len(&self) -> usize {
self.siblings.lock().map(|map| map.len()).unwrap_or(0)
}
}
/// Outcome of one sibling profile verification attempt.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SiblingVerdict {
/// kind:0 fetched and its NIP-OA auth tag cryptographically proves the
/// same owner.
Sibling,
/// The relay returned the author's profile and it does not prove the
/// same owner.
NotSibling,
/// Nothing was attested either way: the relay could not be consulted
/// (timeout, network, or HTTP error), answered with a malformed body, or
/// has no kind:0 for the author yet — a sibling's profile may simply not
/// have been published or replicated, so its absence is not a verdict.
Indeterminate,
}
/// Check if `author` is the owner OR a sibling (same owner via NIP-OA).
///
/// For unknown authors, queries their kind:0 profile to extract the NIP-OA
/// auth tag and verify the owner matches. Result is cached.
/// auth tag and verify the owner matches. Only proven verdicts are cached as
/// verdicts; an [`SiblingVerdict::Indeterminate`] lookup fails closed for the
/// current event and arms a bounded retry deadline, so a transient relay
/// outage cannot poison a sibling for the process lifetime and a persistent
/// one cannot turn every sibling event into an inline relay round-trip.
async fn is_owner_or_sibling(
author: &str,
owner_cache: &OwnerCache,
@@ -233,14 +396,39 @@ async fn is_owner_or_sibling(
}
// Check sibling cache.
if let Some(cached) = owner_cache.is_known_sibling(author) {
return cached;
match owner_cache.sibling_state(author) {
SiblingCacheState::Proven(is_sibling) => return is_sibling,
SiblingCacheState::Throttled => {
tracing::debug!(
author,
"sibling check throttled after an indeterminate lookup — denying without a relay query"
);
return false;
}
SiblingCacheState::Unknown => {}
}
// Query the author's kind:0 profile to check for NIP-OA auth tag.
let is_sibling = check_sibling_via_profile(author, my_owner, rest_client).await;
owner_cache.cache_sibling(author.to_string(), is_sibling);
is_sibling
match check_sibling_via_profile(author, my_owner, rest_client).await {
SiblingVerdict::Sibling => {
owner_cache.cache_sibling(author.to_string(), true);
true
}
SiblingVerdict::NotSibling => {
owner_cache.cache_sibling(author.to_string(), false);
false
}
SiblingVerdict::Indeterminate => {
let backoff = owner_cache.note_indeterminate(author);
tracing::warn!(
author,
retry_after_secs = backoff.as_secs(),
"sibling check indeterminate — relay profile query failed; \
dropping this event without a proven verdict, retrying after backoff"
);
false
}
}
}
/// Return the workflow owner attributed by a relay-signed workflow message.
@@ -550,7 +738,10 @@ mod inbound_author_gate {
)
.await;
if !decision.allowed {
tracing::debug!(
// INFO, not debug: under owner-only fleets a dropped event is
// the whole story of "the agent never answered", and the
// default log filter hides debug.
tracing::info!(
channel_id = %buzz_event.channel_id,
raw_author = %buzz_event.event.pubkey.to_hex(),
effective_author = %decision.effective_author,
@@ -848,16 +1039,22 @@ pub(crate) async fn is_dm_channel(
/// Query an author's kind:0 profile and check if their NIP-OA auth tag
/// proves the same owner as us.
///
/// Only a *fetched* profile yields a verdict the caller may cache: a
/// profile with no verifying tag is definitively [`SiblingVerdict::NotSibling`],
/// while a failed query or a missing profile is
/// [`SiblingVerdict::Indeterminate`] and retried under the bounded backoff.
async fn check_sibling_via_profile(
author: &str,
expected_owner: &str,
rest_client: &relay::RestClient,
) -> bool {
) -> SiblingVerdict {
let filter = nostr::Filter::new()
.kind(nostr::Kind::Metadata)
.author(match nostr::PublicKey::from_hex(author) {
Ok(pk) => pk,
Err(_) => return false,
// A malformed author key can never verify — definitively not a sibling.
Err(_) => return SiblingVerdict::NotSibling,
})
.limit(1);
@@ -865,28 +1062,39 @@ async fn check_sibling_via_profile(
.await
{
Ok(Ok(v)) => v,
_ => return false, // timeout or error — fail closed
Ok(Err(e)) => {
tracing::debug!(author, "sibling profile query failed: {e}");
return SiblingVerdict::Indeterminate;
}
Err(_) => {
tracing::debug!(author, "sibling profile query timed out");
return SiblingVerdict::Indeterminate;
}
};
// Look for an "auth" tag in the profile event.
let events = match resp.as_array() {
Some(arr) => arr,
None => return false,
// Non-array body is a server anomaly, not proof of absence.
None => return SiblingVerdict::Indeterminate,
};
let event = match events.first() {
Some(e) => e,
None => return false,
// No kind:0 yet — not published, or not replicated to this relay.
// Not a verdict: the bounded backoff retries it, so a profile that
// appears later is picked up without a restart.
None => return SiblingVerdict::Indeterminate,
};
let tags = match event.get("tags").and_then(|t| t.as_array()) {
Some(t) => t,
None => return false,
None => return SiblingVerdict::NotSibling,
};
// Find ["auth", owner_pk, conditions, sig] and verify the Schnorr signature.
// Don't trust the relay — verify ourselves.
let agent_pk = match nostr::PublicKey::from_hex(author) {
Ok(pk) => pk,
Err(_) => return false,
Err(_) => return SiblingVerdict::NotSibling,
};
for tag in tags {
@@ -910,7 +1118,7 @@ async fn check_sibling_via_profile(
match buzz_sdk::nip_oa::verify_auth_tag(&tag_json, &agent_pk) {
Ok(_) => {
tracing::debug!(author, expected_owner, "sibling verified via NIP-OA");
return true;
return SiblingVerdict::Sibling;
}
Err(e) => {
tracing::debug!(author, "NIP-OA auth tag verification failed: {e}");
@@ -918,7 +1126,7 @@ async fn check_sibling_via_profile(
}
}
false
SiblingVerdict::NotSibling
}
/// Observer frames are published at a global rate of AT MOST ONE relay frame
@@ -8307,6 +8515,384 @@ mod author_gate_tests {
);
}
// ── sibling verdict caching ───────────────────────────────────────────
//
// A sibling lookup that failed because the relay was unreachable is
// Indeterminate: it fails closed for that one event but must NOT be
// cached, or a single relay blip makes this agent silently ignore a
// same-owner sibling until the process restarts.
/// Serve one JSON body for every request on a loopback port, as the
/// `/query` endpoint the sibling check consults.
async fn sibling_query_server(
body: serde_json::Value,
) -> (relay::RestClient, tokio::task::JoinHandle<()>) {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind sibling query server");
let base_url = format!("http://{}", listener.local_addr().unwrap());
let body = body.to_string();
let server = tokio::spawn(async move {
loop {
let Ok((mut socket, _)) = listener.accept().await else {
break;
};
let mut request = vec![0; 8192];
let _ = socket.read(&mut request).await;
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body
);
let _ = socket.write_all(response.as_bytes()).await;
}
});
let rest = relay::RestClient {
http: reqwest::Client::new(),
base_url,
keys: nostr::Keys::generate(),
auth_tag_json: None,
};
(rest, server)
}
#[tokio::test]
async fn test_sibling_indeterminate_relay_failure_is_not_cached() {
let owner = nostr::Keys::generate().public_key().to_hex();
let author = nostr::Keys::generate().public_key().to_hex();
let cache = OwnerCache::new(Some(owner));
assert!(
!is_owner_or_sibling(&author, &cache, &dummy_rest_client()).await,
"an unverifiable author must fail closed for this event"
);
assert_eq!(
cache.proven_verdict(&author),
None,
"a relay failure must not be cached as a permanent non-sibling verdict"
);
}
#[tokio::test]
async fn test_sibling_absent_profile_is_indeterminate_and_throttled() {
let owner = nostr::Keys::generate().public_key().to_hex();
let author = nostr::Keys::generate().public_key().to_hex();
let cache = OwnerCache::new(Some(owner));
let (rest, server) = sibling_query_server(serde_json::json!([])).await;
assert!(!is_owner_or_sibling(&author, &cache, &rest).await);
assert_eq!(
cache.proven_verdict(&author),
None,
"a missing kind:0 is not a verdict — the profile may not exist or replicate yet"
);
assert_eq!(
cache.sibling_state(&author),
SiblingCacheState::Throttled,
"a missing profile is retried under the bounded backoff, not per event"
);
server.abort();
}
#[tokio::test]
async fn test_sibling_profile_without_matching_tag_is_cached_negative() {
let owner = nostr::Keys::generate().public_key().to_hex();
let author = nostr::Keys::generate().public_key().to_hex();
let cache = OwnerCache::new(Some(owner));
let (rest, server) = sibling_query_server(serde_json::json!([{ "tags": [] }])).await;
assert!(!is_owner_or_sibling(&author, &cache, &rest).await);
assert_eq!(
cache.proven_verdict(&author),
Some(false),
"a fetched profile with no verifying owner attestation is definitive"
);
server.abort();
}
#[tokio::test]
async fn test_sibling_valid_auth_tag_is_verified_and_cached() {
let owner_keys = nostr::Keys::generate();
let agent_keys = nostr::Keys::generate();
let author = agent_keys.public_key().to_hex();
let tag_json =
buzz_sdk::nip_oa::compute_auth_tag(&owner_keys, &agent_keys.public_key(), "")
.expect("compute auth tag");
let tag: serde_json::Value = serde_json::from_str(&tag_json).expect("tag json");
let profile = serde_json::json!([{ "tags": [tag] }]);
let cache = OwnerCache::new(Some(owner_keys.public_key().to_hex()));
let (rest, server) = sibling_query_server(profile).await;
assert!(
is_owner_or_sibling(&author, &cache, &rest).await,
"a kind:0 carrying a valid owner attestation must verify as a sibling"
);
assert_eq!(cache.proven_verdict(&author), Some(true));
server.abort();
}
/// A `/query` stand-in that counts requests and answers HTTP 500 (a
/// non-retriable status, so the harness sees an immediate error rather
/// than the 2s timeout) until `healthy` is flipped, after which it answers
/// `200 [{"tags":[]}]` — a fetched profile with no owner attestation, i.e.
/// a definitive `NotSibling`.
async fn flaky_sibling_query_server() -> (
relay::RestClient,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
std::sync::Arc<std::sync::atomic::AtomicBool>,
tokio::task::JoinHandle<()>,
) {
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind flaky sibling query server");
let base_url = format!("http://{}", listener.local_addr().unwrap());
let requests = std::sync::Arc::new(AtomicUsize::new(0));
let healthy = std::sync::Arc::new(AtomicBool::new(false));
let (server_requests, server_healthy) = (requests.clone(), healthy.clone());
let server = tokio::spawn(async move {
loop {
let Ok((mut socket, _)) = listener.accept().await else {
break;
};
let mut request = vec![0; 8192];
let _ = socket.read(&mut request).await;
server_requests.fetch_add(1, Ordering::SeqCst);
let response = if server_healthy.load(Ordering::SeqCst) {
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 13\r\nConnection: close\r\n\r\n[{\"tags\":[]}]".to_string()
} else {
"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string()
};
let _ = socket.write_all(response.as_bytes()).await;
}
});
let rest = relay::RestClient {
http: reqwest::Client::new(),
base_url,
keys: nostr::Keys::generate(),
auth_tag_json: None,
};
(rest, requests, healthy, server)
}
#[tokio::test]
async fn test_sibling_indeterminate_retry_is_throttled_then_recovers() {
use std::sync::atomic::Ordering;
let owner = nostr::Keys::generate().public_key().to_hex();
let author = nostr::Keys::generate().public_key().to_hex();
let cache = OwnerCache::with_retry_backoff(
Some(owner),
Duration::from_millis(80),
Duration::from_millis(400),
);
let (rest, requests, healthy, server) = flaky_sibling_query_server().await;
// First contact during the outage: one relay call, denied, not proven.
assert!(!is_owner_or_sibling(&author, &cache, &rest).await);
assert_eq!(requests.load(Ordering::SeqCst), 1);
assert_eq!(cache.proven_verdict(&author), None);
assert_eq!(cache.sibling_state(&author), SiblingCacheState::Throttled);
// A burst of further events inside the retry deadline must be denied
// WITHOUT another inline relay round-trip — this is what keeps a
// chatty sibling from stalling the inbound loop during an outage.
for _ in 0..5 {
assert!(!is_owner_or_sibling(&author, &cache, &rest).await);
}
assert_eq!(
requests.load(Ordering::SeqCst),
1,
"events inside the retry deadline must not re-query the relay"
);
// Once the deadline passes the lookup is retried (still failing).
tokio::time::sleep(Duration::from_millis(120)).await;
assert!(!is_owner_or_sibling(&author, &cache, &rest).await);
assert_eq!(
requests.load(Ordering::SeqCst),
2,
"deadline expiry re-arms the lookup"
);
assert_eq!(cache.sibling_state(&author), SiblingCacheState::Throttled);
// Relay recovers: after the (now doubled) deadline the retry lands a
// proven verdict, which is cached and ends the throttling entirely.
healthy.store(true, Ordering::SeqCst);
tokio::time::sleep(Duration::from_millis(220)).await;
assert!(!is_owner_or_sibling(&author, &cache, &rest).await);
assert_eq!(requests.load(Ordering::SeqCst), 3);
assert_eq!(
cache.proven_verdict(&author),
Some(false),
"a relay-confirmed verdict replaces the retry deadline"
);
assert!(!is_owner_or_sibling(&author, &cache, &rest).await);
assert_eq!(
requests.load(Ordering::SeqCst),
3,
"a proven verdict is served from cache with no further relay calls"
);
server.abort();
}
#[test]
fn test_sibling_indeterminate_backoff_doubles_to_cap() {
let cache = OwnerCache::with_retry_backoff(
Some("00".into()),
Duration::from_secs(5),
Duration::from_secs(12),
);
let observed: Vec<u64> = (0..5)
.map(|_| cache.note_indeterminate("author").as_secs())
.collect();
assert_eq!(
observed,
vec![5, 10, 12, 12, 12],
"consecutive indeterminate lookups back off exponentially up to the cap"
);
assert_eq!(cache.sibling_state("author"), SiblingCacheState::Throttled);
// A proven verdict clears the backoff state.
cache.cache_sibling("author".into(), true);
assert_eq!(
cache.sibling_state("author"),
SiblingCacheState::Proven(true)
);
assert_eq!(
cache.note_indeterminate("fresh").as_secs(),
5,
"backoff is tracked per author"
);
}
/// A cache whose retry deadlines expire almost immediately, filled to the
/// eviction threshold with proven verdicts for `authors`.
fn saturated_cache_with_proven(authors: &[String]) -> OwnerCache {
let cache = OwnerCache::with_retry_backoff(
Some("00".into()),
Duration::from_millis(1),
Duration::from_millis(1),
);
for author in authors {
cache.cache_sibling(author.clone(), true);
}
cache
}
#[test]
fn test_sibling_cache_replacing_an_author_at_capacity_keeps_the_rest() {
let authors: Vec<String> = (0..OwnerCache::SIBLING_CACHE_CAP)
.map(|i| format!("author-{i}"))
.collect();
let cache = saturated_cache_with_proven(&authors);
assert_eq!(cache.len(), OwnerCache::SIBLING_CACHE_CAP);
// Re-arming one author's retry deadline at capacity — first arming,
// then re-arming after it expired — replaces that entry only.
cache.note_indeterminate(&authors[0]);
std::thread::sleep(Duration::from_millis(5));
cache.note_indeterminate(&authors[0]);
assert_eq!(
cache.len(),
OwnerCache::SIBLING_CACHE_CAP,
"replacing an existing author must not evict anything"
);
for author in &authors[1..] {
assert_eq!(
cache.sibling_state(author),
SiblingCacheState::Proven(true),
"re-arming {} at capacity must keep {author}'s proven verdict",
authors[0]
);
}
}
#[test]
fn test_sibling_cache_new_author_at_capacity_evicts_expired_deadlines_first() {
let proven: Vec<String> = (1..OwnerCache::SIBLING_CACHE_CAP)
.map(|i| format!("proven-{i}"))
.collect();
let cache = saturated_cache_with_proven(&proven);
cache.note_indeterminate("expired");
assert_eq!(cache.len(), OwnerCache::SIBLING_CACHE_CAP);
std::thread::sleep(Duration::from_millis(5));
// A new author at capacity displaces the one expired deadline, nothing else.
cache.cache_sibling("newcomer".into(), false);
assert_eq!(cache.len(), OwnerCache::SIBLING_CACHE_CAP);
assert_eq!(
cache.sibling_state("newcomer"),
SiblingCacheState::Proven(false)
);
for author in &proven {
assert_eq!(
cache.sibling_state(author),
SiblingCacheState::Proven(true),
"an expired deadline must be evicted before any proven verdict"
);
}
// With no expired deadlines left, one more new author evicts exactly
// one entry — never the whole map.
cache.cache_sibling("newcomer-2".into(), false);
assert_eq!(cache.len(), OwnerCache::SIBLING_CACHE_CAP);
assert_eq!(
cache.sibling_state("newcomer-2"),
SiblingCacheState::Proven(false)
);
let retained = proven
.iter()
.chain(std::iter::once(&"newcomer".to_string()))
.filter(|author| cache.sibling_state(author) != SiblingCacheState::Unknown)
.count();
assert_eq!(
retained,
OwnerCache::SIBLING_CACHE_CAP - 1,
"a full cache evicts a single entry for a new author"
);
}
#[test]
fn test_sibling_cache_evicts_one_expired_record_and_keeps_other_backoff_tiers() {
let cache = OwnerCache::with_retry_backoff(
Some("00".into()),
Duration::from_millis(1),
Duration::from_millis(64),
);
// "old" expires first at the base tier; "tiered" has failed three
// times and climbed to a 4ms deadline.
assert_eq!(cache.note_indeterminate("old").as_millis(), 1);
for _ in 0..3 {
cache.note_indeterminate("tiered");
}
let proven: Vec<String> = (2..OwnerCache::SIBLING_CACHE_CAP)
.map(|i| format!("proven-{i}"))
.collect();
for author in &proven {
cache.cache_sibling(author.clone(), true);
}
assert_eq!(cache.len(), OwnerCache::SIBLING_CACHE_CAP);
std::thread::sleep(Duration::from_millis(10)); // both deadlines expired
// The newcomer takes exactly one slot: the longest-expired record.
cache.cache_sibling("newcomer".into(), false);
assert_eq!(cache.len(), OwnerCache::SIBLING_CACHE_CAP);
for author in &proven {
assert_eq!(cache.sibling_state(author), SiblingCacheState::Proven(true));
}
assert_eq!(
cache.note_indeterminate("tiered").as_millis(),
8,
"the surviving expired record must keep its backoff tier (3 failures → 8ms next)"
);
assert_eq!(
cache.note_indeterminate("old").as_millis(),
1,
"only the longest-expired record was evicted, so it restarts at the base tier"
);
}
// ── is_dm_channel resolution ──────────────────────────────────────────
fn resolver(startup: HashMap<Uuid, relay::ChannelInfo>) -> pool::ChannelInfoResolver {

7
crates/buzz-acp/src/relay.rs
View File

@@ -433,6 +433,13 @@ impl RestClient {
)));
}
Ok(resp) => {
// Surface non-retriable statuses (401/403/500/…): callers
// often swallow the Err, and a silent auth failure here has
// previously masqueraded as "the agent just went quiet".
tracing::warn!(
"{method} {path} returned non-retriable HTTP {}",
resp.status()
);
return Err(RelayError::Http(format!(
"{method} {} returned HTTP {}",
path,