//! `WOW_SYSTEM_TIMING=1`: per-system wall time, once a second, from the `tracing` `"system"` spans
//! Bevy wraps every system in (`function_system.rs`). A measuring lever only — the layer is always
//! installed beside [`crate::log_ring`] and does nothing unless the variable is set.
//!
//! Systems are timed on span enter/exit, so a system's time is its own wall time (which for a
//! parallel system may overlap another's). Sorted costliest-first, with the run count beside the
//! per-run cost, so a crowded-frame hotspot names itself without a profiler.
//!
//! **Why the run count is printed and why names are dropped on close.** Not every `"system"` span
//! belongs to a scheduled system. `SystemMeta::new::<Param>()` builds one too, named by the
//! *parameter tuple* rather than a function, and `bevy_winit`'s cursor pass creates a fresh
//! `SystemState<(ResMut<WinitCustomCursorCache>, Query<…, Changed<PendingCursor>>)>` every frame
//! inside the winit runner — a span created, never entered, and dropped. Those names used to
//! accumulate forever in `SystemTiming::names`, one a frame, and a `tracing_subscriber`
//! registry recycles a closed span's slot: a later span whose metadata name is *not* `"system"`
//! can land on a recycled id, be skipped by [`Layer::on_new_span`], and have its enter/exit
//! charged to whatever name the dead span left behind. That is how a never-entered parameter
//! tuple came second in a report at 27.6 ms a run. [`Layer::on_close`] now drops the name with the
//! span, and the run count makes "27 ms once" unmistakable from "27 ms every frame".

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use bevy::log::tracing::field::{Field, Visit};
use bevy::log::tracing::span::Attributes;
use bevy::log::tracing::{Id, Subscriber};
use bevy::log::tracing_subscriber::layer::Context;
use bevy::log::tracing_subscriber::Layer;

/// How many threads may publish a running span at once ([`Slot`]). Past this a thread still times
/// its systems; it just does not appear in the watchdog's view.
const SLOTS: usize = 128;

/// One thread's "what am I inside right now", published so [`spawn_watchdog`] can read it without
/// taking any lock the stalled thread might be holding.
#[derive(Default)]
struct Slot {
    /// The span id currently entered on this thread, or 0 for none.
    span: AtomicU64,
    /// Nanoseconds since [`epoch`] at which it was entered.
    since_ns: AtomicU64,
    /// Bumped on every enter and every exit: proof of progress. An open span is not a stall —
    /// every enclosing span is open by construction, and `on_exit` republishes the parent, so the
    /// slot shows a span that has been open since the app started whenever an inner one returns.
    /// A thread that is actually wedged stops moving this.
    beat: AtomicU64,
    /// Which thread took this slot, written once when it claims one. The span it is stuck inside
    /// may be unnamed — an enclosing `info_span!` carries no `name` field for `on_new_span` to
    /// record — and then the thread is the only thing a report can name. Behind a lock because it
    /// is read when printing and never while detecting.
    thread: Mutex<Option<String>>,
}

/// The process clock every [`Slot::since_ns`] is measured against.
fn epoch() -> Instant {
    static EPOCH: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
    *EPOCH.get_or_init(Instant::now)
}

/// A thread's own accumulation, merged into the shared totals once a second.
///
/// **Nothing on the hot path touches a shared lock, and that is the whole design.** The first cut
/// of this layer took two mutexes per system *enter* and four per *exit*, from every worker
/// thread, on the order of a hundred thousand times a second. Measured against the same scene
/// with the variable unset, that cost **15 ms a frame — 590 fps became 60**. An instrument that
/// slows its subject tenfold is not measuring it, it is replacing it; several rounds of this
/// project's worst stutter hunt were spent reading stalls that were this layer's own lock
/// contention. Per-thread state, atomics for what the watchdog must see across threads, and one
/// merge a second.
struct Local {
    /// Index into [`SystemTiming::slots`], or `usize::MAX` when none was free.
    slot: usize,
    /// Spans entered on this thread, innermost last; they nest (a system's span, then the
    /// commands span inside it).
    stack: Vec<(u64, Instant)>,
    /// Span id → `(total_ms, runs, worst_ms)` since this thread last merged.
    totals: HashMap<u64, (f64, u64, f64)>,
    merged_at: Option<Instant>,
}

