From ac345950012583352201ba2d5aad7b71c6576e2c Mon Sep 17 00:00:00 2001 From: Peron Date: Mon, 27 Jul 2026 18:04:11 +0800 Subject: [PATCH 1/2] feat(proto): add RunnerLogChunk to the attach stream Vendored ahead of the BSR publish from the platform repo; the next `make fleet-proto-sync` reconciles this copy with the pushed module. --- .../proto/arcbox/fleet/v1/fleet.proto | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto b/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto index 7a4a97e44..c654432d5 100644 --- a/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto +++ b/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto @@ -137,6 +137,7 @@ message AttachRequest { RunnerAccepted runner_accepted = 3; RunnerRejected runner_rejected = 4; RunnerExited runner_exited = 5; + RunnerLogChunk runner_log = 6; } } @@ -210,6 +211,31 @@ message RunnerExited { string ack_token = 2; } +// A slice of a started runner's combined stdout/stderr, for live viewing while +// the job runs. Deliberately the one lossy message on this stream: unlike +// verdicts it is never tracked for resend and never acked, because the attach +// stream is a control plane whose heartbeat must not queue behind runner +// output. A chunk that cannot be sent immediately is dropped, and the bytes it +// carried are reported in the next chunk's `dropped_bytes`. Nothing persists it +// — a consumer sees only what arrives while it is subscribed, and GitHub +// remains the authoritative record of a job's log. +message RunnerLogChunk { + // Prefixed runner job id (`rjob_...`). + string job_id = 1; + // Per-job counter starting at 0, incremented for every chunk the agent + // emits. Consumers detect loss as a gap, never as reordering: one attach + // stream delivers in order. + uint64 seq = 2; + // Raw output bytes. Not line-aligned and not guaranteed UTF-8 — a chunk + // boundary can split a line or a multi-byte character. + bytes data = 3; + // Best-effort count of bytes shed since the previous chunk on this job — + // the rate-budget and queue-full drops the agent observed directly. Zero in + // the steady state. A `seq` gap is the authoritative loss signal: a chunk + // discarded further down the agent's egress path is not counted here. + uint64 dropped_bytes = 4; +} + message AttachResponse { oneof msg { // Handshake success — client proceeds to the heartbeat loop. From 954165196e6e1235487615aa2a4b06023d229073 Mon Sep 17 00:00:00 2001 From: Peron Date: Mon, 27 Jul 2026 18:24:21 +0800 Subject: [PATCH 2/2] feat(agent): stream per-job runner output to the gateway Every backend discarded run.sh output: the VM path dropped ssh Data frames, Docker never attached to the container, interop drained stdout into a debug log, and the host path inherited stdio into the service manager's tee. Capture it per job and ship it as RunnerLogChunk. Output rides its own channel and its own contract. The attach stream is a control plane whose heartbeat writes to a 64-slot channel with a blocking send, and the gateway flips a machine Offline after 60s without one, so the log path never awaits and never touches the pending slot that carries a verdict across a reconnect. A chunk that does not fit is dropped, counted, and reported as a seq gap. Host-runner output no longer reaches the service manager's stdout/stderr capture, where it used to land interleaved and unattributed. --- fleet/arcbox-fleet-agent/src/attach.rs | 271 ++++++++++++++++-- fleet/arcbox-fleet-agent/src/control/mod.rs | 10 +- fleet/arcbox-fleet-agent/src/docker.rs | 36 ++- fleet/arcbox-fleet-agent/src/interop.rs | 36 ++- fleet/arcbox-fleet-agent/src/joblog.rs | 297 ++++++++++++++++++++ fleet/arcbox-fleet-agent/src/main.rs | 8 +- fleet/arcbox-fleet-agent/src/runner.rs | 168 +++++++++-- fleet/arcbox-fleet-agent/src/vm.rs | 18 +- 8 files changed, 783 insertions(+), 61 deletions(-) create mode 100644 fleet/arcbox-fleet-agent/src/joblog.rs diff --git a/fleet/arcbox-fleet-agent/src/attach.rs b/fleet/arcbox-fleet-agent/src/attach.rs index 0bf65f217..07b64e325 100644 --- a/fleet/arcbox-fleet-agent/src/attach.rs +++ b/fleet/arcbox-fleet-agent/src/attach.rs @@ -39,6 +39,11 @@ const VERDICT_RESEND_INTERVAL: Duration = Duration::from_secs(10); const INITIAL_BACKOFF: Duration = Duration::from_secs(1); const MAX_BACKOFF: Duration = Duration::from_secs(60); const OUTBOUND_CAPACITY: usize = 64; +/// Depth of the runner-output queue. Deep enough to ride out a brief stall in +/// the gRPC stream, shallow enough that a job outrunning the network is shed +/// promptly instead of accumulating in agent memory. Separate from +/// `OUTBOUND_CAPACITY` so output can never occupy a slot a heartbeat needs. +const LOG_CAPACITY: usize = 256; const MACHINE_TOKEN_HEADER: &str = "x-arcbox-machine-token"; /// How long to wait for runners to be torn down and reaped on shutdown before /// giving up. Killing a process group or container is near-instant, so this is a @@ -124,7 +129,22 @@ fn telemetry_to_control(t: &HostTelemetry) -> control_proto::HostTelemetry { } } -/// Build the [`RunnerSupervisor`] and its egress queue. The verdict-resend +/// The [`RunnerSupervisor`] and the two outbound queues it feeds. +/// +/// A struct rather than a tuple because the two receivers have the same type +/// but opposite delivery contracts — swapping them would make runner output +/// durable and verdicts lossy, and compile fine. +pub struct SupervisorHandles { + pub supervisor: RunnerSupervisor, + /// Runner lifecycle events: delivered with an awaited send and held in the + /// `pending` slot across a reconnect, so a verdict is never lost. + pub egress_rx: mpsc::Receiver, + /// Runner output: delivered with `try_send` and dropped when the stream + /// cannot take it. See [`crate::joblog`]. + pub log_rx: mpsc::Receiver, +} + +/// Build the [`RunnerSupervisor`] and its outbound queues. The verdict-resend /// loop is started by [`run`] instead, so it shares the attachment's shutdown /// token and is reaped with it. /// @@ -136,32 +156,43 @@ pub fn spawn_supervisor( config: &AgentConfig, backends: Arc, state: AgentState, -) -> (RunnerSupervisor, mpsc::Receiver) { +) -> SupervisorHandles { let (egress_tx, egress_rx) = mpsc::channel::(OUTBOUND_CAPACITY); - let supervisor = - RunnerSupervisor::new(egress_tx, config.runner_script.clone(), backends, state); - (supervisor, egress_rx) + let (log_tx, log_rx) = mpsc::channel::(LOG_CAPACITY); + let supervisor = RunnerSupervisor::new( + egress_tx, + log_tx, + config.runner_script.clone(), + backends, + state, + ); + SupervisorHandles { + supervisor, + egress_rx, + log_rx, + } } /// Connect and serve the attach stream, reconnecting on any failure until /// `shutdown` fires, then stop runners cleanly. /// -/// `supervisor` and `egress_rx` come from [`spawn_supervisor`] and are reused -/// across reconnects, so in-flight jobs survive a dropped connection and -/// their verdicts reach the next live stream. On shutdown the loop exits and -/// hands off to [`RunnerSupervisor::shutdown`], which tears down any -/// in-flight runners. +/// `supervisor`, `egress_rx` and `log_rx` come from [`spawn_supervisor`] and +/// are reused across reconnects, so in-flight jobs survive a dropped +/// connection and their verdicts reach the next live stream. On shutdown the +/// loop exits and hands off to [`RunnerSupervisor::shutdown`], which tears +/// down any in-flight runners. #[allow( clippy::too_many_arguments, reason = "the reconnect loop genuinely needs all of: endpoint config, credential, the \ - persistent supervisor, the cross-reconnect egress queue, the backend \ - registry, the shutdown token, and the observable state handle" + persistent supervisor, the cross-reconnect egress and runner-output queues, \ + the backend registry, the shutdown token, and the observable state handle" )] pub async fn run( config: AgentConfig, credential: Credential, supervisor: RunnerSupervisor, mut egress_rx: mpsc::Receiver, + mut log_rx: mpsc::Receiver, backends: Arc, shutdown: CancellationToken, state: AgentState, @@ -193,6 +224,7 @@ pub async fn run( &credential, &supervisor, &mut egress_rx, + &mut log_rx, &mut pending, &mut backoff, &backends, @@ -317,21 +349,22 @@ pub async fn run( /// /// Outbound traffic is multiplexed onto a fresh per-connection request channel: /// connection-scoped heartbeats are sent directly, while runner lifecycle -/// events are forwarded from the shared egress queue. Inbound orders are routed -/// to the persistent `supervisor`. +/// events are forwarded from the shared egress queue and runner output from the +/// shared log queue. Inbound orders are routed to the persistent `supervisor`. #[allow( clippy::too_many_arguments, reason = "one connection's lifecycle genuinely needs all of: endpoint config, \ - credential, the persistent supervisor, the cross-reconnect egress queue \ - and its pending slot, the mutable backoff, advertised capabilities, the \ - shutdown token, the observable state handle, and the cross-reconnect \ - update-drain flag" + credential, the persistent supervisor, the cross-reconnect egress and \ + runner-output queues and the egress pending slot, the mutable backoff, \ + advertised capabilities, the shutdown token, the observable state handle, \ + and the cross-reconnect update-drain flag" )] async fn connect_and_serve( config: &AgentConfig, credential: &Credential, supervisor: &RunnerSupervisor, egress_rx: &mut mpsc::Receiver, + log_rx: &mut mpsc::Receiver, pending: &mut Option, backoff: &mut Duration, backends: &Backends, @@ -496,6 +529,24 @@ async fn connect_and_serve( // happen while the agent runs; treat it as a clean shutdown. None => break Ok(StreamEnd::Closed), }, + chunk = log_rx.recv() => match chunk { + // Runner output, forwarded on the opposite contract to the arm + // above: `try_send`, never an awaited send, and never parked in + // `pending`. Awaiting here would let a chatty job hold the + // request channel long enough to delay the heartbeat, which the + // gateway reads as a dead machine; `pending` is the single slot + // that makes a verdict survive a reconnect and must not be spent + // on a log chunk. A chunk that does not fit is dropped, and the + // consumer sees the gap in `seq`. + Some(msg) => { + if req_tx.try_send(msg).is_err() { + tracing::trace!("runner log chunk dropped; request channel full"); + } + } + // Same reasoning as the egress arm: the supervisor holds the + // only log sender. + None => break Ok(StreamEnd::Closed), + }, message = inbound.message() => match message { Ok(Some(message)) => { tracing::debug!(msg = ?message.msg, "inbound attach message"); @@ -722,14 +773,14 @@ mod tests { machine_token: "flt_revoked".to_owned(), }; let backends = Backends::fixed(Vec::new(), state.clone()); - let (supervisor, egress_rx) = - spawn_supervisor(&config, Arc::clone(&backends), state.clone()); + let handles = spawn_supervisor(&config, Arc::clone(&backends), state.clone()); let shutdown = CancellationToken::new(); let run = tokio::spawn(run( config, credential, - supervisor, - egress_rx, + handles.supervisor, + handles.egress_rx, + handles.log_rx, backends, shutdown.clone(), state.clone(), @@ -851,6 +902,7 @@ mod tests { let backends = Backends::fixed(Vec::new(), AgentState::new(&seed())); let mut backends_rx = backends.subscribe(); let (_egress_tx, mut egress_rx) = mpsc::channel::(1); + let (_log_tx, mut log_rx) = mpsc::channel::(1); let mut pending = None; let mut backoff = INITIAL_BACKOFF; let shutdown = CancellationToken::new(); @@ -872,6 +924,7 @@ mod tests { &credential, &supervisor, &mut egress_rx, + &mut log_rx, &mut pending, &mut backoff, &backends, @@ -985,14 +1038,14 @@ mod tests { ..config() }; let backends = Backends::new(false, None, None, None, state.clone()); - let (supervisor, egress_rx) = - spawn_supervisor(&config, Arc::clone(&backends), state.clone()); + let handles = spawn_supervisor(&config, Arc::clone(&backends), state.clone()); let shutdown = CancellationToken::new(); let run_task = tokio::spawn(run( config, credential(), - supervisor, - egress_rx, + handles.supervisor, + handles.egress_rx, + handles.log_rx, Arc::clone(&backends), shutdown.clone(), state, @@ -1089,7 +1142,7 @@ mod tests { fn supervisor() -> RunnerSupervisor { let (events, _rx) = mpsc::channel(1); let state = AgentState::new(&seed()); - RunnerSupervisor::new( + RunnerSupervisor::without_logs( events, None, Backends::fixed(Vec::new(), state.clone()), @@ -1097,6 +1150,168 @@ mod tests { ) } + /// A gateway that accepts every `Attach` and records every subsequent + /// inbound message, holding the stream open until the client leaves. + struct MessageRecordingGateway { + seen: Arc>>, + } + + #[tonic::async_trait] + impl arcbox_fleet_proto::v1::fleet_gateway_service_server::FleetGatewayService + for MessageRecordingGateway + { + async fn enroll( + &self, + _: Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::unimplemented("not used by this test")) + } + + type AttachStream = tokio_stream::wrappers::ReceiverStream< + Result, + >; + + async fn attach( + &self, + request: Request>, + ) -> Result, tonic::Status> { + let mut inbound = request.into_inner(); + let first = inbound + .message() + .await + .map_err(|e| tonic::Status::internal(e.to_string()))? + .ok_or_else(|| tonic::Status::internal("closed before Attach"))?; + if !matches!(first.msg, Some(attach_request::Msg::Attach(_))) { + return Err(tonic::Status::internal("first message was not Attach")); + } + + let (tx, rx) = mpsc::channel(4); + let _ = tx + .send(Ok(arcbox_fleet_proto::v1::AttachResponse { + msg: Some(attach_response::Msg::AttachAccepted( + arcbox_fleet_proto::v1::AttachAccepted {}, + )), + })) + .await; + let seen = Arc::clone(&self.seen); + tokio::spawn(async move { + let _hold = tx; + while let Ok(Some(message)) = inbound.message().await { + if let Some(msg) = message.msg { + seen.lock().await.push(msg); + } + } + }); + Ok(tonic::Response::new( + tokio_stream::wrappers::ReceiverStream::new(rx), + )) + } + + async fn unenroll( + &self, + _: Request, + ) -> Result, tonic::Status> + { + Err(tonic::Status::unimplemented("not used by this test")) + } + } + + /// Runner output must reach the gateway on the same stream as verdicts, + /// while staying out of the machinery that makes verdicts durable. + /// + /// The `pending` assertion is the one that matters: it is the single slot + /// that carries an undelivered verdict across a reconnect, and the log arm + /// sits directly beside the egress arm that fills it. A log chunk parked + /// there would evict a verdict — the exact bug a copy-paste of the + /// neighbouring arm would introduce, and one nothing else would catch. + #[tokio::test] + async fn log_chunks_reach_the_stream_without_consuming_the_verdict_pending_slot() { + use arcbox_fleet_proto::v1::fleet_gateway_service_server::FleetGatewayServiceServer; + + let seen = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let gateway = format!("http://{}", listener.local_addr().unwrap()); + tokio::spawn( + tonic::transport::Server::builder() + .add_service(FleetGatewayServiceServer::new(MessageRecordingGateway { + seen: Arc::clone(&seen), + })) + .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)), + ); + + let state = AgentState::new(&PersistedSettings { + gateway: gateway.clone(), + ..seed() + }); + let supervisor = supervisor(); + let backends = Backends::fixed(Vec::new(), AgentState::new(&seed())); + let mut backends_rx = backends.subscribe(); + let (_egress_tx, mut egress_rx) = mpsc::channel::(1); + + // Queue output before the stream exists, the way a job that started on + // a previous connection would. + let (log_tx, mut log_rx) = mpsc::channel::(LOG_CAPACITY); + let sink = crate::joblog::JobLogSink::new("rjob_a", log_tx); + sink.write(b"hello from the runner\n"); + + let mut pending = None; + let mut backoff = INITIAL_BACKOFF; + let shutdown = CancellationToken::new(); + let config = AgentConfig { + gateway, + ..config() + }; + let credential = credential(); + let mut drained_for_update = false; + + // Leave the stream once the chunk has had time to land; the loop only + // ends on shutdown, so drive it with one. + let stopper = shutdown.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(500)).await; + stopper.cancel(); + }); + + tokio::time::timeout( + Duration::from_secs(5), + connect_and_serve( + &config, + &credential, + &supervisor, + &mut egress_rx, + &mut log_rx, + &mut pending, + &mut backoff, + &backends, + &mut backends_rx, + &shutdown, + &state, + &mut drained_for_update, + ), + ) + .await + .expect("the serve loop must exit on shutdown, not hang") + .expect("clean exit"); + + let seen = seen.lock().await; + let logs: Vec<_> = seen + .iter() + .filter_map(|msg| match msg { + attach_request::Msg::RunnerLog(chunk) => Some(chunk), + _ => None, + }) + .collect(); + assert_eq!(logs.len(), 1, "the queued chunk must reach the gateway"); + assert_eq!(logs[0].job_id, "rjob_a"); + assert_eq!(logs[0].data, b"hello from the runner\n"); + + assert!( + pending.is_none(), + "a log chunk must never occupy the slot that makes a verdict survive a reconnect" + ); + } + /// The connect + Attach-RPC handshake has no cancellation awareness of /// its own (see this fn's own doc in the non-test code above) — without /// racing it against `shutdown`, a reconnect attempt could complete the @@ -1110,6 +1325,7 @@ mod tests { let backends = Backends::fixed(Vec::new(), AgentState::new(&seed())); let mut backends_rx = backends.subscribe(); let (_egress_tx, mut egress_rx) = mpsc::channel::(1); + let (_log_tx, mut log_rx) = mpsc::channel::(1); let mut pending = None; let mut backoff = INITIAL_BACKOFF; let shutdown = CancellationToken::new(); @@ -1126,6 +1342,7 @@ mod tests { &credential, &supervisor, &mut egress_rx, + &mut log_rx, &mut pending, &mut backoff, &backends, diff --git a/fleet/arcbox-fleet-agent/src/control/mod.rs b/fleet/arcbox-fleet-agent/src/control/mod.rs index 7c5df4ab3..92bdba8d2 100644 --- a/fleet/arcbox-fleet-agent/src/control/mod.rs +++ b/fleet/arcbox-fleet-agent/src/control/mod.rs @@ -320,14 +320,16 @@ impl AgentSupervisor { // process-lifetime state, so clear any stale draining flag rather than // letting a re-enroll inherit it. self.agent_state.set_draining(false); - let (supervisor, egress_rx) = + let handles = attach::spawn_supervisor(&self.config, self.backends(), self.agent_state.clone()); + let supervisor = handles.supervisor; let shutdown = self.process_shutdown.child_token(); let task = tokio::spawn(attach::run( self.config.clone(), credential.clone(), supervisor.clone(), - egress_rx, + handles.egress_rx, + handles.log_rx, self.backends(), shutdown.clone(), self.agent_state.clone(), @@ -960,7 +962,7 @@ mod tests { Ok(()) }); let (events, _events_rx) = tokio::sync::mpsc::channel(1); - let runner = crate::runner::RunnerSupervisor::new( + let runner = crate::runner::RunnerSupervisor::without_logs( events, None, Backends::new(false, None, None, None, agent_state.clone()), @@ -1106,7 +1108,7 @@ mod tests { Ok(()) }); let (events, _events_rx) = tokio::sync::mpsc::channel(1); - let runner = crate::runner::RunnerSupervisor::new( + let runner = crate::runner::RunnerSupervisor::without_logs( events, None, Backends::new(false, None, None, None, agent_state.clone()), diff --git a/fleet/arcbox-fleet-agent/src/docker.rs b/fleet/arcbox-fleet-agent/src/docker.rs index 86c833a06..582de9e59 100644 --- a/fleet/arcbox-fleet-agent/src/docker.rs +++ b/fleet/arcbox-fleet-agent/src/docker.rs @@ -8,12 +8,14 @@ use anyhow::{Context, Result}; use bollard::Docker; use bollard::models::ContainerCreateBody; use bollard::query_parameters::{ - CreateContainerOptions, CreateImageOptions, RemoveContainerOptions, WaitContainerOptions, + CreateContainerOptions, CreateImageOptions, LogsOptions, RemoveContainerOptions, + WaitContainerOptions, }; use tokio_stream::StreamExt; use tracing::{debug, info, warn}; use crate::host; +use crate::joblog::JobLogSink; /// Everything the Docker runner needs to execute one job. pub struct RunSpec<'a> { @@ -40,6 +42,38 @@ pub struct RunningContainer { } impl RunningContainer { + /// Stream this container's combined output into `sink` for the rest of its + /// life, in a detached task. + /// + /// Detaching is safe because the stream is bounded by the container: it + /// ends when the container exits or is removed, which every exit path from + /// [`wait`](Self::wait) leads to. Docker holds the output itself, so unlike + /// the pipe-backed backends nothing blocks if this task falls behind. + pub fn follow_logs(&self, sink: JobLogSink) { + let client = self.client.clone(); + let id = self.id.clone(); + tokio::spawn(async move { + let options = LogsOptions { + follow: true, + stdout: true, + stderr: true, + ..Default::default() + }; + let mut stream = client.logs(&id, Some(options)); + while let Some(frame) = stream.next().await { + match frame { + Ok(output) => sink.write(&output.into_bytes()), + // Teardown closes the stream mid-read; the job's own exit + // path reports the outcome, so this is not an error here. + Err(e) => { + debug!(container = %id, error = %e, "container log stream ended"); + break; + } + } + } + }); + } + /// Block until the container exits and return its exit code. No cleanup — /// follow with [`remove`](Self::remove). The agent reports no outcome /// upstream (the GitHub webhook is authoritative); the code is for logging. diff --git a/fleet/arcbox-fleet-agent/src/interop.rs b/fleet/arcbox-fleet-agent/src/interop.rs index faa9ec2c7..7591659cc 100644 --- a/fleet/arcbox-fleet-agent/src/interop.rs +++ b/fleet/arcbox-fleet-agent/src/interop.rs @@ -30,6 +30,8 @@ use anyhow::{Context, Result, bail, ensure}; use tokio::io::{AsyncBufReadExt, BufReader}; use tracing::{debug, warn}; +use crate::joblog::JobLogSink; + /// Fixed Windows locations of the interop tools, translated through /// `wslpath` at startup so a non-default automount root still resolves. /// Windows PowerShell 5.1 ships with every Windows 10/11 install (unlike @@ -162,7 +164,10 @@ impl InteropRunner { /// side and carries the PID that can cancel it. Any failure past the /// spawn reaps the relay before returning, so an error never leaks a /// process. - pub async fn spawn(&self, encoded_jit_config: &str) -> Result { + /// + /// Everything the wrapper prints after the handshake is the runner's own + /// output and goes to `sink`. + pub async fn spawn(&self, encoded_jit_config: &str, sink: JobLogSink) -> Result { validate_jit_config(&self.script, encoded_jit_config)?; let mut child = tokio::process::Command::new(&self.powershell) @@ -170,10 +175,18 @@ impl InteropRunner { .arg(wrapper_command(&self.script, encoded_jit_config)) .stdin(Stdio::null()) .stdout(Stdio::piped()) + .stderr(Stdio::piped()) .kill_on_drop(true) .spawn() .context("spawning the interop wrapper")?; + // Piped, so it must be drained for the same reason stdout is. The + // handshake only reads stdout, so this starts immediately rather than + // after it. + if let Some(stderr) = child.stderr.take() { + sink.pipe(stderr); + } + let stdout = child .stdout .take() @@ -206,11 +219,21 @@ impl InteropRunner { // Keep the pipe drained for the rest of the job — the runner's // output flows through the wrapper's console — so a chatty runner // can never fill the pipe and wedge itself. Ends at EOF when the - // relay exits; detaching the handle is deliberate. + // relay exits; detaching the handle is deliberate. The reader stays + // line-oriented rather than switching to `sink.pipe`: the handshake + // already wrapped stdout in a `BufReader`, which may hold buffered + // bytes that a raw read of the underlying pipe would skip. tokio::spawn(async move { loop { match lines.next_line().await { - Ok(Some(line)) => debug!(line, "windows runner output"), + // Restore the terminator `next_line` stripped, and ship the + // line whole: two writes would split every line across two + // chunks, doubling the message count for no benefit. + Ok(Some(line)) => { + let mut framed = line.into_bytes(); + framed.push(b'\n'); + sink.write(&framed); + } Ok(None) => break, Err(e) => { debug!(error = %e, "windows runner output stream failed"); @@ -435,7 +458,7 @@ mod tests { /// spawns long-existing Windows binaries, never freshly written ones. async fn spawn_retrying(runner: &InteropRunner, jit: &str) -> Result { for _ in 0..100 { - match runner.spawn(jit).await { + match runner.spawn(jit, JobLogSink::discarding()).await { Err(e) if format!("{e:#}").contains("Text file busy") => { tokio::time::sleep(Duration::from_millis(20)).await; } @@ -589,7 +612,10 @@ mod tests { std::fs::write(&staged, "@echo off\r\nping -n 60 127.0.0.1 >NUL\r\n").unwrap(); let runner = InteropRunner::new(&script).await.expect("probe"); - let mut job = runner.spawn("dGVzdA==").await.expect("spawn"); + let mut job = runner + .spawn("dGVzdA==", JobLogSink::discarding()) + .await + .expect("spawn"); assert!(job.windows_pid() > 0); let start = tokio::time::Instant::now(); diff --git a/fleet/arcbox-fleet-agent/src/joblog.rs b/fleet/arcbox-fleet-agent/src/joblog.rs new file mode 100644 index 000000000..e866bdf3a --- /dev/null +++ b/fleet/arcbox-fleet-agent/src/joblog.rs @@ -0,0 +1,297 @@ +//! Per-job capture of runner output, shipped to the gateway as +//! [`RunnerLogChunk`]s for live viewing while the job runs. +//! +//! This is the one lossy path in the agent, and deliberately so. The attach +//! stream is a control plane: its heartbeat writes to a 64-slot egress channel +//! with a blocking send, and the gateway flips a machine `Offline` after 60s +//! without one — which stops placement offering it work. Runner output can +//! arrive orders of magnitude faster than that channel drains, so a sink that +//! queued or awaited would trade a machine's liveness for its logs. Instead +//! every write is admitted through a per-job rate budget and a non-blocking +//! `try_send`, and anything that does not fit is dropped and counted. +//! +//! Backpressure therefore never propagates back to the runner: a chatty job +//! loses log lines, not throughput, and the consumer sees the loss as a `seq` +//! gap plus a `dropped_bytes` count on the next chunk. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use arcbox_fleet_proto::v1::{AttachRequest, RunnerLogChunk, attach_request}; +use tokio::io::{AsyncRead, AsyncReadExt}; +use tokio::sync::mpsc; + +/// Largest payload a single chunk carries. Sized so one chunk is a comfortable +/// gRPC message and a burst of them still fits the egress channel. +const MAX_CHUNK_BYTES: usize = 32 * 1024; + +/// Sustained per-job output rate. Well above what a normal CI job produces +/// (a verbose build is a few KiB/s), low enough that a runaway job printing in +/// a tight loop cannot saturate the machine's control stream. +const BUDGET_BYTES_PER_SEC: u64 = 256 * 1024; + +/// Burst allowance on top of the sustained rate, so the bulk output of a +/// single build step is not shed just for arriving all at once. +const BUDGET_BURST_BYTES: u64 = 1024 * 1024; + +/// Read buffer for [`JobLogSink::pipe`]. Matches the chunk size so a full read +/// maps to exactly one chunk. +const PIPE_BUFFER_BYTES: usize = MAX_CHUNK_BYTES; + +/// Captures one job's output and ships it as [`RunnerLogChunk`]s. +/// +/// Cheap to clone — a job's stdout and stderr readers share one sink, so their +/// interleaving on the wire matches the order the agent observed, and `seq` +/// stays monotonic across both. +#[derive(Clone)] +pub struct JobLogSink { + inner: Arc, +} + +struct Inner { + /// Prefixed runner job id (`rjob_...`) stamped on every chunk. + job_id: String, + /// Log-only egress channel, separate from the supervisor's event channel so + /// output can never occupy a slot a verdict or heartbeat needs. + tx: mpsc::Sender, + /// Next chunk's sequence number. Consumed even by a chunk that fails to + /// send, so a drop shows up downstream as a gap. + seq: AtomicU64, + /// Bytes shed since the last chunk that went out, reported on the next one. + dropped: AtomicU64, + budget: Mutex, +} + +impl JobLogSink { + /// Open a sink for `job_id` feeding `tx`. + pub fn new(job_id: &str, tx: mpsc::Sender) -> Self { + Self { + inner: Arc::new(Inner { + job_id: job_id.to_owned(), + tx, + seq: AtomicU64::new(0), + dropped: AtomicU64::new(0), + budget: Mutex::new(TokenBucket::new()), + }), + } + } + + /// A sink with no receiver, so every write is shed. For tests that drive a + /// runner path without asserting on its output. + #[cfg(test)] + pub fn discarding() -> Self { + let (tx, _) = mpsc::channel(1); + Self::new("rjob_test", tx) + } + + /// Ship `data` as one or more chunks. Never blocks and never fails: bytes + /// that exceed the rate budget, or that meet a full channel, are counted + /// into the next chunk's `dropped_bytes` instead. + pub fn write(&self, data: &[u8]) { + for slice in data.chunks(MAX_CHUNK_BYTES) { + self.write_chunk(slice); + } + } + + fn write_chunk(&self, slice: &[u8]) { + let len = slice.len() as u64; + if !self + .inner + .budget + .lock() + .expect("budget mutex poisoned") + .take(len) + { + self.inner.dropped.fetch_add(len, Ordering::Relaxed); + return; + } + // Read rather than swap: the count must survive a failed send, and + // only the chunk that actually goes out may clear it. + let reported = self.inner.dropped.load(Ordering::Relaxed); + let chunk = RunnerLogChunk { + job_id: self.inner.job_id.clone(), + seq: self.inner.seq.fetch_add(1, Ordering::Relaxed), + data: slice.to_vec(), + dropped_bytes: reported, + }; + let message = AttachRequest { + msg: Some(attach_request::Msg::RunnerLog(chunk)), + }; + match self.inner.tx.try_send(message) { + // Subtract rather than store 0: a concurrent writer may have added + // to the count between the load above and here. + Ok(()) => { + self.inner.dropped.fetch_sub(reported, Ordering::Relaxed); + } + Err(_) => { + self.inner.dropped.fetch_add(len, Ordering::Relaxed); + } + } + } + + /// Drain `reader` into this sink until EOF, in a detached task. + /// + /// Detaching is deliberate and matches the lifetime of what is being read: + /// the reader is a pipe from the runner process, so it reaches EOF when + /// that process exits or is torn down, and the task ends with it. A read + /// error ends the task too — a broken pipe is teardown, not a fault. + pub fn pipe(&self, mut reader: R) + where + R: AsyncRead + Unpin + Send + 'static, + { + let sink = self.clone(); + tokio::spawn(async move { + let mut buffer = vec![0u8; PIPE_BUFFER_BYTES]; + loop { + match reader.read(&mut buffer).await { + Ok(0) | Err(_) => break, + Ok(read) => sink.write(&buffer[..read]), + } + } + }); + } +} + +/// Leaky-bucket rate limiter over bytes, refilled continuously from elapsed +/// wall time. Holding the lock never spans an await — every caller is the +/// synchronous [`JobLogSink::write_chunk`]. +struct TokenBucket { + tokens: u64, + last: Instant, +} + +impl TokenBucket { + fn new() -> Self { + Self { + // Start full so a short job's entire output is admitted. + tokens: BUDGET_BURST_BYTES, + last: Instant::now(), + } + } + + /// Admit `want` bytes, or refuse them outright. Never partially admits: + /// a chunk is shed whole so the reported byte count stays exact. + fn take(&mut self, want: u64) -> bool { + self.refill(Instant::now()); + if self.tokens >= want { + self.tokens -= want; + true + } else { + false + } + } + + fn refill(&mut self, now: Instant) { + let elapsed = now.saturating_duration_since(self.last).as_secs_f64(); + self.last = now; + let refilled = (elapsed * BUDGET_BYTES_PER_SEC as f64) as u64; + self.tokens = self.tokens.saturating_add(refilled).min(BUDGET_BURST_BYTES); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Pull every chunk currently queued, as `(seq, data, dropped_bytes)`. + fn drain(rx: &mut mpsc::Receiver) -> Vec<(u64, Vec, u64)> { + let mut chunks = Vec::new(); + while let Ok(message) = rx.try_recv() { + match message.msg { + Some(attach_request::Msg::RunnerLog(chunk)) => { + chunks.push((chunk.seq, chunk.data, chunk.dropped_bytes)); + } + other => panic!("expected RunnerLog, got {other:?}"), + } + } + chunks + } + + #[tokio::test] + async fn splits_oversized_writes_and_numbers_them_in_order() { + let (tx, mut rx) = mpsc::channel(16); + let sink = JobLogSink::new("rjob_a", tx); + + sink.write(&vec![b'x'; MAX_CHUNK_BYTES + 1]); + + let chunks = drain(&mut rx); + assert_eq!(chunks.len(), 2); + assert_eq!(chunks[0].0, 0); + assert_eq!(chunks[0].1.len(), MAX_CHUNK_BYTES); + assert_eq!(chunks[1].0, 1); + assert_eq!(chunks[1].1.len(), 1); + assert!(chunks.iter().all(|(_, _, dropped)| *dropped == 0)); + } + + /// A full channel must not block or error — the bytes are shed and + /// reported on the next chunk that gets through, and the burnt sequence + /// number leaves the gap that tells the consumer output was lost. + #[tokio::test] + async fn full_channel_sheds_and_reports_on_the_next_chunk() { + let (tx, mut rx) = mpsc::channel(1); + let sink = JobLogSink::new("rjob_a", tx); + + sink.write(b"first"); + sink.write(b"lost"); + + // Free the single slot, then send again: the drop is now reportable. + let queued = drain(&mut rx); + assert_eq!(queued.len(), 1); + assert_eq!(queued[0], (0, b"first".to_vec(), 0)); + + sink.write(b"third"); + let chunks = drain(&mut rx); + assert_eq!(chunks.len(), 1); + let (seq, data, dropped) = &chunks[0]; + assert_eq!(*seq, 2, "the shed chunk burnt seq 1, leaving a gap"); + assert_eq!(data, b"third"); + assert_eq!(*dropped, 4, "the shed \"lost\" bytes are reported here"); + } + + /// Sustained output past the burst allowance is shed rather than queued, + /// so a runaway job cannot grow the agent's memory or stall the stream. + #[tokio::test] + async fn output_beyond_the_burst_allowance_is_shed() { + let (tx, mut rx) = mpsc::channel(1024); + let sink = JobLogSink::new("rjob_a", tx); + + let over_budget = (BUDGET_BURST_BYTES as usize) + (4 * MAX_CHUNK_BYTES); + sink.write(&vec![b'x'; over_budget]); + + let chunks = drain(&mut rx); + let admitted: usize = chunks.iter().map(|(_, data, _)| data.len()).sum(); + assert!( + admitted <= BUDGET_BURST_BYTES as usize + MAX_CHUNK_BYTES, + "admitted {admitted} bytes, above the burst allowance" + ); + assert!(admitted > 0, "the burst allowance must admit something"); + + // The shed bytes are pending, reported once the budget refills. + assert_eq!( + sink.inner.dropped.load(Ordering::Relaxed) as usize, + over_budget - admitted + ); + } + + #[tokio::test] + async fn pipe_forwards_reader_contents_until_eof() { + let (tx, mut rx) = mpsc::channel(16); + let sink = JobLogSink::new("rjob_a", tx.clone()); + + sink.pipe(std::io::Cursor::new(b"runner output".to_vec())); + // Drop the local sender so the channel closes once the piped task + // finishes, making the drain below deterministic. + drop(tx); + drop(sink); + + let mut seen = Vec::new(); + while let Some(message) = rx.recv().await { + match message.msg { + Some(attach_request::Msg::RunnerLog(chunk)) => seen.extend(chunk.data), + other => panic!("expected RunnerLog, got {other:?}"), + } + } + assert_eq!(seen, b"runner output"); + } +} diff --git a/fleet/arcbox-fleet-agent/src/main.rs b/fleet/arcbox-fleet-agent/src/main.rs index a795fca0a..a06052f5d 100644 --- a/fleet/arcbox-fleet-agent/src/main.rs +++ b/fleet/arcbox-fleet-agent/src/main.rs @@ -36,6 +36,7 @@ mod enroll; mod fsutil; mod host; mod interop; +mod joblog; #[cfg(test)] mod mock_daemon; mod runner; @@ -326,13 +327,14 @@ async fn run(command: Command, config: AgentConfig) -> Result<()> { shutdown.clone(), ); - let (supervisor, egress_rx) = + let handles = attach::spawn_supervisor(&config, Arc::clone(&backends), agent_state.clone()); attach::run( config, credential, - supervisor, - egress_rx, + handles.supervisor, + handles.egress_rx, + handles.log_rx, backends, shutdown, agent_state, diff --git a/fleet/arcbox-fleet-agent/src/runner.rs b/fleet/arcbox-fleet-agent/src/runner.rs index 95224b2d5..7857edd94 100644 --- a/fleet/arcbox-fleet-agent/src/runner.rs +++ b/fleet/arcbox-fleet-agent/src/runner.rs @@ -16,6 +16,7 @@ //! so a busy host simply rejects and the platform re-offers elsewhere. use std::path::{Path, PathBuf}; +use std::process::Stdio; use std::sync::Arc; use std::time::Duration; @@ -33,6 +34,7 @@ use tracing::{info, warn}; use crate::backends::Backends; use crate::docker::RunSpec; use crate::host; +use crate::joblog::JobLogSink; use crate::state::AgentState; /// Watchdog on a VM job's runtime: GitHub concludes jobs at 6 h, so a @@ -152,6 +154,9 @@ pub struct RunnerSupervisor { struct Inner { /// Outbound channel to the gateway (offer verdicts). events: mpsc::Sender, + /// Outbound channel for runner output. Separate from `events` so a chatty + /// job can never delay a verdict or a heartbeat; see [`crate::joblog`]. + logs: mpsc::Sender, /// Verdicts sent but not yet acknowledged by the gateway, keyed by /// `offer_token`. Resent until an `OfferVerdictAck` arrives so a gateway /// crash-after-read cannot lose the verdict. There is no local expiry: the @@ -209,14 +214,15 @@ impl Drop for ReleaseGuard { } impl RunnerSupervisor { - /// Create a supervisor that emits verdicts on `events`. Routing comes - /// live from `backends` — the same registry the advertised capability - /// set derives from, so routing and advertisement agree by construction. - /// `load_ceiling`/`mem_floor_mib` are not parameters: `admit()` reads - /// them live from `state`, which is the single source of truth settings - /// write through. + /// Create a supervisor that emits verdicts on `events` and runner output + /// on `logs`. Routing comes live from `backends` — the same registry the + /// advertised capability set derives from, so routing and advertisement + /// agree by construction. `load_ceiling`/`mem_floor_mib` are not + /// parameters: `admit()` reads them live from `state`, which is the single + /// source of truth settings write through. pub fn new( events: mpsc::Sender, + logs: mpsc::Sender, runner_script: Option, backends: Arc, state: AgentState, @@ -224,6 +230,7 @@ impl RunnerSupervisor { Self { inner: Arc::new(Inner { events, + logs, outstanding: DashMap::new(), in_flight: DashMap::new(), runner_script, @@ -509,6 +516,26 @@ impl RunnerSupervisor { } } + /// Test constructor with no runner-output consumer, so every chunk is + /// shed. For the many tests that exercise admission, verdicts and job + /// lifecycle without asserting on output. + #[cfg(test)] + pub fn without_logs( + events: mpsc::Sender, + runner_script: Option, + backends: Arc, + state: AgentState, + ) -> Self { + Self::new(events, mpsc::channel(1).0, runner_script, backends, state) + } + + /// Open the runner-output sink for a job. Created before the runner starts + /// so no backend has to retrofit capture onto an already-running process; + /// a job that never starts simply writes nothing to it. + fn log_sink(&self, job_id: &str) -> JobLogSink { + JobLogSink::new(job_id, self.inner.logs.clone()) + } + /// Start the runner for an accepted offer, then accept once it is actually /// running. A failure to start rejects instead, so the platform re-offers. async fn run_job( @@ -613,8 +640,9 @@ impl RunnerSupervisor { Canceled, Watchdog, } + let sink = self.log_sink(job_id); let exited = { - let wait = std::pin::pin!(running.wait()); + let wait = std::pin::pin!(running.wait(&sink)); tokio::select! { exit = wait => VmExit::Exited(exit), () = cancel.cancelled() => VmExit::Canceled, @@ -667,7 +695,8 @@ impl RunnerSupervisor { // on degraded interop and must not make the agent deaf to // CancelRunner — mirror the VM startup path. let spawned = { - let spawn = std::pin::pin!(interop.spawn(&order.encoded_jit_config)); + let spawn = + std::pin::pin!(interop.spawn(&order.encoded_jit_config, self.log_sink(job_id))); tokio::select! { result = spawn => Some(result), () = cancel.cancelled() => None, @@ -749,6 +778,18 @@ impl RunnerSupervisor { } }; + // Drain both pipes for the job's lifetime. This is not optional: with + // `Stdio::piped()` a runner that outruns the reader would block on a + // full pipe buffer, so capture and liveness are the same concern here. + let sink = self.log_sink(job_id); + let inner = child.inner(); + if let Some(stdout) = inner.stdout.take() { + sink.pipe(stdout); + } + if let Some(stderr) = inner.stderr.take() { + sink.pipe(stderr); + } + // group_spawn is synchronous, but cancellation is handled on another // runtime thread. Arbitrate through the slot before accepting. if !self.accept_started(job_id, token) { @@ -833,6 +874,12 @@ impl RunnerSupervisor { } }; + // Follow from here rather than before `start`: the log stream is keyed + // on the container, which only exists now. Docker retains output from + // the container's first instant, so nothing produced during startup is + // missed. + running.follow_logs(self.log_sink(job_id)); + // The container is running. Cancellation and acceptance arbitrate // through the slot before any verdict is sent. if !self.accept_started(job_id, token) { @@ -993,9 +1040,20 @@ impl RunnerSupervisor { /// point (`run.sh`); no `.current_dir()` is set because the wrapper script /// locates its own sibling files via `$0`'s dirname, not the caller's /// working directory. +/// +/// Both output streams are piped rather than inherited, so they can be +/// attributed to this job and streamed to the gateway. The caller must drain +/// them (see [`RunnerSupervisor::run_host_job`]) or a chatty runner blocks on a +/// full pipe. This is why host-runner output no longer appears in the service +/// manager's stdout/stderr capture, where it used to land interleaved and +/// unattributed. fn runner_command(script: &Path, encoded_jit_config: &str) -> tokio::process::Command { let mut command = tokio::process::Command::new(script); - command.arg("--jitconfig").arg(encoded_jit_config); + command + .arg("--jitconfig") + .arg(encoded_jit_config) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); command } @@ -1043,7 +1101,7 @@ mod tests { fn supervisor(capabilities: Vec) -> RunnerSupervisor { let (events, _rx) = mpsc::channel(1); let state = AgentState::new(&seed()); - RunnerSupervisor::new( + RunnerSupervisor::without_logs( events, Some(PathBuf::from("/nonexistent")), Backends::fixed(capabilities, state.clone()), @@ -1091,10 +1149,88 @@ mod tests { interop, state.clone(), ); - let sup = RunnerSupervisor::new(events, None, backends, state); + let sup = RunnerSupervisor::without_logs(events, None, backends, state); (sup, rx) } + /// Like [`windows_supervisor_with_rx`], but also hands back the + /// runner-output receiver so a test can assert on captured output. + fn windows_supervisor_with_logs( + interop: Option, + ) -> (RunnerSupervisor, mpsc::Receiver) { + let (events, _events_rx) = mpsc::channel(8); + let (logs, logs_rx) = mpsc::channel(64); + let state = AgentState::new(&crate::settings::PersistedSettings { + load_ceiling: f64::MAX, + mem_floor_mib: 0, + ..seed() + }); + let backends = Backends::fixed_with_interop( + vec![capability("windows", "amd64", Backend::HostRunner)], + interop, + state.clone(), + ); + // The events receiver is dropped, so verdicts are shed — this test + // is about output, and the supervisor must not depend on a live + // verdict consumer to capture it. + let sup = RunnerSupervisor::new(events, logs, None, backends, state); + (sup, logs_rx) + } + + /// Output a started runner produces must reach the log queue as chunks + /// tagged with the job that produced them. Drives the whole path — + /// `handle_provision` → backend routing → interop drain → `JobLogSink` — + /// rather than calling the sink directly, so a backend that captures + /// nothing fails here. + #[tokio::test] + async fn started_runner_output_reaches_the_log_queue() { + let dir = tempfile::tempdir().unwrap(); + let powershell = interop_stub( + dir.path(), + "powershell", + "echo 'WINPID=4242'\necho 'building the thing'\nexit 0", + ); + let taskkill = interop_stub(dir.path(), "taskkill", "exit 0"); + let interop = InteropRunner::with_paths(powershell, taskkill, r"C:\r\run.cmd"); + + // Same ETXTBSY warm-up as the routing test below: a concurrently + // forking test can hold the freshly written stub open for writing. + for _ in 0..100 { + match interop + .spawn("dGVzdA==", crate::joblog::JobLogSink::discarding()) + .await + { + Ok(mut job) => { + let _ = job.wait().await; + break; + } + Err(_) => tokio::time::sleep(Duration::from_millis(20)).await, + } + } + + let (sup, mut logs_rx) = windows_supervisor_with_logs(Some(interop)); + sup.handle_provision(ProvisionRunner { + job_id: "rjob_win".to_owned(), + os: "windows".to_owned(), + arch: "amd64".to_owned(), + encoded_jit_config: "dGVzdA==".to_owned(), + offer_token: "tok1".to_owned(), + }); + + let chunk = tokio::time::timeout(Duration::from_secs(5), logs_rx.recv()) + .await + .expect("runner output within the handshake budget") + .expect("log queue open"); + match chunk.msg { + Some(attach_request::Msg::RunnerLog(chunk)) => { + assert_eq!(chunk.job_id, "rjob_win"); + assert_eq!(chunk.data, b"building the thing\n"); + assert_eq!(chunk.dropped_bytes, 0); + } + other => panic!("expected RunnerLog, got {other:?}"), + } + } + /// Write an executable stub standing in for powershell/taskkill. fn interop_stub(dir: &std::path::Path, name: &str, body: &str) -> PathBuf { use std::os::unix::fs::PermissionsExt; @@ -1119,7 +1255,7 @@ mod tests { // concurrently forking test can hold it open for writing until its // own exec) so the spawn inside handle_provision can't hit it. for _ in 0..100 { - match interop.spawn("dGVzdA==").await { + match interop.spawn("dGVzdA==", JobLogSink::discarding()).await { Ok(mut job) => { let _ = job.wait().await; break; @@ -1315,7 +1451,7 @@ mod tests { fn supervisor_with_rx(capacity: usize) -> (RunnerSupervisor, mpsc::Receiver) { let (events, rx) = mpsc::channel(capacity); let state = AgentState::new(&seed()); - let sup = RunnerSupervisor::new( + let sup = RunnerSupervisor::without_logs( events, Some(PathBuf::from("/nonexistent")), Backends::fixed( @@ -1340,7 +1476,7 @@ mod tests { mem_floor_mib: 0, ..seed() }); - let sup = RunnerSupervisor::new( + let sup = RunnerSupervisor::without_logs( events, Some(PathBuf::from("/nonexistent")), Backends::fixed( @@ -1693,7 +1829,7 @@ mod tests { async fn shutdown_does_not_flip_observable_draining_but_drain_does() { let drained = AgentState::new(&seed()); let (events, _rx) = mpsc::channel(1); - RunnerSupervisor::new( + RunnerSupervisor::without_logs( events, None, Backends::fixed(Vec::new(), drained.clone()), @@ -1707,7 +1843,7 @@ mod tests { let torn_down = AgentState::new(&seed()); let (events, _rx) = mpsc::channel(1); - RunnerSupervisor::new( + RunnerSupervisor::without_logs( events, None, Backends::fixed(Vec::new(), torn_down.clone()), diff --git a/fleet/arcbox-fleet-agent/src/vm.rs b/fleet/arcbox-fleet-agent/src/vm.rs index 0b3c109a7..4adccdd76 100644 --- a/fleet/arcbox-fleet-agent/src/vm.rs +++ b/fleet/arcbox-fleet-agent/src/vm.rs @@ -37,6 +37,7 @@ use tonic::transport::Channel; use tracing::{debug, info, warn}; use crate::host; +use crate::joblog::JobLogSink; /// Guest login and runner location — the ArcBox macOS runner image contract /// (see the module doc). @@ -380,13 +381,20 @@ impl RunningVm { /// follow with [`destroy`](Self::destroy). The agent reports no outcome /// upstream (the GitHub webhook is authoritative); the code is for /// logging. - pub async fn wait(&mut self) -> Option { + /// + /// The guest is destroyed on every exit path, so this session is the only + /// chance to observe the runner's output: it is forwarded to `sink` as it + /// arrives rather than collected, since nothing here should hold a whole + /// job's log in memory. + pub async fn wait(&mut self, sink: &JobLogSink) -> Option { let mut code = None; while let Some(msg) = self.channel.wait().await { match msg { - // Runner output stays in the disposable guest; nothing to - // collect here. - ChannelMsg::Data { .. } | ChannelMsg::ExtendedData { .. } => {} + // stdout and the stderr extension both carry runner output; + // the guest's own interleaving is the order worth preserving. + ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, .. } => { + sink.write(&data); + } ChannelMsg::ExitStatus { exit_status } => code = Some(exit_status), _ => {} } @@ -508,7 +516,7 @@ mod tests { ) .await .expect("provision guest and exec"); - assert_eq!(running.wait().await, Some(0)); + assert_eq!(running.wait(&JobLogSink::discarding()).await, Some(0)); running.destroy().await; }