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

assay_domain/store/
workflow.rs
