a3s_flow/worker/
task.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3
4use crate::model::JsonValue;
5
6/// Queueable unit of workflow engine work.
7#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
8#[serde(tag = "type", rename_all = "snake_case")]
9pub enum FlowTask {
10    DriveRun {
11        run_id: String,
12    },
13    ResumeWait {
14        run_id: String,
15        wait_id: String,
16    },
17    ResumeHook {
18        run_id: String,
19        hook_id: String,
20        payload: JsonValue,
21    },
22    ResumeHookByToken {
23        token: String,
24        payload: JsonValue,
25    },
26    DisposeHook {
27        run_id: String,
28        hook_id: String,
29    },
30    DisposeHookByToken {
31        token: String,
32    },
33    ResumeScheduledRun {
34        run_id: String,
35        now: DateTime<Utc>,
36    },
37    ResumeDueWaits {
38        now: DateTime<Utc>,
39    },
40    ResumeDueRetries {
41        now: DateTime<Utc>,
42    },
43}
44
45impl FlowTask {
46    /// Return the single run targeted by this task, when one is explicit.
47    ///
48    /// Public-token callbacks and compatibility-wide due scans require host
49    /// resolution before they can participate in exact runtime-build routing.
50    pub fn target_run_id(&self) -> Option<&str> {
51        match self {
52            Self::DriveRun { run_id }
53            | Self::ResumeWait { run_id, .. }
54            | Self::ResumeHook { run_id, .. }
55            | Self::DisposeHook { run_id, .. }
56            | Self::ResumeScheduledRun { run_id, .. } => Some(run_id),
57            Self::ResumeHookByToken { .. }
58            | Self::DisposeHookByToken { .. }
59            | Self::ResumeDueWaits { .. }
60            | Self::ResumeDueRetries { .. } => None,
61        }
62    }
63}
64
65/// Result of handling one queued [`FlowTask`].
66#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
67pub struct FlowTaskOutcome {
68    pub task: FlowTask,
69    pub run_ids: Vec<String>,
70    pub resumed_waits: Vec<(String, String)>,
71    pub resumed_retries: Vec<(String, String)>,
72    pub resumed_hook: Option<(String, String)>,
73    #[serde(default)]
74    pub disposed_hook: Option<(String, String)>,
75}
76
77impl FlowTaskOutcome {
78    pub(super) fn new(task: FlowTask) -> Self {
79        Self {
80            task,
81            run_ids: Vec::new(),
82            resumed_waits: Vec::new(),
83            resumed_retries: Vec::new(),
84            resumed_hook: None,
85            disposed_hook: None,
86        }
87    }
88}
89
90/// Leased task returned by a queue worker before acknowledgement.
91///
92/// [`super::FlowTaskQueue::heartbeat`] replaces `lease_id` with a new fencing
93/// token. Callers that renew leases manually must acknowledge with the latest
94/// returned token.
95#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
96pub struct FlowTaskLease {
97    pub lease_id: String,
98    pub task: FlowTask,
99}
100
101/// Task moved out of inflight dispatch after exceeding a local lease policy.
102#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
103pub struct LocalFileDeadLetteredTask {
104    pub lease_id: String,
105    pub task: FlowTask,
106    pub reason: String,
107    pub dead_lettered_at: DateTime<Utc>,
108}
109
110/// Task moved out of Postgres inflight dispatch after exceeding a lease policy.
111#[cfg(feature = "postgres")]
112#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
113pub struct PostgresDeadLetteredTask {
114    pub lease_id: String,
115    pub task: FlowTask,
116    pub reason: String,
117    pub dead_lettered_at: DateTime<Utc>,
118}
a3s_flow/worker/task.rs

a3s_flow/worker/
task.rs
