fix(maestro): broker fails fast when the registry link is down (NAP-263)
ci / rust (push) Failing after 6m5s
ci / rust (push) Failing after 6m5s
`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) <noreply@anthropic.com>
This commit is contained in:
@@ -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<DaemonOutbound>,
|
||||
pending: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<AttachCandidates>>>>>,
|
||||
/// 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<AtomicBool>,
|
||||
}
|
||||
|
||||
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<DaemonOutbound>) {
|
||||
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<AttachCandidates, BrokerError> {
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user