openlatch-client 0.6.15

//! 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);
    }
}