autumn-web 0.9.0

//! "What did this plugin do in the last hour?" (issue #1632).
//!
//! An operator who granted five capabilities to an artifact they did not audit
//! needs one place to answer that. Not five: a cache metric, an upstream's
//! access log, a job table and a `grep` of the application log is four places
//! and no answer, because none of them knows which plugin was responsible.
//!
//! So every capability call — allowed, denied or over quota — is recorded at the
//! one point that knows all of it — `CapabilityRuntime::record`, the single exit
//! every `dispatch` takes — and the records aggregate here:
//!
//! ```text
//!   request ──► CapabilityEvent × n ──► SandboxOutcome.activity
//!                                            │
//!                                            ▼
//!                                    PluginActivityLog  ──► ActivitySummary
//!                                    (bounded, windowed)     hosts called,
//!                                                            kv/db/job counts,
//!                                                            denials, quota hits
//! ```
//!
//! # What a record carries, and what it deliberately does not
//!
//! A record names the capability, the operation, and the *logical* target — the
//! key, the table, the hostname, the job type. It never carries a value, a row,
//! a request body or a response. An audit surface that logged what a plugin
//! stored would be a second copy of the tenant data the capability system exists
//! to contain, in a place with different access rules — and an operator asking
//! "what did it do" is asking about shape, not contents.

// autumn-determinism-gate: production code in this module must read time and
// mint identifiers through the framework's injected seams (ClockSource /
// Entropy), never `Instant::now()` / `Utc::now()` / `SystemTime::now()` /
// `Uuid::new_v4()` directly. See CONTRIBUTING.md "Determinism seam gate"
// (issue #1797). Justify exceptions with
// #[allow(clippy::disallowed_methods, reason = "…")] at the narrowest scope.
#![cfg_attr(not(test), deny(clippy::disallowed_methods))]

use std::collections::BTreeMap;
use std::fmt;
use std::sync::{Mutex, PoisonError};
use std::time::{Duration, Instant};

use serde::Serialize;

use super::super::manifest::SandboxCapability;
use super::DenialReason;

/// How one capability call ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
#[non_exhaustive]
pub enum CapabilityOutcome {
    /// The call was performed.
    Allowed,
    /// A quota refused it.
    QuotaExceeded,
    /// A grant refused it.
    Denied(DenialReason),
}

impl fmt::Display for CapabilityOutcome {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Allowed => f.write_str("allowed"),
            Self::QuotaExceeded => f.write_str("quota-exceeded"),
            Self::Denied(reason) => write!(f, "denied:{reason}"),
        }
    }
}

/// One thing a plugin did, or was stopped from doing.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[non_exhaustive]
pub struct CapabilityEvent {
    /// Which capability the call needed.
    pub capability: SandboxCapability,
    /// The operation, as the wire spells it (`kv-get`, `db-query`, …).
    pub operation: &'static str,
    /// The logical thing it named: a key, a table, a hostname, a job type.
    pub target: String,
    /// How it ended.
    pub outcome: CapabilityOutcome,
    /// When the call ended, on the ambient clock.
    ///
    /// Taken where the call is recorded, not where the log ingests it. A request
    /// can run for a long time, and stamping its events when it finishes made
    /// its earliest calls look recent.
    #[serde(skip)]
    pub at: Instant,
}

/// Calls one request made that its ledger could not hold, and when.
///
/// It keeps only the first and last times. The ledger had no room for the time
/// of each call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct DroppedCalls {
    /// How many calls.
    pub count: u64,
    /// The time of the first dropped call.
    pub first: Instant,
    /// The time of the last dropped call.
    pub last: Instant,
}

/// The most events one plugin's log holds.
///
/// Bounded, because the log is fed by an untrusted plugin's call rate and an
/// unbounded one would be that plugin's memory-exhaustion channel — the same
/// argument the per-request ledger's ceiling rests on, one level up. Oldest
/// entries are dropped first: an operator asking "in the last hour" wants the
/// recent end.
pub const MAX_LOG_EVENTS: usize = 4096;

/// A bounded, time-windowed record of what each plugin has been doing.
///
/// One log per application, keyed by plugin, because the question is asked per
/// plugin but the ceiling has to hold across all of them.
#[derive(Debug, Default)]
pub struct PluginActivityLog {
    /// Both ledgers under one lock. A summary then never sees an event that has
    /// left the ring and not yet reached the drop ledger. The work under the
    /// lock is constant time per event.
    state: Mutex<LogState>,
}

