ravenrs_report_server/
scheduler.rs

1//! Runs jobs: what is due on the schedule, or what is asked for by hand, with
2//! retries, a history of who received what, and retention of the files.
3
4use std::collections::HashSet;
5use std::fmt;
6use std::sync::{Arc, Mutex};
7use std::time::Duration;
8
9use jiff::tz::TimeZone;
10use jiff::{SignedDuration, Timestamp};
11use ravenrs_report::Report;
12use tokio::sync::Semaphore;
13use tokio::task::JoinHandle;
14
15use crate::config::Config;
16use crate::cron::Schedule;
17use crate::deliver::{Mail, Mailer, write_to_folder};
18use crate::job::{Delivery, Job};
19use crate::render::{Rendered, local_date_time, parameter_values, render, resolve_data};
20use crate::store::{Delivered, JobState, RunRecord, RunStatus, Store};
21
22/// Keeps a job from running twice at once.
23struct RunningGuard {
24    running: Arc<Mutex<HashSet<String>>>,
25    id: String,
26}
27
28impl Drop for RunningGuard {
29    fn drop(&mut self) {
30        if let Ok(mut running) = self.running.lock() {
31            running.remove(&self.id);
32        }
33    }
34}
35
36/// A delivery still to be made.
37#[derive(Clone, Debug)]
38enum Pending {
39    Email { to: Vec<String>, subject: Option<String>, body: Option<String> },
40    Folder { path: String },
41}
42
43/// What a run has done so far, kept across its attempts: a retry renders once
44/// and does not send again to whoever already received the files.
45struct Progress {
46    rendered: Option<Rendered>,
47    layout_version: Option<u32>,
48    pending: Vec<Pending>,
49    delivered: Vec<Delivered>,
50    errors: Vec<String>,
51    skipped_empty: bool,
52    /// The attempt died midway, so what was sent is unknown and retrying could send twice.
53    stopped: bool,
54}
55
56fn pending_deliveries(job: &Job) -> Vec<Pending> {
57    let mut pending = Vec::new();
58    for delivery in &job.deliveries {
59        match delivery {
60            Delivery::Email { to, separately: true, subject, body } => pending.extend(to.iter().map(|one| {
61                Pending::Email { to: vec![one.clone()], subject: subject.clone(), body: body.clone() }
62            })),
63            Delivery::Email { to, subject, body, .. } => {
64                pending.push(Pending::Email { to: to.clone(), subject: subject.clone(), body: body.clone() });
65            }
66            Delivery::Folder { path } => pending.push(Pending::Folder { path: path.clone() }),
67        }
68    }
69    pending
70}
71
72fn fill(template: &str, report: &str, date: &str) -> String {
73    template.replace("{report}", report).replace("{date}", date)
74}
75
76/// Runs jobs. Clones share the same limits.
77#[derive(Clone)]
78pub struct Runner {
79    store: Store,
80    config: Arc<Config>,
81    mailer: Option<Arc<dyn Mailer>>,
82    permits: Arc<Semaphore>,
83    running: Arc<Mutex<HashSet<String>>>,
84}
85
86impl fmt::Debug for Runner {
87    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88        f.debug_struct("Runner").field("store", &self.store).finish_non_exhaustive()
89    }
90}
91
92impl Runner {
93    /// A runner over `store`. Email deliveries fail without a `mailer`.
94    pub fn new(store: Store, config: Arc<Config>, mailer: Option<Arc<dyn Mailer>>) -> Self {
95        let permits = Arc::new(Semaphore::new(config.max_concurrent_renders.max(1)));
96        Self { store, config, mailer, permits, running: Arc::new(Mutex::new(HashSet::new())) }
97    }
98
99    /// The limit on simultaneous renders, shared with on-demand requests.
100    pub fn permits(&self) -> Arc<Semaphore> {
101        self.permits.clone()
102    }
103
104    /// Checks the schedule forever.
105    pub async fn run_forever(&self, period: Duration) {
106        let mut interval = tokio::time::interval(period);
107        loop {
108            interval.tick().await;
109            self.tick(Timestamp::now());
110        }
111    }
112
113    /// Starts whatever is due at `now`, and returns the runs it started.
114    ///
115    /// The next run is saved before a run starts, so a crash cannot repeat one.
116    pub fn tick(&self, now: Timestamp) -> Vec<JoinHandle<RunRecord>> {
117        let ids = match self.store.job_ids() {
118            Ok(ids) => ids,
119            Err(e) => {
120                tracing::error!(error = %e, "could not list jobs");
121                return Vec::new();
122            }
123        };
124        let mut started = Vec::new();
125        for id in ids {
126            match self.check_job(&id, now) {
127                Ok(Some(handle)) => started.push(handle),
128                Ok(None) => {}
129                Err(e) => tracing::error!(job = %id, error = %e, "could not check the job"),
130            }
131        }
132        started
133    }
134
135    fn check_job(&self, id: &str, now: Timestamp) -> Result<Option<JoinHandle<RunRecord>>, String> {
136        let Some(job) = self.store.job(id).map_err(|e| e.to_string())? else { return Ok(None) };
137        if !job.enabled {
138            return Ok(None);
139        }
140        let schedule = Schedule::parse(&job.schedule)?;
141        let zone = TimeZone::get(&job.timezone).map_err(|e| e.to_string())?;
142        let state = self.store.job_state(id).map_err(|e| e.to_string())?;
143        let next = schedule.next_after(&now.to_zoned(zone)).map(|z| z.timestamp());
144        let Some(due) = state.next_run else {
145            self.store.save_job_state(id, &JobState { next_run: next }).map_err(|e| e.to_string())?;
146            return Ok(None);
147        };
148        if due > now {
149            return Ok(None);
150        }
151        self.store.save_job_state(id, &JobState { next_run: next }).map_err(|e| e.to_string())?;
152        let grace = SignedDuration::from_secs(self.config.misfire_grace_minutes.saturating_mul(60) as i64);
153        if now.duration_since(due) > grace {
154            tracing::warn!(job = %id, due = %due, "the run is too late; recording it as missed");
155            self.record_missed(id, due, now, "the server was not running at the scheduled time");
156            return Ok(None);
157        }
158        let runner = self.clone();
159        let id = id.to_owned();
160        Ok(Some(tokio::spawn(async move { runner.run(&id, &job, due, false).await })))
161    }
162
163    fn record_missed(&self, id: &str, due: Timestamp, now: Timestamp, why: &str) -> RunRecord {
164        let record = RunRecord {
165            id: self.store.new_run_id(now),
166            job: id.to_owned(),
167            layout_version: None,
168            scheduled_for: due,
169            started_at: now,
170            finished_at: now,
171            manual: false,
172            status: RunStatus::Missed,
173            attempts: 0,
174            rows: None,
175            delivered: Vec::new(),
176            errors: vec![why.to_owned()],
177            report_warnings: Vec::new(),
178            documents: Vec::new(),
179        };
180        if let Err(e) = self.store.save_run(&record, self.config.history_limit) {
181            tracing::error!(job = %id, error = %e, "could not save the run");
182        }
183        record
184    }
185
186    /// Runs a job now, whatever its schedule, and returns what happened.
187    pub async fn run_now(&self, id: &str, job: &Job) -> RunRecord {
188        self.run(id, job, Timestamp::now(), true).await
189    }
190
191    async fn run(&self, id: &str, job: &Job, scheduled_for: Timestamp, manual: bool) -> RunRecord {
192        let started_at = Timestamp::now();
193        let free = self.running.lock().is_ok_and(|mut running| running.insert(id.to_owned()));
194        if !free {
195            return self.record_missed(id, scheduled_for, started_at, "the previous run was still going");
196        }
197        let guard = RunningGuard { running: self.running.clone(), id: id.to_owned() };
198        let mut progress = Progress {
199            rendered: None,
200            layout_version: None,
201            pending: pending_deliveries(job),
202            delivered: Vec::new(),
203            errors: Vec::new(),
204            skipped_empty: false,
205            stopped: false,
206        };
207        let mut attempts = 0;
208        loop {
209            attempts += 1;
210            progress = self.attempt(job, scheduled_for, progress).await;
211            let done = progress.skipped_empty || (progress.pending.is_empty() && progress.errors.is_empty());
212            if done || progress.stopped || attempts > job.retries {
213                break;
214            }
215            let wait = self.config.retry_delay_seconds.saturating_mul(1 << (attempts - 1).min(16));
216            tracing::warn!(job = %id, attempt = attempts, errors = ?progress.errors, wait, "the run failed; trying again");
217            tokio::time::sleep(Duration::from_secs(wait)).await;
218        }
219        drop(guard);
220        let record = self.finish(id, job, scheduled_for, started_at, manual, attempts, progress);
221        tracing::info!(job = %id, run = %record.id, status = ?record.status, "run finished");
222        record
223    }
224
225    async fn attempt(&self, job: &Job, scheduled_for: Timestamp, progress: Progress) -> Progress {
226        let permit = self.permits.clone().acquire_owned().await;
227        let runner = self.clone();
228        let task_job = job.clone();
229        let task = tokio::task::spawn_blocking(move || {
230            let _permit = permit;
231            runner.attempt_blocking(&task_job, scheduled_for, progress)
232        });
233        match task.await {
234            Ok(progress) => progress,
235            // A panic: this attempt's state is gone with it.
236            Err(e) => Progress {
237                rendered: None,
238                layout_version: None,
239                pending: Vec::new(),
240                delivered: Vec::new(),
241                errors: vec![format!("the run stopped unexpectedly: {e}")],
242                skipped_empty: false,
243                stopped: true,
244            },
245        }
246    }
247
248    fn attempt_blocking(&self, job: &Job, scheduled_for: Timestamp, mut progress: Progress) -> Progress {
249        progress.errors.clear();
250        let Ok(zone) = TimeZone::get(&job.timezone) else {
251            progress.errors.push(format!("unknown time zone '{}'", job.timezone));
252            return progress;
253        };
254        let local = scheduled_for.to_zoned(zone);
255        let date = local.strftime("%Y-%m-%d").to_string();
256        if progress.rendered.is_none() {
257            match self.render_job(job, &local, &date) {
258                Ok((version, rendered)) => {
259                    progress.layout_version = Some(version);
260                    progress.rendered = Some(rendered);
261                }
262                Err(e) => {
263                    progress.errors.push(e);
264                    return progress;
265                }
266            }
267        }
268        let Some(rendered) = progress.rendered.take() else { return progress };
269        if job.skip_if_empty && rendered.rows == 0 {
270            progress.skipped_empty = true;
271            progress.pending.clear();
272            progress.rendered = Some(rendered);
273            return progress;
274        }
275        let mut remaining = Vec::new();
276        for pending in std::mem::take(&mut progress.pending) {
277            match self.deliver(&pending, &rendered, &date) {
278                Ok(delivered) => progress.delivered.push(delivered),
279                Err(e) => {
280                    progress.errors.push(e);
281                    remaining.push(pending);
282                }
283            }
284        }
285        progress.pending = remaining;
286        progress.rendered = Some(rendered);
287        progress
288    }
289
290    fn render_job(&self, job: &Job, local: &jiff::Zoned, date: &str) -> Result<(u32, Rendered), String> {
291        let (version, text) = self
292            .store
293            .layout(&job.layout, None)
294            .map_err(|e| e.to_string())?
295            .ok_or_else(|| format!("the layout '{}' does not exist", job.layout))?;
296        let report = Report::from_json(&text).map_err(|e| format!("layout '{}': {e}", job.layout))?;
297        let now = local_date_time(local).ok_or("the date is out of range")?;
298        let parameters = parameter_values(&job.parameters, now)?;
299        let data = resolve_data(&report, &parameters, &self.config)?;
300        Ok((version, render(&report, &data, parameters, now, &job.formats, date)?))
301    }
302
303    fn deliver(&self, pending: &Pending, rendered: &Rendered, date: &str) -> Result<Delivered, String> {
304        let files: Vec<String> = rendered.files.iter().map(|f| f.name.clone()).collect();
305        let at = Timestamp::now();
306        match pending {
307            Pending::Email { to, subject, body } => {
308                let mailer = self.mailer.as_ref().ok_or("the server has no mail server configured")?;
309                let target = to.join(", ");
310                let subject = fill(subject.as_deref().unwrap_or("{report} - {date}"), &rendered.title, date);
311                let body = fill(
312                    body.as_deref().unwrap_or("{report}, {date}. The report is attached."),
313                    &rendered.title,
314                    date,
315                );
316                mailer
317                    .send(&Mail { to: to.clone(), subject, body, attachments: rendered.files.clone() })
318                    .map_err(|e| format!("email to {target}: {e}"))?;
319                Ok(Delivered { kind: "email".to_owned(), target, files, at })
320            }
321            Pending::Folder { path } => {
322                let base = self.config.output_dir.as_ref().ok_or("the server has no outputDir configured")?;
323                write_to_folder(base, path, &rendered.files)?;
324                Ok(Delivered { kind: "folder".to_owned(), target: path.clone(), files, at })
325            }
326        }
327    }
328
329    #[allow(clippy::too_many_arguments, reason = "one private step of a run")]
330    fn finish(
331        &self,
332        id: &str,
333        job: &Job,
334        scheduled_for: Timestamp,
335        started_at: Timestamp,
336        manual: bool,
337        attempts: u32,
338        progress: Progress,
339    ) -> RunRecord {
340        let now = Timestamp::now();
341        let run_id = self.store.new_run_id(started_at);
342        let status = if progress.skipped_empty {
343            RunStatus::SkippedEmpty
344        } else if progress.errors.is_empty() && progress.pending.is_empty() {
345            RunStatus::Success
346        } else if progress.delivered.is_empty() {
347            RunStatus::Failed
348        } else {
349            RunStatus::Partial
350        };
351        let mut documents = Vec::new();
352        if job.retention_days > 0
353            && !progress.skipped_empty
354            && let Some(rendered) = &progress.rendered
355        {
356            for file in &rendered.files {
357                match self.store.save_document(id, &run_id, &file.name, &file.bytes) {
358                    Ok(()) => documents.push(file.name.clone()),
359                    Err(e) => {
360                        tracing::error!(job = %id, file = %file.name, error = %e, "could not keep the file")
361                    }
362                }
363            }
364        }
365        let record = RunRecord {
366            id: run_id,
367            job: id.to_owned(),
368            layout_version: progress.layout_version,
369            scheduled_for,
370            started_at,
371            finished_at: now,
372            manual,
373            status,
374            attempts,
375            rows: progress.rendered.as_ref().map(|r| r.rows),
376            delivered: progress.delivered,
377            errors: progress.errors,
378            report_warnings: progress.rendered.map(|r| r.warnings).unwrap_or_default(),
379            documents,
380        };
381        if let Err(e) = self.store.save_run(&record, self.config.history_limit) {
382            tracing::error!(job = %id, error = %e, "could not save the run");
383        }
384        let keep = SignedDuration::from_hours(i64::from(job.retention_days).saturating_mul(24));
385        if let Ok(cutoff) = now.checked_sub(keep)
386            && let Err(e) = self.store.purge_documents(id, cutoff)
387        {
388            tracing::error!(job = %id, error = %e, "could not delete old files");
389        }
390        record
391    }
392}
ravenrs_report_server/scheduler.rs

ravenrs_report_server/
scheduler.rs
