Skip to main content

max / synckit

Install the recording subscriber once, globally, instead of per capture `with_default` is thread-local but the callsite interest cache it feeds is process-wide, and it is rebuilt every time any dispatcher is registered or dropped. Under a harness running dozens of threads, another test dropping its dispatcher can reset a callsite to "never" between this one registering and emitting it, and the event is lost with no sign that anything happened. Measured: green on fw13's 8 cores, red on astra's 96, where it failed the unmutated baseline of a cargo-mutants run. Install once and leave it there, so the cached interest is stable. The sink is what moves, and it records only the thread that installed it, so a capture still sees exactly its own events and nobody else's.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session
https://claude.ai/code/session_01MptwXZ8k65v19rFmdGAyki
Author: Max Johnson <me@maxj.phd> · 2026-08-31 23:00 UTC
Signed with PGP, not checked
Commit: 5f8f2533557ddab8d52b04a1644b382669510352
Parent: 2c6c316
1 file changed, +65 insertions, -25 deletions
M synckit-client/src/test_support.rs +65 -25
@@ -55,18 +55,40 @@
55 55 }
56 56 }
57 57
58 - /// Records every event emitted while it is the thread's default subscriber.
59 - struct Recorder(std::sync::Arc<std::sync::Mutex<Vec<CapturedEvent>>>);
58 + /// The buffer one [`events_from`] call is filling, and the thread it belongs to.
59 + struct Sink {
60 + thread: std::thread::ThreadId,
61 + events: Vec<CapturedEvent>,
62 + }
63 +
64 + /// Where the installed subscriber puts what it sees. `None` between captures.
65 + static SINK: std::sync::Mutex<Option<Sink>> = std::sync::Mutex::new(None);
66 +
67 + /// One capture at a time, since there is one sink.
68 + static CAPTURING: std::sync::Mutex<()> = std::sync::Mutex::new(());
69 +
70 + /// The process-wide subscriber, installed once and never removed.
71 + ///
72 + /// Global rather than thread-scoped, which is the opposite of what it looks
73 + /// like it should be. `tracing::subscriber::with_default` is thread-local, but
74 + /// the interest cache it feeds is process-wide and is rebuilt whenever any
75 + /// dispatcher is registered or dropped. Under a test harness running dozens of
76 + /// threads, another test dropping its dispatcher can reset a callsite to
77 + /// "never" between this one registering and emitting, and the event is silently
78 + /// lost. Measured: passes on an 8-core box, fails on a 96-core one.
79 + ///
80 + /// Installing once and leaving it there makes the cached interest stable. The
81 + /// sink is what moves, and it records only the thread that installed it, so a
82 + /// capture still sees exactly its own events.
83 + struct Recorder;
60 84
61 85 impl tracing::Subscriber for Recorder {
62 86 fn register_callsite(
63 87 &self,
64 88 _: &'static tracing::Metadata<'static>,
65 89 ) -> tracing::subscriber::Interest {
66 - // `sometimes`, not the default `always`/`never` verdict: interest is
67 - // cached per callsite for the life of the process, so a verdict recorded
68 - // here would outlive the guard and decide the answer for every later
69 - // caller in this test binary.
90 + // `sometimes` so `enabled` is consulted per event rather than the
91 + // verdict being cached; the sink comes and goes.
70 92 tracing::subscriber::Interest::sometimes()
71 93 }
72 94
@@ -83,15 +105,19 @@
83 105 fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
84 106
85 107 fn event(&self, event: &tracing::Event<'_>) {
108 + let mut sink = SINK
109 + .lock()
110 + .unwrap_or_else(std::sync::PoisonError::into_inner);
111 + let Some(sink) = sink.as_mut() else { return };
112 + if sink.thread != std::thread::current().id() {
113 + return;
114 + }
86 115 let mut visitor = EventVisitor::default();
87 116 event.record(&mut visitor);
88 - self.0
89 - .lock()
90 - .expect("the recorder mutex")
91 - .push(CapturedEvent {
92 - message: visitor.message,
93 - fields: visitor.fields,
94 - });
117 + sink.events.push(CapturedEvent {
118 + message: visitor.message,
119 + fields: visitor.fields,
120 + });
95 121 }
96 122
97 123 fn enter(&self, _: &tracing::span::Id) {}
@@ -99,18 +125,32 @@
99 125 fn exit(&self, _: &tracing::span::Id) {}
100 126 }
101 127
102 - /// Every `tracing` event `f` emitted, in order.
128 + /// Every `tracing` event `f` emitted on this thread, in order.
103 129 ///
104 - /// The subscriber is the thread's default for the duration of `f` only, so
105 - /// nothing global is installed and tests stay independent of each other. What
106 - /// it is for: a branch whose only effect is a log line, which no assertion on a
107 - /// return value can reach.
130 + /// What it is for: a branch whose only effect is a log line, which no assertion
131 + /// on a return value can reach.
108 132 pub(crate) fn events_from(f: impl FnOnce()) -> Vec<CapturedEvent> {
109 - let events = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
110 - let recorder = Recorder(std::sync::Arc::clone(&events));
111 - tracing::subscriber::with_default(recorder, f);
112 - std::sync::Arc::into_inner(events)
113 - .expect("the recorder is dropped with the subscriber")
114 - .into_inner()
115 - .expect("the recorder mutex")
133 + static INSTALL: std::sync::Once = std::sync::Once::new();
134 + INSTALL.call_once(|| {
135 + // Err would mean something else claimed the global default; nothing in
136 + // this crate's tests does, and a capture that recorded nothing would
137 + // fail its own assertions rather than pass quietly.
138 + let _ = tracing::subscriber::set_global_default(Recorder);
139 + });
140 +
141 + let _capturing = CAPTURING
142 + .lock()
143 + .unwrap_or_else(std::sync::PoisonError::into_inner);
144 + *SINK
145 + .lock()
146 + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Sink {
147 + thread: std::thread::current().id(),
148 + events: Vec::new(),
149 + });
150 + f();
151 + SINK.lock()
152 + .unwrap_or_else(std::sync::PoisonError::into_inner)
153 + .take()
154 + .expect("the sink this call installed")
155 + .events
116 156 }