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, FlowTaskQueue};
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    queue: Arc<dyn FlowTaskQueue>,
30}
31
32impl FlowScheduler {
33    pub fn new(engine: FlowEngine, queue: Arc<dyn FlowTaskQueue>) -> Self {
34        Self { engine, queue }
35    }
36
37    pub fn engine(&self) -> &FlowEngine {
38        &self.engine
39    }
40
41    pub fn queue(&self) -> Arc<dyn FlowTaskQueue> {
42        Arc::clone(&self.queue)
43    }
44
45    /// Return the earliest wait or delayed retry that can wake the scheduler.
46    pub async fn next_wakeup(&self, now: DateTime<Utc>) -> Result<Option<WorkflowRunSuspension>> {
47        self.engine.next_wakeup(now).await
48    }
49
50    /// Return how long the host can sleep before the next scheduled wake-up.
51    ///
52    /// Due or overdue wake-ups return `Duration::ZERO`. Active hooks are not
53    /// represented because external callbacks are pushed into the queue by the
54    /// callback router instead of time.
55    pub async fn next_wakeup_delay(&self, now: DateTime<Utc>) -> Result<Option<Duration>> {
56        let Some(wakeup) = self.next_wakeup(now).await? else {
57            return Ok(None);
58        };
59        let Some(scheduled_at) = wakeup.scheduled_at() else {
60            return Ok(None);
61        };
62        Ok(Some(
63            scheduled_at
64                .signed_duration_since(now)
65                .to_std()
66                .unwrap_or(Duration::ZERO),
67        ))
68    }
69
70    pub async fn enqueue_due_work(&self, now: DateTime<Utc>) -> Result<FlowSchedulerTick> {
71        let due_waits = self.engine.list_due_waits(now).await?;
72        let due_retries = self.engine.list_due_retries(now).await?;
73        let mut enqueued_tasks = 0usize;
74
75        if !due_waits.is_empty() {
76            self.queue.enqueue(FlowTask::ResumeDueWaits { now }).await?;
77            enqueued_tasks += 1;
78        }
79        if !due_retries.is_empty() {
80            self.queue
81                .enqueue(FlowTask::ResumeDueRetries { now })
82                .await?;
83            enqueued_tasks += 1;
84        }
85
86        Ok(FlowSchedulerTick {
87            due_waits,
88            due_retries,
89            enqueued_tasks,
90        })
91    }
92}
a3s_flow/scheduler.rs

a3s_flow/
scheduler.rs
