1#![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#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
51#[serde(rename_all = "kebab-case")]
52#[non_exhaustive]
53pub enum CapabilityOutcome {
54 Allowed,
56 QuotaExceeded,
58 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#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
74#[non_exhaustive]
75pub struct CapabilityEvent {
76 pub capability: SandboxCapability,
78 pub operation: &'static str,
80 pub target: String,
82 pub outcome: CapabilityOutcome,
84 #[serde(skip)]
90 pub at: Instant,
91}
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq)]
98#[non_exhaustive]
99pub struct DroppedCalls {
100 pub count: u64,
102 pub first: Instant,
104 pub last: Instant,
106}
107
108pub const MAX_LOG_EVENTS: usize = 4096;
116
117#[derive(Debug, Default)]
122pub struct PluginActivityLog {
123 state: Mutex<LogState>,
127}
128
129#[derive(Debug, Default)]
131struct LogState {
132 entries: BTreeMap<(Instant, u64), (String, CapabilityEvent)>,
136 dropped: BTreeMap<(Instant, u64), DroppedRecord>,
143 seq: u64,
145 folded: BTreeMap<(String, u64), (Instant, u64)>,
150 origin: Option<Instant>,
152 shed: Option<Instant>,
155}
156
157fn apart(a: Instant, b: Instant) -> Duration {
159 a.max(b).saturating_duration_since(a.min(b))
160}
161
162#[derive(Debug)]
164struct DroppedRecord {
165 plugin: String,
166 count: u64,
167 first: Instant,
170}
171
172const FOLD_BUCKET_SECS: u64 = 3600;
174
175const DROPPED_BUCKET: Duration = Duration::from_secs(1);
183
184impl PluginActivityLog {
185 #[must_use]
187 pub fn new() -> Self {
188 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 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 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 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 pub fn ingest_dropped_calls(&self, plugin: &str, calls: DroppedCalls) {
247 if calls.count == 0 {
248 return;
249 }
250 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 #[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 #[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 fn note_dropped(&mut self, first: Instant, last: Instant, plugin: &str, count: u64) {
322 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 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 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#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
382#[non_exhaustive]
383pub struct ActivitySummary {
384 pub allowed: BTreeMap<&'static str, u64>,
386 pub denied: BTreeMap<&'static str, u64>,
388 pub quota_hits: BTreeMap<&'static str, u64>,
390 pub hosts: BTreeMap<String, u64>,
392 pub tables: BTreeMap<String, u64>,
394 pub job_types: BTreeMap<String, u64>,
396 pub refused_targets: BTreeMap<String, u64>,
398 pub plugin: String,
404 pub window: Duration,
406 pub dropped: u64,
414 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 let allowed = matches!(
440 event.outcome,
441 CapabilityOutcome::Allowed
442 | CapabilityOutcome::Denied(
443 DenialReason::ResponseTooLarge
444 | DenialReason::ResultRejected
445 | DenialReason::BackendError
446 )
447 );
448 let ran = matches!(
450 event.outcome,
451 CapabilityOutcome::Allowed
452 | CapabilityOutcome::Denied(
453 DenialReason::ResponseTooLarge | DenialReason::ResultRejected
454 )
455 );
456 let enqueued = ran;
463 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 (
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 #[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 fn write_report(&self, out: &mut String) {
500 use std::fmt::Write as _;
501
502 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 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 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 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 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 assert_eq!(log.summary("shop", 40 * minute).dropped, 4);
647 assert_eq!(log.summary("shop", 20 * minute).dropped, 0);
648 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 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 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 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 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 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 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 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 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 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 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 let old: Vec<_> = (0..MAX_LOG_EVENTS)
943 .map(|_| event("kv-get", CapabilityOutcome::Allowed))
944 .collect();
945 clock.advance(10 * hour);
946 for _ in 0..MAX_LOG_EVENTS {
948 log.ingest_dropped("shop", 1);
949 clock.advance(Duration::from_secs(2));
950 }
951 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 let recent = log.summary("shop", Duration::from_secs(3 * 3600));
964 assert_eq!(recent.dropped, MAX_LOG_EVENTS as u64 + 1);
965 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 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 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 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 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 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 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 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 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 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}