runifold_workflow/store/
model.rs

1use super::{
2    Budget, CheckpointId, Duration, Error, Future, NonZeroU32, NonZeroU64, Pin, Usage, Value,
3    WorkflowInterruptRequest, WorkflowLineage, WorkflowWait, WorkflowWake,
4};
5
6/// A boxed asynchronous workflow-store operation.
7pub type WorkflowStoreFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
8
9/// Validated isolation identity for workflow admission and control-plane access.
10#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
11pub struct WorkflowTenantId(String);
12
13impl WorkflowTenantId {
14    /// Validates a portable tenant identity.
15    ///
16    /// # Errors
17    ///
18    /// Rejects blank, oversized, or non-portable identifiers.
19    pub fn parse(value: impl Into<String>) -> Result<Self, WorkflowStoreError> {
20        let value = value.into();
21        if value.is_empty()
22            || value.len() > 128
23            || !value
24                .bytes()
25                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.'))
26        {
27            return Err(WorkflowStoreError::invalid_input(
28                "workflow tenant must contain 1..=128 portable ASCII characters",
29            ));
30        }
31        Ok(Self(value))
32    }
33
34    /// Returns the validated tenant identity.
35    pub fn as_str(&self) -> &str {
36        &self.0
37    }
38}
39
40impl Default for WorkflowTenantId {
41    fn default() -> Self {
42        Self("default".into())
43    }
44}
45
46/// Per-tenant workflow admission limits.
47#[derive(Clone, Copy, Debug, Eq, PartialEq)]
48pub struct WorkflowTenantPolicy {
49    max_outstanding_tasks: NonZeroU32,
50    max_concurrent_leases: NonZeroU32,
51}
52
53impl WorkflowTenantPolicy {
54    /// Creates a bounded tenant admission policy.
55    ///
56    /// # Errors
57    ///
58    /// Rejects zero limits or a lease limit larger than the outstanding limit.
59    pub fn new(
60        max_outstanding_tasks: u32,
61        max_concurrent_leases: u32,
62    ) -> Result<Self, WorkflowStoreError> {
63        let max_outstanding_tasks = NonZeroU32::new(max_outstanding_tasks).ok_or_else(|| {
64            WorkflowStoreError::invalid_input("tenant outstanding workflow limit must be positive")
65        })?;
66        let max_concurrent_leases = NonZeroU32::new(max_concurrent_leases).ok_or_else(|| {
67            WorkflowStoreError::invalid_input(
68                "tenant concurrent workflow lease limit must be positive",
69            )
70        })?;
71        if max_concurrent_leases > max_outstanding_tasks {
72            return Err(WorkflowStoreError::invalid_input(
73                "tenant concurrent workflow lease limit cannot exceed outstanding limit",
74            ));
75        }
76        Ok(Self {
77            max_outstanding_tasks,
78            max_concurrent_leases,
79        })
80    }
81
82    /// Maximum non-terminal tasks admitted for this tenant.
83    pub const fn max_outstanding_tasks(self) -> u32 {
84        self.max_outstanding_tasks.get()
85    }
86
87    /// Maximum unexpired leases concurrently owned for this tenant.
88    pub const fn max_concurrent_leases(self) -> u32 {
89        self.max_concurrent_leases.get()
90    }
91}
92
93impl Default for WorkflowTenantPolicy {
94    fn default() -> Self {
95        Self {
96            max_outstanding_tasks: NonZeroU32::new(10_000)
97                .expect("default outstanding limit is positive"),
98            max_concurrent_leases: NonZeroU32::new(100)
99                .expect("default concurrent lease limit is positive"),
100        }
101    }
102}
103
104/// Validated page size for discovering tenants with configured budgets.
105#[derive(Clone, Copy, Debug, Eq, PartialEq)]
106pub struct WorkflowTenantListLimit(NonZeroU32);
107
108impl WorkflowTenantListLimit {
109    /// Creates a bounded tenant-discovery page size.
110    ///
111    /// # Errors
112    ///
113    /// Rejects zero or values greater than 1,000.
114    pub fn new(value: u32) -> Result<Self, WorkflowStoreError> {
115        let value = NonZeroU32::new(value).ok_or_else(|| {
116            WorkflowStoreError::invalid_input("workflow tenant list limit must be positive")
117        })?;
118        if value.get() > 1_000 {
119            return Err(WorkflowStoreError::invalid_input(
120                "workflow tenant list limit cannot exceed 1,000",
121            ));
122        }
123        Ok(Self(value))
124    }
125
126    /// Returns the validated maximum number of tenants.
127    pub const fn get(self) -> u32 {
128        self.0.get()
129    }
130}
131
132impl Default for WorkflowTenantListLimit {
133    fn default() -> Self {
134        Self(NonZeroU32::new(100).expect("default tenant page size is positive"))
135    }
136}
137
138/// Persistent aggregate budget policy for one workflow tenant.
139#[derive(Clone, Copy, Debug, Eq, PartialEq)]
140pub struct WorkflowTenantBudgetPolicy {
141    limit: Budget,
142    window_ms: NonZeroU64,
143    recovery_grace_ms: u64,
144}
145
146impl WorkflowTenantBudgetPolicy {
147    /// Creates a fixed-window tenant budget with a crash-recovery grace period.
148    ///
149    /// # Errors
150    ///
151    /// Rejects an unbounded policy, invalid duration units, or overflowing
152    /// window and recovery durations.
153    pub fn new(
154        limit: Budget,
155        window: Duration,
156        recovery_grace: Duration,
157    ) -> Result<Self, WorkflowStoreError> {
158        if budget_is_unbounded(limit) {
159            return Err(WorkflowStoreError::invalid_input(
160                "tenant budget policy must limit at least one resource",
161            ));
162        }
163        validate_budget_duration(limit)?;
164        let window_ms = u64::try_from(window.as_millis())
165            .ok()
166            .and_then(NonZeroU64::new)
167            .ok_or_else(|| {
168                WorkflowStoreError::invalid_input(
169                    "tenant budget window must fit in positive whole milliseconds",
170                )
171            })?;
172        let recovery_grace_ms = u64::try_from(recovery_grace.as_millis()).map_err(|_| {
173            WorkflowStoreError::invalid_input(
174                "tenant budget recovery grace exceeds supported milliseconds",
175            )
176        })?;
177        Ok(Self {
178            limit,
179            window_ms,
180            recovery_grace_ms,
181        })
182    }
183
184    /// Returns the aggregate hard limit.
185    pub const fn limit(self) -> Budget {
186        self.limit
187    }
188
189    /// Returns the fixed-window length in milliseconds.
190    pub const fn window_millis(self) -> u64 {
191        self.window_ms.get()
192    }
193
194    /// Returns the reservation takeover grace in milliseconds.
195    pub const fn recovery_grace_millis(self) -> u64 {
196        self.recovery_grace_ms
197    }
198}
199
200/// Safe point-in-time view of one tenant budget ledger.
201#[derive(Clone, Debug, Eq, PartialEq)]
202pub struct WorkflowTenantBudgetSnapshot {
203    /// Tenant owning the ledger.
204    pub tenant_id: WorkflowTenantId,
205    /// Configured aggregate hard limit.
206    pub limit: Budget,
207    /// Store-authoritative start of the current draining window.
208    pub window_started_at_ms: u64,
209    /// Usage durably settled or conservatively forfeited in this window.
210    pub committed: Usage,
211    /// Upper bounds held by live workflow reservations.
212    pub reserved: Usage,
213    /// Number of reservations awaiting settlement or takeover.
214    pub active_reservations: u64,
215}
216
217/// Stable cursor for incrementally consuming one tenant's durable budget audit.
218#[derive(Clone, Copy, Debug, Default, Eq, Ord, PartialEq, PartialOrd)]
219pub struct WorkflowBudgetAuditCursor(u64);
220
221impl WorkflowBudgetAuditCursor {
222    /// Creates a cursor from a previously observed sequence.
223    pub const fn new(sequence: u64) -> Self {
224        Self(sequence)
225    }
226
227    /// Returns the durable sequence represented by this cursor.
228    pub const fn sequence(self) -> u64 {
229        self.0
230    }
231}
232
233/// Validated page size for tenant budget audit reads.
234#[derive(Clone, Copy, Debug, Eq, PartialEq)]
235pub struct WorkflowBudgetAuditLimit(NonZeroU32);
236
237impl WorkflowBudgetAuditLimit {
238    /// Creates a bounded audit page size.
239    ///
240    /// # Errors
241    ///
242    /// Rejects zero or values greater than 1,000.
243    pub fn new(value: u32) -> Result<Self, WorkflowStoreError> {
244        let value = NonZeroU32::new(value).ok_or_else(|| {
245            WorkflowStoreError::invalid_input("workflow budget audit limit must be positive")
246        })?;
247        if value.get() > 1_000 {
248            return Err(WorkflowStoreError::invalid_input(
249                "workflow budget audit limit cannot exceed 1,000",
250            ));
251        }
252        Ok(Self(value))
253    }
254
255    /// Returns the validated maximum number of events.
256    pub const fn get(self) -> u32 {
257        self.0.get()
258    }
259}
260
261impl Default for WorkflowBudgetAuditLimit {
262    fn default() -> Self {
263        Self(NonZeroU32::new(100).expect("default audit page size is positive"))
264    }
265}
266
267/// Stable identity of one independent tenant-budget audit projection.
268#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
269pub struct WorkflowBudgetAuditProjectionId(String);
270
271impl WorkflowBudgetAuditProjectionId {
272    /// Validates a portable projection identity.
273    ///
274    /// # Errors
275    ///
276    /// Rejects blank, oversized, or non-portable identifiers.
277    pub fn parse(value: impl Into<String>) -> Result<Self, WorkflowStoreError> {
278        let value = value.into();
279        if value.is_empty()
280            || value.len() > 128
281            || !value
282                .bytes()
283                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.'))
284        {
285            return Err(WorkflowStoreError::invalid_input(
286                "workflow budget audit projection must contain 1..=128 portable ASCII characters",
287            ));
288        }
289        Ok(Self(value))
290    }
291
292    /// Returns the validated projection identity.
293    pub fn as_str(&self) -> &str {
294        &self.0
295    }
296}
297
298/// Fenced ownership of one named tenant-budget audit projection.
299#[derive(Clone, Debug, Eq, PartialEq)]
300pub struct WorkflowBudgetAuditProjectionLease {
301    /// Tenant whose audit stream is projected.
302    pub tenant_id: WorkflowTenantId,
303    /// Stable identity of the projection.
304    pub projection_id: WorkflowBudgetAuditProjectionId,
305    /// Worker currently owning the projection.
306    pub owner: WorkerId,
307    /// Last fully acknowledged audit cursor.
308    pub cursor: WorkflowBudgetAuditCursor,
309    /// Monotonic token invalidating superseded projector instances.
310    pub fencing_token: u64,
311    /// Store-authoritative lease expiration in Unix milliseconds.
312    pub expires_at_ms: u64,
313}
314
315impl WorkflowBudgetAuditProjectionLease {
316    /// Tenant whose audit stream is projected.
317    pub fn tenant_id(&self) -> &WorkflowTenantId {
318        &self.tenant_id
319    }
320
321    /// Stable identity of the projection.
322    pub fn projection_id(&self) -> &WorkflowBudgetAuditProjectionId {
323        &self.projection_id
324    }
325
326    /// Worker currently owning the projection.
327    pub fn owner(&self) -> &WorkerId {
328        &self.owner
329    }
330
331    /// Last fully acknowledged audit cursor.
332    pub const fn cursor(&self) -> WorkflowBudgetAuditCursor {
333        self.cursor
334    }
335
336    /// Monotonic token invalidating superseded projector instances.
337    pub const fn fencing_token(&self) -> u64 {
338        self.fencing_token
339    }
340
341    /// Store-authoritative lease expiration in Unix milliseconds.
342    pub const fn expires_at_ms(&self) -> u64 {
343        self.expires_at_ms
344    }
345}
346
347/// Why uncertain reserved capacity was conservatively committed.
348#[derive(Clone, Copy, Debug, Eq, PartialEq)]
349#[non_exhaustive]
350pub enum WorkflowBudgetForfeitReason {
351    /// An operator cancelled the workflow before settlement.
352    Cancelled,
353    /// No fenced successor adopted the reservation before its recovery grace elapsed.
354    RecoveryExpired,
355}
356
357/// One durable tenant-budget decision.
358#[derive(Clone, Copy, Debug, Eq, PartialEq)]
359#[non_exhaustive]
360pub enum WorkflowBudgetAuditKind {
361    /// A policy was created or replaced.
362    PolicyConfigured,
363    /// A new workflow envelope was reserved.
364    Reserved,
365    /// A successor lease adopted a recoverable envelope.
366    Adopted,
367    /// Aggregate capacity rejected a requested envelope.
368    AdmissionDenied,
369    /// Observed cumulative usage exceeded the workflow's reserved envelope.
370    UsageExceeded,
371    /// Actual observed usage was committed and unused capacity released.
372    Settled,
373    /// Uncertain capacity was conservatively committed.
374    Forfeited(WorkflowBudgetForfeitReason),
375    /// A fully drained fixed window advanced.
376    WindowReset,
377}
378
379/// Immutable audit fact recorded with a tenant-budget state transition.
380#[derive(Clone, Debug, Eq, PartialEq)]
381pub struct WorkflowBudgetAuditEvent {
382    /// Monotonic sequence within this tenant's budget ledger.
383    pub cursor: WorkflowBudgetAuditCursor,
384    /// Tenant owning the decision.
385    pub tenant_id: WorkflowTenantId,
386    /// Workflow checkpoint involved, when the decision is workflow-specific.
387    pub checkpoint_id: Option<CheckpointId>,
388    /// Store-authoritative event time in Unix milliseconds.
389    pub occurred_at_ms: u64,
390    /// Stable decision category.
391    pub kind: WorkflowBudgetAuditKind,
392    /// Envelope or committed delta associated with the decision.
393    pub usage: Usage,
394    /// Age of the affected reservation, when applicable.
395    pub reservation_age_ms: Option<u64>,
396    /// Policy active when the event was recorded.
397    pub limit: Budget,
398    /// Committed usage immediately after the decision.
399    pub committed: Usage,
400    /// Reserved usage immediately after the decision.
401    pub reserved: Usage,
402}
403
404/// Outcome of attempting to reserve a tenant budget envelope.
405#[derive(Clone, Copy, Debug, Eq, PartialEq)]
406#[non_exhaustive]
407pub enum WorkflowBudgetReservationOutcome {
408    /// No aggregate tenant budget is configured.
409    NotConfigured,
410    /// The workflow envelope is durably reserved under the current lease.
411    Reserved,
412}
413
414/// Stable identity of a distributed workflow worker.
415#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
416pub struct WorkerId(String);
417
418impl WorkerId {
419    /// Validates a worker identity.
420    ///
421    /// # Errors
422    ///
423    /// Rejects blank, oversized, or non-portable identifiers.
424    pub fn parse(value: impl Into<String>) -> Result<Self, WorkflowStoreError> {
425        let value = value.into();
426        if value.is_empty()
427            || value.len() > 128
428            || !value
429                .bytes()
430                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.'))
431        {
432            return Err(WorkflowStoreError::invalid_input(
433                "worker identity must contain 1..=128 portable ASCII characters",
434            ));
435        }
436        Ok(Self(value))
437    }
438
439    /// Returns the validated worker identity.
440    pub fn as_str(&self) -> &str {
441        &self.0
442    }
443}
444
445/// Positive lease duration represented in whole milliseconds.
446#[derive(Clone, Copy, Debug, Eq, PartialEq)]
447pub struct LeaseDuration(NonZeroU64);
448
449impl LeaseDuration {
450    /// Validates and normalizes one lease duration.
451    ///
452    /// # Errors
453    ///
454    /// Rejects sub-millisecond, zero, or overflowing durations.
455    pub fn new(duration: Duration) -> Result<Self, WorkflowStoreError> {
456        let millis = u64::try_from(duration.as_millis())
457            .ok()
458            .and_then(NonZeroU64::new)
459            .ok_or_else(|| {
460                WorkflowStoreError::invalid_input(
461                    "workflow lease must fit in a positive whole-millisecond duration",
462                )
463            })?;
464        Ok(Self(millis))
465    }
466
467    /// Returns the normalized duration in milliseconds.
468    pub const fn as_millis(self) -> u64 {
469        self.0.get()
470    }
471}
472
473/// One durable workflow task awaiting execution.
474#[derive(Clone, Debug, PartialEq)]
475pub struct WorkflowTask {
476    /// Checkpoint identity shared across every retry and worker claim.
477    pub checkpoint_id: CheckpointId,
478    /// Tenant that owns admission and control-plane authority for this task.
479    pub tenant_id: WorkflowTenantId,
480    /// Stable workflow definition name.
481    pub workflow: String,
482    /// Caller-managed workflow definition version.
483    pub workflow_version: u32,
484    /// Canonical workflow input.
485    pub input: Value,
486    /// Higher values are claimed first.
487    pub priority: i32,
488}
489
490impl WorkflowTask {
491    /// Creates an immediately available workflow task.
492    ///
493    /// # Errors
494    ///
495    /// Rejects blank workflow names or version zero.
496    pub fn new(
497        workflow: impl Into<String>,
498        workflow_version: u32,
499        input: Value,
500    ) -> Result<Self, WorkflowStoreError> {
501        let workflow = workflow.into();
502        if workflow.trim().is_empty() || workflow.len() > 256 {
503            return Err(WorkflowStoreError::invalid_input(
504                "workflow name must contain 1..=256 bytes",
505            ));
506        }
507        if workflow_version == 0 {
508            return Err(WorkflowStoreError::invalid_input(
509                "workflow version must be greater than zero",
510            ));
511        }
512        Ok(Self {
513            checkpoint_id: CheckpointId::new(),
514            tenant_id: WorkflowTenantId::default(),
515            workflow,
516            workflow_version,
517            input,
518            priority: 0,
519        })
520    }
521
522    /// Assigns the task to an explicit tenant.
523    #[must_use]
524    pub fn with_tenant(mut self, tenant_id: WorkflowTenantId) -> Self {
525        self.tenant_id = tenant_id;
526        self
527    }
528
529    /// Uses an existing checkpoint identity.
530    #[must_use]
531    pub const fn with_checkpoint_id(mut self, checkpoint_id: CheckpointId) -> Self {
532        self.checkpoint_id = checkpoint_id;
533        self
534    }
535
536    /// Sets the task priority.
537    #[must_use]
538    pub const fn with_priority(mut self, priority: i32) -> Self {
539        self.priority = priority;
540        self
541    }
542
543    /// Revalidates invariants after deserialization or external construction.
544    ///
545    /// # Errors
546    ///
547    /// Rejects blank workflow names or version zero.
548    pub fn validate(&self) -> Result<(), WorkflowStoreError> {
549        if self.workflow.trim().is_empty() || self.workflow.len() > 256 {
550            return Err(WorkflowStoreError::invalid_input(
551                "workflow name must contain 1..=256 bytes",
552            ));
553        }
554        if self.workflow_version == 0 {
555            return Err(WorkflowStoreError::invalid_input(
556                "workflow version must be greater than zero",
557            ));
558        }
559        Ok(())
560    }
561}
562
563/// Fenced ownership of one workflow task.
564#[derive(Clone, Debug, Eq, PartialEq)]
565pub struct WorkflowLease {
566    /// Claimed workflow checkpoint.
567    pub checkpoint_id: CheckpointId,
568    /// Tenant that owns the leased task.
569    pub tenant_id: WorkflowTenantId,
570    /// Current worker owner.
571    pub worker: WorkerId,
572    /// Monotonic fencing token incremented on every successful claim.
573    pub fencing_token: u64,
574    /// One-based claim attempt.
575    pub attempt: u64,
576    /// Store-authoritative lease expiration in Unix milliseconds.
577    pub expires_at_ms: u64,
578}
579
580/// A workflow task and its fenced worker lease.
581#[derive(Clone, Debug, PartialEq)]
582pub struct ClaimedWorkflow {
583    /// Durable task definition.
584    pub task: WorkflowTask,
585    /// Current fenced ownership.
586    pub lease: WorkflowLease,
587    /// Durable wake value retained across lease takeover until the next wait.
588    pub wake: Option<WorkflowWake>,
589}
590
591/// Durable task disposition accepted from the current lease owner.
592#[derive(Clone, Debug, Eq, PartialEq)]
593#[non_exhaustive]
594pub enum WorkflowDisposition {
595    /// The checkpoint contains a terminal successful outcome.
596    Completed,
597    /// Return the task to the queue after a store-relative delay.
598    RetryAfter(Duration),
599    /// Release the worker while waiting for time or an external signal.
600    Suspend(WorkflowWait),
601    /// Stop automatic execution with a safe operator-facing reason.
602    Failed(String),
603    /// Stop execution because the task was explicitly cancelled.
604    Cancelled,
605}
606
607/// Durable workflow queue state.
608#[derive(Clone, Copy, Debug, Eq, PartialEq)]
609#[non_exhaustive]
610pub enum WorkflowTaskStatus {
611    /// Available now or after a retry delay.
612    Queued,
613    /// Owned by a worker until its lease expires.
614    Leased,
615    /// Persisted without a worker lease until its wake condition is satisfied.
616    Waiting,
617    /// Successfully completed.
618    Completed,
619    /// Permanently failed.
620    Failed,
621    /// Explicitly cancelled.
622    Cancelled,
623}
624
625/// Minimum time a terminal workflow remains queryable before physical cleanup.
626#[derive(Clone, Copy, Debug, Eq, PartialEq)]
627pub struct WorkflowTaskRetention(NonZeroU64);
628
629impl WorkflowTaskRetention {
630    /// Creates a positive whole-millisecond retention.
631    ///
632    /// # Errors
633    ///
634    /// Rejects zero, sub-millisecond, or overflowing durations.
635    pub fn new(duration: Duration) -> Result<Self, WorkflowStoreError> {
636        let millis = u64::try_from(duration.as_millis())
637            .ok()
638            .and_then(NonZeroU64::new)
639            .ok_or_else(|| {
640                WorkflowStoreError::invalid_input(
641                    "workflow Task retention must fit in positive whole milliseconds",
642                )
643            })?;
644        Ok(Self(millis))
645    }
646
647    /// Returns normalized retention milliseconds.
648    pub const fn as_millis(self) -> u64 {
649        self.0.get()
650    }
651}
652
653/// Maximum terminal Tasks removed by one fenced cleanup operation.
654#[derive(Clone, Copy, Debug, Eq, PartialEq)]
655pub struct WorkflowTaskCleanupLimit(NonZeroU32);
656
657impl WorkflowTaskCleanupLimit {
658    /// Creates a bounded cleanup batch size.
659    ///
660    /// # Errors
661    ///
662    /// Rejects zero or values greater than 1,000.
663    pub fn new(value: u32) -> Result<Self, WorkflowStoreError> {
664        let value = NonZeroU32::new(value).ok_or_else(|| {
665            WorkflowStoreError::invalid_input("workflow Task cleanup limit must be positive")
666        })?;
667        if value.get() > 1_000 {
668            return Err(WorkflowStoreError::invalid_input(
669                "workflow Task cleanup limit cannot exceed 1,000",
670            ));
671        }
672        Ok(Self(value))
673    }
674
675    /// Returns the validated cleanup batch size.
676    pub const fn get(self) -> u32 {
677        self.0.get()
678    }
679}
680
681/// Stable cursor for paginating immutable terminal Task tombstones.
682#[derive(Clone, Copy, Debug, Default, Eq, Ord, PartialEq, PartialOrd)]
683pub struct WorkflowTaskTombstoneCursor(u64);
684
685impl WorkflowTaskTombstoneCursor {
686    /// Creates a cursor from its durable sequence.
687    pub const fn new(sequence: u64) -> Self {
688        Self(sequence)
689    }
690
691    /// Returns the durable sequence.
692    pub const fn get(self) -> u64 {
693        self.0
694    }
695}
696
697/// Maximum tombstones returned by one audit page.
698#[derive(Clone, Copy, Debug, Eq, PartialEq)]
699pub struct WorkflowTaskTombstoneLimit(NonZeroU32);
700
701impl WorkflowTaskTombstoneLimit {
702    /// Creates a bounded tombstone page size.
703    ///
704    /// # Errors
705    ///
706    /// Rejects zero or values greater than 1,000.
707    pub fn new(value: u32) -> Result<Self, WorkflowStoreError> {
708        let value = NonZeroU32::new(value).ok_or_else(|| {
709            WorkflowStoreError::invalid_input("workflow Task tombstone limit must be positive")
710        })?;
711        if value.get() > 1_000 {
712            return Err(WorkflowStoreError::invalid_input(
713                "workflow Task tombstone limit cannot exceed 1,000",
714            ));
715        }
716        Ok(Self(value))
717    }
718
719    /// Returns the validated audit page size.
720    pub const fn get(self) -> u32 {
721        self.0.get()
722    }
723}
724
725/// Fenced ownership of one tenant's terminal Task cleanup partition.
726#[derive(Clone, Debug, Eq, PartialEq)]
727pub struct WorkflowTaskCleanupLease {
728    /// Tenant whose terminal Tasks may be removed.
729    pub tenant_id: WorkflowTenantId,
730    /// Exclusive cleanup owner.
731    pub owner: WorkerId,
732    /// Monotonic token incremented on every successful claim.
733    pub fencing_token: u64,
734    /// Store-authoritative expiration in Unix milliseconds.
735    pub expires_at_ms: u64,
736}
737
738/// Immutable audit fact retained after a terminal Task is physically removed.
739#[derive(Clone, Debug, Eq, PartialEq)]
740pub struct WorkflowTaskTombstone {
741    /// Monotonic tenant-independent storage cursor.
742    pub cursor: WorkflowTaskTombstoneCursor,
743    /// Removed workflow identity.
744    pub checkpoint_id: CheckpointId,
745    /// Tenant that owned the removed Task.
746    pub tenant_id: WorkflowTenantId,
747    /// Stable workflow definition.
748    pub workflow: String,
749    /// Exact workflow version.
750    pub workflow_version: u32,
751    /// Terminal state observed atomically with deletion.
752    pub final_status: WorkflowTaskStatus,
753    /// Original store-authoritative creation time.
754    pub created_at_ms: u64,
755    /// Store-authoritative terminal transition time.
756    pub terminal_at_ms: u64,
757    /// Store-authoritative physical deletion time.
758    pub deleted_at_ms: u64,
759}
760
761/// Result of an idempotent external workflow cancellation.
762#[derive(Clone, Copy, Debug, Eq, PartialEq)]
763#[non_exhaustive]
764pub enum WorkflowCancelOutcome {
765    /// A queued, waiting, or leased workflow became cancelled.
766    Cancelled,
767    /// The workflow was already in a terminal state.
768    AlreadyTerminal,
769}
770
771/// Safe operator snapshot of one workflow task.
772#[derive(Clone, Debug, Eq, PartialEq)]
773pub struct WorkflowTaskSnapshot {
774    /// Stable task identity.
775    pub checkpoint_id: CheckpointId,
776    /// Tenant that owns the task.
777    pub tenant_id: WorkflowTenantId,
778    /// Stable workflow definition name.
779    pub workflow: String,
780    /// Exact workflow definition version.
781    pub workflow_version: u32,
782    /// Current durable state.
783    pub status: WorkflowTaskStatus,
784    /// Store-authoritative creation time in Unix milliseconds.
785    pub created_at_ms: u64,
786    /// Store-authoritative last state-transition time in Unix milliseconds.
787    pub updated_at_ms: u64,
788    /// Number of successful claims.
789    pub attempts: u64,
790    /// Current fencing token.
791    pub fencing_token: u64,
792    /// Current lease owner, when leased.
793    pub owner: Option<WorkerId>,
794    /// Lease expiration, when leased.
795    pub lease_expires_at_ms: Option<u64>,
796    /// Pending human-review request, when the task is interrupted.
797    pub interrupt: Option<WorkflowInterruptRequest>,
798    /// Safe terminal failure explanation, when failed.
799    pub failure_message: Option<String>,
800    /// Immutable parent relation when this task was forked from history.
801    pub lineage: Option<WorkflowLineage>,
802}
803
804/// Stable workflow-store failure category.
805#[derive(Clone, Copy, Debug, Eq, PartialEq)]
806#[non_exhaustive]
807pub enum WorkflowStoreErrorKind {
808    /// Input violated a domain invariant.
809    InvalidInput,
810    /// The requested task does not exist.
811    NotFound,
812    /// A create-only operation found an existing task.
813    Conflict,
814    /// The caller no longer owns the current unexpired lease.
815    LeaseLost,
816    /// A tenant admission limit prevented the operation.
817    AdmissionDenied,
818    /// The supplied tenant does not own the requested resource.
819    TenantMismatch,
820    /// The backing store failed.
821    Storage,
822}
823
824/// Typed workflow-store failure.
825#[derive(Clone, Debug, Error, Eq, PartialEq)]
826#[error("{kind:?}: {message}")]
827pub struct WorkflowStoreError {
828    /// Stable failure category.
829    pub kind: WorkflowStoreErrorKind,
830    /// Safe application-facing explanation.
831    pub message: String,
832}
833
834impl WorkflowStoreError {
835    /// Creates a normalized store failure.
836    pub fn new(kind: WorkflowStoreErrorKind, message: impl Into<String>) -> Self {
837        Self {
838            kind,
839            message: message.into(),
840        }
841    }
842
843    pub(super) fn invalid_input(message: impl Into<String>) -> Self {
844        Self::new(WorkflowStoreErrorKind::InvalidInput, message)
845    }
846}
847
848fn budget_is_unbounded(budget: Budget) -> bool {
849    budget.tokens.is_none()
850        && budget.cost_microusd.is_none()
851        && budget.duration.is_none()
852        && budget.turns.is_none()
853        && budget.tool_calls.is_none()
854        && budget.delegations.is_none()
855}
856
857fn validate_budget_duration(budget: Budget) -> Result<(), WorkflowStoreError> {
858    if let Some(duration) = budget.duration {
859        duration_micros(duration)?;
860    }
861    Ok(())
862}
863
864pub(super) fn duration_micros(duration: Duration) -> Result<u64, WorkflowStoreError> {
865    u64::try_from(duration.as_micros()).map_err(|_| {
866        WorkflowStoreError::invalid_input("budget duration exceeds supported microseconds")
867    })
868}
runifold_workflow/store/model.rs

runifold_workflow/store/
model.rs
