decdn_node/
stop_handle.rs

1//! A background task's `CancellationToken` paired with its `JoinHandle` in one
2//! type, so cancel-before-join is structural rather than call-site discipline.
3//!
4//! The receipt writer ([`crate::receipt_log::spawn_receipt_writer`]) and the
5//! warming-credit aggregator
6//! ([`crate::warming_allowance::spawn_warming_creditor`]) are each a drain loop
7//! whose queue senders live inside long-lived holders such as
8//! [`crate::handlers::client::ClientHandler`]. Awaiting such a task's bare
9//! `JoinHandle` without first cancelling its token hangs until every sender
10//! drops — which for those holders is never. [`StopHandle`] carries the token
11//! and the handle as one value, and [`StopHandle::shutdown`] — the only way to
12//! join the task — always cancels before it awaits.
13
14use std::future::Future;
15
16use tokio::task::{AbortHandle, JoinError, JoinHandle};
17use tokio_util::sync::CancellationToken;
18
19/// A spawned background task and the `CancellationToken` that stops it.
20///
21/// [`StopHandle::spawn`] hands the task the same token the handle stores, so
22/// the pairing holds by construction. Dropping the handle detaches the task
23/// (like dropping a bare `JoinHandle`); stop it with [`StopHandle::shutdown`]
24/// instead.
25#[derive(Debug)]
26pub struct StopHandle {
27    shutdown: CancellationToken,
28    handle: JoinHandle<()>,
29}
30
31impl StopHandle {
32    /// Spawn `task` onto the runtime with a fresh stop token. The closure
33    /// receives the token this handle later cancels, so the task observes
34    /// exactly the signal [`Self::cancel`] and [`Self::shutdown`] fire.
35    pub fn spawn<F, Fut>(task: F) -> Self
36    where
37        F: FnOnce(CancellationToken) -> Fut,
38        Fut: Future<Output = ()> + Send + 'static,
39    {
40        let shutdown = CancellationToken::new();
41        let handle = tokio::spawn(task(shutdown.clone()));
42        Self { shutdown, handle }
43    }
44
45    /// Signal the task to stop without awaiting it. Idempotent —
46    /// [`Self::shutdown`] cancels again as a no-op. Use this to start a drain
47    /// early and overlap it with other teardown work before the final
48    /// `shutdown().await`.
49    pub fn cancel(&self) {
50        self.shutdown.cancel();
51    }
52
53    /// Last-resort abort for a drain that overruns its deadline: dropping a
54    /// timed-out [`Self::shutdown`] future merely detaches the task, so the
55    /// caller takes this handle first and fires it when the timeout lands.
56    #[must_use]
57    pub fn abort_handle(&self) -> AbortHandle {
58        self.handle.abort_handle()
59    }
60
61    /// Stop the task and await it: cancel the token, then join. The cancel is
62    /// unconditional and comes first, so this cannot hang on queue senders
63    /// that outlive shutdown the way a bare `JoinHandle` await can.
64    pub async fn shutdown(self) -> Result<(), JoinError> {
65        self.shutdown.cancel();
66        self.handle.await
67    }
68}
decdn_node/stop_handle.rs

decdn_node/
stop_handle.rs
