assay_domain/store/
workflow.rs

1//! `WorkflowStore` trait — every workflow backend implements this.
2
3use std::future::Future;
4
5use crate::types::*;
6
7/// Core storage trait for the workflow engine.
8///
9/// All database access goes through this trait. Methods that operate on
10/// namespace-scoped data take a `namespace` parameter. The "main"
11/// namespace is always available.
12///
13/// All methods return `Send` futures so they can be used from `tokio::spawn`.
14pub trait WorkflowStore: Send + Sync + 'static {
15    // ── Namespaces ─────────────────────────────────────────
16
17    fn create_namespace(&self, name: &str) -> impl Future<Output = anyhow::Result<()>> + Send;
18
19    fn list_namespaces(&self) -> impl Future<Output = anyhow::Result<Vec<NamespaceRecord>>> + Send;
20
21    fn delete_namespace(&self, name: &str) -> impl Future<Output = anyhow::Result<bool>> + Send;
22
23    fn get_namespace_stats(
24        &self,
25        namespace: &str,
26    ) -> impl Future<Output = anyhow::Result<NamespaceStats>> + Send;
27
28    // ── Workflows ──────────────────────────────────────────
29
30    fn create_workflow(
31        &self,
32        workflow: &WorkflowRecord,
33    ) -> impl Future<Output = anyhow::Result<()>> + Send;
34
35    fn get_workflow(
36        &self,
37        id: &str,
38    ) -> impl Future<Output = anyhow::Result<Option<WorkflowRecord>>> + Send;
39
40    fn list_workflows(
41        &self,
42        namespace: &str,
43        status: Option<WorkflowStatus>,
44        workflow_type: Option<&str>,
45        search_attrs_filter: Option<&str>,
46        limit: i64,
47        offset: i64,
48    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowRecord>>> + Send;
49
50    /// List workflows in terminal states whose `completed_at` is older than
51    /// `cutoff` and which haven't been archived yet. Used by the optional
52    /// S3 archival background task to batch candidates.
53    fn list_archivable_workflows(
54        &self,
55        cutoff: f64,
56        limit: i64,
57    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowRecord>>> + Send;
58
59    /// Mark a workflow as archived (records `archived_at` + `archive_uri`)
60    /// and purge its events, activities, timers, signals, and snapshots.
61    /// The workflow record itself is preserved so `GET /workflows/{id}`
62    /// still resolves with an archive pointer.
63    fn mark_archived_and_purge(
64        &self,
65        workflow_id: &str,
66        archive_uri: &str,
67        archived_at: f64,
68    ) -> impl Future<Output = anyhow::Result<()>> + Send;
69
70    /// Merge a JSON object patch into the workflow's `search_attributes`.
71    /// Keys in the patch overwrite existing keys; keys already present but
72    /// not in the patch are preserved. If the current column is NULL, the
73    /// patch becomes the new value.
74    fn upsert_search_attributes(
75        &self,
76        workflow_id: &str,
77        patch_json: &str,
78    ) -> impl Future<Output = anyhow::Result<()>> + Send;
79
80    fn update_workflow_status(
81        &self,
82        id: &str,
83        status: WorkflowStatus,
84        result: Option<&str>,
85        error: Option<&str>,
86    ) -> impl Future<Output = anyhow::Result<()>> + Send;
87
88    fn claim_workflow(
89        &self,
90        id: &str,
91        worker_id: &str,
92    ) -> impl Future<Output = anyhow::Result<bool>> + Send;
93
94    // ── Workflow-task dispatch ────────────────────
95
96    /// Mark a workflow as having new events that need a worker to replay it.
97    /// Idempotent — calling repeatedly is fine. Cleared by `claim_workflow_task`.
98    fn mark_workflow_dispatchable(
99        &self,
100        workflow_id: &str,
101    ) -> impl Future<Output = anyhow::Result<()>> + Send;
102
103    /// Atomically claim the oldest dispatchable workflow on a queue. Sets
104    /// `dispatch_claimed_by` and `dispatch_last_heartbeat`, clears
105    /// `needs_dispatch`. Returns the workflow record or None if nothing
106    /// is available.
107    fn claim_workflow_task(
108        &self,
109        task_queue: &str,
110        worker_id: &str,
111    ) -> impl Future<Output = anyhow::Result<Option<WorkflowRecord>>> + Send;
112
113    /// Release a workflow task's dispatch lease (called when the worker
114    /// submits its commands batch). Only succeeds if `dispatch_claimed_by`
115    /// matches the calling worker.
116    fn release_workflow_task(
117        &self,
118        workflow_id: &str,
119        worker_id: &str,
120    ) -> impl Future<Output = anyhow::Result<()>> + Send;
121
122    /// Forcibly release dispatch leases whose worker hasn't heartbeat'd
123    /// within `timeout_secs`. Used by the engine's background poller to
124    /// recover from worker crashes. Returns how many leases were released
125    /// (each becomes claimable again, with `needs_dispatch=true`).
126    fn release_stale_dispatch_leases(
127        &self,
128        now: f64,
129        timeout_secs: f64,
130    ) -> impl Future<Output = anyhow::Result<u64>> + Send;
131
132    // ── Events ─────────────────────────────────────────────
133
134    fn append_event(
135        &self,
136        event: &WorkflowEvent,
137    ) -> impl Future<Output = anyhow::Result<i64>> + Send;
138
139    fn list_events(
140        &self,
141        workflow_id: &str,
142    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowEvent>>> + Send;
143
144    /// Return a bounded event page in sequence order. A descending page uses
145    /// `cursor` as an exclusive upper bound; an ascending page uses it as an
146    /// exclusive lower bound. Native stores override this to page in SQL.
147    fn list_events_page(
148        &self,
149        workflow_id: &str,
150        cursor: Option<i32>,
151        limit: i64,
152        descending: bool,
153    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowEvent>>> + Send {
154        async move {
155            let limit = limit.clamp(0, 1_000) as usize;
156            if limit == 0 {
157                return Ok(Vec::new());
158            }
159            let mut events = self.list_events(workflow_id).await?;
160            events.retain(|event| {
161                cursor.is_none_or(|sequence| {
162                    if descending {
163                        event.seq < sequence
164                    } else {
165                        event.seq > sequence
166                    }
167                })
168            });
169            if descending {
170                events.reverse();
171            }
172            events.truncate(limit);
173            Ok(events)
174        }
175    }
176
177    fn get_event_count(
178        &self,
179        workflow_id: &str,
180    ) -> impl Future<Output = anyhow::Result<i64>> + Send;
181
182    // ── Activities ──────────────────────────────────────────
183
184    fn create_activity(
185        &self,
186        activity: &WorkflowActivity,
187    ) -> impl Future<Output = anyhow::Result<i64>> + Send;
188
189    /// Look up an activity by its primary key.
190    fn get_activity(
191        &self,
192        id: i64,
193    ) -> impl Future<Output = anyhow::Result<Option<WorkflowActivity>>> + Send;
194
195    /// Look up an activity by its workflow-relative sequence number.
196    /// Used for idempotent scheduling: the engine checks if (workflow_id, seq)
197    /// already exists before creating a new row.
198    fn get_activity_by_workflow_seq(
199        &self,
200        workflow_id: &str,
201        seq: i32,
202    ) -> impl Future<Output = anyhow::Result<Option<WorkflowActivity>>> + Send;
203
204    fn claim_activity(
205        &self,
206        task_queue: &str,
207        worker_id: &str,
208    ) -> impl Future<Output = anyhow::Result<Option<WorkflowActivity>>> + Send;
209
210    /// Re-queue an activity for retry: clears the running state
211    /// (status→PENDING, claimed_by/started_at cleared), bumps `attempt`,
212    /// and sets `scheduled_at = now + backoff` so the next claim_activity
213    /// won't pick it up before the backoff elapses.
214    fn requeue_activity_for_retry(
215        &self,
216        id: i64,
217        next_attempt: i32,
218        next_scheduled_at: f64,
219    ) -> impl Future<Output = anyhow::Result<()>> + Send;
220
221    fn complete_activity(
222        &self,
223        id: i64,
224        result: Option<&str>,
225        error: Option<&str>,
226        failed: bool,
227    ) -> impl Future<Output = anyhow::Result<()>> + Send;
228
229    fn heartbeat_activity(
230        &self,
231        id: i64,
232        details: Option<&str>,
233    ) -> impl Future<Output = anyhow::Result<()>> + Send;
234
235    fn get_timed_out_activities(
236        &self,
237        now: f64,
238    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowActivity>>> + Send;
239
240    // ── Timers ──────────────────────────────────────────────
241
242    /// Mark all PENDING activities of a workflow as CANCELLED so workers
243    /// that haven't claimed them yet won't pick them up. Returns the
244    /// number of rows affected. Does NOT touch RUNNING activities — those
245    /// will see the cancellation when they next heartbeat or complete.
246    fn cancel_pending_activities(
247        &self,
248        workflow_id: &str,
249    ) -> impl Future<Output = anyhow::Result<u64>> + Send;
250
251    /// Mark all unfired timers of a workflow as fired without firing
252    /// (effectively removing them from the timer poller). Returns the
253    /// number of rows affected.
254    fn cancel_pending_timers(
255        &self,
256        workflow_id: &str,
257    ) -> impl Future<Output = anyhow::Result<u64>> + Send;
258
259    fn create_timer(
260        &self,
261        timer: &WorkflowTimer,
262    ) -> impl Future<Output = anyhow::Result<i64>> + Send;
263
264    /// Look up an existing timer by its workflow-relative seq. Used by the
265    /// engine for idempotent ScheduleTimer (deterministic replay can call
266    /// schedule_timer for the same seq more than once on retries).
267    fn get_timer_by_workflow_seq(
268        &self,
269        workflow_id: &str,
270        seq: i32,
271    ) -> impl Future<Output = anyhow::Result<Option<WorkflowTimer>>> + Send;
272
273    fn fire_due_timers(
274        &self,
275        now: f64,
276    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowTimer>>> + Send;
277
278    // ── Signals ─────────────────────────────────────────────
279
280    fn send_signal(
281        &self,
282        signal: &WorkflowSignal,
283    ) -> impl Future<Output = anyhow::Result<i64>> + Send;
284
285    fn consume_signals(
286        &self,
287        workflow_id: &str,
288        name: &str,
289    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowSignal>>> + Send;
290
291    // ── Schedules ───────────────────────────────────────────
292
293    fn create_schedule(
294        &self,
295        schedule: &WorkflowSchedule,
296    ) -> impl Future<Output = anyhow::Result<()>> + Send;
297
298    fn get_schedule(
299        &self,
300        namespace: &str,
301        name: &str,
302    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
303
304    fn list_schedules(
305        &self,
306        namespace: &str,
307    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowSchedule>>> + Send;
308
309    fn update_schedule_last_run(
310        &self,
311        namespace: &str,
312        name: &str,
313        last_run_at: f64,
314        next_run_at: f64,
315        workflow_id: &str,
316    ) -> impl Future<Output = anyhow::Result<()>> + Send;
317
318    fn delete_schedule(
319        &self,
320        namespace: &str,
321        name: &str,
322    ) -> impl Future<Output = anyhow::Result<bool>> + Send;
323
324    /// Apply an in-place patch to a schedule. Only fields present on
325    /// `patch` are updated; the rest keep their current values. Returns
326    /// the updated record, or `None` if the schedule doesn't exist.
327    ///
328    /// The scheduler's `next_run_at` is recomputed from the new
329    /// `cron_expr` + `timezone` on the next evaluation tick, so a PATCH
330    /// takes effect within the scheduler's poll interval.
331    fn update_schedule(
332        &self,
333        namespace: &str,
334        name: &str,
335        patch: &SchedulePatch,
336    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
337
338    /// Flip a schedule's `paused` flag. Returns the updated record, or
339    /// `None` if the schedule doesn't exist.
340    ///
341    /// A paused schedule is skipped by the scheduler; resuming it
342    /// doesn't backfill missed fires — the next fire is whatever the
343    /// cron expression says, starting from now.
344    fn set_schedule_paused(
345        &self,
346        namespace: &str,
347        name: &str,
348        paused: bool,
349    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
350
351    // ── Workers ─────────────────────────────────────────────
352
353    fn register_worker(
354        &self,
355        worker: &WorkflowWorker,
356    ) -> impl Future<Output = anyhow::Result<()>> + Send;
357
358    fn heartbeat_worker(
359        &self,
360        id: &str,
361        now: f64,
362    ) -> impl Future<Output = anyhow::Result<()>> + Send;
363
364    fn list_workers(
365        &self,
366        namespace: &str,
367    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowWorker>>> + Send;
368
369    fn remove_dead_workers(
370        &self,
371        cutoff: f64,
372    ) -> impl Future<Output = anyhow::Result<Vec<String>>> + Send;
373
374    // ── Child Workflows ─────────────────────────────────────
375
376    fn list_child_workflows(
377        &self,
378        parent_id: &str,
379    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowRecord>>> + Send;
380
381    // ── Snapshots ───────────────────────────────────────────
382
383    fn create_snapshot(
384        &self,
385        workflow_id: &str,
386        event_seq: i32,
387        state_json: &str,
388    ) -> impl Future<Output = anyhow::Result<()>> + Send;
389
390    fn get_latest_snapshot(
391        &self,
392        workflow_id: &str,
393    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSnapshot>>> + Send;
394
395    // ── Queue Stats ─────────────────────────────────────────
396
397    fn get_queue_stats(
398        &self,
399        namespace: &str,
400    ) -> impl Future<Output = anyhow::Result<Vec<QueueStats>>> + Send;
401
402    // ── Leader Election ─────────────────────────────────────
403
404    /// Try to acquire the scheduler lock for leader election.
405    /// Returns true if this instance should run the cron scheduler.
406    ///
407    /// - SQLite: always returns true (single-instance assumed)
408    /// - Postgres: uses pg_try_advisory_lock (only one instance wins)
409    fn try_acquire_scheduler_lock(&self) -> impl Future<Output = anyhow::Result<bool>> + Send;
410
411    // Push subscriptions are removed in v0.13.1 — the engine-events
412    // outbox (`assay_domain::events::EngineEventBus`) is the
413    // backend-agnostic replacement. Emits happen at the call site
414    // (lifecycle / tasks / signals / activities) so consumers subscribe
415    // to the bus, not to per-method streams on the store.
416}
assay_domain/store/workflow.rs

assay_domain/store/
workflow.rs
