diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index f9e7bf1ed8a..2437ebc305f 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -373,6 +373,12 @@ pub struct CliArgs { value_parser = clap::value_parser!(u32))] pub max_turns_per_session: u32, + /// Maximum retry attempts before a stuck event batch is dead-lettered. + #[arg(long, env = "BUZZ_ACP_QUEUE_MAX_RETRIES", + default_value_t = crate::queue::MAX_RETRIES, + value_parser = clap::value_parser!(u32).range(1..))] + pub queue_max_retries: u32, + /// Disable automatic presence (online/offline) status. #[arg(long, env = "BUZZ_ACP_NO_PRESENCE")] pub no_presence: bool, @@ -531,6 +537,9 @@ pub struct Config { pub context_message_limit: u32, /// Maximum turns per session before proactive rotation. 0 = disabled. pub max_turns_per_session: u32, + /// Maximum retry attempts before a stuck event batch is dead-lettered. + /// Defaults to [`crate::queue::MAX_RETRIES`] (10). + pub queue_max_retries: u32, pub presence_enabled: bool, pub typing_enabled: bool, /// Whether NIP-AE agent core memory injection is enabled. When false, @@ -1101,6 +1110,7 @@ impl Config { config_path: args.config, context_message_limit: args.context_message_limit, max_turns_per_session: args.max_turns_per_session, + queue_max_retries: args.queue_max_retries, presence_enabled: !args.no_presence, typing_enabled: !args.no_typing, memory_enabled: args.memory && !args.no_memory, @@ -1143,7 +1153,7 @@ impl Config { format!(" allowed_respond_to=[{}]", modes.join(",")) }; format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}", + "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} queue_max_retries={} presence={} typing={} memory={} model={} permission_mode={} {}{}", self.relay_url, self.keys.public_key().to_hex(), self.agent_command, @@ -1159,6 +1169,7 @@ impl Config { self.ignore_self, self.context_message_limit, self.max_turns_per_session, + self.queue_max_retries, self.presence_enabled, self.typing_enabled, self.memory_enabled, @@ -1476,6 +1487,7 @@ mod tests { config_path: PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, + queue_max_retries: crate::queue::MAX_RETRIES, presence_enabled: true, typing_enabled: true, memory_enabled: true, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 27b9000b7bb..310c28e8ceb 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1984,8 +1984,9 @@ async fn tokio_main() -> Result<()> { let runtime_start_nonce = std::env::var("BUZZ_MANAGED_AGENT_START_NONCE").unwrap_or_default(); let dedup_mode = config.dedup_mode; - let mut queue = - EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs); + let mut queue = EventQueue::new(dedup_mode) + .with_in_flight_deadline(config.max_turn_duration_secs) + .with_max_retries(config.queue_max_retries); // Online means the harness can receive work, not merely that its socket is // connected. Publishing after channel subscriptions gives desktop callers @@ -6520,6 +6521,7 @@ mod build_mcp_servers_tests { config_path: std::path::PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, + queue_max_retries: crate::queue::MAX_RETRIES, presence_enabled: true, typing_enabled: true, memory_enabled: false, @@ -6743,6 +6745,7 @@ mod error_outcome_emission_tests { config_path: std::path::PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, + queue_max_retries: crate::queue::MAX_RETRIES, presence_enabled: true, typing_enabled: true, memory_enabled: false, diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index dabee13afd5..88df9659ab3 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -26,7 +26,9 @@ const MAX_PENDING_PER_CHANNEL: usize = 500; /// Maximum events drained into a single batch. const MAX_BATCH_EVENTS: usize = 50; -/// Maximum retry attempts before a batch is dead-lettered. +/// Default maximum retry attempts before a batch is dead-lettered. +/// Configurable via `--queue-max-retries` / `BUZZ_ACP_QUEUE_MAX_RETRIES` +/// ([`Config::queue_max_retries`](crate::config::Config::queue_max_retries)). pub(crate) const MAX_RETRIES: u32 = 10; /// Base retry delay in seconds (doubled each attempt). @@ -168,6 +170,10 @@ pub struct EventQueue { /// Must be strictly greater than `max_turn_duration` so a turn running to /// the hard cap returns via `mark_complete` before the backstop fires. in_flight_deadline: Duration, + /// Maximum retry attempts before a batch is dead-lettered. Defaults to + /// [`MAX_RETRIES`]; overridden via + /// [`with_max_retries`](Self::with_max_retries). + max_retries: u32, } impl EventQueue { @@ -189,6 +195,7 @@ impl EventQueue { cancel_reasons: HashMap::new(), withheld_native_steer: HashMap::new(), in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS), + max_retries: MAX_RETRIES, } } @@ -200,6 +207,13 @@ impl EventQueue { self } + /// Override the maximum retry attempts before a batch is dead-lettered. + /// Defaults to [`MAX_RETRIES`] (10) when not called. + pub fn with_max_retries(mut self, max_retries: u32) -> Self { + self.max_retries = max_retries; + self + } + /// Monotonically extend an existing in-flight deadline for `channel_id`. /// /// Called when a successful steer grants a fresh turn budget. The new @@ -434,13 +448,13 @@ impl EventQueue { *count }; - if attempt > MAX_RETRIES { + if attempt > self.max_retries { tracing::error!( channel_id = %channel_id, attempt, events = batch.events.len(), "dead-lettering batch after {} retries — discarding {} events", - MAX_RETRIES, + self.max_retries, batch.events.len(), ); self.retry_counts.remove(&channel_id); @@ -466,7 +480,7 @@ impl EventQueue { tracing::warn!( channel_id = %channel_id, attempt, - max = MAX_RETRIES, + max = self.max_retries, delay_secs = delay.as_secs_f64(), events = batch.events.len(), "requeueing failed batch with backoff" @@ -3121,6 +3135,34 @@ mod tests { assert!(!q.retry_after.contains_key(&ch)); } + #[test] + fn test_requeue_dead_letters_after_custom_max_retries() { + let custom_max_retries = 2; + let mut q = EventQueue::new(DedupMode::Queue).with_max_retries(custom_max_retries); + let ch = Uuid::new_v4(); + + q.push(make_queued(ch, "poison")); + for attempt in 1..=custom_max_retries { + q.retry_after + .insert(ch, Instant::now() - Duration::from_secs(1)); + let batch = q.flush_next().expect("flush"); + assert!( + q.requeue(batch).is_none(), + "attempt {attempt} should requeue, not dead-letter" + ); + q.mark_complete(ch); + } + + // The (custom_max_retries + 1)'th failure dead-letters: batch is returned. + q.retry_after + .insert(ch, Instant::now() - Duration::from_secs(1)); + let batch = q.flush_next().expect("flush"); + let dead = q + .requeue(batch) + .expect("should dead-letter at custom threshold"); + assert_eq!(dead.channel_id, ch); + } + #[test] fn test_retry_throttle_blocks_requeue_channel() { let mut q = EventQueue::new(DedupMode::Queue);