wsb-daemon 0.248.6265

//! The host, sampled for the strip in the title bar.
//!
//! One stream rather than one route polled: the page wants a reading every
//! couple of seconds, and a poll costs a request, a connection and a set of
//! headers for every one of them. The sampler lives here rather than in the
//! page because the page cannot read `/proc`, and because one sampler serving
//! every open tab costs what one costs.

use axum::{
    extract::State,
    response::sse::{Event, KeepAlive, Sse},
    Json,
};
use futures::stream::Stream;
use wsb::prelude::*;
use std::convert::Infallible;

use crate::web::machine_web::MachineWeb;

/// How often the host is read.
///
/// Two seconds: fast enough that a build starting is visible while somebody is
/// still looking at the strip, slow enough that the reading costs nothing. A
/// processor figure is a delta over this interval, so it is also the window
/// the number describes.
const INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);

/// GET /api/v1/host — one reading, for a page that just opened.
///
/// The processor figure is absent here by construction: it is a delta, and a
/// single request has nothing to compare against. The page gets it from the
/// second frame of the stream, moments later.
pub async fn host_snapshot(State(state): State<MachineWeb>) -> Json<wsb::host::Sample> {
    let root = state.volume_root.clone();
    let sample = tokio::task::spawn_blocking(move || {
        let mut sampler = wsb::host::Sampler::new();
        sampler.sample(&root)
    })
    .await
    .unwrap_or_default();
    Json(sample)
}

/// GET /api/v1/host/stream — a reading every couple of seconds.
pub async fn host_stream(
    State(state): State<MachineWeb>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let root = state.volume_root.clone();
    let stream = async_stream::stream! {
        // One sampler per subscriber, because the processor figure is a delta
        // against the LAST reading this subscriber saw. Shared, a second tab
        // opening would reset the first tab's baseline.
        let mut sampler = wsb::host::Sampler::new();
        loop {
            let root = root.clone();
            // Off the runtime: reading /proc is cheap, but `nvidia-smi` and
            // `df` are processes, and a stalled disk must not stall the
            // server.
            let (next, sample) = match tokio::task::spawn_blocking(move || {
                let mut moved = sampler;
                let sample = moved.sample(&root);
                (moved, sample)
            })
            .await
            {
                Ok(pair) => pair,
                Err(error) => {
                    tracing::warn!(%error, "the host sampler panicked");
                    break;
                }
            };
            sampler = next;

            match serde_json::to_string(&sample) {
                Ok(payload) => yield Ok(Event::default().event("host").data(payload)),
                Err(error) => {
                    tracing::warn!(%error, "a host sample could not be encoded");
                    break;
                }
            }
            tokio::time::sleep(INTERVAL).await;
        }
    };
    Sse::new(stream).keep_alive(KeepAlive::default())
}