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

use std::fmt;
use std::sync::Arc;

use a3s_boot::{BootError, Queue, QueueJob, QueueJobReceipt};
use async_trait::async_trait;

use crate::engine::FlowEngine;
use crate::error::{FlowError, Result};

use super::runner::handle_flow_task;
use super::{FlowTask, FlowTaskDispatcher};

const DEFAULT_FLOW_JOB_NAME: &str = "a3s.flow.task";

/// A3S Boot-backed task manager for Flow scheduler and callback dispatch.
///
/// Boot owns queue processors, worker lifecycle, leasing, job state, and
/// shutdown. Flow owns only task serialization and engine handling semantics.
#[derive(Clone)]
pub struct BootFlowTaskManager {
    engine: FlowEngine,
    queue: Arc<Queue>,
    job_name: String,
}

impl fmt::Debug for BootFlowTaskManager {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("BootFlowTaskManager")
            .field("queue", &self.queue.name())
            .field("job_name", &self.job_name)
            .finish_non_exhaustive()
    }
}

impl BootFlowTaskManager {
    pub fn new(engine: FlowEngine, queue: Arc<Queue>) -> Self {
        Self {
            engine,
            queue,
            job_name: DEFAULT_FLOW_JOB_NAME.to_string(),
        }
    }

    pub fn with_job_name(mut self, job_name: impl Into<String>) -> Result<Self> {
        let job_name = job_name.into().trim().to_string();
        if job_name.is_empty() {
            return Err(FlowError::InvalidWorkerConfiguration(
                "Boot Flow job name cannot be empty".to_string(),
            ));
        }
        self.job_name = job_name;
        Ok(self)
    }

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

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

    pub fn job_name(&self) -> &str {
        &self.job_name
    }

    /// Register the Flow task processor with the Boot queue.
    ///
    /// The host still starts and stops the queue through `QueueModule` or the
    /// corresponding `Queue::start` and `Queue::shutdown` lifecycle calls.
    pub fn register(&self) -> Result<()> {
        let engine = self.engine.clone();
        self.queue
            .process(self.job_name.clone(), move |job: QueueJob, _context| {
                let engine = engine.clone();
                async move {
                    let task = job.data_as::<FlowTask>()?;
                    handle_flow_task(&engine, task).await.map_err(|error| {
                        BootError::Internal(format!("A3S Flow task handling failed: {error}"))
                    })?;
                    Ok(())
                }
            })
            .map_err(boot_error)
    }

    pub async fn enqueue_with_receipt(&self, task: FlowTask) -> Result<QueueJobReceipt> {
        self.queue
            .enqueue(self.job_name.clone(), &task)
            .await
            .map_err(boot_error)
    }
}

#[async_trait]
impl FlowTaskDispatcher for BootFlowTaskManager {
    async fn dispatch(&self, task: FlowTask) -> Result<()> {
        self.enqueue_with_receipt(task).await.map(|_| ())
    }
}

fn boot_error(error: BootError) -> FlowError {
    FlowError::TaskManagement(format!("A3S Boot queue error: {error}"))
}