runifold_workflow/store/
traits.rs

1use super::{
2    Budget, Checkpoint, CheckpointError, CheckpointId, ClaimedWorkflow, LeaseDuration, SystemTime,
3    UNIX_EPOCH, Usage, Value, WorkerId, WorkflowBudgetAuditCursor, WorkflowBudgetAuditEvent,
4    WorkflowBudgetAuditLimit, WorkflowBudgetAuditProjectionId, WorkflowBudgetAuditProjectionLease,
5    WorkflowBudgetReservationOutcome, WorkflowCancelOutcome, WorkflowCheckpointHistoryLimit,
6    WorkflowCheckpointRevision, WorkflowDisposition, WorkflowForkCommand, WorkflowForkOutcome,
7    WorkflowInterruptCommand, WorkflowInterruptDecisionOutcome, WorkflowLease, WorkflowSignal,
8    WorkflowSignalId, WorkflowSignalOutcome, WorkflowSignalRetention, WorkflowSignalSnapshot,
9    WorkflowStoreError, WorkflowStoreFuture, WorkflowTask, WorkflowTaskCleanupLease,
10    WorkflowTaskCleanupLimit, WorkflowTaskRetention, WorkflowTaskSnapshot, WorkflowTaskTombstone,
11    WorkflowTaskTombstoneCursor, WorkflowTaskTombstoneLimit, WorkflowTenantBudgetPolicy,
12    WorkflowTenantBudgetSnapshot, WorkflowTenantId, WorkflowTenantListLimit, WorkflowTenantPolicy,
13};
14
15/// Asynchronous distributed workflow task-control boundary.
16///
17/// Implementations must use a store-authoritative clock for claim expiration.
18/// Every ownership-sensitive mutation must compare both worker identity and
19/// fencing token.
20pub trait WorkflowStore: Send + Sync {
21    /// Returns the store-authoritative Unix time in milliseconds.
22    fn current_time_ms(&self) -> WorkflowStoreFuture<'_, Result<u64, WorkflowStoreError>>;
23
24    /// Creates or replaces one tenant's admission policy.
25    fn set_tenant_policy(
26        &self,
27        tenant_id: WorkflowTenantId,
28        policy: WorkflowTenantPolicy,
29    ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>;
30
31    /// Creates or replaces one tenant's persistent aggregate budget policy.
32    fn set_tenant_budget_policy(
33        &self,
34        tenant_id: WorkflowTenantId,
35        policy: WorkflowTenantBudgetPolicy,
36    ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>;
37
38    /// Discovers budget-enabled tenants in stable identity order.
39    fn list_tenant_budgets(
40        &self,
41        after: Option<WorkflowTenantId>,
42        limit: WorkflowTenantListLimit,
43    ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTenantId>, WorkflowStoreError>>;
44
45    /// Reads a tenant budget after reclaiming expired reservations.
46    fn inspect_tenant_budget(
47        &self,
48        tenant_id: WorkflowTenantId,
49    ) -> WorkflowStoreFuture<'_, Result<WorkflowTenantBudgetSnapshot, WorkflowStoreError>>;
50
51    /// Reads durable budget decisions strictly after an optional cursor.
52    fn list_tenant_budget_audit(
53        &self,
54        tenant_id: WorkflowTenantId,
55        after: Option<WorkflowBudgetAuditCursor>,
56        limit: WorkflowBudgetAuditLimit,
57    ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowBudgetAuditEvent>, WorkflowStoreError>>;
58
59    /// Deletes tenant audit facts at or before an explicitly acknowledged cursor.
60    fn compact_tenant_budget_audit(
61        &self,
62        tenant_id: WorkflowTenantId,
63        through: WorkflowBudgetAuditCursor,
64    ) -> WorkflowStoreFuture<'_, Result<u64, WorkflowStoreError>>;
65
66    /// Loads or atomically registers one named consumer at cursor zero.
67    fn load_or_create_tenant_budget_audit_projection(
68        &self,
69        tenant_id: WorkflowTenantId,
70        projection_id: WorkflowBudgetAuditProjectionId,
71    ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetAuditCursor, WorkflowStoreError>>;
72
73    /// Monotonically advances a projection cursor using compare-and-set.
74    ///
75    /// Returns `false` when another projector changed the cursor after
76    /// `expected` was loaded.
77    fn advance_tenant_budget_audit_projection(
78        &self,
79        tenant_id: WorkflowTenantId,
80        projection_id: WorkflowBudgetAuditProjectionId,
81        expected: WorkflowBudgetAuditCursor,
82        next: WorkflowBudgetAuditCursor,
83    ) -> WorkflowStoreFuture<'_, Result<bool, WorkflowStoreError>>;
84
85    /// Exclusively claims an idle or expired named audit projection.
86    fn claim_tenant_budget_audit_projection(
87        &self,
88        tenant_id: WorkflowTenantId,
89        projection_id: WorkflowBudgetAuditProjectionId,
90        owner: WorkerId,
91        lease: LeaseDuration,
92    ) -> WorkflowStoreFuture<
93        '_,
94        Result<Option<WorkflowBudgetAuditProjectionLease>, WorkflowStoreError>,
95    >;
96
97    /// Extends an active projection lease under its current fencing token.
98    fn heartbeat_tenant_budget_audit_projection(
99        &self,
100        lease: WorkflowBudgetAuditProjectionLease,
101        extension: LeaseDuration,
102    ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetAuditProjectionLease, WorkflowStoreError>>;
103
104    /// Advances a projection cursor only while its fenced lease remains active.
105    fn advance_tenant_budget_audit_projection_lease(
106        &self,
107        lease: WorkflowBudgetAuditProjectionLease,
108        next: WorkflowBudgetAuditCursor,
109    ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetAuditProjectionLease, WorkflowStoreError>>;
110
111    /// Releases a currently fenced projection without changing its cursor.
112    fn release_tenant_budget_audit_projection(
113        &self,
114        lease: WorkflowBudgetAuditProjectionLease,
115    ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>;
116
117    /// Idempotently reserves the remaining workflow envelope under a lease.
118    fn reserve_budget(
119        &self,
120        lease: WorkflowLease,
121        workflow_limit: Budget,
122        baseline: Usage,
123    ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetReservationOutcome, WorkflowStoreError>>;
124
125    /// Commits observed cumulative usage and releases unused reservation.
126    fn settle_budget(
127        &self,
128        lease: WorkflowLease,
129        cumulative: Usage,
130    ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>;
131
132    /// Enqueues a task exactly once.
133    fn enqueue(
134        &self,
135        task: WorkflowTask,
136    ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>;
137
138    /// Atomically claims the highest-priority eligible task.
139    fn claim(
140        &self,
141        worker: WorkerId,
142        lease: LeaseDuration,
143    ) -> WorkflowStoreFuture<'_, Result<Option<ClaimedWorkflow>, WorkflowStoreError>>;
144
145    /// Extends a currently owned, unexpired lease.
146    fn heartbeat(
147        &self,
148        lease: WorkflowLease,
149        extension: LeaseDuration,
150    ) -> WorkflowStoreFuture<'_, Result<WorkflowLease, WorkflowStoreError>>;
151
152    /// Applies a terminal or retry disposition under the current lease.
153    fn finish(
154        &self,
155        lease: WorkflowLease,
156        disposition: WorkflowDisposition,
157    ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>;
158
159    /// Idempotently publishes an external signal, buffering it when necessary.
160    fn publish_signal(
161        &self,
162        tenant_id: WorkflowTenantId,
163        signal: WorkflowSignal,
164    ) -> WorkflowStoreFuture<'_, Result<WorkflowSignalOutcome, WorkflowStoreError>>;
165
166    /// Publishes durable coordination metadata excluded from signal retention.
167    ///
168    /// Custom stores may use the default delivery behavior, but stores that
169    /// implement signal compaction should override this method and preserve the
170    /// accepted record until its parent Task is governed away.
171    fn publish_control_signal(
172        &self,
173        tenant_id: WorkflowTenantId,
174        signal: WorkflowSignal,
175    ) -> WorkflowStoreFuture<'_, Result<WorkflowSignalOutcome, WorkflowStoreError>> {
176        self.publish_signal(tenant_id, signal)
177    }
178
179    /// Idempotently applies a typed human decision to a durable interrupt.
180    fn decide_interrupt(
181        &self,
182        tenant_id: WorkflowTenantId,
183        command: WorkflowInterruptCommand,
184    ) -> WorkflowStoreFuture<'_, Result<WorkflowInterruptDecisionOutcome, WorkflowStoreError>> {
185        Box::pin(async move {
186            let signal = command
187                .into_signal()
188                .map_err(|error| WorkflowStoreError::invalid_input(error.to_string()))?;
189            self.publish_signal(tenant_id, signal).await.map(Into::into)
190        })
191    }
192
193    /// Idempotently cancels queued, waiting, or currently leased work.
194    fn cancel(
195        &self,
196        tenant_id: WorkflowTenantId,
197        checkpoint_id: CheckpointId,
198    ) -> WorkflowStoreFuture<'_, Result<WorkflowCancelOutcome, WorkflowStoreError>>;
199
200    /// Loads safe signal lifecycle metadata without exposing its payload.
201    fn inspect_signal(
202        &self,
203        tenant_id: WorkflowTenantId,
204        signal_id: WorkflowSignalId,
205    ) -> WorkflowStoreFuture<'_, Result<WorkflowSignalSnapshot, WorkflowStoreError>>;
206
207    /// Loads one accepted signal payload under tenant authorization.
208    ///
209    /// Payload access is separate from [`Self::inspect_signal`] so ordinary
210    /// control-plane inspection remains content-free.
211    fn load_signal_payload(
212        &self,
213        tenant_id: WorkflowTenantId,
214        signal_id: WorkflowSignalId,
215    ) -> WorkflowStoreFuture<'_, Result<Value, WorkflowStoreError>>;
216
217    /// Deletes only consumed or dead-letter signals older than retention.
218    fn compact_signals(
219        &self,
220        tenant_id: WorkflowTenantId,
221        retention: WorkflowSignalRetention,
222    ) -> WorkflowStoreFuture<'_, Result<u64, WorkflowStoreError>>;
223
224    /// Loads safe control-plane state for inspection.
225    fn inspect(
226        &self,
227        tenant_id: WorkflowTenantId,
228        checkpoint_id: CheckpointId,
229    ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskSnapshot, WorkflowStoreError>>;
230
231    /// Loads the immutable original task input under tenant authorization.
232    ///
233    /// This is intentionally separate from the safe operator snapshot because
234    /// workflow inputs can contain sensitive application data.
235    fn load_task_input(
236        &self,
237        tenant_id: WorkflowTenantId,
238        checkpoint_id: CheckpointId,
239    ) -> WorkflowStoreFuture<'_, Result<Value, WorkflowStoreError>>;
240
241    /// Lists immutable checkpoint revisions after an optional revision cursor.
242    fn list_checkpoint_history(
243        &self,
244        tenant_id: WorkflowTenantId,
245        checkpoint_id: CheckpointId,
246        after_revision: Option<u64>,
247        limit: WorkflowCheckpointHistoryLimit,
248    ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowCheckpointRevision>, WorkflowStoreError>>;
249
250    /// Loads one exact immutable checkpoint revision for state inspection.
251    fn load_checkpoint_revision(
252        &self,
253        tenant_id: WorkflowTenantId,
254        checkpoint_id: CheckpointId,
255        revision: u64,
256    ) -> WorkflowStoreFuture<'_, Result<WorkflowCheckpointRevision, WorkflowStoreError>>;
257
258    /// Idempotently creates a new execution branch from immutable history.
259    fn fork_workflow(
260        &self,
261        tenant_id: WorkflowTenantId,
262        command: WorkflowForkCommand,
263    ) -> WorkflowStoreFuture<'_, Result<WorkflowForkOutcome, WorkflowStoreError>>;
264
265    /// Loads a checkpoint under a current worker lease.
266    fn load_checkpoint(
267        &self,
268        lease: WorkflowLease,
269    ) -> WorkflowStoreFuture<'_, Result<Checkpoint, CheckpointError>>;
270
271    /// Creates or compare-and-swaps a checkpoint under a current worker lease.
272    fn compare_and_swap_checkpoint(
273        &self,
274        lease: WorkflowLease,
275        checkpoint: Checkpoint,
276        expected_revision: Option<u64>,
277    ) -> WorkflowStoreFuture<'_, Result<(), CheckpointError>>;
278}
279
280/// Optional fenced control plane for physically removing terminal Tasks.
281///
282/// Implementations must write an immutable tombstone in the same atomic
283/// operation that removes execution state. Active Tasks are never eligible.
284pub trait WorkflowTaskRetentionStore: WorkflowStore {
285    /// Discovers tenants that currently own terminal Tasks.
286    ///
287    /// Results are ordered lexicographically and strictly follow `after`.
288    fn list_task_cleanup_tenants(
289        &self,
290        after: Option<WorkflowTenantId>,
291        limit: WorkflowTenantListLimit,
292    ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTenantId>, WorkflowStoreError>>;
293
294    /// Claims one tenant's cleanup partition if it is idle or expired.
295    fn claim_task_cleanup(
296        &self,
297        tenant_id: WorkflowTenantId,
298        owner: WorkerId,
299        lease: LeaseDuration,
300    ) -> WorkflowStoreFuture<'_, Result<Option<WorkflowTaskCleanupLease>, WorkflowStoreError>>;
301
302    /// Atomically tombstones and removes one bounded terminal Task batch.
303    fn compact_terminal_tasks(
304        &self,
305        lease: WorkflowTaskCleanupLease,
306        retention: WorkflowTaskRetention,
307        limit: WorkflowTaskCleanupLimit,
308    ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTaskTombstone>, WorkflowStoreError>>;
309
310    /// Extends an exact current cleanup lease using store-authoritative time.
311    fn heartbeat_task_cleanup(
312        &self,
313        lease: WorkflowTaskCleanupLease,
314        extension: LeaseDuration,
315    ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskCleanupLease, WorkflowStoreError>>;
316
317    /// Lists immutable tombstones strictly after an optional cursor.
318    fn list_task_tombstones(
319        &self,
320        tenant_id: WorkflowTenantId,
321        after: Option<WorkflowTaskTombstoneCursor>,
322        limit: WorkflowTaskTombstoneLimit,
323    ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTaskTombstone>, WorkflowStoreError>>;
324
325    /// Releases a current unexpired cleanup lease.
326    fn release_task_cleanup(
327        &self,
328        lease: WorkflowTaskCleanupLease,
329    ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>;
330}
331
332/// Store-authoritative time source used by the in-memory reference adapter.
333pub trait WorkflowClock: Send + Sync {
334    /// Returns Unix time in milliseconds.
335    fn now_ms(&self) -> u64;
336}
337
338/// System clock used by default for ephemeral workflow queues.
339#[derive(Clone, Copy, Debug, Default)]
340pub struct SystemWorkflowClock;
341
342impl WorkflowClock for SystemWorkflowClock {
343    fn now_ms(&self) -> u64 {
344        SystemTime::now()
345            .duration_since(UNIX_EPOCH)
346            .map_or(0, |duration| {
347                u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
348            })
349    }
350}
runifold_workflow/store/traits.rs

runifold_workflow/store/
traits.rs