impl Local {
    fn new() -> Self {
        Self {
            slot: usize::MAX,
            stack: Vec::new(),
            totals: HashMap::new(),
            merged_at: None,
        }
    }
}

thread_local! {
    /// Per-thread, not per-layer: one `SystemTiming` exists in a process, and the tests that build
    /// more run on threads of their own.
    static LOCAL: std::cell::RefCell<Local> = std::cell::RefCell::new(Local::new());
}

/// The layer; installed always, active only with `WOW_SYSTEM_TIMING`.
pub struct SystemTiming {
    on: bool,
    /// Span id → system name. A scheduled system's span is created at init and never closed,
    /// so its entry is permanent; a `SystemState`'s is dropped by `on_close` with the span.
    ///
    /// `Arc` so [`spawn_watchdog`] can read it from its own thread. Written at span creation and
    /// read once a second per thread, so this lock is cold.
    names: Arc<Mutex<HashMap<u64, String>>>,
    /// Span id → `(total_ms, runs, worst_ms)` since the last print, merged from [`Local`].
    totals: Mutex<HashMap<u64, (f64, u64, f64)>>,
    last: Mutex<Instant>,
    /// What each thread is inside, for the watchdog.
    slots: Arc<Vec<Slot>>,
    next_slot: AtomicUsize,
}

impl SystemTiming {
    pub fn new() -> Self {
        let on = std::env::var_os("WOW_SYSTEM_TIMING").is_some();
        let names = Arc::new(Mutex::new(HashMap::new()));
        let slots: Arc<Vec<Slot>> = Arc::new((0..SLOTS).map(|_| Slot::default()).collect());
        epoch();
        if on {
            eprintln!("[sys-timing] armed — per-system ms once a second");
            spawn_watchdog(&names, &slots);
        }
        Self {
            on,
            names,
            totals: Mutex::new(HashMap::new()),
            last: Mutex::new(Instant::now()),
            slots,
            next_slot: AtomicUsize::new(0),
        }
    }

    /// Give this thread a slot the first time it enters a span.
    fn claim_slot(&self, local: &mut Local) {
        if local.slot != usize::MAX {
            return;
        }
        let taken = self.next_slot.fetch_add(1, Ordering::Relaxed);
        local.slot = if taken < SLOTS { taken } else { usize::MAX };
        if let Some(slot) = self.slots.get(local.slot) {
            *slot.thread.lock().unwrap_or_else(|e| e.into_inner()) =
                std::thread::current().name().map(str::to_string);
        }
    }

    /// Fold this thread's tally into the shared one, dropping spans that are not systems, and
    /// print if a second has passed since the last report.
    fn merge(&self, local: &mut Local) {
        local.merged_at = Some(Instant::now());
        let drained: Vec<(u64, (f64, u64, f64))> = local.totals.drain().collect();
        if !drained.is_empty() {
            let names = self.names.lock().unwrap_or_else(|e| e.into_inner());
            let mut totals = self.totals.lock().unwrap_or_else(|e| e.into_inner());
            for (id, (total, runs, worst)) in drained {
                // Every span passes through this layer, not only the scheduler's; one with no
                // recorded name is not a system and is dropped here rather than carried.
                if !names.contains_key(&id) {
                    continue;
                }
                let entry = totals.entry(id).or_insert((0.0, 0, 0.0));
                entry.0 += total;
                entry.1 += runs;
                entry.2 = entry.2.max(worst);
            }
        }
        let window = {
            let mut last = self.last.lock().unwrap_or_else(|e| e.into_inner());
            if last.elapsed() < Duration::from_secs(1) {
                return;
            }
            let window = last.elapsed().as_secs_f64();
            *last = Instant::now();
            window
        };
        self.report(window);
    }

