strop_trace/
lib.rs

1//! One opt-in diagnostic sink for all strop crates. Producers never perform file
2//! I/O or wait for the writer; an incomplete trace is always reported as such.
3//! Capture is bounded (total bytes, total events, per-record bytes) and every
4//! finished file ends with an explicit terminal `TraceEnd` marker, so a capped
5//! or failed capture can never be mistaken for a complete one. Forensic values
6//! over the per-record cap travel as ordered `replay_chunk` runs (schema 3)
7//! that the reader reassembles strictly — or refuses loudly.
8mod bounded;
9mod chunk;
10mod event;
11pub mod export;
12pub mod replay;
13mod writer;
14
15pub use event::{
16    preview, ContentPolicy, EventKind, Limits, TraceOptions, MAX_CAPTURE_BYTES, MAX_CAPTURE_EVENTS,
17    MAX_RECORD_BYTES, SCHEMA_VERSION, TERMINAL_RESERVE,
18};
19
20use parking_lot::Mutex;
21use serde::Serialize;
22use std::fs::OpenOptions;
23use std::path::{Path, PathBuf};
24use std::sync::atomic::{AtomicBool, Ordering};
25use std::sync::mpsc::{sync_channel, SyncSender};
26use std::sync::{Arc, LazyLock};
27use std::thread::JoinHandle;
28use std::time::Instant;
29
30/// A full queue is a visible capture failure, never silent loss; the writer
31/// drains faster than producers admit, so this only trips on writer stalls.
32const QUEUE_CAPACITY: usize = 64;
33static ACTIVE: LazyLock<Mutex<Option<Arc<Recorder>>>> = LazyLock::new(|| Mutex::new(None));
34static ENABLED: AtomicBool = AtomicBool::new(false);
35static CONTENT: AtomicBool = AtomicBool::new(false);
36
37#[derive(Debug, thiserror::Error)]
38pub enum TraceError {
39    #[error("a trace session is already active")]
40    AlreadyActive,
41    #[error("cannot create trace {path}: {source}")]
42    Open {
43        path: PathBuf,
44        source: std::io::Error,
45    },
46    #[error("cannot start trace writer: {0}")]
47    Spawn(std::io::Error),
48    #[error("incomplete trace: {0}")]
49    Incomplete(String),
50}
51
52#[derive(Default)]
53struct Failure {
54    message: Mutex<Option<String>>,
55    reported: AtomicBool,
56}
57impl Failure {
58    fn set(&self, message: impl FnOnce() -> String) {
59        let mut failure = self.message.lock();
60        if failure.is_none() {
61            *failure = Some(message());
62        }
63    }
64}
65
66struct Record {
67    kind: EventKind,
68    elapsed_us: u128,
69    fields: Vec<u8>,
70}
71struct Recorder {
72    sender: Mutex<Option<SyncSender<Record>>>,
73    failure: Arc<Failure>,
74    started: Instant,
75    max_record: usize,
76}
77
78/// Owning lifetime of a trace. Explicit finish reports errors; Drop still drains.
79pub struct TraceSession {
80    recorder: Arc<Recorder>,
81    worker: Option<JoinHandle<()>>,
82}
83
84pub fn start(path: &Path, options: TraceOptions) -> Result<TraceSession, TraceError> {
85    if !options.limits.valid() {
86        return Err(TraceError::Incomplete("invalid capture limits".into()));
87    }
88    let mut active = ACTIVE.lock();
89    if active.is_some() {
90        return Err(TraceError::AlreadyActive);
91    }
92    let mut open = OpenOptions::new();
93    open.write(true).create_new(true);
94    #[cfg(unix)]
95    {
96        use std::os::unix::fs::OpenOptionsExt;
97        open.mode(0o600);
98    }
99    let file = open.open(path).map_err(|source| TraceError::Open {
100        path: path.to_path_buf(),
101        source,
102    })?;
103    let (sender, receiver) = sync_channel(QUEUE_CAPACITY);
104    let failure = Arc::new(Failure::default());
105    let writer_failure = Arc::clone(&failure);
106    let limits = options.limits;
107    let worker = std::thread::Builder::new()
108        .name("strop-trace".into())
109        .spawn(move || writer::run(file, receiver, writer_failure, limits))
110        .map_err(TraceError::Spawn)?;
111    let recorder = Arc::new(Recorder {
112        sender: Mutex::new(Some(sender)),
113        failure,
114        started: Instant::now(),
115        max_record: limits.record_bytes,
116    });
117    *active = Some(Arc::clone(&recorder));
118    CONTENT.store(options.content == ContentPolicy::Full, Ordering::Release);
119    ENABLED.store(true, Ordering::Release);
120    Ok(TraceSession {
121        recorder,
122        worker: Some(worker),
123    })
124}
125
126#[inline]
127pub fn enabled() -> bool {
128    ENABLED.load(Ordering::Relaxed)
129}
130#[inline]
131pub fn capture_content() -> bool {
132    enabled() && CONTENT.load(Ordering::Relaxed)
133}
134
135/// Lazy producer: no payload construction or allocation when disabled.
136pub fn record_with<T: Serialize>(kind: EventKind, fields: impl FnOnce() -> T) {
137    if enabled() {
138        record(kind, &fields());
139    }
140}
141
142pub fn record<T: Serialize>(kind: EventKind, fields: &T) {
143    if !enabled() {
144        return;
145    }
146    let Some(recorder) = ACTIVE.lock().clone() else {
147        return;
148    };
149    // This lock protects queue admission only, never disk writes. Stamping
150    // under it keeps timestamps nondecreasing in the writer's receive order,
151    // and serializing under it bounds simultaneous trace encodings.
152    let mut sender = recorder.sender.lock();
153    if sender.is_none() {
154        return;
155    }
156    if recorder.failure.message.lock().is_some() {
157        sender.take();
158        return;
159    }
160    let mut bytes = bounded::Bytes::new(recorder.max_record);
161    if serde_json::to_writer(&mut bytes, fields).is_ok() {
162        let record = Record {
163            kind,
164            elapsed_us: recorder.started.elapsed().as_micros(),
165            fields: bytes.into_vec(),
166        };
167        if sender
168            .as_ref()
169            .expect("checked sender")
170            .try_send(record)
171            .is_err()
172        {
173            recorder
174                .failure
175                .set(|| "capture queue full or writer unavailable".into());
176            sender.take();
177        }
178        return;
179    }
180    // Only the forensic substream chunks: replay completeness is contractual
181    // there, and the Full content policy is what may carry payloads at all.
182    // Every other oversize record keeps the honest refusal below.
183    if kind == EventKind::Replay && CONTENT.load(Ordering::Relaxed) {
184        if let Some(chunks) = chunk::serialize(kind, fields, recorder.max_record) {
185            let mut admitted = true;
186            for fields in chunks {
187                let record = Record {
188                    kind: EventKind::ReplayChunk,
189                    elapsed_us: recorder.started.elapsed().as_micros(),
190                    fields,
191                };
192                if sender
193                    .as_ref()
194                    .expect("checked sender")
195                    .try_send(record)
196                    .is_err()
197                {
198                    admitted = false;
199                    break;
200                }
201            }
202            if admitted {
203                return;
204            }
205            recorder
206                .failure
207                .set(|| "capture queue full or writer unavailable".into());
208            sender.take();
209            return;
210        }
211    }
212    recorder
213        .failure
214        .set(|| "record exceeds cap or cannot serialize".into());
215    sender.take();
216}
217
218/// End the capture visibly when honest continuation is impossible (a value
219/// beyond even the assembled-value bound, a writer failure): no further
220/// records are admitted and the terminal marker reports the capture
221/// incomplete instead of silently shrinking.
222pub fn mark_incomplete(message: &'static str) {
223    let Some(recorder) = ACTIVE.lock().clone() else {
224        return;
225    };
226    recorder.failure.set(|| message.into());
227    recorder.sender.lock().take();
228    CONTENT.store(false, Ordering::Release);
229}
230
231/// Report once to the editor's status line; finish still returns the failure.
232pub fn take_failure() -> Option<String> {
233    let recorder = ACTIVE.lock().clone()?;
234    let message = recorder.failure.message.lock().clone()?;
235    (!recorder.failure.reported.swap(true, Ordering::Relaxed)).then_some(message)
236}
237
238impl TraceSession {
239    pub fn finish(mut self) -> Result<(), TraceError> {
240        self.close()
241    }
242
243    fn close(&mut self) -> Result<(), TraceError> {
244        let Some(worker) = self.worker.take() else {
245            return Ok(());
246        };
247        {
248            let mut active = ACTIVE.lock();
249            if active
250                .as_ref()
251                .is_some_and(|value| Arc::ptr_eq(value, &self.recorder))
252            {
253                ENABLED.store(false, Ordering::Release);
254                CONTENT.store(false, Ordering::Release);
255                *active = None;
256            }
257        }
258        self.recorder.sender.lock().take();
259        if worker.join().is_err() {
260            self.recorder
261                .failure
262                .set(|| "writer thread panicked".into());
263        }
264        match self.recorder.failure.message.lock().clone() {
265            Some(error) => Err(TraceError::Incomplete(error)),
266            None => Ok(()),
267        }
268    }
269}
270impl Drop for TraceSession {
271    fn drop(&mut self) {
272        // Explicit finish is the reporting boundary. Drop guarantees durability
273        // during unwinding without risking a second panic or corrupting the TUI.
274        let _ = self.close();
275    }
276}
277
278#[cfg(test)]
279mod tests;
strop_trace/lib.rs

strop_trace/
lib.rs
