everruns_local/
schedule_runner.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use chrono::Utc;
5use futures::StreamExt;
6use tokio::sync::watch;
7use tokio::task::JoinHandle;
8use tokio::time::MissedTickBehavior;
9
10use crate::{LocalError, LocalResult, LocalScheduleStore, LocalSessionRunner};
11
12/// Runtime settings for [`LocalScheduleRunner`].
13#[derive(Clone, Debug)]
14pub struct LocalScheduleRunnerConfig {
15    /// How often the SQLite store is checked for due schedules.
16    pub poll_interval: Duration,
17    /// How long an unfinished claim remains exclusive before another runner may recover it.
18    /// Failed deliveries also wait this long before becoming claimable again.
19    pub claim_timeout: Duration,
20    /// Maximum schedules claimed per poll.
21    pub batch_size: usize,
22}
23
24const MAX_BATCH_SIZE: usize = 1_000;
25
26impl Default for LocalScheduleRunnerConfig {
27    fn default() -> Self {
28        Self {
29            poll_interval: Duration::from_secs(1),
30            claim_timeout: Duration::from_secs(30),
31            batch_size: 32,
32        }
33    }
34}
35
36/// Explicitly started in-process executor for schedules in a [`LocalScheduleStore`].
37pub struct LocalScheduleRunner {
38    store: LocalScheduleStore,
39    session_runner: Arc<dyn LocalSessionRunner>,
40    config: LocalScheduleRunnerConfig,
41}
42
43impl LocalScheduleRunner {
44    pub fn new(store: LocalScheduleStore, session_runner: Arc<dyn LocalSessionRunner>) -> Self {
45        Self {
46            store,
47            session_runner,
48            config: LocalScheduleRunnerConfig::default(),
49        }
50    }
51
52    pub fn with_config(mut self, config: LocalScheduleRunnerConfig) -> Self {
53        self.config = config;
54        self
55    }
56
57    /// Spawn the polling task. Retain the returned handle for the host lifetime.
58    pub fn start(self) -> LocalResult<LocalScheduleRunnerHandle> {
59        if self.config.poll_interval.is_zero() {
60            return Err(LocalError::Config(
61                "schedule poll interval must be positive".into(),
62            ));
63        }
64        if self.config.claim_timeout.is_zero() {
65            return Err(LocalError::Config(
66                "schedule claim timeout must be positive".into(),
67            ));
68        }
69        if self.config.batch_size == 0 {
70            return Err(LocalError::Config(
71                "schedule batch size must be positive".into(),
72            ));
73        }
74        if self.config.batch_size > MAX_BATCH_SIZE {
75            return Err(LocalError::Config(format!(
76                "schedule batch size must not exceed {MAX_BATCH_SIZE}"
77            )));
78        }
79        let runner_id = uuid::Uuid::now_v7().to_string();
80        let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
81        let join = tokio::spawn(async move {
82            tracing::info!(
83                runner_id,
84                poll_interval_ms = self.config.poll_interval.as_millis(),
85                claim_timeout_ms = self.config.claim_timeout.as_millis(),
86                batch_size = self.config.batch_size,
87                "Local schedule runner started"
88            );
89            let mut interval = tokio::time::interval(self.config.poll_interval);
90            interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
91            loop {
92                tokio::select! {
93                    biased;
94                    changed = shutdown_rx.changed() => {
95                        if changed.is_err() || *shutdown_rx.borrow() {
96                            break;
97                        }
98                    }
99                    _ = interval.tick() => {
100                        if let Err(error) = self.poll_once(&runner_id).await {
101                            tracing::error!(runner_id, error = %error, "Local schedule poll failed");
102                        }
103                    }
104                }
105            }
106            tracing::info!(runner_id, "Local schedule runner stopped");
107        });
108        Ok(LocalScheduleRunnerHandle {
109            shutdown_tx,
110            join: Some(join),
111        })
112    }
113
114    async fn poll_once(&self, runner_id: &str) -> everruns_core::Result<()> {
115        let routable_session_ids = self.session_runner.routable_session_ids().await?;
116        let claimed = self.store.claim_due(
117            runner_id,
118            Utc::now(),
119            self.config.claim_timeout,
120            self.config.batch_size,
121            routable_session_ids.as_deref(),
122        )?;
123        if !claimed.is_empty() {
124            tracing::debug!(
125                runner_id,
126                count = claimed.len(),
127                "Claimed due local schedules"
128            );
129        }
130        futures::stream::iter(claimed)
131            .for_each_concurrent(self.config.batch_size, |claim| async move {
132                self.process_claim(claim, runner_id).await;
133            })
134            .await;
135        Ok(())
136    }
137
138    async fn process_claim(&self, claim: crate::schedule_store::ClaimedSchedule, runner_id: &str) {
139        let schedule = &claim.schedule;
140        tracing::debug!(
141            runner_id,
142            schedule_id = %schedule.id,
143            session_id = %schedule.session_id,
144            "Delivering local schedule"
145        );
146        let delivery = self.deliver_with_heartbeat(&claim, runner_id).await;
147        match delivery {
148            Ok(()) => match self.store.complete_delivery(&claim, runner_id, Utc::now()) {
149                Ok(()) => tracing::info!(
150                    runner_id,
151                    schedule_id = %schedule.id,
152                    session_id = %schedule.session_id,
153                    "Local schedule delivered"
154                ),
155                Err(error) => tracing::error!(
156                    runner_id,
157                    schedule_id = %schedule.id,
158                    session_id = %schedule.session_id,
159                    error = %error,
160                    "Local schedule was delivered but its claim could not be completed"
161                ),
162            },
163            Err(error) => {
164                let error_text = bounded_error(&error.to_string());
165                if let Err(release_error) =
166                    self.store
167                        .fail_delivery(&claim, runner_id, Utc::now(), &error_text)
168                {
169                    tracing::error!(
170                        runner_id,
171                        schedule_id = %schedule.id,
172                        session_id = %schedule.session_id,
173                        error = %release_error,
174                        "Local schedule failed and its claim could not be released"
175                    );
176                }
177                tracing::warn!(
178                    runner_id,
179                    schedule_id = %schedule.id,
180                    session_id = %schedule.session_id,
181                    error = %error,
182                    "Local schedule delivery failed; occurrence retained for retry"
183                );
184            }
185        }
186    }
187
188    async fn deliver_with_heartbeat(
189        &self,
190        claim: &crate::schedule_store::ClaimedSchedule,
191        runner_id: &str,
192    ) -> everruns_core::Result<()> {
193        let schedule = &claim.schedule;
194        let delivery = self
195            .session_runner
196            .send_message(schedule.session_id, &schedule.description);
197        tokio::pin!(delivery);
198        let heartbeat_interval = (self.config.claim_timeout / 3).max(Duration::from_millis(1));
199        let mut heartbeat = tokio::time::interval(heartbeat_interval);
200        heartbeat.set_missed_tick_behavior(MissedTickBehavior::Skip);
201        heartbeat.tick().await;
202        loop {
203            tokio::select! {
204                result = &mut delivery => return result,
205                _ = heartbeat.tick() => {
206                    if !self.store.renew_claim(claim, runner_id, Utc::now())? {
207                        return Err(everruns_core::AgentLoopError::store(format!(
208                            "local schedule claim {} was lost during delivery",
209                            claim.claim_id
210                        )));
211                    }
212                }
213            }
214        }
215    }
216}
217
218fn bounded_error(error: &str) -> String {
219    const MAX_ERROR_BYTES: usize = 4_096;
220    if error.len() <= MAX_ERROR_BYTES {
221        return error.to_string();
222    }
223    let mut end = MAX_ERROR_BYTES;
224    while !error.is_char_boundary(end) {
225        end -= 1;
226    }
227    format!("{}…", &error[..end])
228}
229
230/// Owned lifecycle handle for a running local schedule executor.
231pub struct LocalScheduleRunnerHandle {
232    shutdown_tx: watch::Sender<bool>,
233    join: Option<JoinHandle<()>>,
234}
235
236impl LocalScheduleRunnerHandle {
237    /// Stop claiming new schedules and wait for the current delivery to finish.
238    pub async fn shutdown(mut self) -> LocalResult<()> {
239        let _ = self.shutdown_tx.send(true);
240        if let Some(join) = self.join.take() {
241            join.await.map_err(|error| {
242                LocalError::Other(format!("local schedule runner task failed: {error}"))
243            })?;
244        }
245        Ok(())
246    }
247
248    /// Stop immediately. Unfinished claims become eligible after `claim_timeout`.
249    pub fn abort(mut self) {
250        if let Some(join) = self.join.take() {
251            join.abort();
252        }
253    }
254}
255
256impl Drop for LocalScheduleRunnerHandle {
257    fn drop(&mut self) {
258        let _ = self.shutdown_tx.send(true);
259    }
260}
everruns_local/schedule_runner.rs

everruns_local/
schedule_runner.rs
