Skip to main content

max / makenotwork

7.1 KB · 201 lines History Blame Raw
1 //! The gate live log: where a running gate's output goes.
2 //!
3 //! Every gate that shells out streams its bytes through here, so the TUI sees
4 //! the tail as it happens and the finished log is on disk under
5 //! [`super::GateCtx::log_path`] for the failure note to point at.
6
7 use super::GateCtx;
8 use crate::domain::{GateKind, GateRunId};
9 use crate::events::{self, Event, EventTx};
10 use anyhow::Result;
11 use ops_core::live_log::LiveLog;
12 use ops_core::remote::LogSink;
13 use std::path::PathBuf;
14 use std::sync::Arc;
15 use tokio::io::AsyncReadExt;
16 use tokio::process::Command;
17
18 /// The gate live-log callback: emit each chunk as a `GateLogChunk` event so the
19 /// TUI sees the tail stream in real time. `ops_core::live_log::LiveLog` owns the
20 /// disk append and the per-run sequence counter; this closure is the one
21 /// tool-specific bit.
22 pub(super) fn gate_chunk_cb(
23 events: EventTx,
24 run_id: GateRunId,
25 ) -> ops_core::live_log::ChunkCallback {
26 Box::new(move |seq, text| {
27 events::emit(
28 &events,
29 Event::GateLogChunk {
30 run_id,
31 seq,
32 text: text.to_owned(),
33 },
34 );
35 })
36 }
37
38 /// Append raw bytes to a gate log outside the child-streaming path (target
39 /// banners). Best-effort, same as `LiveLog`: a broken log dir never turns a
40 /// passing gate red.
41 pub(super) async fn append_to_log(path: &std::path::Path, bytes: &[u8]) {
42 use tokio::io::AsyncWriteExt;
43 if let Some(parent) = path.parent()
44 && tokio::fs::create_dir_all(parent).await.is_err()
45 {
46 return;
47 }
48 if let Ok(mut f) = tokio::fs::OpenOptions::new()
49 .create(true)
50 .append(true)
51 .open(path)
52 .await
53 {
54 let _ = f.write_all(bytes).await;
55 }
56 }
57
58 /// Drain `stream` into the shared `LiveLog` (which forwards each chunk to
59 /// the on-disk log file AND broadcasts a `GateLogChunk` event), and return
60 /// the concatenated bytes so the classifier can still operate on the full
61 /// output post-hoc.
62 pub(super) async fn stream_into_log<R>(
63 stream: Option<R>,
64 log: std::sync::Arc<tokio::sync::Mutex<LiveLog>>,
65 ) -> Vec<u8>
66 where
67 R: tokio::io::AsyncRead + Unpin + Send + 'static,
68 {
69 let mut total = Vec::new();
70 let Some(mut s) = stream else { return total };
71 let mut buf = [0u8; 4096];
72 loop {
73 match s.read(&mut buf).await {
74 Ok(0) => break,
75 Err(_) => break,
76 Ok(n) => {
77 total.extend_from_slice(&buf[..n]);
78 log.lock().await.write_chunk(&buf[..n]).await;
79 }
80 }
81 }
82 total
83 }
84
85 /// Spawn a child, drain its stdout/stderr through a `LiveLog`, return the
86 /// combined buffers and exit status. Shared by `cargo_test` (no deadline)
87 /// and ad-hoc callers — `boot_smoke` rolls its own variant because of its
88 /// 3s kill window.
89 pub(super) async fn stream_child_to_live_log(
90 child: &mut tokio::process::Child,
91 events: EventTx,
92 run_id: GateRunId,
93 log_path: PathBuf,
94 ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
95 let log = GateLog::new(LiveLog::open(log_path, gate_chunk_cb(events, run_id)).await);
96 let out = log.stream_child(child).await;
97 log.close().await;
98 out
99 }
100
101 /// One gate's live log, held across every step of a multi-step gate.
102 ///
103 /// Single-child gates call [`stream_child_to_live_log`] and are done. The
104 /// staged gates (`code_smoke`, `migration_dry_run`) instead run several
105 /// children plus banner lines between them, and they hold one `GateLog` across
106 /// the lot: a single sink means chunk sequence numbers stay monotonic for the
107 /// whole gate, and the on-disk log reads in the order things actually happened
108 /// rather than as stdout-then-stderr assembled at the end.
109 ///
110 /// Every write is best-effort in the same way `LiveLog` is: a log directory
111 /// that cannot be written degrades to callback-only and never turns a passing
112 /// gate red.
113 pub(super) struct GateLog {
114 sink: Arc<tokio::sync::Mutex<LiveLog>>,
115 }
116
117 impl GateLog {
118 pub(super) fn new(sink: LiveLog) -> Self {
119 Self {
120 sink: Arc::new(tokio::sync::Mutex::new(sink)),
121 }
122 }
123
124 /// Open the sink for `gate`'s log file, streaming to the TUI under `run_id`.
125 pub(super) async fn open(ctx: &GateCtx, run_id: GateRunId, gate: GateKind) -> Self {
126 Self::new(
127 LiveLog::open(
128 ctx.log_path(gate),
129 gate_chunk_cb(ctx.events.clone(), run_id),
130 )
131 .await,
132 )
133 }
134
135 /// Emit a banner (or any line the gate itself produces) through the same
136 /// sink the children stream to, so it lands in sequence with their output.
137 pub(super) async fn write(&self, bytes: &[u8]) {
138 self.sink.lock().await.write_chunk(bytes).await;
139 }
140
141 /// Same, for the common `format!`-a-line case.
142 pub(super) async fn line(&self, s: &str) {
143 self.write(s.as_bytes()).await;
144 }
145
146 /// Spawn `cmd` with both pipes captured, stream them into the sink as they
147 /// arrive, and return the buffers plus the exit status. The buffers are
148 /// what the classifiers still operate on post-hoc.
149 pub(super) async fn run(
150 &self,
151 cmd: &mut Command,
152 ) -> std::io::Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
153 cmd.stdout(std::process::Stdio::piped())
154 .stderr(std::process::Stdio::piped());
155 let mut child = cmd.spawn()?;
156 self.stream_child(&mut child)
157 .await
158 .map_err(std::io::Error::other)
159 }
160
161 /// Drain an already-spawned child's pipes into the sink and wait for it.
162 pub(super) async fn stream_child(
163 &self,
164 child: &mut tokio::process::Child,
165 ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
166 let (stdout_task, stderr_task) = self.drain_pipes(child);
167 let status = child.wait().await?;
168 let stdout_buf = stdout_task.await.unwrap_or_default();
169 let stderr_buf = stderr_task.await.unwrap_or_default();
170 Ok((stdout_buf, stderr_buf, status))
171 }
172
173 /// Start draining a child's pipes into the sink *without* waiting on the
174 /// child. For `code_smoke`'s serve phase, which probes `/health` while the
175 /// server is still up. The caller must await both handles for the buffers
176 /// (and before [`Self::close`], so the flush isn't skipped).
177 #[allow(clippy::type_complexity)]
178 pub(super) fn drain_pipes(
179 &self,
180 child: &mut tokio::process::Child,
181 ) -> (
182 tokio::task::JoinHandle<Vec<u8>>,
183 tokio::task::JoinHandle<Vec<u8>>,
184 ) {
185 (
186 tokio::spawn(stream_into_log(child.stdout.take(), self.sink.clone())),
187 tokio::spawn(stream_into_log(child.stderr.take(), self.sink.clone())),
188 )
189 }
190
191 /// Flush the file. A still-outstanding streaming task (only possible if the
192 /// gate returned without awaiting it) leaves the `Arc` shared, in which case
193 /// the flush is skipped — `LiveLog` writes unbuffered to the OS either way,
194 /// so nothing already written is lost.
195 pub(super) async fn close(self) {
196 if let Ok(mutex) = Arc::try_unwrap(self.sink) {
197 mutex.into_inner().close().await;
198 }
199 }
200 }
201