zeph_durable/
timer.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Durable timers: wakes that survive a process restart.
5//!
6//! [`DurableContext::sleep_until`](crate::DurableContext::sleep_until) journals a `durable_timers`
7//! row at a deterministic [`TimerId`](crate::TimerId) for its program position and parks until the
8//! instant arrives. The [`DurableTimerService`] is a background task that polls the backend for due
9//! timers, marks them fired, and wakes their parked waiters.
10//!
11//! # Restart semantics (FR-DE-06)
12//!
13//! A timer is persisted with its `due_at`, so a process that was down when the instant elapsed
14//! recovers correctly: on the first poll after restart the service sees the timer's `due_at` is in
15//! the past and fires it immediately. The awaiting execution, replaying to the same `sleep_until`
16//! call, re-derives the timer id, finds it already fired, and returns at once instead of sleeping
17//! again.
18
19use std::sync::Arc;
20use std::time::Duration;
21
22use tracing::Instrument as _;
23
24use crate::backend::DurableBackendEnum;
25use crate::backend::local::now_unix_millis;
26
27/// Background task that fires durable timers whose instant has arrived.
28///
29/// Spawn [`DurableTimerService::run`] on a supervised task. It owns no timer state of its own — the
30/// `durable_timers` table is the source of truth — so it is safe to stop and restart: a restarted
31/// service re-reads due timers and fires any that elapsed while it was down.
32#[derive(Debug)]
33pub struct DurableTimerService {
34    backend: Arc<DurableBackendEnum>,
35    poll_interval: Duration,
36}
37
38impl DurableTimerService {
39    /// Build the service from the shared backend and a poll cadence.
40    ///
41    /// `poll_interval` is the worst-case latency between a timer's instant and its firing; the
42    /// `promise_poll_interval_secs` config value (default 2 s) is the natural source.
43    ///
44    /// # Examples
45    ///
46    /// ```no_run
47    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
48    /// use std::sync::Arc;
49    /// use std::time::Duration;
50    /// use zeph_durable::{DurableBackendEnum, DurableTimerService, LocalBackend};
51    ///
52    /// let backend = Arc::new(DurableBackendEnum::Local(Arc::new(
53    ///     LocalBackend::open("durable.db", 1_048_576).await?,
54    /// )));
55    /// let service = DurableTimerService::new(backend, Duration::from_secs(2));
56    /// let task = tokio::spawn(service.run());
57    /// # let _ = task;
58    /// # Ok(()) }
59    /// ```
60    #[must_use]
61    pub fn new(backend: Arc<DurableBackendEnum>, poll_interval: Duration) -> Self {
62        Self {
63            backend,
64            // Tokio's interval panics on a zero period; clamp to at least 1 ms.
65            poll_interval: poll_interval.max(Duration::from_millis(1)),
66        }
67    }
68
69    /// Run the timer poll loop until the task is aborted.
70    ///
71    /// Each tick fires every timer whose `due_at` has elapsed (including, on the first tick after a
72    /// restart, timers that came due during downtime — FR-DE-06). A backend error is logged and the
73    /// loop continues so a transient failure does not strand future timers.
74    #[tracing::instrument(name = "durable.timer.run", skip_all)]
75    pub async fn run(self) {
76        let mut tick = tokio::time::interval(self.poll_interval);
77        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
78        loop {
79            tick.tick().await;
80            self.fire_due()
81                .instrument(tracing::info_span!("durable.timer.run.iter"))
82                .await;
83        }
84    }
85
86    /// Fire every currently-due timer once. Exposed for a deterministic test poll.
87    #[tracing::instrument(name = "durable.timer.fire_due", skip_all)]
88    pub(crate) async fn fire_due(&self) {
89        let now = now_unix_millis();
90        let due = match self.backend.due_timers(now).await {
91            Ok(due) => due,
92            Err(error) => {
93                tracing::warn!(%error, "durable timer poll failed; will retry next tick");
94                return;
95            }
96        };
97        for timer in due {
98            match self.backend.mark_timer_fired(timer).await {
99                Ok(true) => tracing::debug!(timer_id = %timer.as_uuid(), "durable timer fired"),
100                // Already fired by a concurrent poll — nothing to do.
101                Ok(false) => {}
102                Err(error) => {
103                    tracing::warn!(%error, timer_id = %timer.as_uuid(), "failed to mark timer fired");
104                }
105            }
106        }
107    }
108}
zeph_durable/timer.rs

zeph_durable/
timer.rs
