shutdown.rs 80.0%

36 of 45 lines run, 12 of 22 functions · run #482 on proposal/333 · run missed

//! Stopping without cutting anyone off (oxygenhub#233). On SIGTERM (a
//! `systemctl restart`) or Ctrl-C the hub stops taking connections, ends
//! its event streams (browsers reconnect by themselves), and gives the
//! requests in flight up to `DRAIN` to finish. Given a listening socket by
//! systemd (socket activation, docs/deploy.md), it serves that rather than
//! binding: the socket outlives the restart, so a request that arrives
//! between the old process and the new one waits in its queue instead of
//! being refused.

use std::sync::OnceLock;
use tokio::sync::watch;

/// How long the requests in flight get to finish.
pub const DRAIN: std::time::Duration = std::time::Duration::from_secs(30);

/// How long, after the drain, work still running on the runtime's
/// blocking threads (a slow request's repository work, a bucket call) may
/// hold the process. Dropping the runtime would wait for it however long
/// it takes: on 2026-10-17 a restart stood a minute past "stopped", until
/// systemd killed it.
pub const EXIT_GRACE: std::time::Duration = std::time::Duration::from_secs(1);

/// End the runtime once `serve` has returned, cutting off what still runs.
pub fn finish(rt: tokio::runtime::Runtime) {
    rt.shutdown_timeout(EXIT_GRACE);
}

fn sender() -> &'static watch::Sender<bool> {
    static STOP: OnceLock<watch::Sender<bool>> = OnceLock::new();
    STOP.get_or_init(|| watch::channel(false).0)
}

/// Begin stopping.
pub fn begin() {
    sender().send_replace(true);
}

pub fn is_stopping() -> bool {
    *sender().borrow()
}

/// Done when the hub begins stopping: for event streams to end on, and for
/// the servers' graceful shutdown.
pub async fn stopping() {
    let mut rx = sender().subscribe();
    let _ = rx.wait_for(|s| *s).await;
}

/// Wait for SIGTERM or Ctrl-C, then begin stopping. Installed before the
/// servers start, so a signal never kills the process outright.
pub fn on_signal() -> anyhow::Result<impl std::future::Future<Output = ()>> {
    #[cfg(unix)]
    let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
    Ok(async move {
        #[cfg(unix)]
        tokio::select! {
            _ = term.recv() => tracing::info!("SIGTERM: stopping (requests in flight may take {} s)", DRAIN.as_secs()),
            _ = tokio::signal::ctrl_c() => tracing::info!("stopping"),
        }
        #[cfg(not(unix))]
        {
            let _ = tokio::signal::ctrl_c().await;
            tracing::info!("stopping");
        }
        begin();
    })
}

/// The listening socket systemd passed (LISTEN_FDS, first descriptor 3),
/// when it passed one to this process.
pub fn systemd_listener() -> Option<std::net::TcpListener> {
    #[cfg(unix)]
    {
        use std::os::fd::FromRawFd;
        let pid: u32 = std::env::var("LISTEN_PID").ok()?.parse().ok()?;
        let fds: u32 = std::env::var("LISTEN_FDS").ok()?.parse().ok()?;
        if pid != std::process::id() || fds < 1 {
            return None;
        }
        // SAFETY: systemd hands this process descriptor 3 as an open,
        // listening socket (sd_listen_fds(3)); nothing else owns it.
        let listener = unsafe { std::net::TcpListener::from_raw_fd(3) };
        listener.set_nonblocking(true).ok()?;
        Some(listener)
    }
    #[cfg(not(unix))]
    None
}

#[cfg(test)]
mod tests {
    /// Blocking work still running when the hub stops does not hold the
    /// process: the runtime ends within its grace.
    #[test]
    fn finishing_does_not_wait_for_blocking_work() {
        let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
        rt.spawn_blocking(|| std::thread::sleep(std::time::Duration::from_secs(30)));
        let started = std::time::Instant::now();
        super::finish(rt);
        assert!(started.elapsed() < super::EXIT_GRACE * 3, "{:?}", started.elapsed());
    }
}

