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_driver_details(saddle_core::DiagnosticDriverDetails::new(
480                Some(0),
481                Some(4),
482                saddle_core::DiagnosticTypeName::from_metadata("core::option::Option<i64>"),
483                saddle_core::DiagnosticTypeName::from_metadata("BIGINT"),
484            ))
485            .with_input_location(
486                saddle_core::DiagnosticInputLocation::new(Some(17), Some(3))
487                    .with_file(
488                        saddle_core::DiagnosticLocator::from_projection(
489                            "mapping.json",
490                            false,
491                            false,
492                        )
493                        .unwrap(),
494                    )
495                    .with_column(
496                        saddle_core::DiagnosticLocator::from_projection(
497                            "DO_NOT_LOG_SECRET",
498                            true,
499                            true,
500                        )
501                        .unwrap(),
502                    ),
503            ),
504        )
505    }
506    struct Failed;
507    impl Write for Failed {
508        fn write(&mut self, _: &[u8]) -> io::Result<usize> {
509            Err(io::Error::from_raw_os_error(28))
510        }
511        fn flush(&mut self) -> io::Result<()> {
512            Ok(())
513        }
514    }
515    #[test]
516    fn diagnostic_independent_file_after_main_failure() {
517        let _serial = FILE_TEST.lock().unwrap();
518        let directory = std::env::temp_dir().join(format!(
519            "saddle-emergency-{}-{}",
520            std::process::id(),
521            sample().id()
522        ));
523        let config = FileLoggingConfig::new(&directory, crate::Rotation::Daily);
524        let mut emergency = EmergencyDiagnostics::start(&config).unwrap();
525        struct MarkFailure(Arc<AtomicU64>);
526        impl Write for MarkFailure {
527            fn write(&mut self, _: &[u8]) -> io::Result<usize> {
528                self.0.fetch_add(1, Ordering::Release);
529                Err(io::Error::from_raw_os_error(28))
530            }
531            fn flush(&mut self) -> io::Result<()> {
532                Ok(())
533            }
534        }
535        let main_failed = Arc::new(AtomicU64::new(0));
536        let observer = crate::Observer::with_writer(
537            crate::ObserverConfig::default(),
538            MarkFailure(main_failed.clone()),
539        )
540        .unwrap();
541        let diagnostic = sample();
542        let (call, _) = observer
543            .start_external_call_checked("test", "test", "test", "operation", Some("safe-trace"))
544            .unwrap();
545        let event = crate::EventContext::new(
546            crate::RequestIdentity::new("request-1").unwrap(),
547            crate::RouteIdentity::new("route.test").unwrap(),
548            1,
549        )
550        .unwrap();
551        let end = Instant::now() + Duration::from_secs(3);
552        while main_failed.load(Ordering::Acquire) == 0 && Instant::now() < end {
553            thread::sleep(Duration::from_millis(1));
554        }
555        assert!(
556            main_failed.load(Ordering::Acquire) > 0,
557            "primary writer failed before diagnostic submission"
558        );
559        assert_eq!(
560            observer.record_diagnostic(
561                &diagnostic,
562                &emergency.handle(),
563                Some((call.context(), &event))
564            ),
565            DiagnosticSubmission::Enqueued
566        );
567        let end = Instant::now() + Duration::from_secs(3);
568        while emergency.snapshot().written != 1 && Instant::now() < end {
569            thread::sleep(Duration::from_millis(1));
570        }
571        assert_eq!(emergency.snapshot().written, 1);
572        let text = fs::read_to_string(emergency.target()).unwrap();
573        let row: serde_json::Value = serde_json::from_str(text.trim()).unwrap();
574        assert_eq!(row["trace_id"], "safe-trace");
575        assert_eq!(row["request"], "request-1");
576        assert_eq!(row["diagnostic"]["diagnostic_id"], diagnostic.id());
577        let driver = &row["diagnostic"]["causes"][0]["driver_details"];
578        assert_eq!(driver["column_index"], 0);
579        assert_eq!(driver["column_count"], 4);
580        assert_eq!(
581            driver["target_rust_type"]["value"],
582            "core::option::Option<i64>"
583        );
584        assert_eq!(driver["actual_db_type"]["value"], "BIGINT");
585        let input = &row["diagnostic"]["causes"][0]["input_location"];
586        assert_eq!(input["json_line"], 17);
587        assert_eq!(input["json_column"], 3);
588        assert_eq!(input["file"]["value"], "mapping.json");
589        assert_eq!(input["column"]["value"], "[redacted]");
590        assert_eq!(input["locator_truncated"], true);
591        assert_eq!(input["locator_redacted"], true);
592        assert!(!text.contains("DO_NOT_LOG_SECRET"));
593        let end = Instant::now() + Duration::from_secs(3);
594        while emergency.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
595            thread::sleep(Duration::from_millis(1));
596        }
597        assert_eq!(emergency.shutdown(), DiagnosticShutdown::Finished);
598        assert_eq!(
599            emergency.handle().submit(&diagnostic),
600            DiagnosticSubmission::Closed
601        );
602        call.succeed();
603        let _shutdown = observer.shutdown();
604        fs::remove_file(emergency.target()).unwrap();
605        fs::remove_dir(directory).unwrap();
606    }
607    #[test]
608    fn diagnostic_directory_failure_has_real_os_cause() {
609        let _serial = FILE_TEST.lock().unwrap();
610        let path = std::env::temp_dir().join(format!(
611            "saddle-emergency-file-{}-{}",
612            std::process::id(),
613            sample().id()
614        ));
615        fs::write(&path, b"not a directory").unwrap();
616        let mut owner =
617            EmergencyDiagnostics::start(&FileLoggingConfig::new(&path, crate::Rotation::Daily))
618                .unwrap();
619        let end = Instant::now() + Duration::from_secs(3);
620        while !owner.snapshot().initialized && Instant::now() < end {
621            thread::sleep(Duration::from_millis(1));
622        }
623        let failure = owner.snapshot().first_failure.unwrap();
624        assert_eq!(failure.stage, "initialize");
625        assert!(failure.os_code.is_some());
626        assert_eq!(owner.target(), path.join(EMERGENCY_FILE_NAME));
627        while owner.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
628            thread::sleep(Duration::from_millis(1));
629        }
630        assert_eq!(owner.shutdown(), DiagnosticShutdown::Finished);
631        fs::remove_file(path).unwrap();
632    }
633    #[test]
634    fn diagnostic_stuck_worker_shutdown_never_joins_early() {
635        let (sender, _receiver) = mpsc::sync_channel(1);
636        let (release, wait) = mpsc::channel();
637        let (started, reached) = mpsc::channel();
638        let worker = thread::spawn(move || {
639            started.send(()).unwrap();
640            wait.recv().unwrap();
641        });
642        reached.recv_timeout(Duration::from_secs(2)).unwrap();
643        let mut owner = EmergencyDiagnostics {
644            handle: EmergencyDiagnosticHandle {
645                sender,
646                status: Arc::new(Status::default()),
647            },
648            worker: Some(worker),
649            target: PathBuf::from("unused"),
650        };
651        assert_eq!(owner.shutdown(), DiagnosticShutdown::Pending);
652        release.send(()).unwrap();
653        let end = Instant::now() + Duration::from_secs(3);
654        while owner.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
655            thread::yield_now();
656        }
657        assert_eq!(owner.shutdown(), DiagnosticShutdown::Finished);
658    }
659    #[test]
660    fn diagnostic_double_failure_is_counted_without_recursion() {
661        let status = Status::default();
662        let mut output: io::Result<Failed> = Err(io::Error::from_raw_os_error(13));
663        let mut stderr = Vec::new();
664        let mut attempted = false;
665        deliver(
666            &mut output,
667            &mut stderr,
668            std::path::Path::new("/safe/logs/saddle.emergency.log"),
669            b"{\"original\":true}\n",
670            &status,
671            &mut attempted,
672        );
673        let text = String::from_utf8(stderr).unwrap();
674        assert!(text.contains("\"os_code\":13") && text.contains("\"original\":true"));
675        attempted = false;
676        deliver(
677            &mut output,
678            &mut Failed,
679            std::path::Path::new("./logs"),
680            b"{}\n",
681            &status,
682            &mut attempted,
683        );
684        for _ in 0..10 {
685            deliver(
686                &mut output,
687                &mut Failed,
688                std::path::Path::new("./logs"),
689                b"{}\n",
690                &status,
691                &mut attempted,
692            );
693        }
694        assert_eq!(status.snapshot().stderr_failed, 1);
695        assert_eq!(status.snapshot().stderr_suppressed, 10);
696        assert_eq!(status.snapshot().output_failed, 12);
697    }
698    #[test]
699    fn diagnostic_full_queue_submission_does_not_wait() {
700        let (sender, _receiver) = mpsc::sync_channel(1);
701        let status = Arc::new(Status::default());
702        let handle = EmergencyDiagnosticHandle { sender, status };
703        let sample = sample();
704        assert_eq!(handle.submit(&sample), DiagnosticSubmission::Enqueued);
705        assert_eq!(handle.submit(&sample), DiagnosticSubmission::Full);
706        assert_eq!(handle.snapshot().dropped, 1);
707        assert_eq!(handle.snapshot().written, 0);
708    }
709}
saddle_observability/diagnostic.rs

saddle_observability/
diagnostic.rs
