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    // ── API Keys ────────────────────────────────────────────
350
351    fn create_api_key(
352        &self,
353        key_hash: &str,
354        prefix: &str,
355        label: Option<&str>,
356        created_at: f64,
357    ) -> impl Future<Output = anyhow::Result<()>> + Send;
358
359    fn validate_api_key(
360        &self,
361        key_hash: &str,
362    ) -> impl Future<Output = anyhow::Result<bool>> + Send;
363
364    fn list_api_keys(
365        &self,
366    ) -> impl Future<Output = anyhow::Result<Vec<ApiKeyRecord>>> + Send;
367
368    fn revoke_api_key(
369        &self,
370        prefix: &str,
371    ) -> impl Future<Output = anyhow::Result<bool>> + Send;
372
373    /// Return true iff the `api_keys` table has no rows. Used by the HTTP layer
374    /// to identify the first-ever key-creation window (where the `POST
375    /// /api/v1/api-keys` endpoint is callable without authentication).
376    fn api_keys_empty(&self) -> impl Future<Output = anyhow::Result<bool>> + Send;
377
378    /// Find an existing API key by its label. Returns None if no key has this
379    /// label (or `label` is NULL). Used to implement idempotent key creation:
380    /// a second `POST /api/v1/api-keys { label, idempotent: true }` call hits
381    /// this method and returns the existing record's metadata (without a
382    /// plaintext, which is only ever retrievable at generation time).
383    fn get_api_key_by_label(
384        &self,
385        label: &str,
386    ) -> impl Future<Output = anyhow::Result<Option<ApiKeyRecord>>> + Send;
387
388    // ── Child Workflows ─────────────────────────────────────
389
390    fn list_child_workflows(
391        &self,
392        parent_id: &str,
393    ) -> impl Future<Output = anyhow::Result<Vec<WorkflowRecord>>> + Send;
394
395    // ── Snapshots ───────────────────────────────────────────
396
397    fn create_snapshot(
398        &self,
399        workflow_id: &str,
400        event_seq: i32,
401        state_json: &str,
402    ) -> impl Future<Output = anyhow::Result<()>> + Send;
403
404    fn get_latest_snapshot(
405        &self,
406        workflow_id: &str,
407    ) -> impl Future<Output = anyhow::Result<Option<WorkflowSnapshot>>> + Send;
408
409    // ── Queue Stats ─────────────────────────────────────────
410
411    fn get_queue_stats(
412        &self,
413        namespace: &str,
414    ) -> impl Future<Output = anyhow::Result<Vec<QueueStats>>> + Send;
415
416    // ── Leader Election ─────────────────────────────────────
417
418    /// Try to acquire the scheduler lock for leader election.
419    /// Returns true if this instance should run the cron scheduler.
420    ///
421    /// - SQLite: always returns true (single-instance assumed)
422    /// - Postgres: uses pg_try_advisory_lock (only one instance wins)
423    fn try_acquire_scheduler_lock(
424        &self,
425    ) -> impl Future<Output = anyhow::Result<bool>> + Send;
426
427    // Push subscriptions are removed in v0.13.1 — the engine-events
428    // outbox (`assay_domain::events::EngineEventBus`) is the
429    // backend-agnostic replacement. Emits happen at the call site
430    // (lifecycle / tasks / signals / activities) so consumers subscribe
431    // to the bus, not to per-method streams on the store.
432}
assay_domain/store/workflow.rs

assay_domain/store/
workflow.rs