/// What [`PluginActivityLog`] guards.
#[derive(Debug, Default)]
struct LogState {
    /// Keyed by the event's own time, so the oldest event is the first key
    /// whatever order requests finished in. The sequence number separates equal
    /// times.
    entries: BTreeMap<(Instant, u64), (String, CapabilityEvent)>,
    /// Calls the log knows happened but cannot show: those a per-request
    /// ledger could not hold, and those the ring evicted.
    ///
    /// Keyed by time, so the oldest record is the first key whatever order the
    /// records arrived in. The sequence number keeps equal times apart. Windowed
    /// by the same cutoff as `entries`.
    dropped: BTreeMap<(Instant, u64), DroppedRecord>,
    /// The next sequence number for `dropped`.
    seq: u64,
    /// Counts the full `dropped` ledger folded away, by plugin and hour, with
    /// the newest time folded into each bucket. A summary reports a bucket
    /// while that time is in its window. It can over-report by under one hour
    /// plus the longest drop span of one request. It never under-reports.
    folded: BTreeMap<(String, u64), (Instant, u64)>,
    /// The time the hour buckets count from. Set by the first fold.
    origin: Option<Instant>,
    /// The newest time of a drop count the full `folded` ledger shed. Its
    /// owner is unknown, so a summary can only say that counts were lost.
    shed: Option<Instant>,
}

/// The gap between two instants, whichever is later.
fn apart(a: Instant, b: Instant) -> Duration {
    a.max(b).saturating_duration_since(a.min(b))
}

/// Calls the log knows happened but cannot show, for one plugin.
#[derive(Debug)]
struct DroppedRecord {
    plugin: String,
    count: u64,
    /// The time of the first drop. A record spans at most [`DROPPED_BUCKET`]
    /// from here, or the span of one request's drops if that is longer.
    first: Instant,
}

/// How long one `folded` bucket spans.
const FOLD_BUCKET_SECS: u64 = 3600;

/// How coarsely dropped-call records are timestamped.
///
/// One eviction is one dropped call, and evictions arrive at the plugin's whole
/// call rate — so recording each separately would cost as much memory as
/// keeping the events did. Drops for one plugin become one record while they
/// stay inside this span of the record's first drop. The span is far below the
/// "last hour" window, so the windowing stays honest.
const DROPPED_BUCKET: Duration = Duration::from_secs(1);

