wsb-daemon 0.258.6441

//! Keeps the logs directory under its limit, and hands the past back on start.
//!
//! Two jobs, both about the same directory:
//!
//! * On startup, read the operation logs already on disk into history, so a
//!   restarted daemon still knows about the build somebody ran before it was
//!   restarted — and then sweep, because the files that accumulated while it
//!   was not running are still files.
//! * After every operation, sweep. An operation is the only thing that makes
//!   the directory grow by a meaningful amount, so it is the only moment worth
//!   checking; a timer would either check when nothing changed or miss the
//!   build that filled the disk.
//!
//! The sweep is the policy in `wsb::logs`: oldest entries first, whole entries
//! only, and never one that is being written. This module decides WHEN, not
//! what.


use super::ops::OpsEvent;
use super::state::WebState;

/// How many operation logs are read back into history at startup.
///
/// The same order of magnitude as the in-memory history ring: what a page
/// lists is "recent operations", and a year of them is a directory listing
/// rather than a list anybody scrolls. The files are all still there, and all
/// still readable — this is how many the workbench offers without being asked.
const ADOPT_LIMIT: usize = 200;

/// The shortest gap between two sweeps.
///
/// Ten operations finishing in the same second is one directory walk, not ten.
const SWEEP_COOLDOWN: std::time::Duration = std::time::Duration::from_secs(20);

/// Start the janitor. Returns immediately; the work happens in the background.
pub fn spawn(state: WebState) {
    tokio::spawn(async move {
        let adopted = state.ops.adopt_recent_logs(ADOPT_LIMIT).await;
        if adopted > 0 {
            tracing::info!(adopted, "read earlier operation logs back into history");
        }
        sweep_now(&state).await;

        let mut events = state.ops.subscribe();
        let mut last_sweep = std::time::Instant::now();
        loop {
            let event = match events.recv().await {
                Ok(event) => event,
                // Lagging behind ops events is not a reason to stop sweeping:
                // the missed events were operations finishing, which is
                // exactly when a sweep is due.
                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            };
            if !matches!(event, OpsEvent::Completed(_) | OpsEvent::Failed(_)) {
                continue;
            }
            if last_sweep.elapsed() < SWEEP_COOLDOWN {
                continue;
            }
            last_sweep = std::time::Instant::now();
            sweep_now(&state).await;
        }
    });
}

/// Sweep the logs directory, right now.
///
/// Public because lowering the limit in the settings page must take effect
/// immediately: a limit that waits for the next operation is one somebody
/// lowers to reclaim disk and then watches reclaim nothing.
pub async fn sweep_now(state: &WebState) {
    sweep_logs(state).await;
}

async fn sweep_logs(state: &WebState) {
    let Some(root) = state.ops.logs_root().map(std::path::Path::to_path_buf) else {
        return;
    };
    // What the registry is writing right now. Read before the sweep and passed
    // into it, so an operation that starts during the walk is at worst swept
    // on the NEXT pass — never mid-write, because a file that appeared after
    // the listing is not in the listing.
    let protected = state.ops.protected_logs().await;
    let max_bytes = super::handlers::settings::configured_max_bytes();

    let report = match tokio::task::spawn_blocking(move || {
        wsb::logs::sweep(&root, max_bytes, &protected, false)
    })
    .await
    {
        Ok(report) => report,
        Err(error) => {
            tracing::warn!(%error, "the log sweep panicked");
            return;
        }
    };

    if !report.removed.is_empty() {
        tracing::info!(
            removed = report.removed.len(),
            freed = %wsb::logs::human_bytes(report.freed_bytes),
            remaining = %wsb::logs::human_bytes(report.remaining_bytes),
            limit = %wsb::logs::human_bytes(max_bytes),
            "swept old logs"
        );
    }
    if report.still_over {
        // Said out loud, because the limit is not the thing that will fix it.
        //
        // What is left is over the limit for one of two reasons, and they need
        // different answers: it is being WRITTEN (wait, or stop whatever is
        // writing it), or it is bytes no sweep can REACH — a directory from a
        // layout that predated the streams, a file somebody parked at the
        // root. This root held 174 MB of the second kind, in 239 dated
        // directories written before `test/` existed, and the old message told
        // whoever read it that the logs were "in use". They were not; they
        // were unreachable, and no limit anyone could set would ever have
        // taken a byte of them.
        tracing::warn!(
            remaining = %wsb::logs::human_bytes(report.remaining_bytes),
            limit = %wsb::logs::human_bytes(max_bytes),
            protected = report.protected,
            unreachable = %wsb::logs::human_bytes(report.unreachable_bytes),
            "the logs directory is over its limit; what is left is in use or \
             is not something a sweep manages"
        );
    }
}