diff --git a/.gitignore b/.gitignore index f26e74136c0..7f50ba114ac 100644 --- a/.gitignore +++ b/.gitignore @@ -72,3 +72,12 @@ identity.key # Helm dependency tarballs — regenerable from Chart.lock via `helm dependency build` deploy/charts/*/charts/*.tgz + +# Windows test artifacts. The capture writers in crates/buzz-acp build a temp +# path that MSYS translates into a literal FILENAME in the crate root, so +# `cargo test -p buzz-acp` leaves ~25 empty files whose names are mangled +# absolute paths. Matched on the writer-defined stem rather than the path, +# since the mangling is machine-specific. Verified against upstream/main: no +# tracked file matches either pattern. +*buzz-acp-steer-capture* +*buzz-acp-*.ndjson diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 25c6e549052..d1587744118 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2265,6 +2265,14 @@ async fn tokio_main() -> Result<()> { let mut typing_channels: HashMap = HashMap::new(); let mut presence_task: Option> = None; + // (channel, author) pairs already told they are outside the author gate. + // One notice per person per channel: silence strands the sender, but a + // reply per refused event lets a noisy author turn the agent into a + // flooder. Bounded because this process is long-lived and the key space is + // attacker-controlled — on overflow we clear and allow one more notice per + // pair rather than growing without limit. + let mut gate_notified: HashSet<(Uuid, String)> = HashSet::new(); + // Independent of pool readiness: a never-mentioned lazy agent must still // self-terminate. The watch interval is capped so small configured bounds // remain reasonably precise without waking long-lived agents frequently. @@ -2884,13 +2892,53 @@ async fn tokio_main() -> Result<()> { ) .await; if !allowed { - tracing::debug!( + // warn, not debug: a refused request is a + // person who got no answer. At debug level + // this is invisible in practice, so the + // sender is stranded and the owner never + // learns anyone tried. + tracing::warn!( channel_id = %buzz_event.channel_id, - author = %buzz_event.event.pubkey.to_hex(), + author = %author, mode = %config.respond_to, is_dm, - "inbound author gate — dropping event" + "inbound author gate — refusing event" ); + + if remember_gate_notice( + &mut gate_notified, + (buzz_event.channel_id, author.clone()), + ) { + // Mention the owner in channels so they + // learn through the normal mention + // path, in the project where the + // request arrived. Never in a DM: that + // would disclose the owner's pubkey to + // a stranger. + let mentions: Vec = if is_dm { + Vec::new() + } else { + owner_cache + .get() + .map(|o| vec![o.to_string()]) + .unwrap_or_default() + }; + let content = author_gate_notice_text(is_dm); + let thread_tags = + queue::parse_thread_tags(&buzz_event.event); + let rest = ctx.rest_client.clone(); + let channel_id = buzz_event.channel_id; + tokio::spawn(async move { + pool::post_notice( + &rest, + channel_id, + &thread_tags, + &content, + &mentions, + ) + .await; + }); + } continue; } } @@ -3138,6 +3186,19 @@ async fn tokio_main() -> Result<()> { None } _ = shutdown_rx.changed() => { + // A run that discarded work must say so before it exits. + // Otherwise the only trace is scattered mid-run lines that + // nobody was watching, and the loss becomes unanswerable + // after the fact. + let drops = queue.drop_counts(); + if drops.total() > 0 { + tracing::warn!( + in_flight_dedup = drops.in_flight_dedup, + queue_depth_cap = drops.queue_depth_cap, + total = drops.total(), + "shutting down — events were discarded before delivery during this run" + ); + } tracing::info!("shutting down"); break; } @@ -3854,10 +3915,103 @@ fn is_auth_error(error: &acp::AcpError) -> bool { /// /// Shared by the hard-cap immediate dead-letter path and the retries-exhausted /// dead-letter path so neither duplicates the tokio::spawn block. +/// Upper bound on remembered (channel, author) gate refusals. +/// +/// The key space is attacker-controlled — anyone can mint a pubkey and post — +/// so this cannot grow unbounded in a long-lived process. At the cap the memo +/// is cleared rather than evicted per-entry: the cost of forgetting is one +/// extra notice to a person who has already been told, which is far cheaper +/// than the bookkeeping to avoid it. +const MAX_GATE_NOTIFIED: usize = 4_096; + +/// Record that `key` has been told it sits outside the author gate, and report +/// whether this is the first time (so the caller should post a notice). +/// +/// Clearing at the cap, rather than evicting one entry, is deliberate: the cost +/// of forgetting is one repeat notice to someone already told, which is far +/// cheaper than the bookkeeping an LRU would need on a path that exists only to +/// suppress duplicates. +fn remember_gate_notice(seen: &mut HashSet<(Uuid, String)>, key: (Uuid, String)) -> bool { + if seen.contains(&key) { + return false; + } + if seen.len() >= MAX_GATE_NOTIFIED { + tracing::warn!( + tracked = seen.len(), + "author gate notice memo full — clearing" + ); + seen.clear(); + } + seen.insert(key); + true +} + +/// What someone outside the author gate is told. +/// +/// Says what happened, what it means for them, and what happens next — without +/// naming the mechanism. "I'm not set up to take requests from you" is +/// actionable; "inbound author gate rejected event" is not. It never asks them +/// to resend: the message is already in the channel, and re-typing it will not +/// change the outcome. +fn author_gate_notice_text(is_dm: bool) -> String { + if is_dm { + // No project channel to point at, and no owner mention — naming the + // owner here would hand a stranger their pubkey. + "I'm not currently set up to take requests from you, so I can't act on this. \ + Your message hasn't gone anywhere and you don't need to send it again. \ + If you're expecting access, ask whoever runs this workspace to add you." + .to_string() + } else { + "I saw your message but I'm not currently set up to take requests from you, \ + so I can't start on it. Your message is safe in this channel and nothing \ + needs re-typing. I've flagged this for the owner of this project — once \ + they add you, send it again and I'll pick it up." + .to_string() + } +} + +/// Mention list for a notice that someone other than the sender must act on. +/// +/// A failure the requester cannot fix has to reach the person who can — a +/// notice nobody is pinged about is only marginally better than no notice. +/// Empty when no owner is known, which degrades to the previous behaviour +/// rather than failing. +fn owner_mentions(owner: Option<&str>) -> Vec { + owner + .filter(|o| !o.is_empty()) + .map(|o| vec![o.to_string()]) + .unwrap_or_default() +} + +/// Plain-language account of why a batch could not be completed. +/// +/// Names the state rather than the mechanism, and never surfaces a raw error +/// string. Internals belong in the log — the person waiting cannot act on a +/// `Display` impl, and a raw error can carry paths or tokens that should not be +/// posted into a channel. +fn failure_reason_text(outcome: &PromptOutcome) -> String { + match outcome { + PromptOutcome::Timeout(TimeoutKind::Idle) => "it stopped making progress".to_string(), + PromptOutcome::Timeout(TimeoutKind::Hard { .. }) => { + "it ran for a long time without getting closer".to_string() + } + PromptOutcome::AgentExited => "the tool I use to do the work stopped unexpectedly".to_string(), + // Deliberately vague to the channel; the detail is in the log. This is + // the one branch that used to interpolate the raw error. + PromptOutcome::Error(_) => "something went wrong on my side".to_string(), + PromptOutcome::ProjectContextIndeterminate(_) => { + "I couldn't tell which project this belongs to, and I'd rather ask than change the wrong one" + .to_string() + } + _ => "it kept failing".to_string(), + } +} + fn spawn_failure_notice( rest_client: Option<&relay::RestClient>, batch: &FlushBatch, content: String, + mentions: Vec, ) { if let Some(rest) = rest_client { let thread_tags = batch @@ -3868,7 +4022,7 @@ fn spawn_failure_notice( let rest = rest.clone(); let channel_id = batch.channel_id; tokio::spawn(async move { - pool::post_failure_notice(&rest, channel_id, &thread_tags, &content).await; + pool::post_notice(&rest, channel_id, &thread_tags, &content, &mentions).await; }); } } @@ -3964,11 +4118,16 @@ fn handle_prompt_result( "dead-lettering batch after hard-cap timeout (no recent activity) — discarding {} events", batch.events.len(), ); - let content = format!( - "⚠️ I couldn't process the last request (the turn exceeded the maximum duration ({}s)). Please re-send if it's still needed.", - config.max_turn_duration_secs + let content = "⚠️ I couldn't finish this one. I worked on it for a long \ + stretch without getting closer, so I stopped rather than keep going — \ + nothing was changed. Send it again if you'd like me to try once more." + .to_string(); + spawn_failure_notice( + rest_client, + &batch, + content, + owner_mentions(config.agent_owner.as_deref()), ); - spawn_failure_notice(rest_client, &batch, content); hard_timeout_fate_suffix = Some(" — dead-lettered (no recent activity)"); } else if matches!( result.outcome, @@ -3982,11 +4141,17 @@ fn handle_prompt_result( "hard-cap timeout with recent activity — requeueing for retry" ); if let Some(dead) = queue.requeue(batch) { - let content = format!( - "⚠️ I couldn't process the last request after multiple retries (the turn exceeded the maximum duration ({}s)). Please re-send if it's still needed.", - config.max_turn_duration_secs + let content = "⚠️ I tried this several times and couldn't get through it, \ + so I've stopped rather than keep failing quietly. Nothing was changed. \ + Sending it again is unlikely to help on its own — this one needs a person \ + to look at it." + .to_string(); + spawn_failure_notice( + rest_client, + &dead, + content, + owner_mentions(config.agent_owner.as_deref()), ); - spawn_failure_notice(rest_client, &dead, content); hard_timeout_fate_suffix = Some(" — dead-lettered (retry budget exhausted)"); } else { hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)"); @@ -4001,26 +4166,32 @@ fn handle_prompt_result( events = batch.events.len(), "dead-lettering batch immediately — non-retryable auth error" ); - let content = "⚠️ I couldn't process the last request: authentication failed. \ - Please re-authenticate the CLI (e.g. run `claude /login` or `codex login`) \ - and then re-send." + let content = "⚠️ I've lost my ability to sign in to the tool I use to do this \ + work, so I can't act on this — or anything else — until that's restored. \ + Nothing was changed. This isn't something you can fix from here: it needs \ + whoever runs this workspace, and I've flagged them on this message." .to_string(); - spawn_failure_notice(rest_client, &batch, content); + spawn_failure_notice( + rest_client, + &batch, + content, + owner_mentions(config.agent_owner.as_deref()), + ); } else if let Some(dead) = queue.requeue(batch) { - let reason = match &result.outcome { - PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), - PromptOutcome::Timeout(TimeoutKind::Hard { .. }) => { - "the turn exceeded the maximum duration".to_string() - } - PromptOutcome::AgentExited => "the agent process exited".to_string(), - PromptOutcome::Error(e) => format!("{e}"), - PromptOutcome::ProjectContextIndeterminate(reason) => reason.clone(), - _ => "repeated failures".to_string(), - }; + // The raw error stays in the log above; what reaches the + // channel names the state, not the mechanism. + let reason = failure_reason_text(&result.outcome); let content = format!( - "⚠️ I couldn't process the last request after multiple retries ({reason}). Please re-send if it's still needed." + "⚠️ I tried this several times and couldn't get through it — {reason}. \ + I've stopped rather than keep failing quietly, and nothing was changed. \ + This one needs a person to look at it." + ); + spawn_failure_notice( + rest_client, + &dead, + content, + owner_mentions(config.agent_owner.as_deref()), ); - spawn_failure_notice(rest_client, &dead, content); } } else { tracing::debug!( @@ -5508,6 +5679,166 @@ mod author_gate_tests { } } + // ── Author gate notice ──────────────────────────────────────────────── + // + // A refused author used to be dropped at debug level with no reply, so the + // sender was stranded and the owner never learned anyone had tried. These + // lock down the properties that make the refusal visible instead. + + #[test] + fn test_gate_notice_says_what_happened_and_what_next() { + for is_dm in [false, true] { + let text = author_gate_notice_text(is_dm); + assert!( + text.contains("can't"), + "notice must state that the request will not be acted on (is_dm={is_dm})" + ); + assert!( + text.contains("send it again") || text.contains("add you"), + "notice must tell the sender what happens next (is_dm={is_dm})" + ); + assert!( + !text.contains("author gate") + && !text.contains("respond_to") + && !text.contains("pubkey"), + "notice must name the state, not the mechanism (is_dm={is_dm})" + ); + } + } + + #[test] + fn test_gate_notice_never_asks_for_a_resend_of_the_same_message() { + // Re-sending cannot change the outcome — asking for it is the pattern + // that trains people to repeat themselves into silence. + let channel = author_gate_notice_text(false); + assert!( + channel.contains("nothing") && channel.contains("re-typing"), + "channel notice must reassure that the message is not lost: {channel}" + ); + let dm = author_gate_notice_text(true); + assert!( + dm.contains("don't need to send it again"), + "DM notice must reassure that the message is not lost: {dm}" + ); + } + + #[test] + fn test_gate_notice_does_not_leak_owner_in_dm() { + // In a channel the owner is p-tagged so they learn someone tried. In a + // DM there is no such mention, and the text must not compensate by + // naming them — the sender is an unknown party. + let dm = author_gate_notice_text(true); + assert!( + !dm.contains("owner of this project"), + "DM notice must not point an unknown sender at the owner: {dm}" + ); + let channel = author_gate_notice_text(false); + assert!( + channel.contains("owner"), + "channel notice should say the owner was flagged: {channel}" + ); + } + + #[test] + fn test_gate_notice_fires_once_per_person_per_channel() { + let mut seen = HashSet::new(); + let chan_a = Uuid::from_u128(1); + let chan_b = Uuid::from_u128(2); + + assert!( + remember_gate_notice(&mut seen, (chan_a, "alice".into())), + "first refusal must notify" + ); + assert!( + !remember_gate_notice(&mut seen, (chan_a, "alice".into())), + "a repeat refusal must stay quiet — otherwise a noisy author turns the agent into a flooder" + ); + assert!( + remember_gate_notice(&mut seen, (chan_b, "alice".into())), + "the same person in a different channel has not been told there" + ); + assert!( + remember_gate_notice(&mut seen, (chan_a, "bob".into())), + "a different person in the same channel has not been told" + ); + } + + #[test] + fn test_gate_notice_memo_stays_bounded_under_attacker_supplied_keys() { + // Anyone can mint a pubkey and post, so the key space is not ours to + // trust. The memo must not grow without limit in a long-lived process. + let mut seen = HashSet::new(); + let chan = Uuid::from_u128(7); + for i in 0..(MAX_GATE_NOTIFIED * 2 + 5) { + remember_gate_notice(&mut seen, (chan, format!("author-{i}"))); + assert!( + seen.len() <= MAX_GATE_NOTIFIED, + "memo exceeded its bound at iteration {i}: {}", + seen.len() + ); + } + } + + // ── Failure notice wording ──────────────────────────────────────────── + // + // These notices used to name the mechanism ("the turn exceeded the maximum + // duration (7200s)"), interpolate a raw error, and tell whoever was waiting + // to run `claude /login` — an instruction a collaborator cannot act on. + + #[test] + fn test_failure_reason_never_surfaces_a_raw_error() { + // A Display impl can carry paths, tokens, or stack detail. It belongs + // in the log, not in a channel anyone with access can read. + let leaky = PromptOutcome::Error(acp::AcpError::Protocol( + "connect ECONNREFUSED 10.0.0.4:8080 /home/x/.ssh".to_string(), + )); + let text = failure_reason_text(&leaky); + assert!( + !text.contains("ECONNREFUSED") && !text.contains(".ssh") && !text.contains("10.0.0.4"), + "raw error leaked into a user-facing notice: {text}" + ); + } + + #[test] + fn test_failure_reasons_name_the_state_not_the_mechanism() { + let cases = [ + PromptOutcome::Timeout(TimeoutKind::Idle), + PromptOutcome::Timeout(TimeoutKind::Hard { + recently_active: false, + }), + PromptOutcome::AgentExited, + PromptOutcome::Error(acp::AcpError::Protocol("boom".to_string())), + ]; + for outcome in cases { + let text = failure_reason_text(&outcome); + assert!( + !text.is_empty(), + "every outcome needs a plain-language reason" + ); + for jargon in ["turn", "batch", "dead-letter", "PromptOutcome", "requeue"] { + assert!( + !text.to_lowercase().contains(jargon), + "reason leaks internal vocabulary {jargon:?}: {text}" + ); + } + } + } + + #[test] + fn test_owner_is_mentioned_so_a_failure_reaches_someone_who_can_act() { + assert_eq!( + owner_mentions(Some("npub_owner")), + vec!["npub_owner".to_string()], + "a failure the requester cannot fix must ping whoever can" + ); + // Degrade to previous behaviour rather than failing when unknown. + assert!(owner_mentions(None).is_empty()); + assert!( + owner_mentions(Some("")).is_empty(), + "an empty owner must not produce an empty p-tag" + ); + } + // ── DM hardening ────────────────────────────────────────────────────── // // In a DM, clients auto-p-tag every participant, and an agent can be diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index f18f7d6fea2..29f4c814b0b 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -4752,15 +4752,27 @@ pub(crate) async fn reaction_add(rest: &crate::relay::RestClient, event_id: &str } } -/// Best-effort: post a visible failure notice (kind:9) to a channel after a -/// batch is dead-lettered. Replies into the thread of `thread_tags` when the -/// triggering event was threaded. Errors are logged and swallowed — the -/// notice must never take down the main loop. -pub(crate) async fn post_failure_notice( +/// Post a notice from the agent into a channel, optionally p-tagging people who +/// need to know. +/// +/// Replies into the thread of `thread_tags` when the triggering event was +/// threaded, so the notice lands beside the request it concerns. Errors are +/// logged and swallowed — a notice must never take down the main loop. +/// +/// `mentions` is how a notice reaches the person who can act on it: the owner +/// learns through the normal mention path rather than by watching a channel. +/// Notices stay in the channel the work arrived in — keeping the exchange with +/// its project preserves the context instead of stranding it in a private +/// inbox, and the agent never opens a DM of its own. +/// +/// Best-effort by design: the caller must not be blocked waiting on the relay, +/// and a notice that cannot be delivered must not take the turn down with it. +pub(crate) async fn post_notice( rest: &crate::relay::RestClient, channel_id: Uuid, thread_tags: &ThreadTags, content: &str, + mentions: &[String], ) { let thread_ref = thread_tags.root_event_id.as_deref().and_then(|root| { let root_id = nostr::EventId::from_hex(root).ok()?; @@ -4774,25 +4786,32 @@ pub(crate) async fn post_failure_notice( parent_event_id: parent_id, }) }); - let builder = - match buzz_sdk::build_message(channel_id, content, thread_ref.as_ref(), &[], false, &[]) { - Ok(b) => b, - Err(e) => { - tracing::warn!(channel = %channel_id, "failure notice: build failed: {e}"); - return; - } - }; + let mention_refs: Vec<&str> = mentions.iter().map(String::as_str).collect(); + let builder = match buzz_sdk::build_message( + channel_id, + content, + thread_ref.as_ref(), + &mention_refs, + false, + &[], + ) { + Ok(b) => b, + Err(e) => { + tracing::warn!(channel = %channel_id, "notice: build failed: {e}"); + return; + } + }; let event = match builder.sign_with_keys(&rest.keys) { Ok(e) => e, Err(e) => { - tracing::warn!(channel = %channel_id, "failure notice: sign failed: {e}"); + tracing::warn!(channel = %channel_id, "notice: sign failed: {e}"); return; } }; match tokio::time::timeout(Duration::from_secs(5), rest.submit_event(&event)).await { Ok(Ok(_)) => {} - Ok(Err(e)) => tracing::warn!(channel = %channel_id, "failure notice failed: {e}"), - Err(_) => tracing::warn!(channel = %channel_id, "failure notice timed out"), + Ok(Err(e)) => tracing::warn!(channel = %channel_id, "notice failed: {e}"), + Err(_) => tracing::warn!(channel = %channel_id, "notice timed out"), } } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index d62b99114cf..2e0e7854e69 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -136,9 +136,35 @@ pub struct FlushBatch { /// if retry_counts[channel] > MAX_RETRIES: dead-letter (log ERROR, return batch to caller) /// else: push_front with original received_at, set exponential backoff retry_after with jitter /// ``` +/// Events discarded before ever reaching an agent, by cause. +/// +/// Both paths lose work somebody is waiting on, so they are counted and not +/// merely logged. A running total is what makes the loss detectable: a +/// watchdog can alert on any non-zero delta without a metrics pipeline, and +/// "this counter moved" is a question anyone can answer from the logs. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct DropCounts { + /// Discarded because that channel already had a turn in flight, under + /// [`DedupMode::Drop`]. + pub in_flight_dedup: u64, + /// Discarded because the channel queue was at its depth cap. The *oldest* + /// event is evicted, so the request that has waited longest is the one + /// lost — the reverse of what someone waiting would expect. + pub queue_depth_cap: u64, +} + +impl DropCounts { + /// Total events lost across every cause. + pub fn total(&self) -> u64 { + self.in_flight_dedup + self.queue_depth_cap + } +} + pub struct EventQueue { queues: HashMap>, in_flight_channels: HashSet, + /// Running totals of work discarded before delivery. See [`DropCounts`]. + drops: DropCounts, /// Per-channel deadline for auto-expiring stuck in-flight entries. in_flight_deadlines: HashMap, /// Number of events in each in-flight batch (for expiry logging). @@ -182,6 +208,7 @@ impl EventQueue { Self { queues: HashMap::new(), in_flight_channels: HashSet::new(), + drops: DropCounts::default(), in_flight_deadlines: HashMap::new(), in_flight_batch_sizes: HashMap::new(), retry_after: HashMap::new(), @@ -233,19 +260,27 @@ impl EventQueue { if matches!(self.dedup_mode, DedupMode::Drop) && self.in_flight_channels.contains(&event.channel_id) { - tracing::debug!( + self.drops.in_flight_dedup += 1; + // warn, not debug: this discards a request someone sent. At debug + // level the loss is invisible in practice, which is how work goes + // missing without anyone being able to say so afterwards. + tracing::warn!( channel_id = %event.channel_id, + dropped_total = self.drops.in_flight_dedup, "dropping event for in-flight channel (drop mode)" ); return false; } + let drops = &mut self.drops; let queue = self.queues.entry(event.channel_id).or_default(); // Enforce per-channel depth cap: drop oldest to make room. if queue.len() >= MAX_PENDING_PER_CHANNEL { queue.pop_front(); + drops.queue_depth_cap += 1; tracing::warn!( channel_id = %event.channel_id, limit = MAX_PENDING_PER_CHANNEL, + dropped_total = drops.queue_depth_cap, "queue depth cap reached — dropped oldest event" ); } @@ -626,6 +661,15 @@ impl EventQueue { } /// Number of channels with pending events. + /// Running totals of work discarded before delivery. + /// + /// Exposed so a supervisor outside this queue can observe the loss. A + /// counter only the queue can see is not observability — the process that + /// drops the work is the least reliable place to report it from. + pub fn drop_counts(&self) -> DropCounts { + self.drops + } + pub fn pending_channels(&self) -> usize { self.queues.len() } @@ -2005,6 +2049,78 @@ mod tests { use std::time::Duration; /// Build a test event with the given content and kind. + // ── Discarded-work accounting ───────────────────────────────────── + // + // Both drop paths lose a request someone sent. They were previously + // observable only as scattered log lines — one of them at debug level — + // so work could go missing with no way to establish afterwards that it + // had. These lock the counting down. + + #[test] + fn test_dedup_drop_is_counted_not_just_logged() { + let mut q = EventQueue::new(DedupMode::Drop); + let chan = Uuid::new_v4(); + // flush_next() is what puts a channel in flight. + assert!(q.push(make_queued(chan, "first"))); + let _in_flight = q.flush_next().expect("first push is flushable"); + + assert_eq!(q.drop_counts().total(), 0, "nothing lost yet"); + assert!( + !q.push(make_queued(chan, "dropped one")), + "drop mode discards" + ); + assert!( + !q.push(make_queued(chan, "dropped two")), + "drop mode discards" + ); + + let drops = q.drop_counts(); + assert_eq!( + drops.in_flight_dedup, 2, + "every discarded event must be counted" + ); + assert_eq!(drops.queue_depth_cap, 0, "wrong cause attributed"); + assert_eq!(drops.total(), 2); + } + + #[test] + fn test_queue_mode_does_not_count_drops_because_it_does_not_drop() { + // Queue mode is the CLI default. It must not inflate the loss counter, + // or a real drop becomes indistinguishable from normal operation. + let mut q = EventQueue::new(DedupMode::Queue); + let chan = Uuid::new_v4(); + assert!(q.push(make_queued(chan, "first"))); + let _in_flight = q.flush_next().expect("first push is flushable"); + + assert!(q.push(make_queued(chan, "queued behind in-flight"))); + assert_eq!(q.drop_counts().total(), 0, "queue mode loses nothing"); + } + + #[test] + fn test_depth_cap_drop_is_counted_and_evicts_the_oldest() { + let mut q = EventQueue::new(DedupMode::Queue); + let chan = Uuid::new_v4(); + for i in 0..MAX_PENDING_PER_CHANNEL { + assert!(q.push(make_queued(chan, &format!("event {i}")))); + } + assert_eq!(q.drop_counts().total(), 0, "at the cap, nothing lost yet"); + + // One past the cap evicts the OLDEST — the request that has waited + // longest is the one lost, which is worth asserting because it is the + // opposite of what someone waiting would expect. + assert!(q.push(make_queued(chan, "one too many"))); + let drops = q.drop_counts(); + assert_eq!(drops.queue_depth_cap, 1); + assert_eq!(drops.in_flight_dedup, 0, "wrong cause attributed"); + + let batch = q.flush_next().expect("a batch is pending"); + let first = batch.events.first().expect("batch is non-empty"); + assert!( + !first.event.content.contains("event 0"), + "the oldest event should have been evicted, but it is still at the front" + ); + } + fn make_event(content: &str) -> Event { let keys = Keys::generate(); EventBuilder::new(Kind::Custom(9), content)