From 45be83b891f0b860234c3fb9f8fb21ed21f07dc5 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Mon, 28 Sep 2026 14:26:37 -0700 Subject: [PATCH] review: gatewayd drops an approval's answer receiver before it says the approval ended Every path that told the loop `Ended` did so while the handler still held the receiver of the approver's answer; it was dropped only when the handler returned. An answer sent in between was taken, reported to the approver as sent, and lost. `end` now takes the receiver and drops it first. `brokerd_closing_first_ends_it_as_no_longer_pending` failed 10 of 60 runs under load before, and 0 of 100 after. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/gatewayd/src/gateway.rs | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/crates/gatewayd/src/gateway.rs b/crates/gatewayd/src/gateway.rs index 5b544ee..1d65fd8 100644 --- a/crates/gatewayd/src/gateway.rs +++ b/crates/gatewayd/src/gateway.rs @@ -234,7 +234,7 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender, timing: Timi }; let rx = match frames(&stream) { Ok(rx) => rx, - Err(_) => return end(to_loop, id, Ending::NoLonger), + Err(_) => return end(to_loop, id, Ending::NoLonger, answer_rx), }; if write_frame( &mut stream, @@ -242,7 +242,7 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender, timing: Timi ) .is_err() { - return end(to_loop, id, Ending::NoLonger); + return end(to_loop, id, Ending::NoLonger, answer_rx); } // 3. An answer from an approver, or brokerd closing first. let mut answers = Some(answer_rx); @@ -261,13 +261,13 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender, timing: Timi // brokerd closed (the approval ended elsewhere), or sent what it may not. _ => { let _ = stream.shutdown(std::net::Shutdown::Both); - return end(to_loop, id, Ending::NoLonger); + return end(to_loop, id, Ending::NoLonger, answers); } } }; let (by, reason) = (reply.by.clone(), reply.reason.clone()); if write_frame(&mut stream, &frame(1, true, Message::ApprovalReply(reply))).is_err() { - return end(to_loop, id, Ending::NoLonger); + return end(to_loop, id, Ending::NoLonger, answers); } // 4. What came of it. let ending = match rx.recv_timeout(timing.outcome) { @@ -289,10 +289,13 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender, timing: Timi } }; let _ = stream.shutdown(std::net::Shutdown::Both); - end(to_loop, id, ending); + end(to_loop, id, ending, answers); } -fn end(to_loop: &Sender, approval: u64, ending: Ending) { - // `let _ = to_loop.send(ToLoop::Ended { approval, ending });` +/// Tells the loop the approval has ended, after dropping the receiver of its answers: once the +/// loop has the `Ended`, an approver's answer must fail to send ("not pending here"), never be +/// taken by a handler that has already given up on it. +fn end(to_loop: &Sender, approval: u64, ending: Ending, answers: T) { + drop(answers); let _ = to_loop.send(ToLoop::Ended { approval, ending }); }