1use std::fmt;
2use std::sync::atomic::{AtomicBool, Ordering};
3use std::sync::{Arc, Condvar, Mutex};
4use std::thread::{self, JoinHandle};
5use std::time::{Duration, Instant};
6
7use signal_hook::consts::signal::{SIGINT, SIGTERM};
8use signal_hook::iterator::{Handle as SignalIteratorHandle, Signals};
9
10use crate::ServerError;
11use crate::server::connection::{ConnectionSupervisor, WebSocketListener};
12use crate::server::listener::ServerListener;
13
14pub(crate) const WEDGED_CONNECTION_STOP: Duration = Duration::from_millis(500);
34
35#[derive(Clone)]
37pub struct ShutdownHandle {
38 inner: Arc<ShutdownState>,
39}
40
41impl ShutdownHandle {
42 #[must_use]
44 pub fn new() -> Self {
45 Self {
46 inner: Arc::new(ShutdownState::new()),
47 }
48 }
49
50 pub fn initiate(&self) -> bool {
55 if self.inner.initiated.swap(true, Ordering::SeqCst) {
56 tracing::debug!("shutdown request ignored because shutdown is already active");
57 return false;
58 }
59
60 tracing::info!("shutdown requested");
61 self.inner.notify();
62 true
63 }
64
65 pub fn wait(&self) {
67 #[cfg(test)]
70 let _returned = Returned(&self.inner);
71 if self.is_initiated() {
72 return;
73 }
74 let Ok(mut guard) = self.inner.wait_lock.lock() else {
75 tracing::error!(
78 "shutdown wait lock poisoned; returning from wait as if shutdown was requested"
79 );
80 return;
81 };
82 while !self.is_initiated() {
83 #[cfg(test)]
86 self.inner.signal_parked();
87 let Ok(next_guard) = self.inner.waiter.wait(guard) else {
88 tracing::error!(
89 "shutdown wait condvar reacquired a poisoned lock; returning from wait as if shutdown was requested"
90 );
91 return;
92 };
93 guard = next_guard;
94 }
95 }
96
97 #[must_use]
99 pub fn is_initiated(&self) -> bool {
100 self.inner.initiated.load(Ordering::SeqCst)
101 }
102}
103
104impl Default for ShutdownHandle {
105 fn default() -> Self {
106 Self::new()
107 }
108}
109
110impl fmt::Debug for ShutdownHandle {
111 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
112 formatter
113 .debug_struct("ShutdownHandle")
114 .field("initiated", &self.is_initiated())
115 .finish()
116 }
117}
118
119#[derive(Debug)]
120struct ShutdownState {
121 initiated: AtomicBool,
122 wait_lock: Mutex<()>,
123 waiter: Condvar,
124 #[cfg(test)]
126 parked: std::sync::atomic::AtomicUsize,
127 #[cfg(test)]
129 returned: std::sync::atomic::AtomicUsize,
130 #[cfg(test)]
132 notified: std::sync::atomic::AtomicUsize,
133 #[cfg(test)]
136 signal: Condvar,
137 #[cfg(test)]
145 signal_lock: Mutex<()>,
146}
147
148#[cfg(test)]
152struct Returned<'a>(&'a ShutdownState);
153
154#[cfg(test)]
155impl Drop for Returned<'_> {
156 fn drop(&mut self) {
157 self.0.count_and_signal(&self.0.returned);
158 }
159}
160
161impl ShutdownState {
162 const fn new() -> Self {
163 Self {
164 initiated: AtomicBool::new(false),
165 wait_lock: Mutex::new(()),
166 waiter: Condvar::new(),
167 #[cfg(test)]
168 parked: std::sync::atomic::AtomicUsize::new(0),
169 #[cfg(test)]
170 returned: std::sync::atomic::AtomicUsize::new(0),
171 #[cfg(test)]
172 notified: std::sync::atomic::AtomicUsize::new(0),
173 #[cfg(test)]
174 signal: Condvar::new(),
175 #[cfg(test)]
176 signal_lock: Mutex::new(()),
177 }
178 }
179
180 #[cfg(test)]
183 fn signal_parked(&self) {
184 self.count_and_signal(&self.parked);
185 }
186
187 #[cfg(test)]
197 fn count_and_signal(&self, count: &std::sync::atomic::AtomicUsize) {
198 let guard = self.signal_lock.lock();
199 count.fetch_add(1, Ordering::SeqCst);
200 self.signal.notify_all();
201 drop(guard);
202 }
203
204 fn notify(&self) {
205 match self.wait_lock.lock() {
206 Ok(_guard) => self.waiter.notify_all(),
207 Err(poisoned) => {
208 tracing::error!("shutdown wait lock poisoned; notifying waiters anyway");
209 let _guard = poisoned.into_inner();
216 self.waiter.notify_all();
217 }
218 }
219 #[cfg(test)]
221 self.notified.fetch_add(1, Ordering::SeqCst);
222 }
223}
224
225#[derive(Debug)]
227pub struct SignalShutdownRegistration {
228 signal_handle: SignalIteratorHandle,
229 worker: Option<JoinHandle<()>>,
230}
231
232impl SignalShutdownRegistration {
233 const fn new(signal_handle: SignalIteratorHandle, worker: JoinHandle<()>) -> Self {
234 Self {
235 signal_handle,
236 worker: Some(worker),
237 }
238 }
239}
240
241impl Drop for SignalShutdownRegistration {
242 fn drop(&mut self) {
243 self.signal_handle.close();
244 let Some(worker) = self.worker.take() else {
245 return;
246 };
247 if worker.join().is_err() {
248 tracing::warn!("shutdown signal worker terminated unexpectedly");
249 }
250 }
251}
252
253pub fn register_signal_handlers(
258 handle: ShutdownHandle,
259) -> Result<SignalShutdownRegistration, ServerError> {
260 let mut signals =
261 Signals::new([SIGTERM, SIGINT]).map_err(|error| ServerError::ListenerAccept {
262 message: format!("failed to register shutdown signal handlers: {error}"),
263 })?;
264 let signal_handle = signals.handle();
265 let worker = thread::spawn(move || {
266 for signal in signals.forever() {
267 tracing::info!(signal, "received shutdown signal");
268 handle.initiate();
269 }
270 });
271 Ok(SignalShutdownRegistration::new(signal_handle, worker))
272}
273
274pub fn run_shutdown_sequence(
308 listener: &mut ServerListener,
309 websocket_listener: Option<&mut WebSocketListener>,
310 supervisor: &ConnectionSupervisor,
311 drain_timeout: Duration,
312) -> Result<(), ServerError> {
313 run_shutdown_sequence_before_flush(
314 listener,
315 websocket_listener,
316 supervisor,
317 drain_timeout,
318 || Ok(()),
319 )
320}
321
322pub(in crate::server) fn run_shutdown_sequence_before_flush(
323 listener: &mut ServerListener,
324 websocket_listener: Option<&mut WebSocketListener>,
325 supervisor: &ConnectionSupervisor,
326 drain_timeout: Duration,
327 finalize: impl FnOnce() -> Result<(), ServerError>,
328) -> Result<(), ServerError> {
329 let started = Instant::now();
330 tracing::info!(
331 ignored_drain_timeout = ?drain_timeout,
332 "starting shutdown sequence; the configured drain timeout has been ignored since 0.14.3 \
333 because every write is durable before it is acknowledged, so no request is in flight to \
334 drain"
335 );
336 if let Some(websocket_listener) = websocket_listener {
339 websocket_listener.stop_accepting()?;
340 }
341 listener.stop_accepting()?;
342
343 if !supervisor.wait_for_delivery_quiesced(Instant::now() + WEDGED_CONNECTION_STOP) {
354 tracing::warn!(
355 stop = ?WEDGED_CONNECTION_STOP,
356 "delivery flush barrier completion was not observed before the wedged-connection stop; \
357 proceeding to shutdown notification"
358 );
359 }
360
361 supervisor.notify_shutdown_subscribers();
362
363 supervisor.force_close_active_connections();
366 wait_after_force_close(supervisor);
367
368 let settled = supervisor.settle_parked_connection_fates();
374
375 supervisor.shutdown();
376 let finalized = finalize();
377 supervisor.stop_participant_warm();
378 let flushed = flush_durable_state(supervisor);
379 tracing::info!(
380 elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
381 clean = settled.is_ok() && finalized.is_ok() && flushed.is_ok(),
382 "shutdown sequence complete"
383 );
384 let mut failure = None;
385 for result in [finalized, flushed, settled] {
386 if let Err(error) = result {
387 tracing::error!(%error, "shutdown phase failed");
388 if failure.is_none() {
389 failure = Some(error);
390 }
391 }
392 }
393 failure.map_or(Ok(()), Err)
394}
395
396pub(crate) fn wait_after_force_close(supervisor: &ConnectionSupervisor) {
408 let deadline = Instant::now() + WEDGED_CONNECTION_STOP;
409 if supervisor.wait_for_connections_drained(deadline) {
410 return;
411 }
412 let remaining = supervisor.active_connection_count();
413 if remaining > 0 {
414 tracing::warn!(
415 active_connections = remaining,
416 stop = ?WEDGED_CONNECTION_STOP,
417 "connections remained active after the wedged-connection stop"
418 );
419 }
420}
421
422fn flush_durable_state(supervisor: &ConnectionSupervisor) -> Result<(), ServerError> {
423 tracing::info!("flushing durable channel state");
424 supervisor.flush_durable_state().map_err(|error| {
425 tracing::error!(%error, "durable state flush failed during shutdown");
426 match error {
427 ServerError::ShutdownFlush { .. } => error,
428 other => ServerError::ShutdownFlush {
429 message: other.to_string(),
430 },
431 }
432 })?;
433 tracing::info!("durable channel state flushed");
434 Ok(())
435}
436
437#[cfg(test)]
438#[path = "shutdown_inflight_tests.rs"]
439mod shutdown_inflight_tests;
440
441#[cfg(test)]
442mod tests {
443 use std::error::Error;
444 use std::sync::atomic::Ordering;
445 use std::thread;
446 use std::time::Duration;
447
448 use super::{ShutdownHandle, wait_after_force_close};
449 use crate::server::connection::ConnectionSupervisor;
450
451 #[test]
452 fn shutdown_handle_initiates_once() {
453 let handle = ShutdownHandle::new();
454
455 assert!(!handle.is_initiated());
456 assert!(handle.initiate());
457 assert!(handle.is_initiated());
458 assert!(!handle.initiate());
459 }
460
461 const SIGNAL_LOCK_POISONED: &str =
463 "the test signal lock was poisoned: another test thread panicked holding it";
464
465 fn await_parked(handle: &ShutdownHandle) -> Result<(), Box<dyn Error>> {
471 let state = &handle.inner;
472 let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
473 loop {
474 if state.returned.load(Ordering::SeqCst) > 0 {
475 return Err("the waiter returned from wait() without parking".into());
476 }
477 if state.parked.load(Ordering::SeqCst) > 0 {
478 drop(guard);
479 return Ok(());
480 }
481 guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
482 }
483 }
484
485 fn await_returned(handle: &ShutdownHandle, parked_before: usize) -> Result<(), Box<dyn Error>> {
494 let state = &handle.inner;
495 if state.notified.load(Ordering::SeqCst) == 0 {
496 return Err("initiate() never notified the parked waiter".into());
497 }
498 let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
499 while state.returned.load(Ordering::SeqCst) == 0 {
500 if state.parked.load(Ordering::SeqCst) > parked_before {
501 return Err("the waiter woke and parked again: it did not see the shutdown".into());
502 }
503 guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
504 }
505 drop(guard);
506 Ok(())
507 }
508
509 #[test]
510 fn shutdown_handle_wait_unblocks_on_initiate() -> Result<(), Box<dyn std::error::Error>> {
511 let handle = ShutdownHandle::new();
512 let waiter = handle.clone();
513 let worker = thread::spawn(move || {
514 waiter.wait();
515 waiter.is_initiated()
516 });
517
518 await_parked(&handle)?;
519 let parked = handle.inner.parked.load(Ordering::SeqCst);
520 assert!(handle.initiate());
521 await_returned(&handle, parked)?;
522 let observed = worker.join().map_err(|_| "wait worker panicked")?;
523
524 assert!(observed);
525 Ok(())
526 }
527
528 fn wait_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
530 &state.wait_lock
531 }
532
533 fn signal_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
535 &state.signal_lock
536 }
537
538 #[allow(clippy::panic)]
541 fn poison(handle: &ShutdownHandle, lock: fn(&super::ShutdownState) -> &std::sync::Mutex<()>) {
542 let poisoner = handle.clone();
543 let outcome = thread::spawn(move || {
544 let _guard = lock(&poisoner.inner).lock();
545 panic!("poison a shutdown lock");
546 })
547 .join();
548 assert!(outcome.is_err(), "the poisoner thread must have panicked");
549 }
550
551 fn poison_wait_lock(handle: &ShutdownHandle) {
553 poison(handle, wait_lock_of);
554 }
555
556 #[test]
559 fn a_poisoned_test_signal_lock_is_refused_by_name() {
560 let handle = ShutdownHandle::new();
561 poison(&handle, signal_lock_of);
562 assert!(handle.initiate());
563
564 let refused =
565 await_returned(&handle, 0).map_or_else(|error| error.to_string(), |()| String::new());
566 assert_eq!(refused, SIGNAL_LOCK_POISONED);
567 let refused =
568 await_parked(&handle).map_or_else(|error| error.to_string(), |()| String::new());
569 assert_eq!(refused, SIGNAL_LOCK_POISONED);
570 }
571
572 #[test]
576 fn wait_on_poisoned_lock_returns_and_logs_error() {
577 let handle = ShutdownHandle::new();
578 poison_wait_lock(&handle);
579
580 let log = crate::test_log::CapturedLog::default();
581 log.capture(|| handle.wait());
582
583 let output = log.contents();
584 assert!(
585 output.contains("ERROR"),
586 "a poisoned wait lock must be logged at error level, got: {output}"
587 );
588 }
589
590 #[test]
594 fn notify_on_poisoned_lock_still_wakes_parked_waiter() -> Result<(), Box<dyn std::error::Error>>
595 {
596 let handle = ShutdownHandle::new();
597 let waiter = handle.clone();
598 let worker = thread::spawn(move || waiter.wait());
599 await_parked(&handle)?;
602 poison_wait_lock(&handle);
603
604 let parked = handle.inner.parked.load(Ordering::SeqCst);
605 assert!(handle.initiate());
606
607 await_returned(&handle, parked)?;
609 worker.join().map_err(|_| "wait worker panicked")?;
610 Ok(())
611 }
612
613 #[test]
616 fn the_close_settle_returns_immediately_when_no_connections_are_active()
617 -> Result<(), Box<dyn std::error::Error>> {
618 let supervisor = ConnectionSupervisor::new()?;
619
620 let started = std::time::Instant::now();
621 wait_after_force_close(&supervisor);
622 let elapsed = started.elapsed();
623
624 assert!(
625 elapsed < Duration::from_millis(50),
626 "the close settle took {elapsed:?} with no connections tracked"
627 );
628 supervisor.shutdown();
629 Ok(())
630 }
631
632 #[test]
639 fn shutdown_source_has_no_drain_and_no_reap_count_sleep_loop() {
640 let source = include_str!("shutdown.rs");
641 let implementation = source.split("mod tests").next().unwrap_or(source);
644 for forbidden in [
645 "DRAIN_PROGRESS_INTERVAL",
646 "FORCE_CLOSE_SETTLE_TIMEOUT",
647 "FORCE_CLOSE_POLL_INTERVAL",
648 "reap_crashed_connections",
649 "fn drain_connections",
650 ] {
651 assert!(
652 !implementation.contains(forbidden),
653 "retired poll/reap/drain token `{forbidden}` must not appear in the shutdown implementation"
654 );
655 }
656 }
657
658 #[test]
662 fn the_configured_drain_timeout_is_never_turned_into_a_deadline() {
663 let source = include_str!("shutdown.rs");
664 let implementation = source.split("mod tests").next().unwrap_or(source);
665 assert!(
666 !implementation.contains("+ drain_timeout"),
667 "drain_timeout must never be added to an Instant to form a shutdown deadline"
668 );
669 assert!(
670 implementation.contains("ignored_drain_timeout = ?drain_timeout"),
671 "drain_timeout must still be named in the shutdown log line that reports it ignored"
672 );
673 }
674}