didbot_serve/
confinement_poll.rs

1//! The confinement poll: re-derives every inherited lock from the tree of
2//! principals on a schedule.
3//!
4//! A parent's vouch for a child can lapse without anything on this server
5//! being asked — the record is deleted from the parent's repository.
6//! [`Registry::reconcile_confinement`] is what notices,
7//! and this task is what calls it: the same shape as
8//! [`crate::stale_sweep`], one blocking pass per tick, a panic contained to
9//! the tick it happened in.
10
11use std::panic::AssertUnwindSafe;
12use std::sync::Arc;
13use std::time::Duration as StdDuration;
14
15use didbot_pds::Registry;
16use tokio::sync::Notify;
17use tokio::task::JoinHandle;
18use tokio::time::MissedTickBehavior;
19use tracing::{info, warn};
20
21/// How often the tree is reconciled unless configured otherwise.
22///
23/// The operator poll's interval: a lapsed vouch is the same kind of fact
24/// as a lapsed operator claim, noticed on the same clock.
25pub const DEFAULT_CONFINEMENT_INTERVAL: StdDuration = StdDuration::from_secs(5 * 60);
26
27/// The running poll, stopped on shutdown.
28pub struct ConfinementPoll {
29    handle: JoinHandle<()>,
30    stop: Arc<Notify>,
31}
32
33impl ConfinementPoll {
34    /// Starts reconciling every `interval`.
35    pub fn spawn(registry: Arc<dyn Registry>, interval: StdDuration) -> Self {
36        let stop = Arc::new(Notify::new());
37        let notified = stop.clone();
38        let handle = tokio::spawn(async move {
39            let mut ticker = tokio::time::interval(interval);
40            ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
41            loop {
42                tokio::select! {
43                    biased;
44                    () = notified.notified() => break,
45                    _ = ticker.tick() => {}
46                }
47                reconcile_once(registry.clone()).await;
48            }
49        });
50        Self { handle, stop }
51    }
52
53    /// Stops the poll and waits for the tick in flight, if any.
54    pub async fn stop(mut self) {
55        self.stop.notify_one();
56        let _ = (&mut self.handle).await;
57    }
58}
59
60impl Drop for ConfinementPoll {
61    fn drop(&mut self) {
62        self.handle.abort();
63    }
64}
65
66/// One pass, off the async runtime.
67pub async fn reconcile_once(registry: Arc<dyn Registry>) {
68    let outcome = tokio::task::spawn_blocking(move || {
69        std::panic::catch_unwind(AssertUnwindSafe(|| registry.reconcile_confinement()))
70    })
71    .await;
72    match outcome {
73        Ok(Ok(confinement)) => {
74            if !confinement.quarantined.is_empty() {
75                warn!(
76                    accounts = ?confinement.quarantined,
77                    "quarantined accounts whose parent's vouch has lapsed"
78                );
79            }
80            if !confinement.released.is_empty() {
81                info!(
82                    accounts = ?confinement.released,
83                    "released accounts whose parent's vouch returned"
84                );
85            }
86        }
87        Ok(Err(_)) => warn!("the confinement poll panicked; the next tick tries again"),
88        Err(error) => {
89            warn!(%error, "the confinement poll did not finish; the next tick tries again")
90        }
91    }
92}
didbot_serve/confinement_poll.rs

didbot_serve/
confinement_poll.rs
