saddle_observability/
diagnostic.rs

1//! Independent emergency output. Submission never performs filesystem IO.
2use crate::FileLoggingConfig;
3use saddle_core::{CallContext, Diagnostic};
4use serde::Serialize;
5use std::{
6    fmt,
7    fs::{self, OpenOptions},
8    io::{self, Write},
9    path::PathBuf,
10    sync::{
11        Arc,
12        atomic::{AtomicBool, AtomicU64, Ordering},
13        mpsc::{self, SyncSender, TrySendError},
14    },
15    thread::{self, JoinHandle},
16    time::Duration,
17};
18
19pub const EMERGENCY_FILE_NAME: &str = "saddle.emergency.log";
20const QUEUE_CAPACITY: usize = 64;
21const RECORD_BYTES: usize = 32 * 1024;
22static ACTIVE: AtomicBool = AtomicBool::new(false);
23
24#[derive(Clone, Copy, Debug, Eq, PartialEq)]
25pub enum DiagnosticSubmission {
26    Enqueued,
27    Full,
28    Closed,
29    EncodingFailed,
30}
31#[derive(Clone, Copy, Debug, Eq, PartialEq)]
32pub enum DiagnosticShutdown {
33    Pending,
34    Finished,
35    WorkerPanicked,
36}
37#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
38pub struct DiagnosticOutputSnapshot {
39    pub initialized: bool,
40    pub first_failure: Option<DiagnosticIoFailure>,
41    pub enqueued: u64,
42    pub written: u64,
43    pub dropped: u64,
44    pub output_failed: u64,
45    pub stderr_failed: u64,
46    pub stderr_suppressed: u64,
47    pub truncated: u64,
48    pub closed: bool,
49}
50#[derive(Default)]
51struct Status {
52    initialized: AtomicBool,
53    failure: AtomicU64,
54    enqueued: AtomicU64,
55    written: AtomicU64,
56    dropped: AtomicU64,
57    output_failed: AtomicU64,
58    stderr_failed: AtomicU64,
59    stderr_suppressed: AtomicU64,
60    truncated: AtomicU64,
61    closed: AtomicBool,
62    submitting: AtomicU64,
63}
64impl Status {
65    fn fail(&self, stage: u64, error: &io::Error) {
66        let kind = match error.kind() {
67            io::ErrorKind::PermissionDenied => 1,
68            io::ErrorKind::NotFound => 2,
69            io::ErrorKind::NotADirectory => 3,
70            io::ErrorKind::InvalidInput => 4,
71            io::ErrorKind::StorageFull => 5,
72            io::ErrorKind::BrokenPipe => 6,
73            io::ErrorKind::WouldBlock => 7,
74            _ => 8,
75        };
76        let os = error.raw_os_error();
77        let packed = (stage << 56)
78            | (kind << 48)
79            | (u64::from(os.is_some()) << 40)
80            | u64::from(os.unwrap_or(0) as u32);
81        let _ = self
82            .failure
83            .compare_exchange(0, packed, Ordering::AcqRel, Ordering::Acquire);
84    }
85    fn snapshot(&self) -> DiagnosticOutputSnapshot {
86        let failure = self.failure.load(Ordering::Acquire);
87        DiagnosticOutputSnapshot {
88            initialized: self.initialized.load(Ordering::Acquire),
89            first_failure: (failure != 0).then(|| DiagnosticIoFailure {
90                stage: if failure >> 56 == 1 {
91                    "initialize"
92                } else {
93                    "write"
94                },
95                kind: [
96                    "none",
97                    "permission_denied",
98                    "not_found",
99                    "not_a_directory",
100                    "invalid_input",
101                    "storage_full",
102                    "broken_pipe",
103                    "would_block",
104                    "other",
105                ][((failure >> 48) & 255) as usize],
106                os_code: (failure & (1 << 40) != 0).then_some(failure as u32 as i32),
107            }),
108            enqueued: self.enqueued.load(Ordering::Acquire),
109            written: self.written.load(Ordering::Acquire),
110            dropped: self.dropped.load(Ordering::Acquire),
111            output_failed: self.output_failed.load(Ordering::Acquire),
112            stderr_failed: self.stderr_failed.load(Ordering::Acquire),
113            stderr_suppressed: self.stderr_suppressed.load(Ordering::Acquire),
114            truncated: self.truncated.load(Ordering::Acquire),
115            closed: self.closed.load(Ordering::Acquire),
116        }
117    }
118}
119
120#[derive(Clone, Copy, Debug, Eq, PartialEq)]
121pub struct DiagnosticIoFailure {
122    pub stage: &'static str,
123    pub kind: &'static str,
124    pub os_code: Option<i32>,
125}
126
127/// Startup owner: at most one independent writer per process. No join on Drop.
128pub struct EmergencyDiagnostics {
129    handle: EmergencyDiagnosticHandle,
130    worker: Option<JoinHandle<()>>,
131    target: PathBuf,
132}
133#[derive(Clone)]
134pub struct EmergencyDiagnosticHandle {
135    sender: SyncSender<Vec<u8>>,
136    status: Arc<Status>,
137}
138
139#[derive(Debug)]
140pub enum EmergencyInitError {
141    InvalidDirectory,
142    AlreadyActive,
143    Spawn(io::ErrorKind),
144}
145impl fmt::Display for EmergencyInitError {
146    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
147        write!(f, "emergency diagnostic initialization failed: {self:?}")
148    }
149}
150impl std::error::Error for EmergencyInitError {}
151
152struct LimitedBytes(Vec<u8>);
153impl Write for LimitedBytes {
154    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
155        if self.0.len().saturating_add(bytes.len()) > RECORD_BYTES - 1 {
156            return Err(io::Error::new(
157                io::ErrorKind::InvalidData,
158                "diagnostic encoding bound",
159            ));
160        }
161        self.0.extend_from_slice(bytes);
162        Ok(bytes.len())
163    }
164    fn flush(&mut self) -> io::Result<()> {
165        Ok(())
166    }
167}
168#[derive(Serialize)]
169struct Envelope<'a> {
170    event: &'static str,
171    timestamp_unix_ms: u128,
172    diagnostic: &'a Diagnostic,
173    trace_id: Option<&'a str>,
174    rpc_id: Option<String>,
175    request: Option<&'a str>,
176    route: Option<&'a str>,
177}
178
179impl<'a> Envelope<'a> {
180    fn new(
181        diagnostic: &'a Diagnostic,
182        context: Option<(&'a CallContext, &'a crate::EventContext)>,
183    ) -> Self {
184        Self {
185            event: "framework.diagnostic",
186            timestamp_unix_ms: std::time::SystemTime::now()
187                .duration_since(std::time::UNIX_EPOCH)
188                .unwrap_or_default()
189                .as_millis(),
190            diagnostic,
191            trace_id: context.map(|(call, _)| call.trace_correlation_id().as_str()),
192            rpc_id: context.map(|(call, _)| call.span_id().to_string()),
193            request: context.map(|(_, event)| event.diagnostic_request()),
194            route: context.map(|(_, event)| event.diagnostic_route()),
195        }
196    }
197}
198
199impl EmergencyDiagnosticHandle {
200    pub fn snapshot(&self) -> DiagnosticOutputSnapshot {
201        self.status.snapshot()
202    }
203    /// Enqueued means accepted in memory, NOT written or durable.
204    pub fn submit(&self, diagnostic: &Diagnostic) -> DiagnosticSubmission {
205        self.submit_context(diagnostic, None)
206    }
207    pub fn submit_context(
208        &self,
209        diagnostic: &Diagnostic,
210        context: Option<(&CallContext, &crate::EventContext)>,
211    ) -> DiagnosticSubmission {
212        let active = self.status.submitting.fetch_add(1, Ordering::AcqRel);
213        struct Exit<'a>(&'a AtomicU64);
214        impl Drop for Exit<'_> {
215            fn drop(&mut self) {
216                self.0.fetch_sub(1, Ordering::AcqRel);
217            }
218        }
219        let _exit = Exit(&self.status.submitting);
220        if active >= QUEUE_CAPACITY as u64 {
221            self.status.dropped.fetch_add(1, Ordering::Relaxed);
222            return DiagnosticSubmission::Full;
223        }
224        if self.status.closed.load(Ordering::Acquire) {
225            self.status.dropped.fetch_add(1, Ordering::Relaxed);
226            return DiagnosticSubmission::Closed;
227        }
228        let mut bytes = LimitedBytes(Vec::with_capacity(RECORD_BYTES));
229        if serde_json::to_writer(&mut bytes, &Envelope::new(diagnostic, context)).is_err() {
230            bytes.0.clear();
231            let mut reduced = match serde_json::to_value(Envelope::new(diagnostic, context)) {
232                Ok(value) => value,
233                Err(_) => {
234                    self.status.dropped.fetch_add(1, Ordering::Relaxed);
235                    return DiagnosticSubmission::EncodingFailed;
236                }
237            };
238            reduced["diagnostic"]["stack"] = serde_json::Value::String(String::new());
239            reduced["diagnostic"]["stack_truncated"] = true.into();
240            reduced["encoding_truncated"] = true.into();
241            self.status.truncated.fetch_add(1, Ordering::Relaxed);
242            if serde_json::to_writer(&mut bytes, &reduced).is_err() {
243                self.status.dropped.fetch_add(1, Ordering::Relaxed);
244                return DiagnosticSubmission::EncodingFailed;
245            }
246        }
247        bytes.0.push(b'\n');
248        match self.sender.try_send(bytes.0) {
249            Ok(()) => {
250                self.status.enqueued.fetch_add(1, Ordering::Relaxed);
251                DiagnosticSubmission::Enqueued
252            }
253            Err(TrySendError::Full(_)) => {
254                self.status.dropped.fetch_add(1, Ordering::Relaxed);
255                DiagnosticSubmission::Full
256            }
257            Err(TrySendError::Disconnected(_)) => {
258                self.status.dropped.fetch_add(1, Ordering::Relaxed);
259                DiagnosticSubmission::Closed
260            }
261        }
262    }
263}
264
265impl crate::Observer {
266    /// Mirrors one safe diagnostic into independent emergency output and the
267    /// existing logger. The result acknowledges only emergency queue admission.
268    pub fn record_diagnostic(
269        &self,
270        diagnostic: &Diagnostic,
271        emergency: &EmergencyDiagnosticHandle,
272        context: Option<(&CallContext, &crate::EventContext)>,
273    ) -> DiagnosticSubmission {
274        let result = emergency.submit_context(diagnostic, context);
275        if let Ok(serde_json::Value::Object(fields)) =
276            serde_json::to_value(Envelope::new(diagnostic, context))
277        {
278            let level = if matches!(
279                diagnostic.category(),
280                saddle_core::DiagnosticCategory::ExpectedRejection
281            ) {
282                crate::EventLevel::Info
283            } else {
284                crate::EventLevel::Error
285            };
286            let mut record = crate::logger::LogRecord::new(level, "framework.diagnostic");
287            record.data = fields;
288            record.data.remove("event");
289            record.data.remove("timestamp_unix_ms");
290            self.emit(record);
291        }
292        result
293    }
294}
295impl EmergencyDiagnostics {
296    /// Directory setup is performed by the worker, never the submitting request.
297    /// Callers must check snapshot/written, not equate successful start with IO readiness.
298    pub fn start(config: &FileLoggingConfig) -> Result<Self, EmergencyInitError> {
299        let directory = config
300            .directory()
301            .to_str()
302            .ok_or(EmergencyInitError::InvalidDirectory)?;
303        if directory.is_empty()
304            || directory.len() > 4096
305            || directory.chars().any(char::is_control)
306            || directory.contains("://")
307            || directory.contains(['@', '?', '#'])
308        {
309            return Err(EmergencyInitError::InvalidDirectory);
310        }
311        if ACTIVE
312            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
313            .is_err()
314        {
315            return Err(EmergencyInitError::AlreadyActive);
316        }
317        let target = config.directory().join(EMERGENCY_FILE_NAME);
318        let path = target.clone();
319        let (sender, receiver) = mpsc::sync_channel::<Vec<u8>>(QUEUE_CAPACITY);
320        let status = Arc::new(Status::default());
321        let worker_status = status.clone();
322        let worker = thread::Builder::new()
323            .name("saddle-emergency-writer".into())
324            .spawn(move || {
325                struct ActiveGuard;
326                impl Drop for ActiveGuard {
327                    fn drop(&mut self) {
328                        ACTIVE.store(false, Ordering::Release);
329                    }
330                }
331                let _active = ActiveGuard;
332                let file = (|| {
333                    if let Some(parent) = path.parent() {
334                        fs::create_dir_all(parent)?;
335                    }
336                    // Never open a FIFO/device or follow the final symlink.
337                    use std::os::unix::fs::OpenOptionsExt;
338                    let file = OpenOptions::new()
339                        .create(true)
340                        .append(true)
341                        .mode(0o600)
342                        .custom_flags(
343                            rustix::fs::OFlags::NOFOLLOW.bits() as i32
344                                | rustix::fs::OFlags::NONBLOCK.bits() as i32,
345                        )
346                        .open(&path)?;
347                    if !file.metadata()?.is_file() {
348                        return Err(io::Error::new(
349                            io::ErrorKind::InvalidInput,
350                            "non-regular emergency file",
351                        ));
352                    }
353                    rustix::fs::flock(&file, rustix::fs::FlockOperation::NonBlockingLockExclusive)
354                        .map_err(io::Error::from)?;
355                    Ok(file)
356                })();
357                if let Err(error) = &file {
358                    worker_status.fail(1, error);
359                }
360                worker_status.initialized.store(true, Ordering::Release);
361                let mut output = file;
362                let mut stderr_attempted = false;
363                loop {
364                    match receiver.recv_timeout(Duration::from_millis(10)) {
365                        Ok(bytes) => deliver(
366                            &mut output,
367                            &mut io::stderr(),
368                            &path,
369                            &bytes,
370                            &worker_status,
371                            &mut stderr_attempted,
372                        ),
373                        Err(mpsc::RecvTimeoutError::Timeout) => {
374                            if worker_status.closed.load(Ordering::Acquire)
375                                && worker_status.submitting.load(Ordering::Acquire) == 0
376                            {
377                                break;
378                            }
379                        }
380                        Err(mpsc::RecvTimeoutError::Disconnected) => break,
381                    }
382                }
383            })
384            .map_err(|error| {
385                ACTIVE.store(false, Ordering::Release);
386                EmergencyInitError::Spawn(error.kind())
387            })?;
388        Ok(Self {
389            handle: EmergencyDiagnosticHandle { sender, status },
390            worker: Some(worker),
391            target,
392        })
393    }
394    pub fn handle(&self) -> EmergencyDiagnosticHandle {
395        self.handle.clone()
396    }
397    pub fn target(&self) -> &std::path::Path {
398        &self.target
399    }
400    pub fn snapshot(&self) -> DiagnosticOutputSnapshot {
401        self.handle.snapshot()
402    }
403    /// Call using the EXISTING lifecycle deadline; Pending is not a new budget.
404    pub fn shutdown(&mut self) -> DiagnosticShutdown {
405        self.handle.status.closed.store(true, Ordering::Release);
406        if self
407            .worker
408            .as_ref()
409            .is_some_and(|worker| !worker.is_finished())
410        {
411            return DiagnosticShutdown::Pending;
412        }
413        match self.worker.take().map(JoinHandle::join) {
414            Some(Err(_)) => DiagnosticShutdown::WorkerPanicked,
415            _ => DiagnosticShutdown::Finished,
416        }
417    }
418}
419impl Drop for EmergencyDiagnostics {
420    fn drop(&mut self) {
421        self.handle.status.closed.store(true, Ordering::Release);
422    }
423}
424
425fn deliver<W: Write, E: Write>(
426    output: &mut io::Result<W>,
427    stderr: &mut E,
428    target: &std::path::Path,
429    bytes: &[u8],
430    status: &Status,
431    stderr_attempted: &mut bool,
432) {
433    let result = match output {
434        Ok(writer) => writer.write_all(bytes),
435        Err(error) => Err(error
436            .raw_os_error()
437            .map(io::Error::from_raw_os_error)
438            .unwrap_or_else(|| io::Error::from(error.kind()))),
439    };
440    if let Err(error) = result {
441        status.fail(2, &error);
442        status.output_failed.fetch_add(1, Ordering::Relaxed);
443        // Only the independent worker can block here. No recursive logger call.
444        if !*stderr_attempted {
445            *stderr_attempted = true;
446            let failure = serde_json::json!({"event":"framework.diagnostic.output_failed", "target":target.to_string_lossy(), "io_kind":format!("{:?}", error.kind()), "os_code":error.raw_os_error()});
447            if serde_json::to_writer(&mut *stderr, &failure)
448                .and_then(|_| stderr.write_all(b"\n").map_err(serde_json::Error::io))
449                .is_err()
450                || stderr.write_all(bytes).is_err()
451            {
452                status.stderr_failed.fetch_add(1, Ordering::Relaxed);
453            }
454        } else {
455            status.stderr_suppressed.fetch_add(1, Ordering::Relaxed);
456        }
457    } else {
458        status.written.fetch_add(1, Ordering::Release);
459    }
460}
461
462#[cfg(test)]
463mod tests {
464    use super::*;
465    use saddle_core::{
466        CaptureSite, DiagnosticCategory, DiagnosticCause, DiagnosticCode, DiagnosticStage,
467    };
468    use std::time::Instant;
469    static FILE_TEST: std::sync::Mutex<()> = std::sync::Mutex::new(());
470    fn sample() -> Diagnostic {
471        Diagnostic::capture(
472            DiagnosticCategory::UnexpectedError,
473            CaptureSite::FirstObserved,
474            DiagnosticCause::new(
475                DiagnosticStage::RequestDb,
476                DiagnosticCode::new("db.connection_failed").unwrap(),
477            )
478            .with_io(&io::Error::other("DO_NOT_LOG_SECRET"))
479            .with_input_location(
480                saddle_core::DiagnosticInputLocation::new(Some(17), Some(3))
481                    .with_file(
482                        saddle_core::DiagnosticLocator::from_projection(
483                            "mapping.json",
484                            false,
485                            false,
486                        )
487                        .unwrap(),
488                    )
489                    .with_column(
490                        saddle_core::DiagnosticLocator::from_projection(
491                            "DO_NOT_LOG_SECRET",
492                            true,
493                            true,
494                        )
495                        .unwrap(),
496                    ),
497            ),
498        )
499    }
500    struct Failed;
501    impl Write for Failed {
502        fn write(&mut self, _: &[u8]) -> io::Result<usize> {
503            Err(io::Error::from_raw_os_error(28))
504        }
505        fn flush(&mut self) -> io::Result<()> {
506            Ok(())
507        }
508    }
509    #[test]
510    fn diagnostic_independent_file_after_main_failure() {
511        let _serial = FILE_TEST.lock().unwrap();
512        let directory = std::env::temp_dir().join(format!(
513            "saddle-emergency-{}-{}",
514            std::process::id(),
515            sample().id()
516        ));
517        let config = FileLoggingConfig::new(&directory, crate::Rotation::Daily);
518        let mut emergency = EmergencyDiagnostics::start(&config).unwrap();
519        struct MarkFailure(Arc<AtomicU64>);
520        impl Write for MarkFailure {
521            fn write(&mut self, _: &[u8]) -> io::Result<usize> {
522                self.0.fetch_add(1, Ordering::Release);
523                Err(io::Error::from_raw_os_error(28))
524            }
525            fn flush(&mut self) -> io::Result<()> {
526                Ok(())
527            }
528        }
529        let main_failed = Arc::new(AtomicU64::new(0));
530        let observer = crate::Observer::with_writer(
531            crate::ObserverConfig::default(),
532            MarkFailure(main_failed.clone()),
533        )
534        .unwrap();
535        let diagnostic = sample();
536        let (call, _) = observer
537            .start_external_call_checked("test", "test", "test", "operation", Some("safe-trace"))
538            .unwrap();
539        let event = crate::EventContext::new(
540            crate::RequestIdentity::new("request-1").unwrap(),
541            crate::RouteIdentity::new("route.test").unwrap(),
542            1,
543        )
544        .unwrap();
545        let end = Instant::now() + Duration::from_secs(3);
546        while main_failed.load(Ordering::Acquire) == 0 && Instant::now() < end {
547            thread::sleep(Duration::from_millis(1));
548        }
549        assert!(
550            main_failed.load(Ordering::Acquire) > 0,
551            "primary writer failed before diagnostic submission"
552        );
553        assert_eq!(
554            observer.record_diagnostic(
555                &diagnostic,
556                &emergency.handle(),
557                Some((call.context(), &event))
558            ),
559            DiagnosticSubmission::Enqueued
560        );
561        let end = Instant::now() + Duration::from_secs(3);
562        while emergency.snapshot().written != 1 && Instant::now() < end {
563            thread::sleep(Duration::from_millis(1));
564        }
565        assert_eq!(emergency.snapshot().written, 1);
566        let text = fs::read_to_string(emergency.target()).unwrap();
567        let row: serde_json::Value = serde_json::from_str(text.trim()).unwrap();
568        assert_eq!(row["trace_id"], "safe-trace");
569        assert_eq!(row["request"], "request-1");
570        assert_eq!(row["diagnostic"]["diagnostic_id"], diagnostic.id());
571        let input = &row["diagnostic"]["causes"][0]["input_location"];
572        assert_eq!(input["json_line"], 17);
573        assert_eq!(input["json_column"], 3);
574        assert_eq!(input["file"]["value"], "mapping.json");
575        assert_eq!(input["column"]["value"], "[redacted]");
576        assert_eq!(input["locator_truncated"], true);
577        assert_eq!(input["locator_redacted"], true);
578        assert!(!text.contains("DO_NOT_LOG_SECRET"));
579        let end = Instant::now() + Duration::from_secs(3);
580        while emergency.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
581            thread::sleep(Duration::from_millis(1));
582        }
583        assert_eq!(emergency.shutdown(), DiagnosticShutdown::Finished);
584        assert_eq!(
585            emergency.handle().submit(&diagnostic),
586            DiagnosticSubmission::Closed
587        );
588        call.succeed();
589        let _shutdown = observer.shutdown();
590        fs::remove_file(emergency.target()).unwrap();
591        fs::remove_dir(directory).unwrap();
592    }
593    #[test]
594    fn diagnostic_directory_failure_has_real_os_cause() {
595        let _serial = FILE_TEST.lock().unwrap();
596        let path = std::env::temp_dir().join(format!(
597            "saddle-emergency-file-{}-{}",
598            std::process::id(),
599            sample().id()
600        ));
601        fs::write(&path, b"not a directory").unwrap();
602        let mut owner =
603            EmergencyDiagnostics::start(&FileLoggingConfig::new(&path, crate::Rotation::Daily))
604                .unwrap();
605        let end = Instant::now() + Duration::from_secs(3);
606        while !owner.snapshot().initialized && Instant::now() < end {
607            thread::sleep(Duration::from_millis(1));
608        }
609        let failure = owner.snapshot().first_failure.unwrap();
610        assert_eq!(failure.stage, "initialize");
611        assert!(failure.os_code.is_some());
612        assert_eq!(owner.target(), path.join(EMERGENCY_FILE_NAME));
613        while owner.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
614            thread::sleep(Duration::from_millis(1));
615        }
616        assert_eq!(owner.shutdown(), DiagnosticShutdown::Finished);
617        fs::remove_file(path).unwrap();
618    }
619    #[test]
620    fn diagnostic_stuck_worker_shutdown_never_joins_early() {
621        let (sender, _receiver) = mpsc::sync_channel(1);
622        let (release, wait) = mpsc::channel();
623        let (started, reached) = mpsc::channel();
624        let worker = thread::spawn(move || {
625            started.send(()).unwrap();
626            wait.recv().unwrap();
627        });
628        reached.recv_timeout(Duration::from_secs(2)).unwrap();
629        let mut owner = EmergencyDiagnostics {
630            handle: EmergencyDiagnosticHandle {
631                sender,
632                status: Arc::new(Status::default()),
633            },
634            worker: Some(worker),
635            target: PathBuf::from("unused"),
636        };
637        assert_eq!(owner.shutdown(), DiagnosticShutdown::Pending);
638        release.send(()).unwrap();
639        let end = Instant::now() + Duration::from_secs(3);
640        while owner.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
641            thread::yield_now();
642        }
643        assert_eq!(owner.shutdown(), DiagnosticShutdown::Finished);
644    }
645    #[test]
646    fn diagnostic_double_failure_is_counted_without_recursion() {
647        let status = Status::default();
648        let mut output: io::Result<Failed> = Err(io::Error::from_raw_os_error(13));
649        let mut stderr = Vec::new();
650        let mut attempted = false;
651        deliver(
652            &mut output,
653            &mut stderr,
654            std::path::Path::new("/safe/logs/saddle.emergency.log"),
655            b"{\"original\":true}\n",
656            &status,
657            &mut attempted,
658        );
659        let text = String::from_utf8(stderr).unwrap();
660        assert!(text.contains("\"os_code\":13") && text.contains("\"original\":true"));
661        attempted = false;
662        deliver(
663            &mut output,
664            &mut Failed,
665            std::path::Path::new("./logs"),
666            b"{}\n",
667            &status,
668            &mut attempted,
669        );
670        for _ in 0..10 {
671            deliver(
672                &mut output,
673                &mut Failed,
674                std::path::Path::new("./logs"),
675                b"{}\n",
676                &status,
677                &mut attempted,
678            );
679        }
680        assert_eq!(status.snapshot().stderr_failed, 1);
681        assert_eq!(status.snapshot().stderr_suppressed, 10);
682        assert_eq!(status.snapshot().output_failed, 12);
683    }
684    #[test]
685    fn diagnostic_full_queue_submission_does_not_wait() {
686        let (sender, _receiver) = mpsc::sync_channel(1);
687        let status = Arc::new(Status::default());
688        let handle = EmergencyDiagnosticHandle { sender, status };
689        let sample = sample();
690        assert_eq!(handle.submit(&sample), DiagnosticSubmission::Enqueued);
691        assert_eq!(handle.submit(&sample), DiagnosticSubmission::Full);
692        assert_eq!(handle.snapshot().dropped, 1);
693        assert_eq!(handle.snapshot().written, 0);
694    }
695}
saddle_observability/diagnostic.rs

saddle_observability/
diagnostic.rs