    /// Print the window's rows, costliest per run first.
    fn report(&self, window: f64) {
        let mut rows: Vec<(f64, u64, f64, f64, String)> = {
            let names = self.names.lock().unwrap_or_else(|e| e.into_inner());
            let mut totals = self.totals.lock().unwrap_or_else(|e| e.into_inner());
            totals
                .drain()
                .filter_map(|(id, (total, runs, worst))| {
                    let name = names.get(&id)?.clone();
                    #[allow(clippy::cast_precision_loss)]
                    Some((total / runs.max(1) as f64, runs, total, worst, name))
                })
                .collect()
        };
        if rows.is_empty() {
            return;
        }
        rows.sort_by(|a, b| b.0.total_cmp(&a.0));
        eprintln!("[sys-timing] {window:.2}s window · ms per run, costliest first:");
        eprintln!("[sys-timing]   ms/run     runs    ms/s   worst  system");
        for (per_run, runs, total, worst, name) in rows.iter().take(30) {
            if *per_run < 0.02 {
                break;
            }
            // `ms/s` is the row's whole share of the second: a system at 27 ms a run that ran once
            // costs a thirtieth of what the same row costs at 60 runs, and only this column says
            // which it was.
            //
            // `worst` is the single costliest run in the window, and it is here because the run
            // count alone was *not* enough to read a row honestly. A system that early-returns
            // every frame but did one 4.3 s load during the window reported `11.4 ms/run` over
            // 377 runs — indistinguishable, in those two columns, from 11 ms of real work every
            // frame, and read exactly that way. `worst` ≈ `ms/run` is a steady cost; `worst` far
            // above it is one event hiding in an average.
            eprintln!(
                "[sys-timing] {per_run:8.3} {runs:8} {:7.1} {worst:7.1}  {name}",
                total / window.max(f64::EPSILON)
            );
        }
    }

    /// Merge this thread's tally now, for a test that cannot wait a second for it.
    #[cfg(test)]
    fn merge_now(&self) {
        LOCAL.with(|cell| {
            let mut local = cell.borrow_mut();
            let drained: Vec<(u64, (f64, u64, f64))> = local.totals.drain().collect();
            let names = self.names.lock().unwrap_or_else(|e| e.into_inner());
            let mut totals = self.totals.lock().unwrap_or_else(|e| e.into_inner());
            for (id, (total, runs, worst)) in drained {
                if !names.contains_key(&id) {
                    continue;
                }
                let entry = totals.entry(id).or_insert((0.0, 0, 0.0));
                entry.0 += total;
                entry.1 += runs;
                entry.2 = entry.2.max(worst);
            }
        });
    }
}

