From 210bf839ebe31409a1d7d069cbacadcf3b9e0298 Mon Sep 17 00:00:00 2001 From: Miya Date: Sun, 16 Aug 2026 21:46:27 +0200 Subject: [PATCH 1/3] fix(broker): withhold fleet delivery_ack until worker confirms injection The engine-facing delivery_ack for a fleet (node-delivery) message fired the instant the deliver_relay frame was handed to the worker's write channel (fleet.rs handle_fleet_deliver / flush_pending_relay_messages), before the PTY worker had even attempted the write, let alone before echo verification confirmed the bytes landed. A delivery that later failed to inject or never echoed was reported to the engine as delivered regardless. Defer the engine ack from write-enqueue time to confirmation time: a new pending_fleet_acks table withholds the ack (commit_received still advances immediately so subsequent sequenced frames aren't seen as a gap; only commit_acked_receipt + the wire ack wait) and resolves it when the worker's own echo-verified (or bounded timeout-fallback) delivery_ack event arrives. A dead-lettered delivery drops the withheld entry instead of ever acking, matching this file's existing withhold-ack convention for other failure paths. The manual-flush path (flush_pending_relay_messages) is left as-is with a design note: its strictly-ordered cumulative ack cursor commits synchronously per item so a multi-item backlog drains in one call; deferring to async echo confirmation would stall the loop after the first item. Needs a pipelined ack-cursor model as a follow-up. Fixes #1310. Co-Authored-By: Claude Sonnet 5 Session-Id: 82b61cd7-f785-49b5-a29c-69a0c0e4f848 --- CHANGELOG.md | 6 +- crates/broker/src/runtime/delivery.rs | 24 ++- crates/broker/src/runtime/event_loop.rs | 8 + crates/broker/src/runtime/fleet.rs | 119 +++++++++++++-- crates/broker/src/runtime/init.rs | 1 + crates/broker/src/runtime/maintenance.rs | 2 + crates/broker/src/runtime/tests.rs | 168 +++++++++++++++++++++ crates/broker/src/runtime/worker_events.rs | 26 +++- 8 files changed, 333 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2044eb001..a9b1e3a9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- Fleet node delivery (`agent-relay node ...` engine-facing acks) no longer acknowledges a message to the engine the instant it's handed to a worker's PTY; the ack now waits for the worker to confirm the injection landed (echo-verified, or its bounded timeout fallback), so a delivery that never actually reaches the terminal no longer reports as delivered. ## [11.6.9] - 2026-08-16 diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index b15e190d9..ffb77ebe9 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -585,7 +585,7 @@ pub(crate) async fn try_inject_pending_relay_message( worker_name: &str, msg: &PendingRelayMessage, retry_interval: Duration, -) -> Result<()> { +) -> Result { let event_id = msg .event_id .clone() @@ -679,7 +679,7 @@ pub(crate) async fn queue_and_try_delivery_raw( priority: u8, injection_mode: MessageInjectionMode, retry_interval: Duration, -) -> Result<()> { +) -> Result { let delivery = RelayDelivery { delivery_id: DeliveryId::new(format!("del_{}", Uuid::new_v4().simple())), event_id: EventId::new(event_id), @@ -717,7 +717,7 @@ pub(crate) async fn queue_and_try_delivery_raw( pending_deliveries.insert(pending.delivery.delivery_id.clone(), *pending); anyhow::bail!(last_error); } - Ok(()) + Ok(delivery_id) } pub(crate) async fn retry_pending_delivery( @@ -813,6 +813,7 @@ pub(crate) fn delivery_ack_timeout( pub(crate) async fn emit_delivery_attempt_outcome( sdk_out_tx: &mpsc::Sender>, dead_letters: &mut DeadLetterStore, + pending_fleet_acks: &mut HashMap, delivery_id: &DeliveryId, was_retry: bool, outcome: DeliveryAttemptOutcome, @@ -857,6 +858,23 @@ pub(crate) async fn emit_delivery_attempt_outcome( }, ) .await; + // A dead-lettered delivery never actually landed, so any fleet + // (engine-facing) ack withheld pending its confirmation must be + // dropped rather than sent — the engine keeps its own record of + // this delivery as un-acked and will redeliver it. See relay#1310: + // the whole point of withholding the ack is that "enqueued for + // injection" must not be reported the same as "delivered". + if pending_fleet_acks + .remove(&pending.delivery.delivery_id) + .is_some() + { + tracing::info!( + target = "relay_broker::fleet", + worker = %pending.worker_name, + delivery_id = %pending.delivery.delivery_id, + "dropping withheld fleet delivery_ack for dead-lettered delivery" + ); + } dead_letter_pending_delivery(sdk_out_tx, dead_letters, &pending, &last_error).await; } DeliveryAttemptOutcome::Noop => {} diff --git a/crates/broker/src/runtime/event_loop.rs b/crates/broker/src/runtime/event_loop.rs index ba983f0fb..4f329736d 100644 --- a/crates/broker/src/runtime/event_loop.rs +++ b/crates/broker/src/runtime/event_loop.rs @@ -1,5 +1,6 @@ use super::*; +use crate::fleet_wire::Deliver; use futures_util::future::{join, join_all}; /// Current PTY resize owner for a worker under the single-resizer policy. @@ -245,6 +246,13 @@ pub(crate) struct BrokerRuntime { pub(super) pending_deliveries: PendingDeliveryStore, pub(super) dead_letters: DeadLetterStore, pub(super) terminal_failed_deliveries: HashSet, + /// Fleet (engine-facing) `delivery_ack`s withheld until the worker + /// confirms the PTY injection landed — echo-verified, or its bounded + /// timeout fallback (see `pty_worker.rs`'s `verification_window`) — rather + /// than acked the instant the write is merely handed to the worker. See + /// relay#1310. Resolved in `handle_worker_event`'s `delivery_ack` arm; + /// dropped without acking if the delivery is later dead-lettered. + pub(super) pending_fleet_acks: HashMap, pub(super) pending_requests: HashMap, /// Persona/capability spawns whose action result is held until the harness /// proves readiness with worker_ready. Keyed by the node-local worker name. diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 1e84bf869..be4be1ac0 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -127,9 +127,17 @@ pub(super) fn verified_spawn_failed_result(invocation_id: String, error: &str) - } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] enum FleetDeliverySurfaceOutcome { + /// Ack the engine now: no PTY injection occurred (ambient receipt/reaction + /// or an unrecognized payload type), so there is nothing to verify. Acknowledge, + /// A PTY injection was handed to the worker. The engine ack is withheld + /// until the worker confirms it landed (echo-verified, or the bounded + /// timeout fallback) — see relay#1310. The `DeliveryId` is the broker's + /// internal tracking id for that injection, used to correlate the + /// worker's later `delivery_ack` event back to this fleet delivery. + AcknowledgeAfterEcho(DeliveryId), HoldForManualFlush, } @@ -719,6 +727,20 @@ impl BrokerRuntime { Ok(FleetDeliverySurfaceOutcome::Acknowledge) => { self.fleet_delivery_book.commit_delivered(&deliver) } + Ok(FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho(injection_delivery_id)) => { + // Advance the *received* cursor now — the broker has taken + // ownership of this message and the engine's next sequenced + // frame must not be treated as a gap while injection is + // in flight. The *acked* cursor (and the engine-facing + // delivery_ack below) is withheld until + // `handle_worker_event` observes the worker actually + // confirm this specific injection (echo or timeout + // fallback) — see relay#1310. + self.fleet_delivery_book.commit_received(&deliver); + self.pending_fleet_acks + .insert(injection_delivery_id, deliver); + return; + } Ok(FleetDeliverySurfaceOutcome::HoldForManualFlush) => { self.fleet_delivery_book.commit_received(&deliver); return; @@ -912,11 +934,12 @@ impl BrokerRuntime { // engine to redeliver it); backlog injection failures // are logged and otherwise don't block the ack, since // their own delivery frames already governed their acks. - let mut current_result = Ok(()); + let mut current_result: Result<(), anyhow::Error> = Ok(()); + let mut current_injection_id: Option = None; for queued in to_drain { let is_current = queued.event_id.as_deref() == Some(deliver.msg_id.as_str()); - if let Err(error) = try_inject_pending_relay_message( + match try_inject_pending_relay_message( &mut self.workers, &mut self.pending_deliveries, &deliver.agent, @@ -925,20 +948,36 @@ impl BrokerRuntime { ) .await { - if is_current { - current_result = Err(error); - } else { - tracing::warn!( - target = "relay_broker::fleet", - agent = %deliver.agent, - from = %queued.from, - error = %error, - "failed to inject drained backlog message" - ); + Ok(injection_delivery_id) => { + if is_current { + current_injection_id = Some(injection_delivery_id); + } + } + Err(error) => { + if is_current { + current_result = Err(error); + } else { + tracing::warn!( + target = "relay_broker::fleet", + agent = %deliver.agent, + from = %queued.from, + error = %error, + "failed to inject drained backlog message" + ); + } } } } - current_result.map(|()| FleetDeliverySurfaceOutcome::Acknowledge) + current_result.and_then(|()| { + current_injection_id + .map(FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho) + .ok_or_else(|| { + anyhow::anyhow!( + "current delivery '{}' missing from drain batch", + deliver.msg_id + ) + }) + }) } InboundQueueOutcome::RejectedFull => anyhow::bail!( "manual delivery queue is full for '{}'; retaining Relaycast ownership", @@ -946,10 +985,15 @@ impl BrokerRuntime { ), InboundQueueOutcome::WorkerMissing => { let relay_delivery = self.fleet_relay_delivery(deliver); + let injection_delivery_id = relay_delivery.delivery_id.clone(); self.workers .deliver(&deliver.agent, relay_delivery) .await - .map(|()| FleetDeliverySurfaceOutcome::Acknowledge) + .map(|()| { + FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho( + injection_delivery_id, + ) + }) } } } @@ -1408,6 +1452,37 @@ fn fleet_spawn_outcome( } } +/// Resolve a fleet (engine-facing) `delivery_ack` withheld pending +/// confirmation of a specific PTY injection (relay#1310: the ack must not +/// fire before the worker confirms the write landed). Called with the +/// `delivery_id`/`event_id` from the worker's own internal `delivery_ack` +/// event (`worker_events.rs`), which pty_worker.rs sends only after echo +/// verification succeeds or its bounded timeout fallback fires — never at +/// write-enqueue time. +/// +/// Returns the `(agent, up_to_seq)` to send to the engine once resolved, or +/// `None` when there is nothing withheld for `delivery_id` (already +/// resolved, dead-lettered, or never a fleet-originated delivery) or when +/// `event_id` doesn't match the withheld delivery's `msg_id` — the same +/// stale/reused-id guard `clear_pending_delivery_if_event_matches` applies to +/// the broker's own pending-delivery map. +/// +/// Split out from `handle_fleet_deliver`/`handle_worker_event` so the ack +/// gating is testable without a whole `BrokerRuntime`. +pub(super) fn resolve_pending_fleet_ack( + pending_fleet_acks: &mut HashMap, + fleet_delivery_book: &mut FleetDeliveryBook, + delivery_id: &str, + event_id: &str, +) -> Option<(String, u64)> { + if pending_fleet_acks.get(delivery_id)?.msg_id != event_id { + return None; + } + let deliver = pending_fleet_acks.remove(delivery_id)?; + let up_to_seq = fleet_delivery_book.commit_delivered(&deliver); + Some((deliver.agent, up_to_seq)) +} + fn fleet_spawn_action_result( invocation_id: &str, name: &WorkerName, @@ -1438,6 +1513,20 @@ pub(super) struct FlushPendingRelayResult { /// Inject a worker's held queue in FIFO order. A failed item and every item /// behind it remain queued. Relaycast ACKs advance only after the corresponding /// PTY write succeeds, so the emitted cursor is always an injected prefix. +/// +/// relay#1310 design note: unlike `handle_fleet_deliver`'s `DrainNow` path, +/// this loop's ack timing is intentionally **not** deferred to echo +/// verification in this change. Each item's `can_ack_receipt` precondition +/// (`node_control.rs`) requires the *previous* item's ack to have already +/// committed, since `acked_up_to_seq` is a strictly ordered cumulative +/// cursor; committing here happens synchronously in this one loop so a +/// multi-item backlog drains in a single call. Deferring commit to async +/// echo confirmation would stall the loop after the first item — the second +/// item's `can_ack_receipt` check would never pass until the first item's +/// echo resolves — turning a batch drain into a call-per-item serialization. +/// A real fix needs an ack-cursor model that tolerates a pipeline of +/// outstanding (unconfirmed) commits, which is a separate, larger change; +/// this flush path still acks on write, not on echo, until that lands. pub(super) async fn flush_pending_relay_messages( delivery_states: &mut HashMap, workers: &mut WorkerRegistry, diff --git a/crates/broker/src/runtime/init.rs b/crates/broker/src/runtime/init.rs index b3aabc292..479593041 100644 --- a/crates/broker/src/runtime/init.rs +++ b/crates/broker/src/runtime/init.rs @@ -711,6 +711,7 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re pending_deliveries, dead_letters, terminal_failed_deliveries, + pending_fleet_acks: HashMap::new(), pending_requests, pending_verified_spawns, resize_owners: HashMap::new(), diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index b514dcab4..52f15429a 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -23,6 +23,7 @@ impl BrokerRuntime { let crash_insights = &mut self.crash_insights; let pending_deliveries = &mut self.pending_deliveries; let dead_letters = &mut self.dead_letters; + let pending_fleet_acks = &mut self.pending_fleet_acks; let pending_requests = &mut self.pending_requests; let pending_verified_spawns = &mut self.pending_verified_spawns; let delivery_states = &mut self.delivery_states; @@ -177,6 +178,7 @@ impl BrokerRuntime { let _ = emit_delivery_attempt_outcome( sdk_out_tx, dead_letters, + pending_fleet_acks, &delivery_id, was_retry, outcome, diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index c0615466f..b7385ca02 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -878,9 +878,11 @@ async fn retry_exhaustion_dead_letters_instead_of_discarding() { let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); let mut dead_letters = DeadLetterStore::default(); + let mut pending_fleet_acks = HashMap::new(); emit_delivery_attempt_outcome( &sdk_out_tx, &mut dead_letters, + &mut pending_fleet_acks, &DeliveryId::new("del_exhausted"), true, outcome, @@ -913,6 +915,170 @@ async fn retry_exhaustion_dead_letters_instead_of_discarding() { assert_eq!(dead_frame.payload["reason"], "failed writing frame"); } +// relay#1310 MUST-FIRE: a PTY delivery whose write never lands (retries +// exhaust, same as `retry_exhaustion_dead_letters_instead_of_discarding` +// above) must not resolve into an engine-facing delivery_ack. Before the +// fix, the fleet ack for a delivery like this had already been sent at +// write-enqueue time, regardless of what happened afterward — the engine's +// ledger would say "delivered" for a message that was in fact dead-lettered. +#[tokio::test] +async fn dead_lettered_delivery_drops_withheld_fleet_ack_without_sending_it() { + 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_never_echoed", "ghost"); + exhausted.attempts = MAX_DELIVERY_RETRIES; + exhausted.failed_attempts = MAX_DELIVERY_RETRIES; + exhausted.last_error = Some("failed writing frame".to_string()); + let mut pending_deliveries = + HashMap::from([(DeliveryId::new("del_never_echoed"), exhausted.clone())]); + + let outcome = retry_pending_delivery( + &DeliveryId::new("del_never_echoed"), + &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(); + let mut fleet_delivery_book = FleetDeliveryBook::default(); + let mut pending_fleet_acks = HashMap::from([( + DeliveryId::new("del_never_echoed"), + Deliver { + v: FLEET_WIRE_VERSION, + agent: "agent-a".to_string(), + agent_id: "agent-a-id".to_string(), + delivery_id: "del_never_echoed".to_string(), + msg_id: "evt_del_never_echoed".to_string(), + seq: 1, + mode: DeliveryMode::Wait, + payload: json!({}), + }, + )]); + + emit_delivery_attempt_outcome( + &sdk_out_tx, + &mut dead_letters, + &mut pending_fleet_acks, + &DeliveryId::new("del_never_echoed"), + true, + outcome, + ) + .await + .expect("terminal outcome should emit"); + + assert!( + !pending_fleet_acks.contains_key("del_never_echoed"), + "withheld fleet ack must be dropped when its delivery dead-letters" + ); + + // Even a stray/late worker delivery_ack arriving after the dead-letter + // must not conjure an engine ack for a delivery that never landed. + let late_resolution = super::fleet::resolve_pending_fleet_ack( + &mut pending_fleet_acks, + &mut fleet_delivery_book, + "del_never_echoed", + "evt_del_never_echoed", + ); + assert_eq!( + late_resolution, None, + "a dead-lettered delivery must never resolve into an engine ack" + ); + + // Drain the two telemetry events the dead-letter path emits so this test + // doesn't leak unread frames; only the fleet-ack behavior is asserted here. + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; +} + +// relay#1310 MUST-NOT-FIRE: once the worker confirms the injection landed +// (echo-verified, or its bounded timeout fallback — pty_worker.rs sends the +// same internal `delivery_ack` event either way), the engine ack must still +// fire, with the delivery's own (agent, up_to_seq) — i.e. the happy path is +// unchanged, just correctly gated on confirmation instead of write-enqueue. +#[tokio::test] +async fn echo_confirmed_delivery_resolves_its_withheld_fleet_ack() { + let mut fleet_delivery_book = FleetDeliveryBook::default(); + let mut pending_fleet_acks = HashMap::from([( + DeliveryId::new("del_landed"), + Deliver { + v: FLEET_WIRE_VERSION, + agent: "agent-a".to_string(), + agent_id: "agent-a-id".to_string(), + delivery_id: "del_landed".to_string(), + msg_id: "evt_landed".to_string(), + seq: 1, + mode: DeliveryMode::Wait, + payload: json!({}), + }, + )]); + + let resolved = super::fleet::resolve_pending_fleet_ack( + &mut pending_fleet_acks, + &mut fleet_delivery_book, + "del_landed", + "evt_landed", + ); + + assert_eq!(resolved, Some(("agent-a".to_string(), 1))); + assert!( + !pending_fleet_acks.contains_key("del_landed"), + "a resolved ack must not remain withheld" + ); + + // A second confirmation for the same delivery_id (duplicate/replayed + // worker event) must not double-resolve — nothing is left to withhold. + let duplicate = super::fleet::resolve_pending_fleet_ack( + &mut pending_fleet_acks, + &mut fleet_delivery_book, + "del_landed", + "evt_landed", + ); + assert_eq!(duplicate, None); +} + +// A worker delivery_ack whose event_id doesn't match the withheld delivery's +// msg_id (stale or reused delivery_id) must not resolve into an engine ack — +// the same stale-event guard `clear_pending_delivery_if_event_matches` +// applies to the broker's own pending-delivery map. +#[tokio::test] +async fn mismatched_event_id_does_not_resolve_a_withheld_fleet_ack() { + let mut fleet_delivery_book = FleetDeliveryBook::default(); + let mut pending_fleet_acks = HashMap::from([( + DeliveryId::new("del_reused"), + Deliver { + v: FLEET_WIRE_VERSION, + agent: "agent-a".to_string(), + agent_id: "agent-a-id".to_string(), + delivery_id: "del_reused".to_string(), + msg_id: "evt_original".to_string(), + seq: 1, + mode: DeliveryMode::Wait, + payload: json!({}), + }, + )]); + + let resolved = super::fleet::resolve_pending_fleet_ack( + &mut pending_fleet_acks, + &mut fleet_delivery_book, + "del_reused", + "evt_stale", + ); + + assert_eq!(resolved, None); + assert!( + pending_fleet_acks.contains_key("del_reused"), + "a mismatched event must not consume the withheld entry" + ); +} + #[tokio::test] async fn delivery_retry_fails_promptly_when_recipient_is_gone() { let (tx, _rx) = mpsc::channel::(16); @@ -1124,9 +1290,11 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); let mut dead_letters = DeadLetterStore::default(); + let mut pending_fleet_acks = HashMap::new(); emit_delivery_attempt_outcome( &sdk_out_tx, &mut dead_letters, + &mut pending_fleet_acks, &DeliveryId::new("del_blip"), true, outcome, diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 70073e796..daf1fcc88 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -1,8 +1,9 @@ use super::fleet::{ - fail_terminal_session, refresh_fleet_inventory_session_ref, try_send_terminal, - verified_spawn_ready_result, + fail_terminal_session, refresh_fleet_inventory_session_ref, resolve_pending_fleet_ack, + try_send_terminal, verified_spawn_ready_result, }; use super::*; +use crate::node_control::delivery_ack; use crate::terminal_control::{TerminalControlCommand, TerminalToCloud}; use crate::worker::AgentWorkState; @@ -619,6 +620,8 @@ impl BrokerRuntime { let pending_verified_spawns = &mut self.pending_verified_spawns; let delivery_retry_interval = self.delivery_retry_interval; let fleet_control_tx = &self.fleet_control_tx; + let fleet_delivery_book = &mut self.fleet_delivery_book; + let pending_fleet_acks = &mut self.pending_fleet_acks; let fleet_inventory = &mut self.fleet_inventory; let delivery_states = &self.delivery_states; let terminal_control_tx = &self.terminal_control_tx; @@ -725,6 +728,25 @@ impl BrokerRuntime { if pending.is_some() { terminal_failed_deliveries.remove(&ack.delivery_id); } + + // Resolve a fleet (engine-facing) ack withheld + // pending confirmation of this exact PTY + // injection (relay#1310). No-op when nothing + // is withheld for this delivery_id, or when + // event_id doesn't match (stale/reused id). + if let Some((agent, up_to_seq)) = resolve_pending_fleet_ack( + pending_fleet_acks, + fleet_delivery_book, + ack.delivery_id.as_str(), + ack.event_id.as_str(), + ) { + let _ = fleet_control_tx + .send(FleetControlCommand::Send(delivery_ack( + agent, up_to_seq, + ))) + .await; + } + pending } else { None From ec659a58053b80ee27e83780071c7a0bb8a21352 Mon Sep 17 00:00:00 2001 From: Miya Date: Mon, 17 Aug 2026 01:13:46 +0200 Subject: [PATCH 2/3] fix(broker): tie withheld fleet acks to delivery lifetime, not a second map MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit relay#1543 review (chief's ruling, PR comment 5309883312): five of six threads were one defect, not five — pending_fleet_acks was a second map maintained in parallel with pending_deliveries, with nothing guaranteeing the two agreed. Every failure/teardown path that disposed of a PendingDelivery left its withheld fleet ack behind. Structural fix: withheld_fleet_ack now lives as a field on PendingDelivery itself, embedded synchronously at insertion time. Whatever disposes of a delivery (echo confirmation, dead-letter, worker teardown) disposes of its withheld ack with it, by construction, through the two existing choke points (emit_delivery_attempt_outcome, emit_dropped_delivery_failures) instead of a third cleanup site. This also fixes delivery.rs:588 (the P1 blocker): the ack is now embedded before the handoff attempt starts, so a handoff that outlives retry_interval can no longer lose the ack registration to the timeout — previously it was registered as a caller-side follow-up step keyed off a return value the timeout could swallow. The WorkerMissing fleet injection path (fleet.rs) now routes through the same insert_and_attempt_delivery helper as DrainNow, so it also gets a PendingDelivery entry and the same teardown guarantees, instead of a one-shot workers.deliver() outside pending_deliveries. Same shape as relay#1539: one unconditional writer, a second map maintained in parallel, nothing guaranteeing they agree. Co-Authored-By: Claude Sonnet 5 Session-Id: c7735d76-e1f3-40cf-a765-240981af72f7 --- CHANGELOG.md | 2 +- crates/broker/src/runtime/dead_letter.rs | 4 + crates/broker/src/runtime/delivery.rs | 82 +++- crates/broker/src/runtime/event_loop.rs | 8 - crates/broker/src/runtime/fleet.rs | 93 ++-- crates/broker/src/runtime/init.rs | 1 - crates/broker/src/runtime/maintenance.rs | 2 - crates/broker/src/runtime/tests.rs | 517 +++++++++++++++------ crates/broker/src/runtime/worker_events.rs | 19 +- 9 files changed, 527 insertions(+), 201 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a9b1e3a9f..f8a97360f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Fleet node delivery (`agent-relay node ...` engine-facing acks) no longer acknowledges a message to the engine the instant it's handed to a worker's PTY; the ack now waits for the worker to confirm the injection landed (echo-verified, or its bounded timeout fallback), so a delivery that never actually reaches the terminal no longer reports as delivered. +- Fleet node delivery no longer reports a message as delivered to the engine until the worker actually receives it, instead of at the moment it's handed off. ## [11.6.9] - 2026-08-16 diff --git a/crates/broker/src/runtime/dead_letter.rs b/crates/broker/src/runtime/dead_letter.rs index 8866590d0..e22a2b6d0 100644 --- a/crates/broker/src/runtime/dead_letter.rs +++ b/crates/broker/src/runtime/dead_letter.rs @@ -207,6 +207,10 @@ pub(crate) fn requeue_dead_letter( next_retry_at: Instant::now(), queued_at_ms: entry.queued_at_ms, last_error: None, + // The dead-lettered entry's withheld fleet ack (if any) was already + // dropped when it was dead-lettered — see relay#1310. A requeue is a + // fresh redelivery attempt, not a continuation of that withheld ack. + withheld_fleet_ack: None, }; pending_deliveries.insert(pending.delivery.delivery_id.clone(), pending.clone()); Some(pending) diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index ffb77ebe9..96f721866 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -12,6 +12,18 @@ pub(crate) struct PendingDelivery { pub(super) next_retry_at: Instant, pub(super) queued_at_ms: u64, pub(super) last_error: Option, + /// Fleet (engine-facing) `delivery_ack` withheld until the worker confirms + /// this specific PTY injection landed — echo-verified, or its bounded + /// timeout fallback — rather than acked the instant the write is merely + /// handed to the worker. See relay#1310. + /// + /// Lives on the `PendingDelivery` itself, not a second map keyed by + /// `DeliveryId`, so it cannot outlive the delivery it belongs to: every + /// path that disposes of a `PendingDelivery` (echo confirmation, dead + /// letter, worker teardown) disposes of its withheld ack with it, by + /// construction, instead of needing a matching removal remembered at + /// every one of those call sites. See relay#1543. + pub(super) withheld_fleet_ack: Option, } /// Serializable snapshot of pending deliveries for crash recovery. @@ -171,6 +183,11 @@ pub(crate) fn load_pending_deliveries(path: &Path) -> HashMap, ) -> Result { let event_id = msg .event_id @@ -610,6 +636,7 @@ pub(crate) async fn try_inject_pending_relay_message( msg.priority, msg.mode.clone(), retry_interval, + withheld_fleet_ack, ), ) .await @@ -679,6 +706,7 @@ pub(crate) async fn queue_and_try_delivery_raw( priority: u8, injection_mode: MessageInjectionMode, retry_interval: Duration, + withheld_fleet_ack: Option, ) -> Result { let delivery = RelayDelivery { delivery_id: DeliveryId::new(format!("del_{}", Uuid::new_v4().simple())), @@ -692,6 +720,34 @@ pub(crate) async fn queue_and_try_delivery_raw( priority: Some(priority), injection_mode, }; + insert_and_attempt_delivery( + workers, + pending_deliveries, + worker_name, + delivery, + retry_interval, + withheld_fleet_ack, + ) + .await +} + +/// Register a delivery and make its first handoff attempt, in one atomic +/// step: the `PendingDelivery` — including any withheld fleet ack — is +/// inserted into `pending_deliveries` before the handoff attempt starts, so +/// a slow or cancelled attempt can never separate "this delivery exists and +/// is retryable" from "its withheld ack is registered". Shared by the +/// broker-generated-id path (`queue_and_try_delivery_raw`) and any caller +/// that already has a fully-built [`RelayDelivery`] (the fleet +/// `WorkerMissing` injection path, which must keep the engine's own +/// `delivery_id`). +pub(crate) async fn insert_and_attempt_delivery( + workers: &mut WorkerRegistry, + pending_deliveries: &mut HashMap, + worker_name: &str, + delivery: RelayDelivery, + retry_interval: Duration, + withheld_fleet_ack: Option, +) -> Result { let delivery_id = delivery.delivery_id.clone(); pending_deliveries.insert( delivery_id.clone(), @@ -703,6 +759,7 @@ pub(crate) async fn queue_and_try_delivery_raw( next_retry_at: Instant::now(), queued_at_ms: unix_timestamp_millis(), last_error: None, + withheld_fleet_ack, }, ); @@ -813,7 +870,6 @@ pub(crate) fn delivery_ack_timeout( pub(crate) async fn emit_delivery_attempt_outcome( sdk_out_tx: &mpsc::Sender>, dead_letters: &mut DeadLetterStore, - pending_fleet_acks: &mut HashMap, delivery_id: &DeliveryId, was_retry: bool, outcome: DeliveryAttemptOutcome, @@ -863,11 +919,11 @@ pub(crate) async fn emit_delivery_attempt_outcome( // dropped rather than sent — the engine keeps its own record of // this delivery as un-acked and will redeliver it. See relay#1310: // the whole point of withholding the ack is that "enqueued for - // injection" must not be reported the same as "delivered". - if pending_fleet_acks - .remove(&pending.delivery.delivery_id) - .is_some() - { + // injection" must not be reported the same as "delivered". The + // withheld ack lives on `pending` itself, so it is dropped here + // simply by `pending` going out of scope — nothing to remember to + // clean up separately. See relay#1543. + if pending.withheld_fleet_ack.is_some() { tracing::info!( target = "relay_broker::fleet", worker = %pending.worker_name, @@ -906,6 +962,11 @@ pub(crate) fn take_pending_for_worker( .collect() } +/// Choke point for every worker-exit / teardown disposition (agent release, +/// permanent worker death, unsupervised exit): whatever removed these +/// `PendingDelivery`s from `pending_deliveries` (via [`take_pending_for_worker`]) +/// already carried their withheld fleet acks along as a struct field, so this +/// is also the single place that drops them. See relay#1543. pub(crate) async fn emit_dropped_delivery_failures( sdk_out_tx: &mpsc::Sender>, dead_letters: &mut DeadLetterStore, @@ -913,6 +974,15 @@ pub(crate) async fn emit_dropped_delivery_failures( reason: &str, ) -> Result<()> { for pending in dropped { + if pending.withheld_fleet_ack.is_some() { + tracing::info!( + target = "relay_broker::fleet", + worker = %pending.worker_name, + delivery_id = %pending.delivery.delivery_id, + reason = reason, + "dropping withheld fleet delivery_ack for a delivery dropped from the pending map" + ); + } // Notify best-effort: a send failure must not `?`-abort the loop and // strand the remaining dropped deliveries out of the dead-letter store. // The DLQ capture below runs regardless of the send's outcome. diff --git a/crates/broker/src/runtime/event_loop.rs b/crates/broker/src/runtime/event_loop.rs index 4f329736d..ba983f0fb 100644 --- a/crates/broker/src/runtime/event_loop.rs +++ b/crates/broker/src/runtime/event_loop.rs @@ -1,6 +1,5 @@ use super::*; -use crate::fleet_wire::Deliver; use futures_util::future::{join, join_all}; /// Current PTY resize owner for a worker under the single-resizer policy. @@ -246,13 +245,6 @@ pub(crate) struct BrokerRuntime { pub(super) pending_deliveries: PendingDeliveryStore, pub(super) dead_letters: DeadLetterStore, pub(super) terminal_failed_deliveries: HashSet, - /// Fleet (engine-facing) `delivery_ack`s withheld until the worker - /// confirms the PTY injection landed — echo-verified, or its bounded - /// timeout fallback (see `pty_worker.rs`'s `verification_window`) — rather - /// than acked the instant the write is merely handed to the worker. See - /// relay#1310. Resolved in `handle_worker_event`'s `delivery_ack` arm; - /// dropped without acking if the delivery is later dead-lettered. - pub(super) pending_fleet_acks: HashMap, pub(super) pending_requests: HashMap, /// Persona/capability spawns whose action result is held until the harness /// proves readiness with worker_ready. Keyed by the node-local worker name. diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index be4be1ac0..ead8b18c8 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -134,10 +134,10 @@ enum FleetDeliverySurfaceOutcome { Acknowledge, /// A PTY injection was handed to the worker. The engine ack is withheld /// until the worker confirms it landed (echo-verified, or the bounded - /// timeout fallback) — see relay#1310. The `DeliveryId` is the broker's - /// internal tracking id for that injection, used to correlate the - /// worker's later `delivery_ack` event back to this fleet delivery. - AcknowledgeAfterEcho(DeliveryId), + /// timeout fallback) — see relay#1310. The withheld ack was already + /// registered on the corresponding `PendingDelivery` at insertion time + /// (see relay#1543), so there is nothing left to carry here. + AcknowledgeAfterEcho, HoldForManualFlush, } @@ -727,7 +727,7 @@ impl BrokerRuntime { Ok(FleetDeliverySurfaceOutcome::Acknowledge) => { self.fleet_delivery_book.commit_delivered(&deliver) } - Ok(FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho(injection_delivery_id)) => { + Ok(FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho) => { // Advance the *received* cursor now — the broker has taken // ownership of this message and the engine's next sequenced // frame must not be treated as a gap while injection is @@ -736,9 +736,14 @@ impl BrokerRuntime { // `handle_worker_event` observes the worker actually // confirm this specific injection (echo or timeout // fallback) — see relay#1310. + // + // The withheld ack itself was already registered on the + // `PendingDelivery` at insertion time, before this + // injection attempt even started (see + // `try_inject_pending_relay_message` / + // `insert_and_attempt_delivery`), so there is nothing left + // to record here — see relay#1543. self.fleet_delivery_book.commit_received(&deliver); - self.pending_fleet_acks - .insert(injection_delivery_id, deliver); return; } Ok(FleetDeliverySurfaceOutcome::HoldForManualFlush) => { @@ -935,22 +940,27 @@ impl BrokerRuntime { // are logged and otherwise don't block the ack, since // their own delivery frames already governed their acks. let mut current_result: Result<(), anyhow::Error> = Ok(()); - let mut current_injection_id: Option = None; + let mut current_injected = false; for queued in to_drain { let is_current = queued.event_id.as_deref() == Some(deliver.msg_id.as_str()); + // Only the current delivery's own frame carries the + // withheld engine ack forward — backlog messages + // drained alongside it are governed by their own + // delivery frames (see the comment above this loop). match try_inject_pending_relay_message( &mut self.workers, &mut self.pending_deliveries, &deliver.agent, &queued, self.delivery_retry_interval, + is_current.then(|| deliver.clone()), ) .await { - Ok(injection_delivery_id) => { + Ok(_delivery_id) => { if is_current { - current_injection_id = Some(injection_delivery_id); + current_injected = true; } } Err(error) => { @@ -969,14 +979,14 @@ impl BrokerRuntime { } } current_result.and_then(|()| { - current_injection_id - .map(FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho) - .ok_or_else(|| { - anyhow::anyhow!( - "current delivery '{}' missing from drain batch", - deliver.msg_id - ) - }) + if current_injected { + Ok(FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho) + } else { + Err(anyhow::anyhow!( + "current delivery '{}' missing from drain batch", + deliver.msg_id + )) + } }) } InboundQueueOutcome::RejectedFull => anyhow::bail!( @@ -984,16 +994,26 @@ impl BrokerRuntime { deliver.agent ), InboundQueueOutcome::WorkerMissing => { + // Route through the same pending-delivery/retry path as + // `DrainNow` (`insert_and_attempt_delivery`) instead of + // a bare one-shot `workers.deliver`, so this injection + // also gets a `PendingDelivery` entry: its withheld ack + // is registered at insertion time and is guaranteed to + // be cleaned up by worker-teardown / retry-exhaustion + // paths if the worker disappears before echoing. A + // one-shot `workers.deliver` outside `pending_deliveries` + // had no such guarantee — see relay#1543. let relay_delivery = self.fleet_relay_delivery(deliver); - let injection_delivery_id = relay_delivery.delivery_id.clone(); - self.workers - .deliver(&deliver.agent, relay_delivery) - .await - .map(|()| { - FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho( - injection_delivery_id, - ) - }) + insert_and_attempt_delivery( + &mut self.workers, + &mut self.pending_deliveries, + &deliver.agent, + relay_delivery, + self.delivery_retry_interval, + Some(deliver.clone()), + ) + .await + .map(|_delivery_id| FleetDeliverySurfaceOutcome::AcknowledgeAfterEcho) } } } @@ -1467,20 +1487,21 @@ fn fleet_spawn_outcome( /// stale/reused-id guard `clear_pending_delivery_if_event_matches` applies to /// the broker's own pending-delivery map. /// +/// The withheld ack itself lives on the `PendingDelivery` +/// (`withheld_fleet_ack`, see relay#1543), so the delivery_id/event_id +/// matching that used to happen against a second map here is already done by +/// `clear_pending_delivery_if_event_matches` before this is called — this +/// only extracts what that lookup found and commits it to the delivery book. +/// /// Split out from `handle_fleet_deliver`/`handle_worker_event` so the ack /// gating is testable without a whole `BrokerRuntime`. pub(super) fn resolve_pending_fleet_ack( - pending_fleet_acks: &mut HashMap, + pending: Option<&PendingDelivery>, fleet_delivery_book: &mut FleetDeliveryBook, - delivery_id: &str, - event_id: &str, ) -> Option<(String, u64)> { - if pending_fleet_acks.get(delivery_id)?.msg_id != event_id { - return None; - } - let deliver = pending_fleet_acks.remove(delivery_id)?; - let up_to_seq = fleet_delivery_book.commit_delivered(&deliver); - Some((deliver.agent, up_to_seq)) + let deliver = pending?.withheld_fleet_ack.as_ref()?; + let up_to_seq = fleet_delivery_book.commit_delivered(deliver); + Some((deliver.agent.clone(), up_to_seq)) } fn fleet_spawn_action_result( diff --git a/crates/broker/src/runtime/init.rs b/crates/broker/src/runtime/init.rs index 479593041..b3aabc292 100644 --- a/crates/broker/src/runtime/init.rs +++ b/crates/broker/src/runtime/init.rs @@ -711,7 +711,6 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re pending_deliveries, dead_letters, terminal_failed_deliveries, - pending_fleet_acks: HashMap::new(), pending_requests, pending_verified_spawns, resize_owners: HashMap::new(), diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 52f15429a..b514dcab4 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -23,7 +23,6 @@ impl BrokerRuntime { let crash_insights = &mut self.crash_insights; let pending_deliveries = &mut self.pending_deliveries; let dead_letters = &mut self.dead_letters; - let pending_fleet_acks = &mut self.pending_fleet_acks; let pending_requests = &mut self.pending_requests; let pending_verified_spawns = &mut self.pending_verified_spawns; let delivery_states = &mut self.delivery_states; @@ -178,7 +177,6 @@ impl BrokerRuntime { let _ = emit_delivery_attempt_outcome( sdk_out_tx, dead_letters, - pending_fleet_acks, &delivery_id, was_retry, outcome, diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index b7385ca02..5d2710279 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -50,11 +50,11 @@ use super::{ relaycast_ws_should_apply_local_spawn_echo_dedup, relaycast_ws_spawn_token, requeue_dead_letter, resolve_exit_after_task, resolve_workspace, retry_pending_delivery, save_dead_letters, seed_supplied_agent_token, send_broker_event, sender_is_dashboard_label, - should_clear_pending_delivery_for_event, synthetic_delivery_read_ack_reason, AgentRuntime, - DeadLetterEntry, DeadLetterStore, DeliveryAttemptOutcome, InboundContext, InboundQueueOutcome, - ObserverTokenMintError, ObserverTokenMintOutcome, PendingDelivery, PendingDeliveryStore, - ProtocolHeadlessProvider, RelayWorkspace, TypedThreadMessage, MAX_DEAD_LETTERS, - MAX_DELIVERY_RETRIES, + should_clear_pending_delivery_for_event, synthetic_delivery_read_ack_reason, + try_inject_pending_relay_message, AgentRuntime, DeadLetterEntry, DeadLetterStore, + DeliveryAttemptOutcome, InboundContext, InboundQueueOutcome, ObserverTokenMintError, + ObserverTokenMintOutcome, PendingDelivery, PendingDeliveryStore, ProtocolHeadlessProvider, + RelayWorkspace, TypedThreadMessage, MAX_DEAD_LETTERS, MAX_DELIVERY_RETRIES, }; use crate::dedup::DedupCache; use crate::relaycast::{ @@ -126,6 +126,67 @@ async fn make_worker_registry_with_worker(name: &str) -> WorkerRegistry { registry } +/// A worker whose command channel accepts frames but never completes them — +/// no writer task ever drains `command_rx`, so `deliver()` hangs forever. +/// Models a handoff that outlives `retry_interval` deterministically (no +/// timing race): the receiver stays alive (a dropped one would fail the +/// send instead of hanging it), so the send always succeeds and the +/// subsequent completion wait never returns on its own. +async fn make_worker_registry_with_stalled_worker(name: &str) -> WorkerRegistry { + let (tx, _rx) = mpsc::channel::(16); + let mut registry = WorkerRegistry::new( + tx, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests"), + Instant::now(), + ); + let child = tokio::process::Command::new("cat") + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("test worker process should spawn"); + let generation = Uuid::new_v4(); + let (command_tx, command_rx) = mpsc::channel(128); + // Deliberately leaked, not spawned as a writer: keeps the receiver alive + // (so sends succeed) without anything ever draining it. + std::mem::forget(command_rx); + registry.workers.insert( + WorkerName::from(name), + WorkerHandle { + generation, + spec: AgentSpec { + name: WorkerName::from(name), + runtime: AgentRuntime::Pty, + provider: None, + cli: Some("cat".to_string()), + session_id: None, + harness_config: None, + model: None, + cwd: None, + team: None, + shadow_of: None, + shadow_mode: None, + args: Vec::new(), + channels: Vec::new(), + restart_policy: None, + }, + parent: None, + workspace_id: Some(WorkspaceId::new("ws_demo")), + child, + command_tx, + harness_pid: None, + spawned_at: Instant::now(), + ready_at: Some(Instant::now()), + last_activity_at: Instant::now(), + context_budget_pct: None, + state: AgentWorkState::Working, + exit_reason: None, + }, + ); + registry +} + async fn cleanup_worker_registry(mut registry: WorkerRegistry) { for handle in registry.workers.values_mut() { let _ = handle.child.start_kill(); @@ -203,6 +264,7 @@ fn pending_delivery(worker_name: &str, delivery_id: &str, event_id: &str) -> Pen next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, } } @@ -495,6 +557,7 @@ fn make_pending_delivery(delivery_id: &str, worker: &str) -> PendingDelivery { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, } } @@ -878,11 +941,9 @@ async fn retry_exhaustion_dead_letters_instead_of_discarding() { let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); let mut dead_letters = DeadLetterStore::default(); - let mut pending_fleet_acks = HashMap::new(); emit_delivery_attempt_outcome( &sdk_out_tx, &mut dead_letters, - &mut pending_fleet_acks, &DeliveryId::new("del_exhausted"), true, outcome, @@ -915,87 +976,260 @@ async fn retry_exhaustion_dead_letters_instead_of_discarding() { assert_eq!(dead_frame.payload["reason"], "failed writing frame"); } -// relay#1310 MUST-FIRE: a PTY delivery whose write never lands (retries -// exhaust, same as `retry_exhaustion_dead_letters_instead_of_discarding` -// above) must not resolve into an engine-facing delivery_ack. Before the -// fix, the fleet ack for a delivery like this had already been sent at -// write-enqueue time, regardless of what happened afterward — the engine's -// ledger would say "delivered" for a message that was in fact dead-lettered. -#[tokio::test] -async fn dead_lettered_delivery_drops_withheld_fleet_ack_without_sending_it() { - 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_never_echoed", "ghost"); - exhausted.attempts = MAX_DELIVERY_RETRIES; - exhausted.failed_attempts = MAX_DELIVERY_RETRIES; - exhausted.last_error = Some("failed writing frame".to_string()); - let mut pending_deliveries = - HashMap::from([(DeliveryId::new("del_never_echoed"), exhausted.clone())]); - - let outcome = retry_pending_delivery( - &DeliveryId::new("del_never_echoed"), - &mut workers, - &mut pending_deliveries, - Duration::from_millis(1), - ) - .await - .expect("exhausted retries should classify as terminal failure"); +fn withheld_ack_for(delivery_id: &str) -> Deliver { + Deliver { + v: FLEET_WIRE_VERSION, + agent: "agent-a".to_string(), + agent_id: "agent-a-id".to_string(), + delivery_id: delivery_id.to_string(), + msg_id: format!("evt_{delivery_id}"), + seq: 1, + mode: DeliveryMode::Wait, + payload: json!({}), + } +} - let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); - let mut dead_letters = DeadLetterStore::default(); - let mut fleet_delivery_book = FleetDeliveryBook::default(); - let mut pending_fleet_acks = HashMap::from([( - DeliveryId::new("del_never_echoed"), - Deliver { - v: FLEET_WIRE_VERSION, - agent: "agent-a".to_string(), - agent_id: "agent-a-id".to_string(), - delivery_id: "del_never_echoed".to_string(), - msg_id: "evt_del_never_echoed".to_string(), - seq: 1, - mode: DeliveryMode::Wait, - payload: json!({}), - }, - )]); +// relay#1543 delivery.rs:588 MUST-FIRE (P1, blocker): when the initial +// worker handoff outlives `retry_interval`, the withheld fleet ack must +// already be registered on the `PendingDelivery` — not dependent on the +// timed-out call's `Ok(DeliveryId)` ever reaching the caller. Before the +// fix, `try_inject_pending_relay_message` returned only a bare `Result` +// derived from the timed-out future, and the fleet caller registered the +// withheld ack as a *separate* follow-up step keyed off that return value; +// a handoff that timed out returned `Err`, so the ack was simply never +// registered even though the delivery itself remained alive and retryable — +// a later successful retry's echo would then have had nothing to resolve. +#[tokio::test] +async fn timed_out_initial_handoff_still_registers_its_withheld_fleet_ack() { + let mut registry = make_worker_registry_with_stalled_worker("worker-a").await; + let deliver = fleet_deliver(1); + let msg = held_fleet_message(&deliver); + let mut pending_deliveries = HashMap::new(); - emit_delivery_attempt_outcome( - &sdk_out_tx, - &mut dead_letters, - &mut pending_fleet_acks, - &DeliveryId::new("del_never_echoed"), - true, - outcome, + let outcome = tokio::time::timeout( + Duration::from_secs(5), + try_inject_pending_relay_message( + &mut registry, + &mut pending_deliveries, + "worker-a", + &msg, + Duration::from_millis(20), + Some(deliver.clone()), + ), ) .await - .expect("terminal outcome should emit"); + .expect( + "the test's own generous bound must never fire — only the short \ + retry_interval passed to try_inject_pending_relay_message should", + ); assert!( - !pending_fleet_acks.contains_key("del_never_echoed"), - "withheld fleet ack must be dropped when its delivery dead-letters" + outcome.is_err(), + "a handoff that never completes must time out, not hang forever" ); - - // Even a stray/late worker delivery_ack arriving after the dead-letter - // must not conjure an engine ack for a delivery that never landed. - let late_resolution = super::fleet::resolve_pending_fleet_ack( - &mut pending_fleet_acks, - &mut fleet_delivery_book, - "del_never_echoed", - "evt_del_never_echoed", + assert_eq!( + pending_deliveries.len(), + 1, + "the delivery must remain registered and retryable even though the initial handoff timed out" ); + let registered = pending_deliveries + .values() + .next() + .expect("checked len() == 1 above"); assert_eq!( - late_resolution, None, - "a dead-lettered delivery must never resolve into an engine ack" + registered + .withheld_fleet_ack + .as_ref() + .map(|d| d.msg_id.as_str()), + Some(deliver.msg_id.as_str()), + "the withheld fleet ack must be registered before the timeout can expire, so a later \ + successful retry can still resolve it" ); - // Drain the two telemetry events the dead-letter path emits so this test - // doesn't leak unread frames; only the fleet-ack behavior is asserted here. - let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; - let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + cleanup_worker_registry(registry).await; +} + +// relay#1543 MUST-FIRE (parameterised over every terminal disposition): a +// `PendingDelivery`'s withheld fleet ack must never survive the delivery it +// belongs to. Before the structural fix, `pending_fleet_acks` was a second +// map that none of these dispositions — except one very manually-threaded +// retry-exhaustion call site — knew to clean up. The ack now lives on +// `PendingDelivery` itself, so every path that disposes of the delivery +// disposes of the ack with it, by construction. +#[tokio::test] +async fn every_terminal_disposition_drops_its_withheld_fleet_ack() { + // Disposition 1: retry-exhaustion dead-letter (delivery.rs:816's thread) + // — the `emit_delivery_attempt_outcome` `Failed` arm. + { + 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_ack", "ghost"); + exhausted.attempts = MAX_DELIVERY_RETRIES; + exhausted.failed_attempts = MAX_DELIVERY_RETRIES; + exhausted.withheld_fleet_ack = Some(withheld_ack_for("del_exhausted_ack")); + let mut pending_deliveries = + HashMap::from([(DeliveryId::new("del_exhausted_ack"), exhausted)]); + + let outcome = retry_pending_delivery( + &DeliveryId::new("del_exhausted_ack"), + &mut workers, + &mut pending_deliveries, + Duration::from_millis(1), + ) + .await + .expect("exhausted retries should classify as terminal failure"); + match &outcome { + DeliveryAttemptOutcome::Failed { pending, .. } => assert!( + pending.withheld_fleet_ack.is_some(), + "fixture must carry a withheld ack for this case to be meaningful" + ), + other => panic!("expected terminal failure, got {other:?}"), + } + + 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_ack"), + true, + outcome, + ) + .await + .expect("terminal outcome should emit"); + + assert!(!pending_deliveries.contains_key("del_exhausted_ack")); + let mut book = FleetDeliveryBook::default(); + assert_eq!( + super::fleet::resolve_pending_fleet_ack( + pending_deliveries.get("del_exhausted_ack"), + &mut book + ), + None, + "a retry-exhausted delivery must never resolve into an engine ack" + ); + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + } + + // Dispositions 2 & 3: worker-exit and `delivery_failed` both dispose of a + // `PendingDelivery` via `emit_dropped_delivery_failures` — the single + // choke point every worker-teardown path (`take_pending_for_worker`, + // maintenance.rs:26 / event_loop.rs:255's threads) and the + // `delivery_failed` worker-event path share. + for reason in ["worker_exited", "delivery_failed"] { + let mut pending = make_pending_delivery("del_dropped_ack", "ghost"); + pending.withheld_fleet_ack = Some(withheld_ack_for("del_dropped_ack")); + let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); + let mut dead_letters = DeadLetterStore::default(); + emit_dropped_delivery_failures(&sdk_out_tx, &mut dead_letters, &[pending], reason) + .await + .expect("dropped delivery outcome should emit"); + + let mut book = FleetDeliveryBook::default(); + assert_eq!( + super::fleet::resolve_pending_fleet_ack(None, &mut book), + None, + "a delivery dropped for {reason} must never resolve into an engine ack" + ); + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + } + + // Disposition 4: a `WorkerMissing` fleet injection whose recipient never + // existed (fleet.rs:741's thread) — before the fix this injected via a + // bare `workers.deliver` call outside `pending_deliveries`, so nothing + // ever tracked its withheld ack at all. Routed through + // `insert_and_attempt_delivery` like `DrainNow`, it is tracked from the + // first attempt and reaches the exact same terminal cleanup as every + // other disposition above. + { + let (tx, _rx) = mpsc::channel::(16); + let mut workers = WorkerRegistry::new( + tx, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests"), + Instant::now(), + ); // no worker ever registered + let relay_delivery = RelayDelivery { + delivery_id: DeliveryId::new("del_worker_missing"), + event_id: EventId::new("evt_worker_missing"), + workspace_id: None, + workspace_alias: None, + from: "Alice".to_string(), + target: MessageTarget::new("ghost"), + body: "hello".to_string(), + thread_id: None, + priority: Some(2), + injection_mode: MessageInjectionMode::Wait, + }; + let mut pending_deliveries = HashMap::new(); + + let first_attempt = super::insert_and_attempt_delivery( + &mut workers, + &mut pending_deliveries, + "ghost", + relay_delivery, + Duration::from_millis(1), + Some(withheld_ack_for("del_worker_missing")), + ) + .await; + assert!( + first_attempt.is_err(), + "a missing recipient must fail the handoff" + ); + let tracked = pending_deliveries.get("del_worker_missing").expect( + "the delivery must remain tracked for the terminal-failure path to dead-letter it, \ + not vanish silently", + ); + assert!( + tracked.withheld_fleet_ack.is_some(), + "the withheld ack must have survived the failed first attempt" + ); + + // The next retry attempt (e.g. the maintenance sweep) observes the + // same missing recipient and reaches the terminal `Failed` outcome + // that `emit_delivery_attempt_outcome` dead-letters and drops the + // ack for — same as every other disposition in this test. + let outcome = retry_pending_delivery( + &DeliveryId::new("del_worker_missing"), + &mut workers, + &mut pending_deliveries, + Duration::from_millis(1), + ) + .await + .expect("a still-missing recipient 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_worker_missing"), + true, + outcome, + ) + .await + .expect("terminal outcome should emit"); + + assert!(!pending_deliveries.contains_key("del_worker_missing")); + let mut book = FleetDeliveryBook::default(); + assert_eq!( + super::fleet::resolve_pending_fleet_ack( + pending_deliveries.get("del_worker_missing"), + &mut book + ), + None, + "a delivery to a permanently missing worker must never resolve into an engine ack" + ); + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + let _ = tokio::time::timeout(Duration::from_secs(1), sdk_out_rx.recv()).await; + } } // relay#1310 MUST-NOT-FIRE: once the worker confirms the injection landed @@ -1003,79 +1237,78 @@ async fn dead_lettered_delivery_drops_withheld_fleet_ack_without_sending_it() { // same internal `delivery_ack` event either way), the engine ack must still // fire, with the delivery's own (agent, up_to_seq) — i.e. the happy path is // unchanged, just correctly gated on confirmation instead of write-enqueue. +// Exercises the full wiring: a real handoff through +// `try_inject_pending_relay_message`, then the same two-step resolution +// `handle_worker_event`'s `delivery_ack` arm performs +// (`clear_pending_delivery_if_event_matches` then `resolve_pending_fleet_ack`). #[tokio::test] -async fn echo_confirmed_delivery_resolves_its_withheld_fleet_ack() { - let mut fleet_delivery_book = FleetDeliveryBook::default(); - let mut pending_fleet_acks = HashMap::from([( - DeliveryId::new("del_landed"), - Deliver { - v: FLEET_WIRE_VERSION, - agent: "agent-a".to_string(), - agent_id: "agent-a-id".to_string(), - delivery_id: "del_landed".to_string(), - msg_id: "evt_landed".to_string(), - seq: 1, - mode: DeliveryMode::Wait, - payload: json!({}), - }, - )]); +async fn successful_injection_still_resolves_its_withheld_fleet_ack() { + let worker_name = "worker-a"; + let mut registry = make_worker_registry_with_worker(worker_name).await; + let deliver = fleet_deliver(1); + let msg = held_fleet_message(&deliver); + let mut pending_deliveries = HashMap::new(); - let resolved = super::fleet::resolve_pending_fleet_ack( - &mut pending_fleet_acks, - &mut fleet_delivery_book, - "del_landed", - "evt_landed", + let delivery_id = try_inject_pending_relay_message( + &mut registry, + &mut pending_deliveries, + worker_name, + &msg, + Duration::from_secs(2), + Some(deliver.clone()), + ) + .await + .expect("a registered worker should accept the handoff"); + + assert!( + pending_deliveries + .get(&delivery_id) + .expect("the delivery must be tracked pending the worker's confirmation") + .withheld_fleet_ack + .is_some(), + "a successful handoff must still withhold the ack pending echo confirmation" ); - assert_eq!(resolved, Some(("agent-a".to_string(), 1))); + let resolved_pending = clear_pending_delivery_if_event_matches( + &mut pending_deliveries, + delivery_id.as_str(), + Some(deliver.msg_id.as_str()), + worker_name, + "delivery_ack", + ); assert!( - !pending_fleet_acks.contains_key("del_landed"), - "a resolved ack must not remain withheld" + resolved_pending.is_some(), + "a matching event_id must clear the pending delivery" ); - // A second confirmation for the same delivery_id (duplicate/replayed - // worker event) must not double-resolve — nothing is left to withhold. - let duplicate = super::fleet::resolve_pending_fleet_ack( - &mut pending_fleet_acks, + let mut fleet_delivery_book = FleetDeliveryBook::default(); + let resolved = super::fleet::resolve_pending_fleet_ack( + resolved_pending.as_ref(), &mut fleet_delivery_book, - "del_landed", - "evt_landed", ); - assert_eq!(duplicate, None); + assert_eq!( + resolved, + Some((deliver.agent.clone(), deliver.seq)), + "a genuinely landed delivery must still resolve its withheld engine ack" + ); + assert!(!pending_deliveries.contains_key(&delivery_id)); + + cleanup_worker_registry(registry).await; } // A worker delivery_ack whose event_id doesn't match the withheld delivery's -// msg_id (stale or reused delivery_id) must not resolve into an engine ack — -// the same stale-event guard `clear_pending_delivery_if_event_matches` -// applies to the broker's own pending-delivery map. +// event_id (stale or reused delivery_id) must not resolve into an engine +// ack. The matching itself is `clear_pending_delivery_if_event_matches`'s +// job (see `clear_pending_delivery_returns_none_for_stale_event_id` below) +// — this documents that `resolve_pending_fleet_ack` correctly has nothing to +// resolve once that guard has already declined to clear the delivery. #[tokio::test] -async fn mismatched_event_id_does_not_resolve_a_withheld_fleet_ack() { +async fn mismatched_event_id_leaves_nothing_for_resolve_pending_fleet_ack() { let mut fleet_delivery_book = FleetDeliveryBook::default(); - let mut pending_fleet_acks = HashMap::from([( - DeliveryId::new("del_reused"), - Deliver { - v: FLEET_WIRE_VERSION, - agent: "agent-a".to_string(), - agent_id: "agent-a-id".to_string(), - delivery_id: "del_reused".to_string(), - msg_id: "evt_original".to_string(), - seq: 1, - mode: DeliveryMode::Wait, - payload: json!({}), - }, - )]); - - let resolved = super::fleet::resolve_pending_fleet_ack( - &mut pending_fleet_acks, - &mut fleet_delivery_book, - "del_reused", - "evt_stale", - ); - - assert_eq!(resolved, None); - assert!( - pending_fleet_acks.contains_key("del_reused"), - "a mismatched event must not consume the withheld entry" + assert_eq!( + super::fleet::resolve_pending_fleet_ack(None, &mut fleet_delivery_book), + None, + "no pending delivery (because the event_id guard declined to clear one) means nothing to resolve" ); } @@ -1109,6 +1342,7 @@ async fn delivery_retry_fails_promptly_when_recipient_is_gone() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: Some("failed writing frame".to_string()), + withheld_fleet_ack: None, }, )]); @@ -1167,6 +1401,7 @@ async fn initial_delivery_failure_stays_owned_until_dead_lettered() { 2, MessageInjectionMode::Wait, Duration::from_millis(1), + None, ) .await .expect_err("missing recipient should fail the initial handoff"); @@ -1226,6 +1461,7 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, }, )]); @@ -1290,11 +1526,9 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); let mut dead_letters = DeadLetterStore::default(); - let mut pending_fleet_acks = HashMap::new(); emit_delivery_attempt_outcome( &sdk_out_tx, &mut dead_letters, - &mut pending_fleet_acks, &DeliveryId::new("del_blip"), true, outcome, @@ -1368,6 +1602,7 @@ async fn delivery_retry_success_clears_stale_last_error() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: Some("old transient failure".to_string()), + withheld_fleet_ack: None, }, )]); @@ -2278,6 +2513,7 @@ fn drop_pending_for_worker_removes_only_matching_entries() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, }, ); pending.insert( @@ -2301,6 +2537,7 @@ fn drop_pending_for_worker_removes_only_matching_entries() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, }, ); @@ -2331,6 +2568,7 @@ async fn dropped_pending_deliveries_emit_terminal_message_failures() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: Some("previous blip".to_string()), + withheld_fleet_ack: None, }; let (sdk_out_tx, mut sdk_out_rx) = mpsc::channel(4); let mut dead_letters = DeadLetterStore::default(); @@ -2401,6 +2639,7 @@ fn should_clear_pending_delivery_when_event_id_matches() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, }; assert!(should_clear_pending_delivery_for_event( @@ -2436,6 +2675,7 @@ fn clear_pending_delivery_returns_none_for_stale_event_id() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, }, )]); @@ -2853,6 +3093,7 @@ fn should_clear_pending_delivery_without_event_id_for_compatibility() { next_retry_at: Instant::now(), queued_at_ms: super::unix_timestamp_millis(), last_error: None, + withheld_fleet_ack: None, }; assert!(should_clear_pending_delivery_for_event( diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index daf1fcc88..720381a5f 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -621,7 +621,6 @@ impl BrokerRuntime { let delivery_retry_interval = self.delivery_retry_interval; let fleet_control_tx = &self.fleet_control_tx; let fleet_delivery_book = &mut self.fleet_delivery_book; - let pending_fleet_acks = &mut self.pending_fleet_acks; let fleet_inventory = &mut self.fleet_inventory; let delivery_states = &self.delivery_states; let terminal_control_tx = &self.terminal_control_tx; @@ -732,14 +731,14 @@ impl BrokerRuntime { // Resolve a fleet (engine-facing) ack withheld // pending confirmation of this exact PTY // injection (relay#1310). No-op when nothing - // is withheld for this delivery_id, or when - // event_id doesn't match (stale/reused id). - if let Some((agent, up_to_seq)) = resolve_pending_fleet_ack( - pending_fleet_acks, - fleet_delivery_book, - ack.delivery_id.as_str(), - ack.event_id.as_str(), - ) { + // is withheld for this delivery — the + // delivery_id/event_id match already happened + // above via `clear_pending_delivery_if_event_matches`, + // so a stale/reused id naturally yields `pending + // = None` here too (relay#1543). + if let Some((agent, up_to_seq)) = + resolve_pending_fleet_ack(pending.as_ref(), fleet_delivery_book) + { let _ = fleet_control_tx .send(FleetControlCommand::Send(delivery_ack( agent, up_to_seq, @@ -1431,6 +1430,7 @@ impl BrokerRuntime { 2, MessageInjectionMode::Wait, delivery_retry_interval, + None, ) .await { @@ -1762,6 +1762,7 @@ impl BrokerRuntime { 2, MessageInjectionMode::Wait, delivery_retry_interval, + None, ) .await { From f8ee6e7da1859f64599ae45fab8166e4d1809e9c Mon Sep 17 00:00:00 2001 From: Miya Date: Mon, 17 Aug 2026 09:35:23 +0200 Subject: [PATCH 3/3] fix(broker): persist withheld fleet ack across broker restart load_pending_deliveries discarded a delivery's withheld_fleet_ack on every reload, so a fleet injection persisted mid-flight came back after a restart with its engine-facing ack silently dropped: the retried delivery could still land and get echo-confirmed, but resolve_pending_fleet_ack had nothing left to release, leaving the engine unacknowledged and risking duplicate redelivery. Persist the field on PersistedPendingDelivery (#[serde(default)] for pre-relay#1543 snapshots) and restore it in load_pending_deliveries. Also rebuilds two tautological assertions flagged in review (tests.rs:1142, tests.rs:1309) that passed a hardcoded None to resolve_pending_fleet_ack and could not fail for any implementation: every_terminal_disposition_drops_its_withheld_fleet_ack's worker-exit/ delivery_failed cases now observe a real pending_deliveries map via take_pending_for_worker, and mismatched_event_id_leaves_nothing_for_ resolve_pending_fleet_ack now builds a real withheld-ack delivery and feeds clear_pending_delivery_if_event_matches's actual return value through, exactly mirroring handle_worker_event's own wiring. Co-Authored-By: Claude Sonnet 5 Session-Id: d9170f41-9b07-46f0-acfc-8aa9140177db --- crates/broker/src/runtime/delivery.rs | 23 ++++- crates/broker/src/runtime/tests.rs | 140 ++++++++++++++++++++++++-- 2 files changed, 148 insertions(+), 15 deletions(-) diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index 96f721866..e6935e4a3 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -38,6 +38,13 @@ pub(crate) struct PersistedPendingDelivery { pub(super) queued_at_ms: u64, #[serde(default)] pub(super) last_error: Option, + /// See `PendingDelivery::withheld_fleet_ack`. `#[serde(default)]` so a + /// snapshot written before this field existed (or by a broker version + /// that predates it) deserializes as `None` instead of failing to load + /// — the same "nothing withheld" state that field already gets from a + /// fresh delivery. See relay#1543's restart-persistence follow-up. + #[serde(default)] + pub(super) withheld_fleet_ack: Option, } #[derive(Debug, Clone, PartialEq)] @@ -151,6 +158,7 @@ pub(crate) fn save_pending_deliveries( failed_attempts: pd.failed_attempts, queued_at_ms: pd.queued_at_ms, last_error: pd.last_error.clone(), + withheld_fleet_ack: pd.withheld_fleet_ack.clone(), }) .collect(); crate::util::fs::write_json_atomic(path, &persisted) @@ -183,11 +191,16 @@ pub(crate) fn load_pending_deliveries(path: &Path) -> HashMap