Fix two tests that flaked under the parallel harness
testlog: capture through a single global subscriber that routes each event to a thread-local sink and returns Interest::sometimes, so tracing's global per-callsite interest cache can never short-circuit a live capture when another thread rebuilds it. The old scoped with_default lost events at random in the full suite (an empty capture failed the log-accounting assertions ~1 run in 15). labels: capture the escaping test through testlog instead of installing its own process-wide subscriber, which hard-cached every foreign callsite "off" and poisoned the captures above. dirimage: name scratch directories from a process-monotonic counter rather than SystemTime, which resolves to only a microsecond — two of the parallel tests sharing a tag drew the same value, collided on one directory, and the first to finish removed it out from under the other's reads.
This commit is contained in:
Matthew Jackson
+44
-9
@@ -17,18 +17,31 @@ struct Scratch(PathBuf);
|
||||
|
||||
impl Scratch {
|
||||
fn new(tag: &str) -> Self {
|
||||
let mut p = std::env::temp_dir();
|
||||
p.push(format!(
|
||||
"freemkv-dirimage-{tag}-{}-{:?}",
|
||||
std::process::id(),
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_nanos()
|
||||
));
|
||||
let p = Self::unique_path(tag);
|
||||
std::fs::create_dir_all(&p).unwrap();
|
||||
Self(p)
|
||||
}
|
||||
|
||||
/// The path a scratch dir takes — kept separate from directory creation so
|
||||
/// its uniqueness is testable without a syscall between draws. A monotonic
|
||||
/// counter, NOT a timestamp, is what guarantees it: `SystemTime::now()`
|
||||
/// resolves to only a MICROSECOND on macOS, so two of the seven parallel
|
||||
/// tests that share the "bdmv" tag routinely read the same value within one
|
||||
/// microsecond, collide on the same directory, and the first to finish
|
||||
/// `remove_dir_all`s it out from under the others' reads — an intermittent
|
||||
/// `read_sectors` ENOENT. The counter is unique regardless of clock
|
||||
/// granularity; pairing it with the pid keeps it unique across test-binary
|
||||
/// processes.
|
||||
fn unique_path(tag: &str) -> PathBuf {
|
||||
static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
|
||||
let uniq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
let mut p = std::env::temp_dir();
|
||||
p.push(format!(
|
||||
"freemkv-dirimage-{tag}-{}-{uniq}",
|
||||
std::process::id()
|
||||
));
|
||||
p
|
||||
}
|
||||
fn path(&self) -> &Path {
|
||||
&self.0
|
||||
}
|
||||
@@ -69,6 +82,28 @@ fn bdmv_scratch() -> (Scratch, Vec<u8>, Vec<u8>) {
|
||||
(s, index, clip)
|
||||
}
|
||||
|
||||
/// Every `Scratch` must own a DISTINCT directory. Seven tests share the "bdmv"
|
||||
/// tag and run in parallel; if two land the same path, the first to drop
|
||||
/// `remove_dir_all`s it out from under the other's reads, which surfaced as an
|
||||
/// intermittent `read_sectors` failure. Regression: with the old
|
||||
/// `SystemTime::now()` name this is RED, because macOS's clock advances only
|
||||
/// per microsecond, so a tight loop of pure draws — no `create_dir_all` syscall
|
||||
/// between them to nudge the clock, mirroring the real cross-thread collision —
|
||||
/// hands back the same value many times over. The counter-based name is unique
|
||||
/// regardless of clock granularity.
|
||||
#[test]
|
||||
fn scratch_paths_are_unique_even_at_clock_resolution() {
|
||||
let paths: Vec<PathBuf> = (0..1000).map(|_| Scratch::unique_path("uniq")).collect();
|
||||
let distinct: std::collections::HashSet<&PathBuf> = paths.iter().collect();
|
||||
assert_eq!(
|
||||
distinct.len(),
|
||||
paths.len(),
|
||||
"every Scratch must own a distinct path; a timestamp-based name collides \
|
||||
under macOS's microsecond clock and lets one parallel test delete \
|
||||
another's files mid-read"
|
||||
);
|
||||
}
|
||||
|
||||
// ── The de-risking spike ────────────────────────────────────────────────────
|
||||
|
||||
/// THE load-bearing assertion of the whole design: metadata synthesized here
|
||||
|
||||
+18
-80
@@ -1917,81 +1917,10 @@ mod apply_tests {
|
||||
}
|
||||
|
||||
/// Sentinel embedded in the crafted playlist name below. The capture keeps
|
||||
/// ONLY fields whose rendered form contains it, so installing this
|
||||
/// subscriber process-wide costs nothing and cannot accumulate other
|
||||
/// tests' log output.
|
||||
/// ONLY fields whose rendered form contains it, so the assertion cannot be
|
||||
/// fooled by another test's log output.
|
||||
const LOG_INJECTION_SENTINEL: &str = "FMKV-LOG-INJECTION-PROBE";
|
||||
|
||||
fn capture_sink() -> &'static std::sync::Mutex<Vec<(String, String)>> {
|
||||
static SINK: std::sync::OnceLock<std::sync::Mutex<Vec<(String, String)>>> =
|
||||
std::sync::OnceLock::new();
|
||||
SINK.get_or_init(|| std::sync::Mutex::new(Vec::new()))
|
||||
}
|
||||
|
||||
/// Records how a `tracing` field was RENDERED — the question a disc-derived
|
||||
/// log field raises is not whether it is logged but how.
|
||||
///
|
||||
/// This is installed as the process-wide default rather than scoped with
|
||||
/// `with_default`, because `tracing` caches an `Interest` per callsite
|
||||
/// GLOBALLY: a sibling test running the same code on another thread with no
|
||||
/// subscriber caches the callsite as "never", and a thread-local subscriber
|
||||
/// installed afterwards then receives nothing. That failure mode is silent
|
||||
/// — an empty capture reads as "no raw bytes found" — so the test asserts
|
||||
/// the capture is non-empty as well.
|
||||
///
|
||||
/// `register_callsite` answers `never` for every callsite outside this
|
||||
/// module, so the rest of the suite keeps its current no-op logging cost.
|
||||
struct CapturedFields;
|
||||
|
||||
struct FieldVisitor(Vec<(String, String)>);
|
||||
|
||||
impl tracing::field::Visit for FieldVisitor {
|
||||
fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
|
||||
self.0
|
||||
.push((field.name().to_string(), format!("{value:?}")));
|
||||
}
|
||||
fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
|
||||
self.0.push((field.name().to_string(), value.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
fn is_labels_event(meta: &tracing::Metadata<'_>) -> bool {
|
||||
meta.is_event() && meta.target().starts_with("libfreemkv::labels")
|
||||
}
|
||||
|
||||
impl tracing::Subscriber for CapturedFields {
|
||||
fn register_callsite(
|
||||
&self,
|
||||
meta: &'static tracing::Metadata<'static>,
|
||||
) -> tracing::subscriber::Interest {
|
||||
if is_labels_event(meta) {
|
||||
tracing::subscriber::Interest::always()
|
||||
} else {
|
||||
tracing::subscriber::Interest::never()
|
||||
}
|
||||
}
|
||||
fn enabled(&self, meta: &tracing::Metadata<'_>) -> bool {
|
||||
is_labels_event(meta)
|
||||
}
|
||||
fn event(&self, event: &tracing::Event<'_>) {
|
||||
let mut v = FieldVisitor(Vec::new());
|
||||
event.record(&mut v);
|
||||
if v.0
|
||||
.iter()
|
||||
.any(|(_, val)| val.contains(LOG_INJECTION_SENTINEL))
|
||||
{
|
||||
capture_sink().lock().unwrap().extend(v.0);
|
||||
}
|
||||
}
|
||||
fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
|
||||
tracing::span::Id::from_u64(1)
|
||||
}
|
||||
fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
|
||||
fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
|
||||
fn enter(&self, _: &tracing::span::Id) {}
|
||||
fn exit(&self, _: &tracing::span::Id) {}
|
||||
}
|
||||
|
||||
/// A playlist name is a raw UDF directory entry — disc-controlled bytes,
|
||||
/// validated no further than a lossy UTF-8 decode. Logging it through
|
||||
/// tracing's `%` (Display) sigil writes those bytes VERBATIM, so a crafted
|
||||
@@ -2008,7 +1937,6 @@ mod apply_tests {
|
||||
fn a_disc_derived_playlist_name_is_escaped_in_the_log_not_written_verbatim() {
|
||||
// A name whose bytes would clear the line and repaint it.
|
||||
let evil = format!("\u{1b}[2K\u{1b}[31m{LOG_INJECTION_SENTINEL}\u{7}\u{1b}[0m.mpls");
|
||||
let _ = tracing::subscriber::set_global_default(CapturedFields);
|
||||
|
||||
let labels = vec![
|
||||
sub_label(1, "eng", LabelQualifier::None),
|
||||
@@ -2024,19 +1952,29 @@ mod apply_tests {
|
||||
subtitle(0x12A2, "fra"),
|
||||
],
|
||||
)];
|
||||
// Capture through the crate's ONE serialised sink. A process-wide
|
||||
// `set_global_default` here would poison every other test's callsite
|
||||
// interest cache for the rest of the binary — `tracing` caches interest
|
||||
// GLOBALLY, and a global subscriber that answers `never` for foreign
|
||||
// callsites hard-disables them, so `testlog::capture`'s scoped captures
|
||||
// (e.g. the `freemkv::disc` log-accounting tests) then see nothing and
|
||||
// flake. `testlog::capture` serialises every capture under one lock and
|
||||
// installs no global default, which is the invariant those tests rely on.
|
||||
let ((), events) = crate::testlog::capture(|| {
|
||||
apply_labels(&labels, &mut titles);
|
||||
|
||||
let fields = capture_sink().lock().unwrap().clone();
|
||||
let playlist: Vec<&(String, String)> = fields
|
||||
});
|
||||
let playlist: Vec<&str> = events
|
||||
.iter()
|
||||
.filter(|(k, v)| k == "playlist" && v.contains(LOG_INJECTION_SENTINEL))
|
||||
.filter(|e| e.target.starts_with("libfreemkv::labels"))
|
||||
.filter_map(|e| e.field("playlist"))
|
||||
.filter(|v| v.contains(LOG_INJECTION_SENTINEL))
|
||||
.collect();
|
||||
assert!(
|
||||
!playlist.is_empty(),
|
||||
"the anchoring event must actually have fired, or this test proves \
|
||||
nothing; captured: {fields:?}"
|
||||
nothing; captured: {events:?}"
|
||||
);
|
||||
for (_, rendered) in playlist {
|
||||
for rendered in playlist {
|
||||
assert!(
|
||||
!rendered.contains('\u{1b}') && !rendered.contains('\u{7}'),
|
||||
"a disc-controlled playlist name reached the log with its raw \
|
||||
|
||||
+65
-33
@@ -85,11 +85,37 @@ impl tracing::field::Visit for Visitor {
|
||||
}
|
||||
}
|
||||
|
||||
struct Capture(Arc<Mutex<Vec<CapturedEvent>>>);
|
||||
type Sink = Arc<Mutex<Vec<CapturedEvent>>>;
|
||||
|
||||
thread_local! {
|
||||
/// The sink for a capture ACTIVE ON THIS THREAD, if any. Thread-local so
|
||||
/// concurrent captures across `cargo test`'s parallel harness never see
|
||||
/// each other's events, and so a non-capturing thread simply has `None`.
|
||||
static SINK: std::cell::RefCell<Option<Sink>> = const { std::cell::RefCell::new(None) };
|
||||
}
|
||||
|
||||
/// The ONE process-wide subscriber. Installed once and left installed; it is
|
||||
/// offered every event and records into whichever thread's sink is active,
|
||||
/// dropping the event when none is.
|
||||
struct Capture;
|
||||
|
||||
impl tracing::Subscriber for Capture {
|
||||
// `sometimes`, deliberately NOT the default `always`/`never`. A cacheable
|
||||
// interest lets `tracing`'s GLOBAL per-callsite cache short-circuit a
|
||||
// callsite to "off", and under `cargo test`'s parallel harness a cache
|
||||
// rebuild triggered by ANY other thread can leave it there while a capture
|
||||
// is live — the capture then observes NOTHING. That is the race that made
|
||||
// `parse_playlist_unreadable_clip_icb_yields_no_title` flake ~1 run in 15.
|
||||
// `sometimes` forces `enabled` to be consulted on the emitting thread for
|
||||
// every event, so a capture always sees its own.
|
||||
fn register_callsite(
|
||||
&self,
|
||||
_meta: &'static tracing::Metadata<'static>,
|
||||
) -> tracing::subscriber::Interest {
|
||||
tracing::subscriber::Interest::sometimes()
|
||||
}
|
||||
fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool {
|
||||
true
|
||||
SINK.with(|s| s.borrow().is_some())
|
||||
}
|
||||
// Spans are irrelevant here — nothing in this crate asserts on span
|
||||
// structure, only on events — so they get a constant id and no storage.
|
||||
@@ -99,56 +125,62 @@ impl tracing::Subscriber for Capture {
|
||||
fn record(&self, _span: &tracing::span::Id, _values: &tracing::span::Record<'_>) {}
|
||||
fn record_follows_from(&self, _span: &tracing::span::Id, _follows: &tracing::span::Id) {}
|
||||
fn event(&self, event: &tracing::Event<'_>) {
|
||||
SINK.with(|s| {
|
||||
if let Some(sink) = s.borrow().as_ref() {
|
||||
let mut v = Visitor::default();
|
||||
event.record(&mut v);
|
||||
let meta = event.metadata();
|
||||
self.0.lock().expect("capture mutex").push(CapturedEvent {
|
||||
sink.lock().expect("capture mutex").push(CapturedEvent {
|
||||
target: meta.target().to_string(),
|
||||
level: *meta.level(),
|
||||
fields: v.0,
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
fn enter(&self, _span: &tracing::span::Id) {}
|
||||
fn exit(&self, _span: &tracing::span::Id) {}
|
||||
}
|
||||
|
||||
/// Process-wide lock serialising captures. See [`capture`].
|
||||
static CAPTURE_LOCK: Mutex<()> = Mutex::new(());
|
||||
/// Install the single global capturing subscriber, exactly once.
|
||||
fn install() {
|
||||
static INSTALLED: std::sync::OnceLock<()> = std::sync::OnceLock::new();
|
||||
INSTALLED.get_or_init(|| {
|
||||
// This is the crate's ONLY `set_global_default`; a second call cannot
|
||||
// happen. Tolerate it via `.ok()` rather than panic if that ever
|
||||
// changes — capture then no-ops, which the non-empty assertions catch.
|
||||
let _ = tracing::subscriber::set_global_default(Capture);
|
||||
});
|
||||
}
|
||||
|
||||
/// Run `f` with every `tracing` event emitted on THIS thread captured.
|
||||
/// Run `f` with every `tracing` event it emits ON THIS THREAD captured.
|
||||
///
|
||||
/// Returns `f`'s value alongside the events, in emission order.
|
||||
///
|
||||
/// # Why captures are serialised across the whole test binary
|
||||
/// # Why one global subscriber, not scoped `with_default`
|
||||
///
|
||||
/// `tracing`'s per-callsite INTEREST CACHE is global, while
|
||||
/// `with_default` is thread-local, and the two race under `cargo test`'s
|
||||
/// parallel harness. `tracing_core` rebuilds that cache only on the 0 -> 1 and
|
||||
/// 1 -> 0 transitions of its scoped-dispatcher count: entering the first
|
||||
/// capture flips every callsite to "ask the subscriber", leaving the last one
|
||||
/// flips them back to "never" (there being no global subscriber in tests).
|
||||
/// With two capturing tests on two threads, the exiting one's rebuild can land
|
||||
/// AFTER the entering one's, leaving the cache at "never" while a capture is
|
||||
/// live — so the callsite is short-circuited and the capture observes NOTHING.
|
||||
/// `tracing`'s per-callsite INTEREST CACHE is GLOBAL, but `with_default` is
|
||||
/// thread-local. Under `cargo test`'s parallel harness a rebuild of that cache
|
||||
/// — triggered by ANY other thread registering any callsite for the first time
|
||||
/// — re-evaluates the target callsite against the global dispatcher, which a
|
||||
/// scoped subscriber is NOT part of, and can leave it cached "off" while a
|
||||
/// capture is live, so the capture observes NOTHING. Serialising captures
|
||||
/// against each other does not help: the poisoning thread is not itself
|
||||
/// capturing. `parse_playlist_unreadable_clip_icb_yields_no_title` flaked ~1
|
||||
/// run in 15 on exactly this — an empty event list.
|
||||
///
|
||||
/// That was not theoretical: `parse_playlist_unreadable_clip_icb_yields_no_title`
|
||||
/// passed alone and failed in the full suite, with an empty event list, the
|
||||
/// first time two capturing tests existed in one module. A logging assertion
|
||||
/// that fails at random is worse than none — it teaches the next person to
|
||||
/// re-run until green, which is how a real regression gets waved through.
|
||||
///
|
||||
/// Holding this lock across the whole of `with_default` — including the
|
||||
/// guard's drop, which is where the exiting rebuild happens — orders the
|
||||
/// transitions strictly. Captures are few and short, so the serialisation
|
||||
/// costs nothing measurable.
|
||||
///
|
||||
/// The lock is deliberately taken through the poison, not `expect`ed: a
|
||||
/// capturing test that panics (i.e. a genuine assertion failure) must not
|
||||
/// convert every other capturing test into a confusing secondary failure.
|
||||
/// The robust shape is a single subscriber installed globally for the whole
|
||||
/// run, whose `register_callsite` returns `sometimes` (so no callsite is ever
|
||||
/// hard-cached) and which routes each event to the emitting thread's own sink.
|
||||
/// No scoped-dispatcher transitions, no cross-thread cache race, and concurrent
|
||||
/// captures on different threads stay isolated by the thread-local sink.
|
||||
pub(crate) fn capture<T>(f: impl FnOnce() -> T) -> (T, Vec<CapturedEvent>) {
|
||||
let _guard = CAPTURE_LOCK.lock().unwrap_or_else(|e| e.into_inner());
|
||||
let sink: Arc<Mutex<Vec<CapturedEvent>>> = Arc::default();
|
||||
let out = tracing::subscriber::with_default(Capture(sink.clone()), f);
|
||||
install();
|
||||
let sink: Sink = Arc::default();
|
||||
// Save/restore any outer sink so a nested capture on one thread still works.
|
||||
let prev = SINK.with(|s| s.borrow_mut().replace(sink.clone()));
|
||||
let out = f();
|
||||
SINK.with(|s| *s.borrow_mut() = prev);
|
||||
let events = std::mem::take(&mut *sink.lock().expect("capture mutex"));
|
||||
(out, events)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user