impl<S: Subscriber> Layer<S> for SystemTiming {
    fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, _ctx: Context<'_, S>) {
        if !self.on || attrs.metadata().name() != "system" {
            return;
        }
        let mut name = String::new();
        attrs.record(&mut NameVisitor(&mut name));
        if !name.is_empty() {
            self.names
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .insert(id.into_u64(), name);
        }
    }

    fn on_enter(&self, id: &Id, _ctx: Context<'_, S>) {
        if !self.on {
            return;
        }
        let raw = id.into_u64();
        let now = Instant::now();
        LOCAL.with(|cell| {
            // `try_borrow_mut`, not `borrow_mut`: a panic here would be inside a tracing callback
            // on an arbitrary thread, and a dropped sample is the cheaper failure.
            let Ok(mut local) = cell.try_borrow_mut() else {
                return;
            };
            self.claim_slot(&mut local);
            local.stack.push((raw, now));
            if let Some(slot) = self.slots.get(local.slot) {
                // Timestamp first, id second: the watchdog reads the id to decide there is
                // anything to look at, so it must never find one beside a stale timestamp.
                slot.since_ns.store(
                    now.saturating_duration_since(epoch()).as_nanos() as u64,
                    Ordering::Relaxed,
                );
                slot.span.store(raw, Ordering::Relaxed);
                slot.beat.fetch_add(1, Ordering::Relaxed);
            }
        });
    }

    /// Drop the span's name with the span. A closed span's id is returned to the registry's pool
    /// and handed to a later span, so a name kept past the close can be charged another span's
    /// time (see the module header).
    fn on_close(&self, id: Id, _ctx: Context<'_, S>) {
        if !self.on {
            return;
        }
        self.names
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&id.into_u64());
    }

    fn on_exit(&self, id: &Id, _ctx: Context<'_, S>) {
        if !self.on {
            return;
        }
        let raw = id.into_u64();
        LOCAL.with(|cell| {
            let Ok(mut local) = cell.try_borrow_mut() else {
                return;
            };
            // From the top: spans nest, and an exit always matches the innermost frame with this
            // id. A span entered before this layer saw it has no frame and is skipped.
            let Some(at) = local.stack.iter().rposition(|&(span, _)| span == raw) else {
                return;
            };
            let (_, started) = local.stack.remove(at);
            let ms = started.elapsed().as_secs_f64() * 1000.0;
            let entry = local.totals.entry(raw).or_insert((0.0, 0, 0.0));
            entry.0 += ms;
            entry.1 += 1;
            entry.2 = entry.2.max(ms);
            // Republish the enclosing span, so the watchdog keeps seeing something true rather
            // than an idle slot in the middle of a long parent.
            if let Some(slot) = self.slots.get(local.slot) {
                match local.stack.last().copied() {
                    Some((parent, since)) => {
                        slot.since_ns.store(
                            since.saturating_duration_since(epoch()).as_nanos() as u64,
                            Ordering::Relaxed,
                        );
                        slot.span.store(parent, Ordering::Relaxed);
                    }
                    None => slot.span.store(0, Ordering::Relaxed),
                }
                slot.beat.fetch_add(1, Ordering::Relaxed);
            }
            let due = local
                .merged_at
                .is_none_or(|at| at.elapsed() >= Duration::from_secs(1));
            if due {
                self.merge(&mut local);
            }
        });
    }
}

/// Resident set size in MiB and the process's page-fault counters, straight off `/proc/self`.
///
/// Read from the watchdog's thread, because the question it answers — "was the process stopped
/// because it was waiting on memory?" — is one a frozen frame loop cannot be asked. A stall that
/// lands on unrelated systems at once, including systems whose bodies do nothing, is not any
/// system's fault; it is the whole process being stopped, and growing RSS with a burst of faults
/// is what that looks like from the inside.
#[cfg(target_os = "linux")]
fn process_memory() -> Option<(f64, u64, u64)> {
    let statm = std::fs::read_to_string("/proc/self/statm").ok()?;
    // Field 2 of `statm` is resident pages.
    let rss_pages: u64 = statm.split_whitespace().nth(1)?.parse().ok()?;
    #[allow(clippy::cast_precision_loss)]
    let rss_mib = (rss_pages * 4096) as f64 / (1024.0 * 1024.0);
    let stat = std::fs::read_to_string("/proc/self/stat").ok()?;
    // `comm` can contain spaces and parentheses, so fields are counted after the last ')'.
    let rest = &stat[stat.rfind(')')? + 1..];
    let f: Vec<&str> = rest.split_whitespace().collect();
    // After `comm` and `state`, `minflt` and `majflt` are the 3rd and 5th remaining fields.
    let minflt: u64 = f.get(7)?.parse().ok()?;
    let majflt: u64 = f.get(9)?.parse().ok()?;
    Some((rss_mib, minflt, majflt))
}

#[cfg(not(target_os = "linux"))]
fn process_memory() -> Option<(f64, u64, u64)> {
    None
}

