liminal_server/server/connection/supervisor/supervisor_shutdown.rs
1use super::{
2 ConnectionControl, ConnectionSupervisor, Instant, ParticipantServiceFatal, ServerError,
3};
4
5impl ConnectionSupervisor {
6 /// How many connections have departed this supervisor: the count of
7 /// connection records removed, raised by the one removal funnel every exit
8 /// route reaches. A teardown that completes its connection's fate does so
9 /// before the record is removed, so a departure counted here is one whose
10 /// fate has already completed. Pass it to [`Self::await_departures`].
11 ///
12 /// # Errors
13 /// Returns [`ServerError`] when the departure count's lock is poisoned.
14 pub fn departures(&self) -> Result<u64, ServerError> {
15 self.inner.runtime.departures()
16 }
17
18 /// Parks until `count` connections have departed since `since`, a value
19 /// [`Self::departures`] returned, or until `deadline` elapses. Returns
20 /// `true` when they have departed and `false` when the deadline won.
21 ///
22 /// Woken by each departure through the removal `Condvar`, never by a timer
23 /// or a count scan; the deadline is the one bound on the park.
24 ///
25 /// # Errors
26 /// Returns [`ServerError`] when the departure count's lock is poisoned.
27 pub fn await_departures(
28 &self,
29 since: u64,
30 count: u64,
31 deadline: Instant,
32 ) -> Result<bool, ServerError> {
33 self.inner.runtime.await_departures(since, count, deadline)
34 }
35
36 /// Parks until every tracked connection has been removed or `deadline`
37 /// elapses, returning `true` when the drain completed and `false` when the
38 /// single admitted deadline won.
39 ///
40 /// The TOLD drain-completion replacement (W4 leg 3, §4.3): the waiter is woken
41 /// by the one `remove()` funnel every connection exit reaches — never by a
42 /// reap/count timer. Both the graceful drain and the force-close settle call
43 /// this same waiter, each with its own one-shot deadline; there is no second
44 /// settle poll loop.
45 #[must_use]
46 pub(crate) fn wait_for_connections_drained(&self, deadline: Instant) -> bool {
47 let drained = self
48 .inner
49 .runtime
50 .wait_for_active_connections_drained(deadline);
51 let retired = self.inner.retirements.wait(deadline);
52 if !retired {
53 tracing::warn!(
54 "force-closed native connection retirement was not observed before the wedged-connection stop"
55 );
56 }
57 drained && retired
58 }
59
60 /// FIX A-ii shutdown flush barrier: parks until every active connection has
61 /// fanned out its accepted publishes to its socket, or `deadline` elapses.
62 /// Called in `run_shutdown_sequence` BEFORE the shutdown Disconnect broadcast
63 /// so an accepted-but-unfanned-out publish can no longer be overtaken by it.
64 #[must_use]
65 pub(crate) fn wait_for_delivery_quiesced(&self, deadline: Instant) -> bool {
66 self.inner.runtime.wait_for_delivery_quiesced(deadline)
67 }
68
69 /// Returns the first latched post-Open participant fatal, if any.
70 ///
71 /// The production runtime reads this after its existing shutdown handle wakes,
72 /// then returns the typed fatal after the ordinary drain and durable flush.
73 pub(crate) fn participant_service_fatal(
74 &self,
75 ) -> Result<Option<ParticipantServiceFatal>, ServerError> {
76 self.inner.runtime.participant_service_fatal()
77 }
78
79 /// Broadcasts a best-effort shutdown notification to active connections.
80 ///
81 /// Connections with no active subscriptions ignore the notification. Failures
82 /// to enqueue the control message are logged and skipped; they are not retried.
83 pub fn notify_shutdown_subscribers(&self) {
84 self.inner
85 .broadcast_control(&ConnectionControl::NotifyShutdown);
86 }
87
88 /// Sends the orderly close control message to every tracked connection
89 /// process.
90 ///
91 /// Each live process enqueues one protocol `Disconnect`, drains its
92 /// outbound buffer, completes its connection fate as `ServerShutdown` and —
93 /// on the WebSocket transport — writes a close frame carrying
94 /// `CloseCode::Away` and the reason "server shutdown" before exiting
95 /// normally. A peer reads a shutdown, not an error. Enqueue failures are
96 /// logged and skipped.
97 ///
98 /// Since 0.14.3 this is the ONLY close route in `run_shutdown_sequence`:
99 /// nothing waits for a peer to hang up first, so the log line below is the
100 /// ordinary shutdown record rather than a warning about an expired drain.
101 pub fn force_close_active_connections(&self) {
102 for connection in self.inner.runtime.active_connections() {
103 self.inner
104 .runtime
105 .link_life
106 .close(connection.pid, crate::link::LinkEnd::Shutdown);
107 tracing::info!(
108 connection_pid = connection.pid,
109 peer_addr = ?connection.peer_addr,
110 "closing connection for server shutdown"
111 );
112 if !self
113 .inner
114 .enqueue_control(connection.pid, ConnectionControl::ForceClose)
115 {
116 // Observation, not diagnosis — see `broadcast_control`.
117 tracing::warn!(
118 connection_pid = connection.pid,
119 peer_addr = ?connection.peer_addr,
120 "forceful connection close was not published to the process"
121 );
122 }
123 }
124 }
125
126 /// Flushes durable channel state through the configured liminal services.
127 ///
128 /// # Errors
129 /// Returns [`ServerError::ShutdownFlush`] when the underlying service flush fails.
130 pub fn flush_durable_state(&self) -> Result<(), ServerError> {
131 self.inner.runtime.services().flush_durable_state()
132 }
133
134 /// Stops the participant warm and records the clean stop's capacity
135 /// snapshot, before the final flush (DUR-020 R3, R5).
136 pub(crate) fn stop_participant_warm(&self) {
137 self.inner.runtime.services().stop_participant_warm();
138 }
139
140 /// Completes, before a clean stop reports success, every connection fate
141 /// parked on candidate-lane `Precedence`.
142 ///
143 /// # Errors
144 /// Returns [`ServerError::ShutdownUnsettledConnectionFates`] naming every
145 /// durable Open the stop could not complete.
146 pub(crate) fn settle_parked_connection_fates(&self) -> Result<(), ServerError> {
147 self.inner.runtime.settle_parked_connection_fates()
148 }
149
150 /// Stops the beamr scheduler used by connection processes.
151 pub fn shutdown(&self) {
152 // SRV-042 R1: a connection held for the incarnation start is closed
153 // now and never spawned.
154 self.inner.held.stop();
155 // SRV-042 R2: a start still running ends at its next step boundary.
156 if let Some(authority) = &self.inner.incarnations {
157 authority.abandon();
158 }
159 // A connection waiting on a conversation's restore is woken and
160 // answered at once, so the scheduler join below never waits on a
161 // room's replay or rebuild (Gaia, 11 October 2026: 60 s).
162 self.stop_participant_warm();
163 // Remove every host record while the readiness owner is still live. The
164 // removal path ACKs deregistration and only then releases each fd guard;
165 // scheduler shutdown subsequently drops the process-owned handles.
166 let removed: Vec<_> = self
167 .inner
168 .runtime
169 .active_connections()
170 .into_iter()
171 .filter_map(|connection| {
172 self.inner
173 .runtime
174 .remove(connection.pid)
175 .map(|record| (connection.pid, record))
176 })
177 .collect();
178 if let Some(notifier) = self.inner.runtime.notifier.as_ref() {
179 let detached: Vec<_> = removed
180 .iter()
181 .filter_map(|(pid, record)| {
182 record
183 .pass_principal
184 .as_ref()
185 .map(|principal| (*pid, principal))
186 })
187 .collect();
188 let unregistered: Vec<_> = removed
189 .iter()
190 .filter_map(|(pid, record)| record.registration.as_ref().map(|_| *pid))
191 .collect();
192 notifier.on_shutdown(&detached, &unregistered);
193 }
194 self.inner.scheduler.shutdown();
195 }
196}