actix_server/
shutdown.rs

1use std::future::pending;
2
3use tokio::sync::watch;
4
5/// Notification that the server has started a graceful shutdown.
6///
7/// The signal remains notified after graceful shutdown starts. A listener that calls
8/// [`notified`](Self::notified) or clones this signal after that point is notified immediately.
9/// A forced shutdown does not notify this signal.
10///
11/// # Server State
12///
13/// When this signal is notified, the [`Server`](crate::Server) has accepted a graceful shutdown
14/// command, but shutdown is not complete. The server sends this notification before it tells the
15/// accept loop and workers to stop. Therefore, a listener can briefly overlap with connection
16/// acceptance and normal worker operation.
17///
18/// Very shortly after notification, the server stops accepting connections and tells each worker to
19/// stop its services. Active service futures can continue until they finish or the
20/// [`ServerBuilder::shutdown_timeout`](crate::ServerBuilder::shutdown_timeout) expires.
21///
22/// # Use Cases
23///
24/// Connection-oriented services can listen to this signal to coordinate their own drain process.
25/// For example, a protocol dispatcher can:
26///
27/// - close an idle persistent connection;
28/// - stop accepting new logical requests on an active connection;
29/// - let the current request or protocol operation finish; and
30/// - start a protocol-specific close handshake or flush buffered data.
31///
32/// This signal does not replace the worker shutdown timeout. A listener must still finish its
33/// service future for the worker to complete a graceful shutdown.
34#[derive(Clone, Debug)]
35pub struct GracefulShutdownSignal {
36    /// Receiver retained at the pre-shutdown version so each clone observes the shutdown update.
37    rx: watch::Receiver<()>,
38}
39
40impl GracefulShutdownSignal {
41    pub(crate) fn new(rx: watch::Receiver<()>) -> Self {
42        Self { rx }
43    }
44
45    /// Resolves when the server starts a graceful shutdown, or immediately if graceful shutdown
46    /// has already started.
47    pub async fn notified(&self) {
48        let mut rx = self.rx.clone();
49
50        if rx.changed().await.is_err() {
51            pending::<()>().await;
52        }
53    }
54}
55
56#[cfg(test)]
57mod tests {
58    use std::time::Duration;
59
60    use actix_rt::time::timeout;
61
62    use super::GracefulShutdownSignal;
63
64    #[actix_rt::test]
65    async fn set_signal_notifies_listener() {
66        let (tx, rx) = tokio::sync::watch::channel(());
67
68        let signal = GracefulShutdownSignal::new(rx);
69
70        timeout(Duration::from_millis(100), signal.notified())
71            .await
72            .expect_err("signal notified listener before shutdown");
73
74        tx.send_replace(());
75
76        timeout(Duration::from_millis(100), signal.notified())
77            .await
78            .expect("set signal did not notify listener");
79        timeout(Duration::from_millis(100), signal.notified())
80            .await
81            .expect("set signal did not notify later listener");
82    }
83
84    #[actix_rt::test]
85    async fn closed_unset_signal_does_not_notify_listener() {
86        let (tx, rx) = tokio::sync::watch::channel(());
87        let signal = GracefulShutdownSignal::new(rx);
88
89        drop(tx);
90
91        assert!(timeout(Duration::from_millis(10), signal.notified())
92            .await
93            .is_err());
94    }
95}
actix_server/shutdown.rs

actix_server/
shutdown.rs