/// Every thread's scheduling state during a stall, as `name:state` pairs.
///
/// The state letter is the thing worth having: `R` is on-CPU (the process is *doing* something),
/// `S` is an ordinary sleep (waiting on us), and `D` is uninterruptible — blocked in the kernel,
/// which is a driver or a filesystem and not our code at all. A process-wide stop with every
/// thread in `S` is a different bug from one with a thread in `D`, and there is no way to tell
/// them apart after the fact.
#[cfg(target_os = "linux")]
fn thread_states() -> String {
    let Ok(dir) = std::fs::read_dir("/proc/self/task") else {
        return String::new();
    };
    let mut seen: Vec<String> = Vec::new();
    for entry in dir.flatten() {
        let Ok(stat) = std::fs::read_to_string(entry.path().join("stat")) else {
            continue;
        };
        let Some(close) = stat.rfind(')') else {
            continue;
        };
        let name = stat
            .find('(')
            .map(|open| stat[open + 1..close].to_string())
            .unwrap_or_default();
        let state = stat[close + 1..].split_whitespace().next().unwrap_or("?");
        seen.push(format!("{name}:{state}"));
    }
    seen.sort();
    seen.join(" ")
}

#[cfg(not(target_os = "linux"))]
fn thread_states() -> String {
    String::new()
}

