a3s_flow/worker/
queue.rs

1use async_trait::async_trait;
2
3use crate::error::Result;
4use crate::runtime_build::RuntimeBuildId;
5
6use super::{FlowTask, FlowTaskLease};
7
8/// Enqueue-only dispatch boundary used by schedulers and callback routers.
9#[async_trait]
10pub trait FlowTaskDispatcher: Send + Sync {
11    /// Dispatches one Flow task to the configured execution route.
12    async fn dispatch(&self, task: FlowTask) -> Result<()>;
13
14    /// Return whether this dispatcher has an explicit compatible route.
15    fn has_runtime_build_route(&self, required_build_id: Option<&RuntimeBuildId>) -> bool {
16        required_build_id.is_none()
17    }
18
19    /// Fail before dispatch when no compatible route is registered.
20    fn ensure_runtime_build_route(&self, required_build_id: Option<&RuntimeBuildId>) -> Result<()> {
21        if self.has_runtime_build_route(required_build_id) {
22            return Ok(());
23        }
24        Err(crate::FlowError::RuntimeBuildRouteNotFound {
25            required_build_id: required_build_id.cloned(),
26        })
27    }
28
29    /// Dispatch to a route that explicitly serves `required_build_id`.
30    ///
31    /// Ordinary queues accept legacy unpinned tasks only. Pinned workflows
32    /// fail closed unless a build-aware dispatcher such as
33    /// [`RuntimeBuildTaskRouter`](super::RuntimeBuildTaskRouter) selects a
34    /// concrete route.
35    async fn dispatch_for_runtime_build(
36        &self,
37        required_build_id: Option<&RuntimeBuildId>,
38        task: FlowTask,
39    ) -> Result<()> {
40        self.ensure_runtime_build_route(required_build_id)?;
41        self.dispatch(task).await
42    }
43}
44
45/// Queue abstraction for workflow dispatch.
46#[async_trait]
47pub trait FlowTaskQueue: Send + Sync {
48    /// Appends one task to pending dispatch.
49    async fn enqueue(&self, task: FlowTask) -> Result<()>;
50
51    /// Leases the next pending task without acknowledging it.
52    async fn lease(&self) -> Result<Option<FlowTaskLease>>;
53
54    /// Refreshes an active lease and returns its replacement fencing token.
55    ///
56    /// The previous lease ID becomes invalid as soon as this call succeeds.
57    /// Workers must acknowledge with the most recently returned lease ID.
58    async fn heartbeat(&self, lease_id: &str) -> Result<String>;
59
60    /// Acknowledges the active lease identified by its latest fencing token.
61    ///
62    /// Implementations return [`crate::FlowError::LeaseLost`] when the token is
63    /// stale or the task has already been reclaimed, acknowledged, or moved to
64    /// a dead-letter queue.
65    async fn ack(&self, lease_id: &str) -> Result<()>;
66
67    /// Returns inflight tasks to pending dispatch and reports the count.
68    async fn requeue_inflight(&self) -> Result<usize> {
69        Ok(0)
70    }
71
72    /// Leases and immediately acknowledges the next pending task.
73    async fn dequeue(&self) -> Result<Option<FlowTask>> {
74        let Some(lease) = self.lease().await? else {
75            return Ok(None);
76        };
77        let task = lease.task.clone();
78        self.ack(&lease.lease_id).await?;
79        Ok(Some(task))
80    }
81
82    /// Returns the number of tasks pending dispatch.
83    async fn len(&self) -> Result<usize>;
84
85    /// Returns whether no tasks are pending dispatch.
86    async fn is_empty(&self) -> Result<bool> {
87        Ok(self.len().await? == 0)
88    }
89}
90
91#[async_trait]
92impl<T> FlowTaskDispatcher for T
93where
94    T: FlowTaskQueue + ?Sized,
95{
96    async fn dispatch(&self, task: FlowTask) -> Result<()> {
97        FlowTaskQueue::enqueue(self, task).await
98    }
99}
a3s_flow/worker/queue.rs

a3s_flow/worker/
queue.rs
