camel_integration_test/log_capture.rs
1//! Process-global log capture for the scenario tier's document-level
2//! `logs:` assertions (rc-tdgh5).
3//!
4//! The composition root installs a global tracing subscriber at boot,
5//! unconditionally, first-wins with warn-and-skip on loss
6//! (`CamelConfig::configure_context_with_beans`; boot is
7//! caller-owned). A scenario driver that wants `logs:` assertions
8//! therefore claims the process's subscriber seat BEFORE the boot:
9//! [`ensure_capture_subscriber`] installs a registry carrying
10//! `CaptureLayer` through the same first-wins `try_init`.
11//!
12//! When the harness owns the seat, every event flows through
13//! `CaptureLayer`, which appends each event to every open capture
14//! window whose `[opened_at, now)` interval contains the event
15//! timestamp — conservative attribution: a window opened after an
16//! event never sees it. Windows are process-global: [`WindowHandle`]
17//! registers a buffer in a static registry at open and unregisters at
18//! close (RAII on drop); the runner owns open/evaluate/close around
19//! the document run. Each buffer is capped at [`LOG_WINDOW_CAP`](crate::log_capture::LOG_WINDOW_CAP)
20//! events, drop-oldest with a head marker naming the truncation.
21
22use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
23use std::sync::{Arc, Mutex, MutexGuard};
24use std::time::Instant;
25
26use tracing::Level;
27use tracing_subscriber::Layer;
28use tracing_subscriber::layer::Context;
29use tracing_subscriber::prelude::*;
30
31/// One captured tracing event: when it fired (monotonic, for window
32/// attribution), at what level, from which target, with which rendered
33/// message.
34#[derive(Debug, Clone)]
35pub struct LogEvent {
36 /// When the event reached the capture layer (monotonic clock).
37 pub at: Instant,
38 /// The event's level.
39 pub level: Level,
40 /// The event's target — `module_path!` of the emit site, for the
41 /// camel-log component's rendered exchanges:
42 /// `camel_component_log`.
43 pub target: String,
44 /// The rendered `message` field (empty when the event carries
45 /// none). camel-log's exchange lines arrive as composites, for
46 /// example `[marker] Body: <body>`.
47 pub message: String,
48}
49
50/// Events kept per window before the oldest drop: drop-oldest with a
51/// head marker naming the truncation (the marker sits at TRACE level,
52/// so it can never trip a `noLevelAbove` clause).
53pub const LOG_WINDOW_CAP: usize = 10_000;
54
55/// Target identifying the in-band truncation marker event.
56const MARKER_TARGET: &str = "camel_integration_test::log_capture";
57
58/// One registered window: identity, open timestamp (the attribution
59/// boundary), and the shared buffer.
60struct WindowEntry {
61 id: u64,
62 opened_at: Instant,
63 buffer: Arc<Mutex<Vec<LogEvent>>>,
64}
65
66/// Process-global registry of open windows. Attribution takes this
67/// registry lock, then the target buffer's.
68static WINDOWS: Mutex<Vec<WindowEntry>> = Mutex::new(Vec::new());
69/// Monotonic window identity.
70static NEXT_ID: AtomicU64 = AtomicU64::new(0);
71/// Whether the harness's capture subscriber owns the process's tracing
72/// seat (it won the first-wins `try_init`).
73static OWN_INSTALL: AtomicBool = AtomicBool::new(false);
74
75/// An open capture window: identity plus the shared buffer.
76pub struct WindowHandle {
77 id: u64,
78 buffer: Arc<Mutex<Vec<LogEvent>>>,
79}
80
81impl WindowHandle {
82 /// Closes the window: unregisters it and returns the captured
83 /// events in arrival order. Unregistering is idempotent with drop.
84 pub fn close(self) -> Vec<LogEvent> {
85 unregister(self.id);
86 let mut buffer = lock(&self.buffer);
87 std::mem::take(&mut *buffer)
88 }
89}
90
91impl Drop for WindowHandle {
92 fn drop(&mut self) {
93 unregister(self.id);
94 }
95}
96
97/// Opens a capture window and registers it in the process-global
98/// registry; the window's `[opened_at, now)` interval starts at the
99/// registration instant.
100pub fn open_window() -> WindowHandle {
101 let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
102 let buffer = Arc::new(Mutex::new(Vec::new()));
103 let mut windows = lock(&WINDOWS);
104 windows.push(WindowEntry {
105 id,
106 opened_at: Instant::now(),
107 buffer: Arc::clone(&buffer),
108 });
109 WindowHandle { id, buffer }
110}
111
112/// Removes a window from the registry (close or drop); a window no
113/// longer in the registry captures nothing.
114fn unregister(id: u64) {
115 let mut windows = lock(&WINDOWS);
116 windows.retain(|entry| entry.id != id);
117}
118
119/// Poison-tolerant lock: a panic inside one capture path must not take
120/// down every later window.
121fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
122 mutex
123 .lock()
124 .unwrap_or_else(|poisoned| poisoned.into_inner())
125}
126
127/// Whether the harness's capture subscriber owns the process's tracing
128/// seat. `false` means a foreign subscriber won the first-wins race
129/// (or the harness never installed): events bypass the capture layer
130/// and `logs:` assertions cannot run — the runner fails such
131/// documents through `ScenarioFailure::LogCaptureUnavailable`.
132pub fn capture_installed() -> bool {
133 OWN_INSTALL.load(Ordering::Acquire)
134}
135
136/// Claims the process's tracing seat for capture, before any boot can
137/// install its own subscriber. Idempotent once the harness owns the
138/// seat; losing the first-wins `try_init` leaves the foreign
139/// subscriber in place and [`capture_installed`] at `false`.
140///
141/// The subscriber composes the capture layer WITH an `fmt` layer, so
142/// events keep reaching stdout after capture takes the seat. Honest
143/// tradeoff: this passthrough is UNFILTERED — every level prints —
144/// unlike the composition root's config-driven `EnvFilter`, because
145/// the scenario tier must not re-read ambient config (ADR-0069
146/// hermeticity); v1 accepts the verbosity delta.
147pub fn ensure_capture_subscriber() {
148 if OWN_INSTALL.load(Ordering::Acquire) {
149 return;
150 }
151 let capture = tracing_subscriber::registry()
152 .with(CaptureLayer)
153 .with(tracing_subscriber::fmt::layer());
154 if capture.try_init().is_ok() {
155 OWN_INSTALL.store(true, Ordering::Release);
156 }
157}
158
159/// A scoped dispatch carrying the capture layer: exercises `on_event`
160/// and window attribution without touching the process-global
161/// subscriber seat, so tests stay deterministic whatever the binary's
162/// test ordering installed globally (the same escape hatch these unit
163/// tests use).
164#[cfg(test)]
165pub(crate) fn scoped_capture_dispatch() -> tracing::Dispatch {
166 tracing::Dispatch::new(tracing_subscriber::registry().with(CaptureLayer))
167}
168
169/// The capture layer: appends every event to every open window whose
170/// `[opened_at, now)` interval contains the event timestamp.
171struct CaptureLayer;
172
173impl<S> Layer<S> for CaptureLayer
174where
175 S: tracing::Subscriber,
176{
177 fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
178 let at = Instant::now();
179 let mut visitor = MessageVisitor::default();
180 event.record(&mut visitor);
181 let log_event = LogEvent {
182 at,
183 level: *event.metadata().level(),
184 target: event.metadata().target().to_string(),
185 message: visitor.message.unwrap_or_default(),
186 };
187 let windows = lock(&WINDOWS);
188 for window in windows.iter() {
189 // Conservative attribution: only a window already open at
190 // the event timestamp sees the event.
191 if window.opened_at > at {
192 continue;
193 }
194 let mut buffer = lock(&window.buffer);
195 // Cap: drop-oldest until the marker and the event both
196 // fit. The head drain removes any earlier marker, so one
197 // marker stays at the head.
198 if buffer.len() + 2 > LOG_WINDOW_CAP {
199 let drop_count = buffer.len() + 2 - LOG_WINDOW_CAP;
200 buffer.drain(..drop_count);
201 buffer.insert(
202 0,
203 LogEvent {
204 at,
205 level: Level::TRACE,
206 target: MARKER_TARGET.to_string(),
207 message: format!(
208 "window cap {LOG_WINDOW_CAP} reached: earlier events dropped"
209 ),
210 },
211 );
212 }
213 buffer.push(log_event.clone());
214 }
215 }
216}
217
218/// Extracts the rendered `message` field. The `info!("{msg}")`
219/// convention records through `record_debug`, and `format_args!`'s
220/// `Debug` renders the formatted text verbatim — no quoting.
221#[derive(Default)]
222struct MessageVisitor {
223 message: Option<String>,
224}
225
226impl tracing::field::Visit for MessageVisitor {
227 fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
228 if field.name() == "message" {
229 self.message = Some(format!("{value:?}"));
230 }
231 }
232}
233
234#[cfg(test)]
235mod tests {
236 use super::*;
237
238 /// A scoped dispatch carrying the capture layer: exercises
239 /// `on_event` without touching the process-global subscriber, so
240 /// these tests stay deterministic whatever the binary's test
241 /// ordering installed globally.
242 fn scoped_capture() -> tracing::Dispatch {
243 tracing::Dispatch::new(tracing_subscriber::registry().with(CaptureLayer))
244 }
245
246 #[test]
247 fn concurrent_windows_attribute_conservatively() {
248 let dispatch = scoped_capture();
249 let first = open_window();
250 let second = open_window();
251 // One event through the layer: both already-open windows
252 // contain it.
253 tracing::dispatcher::with_default(&dispatch, || tracing::info!("shared marker"));
254 let late = open_window();
255 tracing::dispatcher::with_default(&dispatch, || tracing::info!("later marker"));
256 let first_events = first.close();
257 let late_events = late.close();
258 let second_events = second.close();
259 assert!(
260 first_events
261 .iter()
262 .any(|e| e.message.contains("shared marker")),
263 "first window (open at event time) must capture: {first_events:?}"
264 );
265 assert!(
266 second_events
267 .iter()
268 .any(|e| e.message.contains("shared marker")),
269 "second window (open at event time) must capture: {second_events:?}"
270 );
271 assert!(
272 first_events
273 .iter()
274 .any(|e| e.message.contains("later marker"))
275 );
276 assert!(
277 second_events
278 .iter()
279 .any(|e| e.message.contains("later marker"))
280 );
281 assert!(
282 !late_events
283 .iter()
284 .any(|e| e.message.contains("shared marker")),
285 "conservative attribution: a window opened after the event never sees it: {late_events:?}"
286 );
287 }
288
289 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
290 async fn spawned_task_events_counted() {
291 let dispatch = scoped_capture();
292 let window = open_window();
293 // The spawned task carries the scoped dispatch onto its own
294 // worker thread: capture follows the layer, not the spawning
295 // thread.
296 let task = tokio::spawn(async move {
297 tracing::dispatcher::with_default(&dispatch, || tracing::warn!("spawned task marker"));
298 });
299 task.await.expect("spawned task completes");
300 let events = window.close();
301 assert!(
302 events
303 .iter()
304 .any(|e| e.level == Level::WARN && e.message.contains("spawned task marker")),
305 "the spawned task's warn must land in the open window: {events:?}"
306 );
307 }
308}