robonix_scribe/
sink.rs

1//! Per-tag file sink — JSON-lines output, one file per tag.
2//!
3//! A [`FileSink`] owns a directory (`$SCRIBE_LOG_DIR` or `./logs`) and a
4//! `Mutex<HashMap<tag, File>>`.  The first write for a given `tag` opens
5//! (or creates) `{dir}/{tag}.log`; subsequent writes reuse the handle.
6//!
7//! Every line is a JSON-serialised [`LogRecord`] followed by `\n`.  The
8//! sink flushes after each write so a crash does not lose the last
9//! record.
10
11use crate::LogRecord;
12use std::collections::HashMap;
13use std::fs::{self, File, OpenOptions};
14use std::io::{self, Write};
15use std::path::{Path, PathBuf};
16use std::sync::Mutex;
17
18/// Per-tag file sink.
19///
20/// # Thread safety
21///
22/// All writes go through a single `Mutex`.  Contention is low because
23/// writes are small and every record is flushed immediately.
24pub struct FileSink {
25    dir: PathBuf,
26    writers: Mutex<HashMap<String, File>>,
27}
28
29impl FileSink {
30    /// Create a new file sink rooted at `log_dir`.
31    ///
32    /// Creates the directory (and any missing parents) if it does not
33    /// already exist.
34    pub fn new(log_dir: impl Into<PathBuf>) -> io::Result<Self> {
35        let dir = log_dir.into();
36        fs::create_dir_all(&dir)?;
37        Ok(Self {
38            dir,
39            writers: Mutex::new(HashMap::new()),
40        })
41    }
42
43    /// Write one [`LogRecord`] to `{dir}/{record.tag}.log`.
44    ///
45    /// Opens the file in append mode on first use for a given tag; caches
46    /// the handle thereafter.  Serialises the record as a single JSON
47    /// line, writes it, and flushes.
48    ///
49    /// # Errors
50    ///
51    /// Returns `io::Error` if the file can't be opened or written.
52    /// The caller (`log()` in lib.rs) treats this as best-effort and
53    /// silently drops the file write on failure.
54    pub fn write(&self, record: &LogRecord) -> io::Result<()> {
55        let json_line = serde_json::to_string(record)?;
56        let mut guard = self.writers.lock().unwrap_or_else(|e| e.into_inner());
57        // or_insert_with can't use `?`; use entry + insert to propagate errors.
58        let writer: &mut File = if let Some(w) = guard.get_mut(&record.tag) {
59            w
60        } else {
61            let path = self.tag_path(&record.tag);
62            let f = OpenOptions::new().create(true).append(true).open(&path)?;
63            guard.entry(record.tag.clone()).or_insert(f)
64        };
65        writeln!(writer, "{json_line}")?;
66        writer.flush()?;
67        Ok(())
68    }
69
70    /// Absolute path for a given tag's log file.
71    ///
72    /// Non-safe characters (anything outside `[A-Za-z0-9._-]`) are
73    /// replaced with `_` to prevent path traversal / invalid filenames.
74    fn tag_path(&self, tag: &str) -> PathBuf {
75        let safe: String = tag
76            .chars()
77            .map(|c| {
78                if c.is_ascii_alphanumeric() || c == '.' || c == '_' || c == '-' {
79                    c
80                } else {
81                    '_'
82                }
83            })
84            .collect();
85        self.dir.join(format!("{safe}.log"))
86    }
87
88    /// Return the log directory path (for diagnostics).
89    #[allow(dead_code)]
90    pub fn dir(&self) -> &Path {
91        &self.dir
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use crate::Level;
99    use std::io::BufRead;
100
101    fn test_sink() -> (tempfile::TempDir, FileSink) {
102        let tmp = tempfile::tempdir().unwrap();
103        let sink = FileSink::new(tmp.path()).unwrap();
104        (tmp, sink)
105    }
106
107    fn read_lines(path: &Path) -> Vec<String> {
108        let f = File::open(path).unwrap();
109        io::BufReader::new(f).lines().map(|l| l.unwrap()).collect()
110    }
111
112    #[test]
113    fn first_write_creates_file() {
114        let (tmp, sink) = test_sink();
115        let rec = LogRecord {
116            ts: 1234567890,
117            level: Level::Info,
118            tag: "atlas".into(),
119            msg: "hello".into(),
120        };
121        sink.write(&rec).unwrap();
122
123        let log_path = tmp.path().join("atlas.log");
124        assert!(log_path.exists(), "expected {log_path:?} to exist");
125
126        let lines = read_lines(&log_path);
127        assert_eq!(lines.len(), 1);
128        let parsed: LogRecord = serde_json::from_str(&lines[0]).unwrap();
129        assert_eq!(parsed.tag, "atlas");
130        assert_eq!(parsed.msg, "hello");
131    }
132
133    #[test]
134    fn same_tag_appends() {
135        let (tmp, sink) = test_sink();
136        let rec1 = LogRecord {
137            ts: 1,
138            level: Level::Debug,
139            tag: "a".into(),
140            msg: "m1".into(),
141        };
142        let rec2 = LogRecord {
143            ts: 2,
144            level: Level::Info,
145            tag: "a".into(),
146            msg: "m2".into(),
147        };
148        sink.write(&rec1).unwrap();
149        sink.write(&rec2).unwrap();
150
151        let lines = read_lines(&tmp.path().join("a.log"));
152        assert_eq!(lines.len(), 2);
153        let r1: LogRecord = serde_json::from_str(&lines[0]).unwrap();
154        let r2: LogRecord = serde_json::from_str(&lines[1]).unwrap();
155        assert_eq!(r1.msg, "m1");
156        assert_eq!(r2.msg, "m2");
157    }
158
159    #[test]
160    fn different_tags_different_files() {
161        let (tmp, sink) = test_sink();
162        sink.write(&LogRecord {
163            ts: 0,
164            level: Level::Warn,
165            tag: "x".into(),
166            msg: "mx".into(),
167        })
168        .unwrap();
169        sink.write(&LogRecord {
170            ts: 0,
171            level: Level::Error,
172            tag: "y".into(),
173            msg: "my".into(),
174        })
175        .unwrap();
176
177        let x_path = tmp.path().join("x.log");
178        let y_path = tmp.path().join("y.log");
179        assert!(x_path.exists());
180        assert!(y_path.exists());
181
182        let x_lines = read_lines(&x_path);
183        let y_lines = read_lines(&y_path);
184        assert_eq!(x_lines.len(), 1);
185        assert_eq!(y_lines.len(), 1);
186
187        let rx: LogRecord = serde_json::from_str(&x_lines[0]).unwrap();
188        let ry: LogRecord = serde_json::from_str(&y_lines[0]).unwrap();
189        assert_eq!(rx.tag, "x");
190        assert_eq!(ry.tag, "y");
191    }
192
193    #[test]
194    fn concurrent_writes_no_data_loss() {
195        use std::sync::Arc;
196        use std::thread;
197
198        let tmp = tempfile::tempdir().unwrap();
199        let sink = Arc::new(FileSink::new(tmp.path()).unwrap());
200        const THREADS: usize = 8;
201        const MSGS_PER_THREAD: usize = 100;
202
203        let mut handles = vec![];
204        for t in 0..THREADS {
205            let sink = Arc::clone(&sink);
206            handles.push(thread::spawn(move || {
207                for i in 0..MSGS_PER_THREAD {
208                    sink.write(&LogRecord {
209                        ts: (t * MSGS_PER_THREAD + i) as u64,
210                        level: Level::Info,
211                        tag: "shared".into(),
212                        msg: format!("t{t}-m{i}"),
213                    })
214                    .unwrap();
215                }
216            }));
217        }
218
219        for h in handles {
220            h.join().unwrap();
221        }
222
223        let lines = read_lines(&tmp.path().join("shared.log"));
224        assert_eq!(lines.len(), THREADS * MSGS_PER_THREAD);
225        // Every line should be parseable
226        for line in &lines {
227            let _rec: LogRecord = serde_json::from_str(line).unwrap();
228        }
229    }
230
231    #[test]
232    fn sanitize_tag_replaces_dangerous_chars() {
233        let (tmp, sink) = test_sink();
234        // Tag with `/`, `..`, and spaces — should be sanitized into a safe filename.
235        let rec = LogRecord {
236            ts: 0,
237            level: Level::Info,
238            tag: "a/../b c".into(),
239            msg: "m".into(),
240        };
241        sink.write(&rec).unwrap();
242        // `/` becomes `_`, `.` is safe, ` ` becomes `_` → "a_.._b_c"
243        let safe_path = tmp.path().join("a_.._b_c.log");
244        assert!(
245            safe_path.exists(),
246            "expected {safe_path:?} to exist after sanitization"
247        );
248    }
249
250    #[test]
251    fn write_does_not_panic_on_bad_path() {
252        // Create a sink pointed at a path that can't be a directory.
253        let sink = FileSink::new("/proc/sysrq-trigger"); // existing file, not a dir
254        assert!(
255            sink.is_err(),
256            "should fail to create sink under a non-directory"
257        );
258    }
259
260    #[test]
261    fn sink_creation_fails_gracefully() {
262        // /dev/null is a file, not a directory — create_dir_all should fail.
263        let result = FileSink::new("/dev/null/logs");
264        assert!(result.is_err());
265    }
266}
robonix_scribe/sink.rs

robonix_scribe/
sink.rs
