diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs index 1cfdfa83..55f2a383 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -1207,14 +1207,13 @@ async fn set_nspawn_flags( } /// Execute a container operation via hive-priv and integrate with -/// build_logs.sqlite. hive-priv runs as root and logs output to -/// journald as it arrives; this function captures the final stdout + -/// stderr into build_logs for the dashboard after the operation -/// completes. For `create` and `update` (the long-running ops) -/// hive-priv already logs each line to its own journald stream — -/// streaming into build_logs is deferred to a follow-up that adds a -/// streaming mode to the priv protocol. +/// build_logs.sqlite. hive-priv runs as root and forwards output lines +/// to hive-c0re in real time via the streaming priv protocol. Each line +/// is appended to the build-log row as it arrives, so the dashboard +/// shows live progress during long `nixos-container create` / `update` runs. async fn priv_run(kind: &str, name: &str) -> Result<()> { + use hive_sh4re::priv_proto::PrivStream; + let container = container_name(name); let cmdline = format!("nixos-container {kind} {container}"); @@ -1227,35 +1226,78 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> { .ok() }); - let result: Result<(String, String)> = match kind { - "create" => crate::priv_client::create_container(name).await, - "update" => crate::priv_client::update_container(name).await, - "start" => crate::priv_client::start_container(name).await.map(|()| (String::new(), String::new())), - "stop" => crate::priv_client::stop_container(name).await.map(|()| (String::new(), String::new())), - "kill" => crate::priv_client::kill_container(name).await.map(|()| (String::new(), String::new())), - "destroy" => crate::priv_client::destroy_container(name).await.map(|()| (String::new(), String::new())), + // For long-running ops use the streaming protocol so build_logs + // receives lines in real time rather than as a batch at completion. + let result: Result<()> = match kind { + "create" => { + let h = logs.clone(); + let id = log_id; + crate::priv_client::create_container_streaming(name, move |stream, line| { + match stream { + PrivStream::Stdout => { + tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}"); + if let (Some(h), Some(id)) = (&h, id) { + h.append_stdout(id, line); + } + } + PrivStream::Stderr => { + tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}"); + if let (Some(h), Some(id)) = (&h, id) { + h.append_stderr(id, line); + } + } + } + }) + .await + } + "update" => { + let h = logs.clone(); + let id = log_id; + crate::priv_client::update_container_streaming(name, move |stream, line| { + match stream { + PrivStream::Stdout => { + tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}"); + if let (Some(h), Some(id)) = (&h, id) { + h.append_stdout(id, line); + } + } + PrivStream::Stderr => { + tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}"); + if let (Some(h), Some(id)) = (&h, id) { + h.append_stderr(id, line); + } + } + } + }) + .await + } + "start" => crate::priv_client::start_container(name).await, + "stop" => crate::priv_client::stop_container(name).await, + "kill" => crate::priv_client::kill_container(name).await, + "destroy" => crate::priv_client::destroy_container(name).await, other => Err(anyhow::anyhow!("unknown container op: {other}")), }; - let ok = result.is_ok(); + let succeeded = result.is_ok(); if let (Some(h), Some(id)) = (&logs, log_id) { - if let Ok((ref stdout, ref stderr)) = result { - for line in stdout.lines() { - tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}"); - h.append_stdout(id, line); - } - for line in stderr.lines() { - tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}"); - h.append_stderr(id, line); - } - } - h.finish(id, if ok { crate::build_logs::BuildStatus::Ok } else { crate::build_logs::BuildStatus::Fail }); + h.finish( + id, + if succeeded { + crate::build_logs::BuildStatus::Ok + } else { + crate::build_logs::BuildStatus::Fail + }, + ); } match result { - Ok(_) => Ok(()), + Ok(()) => Ok(()), Err(e) => { - let journal = if kind == "update" { container_journal_tail(&container).await } else { String::new() }; + let journal = if kind == "update" { + container_journal_tail(&container).await + } else { + String::new() + }; match log_id { Some(id) => bail!("{e:#}; see build log #{id}{journal}"), None => bail!("{e:#}{journal}"), diff --git a/hive-c0re/src/priv_client.rs b/hive-c0re/src/priv_client.rs index e9b73e45..e78b48d0 100644 --- a/hive-c0re/src/priv_client.rs +++ b/hive-c0re/src/priv_client.rs @@ -8,12 +8,14 @@ use anyhow::{Context as _, Result, bail}; use hive_sh4re::priv_proto::{ - BindMount, JournalOutput, NetworkIsolation, PRIV_SOCK, PrivRequest, PrivResponse, + BindMount, JournalOutput, NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, + PrivStream, }; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; /// Send a single request to `hive-priv` and return the response. +/// For streaming ops use `call_streaming` instead. pub async fn call(req: &PrivRequest) -> Result { let mut stream = UnixStream::connect(PRIV_SOCK) .await @@ -29,7 +31,46 @@ pub async fn call(req: &PrivRequest) -> Result { .read_line(&mut resp_line) .await .context("read response from hive-priv")?; - serde_json::from_str(&resp_line).context("parse PrivResponse") + // New hive-priv sends `PrivEvent::Done(PrivResponse)` (untagged, wire- + // identical to bare `PrivResponse`) — deserialise as `PrivEvent` to + // handle both the old bare format and the new tagged format. + match serde_json::from_str::(&resp_line).context("parse PrivResponse")? { + PrivEvent::Done(resp) => Ok(resp), + PrivEvent::Line(_) => bail!("unexpected stream line from non-streaming priv op"), + } +} + +/// Send a streaming request to `hive-priv`, calling `on_line` for each +/// `PrivEvent::Line` as it arrives, then returning the terminal +/// `PrivResponse`. Used for long-running ops (`create` / `update`). +pub async fn call_streaming( + req: &PrivRequest, + mut on_line: impl FnMut(PrivStream, &str), +) -> Result { + let mut stream = UnixStream::connect(PRIV_SOCK) + .await + .context("connect to hive-priv socket (streaming)")?; + let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n"; + stream + .write_all(line.as_bytes()) + .await + .context("send request to hive-priv")?; + stream.shutdown().await.context("shutdown write half")?; + let mut reader = BufReader::new(stream); + loop { + let mut event_line = String::new(); + reader + .read_line(&mut event_line) + .await + .context("read event from hive-priv")?; + if event_line.is_empty() { + bail!("hive-priv closed connection before sending Done event"); + } + match serde_json::from_str::(&event_line).context("parse PrivEvent")? { + PrivEvent::Line(l) => on_line(l.stream, &l.data), + PrivEvent::Done(resp) => return Ok(resp), + } + } } pub async fn start_container(name: &str) -> Result<()> { @@ -57,20 +98,55 @@ pub async fn update_container(name: &str) -> Result<(String, String)> { check( call(&PrivRequest::UpdateContainer { name: name.to_owned(), + stream: false, }) .await?, ) } +/// Streaming variant: forward stdout/stderr lines to `on_line` as they +/// arrive. Returns `Ok(())` on success; the callback is responsible for +/// appending lines to build_logs or otherwise capturing the output. +pub async fn update_container_streaming( + name: &str, + on_line: impl FnMut(PrivStream, &str), +) -> Result<()> { + ok(call_streaming( + &PrivRequest::UpdateContainer { + name: name.to_owned(), + stream: true, + }, + on_line, + ) + .await?) +} + pub async fn create_container(name: &str) -> Result<(String, String)> { check( call(&PrivRequest::CreateContainer { name: name.to_owned(), + stream: false, }) .await?, ) } +/// Streaming variant: forward stdout/stderr lines to `on_line` as they +/// arrive. Returns `Ok(())` on success. +pub async fn create_container_streaming( + name: &str, + on_line: impl FnMut(PrivStream, &str), +) -> Result<()> { + ok(call_streaming( + &PrivRequest::CreateContainer { + name: name.to_owned(), + stream: true, + }, + on_line, + ) + .await?) +} + pub async fn destroy_container(name: &str) -> Result<()> { ok(call(&PrivRequest::DestroyContainer { name: name.to_owned(), diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index 26908e91..d6ab8a03 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -22,9 +22,10 @@ use std::path::{Path, PathBuf}; use anyhow::{Context as _, Result, bail}; use hive_sh4re::priv_proto::{ AGENT_PREFIX, BindMount, JournalOutput, MANAGER_NAME, META_DIR, NetworkIsolation, PRIV_SOCK, - PrivRequest, PrivResponse, SIBLING_CONTAINERS, + PrivEvent, PrivRequest, PrivResponse, PrivStream, PrivStreamLine, SIBLING_CONTAINERS, }; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::unix::OwnedWriteHalf; use tokio::net::{UnixListener, UnixStream}; use tokio::process::Command; @@ -100,8 +101,12 @@ async fn handle(stream: UnixStream) { let (reader, mut writer) = stream.into_split(); let mut lines = BufReader::new(reader).lines(); while let Ok(Some(line)) = lines.next_line().await { - let resp = dispatch(&line).await; - let mut json = serde_json::to_string(&resp).unwrap_or_else(|e| { + let resp = dispatch(&line, &mut writer).await; + // Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical + // to a bare PrivResponse (untagged), so old hive-c0re callers that + // deserialise directly to PrivResponse continue to work. + let event = PrivEvent::Done(resp); + let mut json = serde_json::to_string(&event).unwrap_or_else(|e| { format!("{{\"ok\":false,\"stdout\":\"\",\"stderr\":\"\",\"error\":\"serialise failed: {e}\"}}") }); json.push('\n'); @@ -112,9 +117,9 @@ async fn handle(stream: UnixStream) { } } -async fn dispatch(line: &str) -> PrivResponse { +async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse { match serde_json::from_str::(line) { - Ok(req) => match exec(req).await { + Ok(req) => match exec(req, writer).await { Ok((stdout, stderr)) => PrivResponse { ok: true, stdout, @@ -137,8 +142,26 @@ async fn dispatch(line: &str) -> PrivResponse { } } +/// Write one `PrivEvent::Line` to the client. Best-effort: a write +/// failure is logged but doesn't abort the running subprocess. +async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) { + let event = PrivEvent::Line(PrivStreamLine { + stream, + data: data.to_owned(), + }); + if let Ok(mut json) = serde_json::to_string(&event) { + json.push('\n'); + if let Err(e) = writer.write_all(json.as_bytes()).await { + tracing::warn!(error = %e, "write_line_event: write failed"); + } + } +} + /// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success. -async fn exec(req: PrivRequest) -> Result<(String, String)> { +/// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`) +/// output lines are forwarded to `writer` as `PrivEvent::Line` messages and +/// the returned strings are empty. +async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> { match req { PrivRequest::StartContainer { ref name } => { validate_container_name(name)?; @@ -155,28 +178,36 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { container_run(&["kill", &container_system_name(name)]).await } - PrivRequest::UpdateContainer { ref name } => { + PrivRequest::UpdateContainer { ref name, stream } => { validate_container_name(name)?; let flake_ref = agent_flake_ref(name); - container_run(&[ + let args = [ "update", &container_system_name(name), "--flake", &flake_ref, - ]) - .await + ]; + if stream { + container_run_streaming(&args, writer).await + } else { + container_run(&args).await + } } - PrivRequest::CreateContainer { ref name } => { + PrivRequest::CreateContainer { ref name, stream } => { validate_container_name(name)?; let flake_ref = agent_flake_ref(name); - container_run(&[ + let args = [ "create", &container_system_name(name), "--flake", &flake_ref, - ]) - .await + ]; + if stream { + container_run_streaming(&args, writer).await + } else { + container_run(&args).await + } } PrivRequest::DestroyContainer { ref name } => { @@ -332,6 +363,93 @@ async fn container_run(args: &[&str]) -> Result<(String, String)> { Ok((stdout, stderr)) } +/// Invoke `nixos-container` with the given args and forward output lines +/// to the caller as `PrivEvent::Line` messages in real time, logging each +/// line to journald as it arrives. Returns `(String::new(), String::new())` +/// on success (all output was streamed); the error string includes stderr +/// tail on failure. +async fn container_run_streaming( + args: &[&str], + writer: &mut OwnedWriteHalf, +) -> Result<(String, String)> { + use tokio::io::AsyncBufReadExt as _; + use tokio::process::Command; + + let mut child = Command::new("nixos-container") + .args(args) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .context("invoke nixos-container (streaming)")?; + + let stdout = child.stdout.take().expect("stdout piped"); + let stderr = child.stderr.take().expect("stderr piped"); + + let mut stdout_lines = BufReader::new(stdout).lines(); + let mut stderr_lines = BufReader::new(stderr).lines(); + + // Collect stderr for the error message; stream both to the client. + let mut stderr_buf = String::new(); + + // Drive stdout and stderr concurrently. `tokio::select!` interleaves + // them without bias — both streams drain at roughly the same rate + // as the subprocess produces output. + loop { + tokio::select! { + line = stdout_lines.next_line() => { + match line { + Ok(Some(l)) => { + tracing::info!(target: "nixos-container", "{l}"); + write_line_event(writer, PrivStream::Stdout, &l).await; + } + Ok(None) => break, + Err(e) => { + tracing::warn!(error = %e, "nixos-container stdout read error"); + break; + } + } + } + line = stderr_lines.next_line() => { + match line { + Ok(Some(l)) => { + tracing::warn!(target: "nixos-container", "{l}"); + write_line_event(writer, PrivStream::Stderr, &l).await; + if !stderr_buf.is_empty() { + stderr_buf.push('\n'); + } + stderr_buf.push_str(&l); + } + Ok(None) => {} + Err(e) => { + tracing::warn!(error = %e, "nixos-container stderr read error"); + } + } + } + } + } + + // Drain any remaining stderr after stdout closed. + while let Ok(Some(l)) = stderr_lines.next_line().await { + tracing::warn!(target: "nixos-container", "{l}"); + write_line_event(writer, PrivStream::Stderr, &l).await; + if !stderr_buf.is_empty() { + stderr_buf.push('\n'); + } + stderr_buf.push_str(&l); + } + + let status = child.wait().await.context("wait nixos-container")?; + if !status.success() { + bail!( + "nixos-container {} failed ({}): {}", + args.join(" "), + status, + stderr_buf.lines().last().unwrap_or("").trim() + ); + } + Ok((String::new(), String::new())) +} + /// Read a container's journal as root via `journalctl -M`. Returns /// `(stdout, stderr)`. Unlike `container_run` a non-zero exit is *not* a /// hard error — journalctl's own diagnostic (folded into `stderr` with diff --git a/hive-sh4re/src/priv_proto.rs b/hive-sh4re/src/priv_proto.rs index 7f864352..2e551e40 100644 --- a/hive-sh4re/src/priv_proto.rs +++ b/hive-sh4re/src/priv_proto.rs @@ -87,11 +87,23 @@ pub enum PrivRequest { /// `nixos-container update --flake ` /// The flake ref is derived from `name` by hive-priv. - UpdateContainer { name: String }, + /// When `stream` is `true`, hive-priv sends `PrivEvent::Line` messages + /// as the process runs, then a terminal `PrivEvent::Done`. + UpdateContainer { + name: String, + #[serde(default)] + stream: bool, + }, /// `nixos-container create --flake ` /// The flake ref is derived from `name` by hive-priv. - CreateContainer { name: String }, + /// When `stream` is `true`, hive-priv sends `PrivEvent::Line` messages + /// as the process runs, then a terminal `PrivEvent::Done`. + CreateContainer { + name: String, + #[serde(default)] + stream: bool, + }, /// `nixos-container destroy ` DestroyContainer { name: String }, @@ -199,3 +211,44 @@ pub struct PrivResponse { #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, } + +// --------------------------------------------------------------------------- +// Streaming protocol +// --------------------------------------------------------------------------- + +/// Which output stream a `PrivStreamLine` came from. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PrivStream { + Stdout, + Stderr, +} + +/// A single line forwarded from a streaming privileged operation. +/// Wire shape: `{"stream":"stdout","data":"..."}` — distinct from +/// `PrivResponse` (which has `ok` but not `stream`/`data`) so that +/// `PrivEvent` can disambiguate with `#[serde(untagged)]`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PrivStreamLine { + pub stream: PrivStream, + pub data: String, +} + +/// An event in the streaming protocol used by long-running priv ops. +/// +/// Wire format (multiple JSON lines over one connection): +/// - Zero or more `Line` events as the subprocess runs. +/// - One terminal `Done` event carrying the final status. +/// +/// Non-streaming ops (and old hive-priv) send exactly one `Done` line, +/// which is wire-identical to a bare `PrivResponse` — so old `call()` +/// callers that deserialise straight to `PrivResponse` continue to work +/// with new hive-priv's terminal event. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(untagged)] +pub enum PrivEvent { + /// A line of output from the running subprocess. + Line(PrivStreamLine), + /// Terminal event: the operation has finished. + Done(PrivResponse), +}