a3s_flow/worker/
task.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3
4use crate::model::{JsonValue, WorkflowSignal};
5
6/// Queueable unit of workflow engine work.
7#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
8#[non_exhaustive]
9#[serde(tag = "type", rename_all = "snake_case")]
10pub enum FlowTask {
11    /// Replays a run until it suspends or reaches a terminal state.
12    DriveRun {
13        /// Run to replay.
14        run_id: String,
15    },
16    /// Completes one durable timer wait and replays its run.
17    ResumeWait {
18        /// Run that owns the wait.
19        run_id: String,
20        /// Stable identity of the wait.
21        wait_id: String,
22    },
23    /// Delivers a payload to a hook addressed within a run.
24    ResumeHook {
25        /// Run that owns the hook.
26        run_id: String,
27        /// Stable identity of the hook.
28        hook_id: String,
29        /// JSON payload supplied by the external caller.
30        payload: JsonValue,
31    },
32    /// Delivers a payload to a hook addressed by its bearer token.
33    ResumeHookByToken {
34        /// Secret token assigned when the hook was created.
35        token: String,
36        /// JSON payload supplied by the external caller.
37        payload: JsonValue,
38    },
39    /// Delivers one named asynchronous signal to a run.
40    SendSignal {
41        /// Root or active run targeted by the delivery.
42        run_id: String,
43        /// Caller-identified signal delivery.
44        signal: WorkflowSignal,
45    },
46    /// Closes a hook without delivering a payload.
47    DisposeHook {
48        /// Run that owns the hook.
49        run_id: String,
50        /// Stable identity of the hook.
51        hook_id: String,
52    },
53    /// Closes a hook addressed by its bearer token.
54    DisposeHookByToken {
55        /// Secret token assigned when the hook was created.
56        token: String,
57    },
58    /// Drives due waits and retries for one targeted run.
59    ResumeScheduledRun {
60        /// Run whose scheduled work should be inspected.
61        run_id: String,
62        /// UTC cutoff used to determine readiness.
63        now: DateTime<Utc>,
64    },
65    /// Compatibility task that scans all runs for due timer waits.
66    ResumeDueWaits {
67        /// UTC cutoff used to determine readiness.
68        now: DateTime<Utc>,
69    },
70    /// Compatibility task that scans all runs for due delayed retries.
71    ResumeDueRetries {
72        /// UTC cutoff used to determine readiness.
73        now: DateTime<Utc>,
74    },
75}
76
77impl FlowTask {
78    /// Return the single run targeted by this task, when one is explicit.
79    ///
80    /// Public-token callbacks and compatibility-wide due scans require host
81    /// resolution before they can participate in exact runtime-build routing.
82    pub fn target_run_id(&self) -> Option<&str> {
83        match self {
84            Self::DriveRun { run_id }
85            | Self::ResumeWait { run_id, .. }
86            | Self::ResumeHook { run_id, .. }
87            | Self::SendSignal { run_id, .. }
88            | Self::DisposeHook { run_id, .. }
89            | Self::ResumeScheduledRun { run_id, .. } => Some(run_id),
90            Self::ResumeHookByToken { .. }
91            | Self::DisposeHookByToken { .. }
92            | Self::ResumeDueWaits { .. }
93            | Self::ResumeDueRetries { .. } => None,
94        }
95    }
96}
97
98/// Result of handling one queued [`FlowTask`].
99#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
100#[non_exhaustive]
101pub struct FlowTaskOutcome {
102    /// Task whose durable effects are summarized.
103    pub task: FlowTask,
104    /// Runs affected while handling this task.
105    ///
106    /// Run-targeted tasks report the active continuation leaf. Compatibility-wide
107    /// scan tasks retain their legacy scan and commit-report semantics.
108    pub run_ids: Vec<String>,
109    /// Wait completions committed by this task.
110    pub resumed_waits: Vec<(String, String)>,
111    /// Delayed retry wakeups driven by this task.
112    pub resumed_retries: Vec<(String, String)>,
113    /// Hook receipt committed by this task, excluding matching redelivery.
114    pub resumed_hook: Option<(String, String)>,
115    /// Hook disposal committed by this task, excluding matching redelivery.
116    #[serde(default)]
117    pub disposed_hook: Option<(String, String)>,
118    /// Signal receipt committed by this task, excluding matching redelivery.
119    #[serde(default)]
120    pub delivered_signal: Option<(String, String)>,
121}
122
123impl FlowTaskOutcome {
124    pub(super) fn new(task: FlowTask) -> Self {
125        Self {
126            task,
127            run_ids: Vec::new(),
128            resumed_waits: Vec::new(),
129            resumed_retries: Vec::new(),
130            resumed_hook: None,
131            disposed_hook: None,
132            delivered_signal: None,
133        }
134    }
135}
136
137/// Leased task returned by a queue worker before acknowledgement.
138///
139/// [`super::FlowTaskQueue::heartbeat`] replaces `lease_id` with a new fencing
140/// token. Callers that renew leases manually must acknowledge with the latest
141/// returned token.
142#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
143#[non_exhaustive]
144pub struct FlowTaskLease {
145    /// Current fencing token required for heartbeat and acknowledgement.
146    pub lease_id: String,
147    /// Leased Flow task payload.
148    pub task: FlowTask,
149}
150
151impl FlowTaskLease {
152    /// Create a leased task with the queue's current fencing token.
153    pub fn new(lease_id: impl Into<String>, task: FlowTask) -> Self {
154        Self {
155            lease_id: lease_id.into(),
156            task,
157        }
158    }
159}
160
161/// Task moved out of inflight dispatch after exceeding a local lease policy.
162#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
163#[non_exhaustive]
164pub struct LocalFileDeadLetteredTask {
165    /// Final lease token owned before dead-lettering.
166    pub lease_id: String,
167    /// Task removed from inflight dispatch.
168    pub task: FlowTask,
169    /// Queue policy reason for dead-lettering.
170    pub reason: String,
171    /// UTC time at which the task was moved.
172    pub dead_lettered_at: DateTime<Utc>,
173}
174
175/// Task moved out of Postgres inflight dispatch after exceeding a lease policy.
176#[cfg(feature = "postgres")]
177#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
178#[non_exhaustive]
179pub struct PostgresDeadLetteredTask {
180    /// Final lease token owned before dead-lettering.
181    pub lease_id: String,
182    /// Task removed from inflight dispatch.
183    pub task: FlowTask,
184    /// Queue policy reason for dead-lettering.
185    pub reason: String,
186    /// UTC time at which the task was moved.
187    pub dead_lettered_at: DateTime<Utc>,
188}
a3s_flow/worker/task.rs

a3s_flow/worker/
task.rs
