Skip to main content

max / makenotwork

5.6 KB · 168 lines History Blame Raw
1 //! Disk-append + callback live-log sink.
2 //!
3 //! Wraps an on-disk per-run log file with a per-chunk callback. A streamed
4 //! command (via [`crate::remote::RemoteHost::run_streaming`]) pushes its
5 //! merged stdout/stderr through `write_chunk` as it arrives; the sink:
6 //! 1. appends to the on-disk file so a post-mortem `GET /logs/...` route
7 //! still has the full byte stream;
8 //! 2. invokes the chunk callback with `(seq, text)` so the owning daemon can
9 //! broadcast a tool-specific log-chunk event (the callback is where the
10 //! coupling to a concrete `Event` enum lives, keeping this module
11 //! generic).
12 //!
13 //! Chunks reflect tokio read boundaries — they are NOT line-aligned.
14 //! Consumers that want lines must reassemble; the on-disk log preserves the
15 //! exact byte stream for that.
16
17 use crate::remote::LogSink;
18 use async_trait::async_trait;
19 use std::path::{Path, PathBuf};
20 use tokio::fs::File;
21 use tokio::io::AsyncWriteExt;
22
23 /// Callback invoked once per chunk with the monotonic per-run sequence number
24 /// and the UTF-8-lossy text. Boxed so the sink stays a concrete type that can
25 /// be shared behind an `Arc<Mutex<_>>`.
26 pub type ChunkCallback = Box<dyn FnMut(u32, &str) + Send>;
27
28 pub struct LiveLog {
29 file: Option<File>,
30 /// Kept for diagnostic logging when file IO is unavailable.
31 path: PathBuf,
32 on_chunk: ChunkCallback,
33 seq: u32,
34 }
35
36 impl LiveLog {
37 /// Open the log file for append-streaming, creating parent directories as
38 /// needed. If the file can't be opened the sink degrades to "callback
39 /// only" — chunks still reach the callback; the missing on-disk log is a
40 /// warning, never a failure (a broken log dir must not turn a passing run
41 /// red).
42 pub async fn open(path: PathBuf, on_chunk: ChunkCallback) -> Self {
43 let file = open_for_append(&path).await;
44 Self {
45 file,
46 path,
47 on_chunk,
48 seq: 0,
49 }
50 }
51
52 /// Flush and close the file. Best-effort; errors are logged.
53 pub async fn close(mut self) {
54 if let Some(mut f) = self.file.take()
55 && let Err(e) = f.flush().await
56 {
57 tracing::warn!(error = %e, path = %self.path.display(), "live log flush failed");
58 }
59 }
60
61 pub fn chunks_emitted(&self) -> u32 {
62 self.seq
63 }
64 }
65
66 #[async_trait]
67 impl LogSink for LiveLog {
68 /// Append `bytes` to the on-disk log and invoke the callback. The callback
69 /// fires even if the disk write fails — operators watching live still see
70 /// the chunk.
71 async fn write_chunk(&mut self, bytes: &[u8]) {
72 if bytes.is_empty() {
73 return;
74 }
75 if let Some(f) = self.file.as_mut()
76 && let Err(e) = f.write_all(bytes).await
77 {
78 tracing::warn!(error = %e, path = %self.path.display(), "live log write failed");
79 self.file = None;
80 }
81 let text = String::from_utf8_lossy(bytes);
82 (self.on_chunk)(self.seq, &text);
83 self.seq = self.seq.saturating_add(1);
84 }
85 }
86
87 async fn open_for_append(path: &Path) -> Option<File> {
88 if let Some(parent) = path.parent()
89 && let Err(e) = tokio::fs::create_dir_all(parent).await
90 {
91 tracing::warn!(error = %e, dir = %parent.display(), "could not create log dir");
92 return None;
93 }
94 match tokio::fs::OpenOptions::new()
95 .create(true)
96 .append(true)
97 .open(path)
98 .await
99 {
100 Ok(f) => Some(f),
101 Err(e) => {
102 tracing::warn!(error = %e, path = %path.display(), "could not open log file");
103 None
104 }
105 }
106 }
107
108 #[cfg(test)]
109 mod tests {
110 use super::*;
111 use std::sync::{Arc, Mutex};
112
113 #[tokio::test]
114 async fn write_chunk_appends_to_file_and_fires_callback() {
115 let dir = tempfile::tempdir().unwrap();
116 let path = dir.path().join("nested/run.log");
117 let seen: Arc<Mutex<Vec<(u32, String)>>> = Arc::new(Mutex::new(Vec::new()));
118 let cb_seen = seen.clone();
119 let mut log = LiveLog::open(
120 path.clone(),
121 Box::new(move |seq, text| cb_seen.lock().unwrap().push((seq, text.to_string()))),
122 )
123 .await;
124 log.write_chunk(b"hello ").await;
125 log.write_chunk(b"world\n").await;
126 log.close().await;
127
128 assert_eq!(
129 tokio::fs::read_to_string(&path).await.unwrap(),
130 "hello world\n"
131 );
132 assert_eq!(
133 *seen.lock().unwrap(),
134 vec![(0, "hello ".to_string()), (1, "world\n".to_string())]
135 );
136 }
137
138 #[tokio::test]
139 async fn callback_fires_even_when_file_cannot_open() {
140 let dir = tempfile::tempdir().unwrap();
141 let blocker = dir.path().join("blocker");
142 tokio::fs::write(&blocker, b"i am a file").await.unwrap();
143 let path = blocker.join("inside.log"); // parent is a file
144 let seen = Arc::new(Mutex::new(Vec::new()));
145 let cb_seen = seen.clone();
146 let mut log = LiveLog::open(
147 path,
148 Box::new(move |_, text| cb_seen.lock().unwrap().push(text.to_string())),
149 )
150 .await;
151 log.write_chunk(b"streamed despite no file\n").await;
152 log.close().await;
153 assert_eq!(
154 *seen.lock().unwrap(),
155 vec!["streamed despite no file\n".to_string()]
156 );
157 }
158
159 #[tokio::test]
160 async fn empty_chunk_is_noop() {
161 let dir = tempfile::tempdir().unwrap();
162 let path = dir.path().join("empty.log");
163 let mut log = LiveLog::open(path, Box::new(|_, _| {})).await;
164 log.write_chunk(b"").await;
165 assert_eq!(log.chunks_emitted(), 0);
166 }
167 }
168