diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs index 622e64ac..1cfdfa83 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -1206,37 +1206,14 @@ async fn set_nspawn_flags( crate::priv_client::write_nspawn_flags(container, &binds, isolation).await } -/// Build the per-line callback for `create_container_streaming` / -/// `update_container_streaming`. Both ops share identical dispatch logic -/// (stdout → info + append_stdout, stderr → warn + append_stderr); this -/// helper avoids duplicating that match body across the two call sites. -fn make_log_callback( - logs: Option>, - log_id: Option, - cmdline: String, -) -> impl FnMut(hive_sh4re::priv_proto::PrivStream, &str) { - use hive_sh4re::priv_proto::PrivStream; - move |stream, line| match stream { - PrivStream::Stdout => { - tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}"); - if let (Some(h), Some(id)) = (&logs, log_id) { - h.append_stdout(id, line); - } - } - PrivStream::Stderr => { - tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}"); - if let (Some(h), Some(id)) = (&logs, log_id) { - h.append_stderr(id, line); - } - } - } -} - /// Execute a container operation via hive-priv and integrate with -/// 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. +/// 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. async fn priv_run(kind: &str, name: &str) -> Result<()> { let container = container_name(name); let cmdline = format!("nixos-container {kind} {container}"); @@ -1250,50 +1227,35 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> { .ok() }); - // 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" => { - crate::priv_client::create_container_streaming( - name, - make_log_callback(logs, log_id, cmdline), - ) - .await - } - "update" => { - crate::priv_client::update_container_streaming( - name, - make_log_callback(logs, log_id, cmdline), - ) - .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, + 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())), other => Err(anyhow::anyhow!("unknown container op: {other}")), }; - let succeeded = result.is_ok(); + let ok = result.is_ok(); if let (Some(h), Some(id)) = (&logs, log_id) { - h.finish( - id, - if succeeded { - crate::build_logs::BuildStatus::Ok - } else { - crate::build_logs::BuildStatus::Fail - }, - ); + 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 }); } 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 e78b48d0..e9b73e45 100644 --- a/hive-c0re/src/priv_client.rs +++ b/hive-c0re/src/priv_client.rs @@ -8,14 +8,12 @@ use anyhow::{Context as _, Result, bail}; use hive_sh4re::priv_proto::{ - BindMount, JournalOutput, NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, - PrivStream, + BindMount, JournalOutput, NetworkIsolation, PRIV_SOCK, PrivRequest, PrivResponse, }; 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 @@ -31,46 +29,7 @@ pub async fn call(req: &PrivRequest) -> Result { .read_line(&mut resp_line) .await .context("read response from hive-priv")?; - // 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), - } - } + serde_json::from_str(&resp_line).context("parse PrivResponse") } pub async fn start_container(name: &str) -> Result<()> { @@ -98,55 +57,20 @@ 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 df5111b8..26908e91 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -22,10 +22,9 @@ 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, - PrivEvent, PrivRequest, PrivResponse, PrivStream, PrivStreamLine, SIBLING_CONTAINERS, + PrivRequest, PrivResponse, SIBLING_CONTAINERS, }; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::unix::OwnedWriteHalf; use tokio::net::{UnixListener, UnixStream}; use tokio::process::Command; @@ -101,12 +100,8 @@ 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, &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| { + let resp = dispatch(&line).await; + let mut json = serde_json::to_string(&resp).unwrap_or_else(|e| { format!("{{\"ok\":false,\"stdout\":\"\",\"stderr\":\"\",\"error\":\"serialise failed: {e}\"}}") }); json.push('\n'); @@ -117,9 +112,9 @@ async fn handle(stream: UnixStream) { } } -async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse { +async fn dispatch(line: &str) -> PrivResponse { match serde_json::from_str::(line) { - Ok(req) => match exec(req, writer).await { + Ok(req) => match exec(req).await { Ok((stdout, stderr)) => PrivResponse { ok: true, stdout, @@ -142,26 +137,8 @@ async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> 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. -/// 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)> { +async fn exec(req: PrivRequest) -> Result<(String, String)> { match req { PrivRequest::StartContainer { ref name } => { validate_container_name(name)?; @@ -178,36 +155,28 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, container_run(&["kill", &container_system_name(name)]).await } - PrivRequest::UpdateContainer { ref name, stream } => { + PrivRequest::UpdateContainer { ref name } => { validate_container_name(name)?; let flake_ref = agent_flake_ref(name); - let args = [ + container_run(&[ "update", &container_system_name(name), "--flake", &flake_ref, - ]; - if stream { - container_run_streaming(&args, writer).await - } else { - container_run(&args).await - } + ]) + .await } - PrivRequest::CreateContainer { ref name, stream } => { + PrivRequest::CreateContainer { ref name } => { validate_container_name(name)?; let flake_ref = agent_flake_ref(name); - let args = [ + container_run(&[ "create", &container_system_name(name), "--flake", &flake_ref, - ]; - if stream { - container_run_streaming(&args, writer).await - } else { - container_run(&args).await - } + ]) + .await } PrivRequest::DestroyContainer { ref name } => { @@ -363,97 +332,6 @@ 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() { - // Only the last stderr line is embedded — the full stderr was - // already forwarded line-by-line as PrivEvent::Line messages and - // is captured in build_logs.sqlite by the caller. Keeping the - // error message short avoids bloating the anyhow chain. - 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 2e551e40..7f864352 100644 --- a/hive-sh4re/src/priv_proto.rs +++ b/hive-sh4re/src/priv_proto.rs @@ -87,23 +87,11 @@ pub enum PrivRequest { /// `nixos-container update --flake ` /// The flake ref is derived from `name` by hive-priv. - /// 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, - }, + UpdateContainer { name: String }, /// `nixos-container create --flake ` /// The flake ref is derived from `name` by hive-priv. - /// 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, - }, + CreateContainer { name: String }, /// `nixos-container destroy ` DestroyContainer { name: String }, @@ -211,44 +199,3 @@ 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), -}