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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
use super::{
ConnectionControl, ConnectionSupervisor, Instant, ParticipantServiceFatal, ServerError,
};
impl ConnectionSupervisor {
/// How many connections have departed this supervisor: the count of
/// connection records removed, raised by the one removal funnel every exit
/// route reaches. A teardown that completes its connection's fate does so
/// before the record is removed, so a departure counted here is one whose
/// fate has already completed. Pass it to [`Self::await_departures`].
///
/// # Errors
/// Returns [`ServerError`] when the departure count's lock is poisoned.
pub fn departures(&self) -> Result<u64, ServerError> {
self.inner.runtime.departures()
}
/// Parks until `count` connections have departed since `since`, a value
/// [`Self::departures`] returned, or until `deadline` elapses. Returns
/// `true` when they have departed and `false` when the deadline won.
///
/// Woken by each departure through the removal `Condvar`, never by a timer
/// or a count scan; the deadline is the one bound on the park.
///
/// # Errors
/// Returns [`ServerError`] when the departure count's lock is poisoned.
pub fn await_departures(
&self,
since: u64,
count: u64,
deadline: Instant,
) -> Result<bool, ServerError> {
self.inner.runtime.await_departures(since, count, deadline)
}
/// Parks until every tracked connection has been removed or `deadline`
/// elapses, returning `true` when the drain completed and `false` when the
/// single admitted deadline won.
///
/// The TOLD drain-completion replacement (W4 leg 3, §4.3): the waiter is woken
/// by the one `remove()` funnel every connection exit reaches — never by a
/// reap/count timer. Both the graceful drain and the force-close settle call
/// this same waiter, each with its own one-shot deadline; there is no second
/// settle poll loop.
#[must_use]
pub(crate) fn wait_for_connections_drained(&self, deadline: Instant) -> bool {
let drained = self
.inner
.runtime
.wait_for_active_connections_drained(deadline);
let retired = self.inner.retirements.wait(deadline);
if !retired {
tracing::warn!(
"force-closed native connection retirement was not observed before the wedged-connection stop"
);
}
drained && retired
}
/// FIX A-ii shutdown flush barrier: parks until every active connection has
/// fanned out its accepted publishes to its socket, or `deadline` elapses.
/// Called in `run_shutdown_sequence` BEFORE the shutdown Disconnect broadcast
/// so an accepted-but-unfanned-out publish can no longer be overtaken by it.
#[must_use]
pub(crate) fn wait_for_delivery_quiesced(&self, deadline: Instant) -> bool {
self.inner.runtime.wait_for_delivery_quiesced(deadline)
}
/// Returns the first latched post-Open participant fatal, if any.
///
/// The production runtime reads this after its existing shutdown handle wakes,
/// then returns the typed fatal after the ordinary drain and durable flush.
pub(crate) fn participant_service_fatal(
&self,
) -> Result<Option<ParticipantServiceFatal>, ServerError> {
self.inner.runtime.participant_service_fatal()
}
/// Broadcasts a best-effort shutdown notification to active connections.
///
/// Connections with no active subscriptions ignore the notification. Failures
/// to enqueue the control message are logged and skipped; they are not retried.
pub fn notify_shutdown_subscribers(&self) {
self.inner
.broadcast_control(&ConnectionControl::NotifyShutdown);
}
/// Sends the orderly close control message to every tracked connection
/// process.
///
/// Each live process enqueues one protocol `Disconnect`, drains its
/// outbound buffer, completes its connection fate as `ServerShutdown` and —
/// on the WebSocket transport — writes a close frame carrying
/// `CloseCode::Away` and the reason "server shutdown" before exiting
/// normally. A peer reads a shutdown, not an error. Enqueue failures are
/// logged and skipped.
///
/// Since 0.14.3 this is the ONLY close route in `run_shutdown_sequence`:
/// nothing waits for a peer to hang up first, so the log line below is the
/// ordinary shutdown record rather than a warning about an expired drain.
pub fn force_close_active_connections(&self) {
for connection in self.inner.runtime.active_connections() {
self.inner
.runtime
.link_life
.close(connection.pid, crate::link::LinkEnd::Shutdown);
tracing::info!(
connection_pid = connection.pid,
peer_addr = ?connection.peer_addr,
"closing connection for server shutdown"
);
if !self
.inner
.enqueue_control(connection.pid, ConnectionControl::ForceClose)
{
// Observation, not diagnosis — see `broadcast_control`.
tracing::warn!(
connection_pid = connection.pid,
peer_addr = ?connection.peer_addr,
"forceful connection close was not published to the process"
);
}
}
}
/// Flushes durable channel state through the configured liminal services.
///
/// # Errors
/// Returns [`ServerError::ShutdownFlush`] when the underlying service flush fails.
pub fn flush_durable_state(&self) -> Result<(), ServerError> {
self.inner.runtime.services().flush_durable_state()
}
/// Stops the participant warm and records the clean stop's capacity
/// snapshot, before the final flush (DUR-020 R3, R5).
pub(crate) fn stop_participant_warm(&self) {
self.inner.runtime.services().stop_participant_warm();
}
/// Completes, before a clean stop reports success, every connection fate
/// parked on candidate-lane `Precedence`.
///
/// # Errors
/// Returns [`ServerError::ShutdownUnsettledConnectionFates`] naming every
/// durable Open the stop could not complete.
pub(crate) fn settle_parked_connection_fates(&self) -> Result<(), ServerError> {
self.inner.runtime.settle_parked_connection_fates()
}
/// Stops the beamr scheduler used by connection processes.
pub fn shutdown(&self) {
// SRV-042 R1: a connection held for the incarnation start is closed
// now and never spawned.
self.inner.held.stop();
// SRV-042 R2: a start still running ends at its next step boundary.
if let Some(authority) = &self.inner.incarnations {
authority.abandon();
}
// A connection waiting on a conversation's restore is woken and
// answered at once, so the scheduler join below never waits on a
// room's replay or rebuild (Gaia, 11 October 2026: 60 s).
self.stop_participant_warm();
// Remove every host record while the readiness owner is still live. The
// removal path ACKs deregistration and only then releases each fd guard;
// scheduler shutdown subsequently drops the process-owned handles.
let removed: Vec<_> = self
.inner
.runtime
.active_connections()
.into_iter()
.filter_map(|connection| {
self.inner
.runtime
.remove(connection.pid)
.map(|record| (connection.pid, record))
})
.collect();
if let Some(notifier) = self.inner.runtime.notifier.as_ref() {
let detached: Vec<_> = removed
.iter()
.filter_map(|(pid, record)| {
record
.pass_principal
.as_ref()
.map(|principal| (*pid, principal))
})
.collect();
let unregistered: Vec<_> = removed
.iter()
.filter_map(|(pid, record)| record.registration.as_ref().map(|_| *pid))
.collect();
notifier.on_shutdown(&detached, &unregistered);
}
self.inner.scheduler.shutdown();
}
}