a3s-flow 0.4.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

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::WorkflowRunSuspension;
use crate::worker::{FlowTask, FlowTaskQueue};

/// 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,
    queue: Arc<dyn FlowTaskQueue>,
}

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

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

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

    /// 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_waits = self.engine.list_due_waits(now).await?;
        let due_retries = self.engine.list_due_retries(now).await?;
        let mut enqueued_tasks = 0usize;

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

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