a3s-flow 0.9.0

Durable workflow engine and Rust SDK for A3S
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::Duration;

use crate::engine::FlowEngine;
use crate::error::Result;
use crate::model::{ScheduledWakeupKind, WorkflowRunSuspension};
use crate::worker::{FlowTask, FlowTaskDispatcher};

/// Result of one scheduler scan.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct FlowSchedulerTick {
    pub due_waits: Vec<(String, String)>,
    pub due_retries: Vec<(String, String)>,
    pub enqueued_tasks: usize,
}

impl FlowSchedulerTick {
    pub fn has_due_work(&self) -> bool {
        !self.due_waits.is_empty() || !self.due_retries.is_empty()
    }
}

/// Scheduler that scans durable state and enqueues due workflow work.
#[derive(Clone)]
pub struct FlowScheduler {
    engine: FlowEngine,
    dispatcher: Arc<dyn FlowTaskDispatcher>,
}

impl FlowScheduler {
    pub fn new(engine: FlowEngine, dispatcher: Arc<dyn FlowTaskDispatcher>) -> Self {
        Self { engine, dispatcher }
    }

    pub fn engine(&self) -> &FlowEngine {
        &self.engine
    }

    pub fn dispatcher(&self) -> Arc<dyn FlowTaskDispatcher> {
        Arc::clone(&self.dispatcher)
    }

    /// Backward-compatible name for [`Self::dispatcher`].
    #[deprecated(since = "0.4.4", note = "use dispatcher()")]
    pub fn queue(&self) -> Arc<dyn FlowTaskDispatcher> {
        self.dispatcher()
    }

    /// Return the earliest wait or delayed retry that can wake the scheduler.
    pub async fn next_wakeup(&self, now: DateTime<Utc>) -> Result<Option<WorkflowRunSuspension>> {
        self.engine.next_wakeup(now).await
    }

    /// Return how long the host can sleep before the next scheduled wake-up.
    ///
    /// Due or overdue wake-ups return `Duration::ZERO`. Active hooks are not
    /// represented because external callbacks are pushed into the queue by the
    /// callback router instead of time.
    pub async fn next_wakeup_delay(&self, now: DateTime<Utc>) -> Result<Option<Duration>> {
        let Some(wakeup) = self.next_wakeup(now).await? else {
            return Ok(None);
        };
        let Some(scheduled_at) = wakeup.scheduled_at() else {
            return Ok(None);
        };
        Ok(Some(
            scheduled_at
                .signed_duration_since(now)
                .to_std()
                .unwrap_or(Duration::ZERO),
        ))
    }

    pub async fn enqueue_due_work(&self, now: DateTime<Utc>) -> Result<FlowSchedulerTick> {
        let due = self.engine.list_due_wakeups(now).await?;
        let due_waits = due
            .iter()
            .filter(|wakeup| wakeup.kind == ScheduledWakeupKind::Wait)
            .map(|wakeup| (wakeup.run_id.clone(), wakeup.subject_id.clone()))
            .collect::<Vec<_>>();
        let due_retries = due
            .into_iter()
            .filter(|wakeup| wakeup.kind == ScheduledWakeupKind::Retry)
            .map(|wakeup| (wakeup.run_id, wakeup.subject_id))
            .collect::<Vec<_>>();
        let mut enqueued_tasks = 0usize;

        if !due_waits.is_empty() {
            self.dispatcher
                .dispatch(FlowTask::ResumeDueWaits { now })
                .await?;
            enqueued_tasks += 1;
        }
        if !due_retries.is_empty() {
            self.dispatcher
                .dispatch(FlowTask::ResumeDueRetries { now })
                .await?;
            enqueued_tasks += 1;
        }

        Ok(FlowSchedulerTick {
            due_waits,
            due_retries,
            enqueued_tasks,
        })
    }
}