cymule_activation_timer/
lib.rs

1//! Durable logical timer source for Cymule.
2
3use std::path::Path;
4use std::time::{Duration, SystemTime, UNIX_EPOCH};
5
6use cymule_durable::{
7    DurableError, DurableResult, ParkedWaitIndex, WaitActivationSource, WaitDelivery,
8    WaitSourceDriver,
9};
10use rusqlite::{Connection, ErrorCode, OptionalExtension, params};
11use serde_json::Value;
12
13/// Clock observation boundary for timer plugins.
14pub trait Clock {
15    /// Current observed Unix time in milliseconds.
16    fn now_unix_ms(&self) -> DurableResult<u64>;
17}
18
19/// System clock implementation for production timer polling.
20#[derive(Debug, Clone, Copy, Default)]
21pub struct SystemClock;
22
23impl Clock for SystemClock {
24    fn now_unix_ms(&self) -> DurableResult<u64> {
25        let duration = SystemTime::now()
26            .duration_since(UNIX_EPOCH)
27            .map_err(|error| DurableError::Substrate(error.to_string()))?;
28        u64::try_from(duration.as_millis())
29            .map_err(|error| DurableError::Substrate(error.to_string()))
30    }
31}
32
33/// SQLite-backed durable timer source.
34pub struct SqliteTimerDriver<C = SystemClock> {
35    connection: Connection,
36    clock: C,
37}
38
39impl SqliteTimerDriver<SystemClock> {
40    /// Open or create a file-backed timer source.
41    pub fn open(path: impl AsRef<Path>) -> DurableResult<Self> {
42        Self::open_with_clock(path, SystemClock)
43    }
44
45    /// Create an in-memory timer source.
46    pub fn in_memory() -> DurableResult<Self> {
47        Self::in_memory_with_clock(SystemClock)
48    }
49}
50
51impl<C: Clock> SqliteTimerDriver<C> {
52    /// Open a file-backed timer source with an injected clock.
53    pub fn open_with_clock(path: impl AsRef<Path>, clock: C) -> DurableResult<Self> {
54        let connection = Connection::open(path).map_err(sqlite_error)?;
55        Self::initialize(connection, clock, true)
56    }
57
58    /// Create an in-memory timer source with an injected clock.
59    pub fn in_memory_with_clock(clock: C) -> DurableResult<Self> {
60        let connection = Connection::open_in_memory().map_err(sqlite_error)?;
61        Self::initialize(connection, clock, false)
62    }
63
64    fn initialize(connection: Connection, clock: C, file_backed: bool) -> DurableResult<Self> {
65        connection
66            .busy_timeout(Duration::ZERO)
67            .map_err(sqlite_error)?;
68        if file_backed {
69            connection
70                .pragma_update(None, "journal_mode", "WAL")
71                .map_err(sqlite_error)?;
72            connection
73                .pragma_update(None, "synchronous", "FULL")
74                .map_err(sqlite_error)?;
75        }
76        connection
77            .execute_batch(
78                "CREATE TABLE IF NOT EXISTS cymule_timers (
79                    activation_id TEXT PRIMARY KEY NOT NULL,
80                    timer_id TEXT NOT NULL,
81                    due_unix_ms INTEGER NOT NULL,
82                    value_json BLOB NOT NULL,
83                    acknowledged INTEGER NOT NULL DEFAULT 0
84                ) STRICT;
85                CREATE INDEX IF NOT EXISTS cymule_timers_due
86                    ON cymule_timers(acknowledged, due_unix_ms, activation_id);",
87            )
88            .map_err(sqlite_error)?;
89        Ok(Self { connection, clock })
90    }
91
92    /// Schedule or exactly replay one logical timer delivery.
93    pub fn schedule(
94        &mut self,
95        activation_id: &str,
96        timer_id: &str,
97        due_unix_ms: u64,
98        value: &Value,
99    ) -> DurableResult<()> {
100        validate_identity("activation", activation_id)?;
101        validate_identity("timer", timer_id)?;
102        let due = i64::try_from(due_unix_ms)
103            .map_err(|error| DurableError::Validation(error.to_string()))?;
104        let bytes = cymule_core::canonical_bytes(value)?;
105        let existing: Option<(String, i64, Vec<u8>, bool)> = self
106            .connection
107            .query_row(
108                "SELECT timer_id, due_unix_ms, value_json, acknowledged
109                 FROM cymule_timers WHERE activation_id = ?1",
110                [activation_id],
111                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
112            )
113            .optional()
114            .map_err(sqlite_error)?;
115        if let Some((current_timer, current_due, current_value, _acknowledged)) = existing {
116            if current_timer == timer_id && current_due == due && current_value == bytes {
117                return Ok(());
118            }
119            return Err(DurableError::Conflict {
120                expected: Some(activation_id.to_owned()),
121                current: Some("timer-identity-reused".to_owned()),
122            });
123        }
124        self.connection
125            .execute(
126                "INSERT INTO cymule_timers(
127                    activation_id, timer_id, due_unix_ms, value_json, acknowledged
128                 ) VALUES (?1, ?2, ?3, ?4, 0)",
129                params![activation_id, timer_id, due, bytes],
130            )
131            .map_err(contention)?;
132        Ok(())
133    }
134}
135
136impl<C: Clock> WaitSourceDriver for SqliteTimerDriver<C> {
137    fn receive(
138        &mut self,
139        index: &ParkedWaitIndex,
140        max_targets: usize,
141    ) -> DurableResult<Option<WaitDelivery>> {
142        let now = i64::try_from(self.clock.now_unix_ms()?)
143            .map_err(|error| DurableError::Substrate(error.to_string()))?;
144        let mut statement = self
145            .connection
146            .prepare(
147                "SELECT activation_id, timer_id, value_json
148                 FROM cymule_timers
149                 WHERE acknowledged = 0 AND due_unix_ms <= ?1
150                 ORDER BY due_unix_ms, activation_id",
151            )
152            .map_err(sqlite_error)?;
153        let rows = statement
154            .query_map([now], |row| {
155                Ok((
156                    row.get::<_, String>(0)?,
157                    row.get::<_, String>(1)?,
158                    row.get::<_, Vec<u8>>(2)?,
159                ))
160            })
161            .map_err(sqlite_error)?;
162        for row in rows {
163            let (activation_id, timer_id, bytes) = row.map_err(sqlite_error)?;
164            let source = WaitActivationSource::Timer { timer_id };
165            let selection = index.select(&source, max_targets)?;
166            if selection.wait_ids.is_empty() {
167                continue;
168            }
169            let value = serde_json::from_slice(&bytes)?;
170            return Ok(Some(WaitDelivery {
171                activation_id,
172                source,
173                wait_ids: selection.wait_ids,
174                value,
175            }));
176        }
177        Ok(None)
178    }
179
180    fn acknowledge(&mut self, activation_id: &str) -> DurableResult<()> {
181        validate_identity("activation", activation_id)?;
182        self.connection
183            .execute(
184                "UPDATE cymule_timers SET acknowledged = 1
185                 WHERE activation_id = ?1 AND acknowledged = 0",
186                [activation_id],
187            )
188            .map_err(contention)?;
189        let acknowledged = self
190            .connection
191            .query_row(
192                "SELECT acknowledged FROM cymule_timers WHERE activation_id = ?1",
193                [activation_id],
194                |row| row.get::<_, bool>(0),
195            )
196            .optional()
197            .map_err(sqlite_error)?;
198        if acknowledged == Some(true) {
199            Ok(())
200        } else {
201            Err(DurableError::NotFound(format!(
202                "timer activation {activation_id} is missing"
203            )))
204        }
205    }
206}
207
208fn validate_identity(kind: &str, identity: &str) -> DurableResult<()> {
209    if identity.is_empty() || identity.len() > 512 || identity.chars().any(char::is_control) {
210        return Err(DurableError::Validation(format!(
211            "timer {kind} identity must contain 1..=512 printable characters"
212        )));
213    }
214    Ok(())
215}
216
217fn contention(error: rusqlite::Error) -> DurableError {
218    match &error {
219        rusqlite::Error::SqliteFailure(failure, _)
220            if matches!(
221                failure.code,
222                ErrorCode::DatabaseBusy | ErrorCode::DatabaseLocked
223            ) =>
224        {
225            DurableError::Conflict {
226                expected: Some("sqlite-timer-writer-available".to_owned()),
227                current: Some("sqlite-timer-writer-active".to_owned()),
228            }
229        }
230        _ => sqlite_error(error),
231    }
232}
233
234fn sqlite_error(error: impl std::fmt::Display) -> DurableError {
235    DurableError::Substrate(error.to_string())
236}
cymule_activation_timer/lib.rs

cymule_activation_timer/
lib.rs
