wsb-daemon 0.258.6441

//! One log file, followed while it is still being written.
//!
//! A run's bulk output does not go into the op log: a suite writes its own
//! file, and the op log carries the TRANSITIONS between steps. That split is
//! what keeps an op log small enough to keep, but it also means a step that
//! takes twelve minutes to compile CUDA kernels emits nothing at all until it
//! is over — the panel showed a run as a sequence of things that had already
//! happened, never the thing happening now.
//!
//! The file itself was never the problem: it is appended to from the step's
//! first line. This serves it the way `tail -f` does — what is there already,
//! then whatever arrives — so the panel can show a step's output while the
//! step is still producing it.

use axum::{
    extract::{Query, State},
    http::StatusCode,
    response::sse::{Event, KeepAlive, Sse},
    Json,
};
use futures::stream::Stream;
use wsb::prelude::*;
use serde::Deserialize;
use std::convert::Infallible;
use std::io::SeekFrom;
use std::path::{Path, PathBuf};
use tokio::io::{AsyncReadExt, AsyncSeekExt};

use crate::web::state::WebState;

/// How often the file is checked for new bytes.
///
/// A log is appended to by a process on this machine, so there is nothing to
/// wait for but the disk. Half a second reads as immediate to a person and
/// costs one `stat` per subscriber.
const POLL: std::time::Duration = std::time::Duration::from_millis(500);

/// How much of an existing file is replayed when a subscriber attaches.
///
/// A cargo build log runs to megabytes and a person opening a running step
/// wants its END. Everything before this is already on disk for anyone who
/// wants the whole of it; the path is in the record.
const TAIL_BYTES: u64 = 64 * 1024;

#[derive(Debug, Deserialize)]
pub struct TailQuery {
    /// The log, as the workspace addresses it — the `log` field of a
    /// `test_started` or `test_finished` record, verbatim.
    pub path: String,
}

/// Resolve a requested path to a real file inside the logs root.
///
/// A path arriving from a browser names a file this process will read and
/// send, so it is resolved and then CHECKED, never trusted: `..` is not
/// rejected by pattern (a pattern misses `%2e%2e`, symlinks, and the absolute
/// path that happens to start with the right prefix) but defeated by
/// canonicalising both sides and requiring containment. A log that does not
/// exist yet is not an error the caller can distinguish from one that does —
/// both answer "nothing yet" — so only escapes are refused.
pub fn resolve_in_logs(workspace_root: &Path, requested: &str) -> Result<PathBuf, String> {
    let logs_root = wsb::platform::canonicalize(wsb::logs::root(workspace_root))
        .map_err(|error| format!("the logs root cannot be read: {error}"))?;

    let candidate = if Path::new(requested).is_absolute() {
        PathBuf::from(requested)
    } else {
        workspace_root.join(requested)
    };

    // The file may not exist yet — a step that has just started has named its
    // log before writing a byte of it. Canonicalise the deepest ancestor that
    // does exist and re-attach the rest, so containment is decided on a real
    // path either way.
    let (existing, rest) = deepest_existing(&candidate);
    let existing = wsb::platform::canonicalize(&existing)
        .map_err(|error| format!("{} cannot be resolved: {error}", existing.display()))?;
    let resolved = existing.join(rest);

    if !resolved.starts_with(&logs_root) {
        return Err(format!(
            "{requested} is not inside the logs root — only a run's own logs can be followed"
        ));
    }
    Ok(resolved)
}

/// Split `path` into its deepest existing ancestor and the remainder.
fn deepest_existing(path: &Path) -> (PathBuf, PathBuf) {
    let mut existing = path.to_path_buf();
    let mut rest = PathBuf::new();
    while !existing.exists() {
        let Some(name) = existing.file_name().map(|n| n.to_os_string()) else {
            break;
        };
        let Some(parent) = existing.parent().map(|p| p.to_path_buf()) else {
            break;
        };
        rest = if rest.as_os_str().is_empty() {
            PathBuf::from(&name)
        } else {
            Path::new(&name).join(&rest)
        };
        existing = parent;
    }
    (existing, rest)
}

pub async fn tail_stream(
    State(state): State<WebState>,
    Query(query): Query<TailQuery>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, Json<serde_json::Value>)>
{
    let path = resolve_in_logs(&state.workspace_root, &query.path).map_err(|error| {
        (
            StatusCode::BAD_REQUEST,
            Json(serde_json::json!({ "error": error })),
        )
    })?;

    let stream = async_stream::stream! {
        // Where this subscriber starts reading: the last TAIL_BYTES of what is
        // already there, or the whole file when it is shorter.
        let mut offset = match tokio::fs::metadata(&path).await {
            Ok(meta) => meta.len().saturating_sub(TAIL_BYTES),
            Err(_) => 0,
        };
        // A partial last line is held back rather than sent: a log is appended
        // to in whatever sizes the writer flushes, and half a line rendered as
        // a line is a line the file never contained.
        let mut pending = String::new();
        let mut missing_reported = false;

        loop {
            match read_from(&path, offset).await {
                Ok((text, next)) => {
                    offset = next;
                    if !text.is_empty() {
                        pending.push_str(&text);
                        let mut lines: Vec<String> = Vec::new();
                        while let Some(at) = pending.find('\n') {
                            let line: String = pending.drain(..=at).collect();
                            lines.push(line.trim_end_matches(['\n', '\r']).to_string());
                        }
                        if !lines.is_empty() {
                            let payload = serde_json::json!({
                                "event": "lines",
                                "lines": lines,
                                "offset": offset,
                            });
                            yield Ok(Event::default().event("tail").data(payload.to_string()));
                        }
                    }
                }
                Err(error) => {
                    // A log that is not there yet is the ordinary case for a
                    // step that has just started, and it is said ONCE — a
                    // message per poll would be a spinner made of errors.
                    if !missing_reported {
                        missing_reported = true;
                        let payload = serde_json::json!({
                            "event": "waiting",
                            "reason": error.to_string(),
                        });
                        yield Ok(Event::default().event("tail").data(payload.to_string()));
                    }
                }
            }
            tokio::time::sleep(POLL).await;
        }
    };

    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}

/// Read whatever is in `path` beyond `offset`.
///
/// A truncated or replaced file (a rerun writing the same name) is detected by
/// its length going backwards, and restarts from its beginning rather than
/// seeking past its end and reporting silence forever.
async fn read_from(path: &Path, offset: u64) -> std::io::Result<(String, u64)> {
    let mut file = tokio::fs::File::open(path).await?;
    let len = file.metadata().await?.len();
    let from = if len < offset { 0 } else { offset };
    if len == from {
        return Ok((String::new(), from));
    }
    file.seek(SeekFrom::Start(from)).await?;
    let mut buffer = Vec::with_capacity((len - from) as usize);
    file.read_to_end(&mut buffer).await?;
    let read = buffer.len() as u64;
    Ok((String::from_utf8_lossy(&buffer).into_owned(), from + read))
}

#[cfg(test)]
#[path = "log_tail_tests.rs"]
mod log_tail_tests;