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 retry_failed_activity(
222        &self,
223        _workflow_id: &str,
224        _requested_by: &str,
225        _reason: &str,
226        _requested_at: f64,
227    ) -> impl Future<Output = anyhow::Result<RetryFailedActivityResult>> + Send {
228        async { Ok(RetryFailedActivityResult::Unsupported) }
229    }
230
231    fn complete_activity(
232        &self,
233        id: i64,
234        result: Option<&str>,
235        error: Option<&str>,
236        failed: bool,
237    ) -> impl Future<Output = anyhow::Result<()>> + Send;
238
239    fn heartbeat_activity(
240        &self,
241        id: i64,
242        details: Option<&str>,
243    ) -> impl Future<Output = anyhow::Result<()>> + Send;
244
245    fn get_timed_out_activities(
246        &self,
247        now: f64,
248    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowActivity>>> + Send;
249
250    // ── Timers ──────────────────────────────────────────────
251
252    /// Mark all PENDING activities of a workflow as CANCELLED so workers
253    /// that haven't claimed them yet won't pick them up. Returns the
254    /// number of rows affected. Does NOT touch RUNNING activities — those
255    /// will see the cancellation when they next heartbeat or complete.
256    fn cancel_pending_activities(
257        &self,
258        workflow_id: &str,
259    ) -> impl Future<Output = anyhow::Result<u64>> + Send;
260
261    /// Mark all unfired timers of a workflow as fired without firing
262    /// (effectively removing them from the timer poller). Returns the
263    /// number of rows affected.
264    fn cancel_pending_timers(
265        &self,
266        workflow_id: &str,
267    ) -> impl Future<Output = anyhow::Result<u64>> + Send;
268
269    fn create_timer(
270        &self,
271        timer: &WorkflowTimer,
272    ) -> impl Future<Output = anyhow::Result<i64>> + Send;
273
274    /// Look up an existing timer by its workflow-relative seq. Used by the
275    /// engine for idempotent ScheduleTimer (deterministic replay can call
276    /// schedule_timer for the same seq more than once on retries).
277    fn get_timer_by_workflow_seq(
278        &self,
279        workflow_id: &str,
280        seq: i32,
281    ) -> impl Future<Output = anyhow::Result<Option<WorkflowTimer>>> + Send;
282
283    fn fire_due_timers(
284        &self,
285        now: f64,
286    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowTimer>>> + Send;
287
288    // ── Signals ─────────────────────────────────────────────
289
290    fn send_signal(
291        &self,
292        signal: &WorkflowSignal,
293    ) -> impl Future<Output = anyhow::Result<i64>> + Send;
294
295    fn consume_signals(
296        &self,
297        workflow_id: &str,
298        name: &str,
299    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowSignal>>> + Send;
300
301    // ── Schedules ───────────────────────────────────────────
302
303    fn create_schedule(
304        &self,
305        schedule: &WorkflowSchedule,
306    ) -> impl Future<Output = anyhow::Result<()>> + Send;
307
308    fn get_schedule(
309        &self,
310        namespace: &str,
311        name: &str,
312    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
313
314    fn list_schedules(
315        &self,
316        namespace: &str,
317    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowSchedule>>> + Send;
318
319    fn update_schedule_last_run(
320        &self,
321        namespace: &str,
322        name: &str,
323        last_run_at: f64,
324        next_run_at: f64,
325        workflow_id: &str,
326    ) -> impl Future<Output = anyhow::Result<()>> + Send;
327
328    fn delete_schedule(
329        &self,
330        namespace: &str,
331        name: &str,
332    ) -> impl Future<Output = anyhow::Result<bool>> + Send;
333
334    /// Apply an in-place patch to a schedule. Only fields present on
335    /// `patch` are updated; the rest keep their current values. Returns
336    /// the updated record, or `None` if the schedule doesn't exist.
337    ///
338    /// The scheduler's `next_run_at` is recomputed from the new
339    /// `cron_expr` + `timezone` on the next evaluation tick, so a PATCH
340    /// takes effect within the scheduler's poll interval.
341    fn update_schedule(
342        &self,
343        namespace: &str,
344        name: &str,
345        patch: &SchedulePatch,
346    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
347
348    /// Flip a schedule's `paused` flag. Returns the updated record, or
349    /// `None` if the schedule doesn't exist.
350    ///
351    /// A paused schedule is skipped by the scheduler; resuming it
352    /// doesn't backfill missed fires — the next fire is whatever the
353    /// cron expression says, starting from now.
354    fn set_schedule_paused(
355        &self,
356        namespace: &str,
357        name: &str,
358        paused: bool,
359    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
360
361    // ── Workers ─────────────────────────────────────────────
362
363    fn register_worker(
364        &self,
365        worker: &WorkflowWorker,
366    ) -> impl Future<Output = anyhow::Result<()>> + Send;
367
368    fn heartbeat_worker(
369        &self,
370        id: &str,
371        now: f64,
372    ) -> impl Future<Output = anyhow::Result<()>> + Send;
373
374    fn list_workers(
375        &self,
376        namespace: &str,
377    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowWorker>>> + Send;
378
379    fn remove_dead_workers(
380        &self,
381        cutoff: f64,
382    ) -> impl Future<Output = anyhow::Result<Vec<String>>> + Send;
383
384    // ── Child Workflows ─────────────────────────────────────
385
386    fn list_child_workflows(
387        &self,
388        parent_id: &str,
389    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowRecord>>> + Send;
390
391    // ── Snapshots ───────────────────────────────────────────
392
393    fn create_snapshot(
394        &self,
395        workflow_id: &str,
396        event_seq: i32,
397        state_json: &str,
398    ) -> impl Future<Output = anyhow::Result<()>> + Send;
399
400    fn get_latest_snapshot(
401        &self,
402        workflow_id: &str,
403    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSnapshot>>> + Send;
404
405    // ── Queue Stats ─────────────────────────────────────────
406
407    fn get_queue_stats(
408        &self,
409        namespace: &str,
410    ) -> impl Future<Output = anyhow::Result<Vec<QueueStats>>> + Send;
411
412    // ── Leader Election ─────────────────────────────────────
413
414    /// Try to acquire the scheduler lock for leader election.
415    /// Returns true if this instance should run the cron scheduler.
416    ///
417    /// - SQLite: always returns true (single-instance assumed)
418    /// - Postgres: uses pg_try_advisory_lock (only one instance wins)
419    fn try_acquire_scheduler_lock(&self) -> impl Future<Output = anyhow::Result<bool>> + Send;
420
421    // Push subscriptions are removed in v0.13.1 — the engine-events
422    // outbox (`assay_domain::events::EngineEventBus`) is the
423    // backend-agnostic replacement. Emits happen at the call site
424    // (lifecycle / tasks / signals / activities) so consumers subscribe
425    // to the bus, not to per-method streams on the store.
426}
assay_domain/store/workflow.rs

assay_domain/store/
workflow.rs
