1use std::sync::Arc;
11use std::sync::atomic::{AtomicU64, Ordering};
12use std::time::{Duration, Instant};
13
14pub const DEFAULT_RULE_TIMEOUT: Duration = Duration::from_secs(1);
16
17pub const DEFAULT_GLOBAL_TIMEOUT: Duration = Duration::from_secs(15);
19
20pub const DEFAULT_MEMORY_BYTES: usize = 64 * 1024 * 1024;
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub struct Limits {
26 pub rule_timeout: Duration,
32
33 pub global_timeout: Duration,
38
39 pub memory_bytes: usize,
41}
42
43impl Default for Limits {
44 fn default() -> Self {
45 Self {
46 rule_timeout: DEFAULT_RULE_TIMEOUT,
47 global_timeout: DEFAULT_GLOBAL_TIMEOUT,
48 memory_bytes: DEFAULT_MEMORY_BYTES,
49 }
50 }
51}
52
53impl Limits {
54 #[must_use]
59 pub const fn with_rule_timeout(mut self, timeout: Duration) -> Self {
60 self.rule_timeout = timeout;
61 self
62 }
63
64 #[must_use]
66 pub const fn with_global_timeout(mut self, timeout: Duration) -> Self {
67 self.global_timeout = timeout;
68 self
69 }
70
71 #[must_use]
73 pub const fn with_memory_bytes(mut self, bytes: usize) -> Self {
74 self.memory_bytes = bytes;
75 self
76 }
77}
78
79#[derive(Debug)]
84pub struct RunClock {
85 start: Instant,
86 global_timeout: Duration,
87}
88
89impl RunClock {
90 #[must_use]
92 pub fn start(global_timeout: Duration) -> Arc<Self> {
93 Arc::new(Self {
94 start: Instant::now(),
95 global_timeout,
96 })
97 }
98
99 #[must_use]
101 pub fn elapsed(&self) -> Duration {
102 self.start.elapsed()
103 }
104
105 #[must_use]
107 pub const fn global_timeout(&self) -> Duration {
108 self.global_timeout
109 }
110
111 #[must_use]
113 pub fn is_expired(&self) -> bool {
114 self.elapsed() >= self.global_timeout
115 }
116
117 fn elapsed_nanos(&self) -> u64 {
118 u64::try_from(self.start.elapsed().as_nanos()).unwrap_or(u64::MAX)
119 }
120}
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq)]
124pub(crate) enum Trip {
125 Rule,
127 Run,
129}
130
131const TRIP_NONE: u64 = 0;
132const TRIP_RULE: u64 = 1;
133const TRIP_RUN: u64 = 2;
134
135#[derive(Debug)]
143pub(crate) struct Budget {
144 clock: Arc<RunClock>,
145 global_nanos: u64,
146 invocation_deadline_nanos: AtomicU64,
149 tripped: AtomicU64,
150}
151
152impl Budget {
153 pub(crate) fn new(clock: Arc<RunClock>) -> Arc<Self> {
154 let global_nanos = u64::try_from(clock.global_timeout.as_nanos()).unwrap_or(u64::MAX);
155 Arc::new(Self {
156 clock,
157 global_nanos,
158 invocation_deadline_nanos: AtomicU64::new(0),
159 tripped: AtomicU64::new(TRIP_NONE),
160 })
161 }
162
163 pub(crate) fn arm(&self, rule_timeout: Duration) {
165 let now = self.clock.elapsed_nanos();
166 let budget = u64::try_from(rule_timeout.as_nanos()).unwrap_or(u64::MAX);
167 self.invocation_deadline_nanos
170 .store(now.saturating_add(budget).max(1), Ordering::Relaxed);
171 self.tripped.store(TRIP_NONE, Ordering::Relaxed);
172 }
173
174 pub(crate) fn disarm(&self) {
176 self.invocation_deadline_nanos.store(0, Ordering::Relaxed);
177 }
178
179 pub(crate) fn should_interrupt(&self) -> bool {
182 let elapsed = self.clock.elapsed_nanos();
183
184 if elapsed >= self.global_nanos {
185 self.tripped.store(TRIP_RUN, Ordering::Relaxed);
186 return true;
187 }
188
189 let deadline = self.invocation_deadline_nanos.load(Ordering::Relaxed);
190 if deadline != 0 && elapsed >= deadline {
191 self.tripped.store(TRIP_RULE, Ordering::Relaxed);
192 return true;
193 }
194
195 false
196 }
197
198 pub(crate) fn take_trip(&self) -> Option<Trip> {
200 match self.tripped.swap(TRIP_NONE, Ordering::Relaxed) {
201 TRIP_RULE => Some(Trip::Rule),
202 TRIP_RUN => Some(Trip::Run),
203 _ => None,
204 }
205 }
206
207 pub(crate) fn clock(&self) -> &RunClock {
208 &self.clock
209 }
210}
211
212#[cfg(test)]
213mod tests {
214 use super::*;
215
216 #[test]
217 fn defaults_match_the_documented_budgets() {
218 let limits = Limits::default();
219 assert_eq!(limits.rule_timeout, Duration::from_secs(1));
220 assert_eq!(limits.global_timeout, Duration::from_secs(15));
221 assert_eq!(limits.memory_bytes, 64 * 1024 * 1024);
222 }
223
224 #[test]
225 fn the_rule_budget_is_well_under_the_global_one() {
226 let limits = Limits::default();
230 assert!(
231 limits.rule_timeout * 5 < limits.global_timeout,
232 "the per-invocation budget must leave room for the global limit to be a backstop"
233 );
234 }
235
236 #[test]
237 fn a_rule_cannot_raise_the_global_budget() {
238 let limits = Limits::default().with_rule_timeout(Duration::from_secs(60));
239 assert_eq!(limits.rule_timeout, Duration::from_secs(60));
240 assert_eq!(
241 limits.global_timeout, DEFAULT_GLOBAL_TIMEOUT,
242 "raising a rule's own budget must not extend the run"
243 );
244 }
245
246 #[test]
247 fn an_unarmed_budget_never_interrupts() {
248 let budget = Budget::new(RunClock::start(Duration::from_secs(3600)));
249 assert!(!budget.should_interrupt());
250 assert_eq!(budget.take_trip(), None);
251 }
252
253 #[test]
254 fn an_expired_invocation_budget_interrupts_and_records_why() {
255 let budget = Budget::new(RunClock::start(Duration::from_secs(3600)));
256 budget.arm(Duration::ZERO);
257 assert!(budget.should_interrupt());
258 assert_eq!(budget.take_trip(), Some(Trip::Rule));
259 }
260
261 #[test]
262 fn an_expired_run_budget_interrupts_and_records_why() {
263 let budget = Budget::new(RunClock::start(Duration::ZERO));
264 budget.arm(Duration::from_secs(3600));
265 assert!(budget.should_interrupt());
266 assert_eq!(budget.take_trip(), Some(Trip::Run));
267 }
268
269 #[test]
270 fn the_run_budget_wins_when_both_are_spent() {
271 let budget = Budget::new(RunClock::start(Duration::ZERO));
274 budget.arm(Duration::ZERO);
275 assert!(budget.should_interrupt());
276 assert_eq!(budget.take_trip(), Some(Trip::Run));
277 }
278
279 #[test]
280 fn disarming_stops_invocation_enforcement() {
281 let budget = Budget::new(RunClock::start(Duration::from_secs(3600)));
282 budget.arm(Duration::ZERO);
283 budget.disarm();
284 assert!(!budget.should_interrupt(), "no invocation is in flight");
285 }
286
287 #[test]
288 fn taking_the_trip_clears_it() {
289 let budget = Budget::new(RunClock::start(Duration::from_secs(3600)));
290 budget.arm(Duration::ZERO);
291 assert!(budget.should_interrupt());
292 assert_eq!(budget.take_trip(), Some(Trip::Rule));
293 assert_eq!(
294 budget.take_trip(),
295 None,
296 "a trip must not be reported twice"
297 );
298 }
299
300 #[test]
301 fn arming_clears_a_previous_trip() {
302 let budget = Budget::new(RunClock::start(Duration::from_secs(3600)));
305 budget.arm(Duration::ZERO);
306 assert!(budget.should_interrupt());
307
308 budget.arm(Duration::from_secs(3600));
309 assert!(!budget.should_interrupt());
310 assert_eq!(budget.take_trip(), None);
311 }
312
313 #[test]
314 fn an_overflowing_budget_does_not_wrap_into_disarmed() {
315 let budget = Budget::new(RunClock::start(Duration::from_secs(3600)));
318 budget.arm(Duration::MAX);
319 assert_ne!(
320 budget.invocation_deadline_nanos.load(Ordering::Relaxed),
321 0,
322 "an overflowing budget must not read as disarmed"
323 );
324 }
325
326 #[test]
327 fn the_clock_measures_from_one_origin() {
328 let clock = RunClock::start(Duration::from_secs(3600));
329 let a = Arc::clone(&clock);
330 let b = Arc::clone(&clock);
331 assert!(!a.is_expired());
332 assert!(!b.is_expired());
333 assert_eq!(a.global_timeout(), Duration::from_secs(3600));
334 }
335
336 #[test]
337 fn a_zero_global_budget_is_immediately_expired() {
338 assert!(RunClock::start(Duration::ZERO).is_expired());
339 }
340}