made_core/value_objects/delivery/
attention_policy.rs

1use std::collections::BTreeSet;
2
3use serde::{Deserialize, Serialize};
4
5use crate::error::DomainError;
6use crate::value_objects::DurationMs;
7
8use super::{AttentionKind, AttentionOverflowPolicy, LoopLimits, QueueLimit};
9
10/// Five seconds of quiet before two restless kinds are folded into one.
11const DEFAULT_COALESCE_MS: u64 = 5_000;
12const MAX_COALESCE_MS: u64 = 60_000;
13
14/// What an integrator asked to be told about, and how insistently.
15#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
16pub struct AttentionPolicy {
17    kinds: BTreeSet<AttentionKind>,
18    coalesce_window: DurationMs,
19    max_queued: QueueLimit,
20    overflow: AttentionOverflowPolicy,
21    #[serde(default, skip_serializing_if = "Option::is_none")]
22    inactivity_after: Option<DurationMs>,
23    limits: LoopLimits,
24}
25
26impl AttentionPolicy {
27    /// Construct a validated policy.
28    ///
29    /// An empty selection is refused rather than silently meaning
30    /// "everything": a binding that asked for nothing and is woken for
31    /// everything is the kind of surprise a policy exists to prevent.
32    pub fn new(
33        kinds: impl IntoIterator<Item = AttentionKind>,
34        coalesce_window: DurationMs,
35        max_queued: QueueLimit,
36        overflow: AttentionOverflowPolicy,
37        inactivity_after: Option<DurationMs>,
38        limits: LoopLimits,
39    ) -> Result<Self, DomainError> {
40        let kinds: BTreeSet<AttentionKind> = kinds.into_iter().collect();
41        if kinds.is_empty() {
42            return Err(DomainError::EmptyCollection {
43                field: "attention_policy_kinds",
44            });
45        }
46        if coalesce_window.get() > MAX_COALESCE_MS {
47            return Err(DomainError::OutOfRange {
48                field: "attention_coalesce_window",
49                value: coalesce_window.get() as f64,
50                min: 0.0,
51                max: MAX_COALESCE_MS as f64,
52            });
53        }
54        if inactivity_after.is_some_and(|after| after.get() == 0) {
55            return Err(DomainError::MustBeNonZero {
56                field: "attention_inactivity_after",
57            });
58        }
59        Ok(Self {
60            kinds,
61            coalesce_window,
62            max_queued,
63            overflow,
64            inactivity_after,
65            limits,
66        })
67    }
68
69    /// Whether this policy asked to hear about that kind.
70    #[must_use]
71    pub fn admits(&self, kind: AttentionKind) -> bool {
72        self.kinds.contains(&kind)
73    }
74
75    #[must_use]
76    pub const fn kinds(&self) -> &BTreeSet<AttentionKind> {
77        &self.kinds
78    }
79
80    /// How long two restless events of one kind are folded into one.
81    #[must_use]
82    pub const fn coalesce_window(&self) -> DurationMs {
83        self.coalesce_window
84    }
85
86    /// Whether this kind is one the window applies to at all.
87    ///
88    /// Coalescing a result or a decision would lose one; only the kinds
89    /// that repeat while nothing changes are folded.
90    #[must_use]
91    pub const fn coalesces(kind: AttentionKind) -> bool {
92        matches!(
93            kind,
94            AttentionKind::InactivityDetected | AttentionKind::Blocked
95        )
96    }
97
98    #[must_use]
99    pub const fn max_queued(&self) -> QueueLimit {
100        self.max_queued
101    }
102
103    #[must_use]
104    pub const fn overflow(&self) -> AttentionOverflowPolicy {
105        self.overflow
106    }
107
108    /// How long nothing may happen before the quiet is itself reported.
109    #[must_use]
110    pub const fn inactivity_after(&self) -> Option<DurationMs> {
111        self.inactivity_after
112    }
113
114    #[must_use]
115    pub const fn limits(&self) -> LoopLimits {
116        self.limits
117    }
118}
119
120impl Default for AttentionPolicy {
121    fn default() -> Self {
122        Self {
123            kinds: AttentionKind::ALL.into_iter().collect(),
124            coalesce_window: DurationMs::from_millis(DEFAULT_COALESCE_MS),
125            max_queued: QueueLimit::default(),
126            overflow: AttentionOverflowPolicy::default(),
127            inactivity_after: None,
128            limits: LoopLimits::default(),
129        }
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136
137    #[test]
138    fn the_default_hears_everything_and_folds_only_the_restless_kinds() {
139        let policy = AttentionPolicy::default();
140        for kind in AttentionKind::ALL {
141            assert!(policy.admits(kind), "{kind}");
142        }
143        assert_eq!(policy.coalesce_window().get(), 5_000);
144        assert_eq!(policy.max_queued().value(), 200);
145        assert!(AttentionPolicy::coalesces(AttentionKind::Blocked));
146        assert!(!AttentionPolicy::coalesces(AttentionKind::ResultAvailable));
147    }
148
149    #[test]
150    fn a_policy_that_asked_for_nothing_is_refused() {
151        let refused = AttentionPolicy::new(
152            [],
153            DurationMs::ZERO,
154            QueueLimit::default(),
155            AttentionOverflowPolicy::default(),
156            None,
157            LoopLimits::default(),
158        );
159        assert!(refused.is_err());
160    }
161}
made_core/value_objects/delivery/attention_policy.rs

made_core/value_objects/delivery/
attention_policy.rs
