Skip to main content

autumn_web/plugin_sandbox/capability/
audit.rs

1//! "What did this plugin do in the last hour?" (issue #1632).
2//!
3//! An operator who granted five capabilities to an artifact they did not audit
4//! needs one place to answer that. Not five: a cache metric, an upstream's
5//! access log, a job table and a `grep` of the application log is four places
6//! and no answer, because none of them knows which plugin was responsible.
7//!
8//! So every capability call — allowed, denied or over quota — is recorded at the
9//! one point that knows all of it — `CapabilityRuntime::record`, the single exit
10//! every `dispatch` takes — and the records aggregate here:
11//!
12//! ```text
13//!   request ──► CapabilityEvent × n ──► SandboxOutcome.activity
14//!                                            │
15//!                                            ▼
16//!                                    PluginActivityLog  ──► ActivitySummary
17//!                                    (bounded, windowed)     hosts called,
18//!                                                            kv/db/job counts,
19//!                                                            denials, quota hits
20//! ```
21//!
22//! # What a record carries, and what it deliberately does not
23//!
24//! A record names the capability, the operation, and the *logical* target — the
25//! key, the table, the hostname, the job type. It never carries a value, a row,
26//! a request body or a response. An audit surface that logged what a plugin
27//! stored would be a second copy of the tenant data the capability system exists
28//! to contain, in a place with different access rules — and an operator asking
29//! "what did it do" is asking about shape, not contents.
30
31// autumn-determinism-gate: production code in this module must read time and
32// mint identifiers through the framework's injected seams (ClockSource /
33// Entropy), never `Instant::now()` / `Utc::now()` / `SystemTime::now()` /
34// `Uuid::new_v4()` directly. See CONTRIBUTING.md "Determinism seam gate"
35// (issue #1797). Justify exceptions with
36// #[allow(clippy::disallowed_methods, reason = "…")] at the narrowest scope.
37#![cfg_attr(not(test), deny(clippy::disallowed_methods))]
38
39use std::collections::BTreeMap;
40use std::fmt;
41use std::sync::{Mutex, PoisonError};
42use std::time::{Duration, Instant};
43
44use serde::Serialize;
45
46use super::super::manifest::SandboxCapability;
47use super::DenialReason;
48
49/// How one capability call ended.
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
51#[serde(rename_all = "kebab-case")]
52#[non_exhaustive]
53pub enum CapabilityOutcome {
54    /// The call was performed.
55    Allowed,
56    /// A quota refused it.
57    QuotaExceeded,
58    /// A grant refused it.
59    Denied(DenialReason),
60}
61
62impl fmt::Display for CapabilityOutcome {
63    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
64        match self {
65            Self::Allowed => f.write_str("allowed"),
66            Self::QuotaExceeded => f.write_str("quota-exceeded"),
67            Self::Denied(reason) => write!(f, "denied:{reason}"),
68        }
69    }
70}
71
72/// One thing a plugin did, or was stopped from doing.
73#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
74#[non_exhaustive]
75pub struct CapabilityEvent {
76    /// Which capability the call needed.
77    pub capability: SandboxCapability,
78    /// The operation, as the wire spells it (`kv-get`, `db-query`, …).
79    pub operation: &'static str,
80    /// The logical thing it named: a key, a table, a hostname, a job type.
81    pub target: String,
82    /// How it ended.
83    pub outcome: CapabilityOutcome,
84    /// When the call ended, on the ambient clock.
85    ///
86    /// Taken where the call is recorded, not where the log ingests it. A request
87    /// can run for a long time, and stamping its events when it finishes made
88    /// its earliest calls look recent.
89    #[serde(skip)]
90    pub at: Instant,
91}
92
93/// Calls one request made that its ledger could not hold, and when.
94///
95/// It keeps only the first and last times. The ledger had no room for the time
96/// of each call.
97#[derive(Debug, Clone, Copy, PartialEq, Eq)]
98#[non_exhaustive]
99pub struct DroppedCalls {
100    /// How many calls.
101    pub count: u64,
102    /// The time of the first dropped call.
103    pub first: Instant,
104    /// The time of the last dropped call.
105    pub last: Instant,
106}
107
108/// The most events one plugin's log holds.
109///
110/// Bounded, because the log is fed by an untrusted plugin's call rate and an
111/// unbounded one would be that plugin's memory-exhaustion channel — the same
112/// argument the per-request ledger's ceiling rests on, one level up. Oldest
113/// entries are dropped first: an operator asking "in the last hour" wants the
114/// recent end.
115pub const MAX_LOG_EVENTS: usize = 4096;
116
117/// A bounded, time-windowed record of what each plugin has been doing.
118///
119/// One log per application, keyed by plugin, because the question is asked per
120/// plugin but the ceiling has to hold across all of them.
121#[derive(Debug, Default)]
122pub struct PluginActivityLog {
123    /// Both ledgers under one lock. A summary then never sees an event that has
124    /// left the ring and not yet reached the drop ledger. The work under the
125    /// lock is constant time per event.
126    state: Mutex<LogState>,
127}
128
129/// What [`PluginActivityLog`] guards.
130#[derive(Debug, Default)]
131struct LogState {
132    /// Keyed by the event's own time, so the oldest event is the first key
133    /// whatever order requests finished in. The sequence number separates equal
134    /// times.
135    entries: BTreeMap<(Instant, u64), (String, CapabilityEvent)>,
136    /// Calls the log knows happened but cannot show: those a per-request
137    /// ledger could not hold, and those the ring evicted.
138    ///
139    /// Keyed by time, so the oldest record is the first key whatever order the
140    /// records arrived in. The sequence number keeps equal times apart. Windowed
141    /// by the same cutoff as `entries`.
142    dropped: BTreeMap<(Instant, u64), DroppedRecord>,
143    /// The next sequence number for `dropped`.
144    seq: u64,
145    /// Counts the full `dropped` ledger folded away, by plugin and hour, with
146    /// the newest time folded into each bucket. A summary reports a bucket
147    /// while that time is in its window. It can over-report by under one hour
148    /// plus the longest drop span of one request. It never under-reports.
149    folded: BTreeMap<(String, u64), (Instant, u64)>,
150    /// The time the hour buckets count from. Set by the first fold.
151    origin: Option<Instant>,
152    /// The newest time of a drop count the full `folded` ledger shed. Its
153    /// owner is unknown, so a summary can only say that counts were lost.
154    shed: Option<Instant>,
155}
156
157/// The gap between two instants, whichever is later.
158fn apart(a: Instant, b: Instant) -> Duration {
159    a.max(b).saturating_duration_since(a.min(b))
160}
161
162/// Calls the log knows happened but cannot show, for one plugin.
163#[derive(Debug)]
164struct DroppedRecord {
165    plugin: String,
166    count: u64,
167    /// The time of the first drop. A record spans at most [`DROPPED_BUCKET`]
168    /// from here, or the span of one request's drops if that is longer.
169    first: Instant,
170}
171
172/// How long one `folded` bucket spans.
173const FOLD_BUCKET_SECS: u64 = 3600;
174
175/// How coarsely dropped-call records are timestamped.
176///
177/// One eviction is one dropped call, and evictions arrive at the plugin's whole
178/// call rate — so recording each separately would cost as much memory as
179/// keeping the events did. Drops for one plugin become one record while they
180/// stay inside this span of the record's first drop. The span is far below the
181/// "last hour" window, so the windowing stays honest.
182const DROPPED_BUCKET: Duration = Duration::from_secs(1);
183
184impl PluginActivityLog {
185    /// An empty log.
186    #[must_use]
187    pub fn new() -> Self {
188        // The fold buckets count from here. No event can predate its log.
189        let state = LogState {
190            origin: Some(crate::time::ambient_instant()),
191            ..LogState::default()
192        };
193        Self {
194            state: Mutex::new(state),
195        }
196    }
197
198    fn lock(&self) -> std::sync::MutexGuard<'_, LogState> {
199        self.state.lock().unwrap_or_else(PoisonError::into_inner)
200    }
201
202    /// Record everything one request's runtime gathered.
203    ///
204    /// Each event keeps the time it carries, not the time of this call.
205    pub fn ingest(&self, plugin: &str, events: impl IntoIterator<Item = CapabilityEvent>) {
206        let mut state = self.lock();
207        for event in events {
208            if state.entries.len() >= MAX_LOG_EVENTS
209                && let Some(((at, _), (who, _))) = state.entries.pop_first()
210            {
211                // An evicted event is a call the summary can no longer show.
212                // It is recorded under its own time, so a drop that has aged
213                // out of the window stops being reported with it.
214                state.note_dropped(at, at, &who, 1);
215            }
216            state.seq = state.seq.wrapping_add(1);
217            let key = (event.at, state.seq);
218            state.entries.insert(key, (plugin.to_owned(), event));
219        }
220    }
221
222    /// Record that one request made more calls than its ledger could hold.
223    ///
224    /// Separate from [`ingest`](Self::ingest) because the number is not an
225    /// event: there is nothing to say about the calls beyond that they happened
226    /// and are not below. An operator reading a summary needs to know that
227    /// before they read the counts.
228    ///
229    /// Records the drops at the current time. Use
230    /// [`ingest_dropped_calls`](Self::ingest_dropped_calls) when you know the
231    /// times of the drops.
232    pub fn ingest_dropped(&self, plugin: &str, dropped: u64) {
233        if dropped == 0 {
234            return;
235        }
236        let now = crate::time::ambient_instant();
237        self.lock().note_dropped(now, now, plugin, dropped);
238    }
239
240    /// Record the calls one request's ledger could not hold, at their own times.
241    ///
242    /// A slow request ends long after its drops. This method records the drops
243    /// at their own times, not at the time of ingest (#2523). The record counts
244    /// while `calls.last` is in the window. It can over-report by the span of
245    /// the drops. It never under-reports.
246    pub fn ingest_dropped_calls(&self, plugin: &str, calls: DroppedCalls) {
247        if calls.count == 0 {
248            return;
249        }
250        // Ordered, so the key is never before a drop it counts.
251        let (first, last) = (calls.first.min(calls.last), calls.first.max(calls.last));
252        self.lock().note_dropped(first, last, plugin, calls.count);
253    }
254
255    /// What `plugin` did within `window`.
256    #[must_use]
257    pub fn summary(&self, plugin: &str, window: Duration) -> ActivitySummary {
258        let cutoff = crate::time::ambient_instant();
259        let in_window = |at: Instant| cutoff.saturating_duration_since(at) <= window;
260        let mut summary = ActivitySummary {
261            plugin: plugin.to_owned(),
262            window,
263            ..ActivitySummary::default()
264        };
265        let state = self.lock();
266        for ((at, _), (who, event)) in &state.entries {
267            if who == plugin && in_window(*at) {
268                summary.record(event);
269            }
270        }
271        summary.dropped = state
272            .dropped
273            .iter()
274            .filter(|((at, _), record)| record.plugin == plugin && in_window(*at))
275            .fold(0_u64, |sum, (_, record)| sum.saturating_add(record.count));
276        let folded: u64 = state
277            .folded
278            .range((plugin.to_owned(), 0)..=(plugin.to_owned(), u64::MAX))
279            .filter(|(_, (at, _))| in_window(*at))
280            .fold(0_u64, |sum, (_, (_, count))| sum.saturating_add(*count));
281        summary.dropped = summary.dropped.saturating_add(folded);
282        summary.truncated = state.shed.is_some_and(in_window);
283        drop(state);
284        summary
285    }
286
287    /// Every plugin this log has heard from.
288    ///
289    /// All ledgers, not only the event ring. A busy plugin can evict a quiet
290    /// one's every event while the drop ledger still carries the count of what
291    /// was lost, and an operator surface that enumerates plugins here to ask
292    /// each for a [`summary`] would then skip the one plugin whose activity is
293    /// only visible as drops.
294    ///
295    /// [`summary`]: Self::summary
296    #[must_use]
297    pub fn plugins(&self) -> Vec<String> {
298        let mut names: Vec<String> = {
299            let state = self.lock();
300            state
301                .entries
302                .values()
303                .map(|(who, _)| who.clone())
304                .chain(state.dropped.values().map(|record| record.plugin.clone()))
305                .chain(state.folded.keys().map(|(who, _)| who.clone()))
306                .collect()
307        };
308        names.sort();
309        names.dedup();
310        names
311    }
312}
313
314impl LogState {
315    /// Add `count` dropped calls for `plugin`, made from `first` to `last`.
316    ///
317    /// Coalesces into the newest record when it is the same plugin and within
318    /// one [`DROPPED_BUCKET`] of that record's first drop. Bounded by
319    /// [`MAX_LOG_EVENTS`]: the oldest record, by time, folds into `folded`
320    /// rather than vanishing.
321    fn note_dropped(&mut self, first: Instant, last: Instant, plugin: &str, count: u64) {
322        // Measured from the first drop, not the newest. A steady dropper would
323        // otherwise keep one record open for hours.
324        if let Some((&(newest, seq), record)) = self.dropped.last_key_value()
325            && record.plugin == plugin
326            && apart(first, record.first) < DROPPED_BUCKET
327            && apart(last, record.first) < DROPPED_BUCKET
328            && apart(last, newest) < DROPPED_BUCKET
329        {
330            // Keyed by the newer time: the record counts while any drop in it
331            // is inside the window.
332            if let Some(mut record) = self.dropped.remove(&(newest, seq)) {
333                record.count = record.count.saturating_add(count);
334                record.first = record.first.min(first);
335                self.dropped.insert((newest.max(last), seq), record);
336            }
337            return;
338        }
339        self.seq = self.seq.wrapping_add(1);
340        self.dropped.insert(
341            (last, self.seq),
342            DroppedRecord {
343                plugin: plugin.to_owned(),
344                count,
345                first,
346            },
347        );
348        while self.dropped.len() > MAX_LOG_EVENTS {
349            let Some((
350                (oldest, _),
351                DroppedRecord {
352                    plugin: who, count, ..
353                },
354            )) = self.dropped.pop_first()
355            else {
356                break;
357            };
358            let origin = *self.origin.get_or_insert(oldest);
359            let hour = oldest.saturating_duration_since(origin).as_secs() / FOLD_BUCKET_SECS;
360            let folded = self.folded.entry((who, hour)).or_insert((oldest, 0));
361            folded.0 = folded.0.max(oldest);
362            folded.1 = folded.1.saturating_add(count);
363        }
364        // Bounded like the other ledgers. Past this the oldest hour goes, which
365        // takes thousands of plugin-hours of overflow to reach. Keep the newest
366        // shed time. A summary then says its counts are incomplete.
367        while self.folded.len() > MAX_LOG_EVENTS {
368            let Some(oldest) = self.folded.keys().min_by_key(|(_, hour)| *hour).cloned() else {
369                break;
370            };
371            if let Some((at, _)) = self.folded.remove(&oldest) {
372                self.shed = Some(self.shed.map_or(at, |seen| seen.max(at)));
373            }
374        }
375    }
376}
377
378/// The answer to "what did this plugin do".
379///
380/// Counts and names, never values. See the module header.
381#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
382#[non_exhaustive]
383pub struct ActivitySummary {
384    /// Calls that were performed, by operation.
385    pub allowed: BTreeMap<&'static str, u64>,
386    /// Calls a grant refused, by operation.
387    pub denied: BTreeMap<&'static str, u64>,
388    /// Calls a quota refused, by operation.
389    pub quota_hits: BTreeMap<&'static str, u64>,
390    /// Hostnames actually called, and how often.
391    pub hosts: BTreeMap<String, u64>,
392    /// Tables touched, and how often.
393    pub tables: BTreeMap<String, u64>,
394    /// Job types enqueued, and how often.
395    pub job_types: BTreeMap<String, u64>,
396    /// Hosts, tables and job types the plugin reached for and did not get.
397    pub refused_targets: BTreeMap<String, u64>,
398    /// The plugin this summarises and the window it covers.
399    ///
400    /// Carried on the summary rather than passed again to the renderer: taking
401    /// them twice let a caller print a header naming a window the counts were
402    /// never taken over, which is a report that lies about its own scope.
403    pub plugin: String,
404    /// The window the counts were taken over.
405    pub window: Duration,
406    /// Calls that happened and are not counted above, because a per-request
407    /// ledger filled up.
408    ///
409    /// A plugin sets its own quotas — they are in its manifest — so a ledger
410    /// ceiling is reachable by a plugin that wants to reach it. Non-zero here
411    /// means every count above is a **floor**, and an operator must not have to
412    /// infer that from a suspiciously round number.
413    pub dropped: u64,
414    /// True if the log shed drop counts in this window.
415    ///
416    /// The shed counts belong to any plugin. Then every number here is a floor.
417    pub truncated: bool,
418}
419
420impl ActivitySummary {
421    fn record(&mut self, event: &CapabilityEvent) {
422        let bucket = match event.outcome {
423            CapabilityOutcome::Allowed => &mut self.allowed,
424            CapabilityOutcome::QuotaExceeded => &mut self.quota_hits,
425            CapabilityOutcome::Denied(_) => &mut self.denied,
426        };
427        let counter = bucket.entry(event.operation).or_default();
428        *counter = counter.saturating_add(1);
429
430        // The by-target breakdowns cover the three surfaces an operator asks
431        // about by name. KV keys are deliberately absent: a plugin's key space
432        // is chosen by the plugin and is unbounded, so a per-key breakdown would
433        // be both useless and the log's own growth channel. The per-operation
434        // counts above already answer "how much KV".
435        // "Reached" rather than "allowed": a call whose answer was discarded
436        // for being over a byte ceiling, or whose backend errored, still *left
437        // the host*. Filing those under "refused" would make "hosts called"
438        // undercount exactly the calls an operator most wants to see.
439        let allowed = matches!(
440            event.outcome,
441            CapabilityOutcome::Allowed
442                | CapabilityOutcome::Denied(
443                    DenialReason::ResponseTooLarge
444                        | DenialReason::ResultRejected
445                        | DenialReason::BackendError
446                )
447        );
448        // The call ran and the host refused its answer.
449        let ran = matches!(
450            event.outcome,
451            CapabilityOutcome::Allowed
452                | CapabilityOutcome::Denied(
453                    DenialReason::ResponseTooLarge | DenialReason::ResultRejected
454                )
455        );
456        // Jobs are the exception to "reached the host counts": `job_types` is
457        // rendered as *jobs enqueued*, and a sink that refused — a queue at its
458        // depth ceiling — enqueued nothing. Counting it would put a job in the
459        // report that no runner will ever run, which is the one number an
460        // operator reading this would act on.
461        // A rejected result still ran: the queue took the job.
462        let enqueued = ran;
463        // Tables are the other exception: `StoreError::Backend` means the read
464        // or write did not happen, so only a call that ran is a table touched.
465        let touched = ran;
466        let bucket = match (event.capability, allowed) {
467            (SandboxCapability::HttpOutbound, true) => Some(&mut self.hosts),
468            (SandboxCapability::Db, true) if touched => Some(&mut self.tables),
469            (SandboxCapability::Jobs, true) if enqueued => Some(&mut self.job_types),
470            // Every remaining scoped call — refused outright, or a `jobs` call
471            // that reached a sink which would not take it — is a target the
472            // plugin *named* and did not get.
473            (
474                SandboxCapability::HttpOutbound | SandboxCapability::Db | SandboxCapability::Jobs,
475                _,
476            ) => Some(&mut self.refused_targets),
477            _ => None,
478        };
479        if let Some(bucket) = bucket {
480            let counter = bucket.entry(event.target.clone()).or_default();
481            *counter = counter.saturating_add(1);
482        }
483    }
484
485    /// Whether nothing at all was recorded.
486    ///
487    /// Says nothing about [`dropped`](Self::dropped): an empty summary with a
488    /// non-zero drop count is a plugin that was busy, not one that was idle.
489    #[must_use]
490    pub fn is_empty(&self) -> bool {
491        self.allowed.is_empty() && self.denied.is_empty() && self.quota_hits.is_empty()
492    }
493
494    /// One operator-readable block.
495    ///
496    /// `Display` rather than a `render(plugin, window)` method: the summary
497    /// already knows both, and taking them again is how a header comes to name
498    /// a window the counts were not taken over.
499    fn write_report(&self, out: &mut String) {
500        use std::fmt::Write as _;
501
502        // `write!` to a `String` is infallible; results are dropped rather than
503        // unwrapped so this stays panic-free by construction.
504        let _ = writeln!(
505            out,
506            "sandboxed plugin `{plugin}` — last {secs}s",
507            plugin = self.plugin,
508            secs = self.window.as_secs()
509        );
510        if self.is_empty() && self.dropped == 0 && !self.truncated {
511            out.push_str("  no capability calls\n");
512            return;
513        }
514        if self.truncated {
515            out.push_str(
516                "  \u{26A0} the log shed older drop counts. Every number here is a floor\n",
517            );
518        }
519        if self.dropped > 0 {
520            let _ = writeln!(
521                out,
522                "  \u{26A0} {dropped} further call(s) were made and are NOT counted below: \
523                 every number here is a floor",
524                dropped = self.dropped
525            );
526        }
527        for (label, bucket) in [
528            ("performed", &self.allowed),
529            ("denied", &self.denied),
530            ("over quota", &self.quota_hits),
531        ] {
532            if bucket.is_empty() {
533                continue;
534            }
535            let _ = writeln!(out, "  {label}:");
536            for (operation, count) in bucket {
537                let _ = writeln!(out, "    {operation} × {count}");
538            }
539        }
540        for (label, bucket) in [
541            ("hosts called", &self.hosts),
542            ("tables touched", &self.tables),
543            ("jobs enqueued", &self.job_types),
544            ("targets refused", &self.refused_targets),
545        ] {
546            if bucket.is_empty() {
547                continue;
548            }
549            let _ = writeln!(out, "  {label}:");
550            for (target, count) in bucket {
551                let _ = writeln!(out, "    {target} × {count}");
552            }
553        }
554    }
555}
556
557impl fmt::Display for ActivitySummary {
558    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
559        let mut out = String::new();
560        self.write_report(&mut out);
561        f.write_str(&out)
562    }
563}
564
565#[cfg(test)]
566mod tests {
567    use super::*;
568    use crate::plugin_sandbox::manifest::SandboxCapability;
569
570    fn event(operation: &'static str, outcome: CapabilityOutcome) -> CapabilityEvent {
571        CapabilityEvent {
572            capability: SandboxCapability::Kv,
573            operation,
574            target: "cart".to_owned(),
575            outcome,
576            at: crate::time::ambient_instant(),
577        }
578    }
579
580    #[test]
581    fn a_log_from_a_nested_timeline_ages_out_on_the_outer_one() {
582        use std::sync::Arc;
583
584        use chrono::TimeZone as _;
585
586        use crate::time::{TickingClock, install_ambient};
587
588        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
589        let hour = Duration::from_secs(3600);
590        let log = PluginActivityLog::new();
591        let outer = TickingClock::starting_at(epoch);
592        let _outer = install_ambient(Arc::new(outer.clone()));
593
594        // Two hours into a nested timeline, the plugin calls and drops calls.
595        let inner = TickingClock::starting_at(epoch);
596        let guard = install_ambient(Arc::new(inner.clone()));
597        inner.advance(2 * hour);
598        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
599        log.ingest_dropped("shop", 3);
600        drop(guard);
601
602        // The outer timeline goes on from there: those calls were just now.
603        let summary = log.summary("shop", hour);
604        assert_eq!(summary.allowed.get("kv-get").copied(), Some(1));
605        assert_eq!(summary.dropped, 3);
606
607        // An hour and a second later they are out of the last hour.
608        outer.advance(hour + Duration::from_secs(1));
609        let summary = log.summary("shop", hour);
610        assert_eq!(summary.allowed.get("kv-get"), None);
611        assert_eq!(summary.dropped, 0);
612    }
613
614    #[test]
615    fn a_request_s_drops_count_while_its_last_drop_is_in_the_window() {
616        // #2523 finding 18. The span of one request's drops is known, not the
617        // time of each drop. Key the record at the last drop: never under-report.
618        use std::sync::Arc;
619
620        use chrono::TimeZone as _;
621
622        use crate::time::{TickingClock, install_ambient};
623
624        let minute = Duration::from_secs(60);
625        let clock = TickingClock::starting_at(
626            chrono::Utc
627                .with_ymd_and_hms(2020, 1, 1, 0, 0, 0)
628                .single()
629                .expect("a valid date"),
630        );
631        let _clock = install_ambient(Arc::new(clock.clone()));
632        let log = PluginActivityLog::new();
633        let first = crate::time::ambient_instant();
634        clock.advance(90 * minute);
635        let last = crate::time::ambient_instant();
636        clock.advance(30 * minute);
637        log.ingest_dropped_calls(
638            "shop",
639            DroppedCalls {
640                count: 4,
641                first,
642                last,
643            },
644        );
645        // The request ended now, but its last drop was 30 minutes ago.
646        assert_eq!(log.summary("shop", 40 * minute).dropped, 4);
647        assert_eq!(log.summary("shop", 20 * minute).dropped, 0);
648        // A drop on the eviction path later still makes a record of its own.
649        log.ingest_dropped("shop", 1);
650        assert_eq!(log.summary("shop", 20 * minute).dropped, 1);
651        assert_eq!(log.summary("shop", 40 * minute).dropped, 5);
652    }
653
654    #[test]
655    fn the_ring_evicts_the_oldest_rather_than_growing() {
656        let log = PluginActivityLog::new();
657        log.ingest(
658            "shop",
659            (0..MAX_LOG_EVENTS + 100).map(|_| event("kv-get", CapabilityOutcome::Allowed)),
660        );
661        let summary = log.summary("shop", Duration::from_secs(3600));
662        assert_eq!(
663            summary.allowed.get("kv-get").copied(),
664            Some(MAX_LOG_EVENTS as u64),
665            "the log holds its ceiling and no more"
666        );
667    }
668
669    #[test]
670    fn a_summary_covers_one_plugin_and_one_window() {
671        let log = PluginActivityLog::new();
672        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
673        log.ingest("other", [event("kv-set", CapabilityOutcome::Allowed)]);
674
675        let shop = log.summary("shop", Duration::from_secs(3600));
676        assert_eq!(shop.allowed.get("kv-get").copied(), Some(1));
677        assert!(!shop.allowed.contains_key("kv-set"), "{shop:?}");
678
679        // A zero-length window covers nothing, which is the same filter the
680        // "last hour" in the acceptance criterion rests on.
681        assert!(log.summary("shop", Duration::ZERO).is_empty());
682        assert!(log.summary("nobody", Duration::from_secs(3600)).is_empty());
683        assert_eq!(log.plugins(), vec!["other".to_owned(), "shop".to_owned()]);
684    }
685
686    #[test]
687    fn a_truncated_summary_says_so_before_it_says_anything_else() {
688        let log = PluginActivityLog::new();
689        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
690        log.ingest_dropped("shop", 0);
691        assert_eq!(log.summary("shop", Duration::from_secs(3600)).dropped, 0);
692
693        log.ingest_dropped("shop", 7);
694        log.ingest_dropped("shop", 5);
695        let summary = log.summary("shop", Duration::from_secs(3600));
696        assert_eq!(summary.dropped, 12);
697        let rendered = summary.to_string();
698        assert!(rendered.contains("floor"), "{rendered}");
699        // The warning is above the counts, because an operator who reads the
700        // numbers first has already been misled.
701        let warn = rendered.find("floor").unwrap_or(usize::MAX);
702        let count = rendered.find("kv-get").unwrap_or(0);
703        assert!(warn < count, "{rendered}");
704    }
705
706    #[test]
707    fn a_summary_names_the_window_it_was_taken_over() {
708        let log = PluginActivityLog::new();
709        log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
710        let rendered = log.summary("shop", Duration::from_secs(60)).to_string();
711        assert!(rendered.contains("last 60s"), "{rendered}");
712        assert!(rendered.contains("`shop`"), "{rendered}");
713    }
714
715    #[test]
716    fn an_event_the_ring_evicted_is_reported_as_dropped_rather_than_forgotten() {
717        // The ring holds the recent end, which is right. What was wrong was
718        // saying nothing about the rest: a plugin busy enough to fill 4,096
719        // entries in twenty seconds got a "last hour" summary showing twenty
720        // seconds, with counts presented as the hour's.
721        let log = PluginActivityLog::new();
722        for _ in 0..MAX_LOG_EVENTS + 100 {
723            log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
724        }
725        let summary = log.summary("shop", Duration::from_secs(3600));
726        assert_eq!(
727            summary.allowed.get("kv-get").copied(),
728            Some(MAX_LOG_EVENTS as u64),
729            "the ring still holds its ceiling"
730        );
731        assert_eq!(summary.dropped, 100, "and says what it could not hold");
732        assert!(
733            summary.to_string().contains("NOT counted below"),
734            "{summary}"
735        );
736    }
737
738    #[test]
739    fn a_plugin_visible_only_as_drops_is_still_enumerated() {
740        // The ring is shared, so a busy plugin evicts a quiet one's every
741        // event while the drop ledger keeps the count. Enumerating only the
742        // ring then hides the quiet plugin from any operator surface that
743        // walks `plugins()` asking each for a summary — even though
744        // `summary("quiet", …)` has something to say about it.
745        let log = PluginActivityLog::new();
746        log.ingest_dropped("quiet", 7);
747        for _ in 0..MAX_LOG_EVENTS {
748            log.ingest("busy", [event("kv-get", CapabilityOutcome::Allowed)]);
749        }
750
751        let names = log.plugins();
752        assert!(
753            names.contains(&"quiet".to_owned()),
754            "a plugin the log can summarise must be one the log can name: {names:?}"
755        );
756        assert!(names.contains(&"busy".to_owned()), "{names:?}");
757        assert_eq!(
758            log.summary("quiet", Duration::from_secs(3600)).dropped,
759            7,
760            "and the summary it was omitted from is not empty"
761        );
762    }
763
764    #[test]
765    fn a_dropped_count_leaves_the_window_with_the_calls_it_describes() {
766        // A lifetime total attached itself to every later summary. An operator
767        // asking about a window that ended before the overflow was told the
768        // window was incomplete — permanently, and about calls that were not in
769        // it.
770        let log = PluginActivityLog::new();
771        log.ingest_dropped("shop", 9);
772        assert_eq!(log.summary("shop", Duration::from_secs(3600)).dropped, 9);
773        assert_eq!(
774            log.summary("shop", Duration::ZERO).dropped,
775            0,
776            "a window covering nothing covers no drops either"
777        );
778        assert_eq!(
779            log.summary("other", Duration::from_secs(3600)).dropped,
780            0,
781            "and a drop belongs to the plugin that made it"
782        );
783    }
784
785    #[test]
786    fn the_dropped_ledger_is_itself_bounded() {
787        // It reports unbounded growth, so it must not be the growth. Coalescing
788        // is what keeps one record per plugin per bucket rather than one per
789        // evicted event.
790        let log = PluginActivityLog::new();
791        for _ in 0..MAX_LOG_EVENTS * 3 {
792            log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
793        }
794        let held = log
795            .state
796            .lock()
797            .unwrap_or_else(PoisonError::into_inner)
798            .dropped
799            .len();
800        assert!(held <= MAX_LOG_EVENTS, "the dropped ring grew to {held}");
801        assert!(
802            log.summary("shop", Duration::from_secs(3600)).dropped >= 1,
803            "and still reports the drops"
804        );
805    }
806
807    #[test]
808    fn a_refused_enqueue_is_not_reported_as_a_job() {
809        // `job_types` renders as "jobs enqueued". A sink at its depth ceiling
810        // enqueued nothing, so counting it puts a job in the operator's report
811        // that no runner will ever run — and that count is the number they
812        // would act on.
813        let log = PluginActivityLog::new();
814        log.ingest(
815            "shop",
816            [
817                CapabilityEvent {
818                    capability: SandboxCapability::Jobs,
819                    operation: "job-enqueue",
820                    target: "reindex".to_owned(),
821                    outcome: CapabilityOutcome::Allowed,
822                    at: crate::time::ambient_instant(),
823                },
824                CapabilityEvent {
825                    capability: SandboxCapability::Jobs,
826                    operation: "job-enqueue",
827                    target: "reindex".to_owned(),
828                    outcome: CapabilityOutcome::Denied(DenialReason::BackendError),
829                    at: crate::time::ambient_instant(),
830                },
831            ],
832        );
833        let summary = log.summary("shop", Duration::from_secs(3600));
834        assert_eq!(
835            summary.job_types.get("reindex").copied(),
836            Some(1),
837            "only the accepted enqueue is a job: {summary:?}"
838        );
839        assert_eq!(
840            summary.refused_targets.get("reindex").copied(),
841            Some(1),
842            "and the refusal is still visible: {summary:?}"
843        );
844    }
845    fn db_event(outcome: CapabilityOutcome) -> CapabilityEvent {
846        CapabilityEvent {
847            capability: SandboxCapability::Db,
848            operation: "db-insert",
849            target: "orders".to_owned(),
850            outcome,
851            at: crate::time::ambient_instant(),
852        }
853    }
854
855    #[test]
856    fn a_failed_db_call_is_not_a_table_touched() {
857        // `StoreError::Backend` means the read or write did not happen.
858        let log = PluginActivityLog::new();
859        log.ingest(
860            "shop",
861            [
862                db_event(CapabilityOutcome::Allowed),
863                db_event(CapabilityOutcome::Denied(DenialReason::BackendError)),
864            ],
865        );
866        let summary = log.summary("shop", Duration::from_secs(3600));
867        assert_eq!(
868            summary.tables.get("orders").copied(),
869            Some(1),
870            "{summary:?}"
871        );
872        assert_eq!(
873            summary.refused_targets.get("orders").copied(),
874            Some(1),
875            "{summary:?}"
876        );
877    }
878
879    #[test]
880    fn an_event_keeps_the_time_it_happened() {
881        use std::sync::Arc;
882
883        use chrono::TimeZone as _;
884
885        use crate::time::{TickingClock, install_ambient};
886
887        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
888        let hour = Duration::from_secs(3600);
889        let clock = TickingClock::starting_at(epoch);
890        let _guard = install_ambient(Arc::new(clock.clone()));
891        let log = PluginActivityLog::new();
892
893        // A slow request: the call is made, two hours pass, the log ingests.
894        let early = event("kv-get", CapabilityOutcome::Allowed);
895        clock.advance(2 * hour);
896        let late = event("kv-set", CapabilityOutcome::Allowed);
897        log.ingest("shop", [early, late]);
898
899        let summary = log.summary("shop", hour);
900        assert_eq!(summary.allowed.get("kv-get"), None, "{summary:?}");
901        assert_eq!(summary.allowed.get("kv-set").copied(), Some(1));
902    }
903
904    #[test]
905    fn a_full_dropped_ledger_keeps_every_count() {
906        use std::sync::Arc;
907
908        use chrono::TimeZone as _;
909
910        use crate::time::{TickingClock, install_ambient};
911
912        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
913        let clock = TickingClock::starting_at(epoch);
914        let _guard = install_ambient(Arc::new(clock.clone()));
915        let log = PluginActivityLog::new();
916
917        let records = MAX_LOG_EVENTS + 50;
918        for _ in 0..records {
919            log.ingest_dropped("shop", 1);
920            // Past the coalescing span, so each is its own record.
921            clock.advance(Duration::from_secs(2));
922        }
923        let summary = log.summary("shop", Duration::from_secs(86_400));
924        assert_eq!(summary.dropped, records as u64);
925    }
926
927    #[test]
928    fn the_dropped_ledger_evicts_the_oldest_time_not_the_oldest_insert() {
929        use std::sync::Arc;
930
931        use chrono::TimeZone as _;
932
933        use crate::time::{TickingClock, install_ambient};
934
935        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
936        let hour = Duration::from_secs(3600);
937        let clock = TickingClock::starting_at(epoch);
938        let _guard = install_ambient(Arc::new(clock.clone()));
939        let log = PluginActivityLog::new();
940
941        // Old events fill the event ring, then a long time passes.
942        let old: Vec<_> = (0..MAX_LOG_EVENTS)
943            .map(|_| event("kv-get", CapabilityOutcome::Allowed))
944            .collect();
945        clock.advance(10 * hour);
946        // The ledger fills with recent records.
947        for _ in 0..MAX_LOG_EVENTS {
948            log.ingest_dropped("shop", 1);
949            clock.advance(Duration::from_secs(2));
950        }
951        // Ingesting the old events, then new ones, evicts old events: those
952        // drops carry old times and land after the recent records.
953        log.ingest("shop", old);
954        log.ingest(
955            "shop",
956            (0..10).map(|_| event("kv-set", CapabilityOutcome::Allowed)),
957        );
958        log.ingest_dropped("shop", 1);
959
960        // Window: only the recent records. All 4096 + 1 are in it; the evicted
961        // old events are not.
962        // The recent window holds every recent record and nothing old.
963        let recent = log.summary("shop", Duration::from_secs(3 * 3600));
964        assert_eq!(recent.dropped, MAX_LOG_EVENTS as u64 + 1);
965        // A wider window adds the old events the ring evicted.
966        let wide = log.summary("shop", 24 * hour);
967        assert!(wide.dropped > recent.dropped, "{}", wide.dropped);
968    }
969
970    #[test]
971    fn a_steady_dropper_does_not_stretch_one_record_across_hours() {
972        use std::sync::Arc;
973
974        use chrono::TimeZone as _;
975
976        use crate::time::{TickingClock, install_ambient};
977
978        // One drop every 300 ms for five hours. Each lands inside the bucket
979        // of the previous one, so coalescing never opens a new record. The
980        // record must still age: its time may not follow the newest drop.
981        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
982        let clock = TickingClock::starting_at(epoch);
983        let _guard = install_ambient(Arc::new(clock.clone()));
984        let log = PluginActivityLog::new();
985        for _ in 0..60_000 {
986            log.ingest_dropped("shop", 1);
987            clock.advance(Duration::from_millis(300));
988        }
989        let hour = log.summary("shop", Duration::from_secs(3600)).dropped;
990        // 12,000 drops fit in the hour; one bucket of slack, never hours.
991        assert!(hour <= 12_000 + 4, "the last hour reported {hour} drops");
992        assert!(hour >= 12_000, "the last hour lost drops: {hour}");
993        let all = log.summary("shop", Duration::from_secs(6 * 3600)).dropped;
994        assert_eq!(all, 60_000);
995    }
996
997    #[test]
998    fn a_summary_never_sees_a_call_in_neither_ledger() {
999        use std::sync::Arc;
1000
1001        let log = Arc::new(PluginActivityLog::new());
1002        log.ingest(
1003            "shop",
1004            (0..MAX_LOG_EVENTS).map(|_| event("kv-get", CapabilityOutcome::Allowed)),
1005        );
1006        let writer = {
1007            let log = Arc::clone(&log);
1008            std::thread::spawn(move || {
1009                for _ in 0..20_000 {
1010                    log.ingest("shop", [event("kv-get", CapabilityOutcome::Allowed)]);
1011                }
1012            })
1013        };
1014        let mut last = 0_u64;
1015        for _ in 0..2_000 {
1016            let summary = log.summary("shop", Duration::from_secs(3600));
1017            let seen = summary.allowed.get("kv-get").copied().unwrap_or(0) + summary.dropped;
1018            assert!(
1019                seen >= last,
1020                "a call vanished mid-transfer: {seen} < {last}"
1021            );
1022            last = seen;
1023        }
1024        writer.join().expect("the writer finishes");
1025    }
1026    #[test]
1027    fn the_ring_evicts_the_oldest_event_by_time_not_by_arrival() {
1028        use std::sync::Arc;
1029
1030        use chrono::TimeZone as _;
1031
1032        use crate::time::{TickingClock, install_ambient};
1033
1034        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
1035        let hour = Duration::from_secs(3600);
1036        let clock = TickingClock::starting_at(epoch);
1037        let _guard = install_ambient(Arc::new(clock.clone()));
1038        let log = PluginActivityLog::new();
1039
1040        // A slow request made its calls long ago and finishes last.
1041        let old: Vec<_> = (0..MAX_LOG_EVENTS)
1042            .map(|_| event("kv-get", CapabilityOutcome::Allowed))
1043            .collect();
1044        clock.advance(10 * hour);
1045        log.ingest("shop", [event("kv-set", CapabilityOutcome::Allowed)]);
1046        log.ingest("shop", old);
1047        // One more event evicts. The oldest by time must go, not the newest.
1048        log.ingest("shop", [event("kv-delete", CapabilityOutcome::Allowed)]);
1049
1050        let recent = log.summary("shop", hour);
1051        assert_eq!(recent.allowed.get("kv-set").copied(), Some(1), "{recent:?}");
1052        assert_eq!(recent.allowed.get("kv-delete").copied(), Some(1));
1053    }
1054
1055    #[test]
1056    fn folded_counts_leave_the_window_with_their_hour() {
1057        use std::sync::Arc;
1058
1059        use chrono::TimeZone as _;
1060
1061        use crate::time::{TickingClock, install_ambient};
1062
1063        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
1064        let hour = Duration::from_secs(3600);
1065        let clock = TickingClock::starting_at(epoch);
1066        let _guard = install_ambient(Arc::new(clock.clone()));
1067        let log = PluginActivityLog::new();
1068
1069        // Two hours of drop records, each its own record, then overflow.
1070        for _ in 0..MAX_LOG_EVENTS {
1071            log.ingest_dropped("shop", 1);
1072            clock.advance(Duration::from_secs(2));
1073        }
1074        clock.advance(10 * hour);
1075        for _ in 0..MAX_LOG_EVENTS {
1076            log.ingest_dropped("shop", 1);
1077            clock.advance(Duration::from_secs(2));
1078        }
1079        // The first batch folded away, ten hours back. It is not in the last
1080        // two hours, though the second batch is.
1081        let recent = log.summary("shop", 3 * hour);
1082        assert_eq!(recent.dropped, MAX_LOG_EVENTS as u64, "{}", recent.dropped);
1083        let all = log.summary("shop", 100 * hour);
1084        assert_eq!(all.dropped, 2 * MAX_LOG_EVENTS as u64);
1085    }
1086    #[test]
1087    fn a_fold_from_before_the_first_overflow_keeps_its_own_hour() {
1088        use std::sync::Arc;
1089
1090        use chrono::TimeZone as _;
1091
1092        use crate::time::{TickingClock, install_ambient};
1093
1094        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
1095        let hour = Duration::from_secs(3600);
1096        let clock = TickingClock::starting_at(epoch);
1097        let _guard = install_ambient(Arc::new(clock.clone()));
1098        let log = PluginActivityLog::new();
1099
1100        // Old events, then ten hours pass, then the ledger overflows with
1101        // recent records, then the old events are evicted.
1102        let old: Vec<_> = (0..MAX_LOG_EVENTS)
1103            .map(|_| event("kv-get", CapabilityOutcome::Allowed))
1104            .collect();
1105        log.ingest("shop", old);
1106        clock.advance(10 * hour);
1107        for _ in 0..MAX_LOG_EVENTS + 5 {
1108            log.ingest_dropped("shop", 1);
1109            clock.advance(Duration::from_secs(2));
1110        }
1111        log.ingest(
1112            "shop",
1113            (0..5).map(|_| event("kv-set", CapabilityOutcome::Allowed)),
1114        );
1115        // The ten-hour-old evictions are not in the last three hours.
1116        let recent = log.summary("shop", 3 * hour);
1117        assert_eq!(
1118            recent.dropped,
1119            MAX_LOG_EVENTS as u64 + 5,
1120            "{}",
1121            recent.dropped
1122        );
1123    }
1124
1125    #[test]
1126    fn a_summary_says_so_when_the_log_lost_drop_counts() {
1127        let log = PluginActivityLog::new();
1128        let hour = Duration::from_secs(3600);
1129        // This many plugins overflow the dropped ledger. Then the folded
1130        // ledger overflows and sheds a count.
1131        for n in 0..(MAX_LOG_EVENTS * 2 + 10) {
1132            log.ingest_dropped(&format!("p{n}"), 1);
1133        }
1134        assert!(
1135            log.summary("p0", hour).truncated,
1136            "counts were lost, and the summary must say so"
1137        );
1138        assert!(
1139            !PluginActivityLog::new().summary("p0", hour).truncated,
1140            "a log that lost nothing is not truncated"
1141        );
1142    }
1143
1144    #[test]
1145    fn a_truncation_leaves_the_window_with_its_time() {
1146        use std::sync::Arc;
1147
1148        use chrono::TimeZone as _;
1149
1150        use crate::time::{TickingClock, install_ambient};
1151
1152        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
1153        let hour = Duration::from_secs(3600);
1154        let clock = TickingClock::starting_at(epoch);
1155        let _guard = install_ambient(Arc::new(clock.clone()));
1156        let log = PluginActivityLog::new();
1157        for n in 0..(MAX_LOG_EVENTS * 2 + 10) {
1158            log.ingest_dropped(&format!("p{n}"), 1);
1159        }
1160        clock.advance(10 * hour);
1161        assert!(!log.summary("p0", hour).truncated, "the loss is old");
1162        assert!(log.summary("p0", 11 * hour).truncated);
1163    }
1164
1165    #[test]
1166    fn truncation_is_exact_at_the_window_edge_and_reaches_every_plugin() {
1167        use std::sync::Arc;
1168
1169        use chrono::TimeZone as _;
1170
1171        use crate::time::{TickingClock, install_ambient};
1172
1173        let epoch = chrono::Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
1174        let hour = Duration::from_secs(3600);
1175        let clock = TickingClock::starting_at(epoch);
1176        let _guard = install_ambient(Arc::new(clock.clone()));
1177        let log = PluginActivityLog::new();
1178        for n in 0..(MAX_LOG_EVENTS * 2 + 10) {
1179            log.ingest_dropped(&format!("p{n}"), 1);
1180        }
1181        let shed = log.lock().shed.expect("a count was shed");
1182        clock.advance(hour);
1183        let age = crate::time::ambient_instant().saturating_duration_since(shed);
1184        assert!(log.summary("other", age).truncated, "at the edge");
1185        assert!(
1186            !log.summary("other", age.saturating_sub(Duration::from_millis(1)))
1187                .truncated,
1188            "just outside"
1189        );
1190    }
1191
1192    #[test]
1193    fn a_truncated_summary_is_not_reported_as_empty() {
1194        let summary = ActivitySummary {
1195            plugin: "shop".to_owned(),
1196            truncated: true,
1197            ..ActivitySummary::default()
1198        };
1199        let text = summary.to_string();
1200        assert!(text.contains("shed older drop counts"), "{text}");
1201        assert!(!text.contains("no capability calls"), "{text}");
1202    }
1203}