runifold_workflow/store/
model.rs

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

runifold_workflow/store/
model.rs