impl PluginActivityLog {
    /// An empty log.
    #[must_use]
    pub fn new() -> Self {
        // The fold buckets count from here. No event can predate its log.
        let state = LogState {
            origin: Some(crate::time::ambient_instant()),
            ..LogState::default()
        };
        Self {
            state: Mutex::new(state),
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, LogState> {
        self.state.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Record everything one request's runtime gathered.
    ///
    /// Each event keeps the time it carries, not the time of this call.
    pub fn ingest(&self, plugin: &str, events: impl IntoIterator<Item = CapabilityEvent>) {
        let mut state = self.lock();
        for event in events {
            if state.entries.len() >= MAX_LOG_EVENTS
                && let Some(((at, _), (who, _))) = state.entries.pop_first()
            {
                // An evicted event is a call the summary can no longer show.
                // It is recorded under its own time, so a drop that has aged
                // out of the window stops being reported with it.
                state.note_dropped(at, at, &who, 1);
            }
            state.seq = state.seq.wrapping_add(1);
            let key = (event.at, state.seq);
            state.entries.insert(key, (plugin.to_owned(), event));
        }
    }

    /// Record that one request made more calls than its ledger could hold.
    ///
    /// Separate from [`ingest`](Self::ingest) because the number is not an
    /// event: there is nothing to say about the calls beyond that they happened
    /// and are not below. An operator reading a summary needs to know that
    /// before they read the counts.
    ///
    /// Records the drops at the current time. Use
    /// [`ingest_dropped_calls`](Self::ingest_dropped_calls) when you know the
    /// times of the drops.
    pub fn ingest_dropped(&self, plugin: &str, dropped: u64) {
        if dropped == 0 {
            return;
        }
        let now = crate::time::ambient_instant();
        self.lock().note_dropped(now, now, plugin, dropped);
    }

    /// Record the calls one request's ledger could not hold, at their own times.
    ///
    /// A slow request ends long after its drops. This method records the drops
    /// at their own times, not at the time of ingest (#2523). The record counts
    /// while `calls.last` is in the window. It can over-report by the span of
    /// the drops. It never under-reports.
    pub fn ingest_dropped_calls(&self, plugin: &str, calls: DroppedCalls) {
        if calls.count == 0 {
            return;
        }
        // Ordered, so the key is never before a drop it counts.
        let (first, last) = (calls.first.min(calls.last), calls.first.max(calls.last));
        self.lock().note_dropped(first, last, plugin, calls.count);
    }

    /// What `plugin` did within `window`.
    #[must_use]
    pub fn summary(&self, plugin: &str, window: Duration) -> ActivitySummary {
        let cutoff = crate::time::ambient_instant();
        let in_window = |at: Instant| cutoff.saturating_duration_since(at) <= window;
        let mut summary = ActivitySummary {
            plugin: plugin.to_owned(),
            window,
            ..ActivitySummary::default()
        };
        let state = self.lock();
        for ((at, _), (who, event)) in &state.entries {
            if who == plugin && in_window(*at) {
                summary.record(event);
            }
        }
        summary.dropped = state
            .dropped
            .iter()
            .filter(|((at, _), record)| record.plugin == plugin && in_window(*at))
            .fold(0_u64, |sum, (_, record)| sum.saturating_add(record.count));
        let folded: u64 = state
            .folded
            .range((plugin.to_owned(), 0)..=(plugin.to_owned(), u64::MAX))
            .filter(|(_, (at, _))| in_window(*at))
            .fold(0_u64, |sum, (_, (_, count))| sum.saturating_add(*count));
        summary.dropped = summary.dropped.saturating_add(folded);
        summary.truncated = state.shed.is_some_and(in_window);
        drop(state);
        summary
    }

    /// Every plugin this log has heard from.
    ///
    /// All ledgers, not only the event ring. A busy plugin can evict a quiet
    /// one's every event while the drop ledger still carries the count of what
    /// was lost, and an operator surface that enumerates plugins here to ask
    /// each for a [`summary`] would then skip the one plugin whose activity is
    /// only visible as drops.
    ///
    /// [`summary`]: Self::summary
    #[must_use]
    pub fn plugins(&self) -> Vec<String> {
        let mut names: Vec<String> = {
            let state = self.lock();
            state
                .entries
                .values()
                .map(|(who, _)| who.clone())
                .chain(state.dropped.values().map(|record| record.plugin.clone()))
                .chain(state.folded.keys().map(|(who, _)| who.clone()))
                .collect()
        };
        names.sort();
        names.dedup();
        names
    }
}

impl LogState {
    /// Add `count` dropped calls for `plugin`, made from `first` to `last`.
    ///
    /// Coalesces into the newest record when it is the same plugin and within
    /// one [`DROPPED_BUCKET`] of that record's first drop. Bounded by
    /// [`MAX_LOG_EVENTS`]: the oldest record, by time, folds into `folded`
    /// rather than vanishing.
    fn note_dropped(&mut self, first: Instant, last: Instant, plugin: &str, count: u64) {
        // Measured from the first drop, not the newest. A steady dropper would
        // otherwise keep one record open for hours.
        if let Some((&(newest, seq), record)) = self.dropped.last_key_value()
            && record.plugin == plugin
            && apart(first, record.first) < DROPPED_BUCKET
            && apart(last, record.first) < DROPPED_BUCKET
            && apart(last, newest) < DROPPED_BUCKET
        {
            // Keyed by the newer time: the record counts while any drop in it
            // is inside the window.
            if let Some(mut record) = self.dropped.remove(&(newest, seq)) {
                record.count = record.count.saturating_add(count);
                record.first = record.first.min(first);
                self.dropped.insert((newest.max(last), seq), record);
            }
            return;
        }
        self.seq = self.seq.wrapping_add(1);
        self.dropped.insert(
            (last, self.seq),
            DroppedRecord {
                plugin: plugin.to_owned(),
                count,
                first,
            },
        );
        while self.dropped.len() > MAX_LOG_EVENTS {
            let Some((
                (oldest, _),
                DroppedRecord {
                    plugin: who, count, ..
                },
            )) = self.dropped.pop_first()
            else {
                break;
            };
            let origin = *self.origin.get_or_insert(oldest);
            let hour = oldest.saturating_duration_since(origin).as_secs() / FOLD_BUCKET_SECS;
            let folded = self.folded.entry((who, hour)).or_insert((oldest, 0));
            folded.0 = folded.0.max(oldest);
            folded.1 = folded.1.saturating_add(count);
        }
        // Bounded like the other ledgers. Past this the oldest hour goes, which
        // takes thousands of plugin-hours of overflow to reach. Keep the newest
        // shed time. A summary then says its counts are incomplete.
        while self.folded.len() > MAX_LOG_EVENTS {
            let Some(oldest) = self.folded.keys().min_by_key(|(_, hour)| *hour).cloned() else {
                break;
            };
            if let Some((at, _)) = self.folded.remove(&oldest) {
                self.shed = Some(self.shed.map_or(at, |seen| seen.max(at)));
            }
        }
    }
}

/// The answer to "what did this plugin do".
///
/// Counts and names, never values. See the module header.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
#[non_exhaustive]
pub struct ActivitySummary {
    /// Calls that were performed, by operation.
    pub allowed: BTreeMap<&'static str, u64>,
    /// Calls a grant refused, by operation.
    pub denied: BTreeMap<&'static str, u64>,
    /// Calls a quota refused, by operation.
    pub quota_hits: BTreeMap<&'static str, u64>,
    /// Hostnames actually called, and how often.
    pub hosts: BTreeMap<String, u64>,
    /// Tables touched, and how often.
    pub tables: BTreeMap<String, u64>,
    /// Job types enqueued, and how often.
    pub job_types: BTreeMap<String, u64>,
    /// Hosts, tables and job types the plugin reached for and did not get.
    pub refused_targets: BTreeMap<String, u64>,
    /// The plugin this summarises and the window it covers.
    ///
    /// Carried on the summary rather than passed again to the renderer: taking
    /// them twice let a caller print a header naming a window the counts were
    /// never taken over, which is a report that lies about its own scope.
    pub plugin: String,
    /// The window the counts were taken over.
    pub window: Duration,
    /// Calls that happened and are not counted above, because a per-request
    /// ledger filled up.
    ///
    /// A plugin sets its own quotas — they are in its manifest — so a ledger
    /// ceiling is reachable by a plugin that wants to reach it. Non-zero here
    /// means every count above is a **floor**, and an operator must not have to
    /// infer that from a suspiciously round number.
    pub dropped: u64,
    /// True if the log shed drop counts in this window.
    ///
    /// The shed counts belong to any plugin. Then every number here is a floor.
    pub truncated: bool,
}

impl ActivitySummary {
    fn record(&mut self, event: &CapabilityEvent) {
        let bucket = match event.outcome {
            CapabilityOutcome::Allowed => &mut self.allowed,
            CapabilityOutcome::QuotaExceeded => &mut self.quota_hits,
            CapabilityOutcome::Denied(_) => &mut self.denied,
        };
        let counter = bucket.entry(event.operation).or_default();
        *counter = counter.saturating_add(1);

        // The by-target breakdowns cover the three surfaces an operator asks
        // about by name. KV keys are deliberately absent: a plugin's key space
        // is chosen by the plugin and is unbounded, so a per-key breakdown would
        // be both useless and the log's own growth channel. The per-operation
        // counts above already answer "how much KV".
        // "Reached" rather than "allowed": a call whose answer was discarded
        // for being over a byte ceiling, or whose backend errored, still *left
        // the host*. Filing those under "refused" would make "hosts called"
        // undercount exactly the calls an operator most wants to see.
        let allowed = matches!(
            event.outcome,
            CapabilityOutcome::Allowed
                | CapabilityOutcome::Denied(
                    DenialReason::ResponseTooLarge
                        | DenialReason::ResultRejected
                        | DenialReason::BackendError
                )
        );
        // The call ran and the host refused its answer.
        let ran = matches!(
            event.outcome,
            CapabilityOutcome::Allowed
                | CapabilityOutcome::Denied(
                    DenialReason::ResponseTooLarge | DenialReason::ResultRejected
                )
        );
        // Jobs are the exception to "reached the host counts": `job_types` is
        // rendered as *jobs enqueued*, and a sink that refused — a queue at its
        // depth ceiling — enqueued nothing. Counting it would put a job in the
        // report that no runner will ever run, which is the one number an
        // operator reading this would act on.
        // A rejected result still ran: the queue took the job.
        let enqueued = ran;
        // Tables are the other exception: `StoreError::Backend` means the read
        // or write did not happen, so only a call that ran is a table touched.
        let touched = ran;
        let bucket = match (event.capability, allowed) {
            (SandboxCapability::HttpOutbound, true) => Some(&mut self.hosts),
            (SandboxCapability::Db, true) if touched => Some(&mut self.tables),
            (SandboxCapability::Jobs, true) if enqueued => Some(&mut self.job_types),
            // Every remaining scoped call — refused outright, or a `jobs` call
            // that reached a sink which would not take it — is a target the
            // plugin *named* and did not get.
            (
                SandboxCapability::HttpOutbound | SandboxCapability::Db | SandboxCapability::Jobs,
                _,
            ) => Some(&mut self.refused_targets),
            _ => None,
        };
        if let Some(bucket) = bucket {
            let counter = bucket.entry(event.target.clone()).or_default();
            *counter = counter.saturating_add(1);
        }
    }

    /// Whether nothing at all was recorded.
    ///
    /// Says nothing about [`dropped`](Self::dropped): an empty summary with a
    /// non-zero drop count is a plugin that was busy, not one that was idle.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.allowed.is_empty() && self.denied.is_empty() && self.quota_hits.is_empty()
    }

    /// One operator-readable block.
    ///
    /// `Display` rather than a `render(plugin, window)` method: the summary
    /// already knows both, and taking them again is how a header comes to name
    /// a window the counts were not taken over.
    fn write_report(&self, out: &mut String) {
        use std::fmt::Write as _;

        // `write!` to a `String` is infallible; results are dropped rather than
        // unwrapped so this stays panic-free by construction.
        let _ = writeln!(
            out,
            "sandboxed plugin `{plugin}` — last {secs}s",
            plugin = self.plugin,
            secs = self.window.as_secs()
        );
        if self.is_empty() && self.dropped == 0 && !self.truncated {
            out.push_str("  no capability calls\n");
            return;
        }
        if self.truncated {
            out.push_str(
                "  \u{26A0} the log shed older drop counts. Every number here is a floor\n",
            );
        }
        if self.dropped > 0 {
            let _ = writeln!(
                out,
                "  \u{26A0} {dropped} further call(s) were made and are NOT counted below: \
                 every number here is a floor",
                dropped = self.dropped
            );
        }
        for (label, bucket) in [
            ("performed", &self.allowed),
            ("denied", &self.denied),
            ("over quota", &self.quota_hits),
        ] {
            if bucket.is_empty() {
                continue;
            }
            let _ = writeln!(out, "  {label}:");
            for (operation, count) in bucket {
                let _ = writeln!(out, "    {operation} × {count}");
            }
        }
        for (label, bucket) in [
            ("hosts called", &self.hosts),
            ("tables touched", &self.tables),
            ("jobs enqueued", &self.job_types),
            ("targets refused", &self.refused_targets),
        ] {
            if bucket.is_empty() {
                continue;
            }
            let _ = writeln!(out, "  {label}:");
            for (target, count) in bucket {
                let _ = writeln!(out, "    {target} × {count}");
            }
        }
    }
}

impl fmt::Display for ActivitySummary {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut out = String::new();
        self.write_report(&mut out);
        f.write_str(&out)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::plugin_sandbox::manifest::SandboxCapability;

    fn event(operation: &'static str, outcome: CapabilityOutcome) -> CapabilityEvent {
        CapabilityEvent {
            capability: SandboxCapability::Kv,
            operation,
            target: "cart".to_owned(),
            outcome,
            at: crate::time::ambient_instant(),
        }
    }

    #[test]
    fn a_log_from_a_nested_timeline_ages_out_on_the_outer_one() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let log = PluginActivityLog::new();
        let outer = TickingClock::starting_at(epoch);
        let _outer = install_ambient(Arc::new(outer.clone()));

        // Two hours into a nested timeline, the plugin calls and drops calls.
        let inner = TickingClock::starting_at(epoch);
        let guard = install_ambient(Arc::new(inner.clone()));
        inner.advance(2 * hour);
        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
        log.ingest_dropped("shop", 3);
        drop(guard);

        // The outer timeline goes on from there: those calls were just now.
        let summary = log.summary("shop", hour);
        assert_eq!(summary.allowed.get("kv-get").copied(), Some(1));
        assert_eq!(summary.dropped, 3);

        // An hour and a second later they are out of the last hour.
        outer.advance(hour + Duration::from_secs(1));
        let summary = log.summary("shop", hour);
        assert_eq!(summary.allowed.get("kv-get"), None);
        assert_eq!(summary.dropped, 0);
    }

    #[test]
    fn a_request_s_drops_count_while_its_last_drop_is_in_the_window() {
        // #2523 finding 18. The span of one request's drops is known, not the
        // time of each drop. Key the record at the last drop: never under-report.
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let minute = Duration::from_secs(60);
        let clock = TickingClock::starting_at(
            chrono::Utc
                .with_ymd_and_hms(2020, 1, 1, 0, 0, 0)
                .single()
                .expect("a valid date"),
        );
        let _clock = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();
        let first = crate::time::ambient_instant();
        clock.advance(90 * minute);
        let last = crate::time::ambient_instant();
        clock.advance(30 * minute);
        log.ingest_dropped_calls(
            "shop",
            DroppedCalls {
                count: 4,
                first,
                last,
            },
        );
        // The request ended now, but its last drop was 30 minutes ago.
        assert_eq!(log.summary("shop", 40 * minute).dropped, 4);
        assert_eq!(log.summary("shop", 20 * minute).dropped, 0);
        // A drop on the eviction path later still makes a record of its own.
        log.ingest_dropped("shop", 1);
        assert_eq!(log.summary("shop", 20 * minute).dropped, 1);
        assert_eq!(log.summary("shop", 40 * minute).dropped, 5);
    }

    #[test]
    fn the_ring_evicts_the_oldest_rather_than_growing() {
        let log = PluginActivityLog::new();
        log.ingest(
            "shop",
            (0..MAX_LOG_EVENTS + 100).map(|_| event("kv-get", CapabilityOutcome::Allowed)),
        );
        let summary = log.summary("shop", Duration::from_secs(3600));
        assert_eq!(
            summary.allowed.get("kv-get").copied(),
            Some(MAX_LOG_EVENTS as u64),
            "the log holds its ceiling and no more"
        );
    }

    #[test]
    fn a_summary_covers_one_plugin_and_one_window() {
        let log = PluginActivityLog::new();
        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
        log.ingest("other", [event("kv-set", CapabilityOutcome::Allowed)]);

        let shop = log.summary("shop", Duration::from_secs(3600));
        assert_eq!(shop.allowed.get("kv-get").copied(), Some(1));
        assert!(!shop.allowed.contains_key("kv-set"), "{shop:?}");

        // A zero-length window covers nothing, which is the same filter the
        // "last hour" in the acceptance criterion rests on.
        assert!(log.summary("shop", Duration::ZERO).is_empty());
        assert!(log.summary("nobody", Duration::from_secs(3600)).is_empty());
        assert_eq!(log.plugins(), vec!["other".to_owned(), "shop".to_owned()]);
    }

    #[test]
    fn a_truncated_summary_says_so_before_it_says_anything_else() {
        let log = PluginActivityLog::new();
        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
        log.ingest_dropped("shop", 0);
        assert_eq!(log.summary("shop", Duration::from_secs(3600)).dropped, 0);

        log.ingest_dropped("shop", 7);
        log.ingest_dropped("shop", 5);
        let summary = log.summary("shop", Duration::from_secs(3600));
        assert_eq!(summary.dropped, 12);
        let rendered = summary.to_string();
        assert!(rendered.contains("floor"), "{rendered}");
        // The warning is above the counts, because an operator who reads the
        // numbers first has already been misled.
        let warn = rendered.find("floor").unwrap_or(usize::MAX);
        let count = rendered.find("kv-get").unwrap_or(0);
        assert!(warn < count, "{rendered}");
    }

    #[test]
    fn a_summary_names_the_window_it_was_taken_over() {
        let log = PluginActivityLog::new();
        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
        let rendered = log.summary("shop", Duration::from_secs(60)).to_string();
        assert!(rendered.contains("last 60s"), "{rendered}");
        assert!(rendered.contains("`shop`"), "{rendered}");
    }

    #[test]
    fn an_event_the_ring_evicted_is_reported_as_dropped_rather_than_forgotten() {
        // The ring holds the recent end, which is right. What was wrong was
        // saying nothing about the rest: a plugin busy enough to fill 4,096
        // entries in twenty seconds got a "last hour" summary showing twenty
        // seconds, with counts presented as the hour's.
        let log = PluginActivityLog::new();
        for _ in 0..MAX_LOG_EVENTS + 100 {
            log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
        }
        let summary = log.summary("shop", Duration::from_secs(3600));
        assert_eq!(
            summary.allowed.get("kv-get").copied(),
            Some(MAX_LOG_EVENTS as u64),
            "the ring still holds its ceiling"
        );
        assert_eq!(summary.dropped, 100, "and says what it could not hold");
        assert!(
            summary.to_string().contains("NOT counted below"),
            "{summary}"
        );
    }

    #[test]
    fn a_plugin_visible_only_as_drops_is_still_enumerated() {
        // The ring is shared, so a busy plugin evicts a quiet one's every
        // event while the drop ledger keeps the count. Enumerating only the
        // ring then hides the quiet plugin from any operator surface that
        // walks `plugins()` asking each for a summary — even though
        // `summary("quiet", …)` has something to say about it.
        let log = PluginActivityLog::new();
        log.ingest_dropped("quiet", 7);
        for _ in 0..MAX_LOG_EVENTS {
            log.ingest("busy", [event("kv-get", CapabilityOutcome::Allowed)]);
        }

        let names = log.plugins();
        assert!(
            names.contains(&"quiet".to_owned()),
            "a plugin the log can summarise must be one the log can name: {names:?}"
        );
        assert!(names.contains(&"busy".to_owned()), "{names:?}");
        assert_eq!(
            log.summary("quiet", Duration::from_secs(3600)).dropped,
            7,
            "and the summary it was omitted from is not empty"
        );
    }

    #[test]
    fn a_dropped_count_leaves_the_window_with_the_calls_it_describes() {
        // A lifetime total attached itself to every later summary. An operator
        // asking about a window that ended before the overflow was told the
        // window was incomplete — permanently, and about calls that were not in
        // it.
        let log = PluginActivityLog::new();
        log.ingest_dropped("shop", 9);
        assert_eq!(log.summary("shop", Duration::from_secs(3600)).dropped, 9);
        assert_eq!(
            log.summary("shop", Duration::ZERO).dropped,
            0,
            "a window covering nothing covers no drops either"
        );
        assert_eq!(
            log.summary("other", Duration::from_secs(3600)).dropped,
            0,
            "and a drop belongs to the plugin that made it"
        );
    }

    #[test]
    fn the_dropped_ledger_is_itself_bounded() {
        // It reports unbounded growth, so it must not be the growth. Coalescing
        // is what keeps one record per plugin per bucket rather than one per
        // evicted event.
        let log = PluginActivityLog::new();
        for _ in 0..MAX_LOG_EVENTS * 3 {
            log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
        }
        let held = log
            .state
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .dropped
            .len();
        assert!(held <= MAX_LOG_EVENTS, "the dropped ring grew to {held}");
        assert!(
            log.summary("shop", Duration::from_secs(3600)).dropped >= 1,
            "and still reports the drops"
        );
    }

    #[test]
    fn a_refused_enqueue_is_not_reported_as_a_job() {
        // `job_types` renders as "jobs enqueued". A sink at its depth ceiling
        // enqueued nothing, so counting it puts a job in the operator's report
        // that no runner will ever run — and that count is the number they
        // would act on.
        let log = PluginActivityLog::new();
        log.ingest(
            "shop",
            [
                CapabilityEvent {
                    capability: SandboxCapability::Jobs,
                    operation: "job-enqueue",
                    target: "reindex".to_owned(),
                    outcome: CapabilityOutcome::Allowed,
                    at: crate::time::ambient_instant(),
                },
                CapabilityEvent {
                    capability: SandboxCapability::Jobs,
                    operation: "job-enqueue",
                    target: "reindex".to_owned(),
                    outcome: CapabilityOutcome::Denied(DenialReason::BackendError),
                    at: crate::time::ambient_instant(),
                },
            ],
        );
        let summary = log.summary("shop", Duration::from_secs(3600));
        assert_eq!(
            summary.job_types.get("reindex").copied(),
            Some(1),
            "only the accepted enqueue is a job: {summary:?}"
        );
        assert_eq!(
            summary.refused_targets.get("reindex").copied(),
            Some(1),
            "and the refusal is still visible: {summary:?}"
        );
    }
    fn db_event(outcome: CapabilityOutcome) -> CapabilityEvent {
        CapabilityEvent {
            capability: SandboxCapability::Db,
            operation: "db-insert",
            target: "orders".to_owned(),
            outcome,
            at: crate::time::ambient_instant(),
        }
    }

    #[test]
    fn a_failed_db_call_is_not_a_table_touched() {
        // `StoreError::Backend` means the read or write did not happen.
        let log = PluginActivityLog::new();
        log.ingest(
            "shop",
            [
                db_event(CapabilityOutcome::Allowed),
                db_event(CapabilityOutcome::Denied(DenialReason::BackendError)),
            ],
        );
        let summary = log.summary("shop", Duration::from_secs(3600));
        assert_eq!(
            summary.tables.get("orders").copied(),
            Some(1),
            "{summary:?}"
        );
        assert_eq!(
            summary.refused_targets.get("orders").copied(),
            Some(1),
            "{summary:?}"
        );
    }

    #[test]
    fn an_event_keeps_the_time_it_happened() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();

        // A slow request: the call is made, two hours pass, the log ingests.
        let early = event("kv-get", CapabilityOutcome::Allowed);
        clock.advance(2 * hour);
        let late = event("kv-set", CapabilityOutcome::Allowed);
        log.ingest("shop", [early, late]);

        let summary = log.summary("shop", hour);
        assert_eq!(summary.allowed.get("kv-get"), None, "{summary:?}");
        assert_eq!(summary.allowed.get("kv-set").copied(), Some(1));
    }

