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 /// Terminally settle an activity: write its status and result, append
240 /// the matching history event, and arm the workflow for dispatch — all
241 /// in one transaction. A `COMPLETED` activity whose workflow never got
242 /// the event (and so replays it as still pending) is not reachable
243 /// through this call.
244 ///
245 /// Idempotent, and the documented repair path: re-settling an activity
246 /// whose event is already durable re-applies only the dispatch arming,
247 /// which recovers a workflow task lost after the event landed. The
248 /// first settle wins the activity row — a later call never rewrites a
249 /// result the history already carries.
250 fn settle_activity(
251 &self,
252 settlement: &ActivitySettlement<'_>,
253 ) -> impl Future<Output = anyhow::Result<SettleOutcome>> + Send;
254
255 /// Activities that reached a terminal status while their workflow is
256 /// still live and their terminal history event is missing. Rows written
257 /// by an engine older than the transactional settle path, or by any
258 /// half-applied write, surface here so the engine can converge them.
259 /// Oldest first, bounded by `limit`.
260 fn list_unsettled_activities(
261 &self,
262 limit: i64,
263 ) -> impl Future<Output = anyhow::Result<Vec<WorkflowActivity>>> + Send;
264
265 fn heartbeat_activity(
266 &self,
267 id: i64,
268 details: Option<&str>,
269 ) -> impl Future<Output = anyhow::Result<()>> + Send;
270
271 fn get_timed_out_activities(
272 &self,
273 now: f64,
274 ) -> impl Future<Output = anyhow::Result<Vec<WorkflowActivity>>> + Send;
275
276 // ── Timers ──────────────────────────────────────────────
277
278 /// Mark all PENDING activities of a workflow as CANCELLED so workers
279 /// that haven't claimed them yet won't pick them up. Returns the
280 /// number of rows affected. Does NOT touch RUNNING activities — those
281 /// will see the cancellation when they next heartbeat or complete.
282 fn cancel_pending_activities(
283 &self,
284 workflow_id: &str,
285 ) -> impl Future<Output = anyhow::Result<u64>> + Send;
286
287 /// Mark all unfired timers of a workflow as fired without firing
288 /// (effectively removing them from the timer poller). Returns the
289 /// number of rows affected.
290 fn cancel_pending_timers(
291 &self,
292 workflow_id: &str,
293 ) -> impl Future<Output = anyhow::Result<u64>> + Send;
294
295 fn create_timer(
296 &self,
297 timer: &WorkflowTimer,
298 ) -> impl Future<Output = anyhow::Result<i64>> + Send;
299
300 /// Look up an existing timer by its workflow-relative seq. Used by the
301 /// engine for idempotent ScheduleTimer (deterministic replay can call
302 /// schedule_timer for the same seq more than once on retries).
303 fn get_timer_by_workflow_seq(
304 &self,
305 workflow_id: &str,
306 seq: i32,
307 ) -> impl Future<Output = anyhow::Result<Option<WorkflowTimer>>> + Send;
308
309 fn fire_due_timers(
310 &self,
311 now: f64,
312 ) -> impl Future<Output = anyhow::Result<Vec<WorkflowTimer>>> + Send;
313
314 // ── Signals ─────────────────────────────────────────────
315
316 fn send_signal(
317 &self,
318 signal: &WorkflowSignal,
319 ) -> impl Future<Output = anyhow::Result<i64>> + Send;
320
321 /// Record a signal, append its `SignalReceived` event and arm the
322 /// workflow for dispatch in one transaction, so a stored signal can
323 /// never be invisible to the workflow that has to react to it.
324 /// `payload_json` is the serialised event payload.
325 fn deliver_signal(
326 &self,
327 signal: &WorkflowSignal,
328 payload_json: &str,
329 ) -> impl Future<Output = anyhow::Result<i64>> + Send;
330
331 fn consume_signals(
332 &self,
333 workflow_id: &str,
334 name: &str,
335 ) -> impl Future<Output = anyhow::Result<Vec<WorkflowSignal>>> + Send;
336
337 // ── Schedules ───────────────────────────────────────────
338
339 fn create_schedule(
340 &self,
341 schedule: &WorkflowSchedule,
342 ) -> impl Future<Output = anyhow::Result<()>> + Send;
343
344 fn get_schedule(
345 &self,
346 namespace: &str,
347 name: &str,
348 ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
349
350 fn list_schedules(
351 &self,
352 namespace: &str,
353 ) -> impl Future<Output = anyhow::Result<Vec<WorkflowSchedule>>> + Send;
354
355 fn update_schedule_last_run(
356 &self,
357 namespace: &str,
358 name: &str,
359 last_run_at: f64,
360 next_run_at: f64,
361 workflow_id: &str,
362 ) -> impl Future<Output = anyhow::Result<()>> + Send;
363
364 fn delete_schedule(
365 &self,
366 namespace: &str,
367 name: &str,
368 ) -> impl Future<Output = anyhow::Result<bool>> + Send;
369
370 /// Apply an in-place patch to a schedule. Only fields present on
371 /// `patch` are updated; the rest keep their current values. Returns
372 /// the updated record, or `None` if the schedule doesn't exist.
373 ///
374 /// The scheduler's `next_run_at` is recomputed from the new
375 /// `cron_expr` + `timezone` on the next evaluation tick, so a PATCH
376 /// takes effect within the scheduler's poll interval.
377 fn update_schedule(
378 &self,
379 namespace: &str,
380 name: &str,
381 patch: &SchedulePatch,
382 ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
383
384 /// Flip a schedule's `paused` flag. Returns the updated record, or
385 /// `None` if the schedule doesn't exist.
386 ///
387 /// A paused schedule is skipped by the scheduler; resuming it
388 /// doesn't backfill missed fires — the next fire is whatever the
389 /// cron expression says, starting from now.
390 fn set_schedule_paused(
391 &self,
392 namespace: &str,
393 name: &str,
394 paused: bool,
395 ) -> impl Future<Output = anyhow::Result<Option<WorkflowSchedule>>> + Send;
396
397 // ── Workers ─────────────────────────────────────────────
398
399 fn register_worker(
400 &self,
401 worker: &WorkflowWorker,
402 ) -> impl Future<Output = anyhow::Result<()>> + Send;
403
404 /// Record a worker heartbeat. Returns false when no registration with
405 /// this id exists — the reaper removed it while the worker was silent,
406 /// and the worker has to register again to be dispatchable.
407 fn heartbeat_worker(
408 &self,
409 id: &str,
410 now: f64,
411 ) -> impl Future<Output = anyhow::Result<bool>> + Send;
412
413 fn list_workers(
414 &self,
415 namespace: &str,
416 ) -> impl Future<Output = anyhow::Result<Vec<WorkflowWorker>>> + Send;
417
418 fn remove_dead_workers(
419 &self,
420 cutoff: f64,
421 ) -> impl Future<Output = anyhow::Result<Vec<String>>> + Send;
422
423 // ── Child Workflows ─────────────────────────────────────
424
425 fn list_child_workflows(
426 &self,
427 parent_id: &str,
428 ) -> impl Future<Output = anyhow::Result<Vec<WorkflowRecord>>> + Send;
429
430 // ── Snapshots ───────────────────────────────────────────
431
432 fn create_snapshot(
433 &self,
434 workflow_id: &str,
435 event_seq: i32,
436 state_json: &str,
437 ) -> impl Future<Output = anyhow::Result<()>> + Send;
438
439 fn get_latest_snapshot(
440 &self,
441 workflow_id: &str,
442 ) -> impl Future<Output = anyhow::Result<Option<WorkflowSnapshot>>> + Send;
443
444 // ── Queue Stats ─────────────────────────────────────────
445
446 fn get_queue_stats(
447 &self,
448 namespace: &str,
449 ) -> impl Future<Output = anyhow::Result<Vec<QueueStats>>> + Send;
450
451 // ── Leader Election ─────────────────────────────────────
452
453 /// Try to acquire the scheduler lock for leader election.
454 /// Returns true if this instance should run the cron scheduler.
455 ///
456 /// - SQLite: always returns true (single-instance assumed)
457 /// - Postgres: uses pg_try_advisory_lock (only one instance wins)
458 fn try_acquire_scheduler_lock(&self) -> impl Future<Output = anyhow::Result<bool>> + Send;
459
460 // Push subscriptions are removed in v0.13.1 — the engine-events
461 // outbox (`assay_domain::events::EngineEventBus`) is the
462 // backend-agnostic replacement. Emits happen at the call site
463 // (lifecycle / tasks / signals / activities) so consumers subscribe
464 // to the bus, not to per-method streams on the store.
465}