didbot_serve/

stale_sweep.rs

//! The timer that runs `didbot_pds::Provisioner::sweep_stale_bounded`.
//!
//! A process that stops between storing an account and activating it — a
//! `SIGKILL`, a lost machine, an OOM kill — leaves a row behind that nothing
//! else comes back for. The sweep removes that row, whatever the account's
//! kind, once it is older than [`DEFAULT_UNFINISHED_AGE`], and its hostname
//! answers 404 from then on the way every name no account holds does. It
//! also finishes an erasure a crash left half done. Every other account
//! stays, whatever its age.
//!
//! # Why a timer rather than a startup pass
//!
//! Both, really — the first tick fires immediately, so a restart reaps
//! without waiting an interval, and the ticks after it cover a process that
//! stays up for weeks. A startup-only pass would leave a long-lived server
//! accumulating stale rows until somebody restarted it, which is the
//! opposite of what a deployment that must not be restarted wants.
//!
//! # What one tick costs
//!
//! Each deletion takes the write lock every request path takes.
//! [`SweepSchedule::limit`] is what keeps a deployment that has accumulated
//! a backlog from holding it for one tick of thousands of deletions back to
//! back; what a tick leaves behind is taken by the next one.

use std::panic::AssertUnwindSafe;
use std::sync::Arc;
use std::time::Duration as StdDuration;

use didbot_pds::StaleSweep;
use tokio::sync::Notify;
use tokio::task::JoinHandle;
use tokio::time::MissedTickBehavior;
use tracing::{info, warn};

/// How often the sweep runs, absent configuration.
///
/// An hour. [`DEFAULT_UNFINISHED_AGE`] is days rather than minutes, so the
/// interval is what bounds how long a stranded row stays past that window —
/// an hour of it is well inside the noise of a window measured in days, and
/// cheap enough to run on a server that is otherwise idle.
pub const DEFAULT_SWEEP_INTERVAL: StdDuration = StdDuration::from_secs(60 * 60);

/// How old a row that never became an account gets before the sweep
/// removes it, absent configuration.
///
/// Seven days. Provisioning finishes in one call, so a row still `reserved`
/// or `provisioning` a week later was left by a process that died.
pub const DEFAULT_UNFINISHED_AGE: time::Duration = time::Duration::days(7);

/// How many rows one tick takes, absent configuration.
///
/// Sized against the write lock each removal takes: a hundred removals is a
/// sweep that finishes well inside a request's patience, and a backlog
/// larger than that is drained an hour at a time.
pub const DEFAULT_SWEEP_LIMIT: usize = 100;

/// When the sweep runs and how much it takes.
///
/// The three together because they are one decision: how quickly this
/// deployment reclaims what a killed process left behind, and how long it
/// is willing to hold its own write lock doing so.
#[derive(Debug, Clone, Copy)]
pub struct SweepSchedule {
    /// How often the sweep fires.
    pub interval: StdDuration,
    /// How old a row that never became an account gets before the sweep
    /// removes it.
    pub unfinished_age: time::Duration,
    /// The most rows one tick removes.
    pub limit: usize,
}

impl Default for SweepSchedule {
    /// [`DEFAULT_SWEEP_INTERVAL`], [`DEFAULT_UNFINISHED_AGE`] and
    /// [`DEFAULT_SWEEP_LIMIT`].
    fn default() -> Self {
        Self {
            interval: DEFAULT_SWEEP_INTERVAL,
            unfinished_age: DEFAULT_UNFINISHED_AGE,
            limit: DEFAULT_SWEEP_LIMIT,
        }
    }
}

/// A running sweep, stopped by [`StaleSweeper::stop`] or by dropping it.
pub struct StaleSweeper {
    handle: JoinHandle<()>,
    stop: Arc<Notify>,
}

impl StaleSweeper {
    /// Starts sweeping `registry` on `schedule`.
    ///
    /// The first tick fires immediately, so a process that has just come back
    /// from the crash that stranded an account starts reaping at once. That
    /// pass runs alongside serving.
    ///
    /// The sweep itself runs on a blocking thread: it takes the same write
    /// lock every write path takes, and a task doing that on a runtime worker
    /// holds the worker for the duration.
    /// That is also what makes a panicking sweep survivable — the panic
    /// arrives here as a join error, is logged, and the next tick tries again
    /// rather than the timer dying with it.
    pub fn spawn(registry: Arc<dyn StaleSweep>, schedule: SweepSchedule) -> Self {
        let stop = Arc::new(Notify::new());
        let notified = stop.clone();
        let handle = tokio::spawn(async move {
            let mut ticker = tokio::time::interval(schedule.interval);
            ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
            loop {
                tokio::select! {
                    // Biased so a stop that arrives with a due tick wins,
                    // rather than the shutdown waiting out one more sweep.
                    biased;
                    () = notified.notified() => break,
                    _ = ticker.tick() => {}
                }
                sweep_once(registry.clone(), schedule).await;
            }
        });
        Self { handle, stop }
    }

    /// Stops the sweep and waits for the task to finish.
    pub async fn stop(mut self) {
        self.stop.notify_one();
        // Borrowed rather than moved: `Drop` below aborts the handle, and a
        // type that owns a `Drop` cannot give its field away.
        let _ = (&mut self.handle).await;
    }
}

impl Drop for StaleSweeper {
    fn drop(&mut self) {
        self.handle.abort();
    }
}

/// One tick: sweep, say what it took.
///
/// Public and separate from the timer so a caller — a test, or a deployment
/// reaping once from somewhere else — can run exactly one pass without
/// waiting on a clock. The spawned sweep calls nothing else.
pub async fn sweep_once(registry: Arc<dyn StaleSweep>, schedule: SweepSchedule) {
    let swept = tokio::task::spawn_blocking(move || {
        // `catch_unwind` as well as the join error below: a panic inside a
        // blocking task is already surfaced as a join error, and catching it
        // here is what keeps the panic's own message attached to the sweep
        // that raised it rather than to the timer that scheduled it.
        std::panic::catch_unwind(AssertUnwindSafe(|| {
            registry.sweep_stale_bounded(schedule.unfinished_age, schedule.limit)
        }))
    })
    .await;

    match swept {
        Ok(Ok(swept)) if swept.is_empty() => {}
        Ok(Ok(swept)) => info!(
            rows = swept.len(),
            limit = schedule.limit,
            unfinished_age = %schedule.unfinished_age,
            "swept stale rows"
        ),
        // Both arms leave the timer running: whatever refused this pass, the
        // rows it was about to take are still there to take next hour, and a
        // deployment whose sweep has stopped is one accumulating rows nobody
        // finished.
        Ok(Err(_)) => warn!("the stale-account sweep panicked; the next tick tries again"),
        Err(error) => {
            warn!(%error, "the stale-account sweep did not finish; the next tick tries again")
        }
    }
}