    #[test]
    fn a_full_dropped_ledger_keeps_every_count() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();

        let records = MAX_LOG_EVENTS + 50;
        for _ in 0..records {
            log.ingest_dropped("shop", 1);
            // Past the coalescing span, so each is its own record.
            clock.advance(Duration::from_secs(2));
        }
        let summary = log.summary("shop", Duration::from_secs(86_400));
        assert_eq!(summary.dropped, records as u64);
    }

    #[test]
    fn the_dropped_ledger_evicts_the_oldest_time_not_the_oldest_insert() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();

        // Old events fill the event ring, then a long time passes.
        let old: Vec<_> = (0..MAX_LOG_EVENTS)
            .map(|_| event("kv-get", CapabilityOutcome::Allowed))
            .collect();
        clock.advance(10 * hour);
        // The ledger fills with recent records.
        for _ in 0..MAX_LOG_EVENTS {
            log.ingest_dropped("shop", 1);
            clock.advance(Duration::from_secs(2));
        }
        // Ingesting the old events, then new ones, evicts old events: those
        // drops carry old times and land after the recent records.
        log.ingest("shop", old);
        log.ingest(
            "shop",
            (0..10).map(|_| event("kv-set", CapabilityOutcome::Allowed)),
        );
        log.ingest_dropped("shop", 1);

        // Window: only the recent records. All 4096 + 1 are in it; the evicted
        // old events are not.
        // The recent window holds every recent record and nothing old.
        let recent = log.summary("shop", Duration::from_secs(3 * 3600));
        assert_eq!(recent.dropped, MAX_LOG_EVENTS as u64 + 1);
        // A wider window adds the old events the ring evicted.
        let wide = log.summary("shop", 24 * hour);
        assert!(wide.dropped > recent.dropped, "{}", wide.dropped);
    }

    #[test]
    fn a_steady_dropper_does_not_stretch_one_record_across_hours() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        // One drop every 300 ms for five hours. Each lands inside the bucket
        // of the previous one, so coalescing never opens a new record. The
        // record must still age: its time may not follow the newest drop.
        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();
        for _ in 0..60_000 {
            log.ingest_dropped("shop", 1);
            clock.advance(Duration::from_millis(300));
        }
        let hour = log.summary("shop", Duration::from_secs(3600)).dropped;
        // 12,000 drops fit in the hour; one bucket of slack, never hours.
        assert!(hour <= 12_000 + 4, "the last hour reported {hour} drops");
        assert!(hour >= 12_000, "the last hour lost drops: {hour}");
        let all = log.summary("shop", Duration::from_secs(6 * 3600)).dropped;
        assert_eq!(all, 60_000);
    }

    #[test]
    fn a_summary_never_sees_a_call_in_neither_ledger() {
        use std::sync::Arc;

        let log = Arc::new(PluginActivityLog::new());
        log.ingest(
            "shop",
            (0..MAX_LOG_EVENTS).map(|_| event("kv-get", CapabilityOutcome::Allowed)),
        );
        let writer = {
            let log = Arc::clone(&log);
            std::thread::spawn(move || {
                for _ in 0..20_000 {
                    log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
                }
            })
        };
        let mut last = 0_u64;
        for _ in 0..2_000 {
            let summary = log.summary("shop", Duration::from_secs(3600));
            let seen = summary.allowed.get("kv-get").copied().unwrap_or(0) + summary.dropped;
            assert!(
                seen >= last,
                "a call vanished mid-transfer: {seen} < {last}"
            );
            last = seen;
        }
        writer.join().expect("the writer finishes");
    }
    #[test]
    fn the_ring_evicts_the_oldest_event_by_time_not_by_arrival() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();

        // A slow request made its calls long ago and finishes last.
        let old: Vec<_> = (0..MAX_LOG_EVENTS)
            .map(|_| event("kv-get", CapabilityOutcome::Allowed))
            .collect();
        clock.advance(10 * hour);
        log.ingest("shop", [event("kv-set", CapabilityOutcome::Allowed)]);
        log.ingest("shop", old);
        // One more event evicts. The oldest by time must go, not the newest.
        log.ingest("shop", [event("kv-delete", CapabilityOutcome::Allowed)]);

        let recent = log.summary("shop", hour);
        assert_eq!(recent.allowed.get("kv-set").copied(), Some(1), "{recent:?}");
        assert_eq!(recent.allowed.get("kv-delete").copied(), Some(1));
    }

    #[test]
    fn folded_counts_leave_the_window_with_their_hour() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();

        // Two hours of drop records, each its own record, then overflow.
        for _ in 0..MAX_LOG_EVENTS {
            log.ingest_dropped("shop", 1);
            clock.advance(Duration::from_secs(2));
        }
        clock.advance(10 * hour);
        for _ in 0..MAX_LOG_EVENTS {
            log.ingest_dropped("shop", 1);
            clock.advance(Duration::from_secs(2));
        }
        // The first batch folded away, ten hours back. It is not in the last
        // two hours, though the second batch is.
        let recent = log.summary("shop", 3 * hour);
        assert_eq!(recent.dropped, MAX_LOG_EVENTS as u64, "{}", recent.dropped);
        let all = log.summary("shop", 100 * hour);
        assert_eq!(all.dropped, 2 * MAX_LOG_EVENTS as u64);
    }
    #[test]
    fn a_fold_from_before_the_first_overflow_keeps_its_own_hour() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();

        // Old events, then ten hours pass, then the ledger overflows with
        // recent records, then the old events are evicted.
        let old: Vec<_> = (0..MAX_LOG_EVENTS)
            .map(|_| event("kv-get", CapabilityOutcome::Allowed))
            .collect();
        log.ingest("shop", old);
        clock.advance(10 * hour);
        for _ in 0..MAX_LOG_EVENTS + 5 {
            log.ingest_dropped("shop", 1);
            clock.advance(Duration::from_secs(2));
        }
        log.ingest(
            "shop",
            (0..5).map(|_| event("kv-set", CapabilityOutcome::Allowed)),
        );
        // The ten-hour-old evictions are not in the last three hours.
        let recent = log.summary("shop", 3 * hour);
        assert_eq!(
            recent.dropped,
            MAX_LOG_EVENTS as u64 + 5,
            "{}",
            recent.dropped
        );
    }

    #[test]
    fn a_summary_says_so_when_the_log_lost_drop_counts() {
        let log = PluginActivityLog::new();
        let hour = Duration::from_secs(3600);
        // This many plugins overflow the dropped ledger. Then the folded
        // ledger overflows and sheds a count.
        for n in 0..(MAX_LOG_EVENTS * 2 + 10) {
            log.ingest_dropped(&format!("p{n}"), 1);
        }
        assert!(
            log.summary("p0", hour).truncated,
            "counts were lost, and the summary must say so"
        );
        assert!(
            !PluginActivityLog::new().summary("p0", hour).truncated,
            "a log that lost nothing is not truncated"
        );
    }

    #[test]
    fn a_truncation_leaves_the_window_with_its_time() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();
        for n in 0..(MAX_LOG_EVENTS * 2 + 10) {
            log.ingest_dropped(&format!("p{n}"), 1);
        }
        clock.advance(10 * hour);
        assert!(!log.summary("p0", hour).truncated, "the loss is old");
        assert!(log.summary("p0", 11 * hour).truncated);
    }

    #[test]
    fn truncation_is_exact_at_the_window_edge_and_reaches_every_plugin() {
        use std::sync::Arc;

        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let hour = Duration::from_secs(3600);
        let clock = TickingClock::starting_at(epoch);
        let _guard = install_ambient(Arc::new(clock.clone()));
        let log = PluginActivityLog::new();
        for n in 0..(MAX_LOG_EVENTS * 2 + 10) {
            log.ingest_dropped(&format!("p{n}"), 1);
        }
        let shed = log.lock().shed.expect("a count was shed");
        clock.advance(hour);
        let age = crate::time::ambient_instant().saturating_duration_since(shed);
        assert!(log.summary("other", age).truncated, "at the edge");
        assert!(
            !log.summary("other", age.saturating_sub(Duration::from_millis(1)))
                .truncated,
            "just outside"
        );
    }

    #[test]
    fn a_truncated_summary_is_not_reported_as_empty() {
        let summary = ActivitySummary {
            plugin: "shop".to_owned(),
            truncated: true,
            ..ActivitySummary::default()
        };
        let text = summary.to_string();
        assert!(text.contains("shed older drop counts"), "{text}");
        assert!(!text.contains("no capability calls"), "{text}");
    }
}