a3s_flow/
scheduler.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use std::sync::Arc;
4use std::time::Duration;
5
6use crate::engine::FlowEngine;
7use crate::error::Result;
8use crate::model::WorkflowRunSuspension;
9use crate::worker::{FlowTask, FlowTaskDispatcher};
10
11/// Result of one scheduler scan.
12#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
13pub struct FlowSchedulerTick {
14    pub due_waits: Vec<(String, String)>,
15    pub due_retries: Vec<(String, String)>,
16    pub enqueued_tasks: usize,
17}
18
19impl FlowSchedulerTick {
20    pub fn has_due_work(&self) -> bool {
21        !self.due_waits.is_empty() || !self.due_retries.is_empty()
22    }
23}
24
25/// Scheduler that scans durable state and enqueues due workflow work.
26#[derive(Clone)]
27pub struct FlowScheduler {
28    engine: FlowEngine,
29    dispatcher: Arc<dyn FlowTaskDispatcher>,
30}
31
32impl FlowScheduler {
33    pub fn new(engine: FlowEngine, dispatcher: Arc<dyn FlowTaskDispatcher>) -> Self {
34        Self { engine, dispatcher }
35    }
36
37    pub fn engine(&self) -> &FlowEngine {
38        &self.engine
39    }
40
41    pub fn dispatcher(&self) -> Arc<dyn FlowTaskDispatcher> {
42        Arc::clone(&self.dispatcher)
43    }
44
45    /// Backward-compatible name for [`Self::dispatcher`].
46    #[deprecated(since = "0.4.4", note = "use dispatcher()")]
47    pub fn queue(&self) -> Arc<dyn FlowTaskDispatcher> {
48        self.dispatcher()
49    }
50
51    /// Return the earliest wait or delayed retry that can wake the scheduler.
52    pub async fn next_wakeup(&self, now: DateTime<Utc>) -> Result<Option<WorkflowRunSuspension>> {
53        self.engine.next_wakeup(now).await
54    }
55
56    /// Return how long the host can sleep before the next scheduled wake-up.
57    ///
58    /// Due or overdue wake-ups return `Duration::ZERO`. Active hooks are not
59    /// represented because external callbacks are pushed into the queue by the
60    /// callback router instead of time.
61    pub async fn next_wakeup_delay(&self, now: DateTime<Utc>) -> Result<Option<Duration>> {
62        let Some(wakeup) = self.next_wakeup(now).await? else {
63            return Ok(None);
64        };
65        let Some(scheduled_at) = wakeup.scheduled_at() else {
66            return Ok(None);
67        };
68        Ok(Some(
69            scheduled_at
70                .signed_duration_since(now)
71                .to_std()
72                .unwrap_or(Duration::ZERO),
73        ))
74    }
75
76    pub async fn enqueue_due_work(&self, now: DateTime<Utc>) -> Result<FlowSchedulerTick> {
77        let due_waits = self.engine.list_due_waits(now).await?;
78        let due_retries = self.engine.list_due_retries(now).await?;
79        let mut enqueued_tasks = 0usize;
80
81        if !due_waits.is_empty() {
82            self.dispatcher
83                .dispatch(FlowTask::ResumeDueWaits { now })
84                .await?;
85            enqueued_tasks += 1;
86        }
87        if !due_retries.is_empty() {
88            self.dispatcher
89                .dispatch(FlowTask::ResumeDueRetries { now })
90                .await?;
91            enqueued_tasks += 1;
92        }
93
94        Ok(FlowSchedulerTick {
95            due_waits,
96            due_retries,
97            enqueued_tasks,
98        })
99    }
100}
a3s_flow/scheduler.rs

a3s_flow/
scheduler.rs
