reifydb_runtime/actor/timers/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	fmt,
6	fmt::Debug,
7	sync::{
8		Arc,
9		atomic::{AtomicBool, AtomicU64, Ordering},
10	},
11};
12
13#[cfg(reifydb_dst)]
14pub(crate) mod dst;
15#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
16pub mod scheduler;
17#[cfg(reifydb_target = "wasi")]
18pub(crate) mod wasi;
19#[cfg(reifydb_target = "wasm")]
20pub(crate) mod wasm;
21
22#[cfg(reifydb_target = "wasi")]
23use wasi::drain_expired_timers as wasi_drain;
24
25use super::mailbox::SendError;
26
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum Repeat {
29	Keep,
30
31	Cancel,
32}
33
34impl Repeat {
35	pub fn after_send<M>(result: Result<(), SendError<M>>) -> Self {
36		match result {
37			Ok(()) => Self::Keep,
38			Err(SendError::Full(_)) => Self::Keep,
39			Err(SendError::Closed(_)) => Self::Cancel,
40		}
41	}
42
43	pub fn is_keep(&self) -> bool {
44		matches!(self, Self::Keep)
45	}
46}
47
48#[derive(Clone)]
49pub struct TimerHandle {
50	id: u64,
51	cancelled: Arc<AtomicBool>,
52}
53
54impl TimerHandle {
55	pub(crate) fn new(id: u64) -> Self {
56		Self {
57			id,
58			cancelled: Arc::new(AtomicBool::new(false)),
59		}
60	}
61
62	pub fn cancel(&self) -> bool {
63		self.cancelled.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst).is_ok()
64	}
65
66	pub fn is_cancelled(&self) -> bool {
67		self.cancelled.load(Ordering::SeqCst)
68	}
69
70	pub fn id(&self) -> u64 {
71		self.id
72	}
73
74	pub(crate) fn cancelled_flag(&self) -> Arc<AtomicBool> {
75		self.cancelled.clone()
76	}
77}
78
79impl Debug for TimerHandle {
80	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
81		f.debug_struct("TimerHandle").field("id", &self.id).field("cancelled", &self.is_cancelled()).finish()
82	}
83}
84
85static TIMER_ID_COUNTER: AtomicU64 = AtomicU64::new(0);
86
87pub(crate) fn next_timer_id() -> u64 {
88	TIMER_ID_COUNTER.fetch_add(1, Ordering::Relaxed)
89}
90
91#[cfg(reifydb_target = "wasi")]
92pub fn drain_expired_timers() {
93	wasi_drain();
94}
95
96#[cfg(not(reifydb_target = "wasi"))]
97pub fn drain_expired_timers() {}
98
99#[cfg(all(test, reifydb_target = "host", not(reifydb_dst)))]
100mod tests {
101	use super::*;
102	use crate::actor::mailbox::create_mailbox;
103
104	#[test]
105	fn a_full_mailbox_drops_the_tick_but_keeps_the_timer_armed() {
106		// The defect this pins: every repeating timer collapsed send() into `.is_ok()`, so one
107		// transient full mailbox retired the timer for the life of the process with no log line.
108		// Six of the nine lifecycle maintenance tasks were dead this way, the tombstone reaper
109		// among them, which is how operator state came to be 80% unreaped tombstones.
110		let (actor, _mailbox) = create_mailbox::<u8>(Some(1));
111		assert_eq!(Repeat::after_send(actor.send(1)), Repeat::Keep, "precondition: the first send fits");
112
113		let full = actor.send(2);
114		assert!(
115			matches!(full, Err(SendError::Full(_))),
116			"precondition: capacity 1 must reject the second send"
117		);
118		assert_eq!(Repeat::after_send(full), Repeat::Keep, "backpressure must not retire a repeating timer");
119	}
120
121	#[test]
122	fn a_closed_mailbox_retires_the_timer() {
123		// The one case that must still cancel. Without it a timer fires forever into a dead
124		// channel, and nothing else ever removes it from the heap.
125		let (actor, mailbox) = create_mailbox::<u8>(Some(1));
126		drop(mailbox);
127
128		let closed = actor.send(1);
129		assert!(
130			matches!(closed, Err(SendError::Closed(_))),
131			"precondition: a dropped mailbox closes the channel"
132		);
133		assert_eq!(Repeat::after_send(closed), Repeat::Cancel, "a dead actor must retire its timer");
134	}
135}
reifydb_runtime/actor/timers/mod.rs

reifydb_runtime/actor/timers/
mod.rs
