1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
//! One wiring pass, on request: what lets a Windows launcher shim (`hooks::cli_shim`) wire a CLI
//! installed after the daemon started on that CLI's very first run, instead of waiting out the
//! supervisor's tick.
//!
//! `POST /admin/wiring/refresh` calls [`WiringRequests::request`] and awaits the pass that STARTED
//! after it ([`WiringRequests::wait_for`]): a pass already running when the request lands detected
//! its agents before the CLI existed, so it does not count. The supervisor reads the request
//! number as a pass starts and publishes it once the pass is done; the [`Notify`] permit wakes it
//! from its tick wait, or right after a pass that was running.
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use tokio::sync::{watch, Notify};
/// The wiring supervisor's request inbox. One per daemon, shared with the admin router.
#[derive(Debug)]
pub struct WiringRequests {
requested: AtomicU64,
wake: Notify,
completed: watch::Sender<u64>,
}
impl Default for WiringRequests {
fn default() -> Self {
Self {
requested: AtomicU64::new(0),
wake: Notify::new(),
completed: watch::Sender::new(0),
}
}
}
impl WiringRequests {
/// Ask for a pass. Returns the number [`Self::wait_for`] awaits.
pub fn request(&self) -> u64 {
let seq = self.requested.fetch_add(1, Ordering::SeqCst) + 1;
self.wake.notify_one();
seq
}
/// Resolves once a pass is requested (immediately when one was asked for since the last wait).
pub async fn requested(&self) {
self.wake.notified().await;
}
/// Called as a pass starts: the requests this pass will answer.
pub fn pass_starting(&self) -> u64 {
self.requested.load(Ordering::SeqCst)
}
/// Called once that pass is done, with what [`Self::pass_starting`] returned.
pub fn pass_done(&self, answered: u64) {
self.completed.send_if_modified(|c| {
let newer = answered > *c;
if newer {
*c = answered;
}
newer
});
}
/// Whether a pass answering request `seq` completed within `budget`.
pub async fn wait_for(&self, seq: u64, budget: Duration) -> bool {
let mut rx = self.completed.subscribe();
tokio::time::timeout(budget, rx.wait_for(|c| *c >= seq))
.await
.is_ok_and(|r| r.is_ok())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
#[tokio::test]
async fn a_pass_running_when_the_request_lands_does_not_answer_it() {
let r = Arc::new(WiringRequests::default());
let running = r.pass_starting(); // started before the request
let seq = r.request();
r.pass_done(running);
assert!(!r.wait_for(seq, Duration::from_millis(50)).await);
r.requested().await; // the permit left by `request`
let next = r.pass_starting();
r.pass_done(next);
assert!(r.wait_for(seq, Duration::from_millis(50)).await);
}
#[tokio::test]
async fn a_wait_ends_when_the_pass_completes() {
let r = Arc::new(WiringRequests::default());
let seq = r.request();
let supervisor = {
let r = r.clone();
tokio::spawn(async move {
r.requested().await;
let n = r.pass_starting();
tokio::time::sleep(Duration::from_millis(20)).await;
r.pass_done(n);
})
};
assert!(r.wait_for(seq, Duration::from_secs(5)).await);
supervisor.await.unwrap();
}
#[tokio::test]
async fn no_pass_means_the_bound_answers() {
let r = WiringRequests::default();
let seq = r.request();
assert!(!r.wait_for(seq, Duration::from_millis(30)).await);
}
}