/// The stall watchdog (`WOW_STALL_MS=<ms>`, default 150): a thread that names a system which has
/// entered and not returned, **while it is still in there**.
///
/// Every other report in this module — and in `perf::phases`, `script::tick_prof`,
/// `script::handler_prof`, all of them — is printed from inside the frame loop, once a window has
/// elapsed. That makes them structurally blind to the thing most worth seeing: if the client is
/// wedged, no frame completes, no window closes, and nothing prints. A freeze leaves no sample,
/// so profiles read from the moments *between* freezes, and those look healthy — which is exactly
/// how a session that was plainly stuttering produced a clean profile over and over.
///
/// This thread does not wait for anything. It wakes on its own clock, reads what each thread is
/// inside, and prints the ones that have stopped making progress, then prints again when one
/// finally moves with the full duration it was in there. A wedged client therefore narrates itself.
///
/// "Stopped making progress" is [`Slot::beat`], not "has an open span". An open span is the normal
/// case: spans nest, `on_exit` republishes the enclosing one, and the outermost is open from the
/// first frame to the last. Keying on openness alone reported a 200-second-old parent as a stall
/// every time an inner span returned — 13k lines in one session's log, about a span that was never
/// in trouble, from the instrument whose whole job is to be trustworthy when nothing else is.
fn spawn_watchdog(names: &Arc<Mutex<HashMap<u64, String>>>, slots: &Arc<Vec<Slot>>) {
    let threshold_ms: f64 = std::env::var("WOW_STALL_MS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(150.0);
    let (names, slots) = (Arc::clone(names), Arc::clone(slots));
    eprintln!("[sys-timing] stall watchdog armed — a system still running after {threshold_ms:.0} ms is named (WOW_STALL_MS)");
    let spawned = std::thread::Builder::new()
        .name("benilla stall watch".to_string())
        .spawn(move || {
            // Slot → who was reported, so a stall is announced once and its end once. Keyed by
            // slot, not span: what is stalled is the thread, and the span it is stuck in is only
            // how we name it.
            let mut reported: HashMap<usize, String> = HashMap::new();
            // Per slot, the beat last seen and when it last moved. See [`Slot::beat`].
            let mut beats: Vec<(u64, Instant)> = vec![(0, Instant::now()); slots.len()];
            let mut mem_at = Instant::now();
            let mut was = process_memory();
            loop {
                std::thread::sleep(Duration::from_millis(25));
                // The memory line, once a second, from this thread: it keeps printing through a
                // freeze, which is the whole reason it lives here and not in a system.
                if mem_at.elapsed() >= Duration::from_secs(1) {
                    let secs = mem_at.elapsed().as_secs_f64();
                    mem_at = Instant::now();
                    if let (Some((rss, min_now, maj_now)), Some((prev, min_was, maj_was))) =
                        (process_memory(), was)
                    {
                        eprintln!(
                            "[sys-timing] mem: rss {rss:.0} MiB ({:+.1} MiB/s) · faults {:.0}/s minor, {:.0}/s major",
                            (rss - prev) / secs,
                            (min_now.saturating_sub(min_was)) as f64 / secs,
                            (maj_now.saturating_sub(maj_was)) as f64 / secs,
                        );
                        was = Some((rss, min_now, maj_now));
                    }
                }
                // Plain atomic loads, no lock at all: a thread wedged while holding one must
                // not be able to hide from the instrument meant to catch it.
                let now_ns = epoch().elapsed().as_nanos() as u64;
                let mut fresh: Vec<String> = Vec::new();
                let mut done: Vec<(String, f64)> = Vec::new();
                for (i, slot) in slots.iter().enumerate() {
                    let beat = slot.beat.load(Ordering::Relaxed);
                    // The thread entered or exited something since the last poll, so whatever it
                    // is inside now, it is not stuck in it.
                    if beat != beats[i].0 {
                        let stuck_ms = beats[i].1.elapsed().as_secs_f64() * 1000.0;
                        beats[i] = (beat, Instant::now());
                        if let Some(who) = reported.remove(&i) {
                            done.push((who, stuck_ms));
                        }
                        continue;
                    }
                    if beats[i].1.elapsed().as_secs_f64() * 1000.0 < threshold_ms
                        || reported.contains_key(&i)
                    {
                        continue;
                    }
                    let span = slot.span.load(Ordering::Relaxed);
                    if span == 0 {
                        // An idle worker with nothing entered. It is not inside anything, so
                        // there is nothing to name and nothing is wrong.
                        continue;
                    }
                    let since = slot.since_ns.load(Ordering::Relaxed);
                    // Written timestamp-then-id, read id-then-timestamp-then-id: if the id moved
                    // under us the timestamp may belong to the other span, so drop the sample
                    // rather than report a duration nobody spent.
                    if slot.span.load(Ordering::Relaxed) != span {
                        continue;
                    }
                    let name = names
                        .lock()
                        .ok()
                        .and_then(|n| n.get(&span).cloned())
                        .unwrap_or_else(|| format!("span #{span}"));
                    let thread = slot
                        .thread
                        .lock()
                        .ok()
                        .and_then(|t| t.clone())
                        .unwrap_or_else(|| format!("slot {i}"));
                    // Two durations, and they answer different questions. How long the thread has
                    // been stuck is the defect; how long the span has been open is only where it
                    // is stuck, and for an enclosing span that is the whole session. Printing the
                    // second as though it were the first is what the old report did.
                    #[allow(clippy::cast_precision_loss)]
                    let open_ms = now_ns.saturating_sub(since) as f64 / 1.0e6;
                    let stuck_ms = beats[i].1.elapsed().as_secs_f64() * 1000.0;
                    reported.insert(i, format!("{thread} in {name}"));
                    fresh.push(format!(
                        "thread {thread:?} has made no progress for {stuck_ms:.0} ms — inside {name}, open {open_ms:.0} ms"
                    ));
                }
                for sentence in fresh {
                    // RSS beside the stall: a process-wide stop and a climbing resident set in the
                    // same breath is the memory answer; a stall with flat memory is not.
                    let mem = process_memory()
                        .map(|(rss, _, _)| format!(" · rss {rss:.0} MiB"))
                        .unwrap_or_default();
                    eprintln!("[sys-timing] STALL: {sentence}{mem}");
                    let states = thread_states();
                    if !states.is_empty() {
                        eprintln!("[sys-timing] STALL threads: {states}");
                    }
                }
                for (who, ms) in done {
                    eprintln!("[sys-timing] STALL ended: {who} moved again after {ms:.0} ms stuck");
                }
            }
        });
    if spawned.is_err() {
        eprintln!("[sys-timing] stall watchdog could not start — the periodic report still works");
    }
}

impl Default for SystemTiming {
    fn default() -> Self {
        Self::new()
    }
}

/// The log ring and this timer as one layer for `LogPlugin::custom_layer`. A free function so
/// `Layer::and` resolves here, where Bevy's prelude (`SystemCondition::and`) is not in scope.
pub fn log_layers() -> impl Layer<bevy::log::tracing_subscriber::Registry> + Send + Sync + 'static {
    use bevy::log::tracing_subscriber::Registry;
    <crate::log_ring::LogRing as Layer<Registry>>::and_then(
        crate::log_ring::LogRing,
        SystemTiming::new(),
    )
}

