kyle/boxmaker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

review: gatewayd drops an approval's answer receiver before it says the approval ended

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
kyle
2026-09-28 14:26:37 -07:00
co-authored by Claude Opus 5.5
parent 161b16353b
commit 45be83b891
1 changed files with 10 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/gatewayd/src/gateway.rs
+10 -7
View File
@@ -234,7 +234,7 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender<ToLoop>, timing: Timi
}; };
let rx = match frames(&stream) { let rx = match frames(&stream) {
Ok(rx) => rx, Ok(rx) => rx,
Err(_) => return end(to_loop, id, Ending::NoLonger), Err(_) => return end(to_loop, id, Ending::NoLonger, answer_rx),
}; };
if write_frame( if write_frame(
&mut stream, &mut stream,
@@ -242,7 +242,7 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender<ToLoop>, timing: Timi
) )
.is_err() .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. // 3. An answer from an approver, or brokerd closing first.
let mut answers = Some(answer_rx); let mut answers = Some(answer_rx);
@@ -261,13 +261,13 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender<ToLoop>, timing: Timi
// brokerd closed (the approval ended elsewhere), or sent what it may not. // brokerd closed (the approval ended elsewhere), or sent what it may not.
_ => { _ => {
let _ = stream.shutdown(std::net::Shutdown::Both); 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()); let (by, reason) = (reply.by.clone(), reply.reason.clone());
if write_frame(&mut stream, &frame(1, true, Message::ApprovalReply(reply))).is_err() { 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. // 4. What came of it.
let ending = match rx.recv_timeout(timing.outcome) { let ending = match rx.recv_timeout(timing.outcome) {
@@ -289,10 +289,13 @@ pub fn connection(mut stream: UnixStream, to_loop: &Sender<ToLoop>, timing: Timi
} }
}; };
let _ = stream.shutdown(std::net::Shutdown::Both); let _ = stream.shutdown(std::net::Shutdown::Both);
end(to_loop, id, ending); end(to_loop, id, ending, answers);
} }
fn end(to_loop: &Sender<ToLoop>, approval: u64, ending: Ending) { /// Tells the loop the approval has ended, after dropping the receiver of its answers: once the
// `let _ = to_loop.send(ToLoop::Ended { approval, ending });` /// 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<T>(to_loop: &Sender<ToLoop>, approval: u64, ending: Ending, answers: T) {
drop(answers);
let _ = to_loop.send(ToLoop::Ended { approval, ending }); let _ = to_loop.send(ToLoop::Ended { approval, ending });
} }