runifold_workflow/store/
traits.rs

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

runifold_workflow/store/
traits.rs
