From ae3775946de5e1aeed846327774ca693020405ad Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Sun, 31 May 2026 23:10:43 +0000 Subject: [PATCH] fix(maestro): broker fails fast when the registry link is down (NAP-263) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `BrokerHandle::resolve_attach` queued ResolveAttach frames with `out_tx.send().await`. While the registry was disconnected nothing drained the bounded channel, so after 64 queued frames every caller blocked on the send — HTTP `/api/attach/:id` requests for peer sessions hung instead of erroring. Track link state with an `AtomicBool` the registry-client flips on connect / disconnect: `resolve_attach` returns `NoControlLink` immediately while down, and uses a non-blocking `try_send` so a back-pressured-but-connected channel can't block either. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/maestro-daemon/src/broker.rs | 50 +++++++++++++++++++- crates/maestro-daemon/src/registry_client.rs | 10 +++- 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/crates/maestro-daemon/src/broker.rs b/crates/maestro-daemon/src/broker.rs index 05792a9..6ca94f3 100644 --- a/crates/maestro-daemon/src/broker.rs +++ b/crates/maestro-daemon/src/broker.rs @@ -7,6 +7,7 @@ use std::collections::HashMap; use std::hash::Hash; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -56,23 +57,45 @@ pub enum BrokerError { pub struct BrokerHandle { out_tx: mpsc::Sender, pending: Arc>>>>, + /// Whether the registry control link is currently up. Flipped by the + /// registry-client task via [`BrokerHandle::set_connected`]. When down, + /// `resolve_attach` fails fast instead of queueing onto an undrained + /// channel that would otherwise fill and block every caller (NAP-263). + connected: Arc, } impl BrokerHandle { /// `(handle, receiver)` — the registry-client task owns the receiver and /// drains it into the control-link WS sink. The handle is shared with /// HTTP handlers and is what they use to issue requests. + /// + /// Starts **disconnected**: the registry-client task calls + /// [`set_connected(true)`](Self::set_connected) once the handshake lands. pub fn new() -> (Self, mpsc::Receiver) { let (out_tx, out_rx) = mpsc::channel(64); let handle = Self { out_tx, pending: Arc::new(Mutex::new(HashMap::new())), + connected: Arc::new(AtomicBool::new(false)), }; (handle, out_rx) } + /// Mark the control link up/down. Called by the registry-client task on + /// every connect (`true`) and disconnect (`false`). + pub fn set_connected(&self, connected: bool) { + self.connected.store(connected, Ordering::Release); + } + /// Resolve a peer session: ask the registry for an [`AttachCandidates`]. pub async fn resolve_attach(&self, session_id: Uuid) -> Result { + // Fail fast when the registry link is down — otherwise the bounded + // outbound channel (nothing draining it while disconnected) fills up + // and every subsequent caller blocks on `send().await` (NAP-263). + if !self.connected.load(Ordering::Acquire) { + return Err(BrokerError::NoControlLink); + } + let reply_to = Uuid::new_v4(); let (tx, rx) = oneshot::channel(); self.pending.lock().unwrap().insert(reply_to, tx); @@ -83,13 +106,14 @@ impl BrokerHandle { key: reply_to, }; + // Non-blocking send: a full channel (registry up but back-pressured) + // is treated as "link unavailable" rather than parking the caller. if self .out_tx - .send(DaemonOutbound::ResolveAttach { + .try_send(DaemonOutbound::ResolveAttach { reply_to, session_id, }) - .await .is_err() { return Err(BrokerError::NoControlLink); @@ -136,6 +160,7 @@ mod tests { // Keep the receiver alive so `out_tx.send()` succeeds and the call // parks on the reply oneshot rather than failing fast. let (handle, _out_rx) = BrokerHandle::new(); + handle.set_connected(true); { let fut = handle.resolve_attach(Uuid::new_v4()); @@ -158,4 +183,25 @@ mod tests { "pending entry must be removed on cancellation (NAP-259)" ); } + + /// NAP-263: while the registry link is down, `resolve_attach` must return + /// promptly instead of queueing onto the undrained outbound channel (which + /// would fill and deadlock callers). The receiver is alive with free + /// capacity here, so a `NoControlLink` proves the disconnected gate — not a + /// closed/full channel — short-circuited the call. + #[tokio::test] + async fn resolve_attach_fails_fast_while_disconnected() { + let (handle, _out_rx) = BrokerHandle::new(); // starts disconnected + + let res = tokio::time::timeout( + Duration::from_secs(1), + handle.resolve_attach(Uuid::new_v4()), + ) + .await + .expect("resolve_attach must return promptly, not block on a full channel"); + + assert!(matches!(res, Err(BrokerError::NoControlLink)), "{res:?}"); + // It must not leave a pending entry behind either. + assert_eq!(handle.pending.lock().unwrap().len(), 0); + } } diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index 838ea56..c878d51 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -146,7 +146,11 @@ impl RegistryClient { pub async fn run(mut self) { let mut backoff = BACKOFF_MIN; loop { - match self.connect_once().await { + let outcome = self.connect_once().await; + // Whatever happened, the link is down now — make the broker fail + // fast until the next successful handshake re-marks it up (NAP-263). + self.broker.set_connected(false); + match outcome { Ok(()) => { tracing::info!("registry connection ended cleanly"); backoff = BACKOFF_MIN; @@ -194,6 +198,10 @@ impl RegistryClient { } } + // Link is up — let the broker queue ResolveAttach frames again + // (NAP-263). Cleared in `run` once this connection ends. + self.broker.set_connected(true); + // Subscribe to the session-event stream BEFORE taking the snapshot. A // reap or kill that races the handshake would otherwise slip between // the snapshot and the subscription — present in neither — leaving the