1		//! Stopping without cutting anyone off (oxygenhub#233). On SIGTERM (a
2		//! `systemctl restart`) or Ctrl-C the hub stops taking connections, ends
3		//! its event streams (browsers reconnect by themselves), and gives the
4		//! requests in flight up to `DRAIN` to finish. Given a listening socket by
5		//! systemd (socket activation, docs/deploy.md), it serves that rather than
6		//! binding: the socket outlives the restart, so a request that arrives
7		//! between the old process and the new one waits in its queue instead of
8		//! being refused.
9
10		use std::sync::OnceLock;
11		use tokio::sync::watch;
12
13		/// How long the requests in flight get to finish.
14		pub const DRAIN: std::time::Duration = std::time::Duration::from_secs(30);
15
16		/// How long, after the drain, work still running on the runtime's
17		/// blocking threads (a slow request's repository work, a bucket call) may
18		/// hold the process. Dropping the runtime would wait for it however long
19		/// it takes: on 2026-10-17 a restart stood a minute past "stopped", until
20		/// systemd killed it.
21		pub const EXIT_GRACE: std::time::Duration = std::time::Duration::from_secs(1);
22
23		/// End the runtime once `serve` has returned, cutting off what still runs.
24	1×	pub fn finish(rt: tokio::runtime::Runtime) {
25	1×	rt.shutdown_timeout(EXIT_GRACE);
26	1×	}
27
28	7×	fn sender() -> &'static watch::Sender<bool> {
29		static STOP: OnceLock<watch::Sender<bool>> = OnceLock::new();
30	7×	STOP.get_or_init(\|\| watch::channel(false).0)
31	7×	}
32
33		/// Begin stopping.
34	1×	pub fn begin() {
35	1×	sender().send_replace(true);
36	1×	}
37
38	1×	pub fn is_stopping() -> bool {
39	1×	*sender().borrow()
40	1×	}
41
42		/// Done when the hub begins stopping: for event streams to end on, and for
43		/// the servers' graceful shutdown.
44	5×	pub async fn stopping() {
45	5×	let mut rx = sender().subscribe();
46	5×	let _ = rx.wait_for(\|s\| *s).await;
47	2×	}
48
49		/// Wait for SIGTERM or Ctrl-C, then begin stopping. Installed before the
50		/// servers start, so a signal never kills the process outright.
51	1×	pub fn on_signal() -> anyhow::Result<impl std::future::Future<Output = ()>> {
52		#[cfg(unix)]
53	1×	let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
54	1×	Ok(async move {
55		#[cfg(unix)]
56	1×	tokio::select! {
57	1×	_ = term.recv() => tracing::info!("SIGTERM: stopping (requests in flight may take {} s)", DRAIN.as_secs()),
58	1×	_ = tokio::signal::ctrl_c() => tracing::info!("stopping"),
59		}
60		#[cfg(not(unix))]
61		{
62		let _ = tokio::signal::ctrl_c().await;
63		tracing::info!("stopping");
64		}
65	0×	begin();
66	0×	})
67	1×	}
68
69		/// The listening socket systemd passed (LISTEN_FDS, first descriptor 3),
70		/// when it passed one to this process.
71	1×	pub fn systemd_listener() -> Option<std::net::TcpListener> {
72		#[cfg(unix)]
73		{
74		use std::os::fd::FromRawFd;
75	1×	let pid: u32 = std::env::var("LISTEN_PID").ok()?.parse().ok()?;
76	0×	let fds: u32 = std::env::var("LISTEN_FDS").ok()?.parse().ok()?;
77	0×	if pid != std::process::id() \|\| fds < 1 {
78	0×	return None;
79	0×	}
80		// SAFETY: systemd hands this process descriptor 3 as an open,
81		// listening socket (sd_listen_fds(3)); nothing else owns it.
82	0×	let listener = unsafe { std::net::TcpListener::from_raw_fd(3) };
83	0×	listener.set_nonblocking(true).ok()?;
84	0×	Some(listener)
85		}
86		#[cfg(not(unix))]
87		None
88	1×	}
89
90		#[cfg(test)]
91		mod tests {
92		/// Blocking work still running when the hub stops does not hold the
93		/// process: the runtime ends within its grace.
94		#[test]
95	1×	fn finishing_does_not_wait_for_blocking_work() {
96	1×	let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
97	1×	rt.spawn_blocking(\|\| std::thread::sleep(std::time::Duration::from_secs(30)));
98	1×	let started = std::time::Instant::now();
99	1×	super::finish(rt);
100	1×	assert!(started.elapsed() < super::EXIT_GRACE * 3, "{:?}", started.elapsed());
101	1×	}
102		}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/hub/src/shutdown.rs 80.0%

Keyboard