diff --git a/CHANGELOG.md b/CHANGELOG.md index 997d9c2f5..7105ca1df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `agent-relay-broker` retains terminally-failed deliveries (retry cap exhausted or recipient gone) in a persisted, capped dead-letter queue instead of discarding them, emitting `dead_letter_added`/`dead_letter_redelivered` broker events and exposing `GET /api/dead-letters` and `POST /api/dead-letters/redeliver`. +- `agent-relay local deadletters` lists dead-letter deliveries and `agent-relay local redeliver ` requeues them through the normal delivery path with a reset retry count; `@agent-relay/harness-driver` adds matching `getDeadLetters()`/`redeliverDeadLetters()` client methods. +- `agent-relay-broker` persists the inbound-event dedup cache alongside pending deliveries and reloads it on startup (dropping expired entries), so a crash + restart no longer re-injects duplicates when the persisted pending file replays. - `@agent-relay/sdk` wires the durable delivery surface to the Relaycast backend: `inbox.list`, `inbox.subscribe`, `inbox.ack/fail/defer`, and `deliveries.ack/fail/defer` now use the hosted delivery ledger, agent-scoped capabilities report `serverDeliveryState: true`, and `DeliveryRunner` works against Relaycast-backed inbox items. ### Changed diff --git a/crates/broker/src/dedup.rs b/crates/broker/src/dedup.rs index d1f1a1e61..abdd0e809 100644 --- a/crates/broker/src/dedup.rs +++ b/crates/broker/src/dedup.rs @@ -1,14 +1,27 @@ use std::{ collections::{HashMap, VecDeque}, + path::Path, time::{Duration, Instant}, }; +use serde::{Deserialize, Serialize}; + +/// Wall-clock snapshot of one dedup entry for crash recovery. `Instant`s +/// cannot be serialized, so insertion times are persisted as unix millis and +/// converted back relative to load time. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PersistedDedupEntry { + pub key: String, + pub inserted_at_ms: u64, +} + #[derive(Debug)] pub struct DedupCache { ttl: Duration, max_entries: usize, seen: HashMap, order: VecDeque<(String, Instant)>, + dirty: bool, } impl DedupCache { @@ -18,6 +31,7 @@ impl DedupCache { max_entries, seen: HashMap::new(), order: VecDeque::new(), + dirty: false, } } @@ -27,6 +41,7 @@ impl DedupCache { return false; } + self.dirty = true; self.seen.insert(id.to_string(), now); self.order.push_back((id.to_string(), now)); @@ -51,11 +66,14 @@ impl DedupCache { } self.order.pop_front(); self.seen.remove(&id); + self.dirty = true; } } pub fn remove(&mut self, id: &str) { - self.seen.remove(id); + if self.seen.remove(id).is_some() { + self.dirty = true; + } self.order.retain(|(key, _)| key != id); } @@ -66,13 +84,83 @@ impl DedupCache { pub fn is_empty(&self) -> bool { self.seen.is_empty() } + + /// Return whether the cache was mutated since the last call, clearing the flag. + pub fn take_dirty(&mut self) -> bool { + std::mem::take(&mut self.dirty) + } + + /// Snapshot live entries as wall-clock timestamps (oldest first). + /// `now`/`now_ms` must describe the same moment on both clocks. + pub fn to_persisted(&self, now: Instant, now_ms: u64) -> Vec { + self.order + .iter() + .map(|(key, inserted_at)| PersistedDedupEntry { + key: key.clone(), + inserted_at_ms: now_ms + .saturating_sub(now.duration_since(*inserted_at).as_millis() as u64), + }) + .collect() + } + + /// Rebuild a cache from persisted entries, dropping anything whose TTL + /// already elapsed. `now`/`now_ms` must describe the same moment on both + /// clocks. + pub fn from_persisted( + ttl: Duration, + max_entries: usize, + entries: Vec, + now: Instant, + now_ms: u64, + ) -> Self { + let mut cache = Self::new(ttl, max_entries); + let mut entries = entries; + entries.sort_by_key(|entry| entry.inserted_at_ms); + for entry in entries { + let elapsed = Duration::from_millis(now_ms.saturating_sub(entry.inserted_at_ms)); + if elapsed >= ttl { + continue; + } + // `checked_sub` can fail close to process/system start; treating + // the entry as just-inserted only extends its dedup window. + let inserted_at = now.checked_sub(elapsed).unwrap_or(now); + if cache.seen.contains_key(&entry.key) { + continue; + } + cache.seen.insert(entry.key.clone(), inserted_at); + cache.order.push_back((entry.key, inserted_at)); + while cache.seen.len() > cache.max_entries { + if let Some((old_id, _)) = cache.order.pop_front() { + cache.seen.remove(&old_id); + } + } + } + cache + } +} + +pub(crate) fn unix_now_millis() -> u64 { + chrono::Utc::now().timestamp_millis().max(0) as u64 +} + +pub(crate) fn save_dedup_cache(path: &Path, cache: &DedupCache) -> anyhow::Result<()> { + let persisted = cache.to_persisted(Instant::now(), unix_now_millis()); + crate::util::fs::write_json_atomic(path, &persisted) +} + +pub(crate) fn load_dedup_cache(path: &Path, ttl: Duration, max_entries: usize) -> DedupCache { + let entries: Vec = std::fs::read_to_string(path) + .ok() + .and_then(|data| serde_json::from_str(&data).ok()) + .unwrap_or_default(); + DedupCache::from_persisted(ttl, max_entries, entries, Instant::now(), unix_now_millis()) } #[cfg(test)] mod tests { use std::time::{Duration, Instant}; - use super::DedupCache; + use super::{load_dedup_cache, save_dedup_cache, DedupCache, PersistedDedupEntry}; #[test] fn drops_duplicates() { @@ -101,4 +189,81 @@ mod tests { // After TTL expires, should be insertable again assert!(dedup.insert_if_new("x", now + Duration::from_secs(6))); } + + #[test] + fn tracks_dirty_on_mutation() { + let mut dedup = DedupCache::new(Duration::from_secs(60), 100); + assert!(!dedup.take_dirty(), "fresh cache starts clean"); + + assert!(dedup.insert_if_new("a", Instant::now())); + assert!(dedup.take_dirty(), "new insert marks dirty"); + assert!(!dedup.take_dirty(), "take_dirty clears the flag"); + + assert!(!dedup.insert_if_new("a", Instant::now())); + assert!(!dedup.take_dirty(), "duplicate insert is not a mutation"); + + dedup.remove("a"); + assert!(dedup.take_dirty(), "remove marks dirty"); + dedup.remove("a"); + assert!( + !dedup.take_dirty(), + "removing a missing key is not a mutation" + ); + } + + #[test] + fn persistence_round_trips_and_drops_expired() { + let dir = tempfile::tempdir().expect("tempdir should create"); + let path = dir.path().join("dedup.json"); + let ttl = Duration::from_secs(300); + + let mut dedup = DedupCache::new(ttl, 100); + let now = Instant::now(); + dedup.insert_if_new("ws_demo:evt_1", now); + dedup.insert_if_new("ws_demo:evt_2", now); + save_dedup_cache(&path, &dedup).expect("dedup cache should save"); + + let mut reloaded = load_dedup_cache(&path, ttl, 100); + assert_eq!(reloaded.len(), 2, "live entries survive a reload"); + assert!( + !reloaded.insert_if_new("ws_demo:evt_1", Instant::now()), + "reloaded entries still deduplicate replayed events" + ); + assert!( + reloaded.insert_if_new("ws_demo:evt_3", Instant::now()), + "new events still insert after reload" + ); + } + + #[test] + fn from_persisted_drops_expired_entries() { + let ttl = Duration::from_secs(300); + let now_ms: u64 = 1_700_000_000_000; + let entries = vec![ + PersistedDedupEntry { + key: "fresh".to_string(), + inserted_at_ms: now_ms - 1_000, + }, + PersistedDedupEntry { + key: "expired".to_string(), + inserted_at_ms: now_ms - ttl.as_millis() as u64 - 1, + }, + ]; + + let mut cache = DedupCache::from_persisted(ttl, 100, entries, Instant::now(), now_ms); + assert_eq!(cache.len(), 1, "expired entries are dropped on load"); + assert!(!cache.insert_if_new("fresh", Instant::now())); + assert!(cache.insert_if_new("expired", Instant::now())); + } + + #[test] + fn load_missing_or_corrupt_file_yields_empty_cache() { + let dir = tempfile::tempdir().expect("tempdir should create"); + let missing = dir.path().join("missing.json"); + assert!(load_dedup_cache(&missing, Duration::from_secs(300), 100).is_empty()); + + let corrupt = dir.path().join("corrupt.json"); + std::fs::write(&corrupt, "not json").expect("seed file should write"); + assert!(load_dedup_cache(&corrupt, Duration::from_secs(300), 100).is_empty()); + } } diff --git a/crates/broker/src/listen_api.rs b/crates/broker/src/listen_api.rs index 7f985f69f..ea7517116 100644 --- a/crates/broker/src/listen_api.rs +++ b/crates/broker/src/listen_api.rs @@ -11,7 +11,9 @@ use std::{ }; use crate::{ - ids::{ChannelName, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId}, + ids::{ + ChannelName, DeliveryId, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId, + }, protocol::{MessageInjectionMode, ResolvedHarnessConfig}, relaycast::WorkspaceMembershipSummary, replay_buffer::ReplayBuffer, @@ -130,6 +132,18 @@ pub enum ListenApiRequest { GetCrashInsights { reply: tokio::sync::oneshot::Sender>, }, + /// `GET /api/dead-letters` — list terminally-failed deliveries retained + /// in the dead-letter queue. + GetDeadLetters { + reply: tokio::sync::oneshot::Sender>, + }, + /// `POST /api/dead-letters/redeliver` — requeue dead-letter entries + /// through the normal delivery path with a reset retry count. `id: None` + /// redelivers every entry whose recipient is currently running. + RedeliverDeadLetters { + id: Option, + reply: tokio::sync::oneshot::Sender>, + }, Preflight { agents: Vec, reply: tokio::sync::oneshot::Sender>, @@ -401,6 +415,11 @@ fn listen_api_router_with_auth( "/api/crash-insights", routing::get(listen_api_crash_insights), ) + .route("/api/dead-letters", routing::get(listen_api_dead_letters)) + .route( + "/api/dead-letters/redeliver", + routing::post(listen_api_redeliver_dead_letters), + ) .route("/api/preflight", routing::post(listen_api_preflight)) .route("/api/shutdown", routing::post(listen_api_shutdown)) .route( @@ -1887,6 +1906,72 @@ async fn listen_api_crash_insights( } } +async fn listen_api_dead_letters( + axum::extract::State(state): axum::extract::State, +) -> (axum::http::StatusCode, axum::Json) { + let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); + if state + .tx + .send(ListenApiRequest::GetDeadLetters { reply: reply_tx }) + .await + .is_err() + { + return internal_error(); + } + match reply_rx.await { + Ok(Ok(val)) => (axum::http::StatusCode::OK, axum::Json(val)), + Ok(Err(err)) => api_error(axum::http::StatusCode::INTERNAL_SERVER_ERROR, "error", err), + Err(_) => internal_error(), + } +} + +#[derive(Deserialize, Default)] +struct RedeliverBody { + /// Dead-letter delivery id to requeue; omit (with `all: true`) to + /// requeue everything. + id: Option, + #[serde(default)] + all: bool, +} + +async fn listen_api_redeliver_dead_letters( + axum::extract::State(state): axum::extract::State, + axum::Json(body): axum::Json, +) -> (axum::http::StatusCode, axum::Json) { + let id = match (&body.id, body.all) { + (Some(id), false) => Some(DeliveryId::from(id.as_str())), + (None, true) => None, + _ => { + return api_error( + axum::http::StatusCode::BAD_REQUEST, + "invalid_request", + "provide exactly one of 'id' or 'all: true'".to_string(), + ); + } + }; + let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); + if state + .tx + .send(ListenApiRequest::RedeliverDeadLetters { + id, + reply: reply_tx, + }) + .await + .is_err() + { + return internal_error(); + } + match reply_rx.await { + Ok(Ok(val)) => (axum::http::StatusCode::OK, axum::Json(val)), + Ok(Err(err)) => api_error( + axum::http::StatusCode::NOT_FOUND, + "dead_letter_not_found", + err, + ), + Err(_) => internal_error(), + } +} + // --------------------------------------------------------------------------- // Lifecycle // --------------------------------------------------------------------------- diff --git a/crates/broker/src/protocol.rs b/crates/broker/src/protocol.rs index 522b7aea4..01c6bcedd 100644 --- a/crates/broker/src/protocol.rs +++ b/crates/broker/src/protocol.rs @@ -447,6 +447,20 @@ pub enum BrokerEvent { #[serde(rename = "lastError")] last_error: String, }, + DeadLetterAdded { + name: WorkerName, + delivery_id: DeliveryId, + event_id: EventId, + from: String, + to: MessageTarget, + attempts: u32, + reason: String, + }, + DeadLetterRedelivered { + name: WorkerName, + delivery_id: DeliveryId, + event_id: EventId, + }, DeliveryQueued { delivery_id: DeliveryId, agent: WorkerName, diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 5cb6aab82..1e85ca52b 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -16,6 +16,7 @@ impl BrokerRuntime { let telemetry = &self.telemetry; let agent_spawn_count = &mut self.agent_spawn_count; let pending_deliveries = &mut self.pending_deliveries; + let dead_letters = &mut self.dead_letters; let pending_requests = &mut self.pending_requests; let delivery_states = &mut self.delivery_states; let agent_result_tokens = &mut self.agent_result_tokens; @@ -425,6 +426,7 @@ impl BrokerRuntime { ).await; let _ = emit_dropped_delivery_failures( sdk_out_tx, + dead_letters, &dropped, "agent_released", ) @@ -1182,6 +1184,7 @@ impl BrokerRuntime { "agents": workers.list(), "pending_delivery_count": pending.len(), "pending_deliveries": pending, + "dead_letter_count": dead_letters.len(), "auth": { "authenticated": !auth_workspaces.is_empty(), "workspace_count": auth_workspaces.len(), @@ -1193,6 +1196,85 @@ impl BrokerRuntime { ListenApiRequest::GetCrashInsights { reply } => { let _ = reply.send(Ok(crash_insights.to_json())); } + ListenApiRequest::GetDeadLetters { reply } => { + let now_ms = unix_timestamp_millis(); + let entries: Vec = dead_letters + .iter() + .map(|entry| { + json!({ + "delivery_id": entry.delivery.delivery_id, + "worker_name": entry.worker_name, + "event_id": entry.delivery.event_id, + "from": entry.delivery.from, + "to": entry.delivery.target, + "attempts": entry.attempts, + "reason": entry.reason, + "queued_at_ms": entry.queued_at_ms, + "failed_at_ms": entry.failed_at_ms, + "age_ms": now_ms.saturating_sub(entry.failed_at_ms), + }) + }) + .collect(); + let _ = reply.send(Ok(json!({ + "count": entries.len(), + "dead_letters": entries, + }))); + } + ListenApiRequest::RedeliverDeadLetters { id, reply } => { + let candidate_ids: Vec = match id { + Some(id) => { + if dead_letters.get(&id).is_none() { + let _ = + reply.send(Err(format!("no dead-letter delivery with id '{id}'"))); + return; + } + vec![id] + } + None => dead_letters.delivery_ids(), + }; + + let mut redelivered: Vec = Vec::new(); + let mut skipped: Vec = Vec::new(); + for delivery_id in candidate_ids { + let Some(entry) = dead_letters.get(&delivery_id) else { + continue; + }; + // Leave entries for absent recipients in the queue — + // requeueing them would only bounce straight back here + // with `recipient gone` on the next maintenance tick. + if !workers.has_worker(&entry.worker_name) { + skipped.push(json!({ + "delivery_id": delivery_id, + "worker_name": entry.worker_name, + "reason": "recipient not running", + })); + continue; + } + let Some(pending) = + requeue_dead_letter(dead_letters, pending_deliveries, &delivery_id) + else { + continue; + }; + let _ = send_broker_event( + sdk_out_tx, + BrokerEvent::DeadLetterRedelivered { + name: pending.worker_name.clone(), + delivery_id: pending.delivery.delivery_id.clone(), + event_id: pending.delivery.event_id.clone(), + }, + ) + .await; + redelivered.push(json!({ + "delivery_id": pending.delivery.delivery_id, + "worker_name": pending.worker_name, + "event_id": pending.delivery.event_id, + })); + } + let _ = reply.send(Ok(json!({ + "redelivered": redelivered, + "skipped": skipped, + }))); + } ListenApiRequest::Preflight { agents, reply } => { let count = agents.len(); let _ = reply.send(Ok(json!({ "queued": count }))); diff --git a/crates/broker/src/runtime/dead_letter.rs b/crates/broker/src/runtime/dead_letter.rs new file mode 100644 index 000000000..0cf8d6417 --- /dev/null +++ b/crates/broker/src/runtime/dead_letter.rs @@ -0,0 +1,198 @@ +use super::*; + +/// Maximum dead letters retained before the oldest entry is evicted. +pub(crate) const MAX_DEAD_LETTERS: usize = 500; + +/// A delivery that terminally failed (retry cap exhausted or the recipient +/// is gone) and was moved out of the pending map instead of being discarded. +/// Retains the full [`RelayDelivery`] so `redeliver` can requeue it through +/// the normal delivery path. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub(crate) struct DeadLetterEntry { + pub(super) worker_name: WorkerName, + pub(super) delivery: RelayDelivery, + pub(super) attempts: u32, + pub(super) reason: String, + /// When the delivery was first queued (unix millis). + #[serde(default)] + pub(super) queued_at_ms: u64, + /// When the delivery terminally failed (unix millis). + #[serde(default)] + pub(super) failed_at_ms: u64, +} + +impl DeadLetterEntry { + pub(crate) fn from_pending(pending: &PendingDelivery, reason: &str) -> Self { + Self { + worker_name: pending.worker_name.clone(), + delivery: pending.delivery.clone(), + attempts: pending.attempts, + reason: reason.to_string(), + queued_at_ms: pending.queued_at_ms, + failed_at_ms: unix_timestamp_millis(), + } + } +} + +/// Bounded FIFO of terminally-failed deliveries with dirty tracking, mirroring +/// [`PendingDeliveryStore`]: mutations mark the store dirty and the event loop +/// persists the snapshot right after the mutating event. Entries are ordered +/// oldest-first; the cap evicts the oldest with a warning. +#[derive(Debug, Default)] +pub(crate) struct DeadLetterStore { + entries: VecDeque, + dirty: bool, +} + +impl DeadLetterStore { + pub(crate) fn new(entries: Vec) -> Self { + Self { + entries: entries.into(), + dirty: false, + } + } + + /// Return whether the store was mutated since the last call, clearing the flag. + pub(crate) fn take_dirty(&mut self) -> bool { + std::mem::take(&mut self.dirty) + } + + /// Append an entry, evicting (and returning) the oldest one when the + /// store is at [`MAX_DEAD_LETTERS`]. + pub(crate) fn push(&mut self, entry: DeadLetterEntry) -> Option { + self.dirty = true; + let evicted = if self.entries.len() >= MAX_DEAD_LETTERS { + let evicted = self.entries.pop_front(); + if let Some(ref old) = evicted { + tracing::warn!( + target = "agent_relay::broker", + delivery_id = %old.delivery.delivery_id, + worker = %old.worker_name, + max_dead_letters = MAX_DEAD_LETTERS, + "dead-letter queue full — evicting oldest entry" + ); + } + evicted + } else { + None + }; + self.entries.push_back(entry); + evicted + } + + pub(crate) fn remove(&mut self, delivery_id: &str) -> Option { + let index = self + .entries + .iter() + .position(|entry| entry.delivery.delivery_id.as_str() == delivery_id)?; + self.dirty = true; + self.entries.remove(index) + } + + pub(crate) fn get(&self, delivery_id: &str) -> Option<&DeadLetterEntry> { + self.entries + .iter() + .find(|entry| entry.delivery.delivery_id.as_str() == delivery_id) + } + + pub(crate) fn iter(&self) -> impl Iterator { + self.entries.iter() + } + + pub(crate) fn delivery_ids(&self) -> Vec { + self.entries + .iter() + .map(|entry| entry.delivery.delivery_id.clone()) + .collect() + } + + pub(crate) fn len(&self) -> usize { + self.entries.len() + } + + pub(crate) fn is_empty(&self) -> bool { + self.entries.is_empty() + } +} + +pub(crate) fn save_dead_letters(path: &Path, store: &DeadLetterStore) -> Result<()> { + let entries: Vec<&DeadLetterEntry> = store.iter().collect(); + crate::util::fs::write_json_atomic(path, &entries) +} + +pub(crate) fn load_dead_letters(path: &Path) -> Vec { + let data = match std::fs::read_to_string(path) { + Ok(d) => d, + Err(_) => return Vec::new(), + }; + serde_json::from_str(&data).unwrap_or_default() +} + +/// Persist or remove the dead-letter file during graceful shutdown. Unlike +/// pending deliveries the entries are terminal, so this only keeps the +/// snapshot in sync: write when non-empty, remove when empty. +pub(crate) fn persist_dead_letters_on_shutdown( + path: &Path, + persist: bool, + store: &DeadLetterStore, +) { + if !persist { + return; + } + if store.is_empty() { + let _ = std::fs::remove_file(path); + return; + } + if let Err(error) = save_dead_letters(path, store) { + tracing::warn!( + path = %path.display(), + error = %error, + "failed to persist dead letters during shutdown" + ); + } +} + +/// Move a dead-letter entry back into the pending map so the normal +/// delivery/retry path picks it up on the next maintenance tick. Resets the +/// retry count and clears the recorded error. Returns the requeued +/// [`PendingDelivery`] or `None` when the id is not in the dead-letter store. +pub(crate) fn requeue_dead_letter( + dead_letters: &mut DeadLetterStore, + pending_deliveries: &mut HashMap, + delivery_id: &str, +) -> Option { + let entry = dead_letters.remove(delivery_id)?; + let pending = PendingDelivery { + worker_name: entry.worker_name, + delivery: entry.delivery, + attempts: 0, + next_retry_at: Instant::now(), + queued_at_ms: entry.queued_at_ms, + last_error: None, + }; + pending_deliveries.insert(pending.delivery.delivery_id.clone(), pending.clone()); + Some(pending) +} + +/// Push a terminally-failed pending delivery into the dead-letter store and +/// emit the `dead_letter_added` broker event (plus the cap-eviction warning +/// handled inside [`DeadLetterStore::push`]). +pub(crate) async fn dead_letter_pending_delivery( + sdk_out_tx: &mpsc::Sender>, + dead_letters: &mut DeadLetterStore, + pending: &PendingDelivery, + reason: &str, +) { + let entry = DeadLetterEntry::from_pending(pending, reason); + let event = BrokerEvent::DeadLetterAdded { + name: entry.worker_name.clone(), + delivery_id: entry.delivery.delivery_id.clone(), + event_id: entry.delivery.event_id.clone(), + from: entry.delivery.from.clone(), + to: entry.delivery.target.clone(), + attempts: entry.attempts, + reason: entry.reason.clone(), + }; + dead_letters.push(entry); + let _ = send_broker_event(sdk_out_tx, event).await; +} diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index 5b0b32408..a82848af7 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -1,6 +1,6 @@ use super::*; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq)] pub(crate) struct PendingDelivery { pub(super) worker_name: WorkerName, pub(super) delivery: RelayDelivery, @@ -22,20 +22,18 @@ pub(crate) struct PersistedPendingDelivery { pub(super) last_error: Option, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq)] pub(crate) enum DeliveryAttemptOutcome { Attempted { worker_name: WorkerName, attempts: u32, event_id: EventId, }, + /// Terminal failure: the entry was removed from the pending map. The full + /// [`PendingDelivery`] rides along so the caller can move it into the + /// dead-letter store instead of discarding the message. Failed { - worker_name: WorkerName, - delivery_id: DeliveryId, - event_id: EventId, - from: String, - to: MessageTarget, - attempts: u32, + pending: Box, last_error: String, }, Noop, @@ -130,14 +128,7 @@ pub(crate) fn save_pending_deliveries( last_error: pd.last_error.clone(), }) .collect(); - let json = serde_json::to_string_pretty(&persisted)?; - let dir = path.parent().unwrap_or(path); - let mut tmp = tempfile::NamedTempFile::new_in(dir) - .with_context(|| format!("failed creating temp file in {}", dir.display()))?; - std::io::Write::write_all(&mut tmp, json.as_bytes())?; - tmp.persist(path) - .with_context(|| format!("failed persisting pending deliveries to {}", path.display()))?; - Ok(()) + crate::util::fs::write_json_atomic(path, &persisted) } pub(crate) fn load_pending_deliveries(path: &Path) -> HashMap { @@ -686,28 +677,20 @@ pub(crate) async fn retry_pending_delivery( if pending.attempts >= MAX_DELIVERY_RETRIES { let removed = pending_deliveries.remove(delivery_id).unwrap_or(pending); + let last_error = removed + .last_error + .clone() + .unwrap_or_else(|| "max delivery retries exceeded".to_string()); return Ok(DeliveryAttemptOutcome::Failed { - worker_name: removed.worker_name, - delivery_id: removed.delivery.delivery_id, - event_id: removed.delivery.event_id, - from: removed.delivery.from, - to: removed.delivery.target, - attempts: removed.attempts, - last_error: removed - .last_error - .unwrap_or_else(|| "max delivery retries exceeded".to_string()), + pending: Box::new(removed), + last_error, }); } if !workers.has_worker(&pending.worker_name) { let removed = pending_deliveries.remove(delivery_id).unwrap_or(pending); return Ok(DeliveryAttemptOutcome::Failed { - worker_name: removed.worker_name, - delivery_id: removed.delivery.delivery_id, - event_id: removed.delivery.event_id, - from: removed.delivery.from, - to: removed.delivery.target, - attempts: removed.attempts, + pending: Box::new(removed), last_error: "recipient gone".to_string(), }); } @@ -741,16 +724,13 @@ pub(crate) async fn retry_pending_delivery( if should_fail { if let Some(removed) = pending_deliveries.remove(delivery_id) { + let last_error = removed + .last_error + .clone() + .unwrap_or_else(|| "max delivery retries exceeded".to_string()); return Ok(DeliveryAttemptOutcome::Failed { - worker_name: removed.worker_name, - delivery_id: removed.delivery.delivery_id, - event_id: removed.delivery.event_id, - from: removed.delivery.from, - to: removed.delivery.target, - attempts: removed.attempts, - last_error: removed - .last_error - .unwrap_or_else(|| "max delivery retries exceeded".to_string()), + pending: Box::new(removed), + last_error, }); } return Ok(DeliveryAttemptOutcome::Noop); @@ -762,6 +742,7 @@ pub(crate) async fn retry_pending_delivery( pub(crate) async fn emit_delivery_attempt_outcome( sdk_out_tx: &mpsc::Sender>, + dead_letters: &mut DeadLetterStore, delivery_id: &DeliveryId, was_retry: bool, outcome: DeliveryAttemptOutcome, @@ -786,27 +767,23 @@ pub(crate) async fn emit_delivery_attempt_outcome( } } DeliveryAttemptOutcome::Failed { - worker_name, - delivery_id, - event_id, - from, - to, - attempts, + pending, last_error, } => { send_broker_event( sdk_out_tx, BrokerEvent::MessageDeliveryFailed { - name: worker_name, - delivery_id: Some(delivery_id), - event_id: Some(event_id), - from, - to, - attempts, - last_error, + name: pending.worker_name.clone(), + delivery_id: Some(pending.delivery.delivery_id.clone()), + event_id: Some(pending.delivery.event_id.clone()), + from: pending.delivery.from.clone(), + to: pending.delivery.target.clone(), + attempts: pending.attempts, + last_error: last_error.clone(), }, ) .await?; + dead_letter_pending_delivery(sdk_out_tx, dead_letters, &pending, &last_error).await; } DeliveryAttemptOutcome::Noop => {} } @@ -839,6 +816,7 @@ pub(crate) fn take_pending_for_worker( pub(crate) async fn emit_dropped_delivery_failures( sdk_out_tx: &mpsc::Sender>, + dead_letters: &mut DeadLetterStore, dropped: &[PendingDelivery], reason: &str, ) -> Result<()> { @@ -856,6 +834,7 @@ pub(crate) async fn emit_dropped_delivery_failures( }, ) .await?; + dead_letter_pending_delivery(sdk_out_tx, dead_letters, pending, reason).await; } Ok(()) } diff --git a/crates/broker/src/runtime/event_loop.rs b/crates/broker/src/runtime/event_loop.rs index c322aef6a..727a0fb73 100644 --- a/crates/broker/src/runtime/event_loop.rs +++ b/crates/broker/src/runtime/event_loop.rs @@ -29,6 +29,7 @@ pub(crate) struct BrokerRuntime { pub(super) dedup: DedupCache, pub(super) delivery_retry_interval: Duration, pub(super) pending_deliveries: PendingDeliveryStore, + pub(super) dead_letters: DeadLetterStore, pub(super) terminal_failed_deliveries: HashSet, pub(super) pending_requests: HashMap, pub(super) delivery_states: HashMap, @@ -114,26 +115,52 @@ impl BrokerRuntime { } } - self.flush_pending_deliveries(); + self.flush_persisted_stores(); } self.shutdown_runtime().await } - /// Persist pending deliveries whenever the map was mutated by the event - /// just handled. Keeps the on-disk snapshot in lockstep with the - /// in-memory map so a crash between maintenance ticks cannot lose - /// queued deliveries. - fn flush_pending_deliveries(&mut self) { - if !self.pending_deliveries.take_dirty() || !self.paths.persist { + /// Persist pending deliveries, dead letters, and the dedup cache whenever + /// they were mutated by the event just handled. Keeps the on-disk + /// snapshots in lockstep with the in-memory state so a crash between + /// maintenance ticks cannot lose queued deliveries, dead letters, or + /// already-seen event ids. + fn flush_persisted_stores(&mut self) { + let pending_dirty = self.pending_deliveries.take_dirty(); + let dead_letters_dirty = self.dead_letters.take_dirty(); + let dedup_dirty = self.dedup.take_dirty(); + if !self.paths.persist { return; } - if let Err(error) = save_pending_deliveries(&self.paths.pending, &self.pending_deliveries) { - tracing::warn!( - path = %self.paths.pending.display(), - error = %error, - "failed to persist pending deliveries" - ); + if pending_dirty { + if let Err(error) = + save_pending_deliveries(&self.paths.pending, &self.pending_deliveries) + { + tracing::warn!( + path = %self.paths.pending.display(), + error = %error, + "failed to persist pending deliveries" + ); + } + } + if dead_letters_dirty { + if let Err(error) = save_dead_letters(&self.paths.dead_letters, &self.dead_letters) { + tracing::warn!( + path = %self.paths.dead_letters.display(), + error = %error, + "failed to persist dead letters" + ); + } + } + if dedup_dirty { + if let Err(error) = crate::dedup::save_dedup_cache(&self.paths.dedup, &self.dedup) { + tracing::warn!( + path = %self.paths.dedup.display(), + error = %error, + "failed to persist dedup cache" + ); + } } } @@ -190,6 +217,20 @@ impl BrokerRuntime { self.paths.persist, &self.pending_deliveries, ); + persist_dead_letters_on_shutdown( + &self.paths.dead_letters, + self.paths.persist, + &self.dead_letters, + ); + if self.paths.persist { + if let Err(error) = crate::dedup::save_dedup_cache(&self.paths.dedup, &self.dedup) { + tracing::warn!( + path = %self.paths.dedup.display(), + error = %error, + "failed to persist dedup cache during shutdown" + ); + } + } self.workers.shutdown_all().await?; // Clean up state and connection files on graceful shutdown diff --git a/crates/broker/src/runtime/init.rs b/crates/broker/src/runtime/init.rs index 1c76d3ff3..614fdb641 100644 --- a/crates/broker/src/runtime/init.rs +++ b/crates/broker/src/runtime/init.rs @@ -416,9 +416,27 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re let stdin_open = true; let mut reap_tick = tokio::time::interval(Duration::from_millis(500)); reap_tick.set_missed_tick_behavior(MissedTickBehavior::Skip); - let dedup = DedupCache::new(Duration::from_secs(300), 8192); + // Reload the dedup cache persisted by the previous session so the + // pending-delivery replay below cannot re-inject events the workers + // already saw before a crash/restart. Expired entries drop on load. + let dedup = crate::dedup::load_dedup_cache(&paths.dedup, Duration::from_secs(300), 8192); + if !dedup.is_empty() { + tracing::info!( + count = dedup.len(), + "loaded {} dedup entries from previous session", + dedup.len() + ); + } let delivery_retry_interval = delivery_retry_interval(); let pending_deliveries = PendingDeliveryStore::new(load_pending_deliveries(&paths.pending)); + let dead_letters = DeadLetterStore::new(load_dead_letters(&paths.dead_letters)); + if !dead_letters.is_empty() { + tracing::info!( + count = dead_letters.len(), + "loaded {} dead-letter deliveries from previous session", + dead_letters.len() + ); + } let terminal_failed_deliveries: HashSet = HashSet::new(); // Outstanding worker-bound RPC requests waiting on a `*_response` // frame from the wrapped worker. Keyed by the `request_id` we put on @@ -499,6 +517,7 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re dedup, delivery_retry_interval, pending_deliveries, + dead_letters, terminal_failed_deliveries, pending_requests, delivery_states, diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 246998a7f..f74cf87d7 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -11,6 +11,7 @@ impl BrokerRuntime { let telemetry = &self.telemetry; let crash_insights = &mut self.crash_insights; let pending_deliveries = &mut self.pending_deliveries; + let dead_letters = &mut self.dead_letters; let pending_requests = &mut self.pending_requests; let delivery_states = &mut self.delivery_states; let agent_result_tokens = &mut self.agent_result_tokens; @@ -59,9 +60,14 @@ impl BrokerRuntime { .await { Ok(outcome) => { - let _ = - emit_delivery_attempt_outcome(sdk_out_tx, &delivery_id, was_retry, outcome) - .await; + let _ = emit_delivery_attempt_outcome( + sdk_out_tx, + dead_letters, + &delivery_id, + was_retry, + outcome, + ) + .await; } Err(error) => { let _ = send_error( @@ -159,6 +165,7 @@ impl BrokerRuntime { .await; let _ = emit_dropped_delivery_failures( sdk_out_tx, + dead_letters, &dropped, "worker_permanently_dead", ) @@ -211,9 +218,13 @@ impl BrokerRuntime { }), ) .await; - let _ = - emit_dropped_delivery_failures(sdk_out_tx, &dropped, "worker_exited") - .await; + let _ = emit_dropped_delivery_failures( + sdk_out_tx, + dead_letters, + &dropped, + "worker_exited", + ) + .await; } fail_pending_requests_for_worker(pending_requests, name, "worker_exited"); delivery_states.remove(name); diff --git a/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs index 7153731d0..1e0342e22 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -67,6 +67,7 @@ static TRACING_GUARD: OnceLock = On mod api; mod app_server; mod connection; +mod dead_letter; mod delivery; mod event_loop; mod headless; @@ -86,6 +87,7 @@ mod worker_events; pub(crate) use app_server::*; pub(crate) use connection::*; +pub(crate) use dead_letter::*; pub(crate) use delivery::*; pub(crate) use event_loop::*; pub(crate) use headless::*; diff --git a/crates/broker/src/runtime/paths.rs b/crates/broker/src/runtime/paths.rs index 9a7bdb0c3..6cefe44c4 100644 --- a/crates/broker/src/runtime/paths.rs +++ b/crates/broker/src/runtime/paths.rs @@ -5,6 +5,11 @@ pub(crate) struct RuntimePaths { pub(super) persist: bool, pub(super) state: PathBuf, pub(super) pending: PathBuf, + /// Terminally-failed deliveries retained for `redeliver`. + pub(super) dead_letters: PathBuf, + /// Inbound-event dedup cache snapshot, so a restart that replays the + /// persisted pending file cannot re-inject duplicates. + pub(super) dedup: PathBuf, /// Held for process lifetime to prevent concurrent broker instances (persist mode only). #[allow(dead_code)] pub(super) _lock: Option, @@ -63,6 +68,8 @@ pub(crate) fn ensure_ephemeral_paths(_cwd: &Path, broker_name: &str) -> Result DeadLetterEntry { + DeadLetterEntry::from_pending(&make_pending_delivery(delivery_id, worker), reason) +} + +#[test] +fn dead_letter_store_caps_size_and_evicts_oldest() { + let mut store = DeadLetterStore::default(); + for index in 0..MAX_DEAD_LETTERS { + assert!( + store + .push(make_dead_letter(&format!("del_{index}"), "worker-a", "x")) + .is_none(), + "no eviction below the cap" + ); + } + assert_eq!(store.len(), MAX_DEAD_LETTERS); + + let evicted = store + .push(make_dead_letter("del_overflow", "worker-a", "x")) + .expect("push past the cap evicts the oldest entry"); + assert_eq!(evicted.delivery.delivery_id, DeliveryId::new("del_0")); + assert_eq!(store.len(), MAX_DEAD_LETTERS, "store stays at the cap"); + assert!(store.get("del_0").is_none(), "oldest entry is gone"); + assert!(store.get("del_overflow").is_some(), "newest entry is kept"); +} + +#[test] +fn dead_letter_store_tracks_mutations() { + let mut store = DeadLetterStore::default(); + assert!(!store.take_dirty(), "fresh store starts clean"); + + store.push(make_dead_letter("del_1", "worker-a", "recipient gone")); + assert!(store.take_dirty(), "push marks the store dirty"); + assert!(!store.take_dirty(), "take_dirty clears the flag"); + + assert!(store.get("del_1").is_some()); + assert!(!store.take_dirty(), "reads stay clean"); + + assert!(store.remove("del_missing").is_none()); + assert!( + !store.take_dirty(), + "removing a missing id is not a mutation" + ); + + assert!(store.remove("del_1").is_some()); + assert!(store.take_dirty(), "remove marks the store dirty"); +} + +#[test] +fn redeliver_requeues_dead_letter_and_resets_retries() { + let mut dead_letters = DeadLetterStore::default(); + let mut pending_deliveries: HashMap = HashMap::new(); + let mut source = make_pending_delivery("del_retry", "worker-a"); + source.attempts = MAX_DELIVERY_RETRIES; + source.last_error = Some("max delivery retries exceeded".to_string()); + dead_letters.push(DeadLetterEntry::from_pending( + &source, + "max delivery retries exceeded", + )); + + let requeued = requeue_dead_letter(&mut dead_letters, &mut pending_deliveries, "del_retry") + .expect("dead letter should requeue by id"); + + assert!(dead_letters.is_empty(), "requeued entry leaves the DLQ"); + assert_eq!(requeued.attempts, 0, "retry count resets on redeliver"); + assert_eq!(requeued.last_error, None, "stale error clears on redeliver"); + let pending = pending_deliveries + .get("del_retry") + .expect("requeued delivery joins the pending map"); + assert_eq!(pending.delivery.body, source.delivery.body); + assert_eq!(pending.worker_name, source.worker_name); + assert_eq!( + pending.queued_at_ms, source.queued_at_ms, + "original queue time is preserved for age reporting" + ); + + assert!( + requeue_dead_letter(&mut dead_letters, &mut pending_deliveries, "del_retry").is_none(), + "redelivering an unknown id is a no-op" + ); +} + +#[test] +fn dead_letters_round_trip_persistence() { + let dir = tempfile::tempdir().expect("tempdir should create"); + let path = dir.path().join("dead-letters.json"); + let mut store = DeadLetterStore::default(); + store.push(make_dead_letter("del_a", "worker-a", "recipient gone")); + store.push(make_dead_letter("del_b", "worker-b", "worker_exited")); + + save_dead_letters(&path, &store).expect("dead letters should save"); + + let reloaded = DeadLetterStore::new(load_dead_letters(&path)); + assert_eq!(reloaded.len(), 2); + let entry = reloaded.get("del_a").expect("entry reloads by id"); + assert_eq!(entry.worker_name, WorkerName::from("worker-a")); + assert_eq!(entry.reason, "recipient gone"); + assert_eq!(entry.delivery.event_id, EventId::new("evt_del_a")); + assert!(entry.failed_at_ms > 0, "failure timestamp survives reload"); +} + +#[test] +fn shutdown_persists_dead_letters_and_removes_empty_file() { + let dir = tempfile::tempdir().expect("tempdir should create"); + let path = dir.path().join("dead-letters.json"); + + let mut store = DeadLetterStore::default(); + store.push(make_dead_letter("del_keep", "worker-a", "worker_exited")); + persist_dead_letters_on_shutdown(&path, true, &store); + assert_eq!( + DeadLetterStore::new(load_dead_letters(&path)).len(), + 1, + "dead letters survive shutdown" + ); + + persist_dead_letters_on_shutdown(&path, true, &DeadLetterStore::default()); + assert!(!path.exists(), "empty store removes the file"); + + let mut store = DeadLetterStore::default(); + store.push(make_dead_letter("del_lost", "worker-a", "worker_exited")); + persist_dead_letters_on_shutdown(&path, false, &store); + assert!(!path.exists(), "persistence disabled writes nothing"); +} + +#[tokio::test] +async fn retry_exhaustion_dead_letters_instead_of_discarding() { + let (tx, _rx) = mpsc::channel::(16); + let mut workers = WorkerRegistry::new( + tx, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests"), + Instant::now(), + ); + let mut exhausted = make_pending_delivery("del_exhausted", "ghost"); + exhausted.attempts = MAX_DELIVERY_RETRIES; + exhausted.last_error = Some("failed writing frame".to_string()); + let mut pending_deliveries = + HashMap::from([(DeliveryId::new("del_exhausted"), exhausted.clone())]); + + let outcome = retry_pending_delivery( + &DeliveryId::new("del_exhausted"), + &mut workers, + &mut pending_deliveries, + Duration::from_millis(1), + ) + .await + .expect("exhausted retries should classify as terminal failure"); + + let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); + let mut dead_letters = DeadLetterStore::default(); + emit_delivery_attempt_outcome( + &sdk_out_tx, + &mut dead_letters, + &DeliveryId::new("del_exhausted"), + true, + outcome, + ) + .await + .expect("terminal outcome should emit"); + + assert!( + pending_deliveries.is_empty(), + "entry leaves the pending map" + ); + let entry = dead_letters + .get("del_exhausted") + .expect("exhausted delivery is retained in the dead-letter store"); + assert_eq!(entry.attempts, MAX_DELIVERY_RETRIES); + assert_eq!(entry.reason, "failed writing frame"); + assert_eq!(entry.delivery.body, exhausted.delivery.body); + + let failed_frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) + .await + .expect("message_delivery_failed should emit") + .expect("sdk_out_tx should remain open"); + assert_eq!(failed_frame.payload["kind"], "message_delivery_failed"); + let dead_frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) + .await + .expect("dead_letter_added should emit") + .expect("sdk_out_tx should remain open"); + assert_eq!(dead_frame.payload["kind"], "dead_letter_added"); + assert_eq!(dead_frame.payload["delivery_id"], "del_exhausted"); + assert_eq!(dead_frame.payload["reason"], "failed writing frame"); +} + #[tokio::test] async fn delivery_retry_fails_promptly_when_recipient_is_gone() { let (tx, _rx) = mpsc::channel::(16); @@ -427,18 +612,21 @@ async fn delivery_retry_fails_promptly_when_recipient_is_gone() { .await .expect("retry should classify missing recipient"); - assert_eq!( - outcome, + match outcome { DeliveryAttemptOutcome::Failed { - worker_name: WorkerName::from("ghost"), - delivery_id: DeliveryId::new("del_gone"), - event_id: EventId::new("evt_gone"), - from: "Lead".to_string(), - to: MessageTarget::new("Worker"), - attempts: 3, - last_error: "recipient gone".to_string(), + pending, + last_error, + } => { + assert_eq!(pending.worker_name, WorkerName::from("ghost")); + assert_eq!(pending.delivery.delivery_id, DeliveryId::new("del_gone")); + assert_eq!(pending.delivery.event_id, EventId::new("evt_gone")); + assert_eq!(pending.delivery.from, "Lead"); + assert_eq!(pending.delivery.target, MessageTarget::new("Worker")); + assert_eq!(pending.attempts, 3); + assert_eq!(last_error, "recipient gone"); } - ); + other => panic!("missing recipient should fail terminally, got {other:?}"), + } assert!( pending_deliveries.is_empty(), "terminal failed deliveries are removed so they cannot retry forever" @@ -495,8 +683,11 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { ) .await { - Ok(outcome @ DeliveryAttemptOutcome::Failed { attempts, .. }) => { - assert_eq!(attempts, MAX_DELIVERY_RETRIES); + Ok(outcome @ DeliveryAttemptOutcome::Failed { .. }) => { + let DeliveryAttemptOutcome::Failed { ref pending, .. } = outcome else { + unreachable!(); + }; + assert_eq!(pending.attempts, MAX_DELIVERY_RETRIES); // Some platforms can accept a final pipe write after the child exits, // so terminal failure may arrive on the immediate post-cap check. assert!( @@ -542,9 +733,16 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { ); let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); - emit_delivery_attempt_outcome(&sdk_out_tx, &DeliveryId::new("del_blip"), true, outcome) - .await - .expect("failed outcome should emit to sdk_out_tx"); + let mut dead_letters = DeadLetterStore::default(); + emit_delivery_attempt_outcome( + &sdk_out_tx, + &mut dead_letters, + &DeliveryId::new("del_blip"), + true, + outcome, + ) + .await + .expect("failed outcome should emit to sdk_out_tx"); let frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) .await @@ -570,6 +768,21 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { frame.payload.get("last_error").is_none(), "wire event should use the typed lastError field only" ); + + let dead_frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) + .await + .expect("terminal failure should also emit dead_letter_added") + .expect("sdk_out_tx should remain open"); + assert_eq!(dead_frame.payload["kind"], "dead_letter_added"); + assert_eq!(dead_frame.payload["delivery_id"], "del_blip"); + assert_eq!( + dead_letters.len(), + 1, + "terminal failure is retained in the dead-letter store, not discarded" + ); + let entry = dead_letters.get("del_blip").expect("dead letter by id"); + assert_eq!(entry.delivery.body, "transient auth blip"); + assert_eq!(entry.attempts, MAX_DELIVERY_RETRIES); } #[tokio::test] @@ -1433,10 +1646,16 @@ async fn dropped_pending_deliveries_emit_terminal_message_failures() { last_error: Some("previous blip".to_string()), }; let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); + let mut dead_letters = DeadLetterStore::default(); - emit_dropped_delivery_failures(&sdk_out_tx, &[pending], "worker_permanently_dead") - .await - .expect("dropped delivery failure should emit"); + emit_dropped_delivery_failures( + &sdk_out_tx, + &mut dead_letters, + &[pending], + "worker_permanently_dead", + ) + .await + .expect("dropped delivery failure should emit"); let frame = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()) .await @@ -1451,6 +1670,15 @@ async fn dropped_pending_deliveries_emit_terminal_message_failures() { assert_eq!(frame.payload["to"], "A"); assert_eq!(frame.payload["attempts"].as_u64(), Some(2)); assert_eq!(frame.payload["lastError"], "worker_permanently_dead"); + assert_eq!( + dead_letters.len(), + 1, + "dropped deliveries are retained in the dead-letter store" + ); + assert_eq!( + dead_letters.get("del_1").expect("dead letter by id").reason, + "worker_permanently_dead" + ); } #[test] diff --git a/crates/broker/src/util/fs.rs b/crates/broker/src/util/fs.rs new file mode 100644 index 000000000..15f65dd2c --- /dev/null +++ b/crates/broker/src/util/fs.rs @@ -0,0 +1,18 @@ +use std::path::Path; + +use anyhow::{Context, Result}; + +/// Serialize `value` as pretty JSON and atomically replace `path` via a +/// temp-file-rename in the same directory. Used for broker state snapshots +/// (pending deliveries, dead letters, dedup cache) so a crash mid-write can +/// never leave a truncated file behind. +pub(crate) fn write_json_atomic(path: &Path, value: &T) -> Result<()> { + let json = serde_json::to_string_pretty(value)?; + let dir = path.parent().unwrap_or(path); + let mut tmp = tempfile::NamedTempFile::new_in(dir) + .with_context(|| format!("failed creating temp file in {}", dir.display()))?; + std::io::Write::write_all(&mut tmp, json.as_bytes())?; + tmp.persist(path) + .with_context(|| format!("failed persisting {}", path.display()))?; + Ok(()) +} diff --git a/crates/broker/src/util/mod.rs b/crates/broker/src/util/mod.rs index fe0b6feba..adf1b1664 100644 --- a/crates/broker/src/util/mod.rs +++ b/crates/broker/src/util/mod.rs @@ -1,4 +1,5 @@ pub(crate) mod ansi; +pub(crate) mod fs; pub(crate) mod terminal; pub(crate) mod utf8_stream; pub(crate) mod version; diff --git a/packages/cli/src/cli/bootstrap.test.ts b/packages/cli/src/cli/bootstrap.test.ts index 3906613a8..6bb0f8252 100644 --- a/packages/cli/src/cli/bootstrap.test.ts +++ b/packages/cli/src/cli/bootstrap.test.ts @@ -9,6 +9,8 @@ const expectedLeafCommands = [ 'local down', 'local status', 'local metrics', + 'local deadletters', + 'local redeliver', 'local run', 'local logs', 'local sync', diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index 098e0749a..5ff3b422b 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -8,10 +8,15 @@ const sdkStatusClient = { disconnect: vi.fn(() => undefined), }; +const harnessConnectMock = vi.hoisted(() => vi.fn()); + vi.mock('@agent-relay/harness-driver', () => ({ - HarnessDriverClient: vi.fn(function () { - return sdkStatusClient; - }), + HarnessDriverClient: Object.assign( + vi.fn(function () { + return sdkStatusClient; + }), + { connect: harnessConnectMock } + ), })); const telemetryMocks = vi.hoisted(() => ({ @@ -28,6 +33,7 @@ beforeEach(() => { sdkStatusClient.getSession.mockReset(); sdkStatusClient.getSession.mockResolvedValue({ workspace_key: '' }); sdkStatusClient.disconnect.mockClear(); + harnessConnectMock.mockReset(); telemetryMocks.track.mockClear(); }); @@ -203,7 +209,9 @@ describe('registerCoreCommands', () => { const { program } = createHarness(); const commandNames = program.commands.map((cmd) => cmd.name()); - expect(commandNames).toEqual(expect.arrayContaining(['up', 'down', 'status', 'metrics'])); + expect(commandNames).toEqual( + expect.arrayContaining(['up', 'down', 'status', 'metrics', 'deadletters', 'redeliver']) + ); expect(commandNames).not.toEqual(expect.arrayContaining(['bridge', 'uninstall', 'version', 'update'])); }); @@ -1425,3 +1433,113 @@ describe('registerCoreCommands', () => { expect(deps.log).toHaveBeenCalledWith('Workspace Key: rk_live_new'); }); }); + +describe('dead-letter commands', () => { + function deadLetterEntry(overrides: Record = {}) { + return { + delivery_id: 'del_1', + worker_name: 'Worker', + event_id: 'evt_1', + from: 'Lead', + to: 'Worker', + attempts: 10, + reason: 'max delivery retries exceeded', + queued_at_ms: 1, + failed_at_ms: 2, + age_ms: 65_000, + ...overrides, + }; + } + + it('deadletters lists entries from the broker', async () => { + const client = { + getDeadLetters: vi.fn(async () => ({ count: 1, dead_letters: [deadLetterEntry()] })), + disconnect: vi.fn(), + }; + harnessConnectMock.mockReturnValueOnce(client); + const { program, deps } = createHarness(); + + const exitCode = await runCommand(program, ['deadletters']); + + expect(exitCode).toBeUndefined(); + expect(client.getDeadLetters).toHaveBeenCalled(); + expect(deps.log).toHaveBeenCalledWith(expect.stringContaining('del_1')); + expect(deps.log).toHaveBeenCalledWith(expect.stringContaining('recipient=Worker')); + expect(deps.log).toHaveBeenCalledWith(expect.stringContaining('reason=max delivery retries exceeded')); + expect(deps.log).toHaveBeenCalledWith(expect.stringContaining('age=1m')); + expect(client.disconnect).toHaveBeenCalled(); + }); + + it('deadletters reports an empty queue', async () => { + const client = { + getDeadLetters: vi.fn(async () => ({ count: 0, dead_letters: [] })), + disconnect: vi.fn(), + }; + harnessConnectMock.mockReturnValueOnce(client); + const { program, deps } = createHarness(); + + await runCommand(program, ['deadletters']); + + expect(deps.log).toHaveBeenCalledWith('No dead-letter deliveries.'); + }); + + it('deadletters --json prints the raw response', async () => { + const response = { count: 1, dead_letters: [deadLetterEntry()] }; + const client = { + getDeadLetters: vi.fn(async () => response), + disconnect: vi.fn(), + }; + harnessConnectMock.mockReturnValueOnce(client); + const { program, deps } = createHarness(); + + await runCommand(program, ['deadletters', '--json']); + + expect(deps.log).toHaveBeenCalledWith(JSON.stringify(response, null, 2)); + }); + + it('redeliver requeues a single entry by id', async () => { + const client = { + redeliverDeadLetters: vi.fn(async () => ({ + redelivered: [{ delivery_id: 'del_1', worker_name: 'Worker', event_id: 'evt_1' }], + skipped: [], + })), + disconnect: vi.fn(), + }; + harnessConnectMock.mockReturnValueOnce(client); + const { program, deps } = createHarness(); + + const exitCode = await runCommand(program, ['redeliver', 'del_1']); + + expect(exitCode).toBeUndefined(); + expect(client.redeliverDeadLetters).toHaveBeenCalledWith({ id: 'del_1' }); + expect(deps.log).toHaveBeenCalledWith('Redelivered del_1 to Worker'); + expect(client.disconnect).toHaveBeenCalled(); + }); + + it('redeliver --all requeues everything and reports skips', async () => { + const client = { + redeliverDeadLetters: vi.fn(async () => ({ + redelivered: [{ delivery_id: 'del_1', worker_name: 'Worker', event_id: 'evt_1' }], + skipped: [{ delivery_id: 'del_2', worker_name: 'Ghost', reason: 'recipient not running' }], + })), + disconnect: vi.fn(), + }; + harnessConnectMock.mockReturnValueOnce(client); + const { program, deps } = createHarness(); + + await runCommand(program, ['redeliver', '--all']); + + expect(client.redeliverDeadLetters).toHaveBeenCalledWith({ all: true }); + expect(deps.log).toHaveBeenCalledWith('Redelivered del_1 to Worker'); + expect(deps.warn).toHaveBeenCalledWith('Skipped del_2 (Ghost): recipient not running'); + }); + + it('redeliver requires exactly one of id or --all', async () => { + const { program, deps } = createHarness(); + + expect(await runCommand(program, ['redeliver'])).toBe(1); + expect(await runCommand(program, ['redeliver', 'del_1', '--all'])).toBe(1); + expect(deps.error).toHaveBeenCalledWith('Provide exactly one of or --all.'); + expect(harnessConnectMock).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/cli/src/cli/commands/core.ts b/packages/cli/src/cli/commands/core.ts index a7884cac3..afaa737d5 100644 --- a/packages/cli/src/cli/commands/core.ts +++ b/packages/cli/src/cli/commands/core.ts @@ -399,6 +399,79 @@ export function registerCoreCommands(program: Command, overrides: Partial { + let client: HarnessDriverClient | undefined; + try { + client = HarnessDriverClient.connect({ cwd: process.cwd() }); + const result = await client.getDeadLetters(); + if (options.json) { + deps.log(JSON.stringify(result, null, 2)); + return; + } + if (result.count === 0) { + deps.log('No dead-letter deliveries.'); + return; + } + for (const entry of result.dead_letters) { + deps.log( + `${entry.delivery_id} recipient=${entry.worker_name} from=${entry.from} ` + + `age=${formatAgeMs(entry.age_ms)} attempts=${entry.attempts} reason=${entry.reason}` + ); + } + } catch (err) { + deps.error(err instanceof Error ? err.message : String(err)); + deps.exit(1); + } finally { + client?.disconnect(); + } + }); + + program + .command('redeliver [id]') + .description('Requeue dead-letter deliveries through the normal delivery path') + .option('--all', 'Redeliver every dead-letter entry') + .action(async (id: string | undefined, options: { all?: boolean }) => { + if (Boolean(id) === Boolean(options.all)) { + deps.error('Provide exactly one of or --all.'); + deps.exit(1); + } + let client: HarnessDriverClient | undefined; + try { + client = HarnessDriverClient.connect({ cwd: process.cwd() }); + const result = await client.redeliverDeadLetters(id ? { id } : { all: true }); + for (const entry of result.redelivered) { + deps.log(`Redelivered ${entry.delivery_id} to ${entry.worker_name}`); + } + for (const entry of result.skipped) { + deps.warn(`Skipped ${entry.delivery_id} (${entry.worker_name}): ${entry.reason}`); + } + if (result.redelivered.length === 0 && result.skipped.length === 0) { + deps.log('No dead-letter deliveries to redeliver.'); + } + } catch (err) { + deps.error(err instanceof Error ? err.message : String(err)); + deps.exit(1); + } finally { + client?.disconnect(); + } + }); +} + +/** Render a millisecond age as a compact human-readable duration. */ +function formatAgeMs(ageMs: number): string { + const seconds = Math.floor(ageMs / 1000); + if (seconds < 60) return `${seconds}s`; + const minutes = Math.floor(seconds / 60); + if (minutes < 60) return `${minutes}m`; + const hours = Math.floor(minutes / 60); + if (hours < 24) return `${hours}h${minutes % 60 ? `${minutes % 60}m` : ''}`; + const days = Math.floor(hours / 24); + return `${days}d${hours % 24 ? `${hours % 24}h` : ''}`; } /** diff --git a/packages/harness-driver/src/client.ts b/packages/harness-driver/src/client.ts index b5776f06f..970a58f4f 100644 --- a/packages/harness-driver/src/client.ts +++ b/packages/harness-driver/src/client.ts @@ -28,6 +28,8 @@ import type { BrokerStats, BrokerStatus, CrashInsightsResponse, + DeadLettersResponse, + RedeliverDeadLettersResponse, HeadlessProvider, PendingRelayMessage, PtySnapshot, @@ -935,6 +937,23 @@ export class HarnessDriverClient { return this.transport.request('/api/crash-insights'); } + /** List terminally-failed deliveries retained in the broker's dead-letter queue. */ + async getDeadLetters(): Promise { + return this.transport.request('/api/dead-letters'); + } + + /** + * Requeue dead-letter entries through the normal delivery path with a + * reset retry count. Pass an id for a single entry, or `{ all: true }` + * for every entry whose recipient is currently running. + */ + async redeliverDeadLetters(input: { id: string } | { all: true }): Promise { + return this.transport.request('/api/dead-letters/redeliver', { + method: 'POST', + body: JSON.stringify(input), + }); + } + // ── Lifecycle ────────────────────────────────────────────────────── async preflight(agents: Array<{ name: string; cli: string }>): Promise<{ queued: number }> { diff --git a/packages/harness-driver/src/protocol.ts b/packages/harness-driver/src/protocol.ts index c3c850d8d..929e28b60 100644 --- a/packages/harness-driver/src/protocol.ts +++ b/packages/harness-driver/src/protocol.ts @@ -180,9 +180,34 @@ export interface BrokerStatus { }>; pending_delivery_count: number; pending_deliveries: PendingDeliveryInfo[]; + dead_letter_count?: number; auth?: BrokerAuthStatus; } +/** A terminally-failed delivery retained in the broker's dead-letter queue. */ +export interface DeadLetterInfo { + delivery_id: string; + worker_name: string; + event_id: string; + from: string; + to: string; + attempts: number; + reason: string; + queued_at_ms: number; + failed_at_ms: number; + age_ms: number; +} + +export interface DeadLettersResponse { + count: number; + dead_letters: DeadLetterInfo[]; +} + +export interface RedeliverDeadLettersResponse { + redelivered: Array<{ delivery_id: string; worker_name: string; event_id: string }>; + skipped: Array<{ delivery_id: string; worker_name: string; reason: string }>; +} + export type AgentCurrentState = 'working' | 'idle' | 'blocked_on_send'; export interface BrokerAuthStatus { @@ -382,6 +407,22 @@ export type BrokerEvent = attempts: number; lastError: string; } + | { + kind: 'dead_letter_added'; + name: string; + delivery_id: string; + event_id: string; + from: string; + to: string; + attempts: number; + reason: string; + } + | { + kind: 'dead_letter_redelivered'; + name: string; + delivery_id: string; + event_id: string; + } | { kind: 'delivery_active'; name: string;