haematite-cli 0.21.7

use super::{Config, Launch, Phase, Shared};
use std::io::{self, BufRead, BufReader, Write};
use std::sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
};

#[cfg(test)]
macro_rules! eprintln {
    ($($argument:tt)*) => { super::ready_tests::observe(format_args!($($argument)*)) };
}

const LINE_BYTES: usize = 65_536;

struct Log<W> {
    writer: W,
    failed: Arc<AtomicBool>,
}
impl<W: Write> Log<W> {
    fn append(&mut self, bytes: &[u8]) -> Option<io::Error> {
        match self
            .writer
            .write_all(bytes)
            .and_then(|()| self.writer.flush())
        {
            Ok(()) => {
                self.failed.store(false, Ordering::Release);
                None
            }
            Err(error) => {
                // One full log is one incident until a later write succeeds.
                (!self.failed.swap(true, Ordering::AcqRel)).then_some(error)
            }
        }
    }
}

pub(super) fn read_lines(
    shared: &Shared,
    index: usize,
    launch: Launch,
    config: &Config,
    pipe: impl io::Read,
    log: impl Write,
    failed: Arc<AtomicBool>,
) -> Result<(), String> {
    let mut log = Log {
        writer: log,
        failed,
    };
    drain(&mut BufReader::new(pipe), &mut log, &config.name, |line| {
        if !line.starts_with(config.ready.as_bytes()) {
            return Ok(());
        }
        let mut state = shared
            .state
            .lock()
            .map_err(|e| format!("door_children_state_failed: {e}"))?;
        let entry = &mut state.entries[index];
        if entry.generation != launch.generation || entry.phase != Phase::NotReady {
            return Ok(());
        }
        entry.pending_ready = entry
            .pending_ready
            .checked_add(1)
            .ok_or("door_children_state_failed: ready announcements exhausted")?;
        entry.phase = Phase::Ready;
        let line = format!(
            "door_child_ready: {} elapsed_ms={}",
            config.name,
            launch.started.elapsed().as_millis()
        );
        drop(state);
        eprintln!("{line}");
        let mut state = shared
            .state
            .lock()
            .map_err(|e| format!("door_children_state_failed: {e}"))?;
        let entry = &mut state.entries[index];
        // Publishing follows output even if this launch has already exited.
        entry.seen_ready = true;
        entry.pending_ready -= 1;
        drop(state);
        shared.changed.notify_all();
        Ok(())
    })
}

fn drain(
    reader: &mut impl BufRead,
    log: &mut Log<impl Write>,
    name: &str,
    mut line_ready: impl FnMut(&[u8]) -> Result<(), String>,
) -> Result<(), String> {
    let mut line = Vec::with_capacity(LINE_BYTES + 1);
    let mut discarding = false;
    loop {
        let bytes = reader
            .fill_buf()
            .map_err(|error| format!("door_child_stderr_failed: {name}: {error}"))?;
        if bytes.is_empty() {
            if !discarding && !line.is_empty() {
                write_log(log, &line, name);
                line_ready(&line)?;
            }
            return Ok(());
        }
        let newline = bytes.iter().position(|byte| *byte == b'\n');
        let payload = newline.unwrap_or(bytes.len());
        let consumed = payload + usize::from(newline.is_some());
        if !discarding {
            let retained = payload.min(LINE_BYTES - line.len());
            line.extend_from_slice(&bytes[..retained]);
            if retained < payload {
                write_log(log, &line, name);
                line.clear();
                discarding = true;
                eprintln!(
                    "door_child_stderr_line_too_long: {name}: discarded beyond {LINE_BYTES} bytes"
                );
            } else if newline.is_some() {
                line.push(b'\n');
                write_log(log, &line, name);
                line_ready(&line)?;
                line.clear();
            }
        }
        reader.consume(consumed);
        if newline.is_some() {
            discarding = false;
        }
    }
}

fn write_log(log: &mut Log<impl Write>, bytes: &[u8], name: &str) {
    if let Some(error) = log.append(bytes) {
        eprintln!("door_child_log_failed: {name}: {error}");
    }
}

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