shep_core/barks.rs
1//! `barks.jsonl`: the size-capped ring of fired alerts.
2//!
3//! One [`Bark`] per line, appended by the bark dog on a rule fire and by
4//! the shepherd on a dog's exhausted restart budget, and read by `shep
5//! barks`. [`append`] evicts oldest-first under a byte cap by rewriting
6//! survivors to a temp file and `rename`ing it over the original, so a
7//! writer that dies mid-rewrite never leaves a fragment. [`read`] skips
8//! any unparseable line, since this file is read during an incident, and
9//! [`read_last`] answers the same question over a window of the file's
10//! end so a caller that wants only the recent records does not pay for
11//! the whole ring.
12//!
13//! The two writers are separate OS processes, so a [`crate::file_lock`] on
14//! a sibling `<path>.lock` serializes them.
15
16// The lock type lives in shep-core rather than shep-daemon so both writers
17// can name it.
18use core::fmt;
19use std::fs::File;
20use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
21use std::path::Path;
22
23use serde::{Deserialize, Serialize};
24
25use crate::file_lock::FileLock;
26
27/// Cap the ring keeps itself under when nobody configured one.
28pub const DEFAULT_MAX_BYTES: u64 = 1024 * 1024;
29
30/// The first window [`read_last`] reads from the end of the ring, and the
31/// factor it grows by while the window holds fewer records than were
32/// asked for. 64 KiB (65,536 bytes). Serialised records measure 149 to
33/// 173 bytes (a rule firing with one sink, delivered or failed; a
34/// shepherd-written restart-budget bark is 160), so one window holds
35/// roughly 375 records and covers a typical tail request in one read.
36/// It is 1/16 of a [`DEFAULT_MAX_BYTES`] ring: four doublings reach the
37/// whole of one.
38const TAIL_WINDOW_BYTES: u64 = 64 * 1024;
39
40/// One fired alert, as it lands in `$SHEP_HOME/barks.jsonl`.
41///
42/// One JSON object per line: an interrupted write then costs the reader
43/// one record, not the whole file.
44///
45/// `Debug` is derived, not redacted. Every field here is shep's own prose
46/// or a config key, never a sink's target; a field that ever carried a
47/// webhook URL or token would need its own redacted `Debug`.
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49pub struct Bark {
50 /// Unix millis when the alert fired.
51 pub at_ms: u64,
52 /// The rule that fired, or `daemon` when the shepherd wrote this
53 /// itself.
54 pub rule: String,
55 /// What it is about: a sheep's name, or a dog's.
56 pub subject: String,
57 /// The human-readable line. Plain English, no theme: this is read
58 /// during an incident.
59 pub message: String,
60 /// Which sinks the alert was delivered to, and whether each took it.
61 /// Empty when the shepherd wrote the record itself: it has no sinks
62 /// and no webhook code, and says so by carrying none.
63 pub sinks: Vec<SinkOutcome>,
64}
65
66/// What one sink made of one alert.
67///
68/// Names the sink by its `[dog.bark.sinks]` config key, never by its
69/// webhook URL or bearer token, so [`Bark`] stays safe to print with a
70/// derived `Debug`.
71#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
72pub struct SinkOutcome {
73 /// The sink's name from `[dog.bark.sinks]`.
74 pub sink: String,
75 /// `None` when it was delivered; the failure otherwise.
76 pub error: Option<String>,
77}
78
79/// Error type returned by [`append`] and [`read`].
80///
81/// Wraps `io::Error`/`serde_json::Error` directly so callers keep the
82/// underlying diagnostic via [`core::error::Error::source`]; this enum
83/// cannot derive `Clone`/`PartialEq`/`Eq` as a result.
84///
85/// `#[non_exhaustive]`: shep-core is a published library, so a future
86/// failure variant must not break an out-of-tree consumer's `match`.
87#[non_exhaustive]
88#[derive(Debug)]
89pub enum BarkError {
90 /// The ring file could not be read, written, or replaced.
91 Io(std::io::Error),
92 /// A [`Bark`] could not be serialized to JSON.
93 Encode(serde_json::Error),
94}
95
96impl fmt::Display for BarkError {
97 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
98 match self {
99 Self::Io(err) => write!(f, "bark ring I/O failed: {err}"),
100 Self::Encode(err) => write!(f, "bark record failed to serialize: {err}"),
101 }
102 }
103}
104
105impl core::error::Error for BarkError {
106 fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
107 match self {
108 Self::Io(err) => Some(err),
109 Self::Encode(err) => Some(err),
110 }
111 }
112}
113
114impl From<std::io::Error> for BarkError {
115 fn from(source: std::io::Error) -> Self {
116 Self::Io(source)
117 }
118}
119
120impl From<serde_json::Error> for BarkError {
121 fn from(source: serde_json::Error) -> Self {
122 Self::Encode(source)
123 }
124}
125
126/// Appends `bark` to `path`, evicting oldest-first to keep the file under
127/// `max_bytes`.
128///
129/// Eviction drops a prefix of whole lines, atomically; an oversized
130/// record is written anyway, since dropping it would silently lose the
131/// one alert too big to fit. Serialized against other appenders by an
132/// advisory lock on a sibling `<path>.lock`. Concurrent [`read`]s are not
133/// blocked: the ring is only ever replaced whole.
134///
135/// # Errors
136/// - [`BarkError::Io`]: the file or its lock could not be read, written, or replaced.
137/// - [`BarkError::Encode`]: the record could not be serialized.
138pub fn append(path: &Path, bark: &Bark, max_bytes: u64) -> Result<(), BarkError> {
139 // Held until this returns, so the read and the final rename are one
140 // transaction as far as any other writer is concerned.
141 let _lock = FileLock::acquire(path)?;
142
143 let mut lines = read_lines(path)?;
144 let new_line = serde_json::to_string(bark)?;
145 lines.push(new_line);
146
147 // Oldest-out: count the prefix that has to go, then drop it in one
148 // `drain`. Line-at-a-time removal is quadratic in the lines evicted.
149 // A lowered cap makes that nearly the whole ring. The last line
150 // stays whatever its size: it is the record just appended.
151 let mut ring = ring_bytes(&lines);
152 let mut evicted = 0;
153 while ring > max_bytes && lines.len() - evicted > 1 {
154 ring -= line_bytes(&lines[evicted]);
155 evicted += 1;
156 }
157 lines.drain(..evicted);
158
159 write_ring(path, &lines)
160}
161
162/// Reads every bark in `path`, oldest first, skipping any line that will
163/// not parse.
164///
165/// A line that will not parse is a partially-written record from a writer
166/// that died mid-append, or a record from a future shep. Neither refuses
167/// the whole history during an incident, which is the one time this file
168/// is read.
169///
170/// # Errors
171/// - [`BarkError::Io`]: the file exists and could not be read. A missing
172/// file is `Ok(Vec::new())`: no barks yet is not a fault.
173pub fn read(path: &Path) -> Result<Vec<Bark>, BarkError> {
174 Ok(read_text(path)?
175 .lines()
176 .filter_map(|line| serde_json::from_str(line).ok())
177 .collect())
178}
179
180/// The last `limit` barks in `path`, oldest first, without reading the
181/// whole ring: a byte window from the end of the file, grown until it
182/// holds `limit` records or the file runs out.
183///
184/// The same answer [`read`] gives for its last `limit` records — a
185/// missing file is an empty ring, an unparseable line is skipped — so the
186/// only difference is what the reader pays for. A byte sequence the
187/// window cuts mid-character is replaced rather than refused: this file
188/// is read during an incident, and a record the reader cannot have is
189/// not worth failing the read over.
190///
191/// # Errors
192/// - [`BarkError::Io`]: the file exists and could not be opened, stat'ed,
193/// seeked or read.
194pub fn read_last(path: &Path, limit: usize) -> Result<Vec<Bark>, BarkError> {
195 if limit == 0 {
196 return Ok(Vec::new());
197 }
198 let mut file = match File::open(path) {
199 Ok(file) => file,
200 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
201 Err(err) => return Err(BarkError::Io(err)),
202 };
203 let len = file.metadata()?.len();
204 let mut window = TAIL_WINDOW_BYTES.min(len);
205 loop {
206 let mut barks = read_window(&mut file, len, window)?;
207 if barks.len() >= limit || window >= len {
208 let keep_from = barks.len().saturating_sub(limit);
209 barks.drain(..keep_from);
210 return Ok(barks);
211 }
212 // Doubling rather than a fixed step: the window that finally
213 // suffices is within a factor of two of this one, so the total
214 // read is bounded by twice the answer's size rather than by the
215 // file's.
216 window = window.saturating_mul(2).min(len);
217 }
218}
219
220/// The parseable records in the last `window` bytes of the file, oldest
221/// first. A window that starts mid-file may start mid-record, so
222/// everything up to the first newline is that record's tail and is
223/// dropped rather than offered to the parser as a fragment.
224fn read_window(file: &mut File, len: u64, window: u64) -> Result<Vec<Bark>, BarkError> {
225 let start = len - window;
226 // Reads from one byte before the window when there is one. A window
227 // starting exactly on a record boundary then finds the previous
228 // record's newline first and drops only that byte, not its own first
229 // record.
230 let probe = u64::from(start > 0);
231 // Every iteration seeks, including the first and including a grown
232 // window that now covers the whole file: the previous read left the
233 // position at the end, and a `read_to_end` from there would see
234 // nothing.
235 file.seek(SeekFrom::Start(start - probe))?;
236 let mut bytes = Vec::new();
237 file.read_to_end(&mut bytes)?;
238 let text = if start > 0 {
239 match bytes.iter().position(|&b| b == b'\n') {
240 Some(newline) => String::from_utf8_lossy(&bytes[newline + 1..]),
241 None => return Ok(Vec::new()),
242 }
243 } else {
244 String::from_utf8_lossy(&bytes)
245 };
246 Ok(text
247 .lines()
248 .filter_map(|line| serde_json::from_str(line).ok())
249 .collect())
250}
251
252/// `path`'s existing lines, raw and unparsed, or an empty ring if the
253/// file does not exist yet.
254///
255/// Does not parse: eviction operates on whole lines exactly as they sit
256/// on disk, so a line a future shep wrote, or a fragment a dead writer
257/// left, still counts toward the byte cap and survives an eviction it
258/// does not trigger. [`read`] is where an unparseable line is dropped.
259fn read_lines(path: &Path) -> Result<Vec<String>, BarkError> {
260 Ok(read_text(path)?.lines().map(str::to_owned).collect())
261}
262
263/// `path`'s bytes, or an empty string if the file is not there yet.
264///
265/// A ring nobody has written to is an empty ring, not a fault, and both
266/// readers above want exactly that reading before they go their separate
267/// ways on what to do with the lines.
268fn read_text(path: &Path) -> Result<String, BarkError> {
269 match std::fs::read_to_string(path) {
270 Ok(text) => Ok(text),
271 Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(String::new()),
272 Err(err) => Err(BarkError::Io(err)),
273 }
274}
275
276/// On-disk size, in bytes, of one line: its own bytes plus the `\n`
277/// [`write_ring`] writes after it.
278fn line_bytes(line: &str) -> u64 {
279 line.len() as u64 + 1
280}
281
282/// Total on-disk size, in bytes, if `lines` were written one per line.
283fn ring_bytes(lines: &[String]) -> u64 {
284 lines.iter().map(|line| line_bytes(line)).sum()
285}
286
287/// Rewrites `path` to hold exactly `lines`: the content lands in a
288/// uniquely-named sibling temp file, is `fsync`ed, then `rename`d over
289/// `path`, so an interrupted write leaves the original untouched.
290///
291/// The name is unique per call, not a fixed `<path>.tmp`: two writers
292/// racing on a shared name can have one `rename` consume the other's
293/// staging file. [`FileLock`] already keeps two appenders apart; this is
294/// the second lock for a caller that reaches `write_ring` another way.
295fn write_ring(path: &Path, lines: &[String]) -> Result<(), BarkError> {
296 let parent = crate::atomic_file::parent_of(path);
297 let mut tmp = crate::atomic_file::create_staging_file(parent, "barks", ".tmp")?;
298
299 for line in lines {
300 tmp.write_all(line.as_bytes())?;
301 tmp.write_all(b"\n")?;
302 }
303 crate::atomic_file::publish(tmp, path).map_err(BarkError::Io)
304}
305
306#[cfg(test)]
307mod tests {
308 use super::*;
309
310 /// A representative fired alert. `at_ms` is a caller-chosen tag, not a
311 /// real timestamp: tests use it to tell records apart.
312 fn bark_for(subject: &str, at_ms: u64) -> Bark {
313 Bark {
314 at_ms,
315 rule: "watchdog".to_string(),
316 subject: subject.to_string(),
317 message: "restart budget exhausted".to_string(),
318 sinks: vec![SinkOutcome {
319 sink: "discord".to_string(),
320 error: None,
321 }],
322 }
323 }
324
325 /// The serialized length, plus its trailing newline, of one
326 /// `bark_for`-shaped line, computed here so it cannot happen to equal
327 /// the implementation's own byte count.
328 fn one_bark_len() -> u64 {
329 let line = serde_json::to_string(&bark_for("second", 1)).unwrap();
330 line.len() as u64 + 1
331 }
332
333 /// Cap set to force eviction on the third write.
334 #[test]
335 fn the_ring_drops_the_oldest_bark_to_stay_under_its_cap() {
336 let dir = tempfile::tempdir().unwrap();
337 let path = dir.path().join("barks.jsonl");
338 let cap = 2 * one_bark_len();
339
340 for (i, subject) in ["first", "second", "third"].iter().enumerate() {
341 append(&path, &bark_for(subject, i as u64), cap).unwrap();
342 }
343
344 let barks = read(&path).unwrap();
345 let subjects: Vec<&str> = barks.iter().map(|b| b.subject.as_str()).collect();
346 assert_eq!(subjects, ["second", "third"], "oldest out, newest kept");
347 assert!(
348 std::fs::metadata(&path).unwrap().len() <= cap,
349 "the cap is a cap"
350 );
351 }
352
353 /// Dropping one line too many leaves the ring under its cap too, so
354 /// this pins the survivors rather than the size alone.
355 #[test]
356 fn a_cap_crossed_by_many_lines_at_once_takes_the_prefix_and_no_more() {
357 let dir = tempfile::tempdir().unwrap();
358 let path = dir.path().join("barks.jsonl");
359 for i in 0..20 {
360 append(&path, &bark_for("web", i), DEFAULT_MAX_BYTES).unwrap();
361 }
362
363 // Room for the last four of the twenty-one lines the next append
364 // leaves behind, so seventeen have to go in one call.
365 let mut expected = read_lines(&path).unwrap();
366 expected.push(serde_json::to_string(&bark_for("web", 20)).unwrap());
367 let evicted = expected.len() - 4;
368 expected.drain(..evicted);
369 let cap = ring_bytes(&expected);
370
371 append(&path, &bark_for("web", 20), cap).unwrap();
372
373 let kept = read_lines(&path).unwrap();
374 assert_eq!(kept, expected, "the whole prefix goes, and nothing past it");
375 // Against the bytes on disk, not against `ring_bytes` again: the
376 // count the eviction trusts is the one under test here.
377 assert_eq!(
378 std::fs::metadata(&path).unwrap().len(),
379 cap,
380 "the byte accounting stays exact"
381 );
382 }
383
384 #[test]
385 fn a_bark_bigger_than_the_cap_is_written_anyway() {
386 let dir = tempfile::tempdir().unwrap();
387 let path = dir.path().join("barks.jsonl");
388 let huge = Bark {
389 message: "x".repeat(4096),
390 ..bark_for("web", 0)
391 };
392 append(&path, &huge, 64).unwrap();
393 assert_eq!(read(&path).unwrap().len(), 1);
394 }
395
396 /// The cap yields to the record rather than the record to the cap:
397 /// the ring empties ahead of it, and the alert still lands.
398 #[test]
399 fn an_oversized_bark_evicts_every_line_before_it() {
400 let dir = tempfile::tempdir().unwrap();
401 let path = dir.path().join("barks.jsonl");
402 append(&path, &bark_for("web", 0), DEFAULT_MAX_BYTES).unwrap();
403
404 let huge = Bark {
405 message: "x".repeat(4096),
406 ..bark_for("api", 1)
407 };
408 append(&path, &huge, 64).unwrap();
409
410 assert_eq!(
411 read(&path).unwrap(),
412 vec![huge],
413 "the one record too big to fit, alone"
414 );
415 }
416
417 #[test]
418 fn a_line_that_will_not_parse_costs_one_record_and_not_the_file() {
419 let dir = tempfile::tempdir().unwrap();
420 let path = dir.path().join("barks.jsonl");
421 append(&path, &bark_for("web", 1), DEFAULT_MAX_BYTES).unwrap();
422 std::fs::OpenOptions::new()
423 .append(true)
424 .open(&path)
425 .unwrap()
426 .write_all(b"{\"at_ms\": 2, \"rul\n")
427 .unwrap();
428 append(&path, &bark_for("api", 3), DEFAULT_MAX_BYTES).unwrap();
429
430 let barks = read(&path).unwrap();
431 assert_eq!(
432 barks.iter().map(|b| b.subject.as_str()).collect::<Vec<_>>(),
433 ["web", "api"]
434 );
435 }
436
437 #[test]
438 fn no_file_yet_is_no_barks_rather_than_a_failure() {
439 let dir = tempfile::tempdir().unwrap();
440 assert_eq!(read(&dir.path().join("nothing.jsonl")).unwrap(), vec![]);
441 }
442
443 /// A window that begins exactly at a record boundary holds that record
444 /// whole, so the scan that drops a partial record's tail must not eat
445 /// it.
446 #[test]
447 fn a_window_starting_on_a_boundary_keeps_its_first_record() {
448 let dir = tempfile::tempdir().unwrap();
449 let path = dir.path().join("barks.jsonl");
450 for i in 0..3u64 {
451 append(&path, &bark_for(&format!("s{i}"), i), DEFAULT_MAX_BYTES).unwrap();
452 }
453 let bytes = std::fs::read(&path).unwrap();
454 let len = bytes.len() as u64;
455 let first = bytes.iter().position(|&b| b == b'\n').unwrap() as u64;
456 let window = len - (first + 1);
457 let mut file = File::open(&path).unwrap();
458 let barks = read_window(&mut file, len, window).unwrap();
459 assert_eq!(
460 barks.iter().map(|b| b.subject.as_str()).collect::<Vec<_>>(),
461 ["s1", "s2"]
462 );
463 }
464
465 #[test]
466 fn read_last_on_a_missing_file_is_no_barks_rather_than_a_failure() {
467 let dir = tempfile::tempdir().unwrap();
468 assert_eq!(
469 read_last(&dir.path().join("nothing.jsonl"), 20).unwrap(),
470 vec![]
471 );
472 }
473
474 /// Zero asked for is zero returned, and the file is not even opened.
475 #[test]
476 fn read_last_of_zero_records_is_empty() {
477 let dir = tempfile::tempdir().unwrap();
478 let path = dir.path().join("barks.jsonl");
479 append(&path, &bark_for("web", 1), DEFAULT_MAX_BYTES).unwrap();
480 assert_eq!(read_last(&path, 0).unwrap(), vec![]);
481 }
482
483 /// The window reads from the end of the file, so a ring larger than
484 /// the window still answers with its newest records.
485 #[test]
486 fn read_last_returns_the_newest_records_oldest_first() {
487 let dir = tempfile::tempdir().unwrap();
488 let path = dir.path().join("barks.jsonl");
489 for i in 0..10u64 {
490 append(&path, &bark_for(&format!("s{i}"), i), DEFAULT_MAX_BYTES).unwrap();
491 }
492 let barks = read_last(&path, 3).unwrap();
493 assert_eq!(
494 barks.iter().map(|b| b.subject.as_str()).collect::<Vec<_>>(),
495 ["s7", "s8", "s9"]
496 );
497 }
498
499 /// Records large enough that the first window cannot hold the number
500 /// asked for: the window must grow, and the answer must still be the
501 /// last `limit` records — the oldest of them absent, not truncated.
502 #[test]
503 fn read_last_grows_the_window_until_it_holds_enough_records() {
504 let dir = tempfile::tempdir().unwrap();
505 let path = dir.path().join("barks.jsonl");
506 // About 20 KiB a record, so the 64 KiB first window holds three
507 // of the six — fewer than the five asked for below.
508 let big = |subject: &str, at_ms: u64| Bark {
509 message: "x".repeat(20 * 1024),
510 ..bark_for(subject, at_ms)
511 };
512 for (i, subject) in ["s0", "s1", "s2", "s3", "s4", "s5"].iter().enumerate() {
513 append(&path, &big(subject, i as u64), DEFAULT_MAX_BYTES).unwrap();
514 }
515 assert_eq!(read(&path).unwrap().len(), 6, "the fixture is six records");
516
517 let barks = read_last(&path, 5).unwrap();
518 assert_eq!(
519 barks.iter().map(|b| b.subject.as_str()).collect::<Vec<_>>(),
520 ["s1", "s2", "s3", "s4", "s5"]
521 );
522 }
523
524 /// `read_last` must answer what `read` plus a drain answers, for every
525 /// limit: the point of the window is what the reader pays, not what it
526 /// returns. The file carries the two kinds of line no parser takes — a
527 /// fragment a dead writer left, and a well-formed record from a shep
528 /// that grew a field — since those are what a real ring accumulates.
529 /// The records are padded past the first window, so the limits that
530 /// force a grow are in the range too.
531 #[test]
532 fn read_last_answers_what_a_full_read_would_for_every_limit() {
533 let dir = tempfile::tempdir().unwrap();
534 let path = dir.path().join("barks.jsonl");
535 let mut file = std::fs::File::create(&path).unwrap();
536 for i in 0..12u64 {
537 let bark = Bark {
538 message: "x".repeat(8 * 1024),
539 ..bark_for(&format!("s{i}"), i)
540 };
541 writeln!(file, "{}", serde_json::to_string(&bark).unwrap()).unwrap();
542 if i % 3 == 0 {
543 writeln!(file, "{{\"at_ms\": 99, \"subj").unwrap();
544 }
545 if i % 4 == 0 {
546 writeln!(file, "{{\"at_ms\": 99, \"future\": []}}").unwrap();
547 }
548 }
549 drop(file);
550
551 let everything = read(&path).unwrap();
552 assert_eq!(
553 everything.len(),
554 12,
555 "the fixture is twelve parseable records"
556 );
557 for limit in 0..everything.len() + 3 {
558 let keep_from = everything.len().saturating_sub(limit);
559 assert_eq!(
560 read_last(&path, limit).unwrap(),
561 &everything[keep_from..],
562 "limit {limit} must answer exactly what a full read plus a drain answers"
563 );
564 }
565 }
566
567 /// Env var naming the ring file the re-executed child should append
568 /// to. Its presence is also what tells the child it is a child.
569 #[cfg(any(unix, windows))]
570 const CHILD_PATH_VAR: &str = "SHEP_BARK_RACE_PATH";
571 /// Env var carrying the child's tag, which it stamps into every
572 /// record's `subject` so the parent can tell the two writers apart.
573 #[cfg(any(unix, windows))]
574 const CHILD_TAG_VAR: &str = "SHEP_BARK_RACE_TAG";
575 /// How many records each of the two writers appends. Large enough
576 /// that the two read-modify-rename sequences overlap many times over.
577 #[cfg(any(unix, windows))]
578 const RECORDS_PER_WRITER: u64 = 200;
579
580 /// Not a test: the child half of
581 /// [`two_writer_processes_do_not_lose_each_other_s_barks`], re-executed
582 /// as a separate OS process via `--ignored --exact`. Asserts nothing;
583 /// its job is to hammer [`append`] while the parent judges the result.
584 #[cfg(any(unix, windows))]
585 #[test]
586 #[ignore = "child process of two_writer_processes_do_not_lose_each_other_s_barks"]
587 fn bark_race_child() {
588 let Ok(path) = std::env::var(CHILD_PATH_VAR) else {
589 panic!("{CHILD_PATH_VAR} unset — this test is only run as a child process");
590 };
591 let tag = std::env::var(CHILD_TAG_VAR).expect("child needs a tag");
592 let path = std::path::PathBuf::from(path);
593
594 for i in 0..RECORDS_PER_WRITER {
595 append(&path, &bark_for(&tag, i), DEFAULT_MAX_BYTES).expect("child append");
596 }
597 }
598
599 /// Two OS processes, not threads: an in-process mutex would prove
600 /// nothing about a race that crosses address spaces via `rename`.
601 /// Covers Windows too: reverting `acquire`'s Windows arm to
602 /// `Ok(Self {})` reddens this rather than passing quietly. Without
603 /// the lock this can still pass on a lucky serial schedule.
604 #[cfg(any(unix, windows))]
605 #[test]
606 fn two_writer_processes_do_not_lose_each_other_s_barks() {
607 let dir = tempfile::tempdir().unwrap();
608 let path = dir.path().join("barks.jsonl");
609 let exe = std::env::current_exe().expect("test binary path");
610
611 let children: Vec<_> = ["alpha", "beta"]
612 .iter()
613 .map(|tag| {
614 std::process::Command::new(&exe)
615 .args(["--exact", "--ignored", "barks::tests::bark_race_child"])
616 .env(CHILD_PATH_VAR, &path)
617 .env(CHILD_TAG_VAR, tag)
618 // Piped, not inherited: a passing run should not
619 // interleave two child harnesses' output into this
620 // one's, and a failing child's harness output is
621 // exactly what the assertion below needs to show.
622 .stdout(std::process::Stdio::piped())
623 .spawn()
624 .expect("spawn writer")
625 })
626 .collect();
627
628 for child in children {
629 let out = child.wait_with_output().expect("wait for writer");
630 assert!(
631 out.status.success(),
632 "a writer process failed: {}\n{}",
633 out.status,
634 String::from_utf8_lossy(&out.stdout)
635 );
636 }
637
638 let barks = read(&path).unwrap();
639 for tag in ["alpha", "beta"] {
640 let mut seen: Vec<u64> = barks
641 .iter()
642 .filter(|b| b.subject == tag)
643 .map(|b| b.at_ms)
644 .collect();
645 seen.sort_unstable();
646 let expected: Vec<u64> = (0..RECORDS_PER_WRITER).collect();
647 assert_eq!(
648 seen, expected,
649 "{tag}'s records did not all survive the other writer"
650 );
651 }
652 assert_eq!(
653 barks.len() as u64,
654 2 * RECORDS_PER_WRITER,
655 "the ring holds records nobody wrote"
656 );
657 }
658
659 /// No field here is a credential today; the mode stays narrow so a
660 /// future one that is arrives already protected.
661 #[cfg(unix)]
662 #[test]
663 fn append_creates_the_ring_owner_only_on_unix() {
664 use std::os::unix::fs::PermissionsExt;
665 let dir = tempfile::tempdir().unwrap();
666 let path = dir.path().join("barks.jsonl");
667
668 append(&path, &bark_for("web", 0), DEFAULT_MAX_BYTES).unwrap();
669
670 let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
671 assert_eq!(
672 mode, 0o600,
673 "barks.jsonl is not the credential file, but stays narrow anyway"
674 );
675 }
676}