1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
// SPDX-License-Identifier: MIT OR Apache-2.0
//! 2026-09-26: One shutdown path for every trigger. [`request`] is called by
//! the `SIGINT`/`SIGTERM` task from [`install_signal_listeners`], by the TUI
//! on Ctrl+C (raw mode delivers it as a key event), by `/quit`, by the TUI's
//! own exit, by the scheduler when the fault latch reports a fault, and by the
//! bench self-start teardown.
//!
//! During startup a request is sent on `main`'s one-shot escape. Once the
//! router's accept loop has called [`disarm_startup_escape`], a request wakes
//! [`wait`] in that loop, which stops accepting, drains in-flight requests for
//! a bounded grace and returns, so `serve()` unwinds normally.
//!
//! Owner: server tui.
//! Invariants: the shutdown latch is one-way: once [`requested`] is true it
//! stays true for the life of the process.
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::{Notify, oneshot};
/// 2026-09-26: Where the process is in its shutdown sequence.
struct Phase {
/// 2026-09-26: Set once a shutdown has been requested, from any trigger.
requested: AtomicBool,
/// 2026-09-26: Whether a request still takes the startup escape.
/// Cleared by [`disarm_startup_escape`].
in_startup: AtomicBool,
}
// 2026-09-26: Static because a shutdown request is process-wide, and the
// callers of `request` (the signal task, the TUI thread, the scheduler's fault
// paths, the bench self-start) share no object it could be carried on.
static PHASE: Phase = Phase {
requested: AtomicBool::new(false),
in_startup: AtomicBool::new(true),
};
/// 2026-09-26: The wakeup for [`wait`] and the startup escape. `request`
/// always wakes the waiters, and while `in_startup` is set it also sends on
/// the escape.
#[derive(Default)]
struct Channels {
notify: Notify,
escape: std::sync::Mutex<Option<oneshot::Sender<&'static str>>>,
}
static CHANNELS: OnceLock<Channels> = OnceLock::new();
fn channels() -> &'static Channels {
CHANNELS.get_or_init(Channels::default)
}
fn notify() -> &'static Notify {
&channels().notify
}
/// 2026-09-26: The startup escape: the sending half of `main`'s one-shot.
/// The first request during startup takes it; later ones find the slot empty.
///
/// `main` races `serve()` against the receiving half, so a shutdown during
/// model load does not wait for the accept loop. Once
/// [`disarm_startup_escape`] has run, a request goes through the draining
/// path instead, so a live server does not exit with requests in flight.
fn escape() -> &'static std::sync::Mutex<Option<oneshot::Sender<&'static str>>> {
&channels().escape
}
/// 2026-09-26: Arm the startup escape with `main`'s sender. `main` calls it
/// once, before `serve()`.
pub fn arm_startup_escape(tx: oneshot::Sender<&'static str>) {
*escape().lock().expect("shutdown escape poisoned") = Some(tx);
}
/// 2026-09-26: Close the startup escape once the server is accepting: from
/// here on, a shutdown means "stop accepting and drain", not "return from
/// `main`".
///
/// It only clears `in_startup` and leaves the sender parked, so the channel
/// never closes and `main`'s receiver never resolves with `Err`.
pub fn disarm_startup_escape() {
PHASE.in_startup.store(false, Ordering::SeqCst);
}
/// 2026-09-26: Request a clean shutdown. Idempotent; callable from any
/// thread.
pub fn request(reason: &'static str) {
if !PHASE.requested.swap(true, Ordering::SeqCst) {
tracing::info!("Shutdown requested ({reason}) — draining in-flight requests");
}
// 2026-09-26: In startup, hand the reason to `main`'s `select!`. Only the
// first request takes the sender. After disarming, the sender stays
// parked and the notification below does the work.
if PHASE.in_startup.load(Ordering::SeqCst)
&& let Ok(mut slot) = escape().lock()
&& let Some(tx) = slot.take()
{
let _ = tx.send(reason);
}
notify().notify_waiters();
}
/// 2026-09-26: Whether a shutdown has been requested.
pub fn requested() -> bool {
PHASE.requested.load(Ordering::SeqCst)
}
/// 2026-09-26: Resolves when a shutdown is requested, at once if one
/// already was.
pub async fn wait() {
if requested() {
return;
}
// 2026-09-26: Create the `Notified` future before re-checking, so a
// `request` between the two checks still wakes it.
let notified = notify().notified();
if requested() {
return;
}
notified.await;
}
/// 2026-09-26: Install `SIGINT` and `SIGTERM` listeners on the tokio
/// runtime (`SIGINT` only off Unix). Called from `serve()` and from the bench
/// runner. `SIGTERM` still arrives while the TUI owns the keyboard.
pub fn install_signal_listeners() {
tokio::spawn(async {
let ctrl_c = tokio::signal::ctrl_c();
#[cfg(unix)]
{
let mut term =
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
Ok(s) => s,
Err(e) => {
tracing::warn!("SIGTERM listener unavailable: {e}");
if ctrl_c.await.is_ok() {
request("SIGINT");
}
return;
}
};
tokio::select! {
r = ctrl_c => { if r.is_ok() { request("SIGINT"); } }
_ = term.recv() => { request("SIGTERM"); }
}
}
#[cfg(not(unix))]
{
if ctrl_c.await.is_ok() {
request("SIGINT");
}
}
});
}
/// 2026-09-26: After the accept loop stops, wait until the
/// `REQUESTS_ACTIVE` gauge reads zero or the grace window expires, polling
/// every 200 ms.
pub async fn drain_in_flight(grace: Duration) {
let start = std::time::Instant::now();
loop {
let active = crate::metrics::REQUESTS_ACTIVE.get();
if active <= 0 {
tracing::info!("Drain complete — no active requests");
return;
}
if start.elapsed() >= grace {
tracing::warn!(
"Drain grace ({}s) expired with {active} request(s) still active — exiting",
grace.as_secs()
);
return;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
}
#[cfg(test)]
#[path = "shutdown_tests.rs"]
mod tests;