runtime/backlog/
tail.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4// Copyright (c) 2026 2youg1 and the sprawling contributors
5
6//! What a running command has written since the last poll, handed to a
7//! sink while the command still runs (`crates/runtime/spec/Backlog.lean` §8-28-3).
8//!
9//! A child writes to two files, never to pipes, so reading its output
10//! live needs nobody to change how it writes: the window's poll loop
11//! reads what the files grew by since its last visit. What it reads is
12//! a preview a page may drop. The tool result that lands in the ledger
13//! is the one authority on this output, so a failed read is not a
14//! failure of the command, only a piece the page sees one poll later.
15
16use std::io::{Read, Seek, SeekFrom};
17use std::sync::Arc;
18
19use kernel::RunId;
20
21use super::BacklogId;
22
23/// Which of a command's two outputs a piece came from.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum Stream {
26    Out,
27    Err,
28}
29
30/// One piece of output, in the order it was written to its stream.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct Chunk {
33    pub run: RunId,
34    pub member: BacklogId,
35    pub stream: Stream,
36    pub bytes: Vec<u8>,
37}
38
39/// Where the pieces go. A backlog without one reads nothing at all, so a
40/// city nobody watches pays for none of this.
41#[derive(Clone)]
42pub struct Sink(Arc<dyn Fn(Chunk) + Send + Sync>);
43
44impl Sink {
45    #[must_use]
46    pub fn new(deliver: impl Fn(Chunk) + Send + Sync + 'static) -> Sink {
47        Sink(Arc::new(deliver))
48    }
49
50    pub(super) fn deliver(&self, chunks: Vec<Chunk>) {
51        chunks.into_iter().for_each(|chunk| (self.0)(chunk));
52    }
53}
54
55/// How far one member's two files have been read.
56#[derive(Debug, Default)]
57pub(super) struct Tail {
58    out: u64,
59    err: u64,
60}
61
62impl Tail {
63    /// Takes what each file grew by since the last take, at most half
64    /// of `per_poll` bytes from each, so a flood on stdout never starves
65    /// stderr and a command that writes faster than a page reads leaves
66    /// the page behind rather than the poll slower. The caller delivers
67    /// them, which lets it release the table first.
68    pub(super) fn take(
69        &mut self,
70        dir: &std::path::Path,
71        from: (RunId, BacklogId),
72        per_poll: usize,
73    ) -> Vec<Chunk> {
74        let half = per_poll.div_ceil(2);
75        let mut chunks = Vec::new();
76        for (stream, name, offset) in [
77            (Stream::Out, "out", &mut self.out),
78            (Stream::Err, "err", &mut self.err),
79        ] {
80            let bytes = grown(&dir.join(name), *offset, half);
81            if bytes.is_empty() {
82                continue;
83            }
84            *offset = offset.saturating_add(u64::try_from(bytes.len()).unwrap_or(u64::MAX));
85            chunks.push(Chunk {
86                run: from.0,
87                member: from.1,
88                stream,
89                bytes,
90            });
91        }
92        chunks
93    }
94}
95
96/// Up to `cap` bytes of `file` past `offset`; nothing when the file
97/// cannot be read, because the next poll asks again from the same place.
98fn grown(file: &std::path::Path, offset: u64, cap: usize) -> Vec<u8> {
99    let mut bytes = Vec::new();
100    let limit = u64::try_from(cap).unwrap_or(u64::MAX);
101    let read = std::fs::File::open(file).and_then(|mut opened| {
102        opened.seek(SeekFrom::Start(offset))?;
103        opened.take(limit).read_to_end(&mut bytes)
104    });
105    match read {
106        Ok(_) => bytes,
107        Err(_) => Vec::new(),
108    }
109}
runtime/backlog/tail.rs

runtime/backlog/
tail.rs
