a3s_flow/worker/
queue.rs

1use async_trait::async_trait;
2
3use crate::error::Result;
4
5use super::{FlowTask, FlowTaskLease};
6
7/// Enqueue-only dispatch boundary used by schedulers and callback routers.
8#[async_trait]
9pub trait FlowTaskDispatcher: Send + Sync {
10    async fn dispatch(&self, task: FlowTask) -> Result<()>;
11}
12
13/// Queue abstraction for workflow dispatch.
14#[async_trait]
15pub trait FlowTaskQueue: Send + Sync {
16    async fn enqueue(&self, task: FlowTask) -> Result<()>;
17
18    async fn lease(&self) -> Result<Option<FlowTaskLease>>;
19
20    /// Refreshes an active lease and returns its replacement fencing token.
21    ///
22    /// The previous lease ID becomes invalid as soon as this call succeeds.
23    /// Workers must acknowledge with the most recently returned lease ID.
24    async fn heartbeat(&self, lease_id: &str) -> Result<String>;
25
26    /// Acknowledges the active lease identified by its latest fencing token.
27    ///
28    /// Implementations return [`crate::FlowError::LeaseLost`] when the token is
29    /// stale or the task has already been reclaimed, acknowledged, or moved to
30    /// a dead-letter queue.
31    async fn ack(&self, lease_id: &str) -> Result<()>;
32
33    async fn requeue_inflight(&self) -> Result<usize> {
34        Ok(0)
35    }
36
37    async fn dequeue(&self) -> Result<Option<FlowTask>> {
38        let Some(lease) = self.lease().await? else {
39            return Ok(None);
40        };
41        let task = lease.task.clone();
42        self.ack(&lease.lease_id).await?;
43        Ok(Some(task))
44    }
45
46    async fn len(&self) -> Result<usize>;
47
48    async fn is_empty(&self) -> Result<bool> {
49        Ok(self.len().await? == 0)
50    }
51}
52
53#[async_trait]
54impl<T> FlowTaskDispatcher for T
55where
56    T: FlowTaskQueue + ?Sized,
57{
58    async fn dispatch(&self, task: FlowTask) -> Result<()> {
59        FlowTaskQueue::enqueue(self, task).await
60    }
61}
a3s_flow/worker/queue.rs

a3s_flow/worker/
queue.rs
