feat(agent): stream per-job runner output to the gateway - #503
Conversation
Vendored ahead of the BSR publish from the platform repo; the next `make fleet-proto-sync` reconciles this copy with the pushed module.
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.
![pullfrog[bot]](https://avatars.githubusercontent.com/in/1768019?s=60&v=4){kind=small}
There was a problem hiding this comment.
ℹ️ No critical issues — one concurrency rough edge on a best-effort counter, inline.
Reviewed changes — the fleet agent now captures each runner's stdout/stderr and streams it to the gateway on the existing attach stream for live tailing, as a deliberately lossy path that never delays the control-plane heartbeat.
- Add
RunnerLogChunkto the attach oneof — new fieldrunner_log = 6onAttachRequest, carryingjob_id, per-jobseq, rawdata, and a best-effortdropped_bytes; documented as the one lossy, never-acked message on the stream. - New
joblog.rssink —JobLogSinksplits output into 32KiB chunks, gates each through aTokenBucket(256KiB/s sustained, 1MiB burst), thentry_sends onto a dedicated log channel; shed bytes are counted and reported on the next successful chunk. - Split egress from output in
attach.rs—spawn_supervisorreturnsSupervisorHandles(a named struct, not a tuple of two same-typed receivers) andconnect_and_servegains alog_rxarm that usestry_send, is never awaited, and never parks in the verdict-durabilitypendingslot. - Wire all four backends — SSH
Data/ExtendedData(vm.rs),docker logs --follow(docker.rs), and piped stdout/stderr for host (runner.rs) and Windows-interop (interop.rs) runners; host output is now piped rather than inherited (behavior change, documented). - Tests — chunks reach a real gateway stream while leaving
pendinguntouched; oversized writes split and number in order; a full channel sheds, burns aseq, and reports the loss; sustained output past the burst allowance is shed; end-to-end throughhandle_provision→ backend routing → sink.
Claude Opus ｜ 𝕏
| // 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); |
There was a problem hiding this comment.
pipe/reader tasks, so two write_chunk calls can both read the same reported at line 111, both try_send OK, and both fetch_sub(reported) here. The first zeroes the counter; the second underflows the AtomicU64 to ~u64::MAX, so the next successful chunk claims exabytes dropped — and both chunks already over-reported the same reported value to the consumer.
This only corrupts the best-effort dropped_bytes field (the seq gap remains the authoritative loss signal), so it is non-blocking, but it defeats the care the surrounding accounting takes.
Technical details
# Concurrent write_chunk double-subtracts the shared dropped counter
## Affected sites
- `fleet/arcbox-fleet-agent/src/joblog.rs:111` — `reported = dropped.load(Relaxed)`
- `fleet/arcbox-fleet-agent/src/joblog.rs:124-125` — success path `dropped.fetch_sub(reported, Relaxed)`
## Why it triggers
- The `budget` mutex is released after `take()`, so steps load→build→send→clear run unserialized.
- Host runner (`run_host_job`) pipes stdout and stderr as two tasks sharing one `JobLogSink` clone; interop pipes stderr while a line-reader task writes stdout to the same sink. Both are real concurrent `write_chunk` callers.
- Interleave: dropped=D>0; A reads reported=D; B reads reported=D; A send OK, fetch_sub(D)→0; B send OK, fetch_sub(D)→wraps to u64::MAX-D+1.
## Required outcome
- A successful send must clear only the amount that chunk actually reported, and the counter must never underflow, even when two sends observe the same `reported`.
## Suggested approach (optional)
- Move the load + build-`dropped_bytes` + send + clear under the existing `budget` mutex (it is already the single serialization point and holds no await), so `reported` and its subtraction are atomic w.r.t. other writers.
- Or replace the `load`/`fetch_sub` pair with a CAS loop that subtracts `min(reported, current)`, so the clear is claim-once and saturating.
Greptile SummaryAdds lossy live streaming of per-job runner output over the attach stream.
Confidence Score: 2/5The PR should not merge until dropped-byte accounting is made concurrency-safe and log traffic is prevented from occupying heartbeat and verdict capacity. Concurrent pipe readers can underflow the shared loss counter, and a slow attach stream allows log chunks to fill the same downstream queue used by control-plane messages. Files Needing Attention: fleet/arcbox-fleet-agent/src/joblog.rs, fleet/arcbox-fleet-agent/src/attach.rs
 T-Rex LogsWhat T-Rex did
|
| Filename | Overview |
|---|---|
| fleet/arcbox-fleet-agent/src/joblog.rs | Introduces bounded chunking and loss accounting, but concurrent stdout/stderr writers can corrupt dropped-byte metadata. |
| fleet/arcbox-fleet-agent/src/attach.rs | Multiplexes logs onto the attach stream, but the final request queue remains shared with heartbeat and verdict traffic. |
| fleet/arcbox-fleet-agent/src/runner.rs | Wires per-job sinks into all runner lifecycle paths and drains pipe-backed stdout and stderr. |
| fleet/arcbox-fleet-agent/src/docker.rs | Adds detached Docker log following for combined container stdout and stderr. |
| fleet/arcbox-fleet-agent/src/interop.rs | Captures Windows wrapper output after the PID handshake while continuously draining both pipes. |
| fleet/arcbox-fleet-agent/src/vm.rs | Forwards SSH data and extended-data frames through the per-job sink. |
| fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto | Adds the RunnerLogChunk attach variant with job, sequence, raw data, and loss fields. |
Flowchart
%%{init: {'theme': 'neutral'}}%%
flowchart LR
R[Runner stdout/stderr] --> S[Per-job JobLogSink]
S -->|token budget + try_send| L[Log queue]
L --> A[Attach connection loop]
A --> Q[Shared request queue]
H[Heartbeat task] --> Q
V[Lifecycle events] --> A
Q --> G[Gateway attach stream]
Reviews (1): Last reviewed commit: "feat(agent): stream per-job runner outpu..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| 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); |
There was a problem hiding this comment.
When host or Windows-interop jobs write stdout and stderr concurrently after a drop, both writers can load and report the same nonzero counter before each subtracts it. This underflows the AtomicU64, causing a later chunk to report a dropped_bytes value near u64::MAX.
Artifacts
Repro: focused test injector that executes concurrent writes through the repository JobLogSink
- Evidence file captured while the check ran.
- The full command output behind this check.
| Some(msg) => { | ||
| if req_tx.try_send(msg).is_err() { | ||
| tracing::trace!("runner log chunk dropped; request channel full"); |
There was a problem hiding this comment.
When a chatty job fills req_tx while the gRPC stream drains slowly, heartbeats and verdicts block behind queued log chunks because they share this downstream channel. Heartbeat delay can mark the machine offline, and a blocked verdict send stalls cancellation, drain, and acknowledgement handling in the connection loop.
Artifacts
Repro: focused Tokio and tonic test harness patch
- Evidence file captured while the check ran.
Repro: focused test output showing the verdict blocked for 750 ms and delivered behind 96 log chunks
- The full command output behind this check.
Capture each runner's output on the machine and report it on the existing attach
stream, so a job's log can be tailed live while it runs. Nothing is stored on
the machine.
Shape
RunnerLogChunkadded to theAttachRequestoneof, carryingjob_id, aper-job
seq, raw bytes, and adropped_bytescount.Data/ExtendedDatafor the VM runner,docker logs --followfor the container runner, and piped stdout/stderr forthe host and Windows-interop runners.
is a control plane: a heartbeat later than 60s marks the machine offline and
stops placement. Output can arrive orders of magnitude faster than the egress
channel drains, so every write passes a per-job token budget and a
non-blocking
try_send, and anything that does not fit is dropped andcounted. A chatty job loses log lines, never throughput.
try_send,never awaited, and never parked in the
pendingslot that carries anundelivered verdict across a reconnect.
spawn_supervisorreturns a namedstruct rather than a tuple of two same-typed receivers, because swapping them
would make output durable and verdicts lossy and still compile.
Stdio::piped()that is a liveness requirement, not just capture.
Behaviour change
Host-runner output is now piped rather than inherited, so it no longer appears
in the service manager's stdout/stderr capture, where it used to land
interleaved and unattributed. Use the live tail instead.
Tests
pendingslot untouched — the copy-paste bug a neighbouring
select!arm invites, andthe one thing nothing else would catch.
seq, and reports the loss on the next chunk; sustained output past the burstallowance is shed rather than queued.
handle_provision→ backend routing → sink, so a backendthat captures nothing fails.