/// Reads the `name` field off a span.
struct NameVisitor<'a>(&'a mut String);

impl Visit for NameVisitor<'_> {
    fn record_str(&mut self, field: &Field, value: &str) {
        if field.name() == "name" {
            *self.0 = value.to_string();
        }
    }
    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
        if field.name() == "name" && self.0.is_empty() {
            *self.0 = format!("{value:?}");
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use bevy::log::tracing;
    use bevy::log::tracing_subscriber::layer::SubscriberExt;

    use super::*;

    /// Forwards to a shared [`SystemTiming`], so the test can read its totals after the subscriber
    /// that drove it is gone. `tracing_subscriber` takes the layer by value.
    struct Shared(Arc<SystemTiming>);

    impl<S: Subscriber> Layer<S> for Shared {
        fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
            self.0.on_new_span(attrs, id, ctx);
        }
        fn on_enter(&self, id: &Id, ctx: Context<'_, S>) {
            self.0.on_enter(id, ctx);
        }
        fn on_exit(&self, id: &Id, ctx: Context<'_, S>) {
            self.0.on_exit(id, ctx);
        }
        fn on_close(&self, id: Id, ctx: Context<'_, S>) {
            self.0.on_close(id, ctx);
        }
    }

    fn armed() -> Arc<SystemTiming> {
        Arc::new(SystemTiming {
            on: true,
            names: Arc::new(Mutex::new(HashMap::new())),
            totals: Mutex::new(HashMap::new()),
            last: Mutex::new(Instant::now()),
            slots: Arc::new((0..SLOTS).map(|_| Slot::default()).collect()),
            next_slot: AtomicUsize::new(0),
        })
    }

    /// The tally by system name. The layer keys everything on span ids and resolves names only
    /// when it prints, so a test reads it the same way — and merges this thread's share first,
    /// since the hot path no longer publishes anything shared until it has to.
    fn rows(timing: &SystemTiming) -> HashMap<String, (f64, u64, f64)> {
        timing.merge_now();
        let names = timing.names.lock().unwrap();
        let totals = timing.totals.lock().unwrap();
        totals
            .iter()
            .filter_map(|(id, v)| names.get(id).map(|name| (name.clone(), *v)))
            .collect()
    }

    /// How many spans it takes to make `tracing_subscriber`'s registry hand out an id it has used
    /// before. The pool packs a 13-bit generation beside the slot address, so a slot's id repeats
    /// every 8,191 reuses — a couple of minutes of play, not a lifetime. Twice that, so the test
    /// does not sit on the boundary.
    const UNTIL_IDS_REPEAT: usize = 16_400;

    /// The property this layer has to hold: a span's name belongs to that span only. A span that
    /// Bevy creates and drops without ever entering it — `SystemMeta::new::<Param>()` makes one
    /// per frame inside `bevy_winit`'s cursor pass, named by its parameter tuple — must not end up
    /// charged the time of whatever later span inherits its id.
    ///
    /// Without [`Layer::on_close`] dropping the name, it does, and the report lies about which
    /// system is expensive: a user's session had the winit cursor parameter tuple second at 27.6
    /// ms a run, which was the UI tick's `ui_script: tick` span wearing a dead span's name.
    #[test]
    fn a_closed_spans_name_is_not_charged_a_later_spans_time() {
        let timing = armed();
        let subscriber =
            bevy::log::tracing_subscriber::registry().with(Shared(Arc::clone(&timing)));
        let totals = tracing::subscriber::with_default(subscriber, || {
            for _ in 0..UNTIL_IDS_REPEAT {
                // The ghost: metadata name `"system"`, so the layer records its name, and never
                // entered, so it can only ever be charged somebody else's time.
                drop(tracing::info_span!(parent: None, "system", name = "never-entered"));
                // The real work, under a metadata name the layer does not record. On a recycled
                // id, `on_new_span` skips it and the stale name is all `on_exit` can find.
                let work = tracing::info_span!(parent: None, "ui_script: tick");
                drop(work.enter());
            }
            rows(&timing)
        });
        assert!(
            !totals.contains_key("never-entered"),
            "a span that was never entered was charged {:?} — a closed span's name outlived it \
             and a later span inherited its id",
            totals.get("never-entered")
        );
    }

    /// The other half, so the test above cannot pass by the layer having stopped measuring: a span
    /// that *is* entered is still timed, and counted once per enter.
    #[test]
    fn a_system_span_that_is_entered_is_still_timed_once_per_run() {
        let timing = armed();
        let subscriber =
            bevy::log::tracing_subscriber::registry().with(Shared(Arc::clone(&timing)));
        // Read inside the closure, while the span is still alive: the layer keys its tally on
        // span ids and resolves names when it prints, so a span that has closed — and had its
        // name dropped with it, which is the point of `on_close` — has no row left to read. In a
        // session that costs nothing: a scheduled system's span is created at init and never
        // closes, which is exactly the span this instrument is for.
        let totals = tracing::subscriber::with_default(subscriber, || {
            let span = tracing::info_span!(parent: None, "system", name = "a::real::system");
            for _ in 0..7 {
                drop(span.enter());
            }
            rows(&timing)
        });
        let (total, runs, worst) = totals
            .get("a::real::system")
            .copied()
            .expect("an entered system span is timed");
        assert_eq!(runs, 7, "one run per enter");
        assert!(
            worst <= total,
            "the worst single run cannot exceed the window's total"
        );
    }

    /// What tells a working thread from a wedged one. A thread inside a long-lived outer span
    /// publishes that span, and its start, for as long as it lives: `on_exit` republishes the
    /// parent, so between two inner spans the slot shows one that has been open since the first
    /// frame. The watchdog used to read that as a stall and said so every 75 ms for the length of
    /// a session.
    ///
    /// So the published span is asserted to hold *still* across the inner work — the false
    /// positive, reproduced — while [`Slot::beat`] is asserted to move, which is the only thing
    /// separating the two cases.
    #[test]
    fn an_enclosing_span_holds_still_while_the_beat_moves() {
        let timing = armed();
        let subscriber =
            bevy::log::tracing_subscriber::registry().with(Shared(Arc::clone(&timing)));
        let (before, after, span_held, since_held) =
            tracing::subscriber::with_default(subscriber, || {
                let outer = tracing::info_span!(parent: None, "system", name = "the::whole::frame");
                let _held = outer.enter();
                // Whatever slot this thread took; `claim_slot` hands them out in order, but a
                // thread that entered a span under some other instance keeps the slot it had.
                let slot = |t: &SystemTiming| {
                    let i = (0..SLOTS)
                        .max_by_key(|&i| t.slots[i].beat.load(Ordering::Relaxed))
                        .expect("SLOTS is not zero");
                    (
                        t.slots[i].beat.load(Ordering::Relaxed),
                        t.slots[i].span.load(Ordering::Relaxed),
                        t.slots[i].since_ns.load(Ordering::Relaxed),
                    )
                };
                let (before, span_was, since_was) = slot(&timing);
                for _ in 0..5 {
                    let inner = tracing::info_span!(parent: None, "system", name = "a::system");
                    drop(inner.enter());
                }
                let (after, span_now, since_now) = slot(&timing);
                (before, after, span_was == span_now, since_was == since_now)
            });
        assert!(
            span_held && since_held,
            "the enclosing span and its start should be republished unchanged — without that \
             there is no false positive left to guard against"
        );
        assert!(
            after > before,
            "five enter/exit pairs moved the beat {before} -> {after}: a thread that is working \
             must be distinguishable from one stuck in the same span"
        );
    }
}