mirror of
https://github.com/block/buzz.git
synced 2026-11-02 23:41:08 +00:00
test(buzz-acp): cover dispatch_pending leader gate, harden mutex, document fail-safe scope
The primary suppression surface — dispatch_pending — had no test; only the heartbeat gate was covered, so a refactor of the shared leader gate could silently ship a non-leader that still dispatches. Add the missing leader/non-leader test pair. Read-side mutex locks now read through poison rather than panic the event loop, removing a latent footgun for when Phase 2 extends the critical sections. Module-doc and NIP-LE now state that any IO error (not just an absent lock) fails safe to leader, that this can produce a bounded ≤5s transient duplicate during a concurrent rewrite, and that pid/claimed_at are read-side-ignored. Co-authored-by: Will Pfleger <wpfleger@block.xyz> Signed-off-by: Will Pfleger <wpfleger@block.xyz>
This commit is contained in:
committed by
npub1fgdl5qqnh3k3f2xkqrvt7cujalhm623x4s7fdjdj5yrtp5fzjl9qrjpucw
parent
6685497bc8
commit
305cda5be1
@@ -15,8 +15,9 @@
|
||||
//!
|
||||
//! - Lock dir: `~/.buzz/leader-locks/`, one file per agent pubkey:
|
||||
//! `<pubkey-hex>.lock`, JSON `{"instance_id","pid","claimed_at"}`.
|
||||
//! - **Absent** lock file → this process is leader. Single-instance dev is
|
||||
//! thereby unaffected: no lock, no suppression.
|
||||
//! - **Absent or unreadable** lock file → this process is leader. Single-instance
|
||||
//! dev is thereby unaffected: no lock, no suppression. Any IO error reading the
|
||||
//! lock (permission, mid-write truncation) fails safe to leader for the same reason.
|
||||
//! - **Present** → leader iff the lock's `instance_id` equals this process's
|
||||
//! own election id ([`ELECTION_ID_ENV`]).
|
||||
//! - **No instance id** (env unset, e.g. solo CLI use or pre-Phase-2 where
|
||||
@@ -119,13 +120,18 @@ impl FileLeaderCheck {
|
||||
|
||||
impl LeaderCheck for FileLeaderCheck {
|
||||
fn is_leader(&self, agent_pubkey_hex: &str) -> bool {
|
||||
if let Some(&cached) = self.cache.lock().unwrap().get(agent_pubkey_hex) {
|
||||
if let Some(&cached) = self
|
||||
.cache
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.get(agent_pubkey_hex)
|
||||
{
|
||||
return cached;
|
||||
}
|
||||
let status = self.read_status(agent_pubkey_hex);
|
||||
self.cache
|
||||
.lock()
|
||||
.unwrap()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.insert(agent_pubkey_hex.to_string(), status);
|
||||
status
|
||||
}
|
||||
@@ -133,10 +139,19 @@ impl LeaderCheck for FileLeaderCheck {
|
||||
fn refresh(&self) {
|
||||
// Re-read only keys we've already been asked about — those are the
|
||||
// agent pubkeys this process actually dispatches for.
|
||||
let keys: Vec<String> = self.cache.lock().unwrap().keys().cloned().collect();
|
||||
let keys: Vec<String> = self
|
||||
.cache
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.keys()
|
||||
.cloned()
|
||||
.collect();
|
||||
for key in keys {
|
||||
let status = self.read_status(&key);
|
||||
self.cache.lock().unwrap().insert(key, status);
|
||||
self.cache
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.insert(key, status);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3684,6 +3684,57 @@ mod error_outcome_emission_tests {
|
||||
assert!(!pool.any_idle(), "leader must claim the idle agent");
|
||||
assert!(heartbeat_in_flight, "leader heartbeat must fire");
|
||||
}
|
||||
|
||||
/// Seed `queue` with one pending event so `dispatch_pending` has work to
|
||||
/// promote when the leader gate allows it.
|
||||
fn queue_one_pending(queue: &mut EventQueue) {
|
||||
use nostr::{EventBuilder, Keys, Kind};
|
||||
let event = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "@agent hello")
|
||||
.sign_with_keys(&Keys::generate())
|
||||
.unwrap();
|
||||
queue.push(QueuedEvent {
|
||||
channel_id: Uuid::new_v4(),
|
||||
event,
|
||||
received_at: std::time::Instant::now(),
|
||||
prompt_tag: "@mention".into(),
|
||||
});
|
||||
}
|
||||
|
||||
/// Pins the primary suppression surface: a non-leader must never promote a
|
||||
/// queued event to a prompt, even with an idle agent and pending work.
|
||||
/// Without this gate, N instances sharing one key would each dispatch the
|
||||
/// same mention — the duplicate-actor bug leader election exists to prevent.
|
||||
#[tokio::test]
|
||||
async fn test_non_leader_dispatch_pending_promotes_nothing() {
|
||||
let ctx = prompt_context_with_leader(nostr::Keys::generate(), false);
|
||||
let mut pool = AgentPool::from_slots(vec![Some(dummy_agent(0).await)]);
|
||||
let mut queue = EventQueue::new(config::DedupMode::Queue);
|
||||
queue_one_pending(&mut queue);
|
||||
|
||||
let dispatched = dispatch_pending(&mut pool, &mut queue, &ctx);
|
||||
|
||||
assert!(
|
||||
dispatched.is_empty(),
|
||||
"non-leader must not promote queued events"
|
||||
);
|
||||
assert!(pool.any_idle(), "non-leader must not claim the idle agent");
|
||||
}
|
||||
|
||||
/// Companion to the gate test: the same idle agent + pending event, but as
|
||||
/// the leader, DOES promote — proving the gate test catches the gate, not a
|
||||
/// broken dispatch path.
|
||||
#[tokio::test]
|
||||
async fn test_leader_dispatch_pending_promotes_queued_event() {
|
||||
let ctx = prompt_context_with_leader(nostr::Keys::generate(), true);
|
||||
let mut pool = AgentPool::from_slots(vec![Some(dummy_agent(0).await)]);
|
||||
let mut queue = EventQueue::new(config::DedupMode::Queue);
|
||||
queue_one_pending(&mut queue);
|
||||
|
||||
let dispatched = dispatch_pending(&mut pool, &mut queue, &ctx);
|
||||
|
||||
assert_eq!(dispatched.len(), 1, "leader must promote the queued event");
|
||||
assert!(!pool.any_idle(), "leader must claim the idle agent");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -85,9 +85,24 @@ The lock file contains a JSON object with the following shape
|
||||
- `pid`: the operating-system process id of the leading window.
|
||||
- `claimed_at`: when the current leader claimed the lock.
|
||||
|
||||
Only `instance_id` is significant to the read side; `pid` and `claimed_at` are
|
||||
informational, written by the Phase-2 acquire path and ignored when deciding
|
||||
leadership.
|
||||
|
||||
An instance is the leader for an agent identity iff a lock file exists and its
|
||||
`instance_id` equals that instance's own election identity. A malformed or
|
||||
unparseable lock file MUST fail safe to leader, preserving solo-dev behavior.
|
||||
An absent or unreadable lock file (permission error, or a partial read while
|
||||
another instance is mid-rewrite) likewise fails safe to leader. Because the read
|
||||
side is unguarded (acquire/steal `flock` is Phase 2), a concurrent rewrite read as
|
||||
malformed can briefly produce a duplicate responder; this self-heals on the next
|
||||
refresh (≤5s, see below) and is the intended trade — a transient duplicate beats
|
||||
silencing the only responder.
|
||||
|
||||
Leadership changes (claim, release, failover) take effect within a bounded refresh
|
||||
interval of 5 seconds: each instance re-reads its leader lock(s) every 5s, so a
|
||||
stale-cache window after a leadership change is capped at that interval. Phase 2
|
||||
MUST NOT regress this bound.
|
||||
|
||||
Acquire and steal MUST be `flock`-guarded to close the read-check-write TOCTOU
|
||||
window between two instances racing to claim or steal the same lock.
|
||||
|
||||
Reference in New Issue
Block a user