basis_tasks/
stop.rs

1//! A host's word that it is going away.
2//!
3//! A process that drives tasks on behalf of others — the A2A server is the
4//! first — outlives none of them on purpose: when it shuts down, every task
5//! it was driving must be left exactly as resumable as a crash would leave
6//! it, and nothing it was in the middle of may be recorded as an outcome. A
7//! turn cut short by a shutdown did not finish, and a turn that finished
8//! after the host stopped listening answered nobody.
9//!
10//! [`Stop`] is that word. Given to [`Tasks::with_stop`](crate::Tasks::with_stop),
11//! it is honored at once rather than at the next turn boundary: the turn in
12//! flight is abandoned where it stands (its conversation keeps what it
13//! committed, which the next attach continues or re-drives), no terminal
14//! record is written, and every wait driving under it returns.
15
16use std::sync::Arc;
17
18use tokio::sync::watch;
19
20/// A one-way switch from running to stopping, shared by every clone.
21#[derive(Clone, Debug)]
22pub struct Stop {
23    stopping: Arc<watch::Sender<bool>>,
24}
25
26impl Default for Stop {
27    fn default() -> Self {
28        Self::new()
29    }
30}
31
32impl Stop {
33    pub fn new() -> Self {
34        Self {
35            stopping: Arc::new(watch::channel(false).0),
36        }
37    }
38
39    /// Flips the switch. Idempotent; there is no way back.
40    pub fn stop(&self) {
41        self.stopping.send_replace(true);
42    }
43
44    pub fn is_stopping(&self) -> bool {
45        *self.stopping.borrow()
46    }
47
48    /// Resolves once [`stop`](Self::stop) has been called.
49    pub async fn stopping(&self) {
50        let mut receiver = self.stopping.subscribe();
51        // The sender lives as long as `self`, so this cannot fail.
52        let _ = receiver.wait_for(|stopping| *stopping).await;
53    }
54}
55
56#[cfg(test)]
57mod tests {
58    use std::time::Duration;
59
60    use super::Stop;
61
62    #[tokio::test]
63    async fn every_clone_sees_the_stop_and_waiting_on_it_returns() {
64        let stop = Stop::new();
65        let clone = stop.clone();
66        assert!(!clone.is_stopping());
67        let waiting = tokio::spawn(async move { clone.stopping().await });
68        tokio::time::sleep(Duration::from_millis(10)).await;
69        assert!(!waiting.is_finished());
70
71        stop.stop();
72        stop.stop();
73        tokio::time::timeout(Duration::from_secs(5), waiting)
74            .await
75            .expect("the wait returns")
76            .unwrap();
77        assert!(stop.is_stopping());
78        stop.stopping().await;
79    }
80}
basis_tasks/stop.rs

basis_tasks/
stop.rs
