traverse_runtime/
durable_orchestration.rs

1//! Host-owned durable orchestration controls (spec `111-durable-dynamic-orchestration`).
2//!
3//! This module deliberately stores only governing identities and redacted execution
4//! state. Hosts own persistence and checkpoint authentication; an invalid checkpoint
5//! is never partially recovered or replanned.
6
7use serde::{Deserialize, Serialize};
8use std::collections::BTreeMap;
9
10const GOVERNING_SPEC: &str = "111-durable-dynamic-orchestration";
11
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
13pub struct GoverningSnapshots {
14    pub proposal_digest: String,
15    pub manifest_digest: String,
16    pub registry_digest: String,
17    pub policy_digest: String,
18    pub authorization_digest: String,
19}
20
21impl GoverningSnapshots {
22    #[must_use]
23    pub fn is_complete(&self) -> bool {
24        [
25            &self.proposal_digest,
26            &self.manifest_digest,
27            &self.registry_digest,
28            &self.policy_digest,
29            &self.authorization_digest,
30        ]
31        .iter()
32        .all(|digest| !digest.trim().is_empty())
33    }
34}
35
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37pub struct ExecutionLease {
38    pub owner_id: String,
39    pub fencing_token: u64,
40    /// Host-provided, comparable logical expiry; no wall clock is read here.
41    pub expires_at: u64,
42}
43
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct DurableWait {
46    pub kind: DurableWaitKind,
47    pub owner_id: String,
48    pub deadline: u64,
49    pub cancellation_id: String,
50}
51
52#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
53#[serde(rename_all = "snake_case")]
54pub enum DurableWaitKind {
55    Event,
56    Schedule,
57    HumanApproval,
58}
59
60#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
61pub struct RetryPolicy {
62    pub retryable: bool,
63    pub max_attempts: u32,
64    pub backoff_units: u64,
65    pub budget_units: u64,
66    pub idempotency_key: String,
67}
68
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
70pub struct CompensationStep {
71    pub capability_id: String,
72    pub authorization_digest: String,
73    pub order: u32,
74}
75
76#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
77pub struct DurableCheckpoint {
78    pub governing_spec: String,
79    pub execution_id: String,
80    pub snapshots: GoverningSnapshots,
81    pub lease: ExecutionLease,
82    pub completed_node_ids: Vec<String>,
83    pub wait: Option<DurableWait>,
84    pub retry: Option<RetryPolicy>,
85    pub compensation: Vec<CompensationStep>,
86    /// Host-authenticated opaque tag; never raw credentials or payloads.
87    pub authentication_tag: String,
88}
89
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91pub enum DurableOrchestrationErrorCode {
92    InvalidCheckpoint,
93    AuthenticationFailed,
94    SnapshotMismatch,
95    StaleLease,
96    WaitExpired,
97    Cancelled,
98    RetryNotDeclared,
99    RetryBudgetExhausted,
100    CompensationUnauthorized,
101}
102
103#[derive(Debug, Clone, PartialEq, Eq)]
104pub struct DurableOrchestrationError {
105    pub code: DurableOrchestrationErrorCode,
106    pub message: String,
107}
108
109type Result<T> = std::result::Result<T, DurableOrchestrationError>;
110
111/// Host boundary for authenticated persistence. Implementations must keep signing
112/// material outside `DurableCheckpoint` and atomically replace by fencing token.
113pub trait CheckpointStore {
114    /// # Errors
115    ///
116    /// Returns a stable persistence failure when the host cannot atomically save.
117    fn save(&mut self, checkpoint: DurableCheckpoint) -> Result<()>;
118    /// # Errors
119    ///
120    /// Returns a stable host storage failure when loading cannot complete.
121    fn load(&self, execution_id: &str) -> Result<Option<DurableCheckpoint>>;
122    fn verify(&self, checkpoint: &DurableCheckpoint) -> bool;
123}
124
125/// In-memory store for deterministic conformance tests. Production hosts provide
126/// authenticated durable storage through [`CheckpointStore`].
127#[derive(Default)]
128pub struct MemoryCheckpointStore {
129    checkpoints: BTreeMap<String, DurableCheckpoint>,
130}
131
132impl CheckpointStore for MemoryCheckpointStore {
133    fn save(&mut self, checkpoint: DurableCheckpoint) -> Result<()> {
134        self.checkpoints
135            .insert(checkpoint.execution_id.clone(), checkpoint);
136        Ok(())
137    }
138    fn load(&self, execution_id: &str) -> Result<Option<DurableCheckpoint>> {
139        Ok(self.checkpoints.get(execution_id).cloned())
140    }
141    fn verify(&self, checkpoint: &DurableCheckpoint) -> bool {
142        checkpoint.authentication_tag == authenticated_tag(checkpoint)
143    }
144}
145
146/// Constructs and persists a secret-free checkpoint before a wait or effect is
147/// reported. The caller supplies no authentication tag; it is derived by the host
148/// store's test implementation here and production stores replace it atomically.
149///
150/// # Errors
151///
152/// Returns an error for an incomplete checkpoint or a host persistence failure.
153pub fn persist_checkpoint(
154    store: &mut impl CheckpointStore,
155    mut checkpoint: DurableCheckpoint,
156) -> Result<()> {
157    validate_checkpoint(&checkpoint)?;
158    if checkpoint.authentication_tag.is_empty() {
159        checkpoint.authentication_tag = authenticated_tag(&checkpoint);
160    }
161    store.save(checkpoint)
162}
163
164/// Recovers only when the authenticated stored checkpoint exactly binds the current
165/// governing snapshots and has a non-expired lease. It never re-resolves artifacts.
166///
167/// # Errors
168///
169/// Returns a fail-closed error for missing, unauthenticated, mismatched, or stale state.
170pub fn recover_checkpoint(
171    store: &impl CheckpointStore,
172    execution_id: &str,
173    expected: &GoverningSnapshots,
174    now: u64,
175) -> Result<DurableCheckpoint> {
176    let checkpoint = store.load(execution_id)?.ok_or_else(|| {
177        error(
178            DurableOrchestrationErrorCode::InvalidCheckpoint,
179            "checkpoint is absent",
180        )
181    })?;
182    validate_checkpoint(&checkpoint)?;
183    if !store.verify(&checkpoint) {
184        return Err(error(
185            DurableOrchestrationErrorCode::AuthenticationFailed,
186            "checkpoint authentication failed",
187        ));
188    }
189    if &checkpoint.snapshots != expected {
190        return Err(error(
191            DurableOrchestrationErrorCode::SnapshotMismatch,
192            "checkpoint governing snapshots differ",
193        ));
194    }
195    if checkpoint.lease.expires_at <= now {
196        return Err(error(
197            DurableOrchestrationErrorCode::StaleLease,
198            "checkpoint lease has expired",
199        ));
200    }
201    Ok(checkpoint)
202}
203
204/// Transfers execution ownership only to a strictly newer fencing token.
205///
206/// # Errors
207///
208/// Returns `StaleLease` unless ownership advances both token and expiry.
209pub fn acquire_lease(
210    checkpoint: &mut DurableCheckpoint,
211    owner_id: &str,
212    fencing_token: u64,
213    expires_at: u64,
214) -> Result<()> {
215    if owner_id.trim().is_empty()
216        || fencing_token <= checkpoint.lease.fencing_token
217        || expires_at <= checkpoint.lease.expires_at
218    {
219        return Err(error(
220            DurableOrchestrationErrorCode::StaleLease,
221            "lease transfer requires a newer fencing token and expiry",
222        ));
223    }
224    checkpoint.lease = ExecutionLease {
225        owner_id: owner_id.to_string(),
226        fencing_token,
227        expires_at,
228    };
229    checkpoint.authentication_tag.clear();
230    Ok(())
231}
232
233/// Enforces declared retryability, attempt and budget bounds, and a stable idempotency key.
234///
235/// # Errors
236///
237/// Returns an error when retryability, idempotency, attempts, or budget are exhausted.
238pub fn authorize_retry(policy: &RetryPolicy, attempts_used: u32, budget_used: u64) -> Result<u64> {
239    if !policy.retryable || policy.idempotency_key.trim().is_empty() {
240        return Err(error(
241            DurableOrchestrationErrorCode::RetryNotDeclared,
242            "retryability and idempotency key must be declared",
243        ));
244    }
245    if attempts_used >= policy.max_attempts || budget_used >= policy.budget_units {
246        return Err(error(
247            DurableOrchestrationErrorCode::RetryBudgetExhausted,
248            "retry attempt or budget limit reached",
249        ));
250    }
251    Ok(policy.backoff_units)
252}
253
254/// Records a bounded wake decision. Cancellation wins over wake; expired waits do not wake.
255///
256/// # Errors
257///
258/// Returns a cancellation, expiry, or ownership error when the wake is not valid.
259pub fn wake_wait(wait: &DurableWait, wake_owner: &str, cancelled: bool, now: u64) -> Result<()> {
260    if cancelled {
261        return Err(error(
262            DurableOrchestrationErrorCode::Cancelled,
263            "wait was cancelled before wake",
264        ));
265    }
266    if now > wait.deadline {
267        return Err(error(
268            DurableOrchestrationErrorCode::WaitExpired,
269            "wait deadline elapsed before wake",
270        ));
271    }
272    if wake_owner != wait.owner_id {
273        return Err(error(
274            DurableOrchestrationErrorCode::StaleLease,
275            "wake owner does not own the wait",
276        ));
277    }
278    Ok(())
279}
280
281/// Returns compensation in reverse completed order after authorization binding checks.
282///
283/// # Errors
284///
285/// Returns `CompensationUnauthorized` when a step is not bound to the authorization snapshot.
286pub fn authorized_compensation(
287    steps: &[CompensationStep],
288    authorization_digest: &str,
289) -> Result<Vec<CompensationStep>> {
290    if steps.iter().any(|step| {
291        step.authorization_digest != authorization_digest || step.capability_id.trim().is_empty()
292    }) {
293        return Err(error(
294            DurableOrchestrationErrorCode::CompensationUnauthorized,
295            "compensation is not authorized by the execution snapshot",
296        ));
297    }
298    let mut ordered = steps.to_vec();
299    ordered.sort_by_key(|step| std::cmp::Reverse(step.order));
300    Ok(ordered)
301}
302
303fn validate_checkpoint(checkpoint: &DurableCheckpoint) -> Result<()> {
304    if checkpoint.governing_spec != GOVERNING_SPEC
305        || checkpoint.execution_id.trim().is_empty()
306        || !checkpoint.snapshots.is_complete()
307        || checkpoint.lease.owner_id.trim().is_empty()
308        || checkpoint.lease.fencing_token == 0
309    {
310        return Err(error(
311            DurableOrchestrationErrorCode::InvalidCheckpoint,
312            "checkpoint is incomplete or uses an unsupported governing spec",
313        ));
314    }
315    Ok(())
316}
317
318fn authenticated_tag(checkpoint: &DurableCheckpoint) -> String {
319    // A deterministic test tag, deliberately scoped to the in-memory adapter.
320    // Production adapters verify host-held MACs/signatures via `CheckpointStore::verify`.
321    format!(
322        "{}:{}:{}",
323        checkpoint.execution_id,
324        checkpoint.lease.fencing_token,
325        checkpoint.snapshots.proposal_digest
326    )
327}
328fn error(code: DurableOrchestrationErrorCode, message: &str) -> DurableOrchestrationError {
329    DurableOrchestrationError {
330        code,
331        message: message.to_string(),
332    }
333}
334
335#[cfg(test)]
336mod tests {
337    use super::*;
338    fn checkpoint() -> DurableCheckpoint {
339        DurableCheckpoint {
340            governing_spec: GOVERNING_SPEC.to_string(),
341            execution_id: "exec-1".to_string(),
342            snapshots: GoverningSnapshots {
343                proposal_digest: "proposal".to_string(),
344                manifest_digest: "manifest".to_string(),
345                registry_digest: "registry".to_string(),
346                policy_digest: "policy".to_string(),
347                authorization_digest: "auth".to_string(),
348            },
349            lease: ExecutionLease {
350                owner_id: "worker-a".to_string(),
351                fencing_token: 1,
352                expires_at: 20,
353            },
354            completed_node_ids: vec!["charge".to_string()],
355            wait: None,
356            retry: None,
357            compensation: vec![],
358            authentication_tag: String::new(),
359        }
360    }
361    #[test]
362    fn recovery_fails_closed_for_tamper_snapshot_and_stale_lease() {
363        let mut store = MemoryCheckpointStore::default();
364        let c = checkpoint();
365        assert!(persist_checkpoint(&mut store, c.clone()).is_ok());
366        assert!(matches!(
367            recover_checkpoint(&store, "exec-1", &c.snapshots, 10),
368            Ok(DurableCheckpoint { execution_id, .. }) if execution_id == "exec-1"
369        ));
370        let mut wrong = c.snapshots.clone();
371        wrong.policy_digest = "other".to_string();
372        assert!(matches!(
373            recover_checkpoint(&store, "exec-1", &wrong, 10),
374            Err(DurableOrchestrationError {
375                code: DurableOrchestrationErrorCode::SnapshotMismatch,
376                ..
377            })
378        ));
379        assert!(matches!(
380            recover_checkpoint(&store, "exec-1", &c.snapshots, 20),
381            Err(DurableOrchestrationError {
382                code: DurableOrchestrationErrorCode::StaleLease,
383                ..
384            })
385        ));
386    }
387    #[test]
388    fn retries_waits_and_compensation_are_bounded_and_declared() {
389        let policy = RetryPolicy {
390            retryable: true,
391            max_attempts: 2,
392            backoff_units: 3,
393            budget_units: 5,
394            idempotency_key: "key".to_string(),
395        };
396        assert_eq!(authorize_retry(&policy, 0, 0), Ok(3));
397        assert!(matches!(
398            authorize_retry(&policy, 2, 0),
399            Err(DurableOrchestrationError {
400                code: DurableOrchestrationErrorCode::RetryBudgetExhausted,
401                ..
402            })
403        ));
404        let wait = DurableWait {
405            kind: DurableWaitKind::Event,
406            owner_id: "worker".to_string(),
407            deadline: 10,
408            cancellation_id: "cancel".to_string(),
409        };
410        assert!(wake_wait(&wait, "worker", false, 10).is_ok());
411        assert!(matches!(
412            wake_wait(&wait, "worker", true, 1),
413            Err(DurableOrchestrationError {
414                code: DurableOrchestrationErrorCode::Cancelled,
415                ..
416            })
417        ));
418        let steps = vec![
419            CompensationStep {
420                capability_id: "undo-second".to_string(),
421                authorization_digest: "auth".to_string(),
422                order: 2,
423            },
424            CompensationStep {
425                capability_id: "undo-first".to_string(),
426                authorization_digest: "auth".to_string(),
427                order: 1,
428            },
429        ];
430        assert!(matches!(
431            authorized_compensation(&steps, "auth"),
432            Ok(ordered) if ordered[0].capability_id == "undo-second"
433        ));
434    }
435
436    #[test]
437    fn checkpoint_and_recovery_guards_cover_all_fail_closed_paths()
438    -> std::result::Result<(), String> {
439        let mut store = MemoryCheckpointStore::default();
440        let mut incomplete = checkpoint();
441        incomplete.snapshots.policy_digest.clear();
442        assert!(!incomplete.snapshots.is_complete());
443        assert!(matches!(
444            persist_checkpoint(&mut store, incomplete),
445            Err(DurableOrchestrationError {
446                code: DurableOrchestrationErrorCode::InvalidCheckpoint,
447                ..
448            })
449        ));
450        let checkpoint = checkpoint();
451        assert!(matches!(
452            recover_checkpoint(&store, "missing", &checkpoint.snapshots, 0),
453            Err(DurableOrchestrationError {
454                code: DurableOrchestrationErrorCode::InvalidCheckpoint,
455                ..
456            })
457        ));
458        assert!(persist_checkpoint(&mut store, checkpoint.clone()).is_ok());
459        let Ok(Some(mut tampered)) = store.load("exec-1") else {
460            return Err("saved checkpoint must remain loadable".to_string());
461        };
462        tampered.authentication_tag = "invalid".to_string();
463        store.checkpoints.insert("exec-1".to_string(), tampered);
464        assert!(matches!(
465            recover_checkpoint(&store, "exec-1", &checkpoint.snapshots, 0),
466            Err(DurableOrchestrationError {
467                code: DurableOrchestrationErrorCode::AuthenticationFailed,
468                ..
469            })
470        ));
471        Ok(())
472    }
473
474    #[test]
475    fn lease_retry_wait_and_compensation_error_paths_are_stable() {
476        let mut checkpoint = checkpoint();
477        assert!(matches!(
478            acquire_lease(&mut checkpoint, "", 2, 21),
479            Err(DurableOrchestrationError {
480                code: DurableOrchestrationErrorCode::StaleLease,
481                ..
482            })
483        ));
484        assert!(acquire_lease(&mut checkpoint, "worker-b", 2, 21).is_ok());
485        assert_eq!(checkpoint.lease.owner_id, "worker-b");
486        assert!(checkpoint.authentication_tag.is_empty());
487
488        let policy = RetryPolicy {
489            retryable: false,
490            max_attempts: 1,
491            backoff_units: 1,
492            budget_units: 1,
493            idempotency_key: String::new(),
494        };
495        assert!(matches!(
496            authorize_retry(&policy, 0, 0),
497            Err(DurableOrchestrationError {
498                code: DurableOrchestrationErrorCode::RetryNotDeclared,
499                ..
500            })
501        ));
502        let bounded_policy = RetryPolicy {
503            retryable: true,
504            idempotency_key: "key".to_string(),
505            ..policy
506        };
507        assert!(matches!(
508            authorize_retry(&bounded_policy, 0, 1),
509            Err(DurableOrchestrationError {
510                code: DurableOrchestrationErrorCode::RetryBudgetExhausted,
511                ..
512            })
513        ));
514
515        let wait = DurableWait {
516            kind: DurableWaitKind::Schedule,
517            owner_id: "worker-b".to_string(),
518            deadline: 10,
519            cancellation_id: "cancel".to_string(),
520        };
521        assert!(matches!(
522            wake_wait(&wait, "worker-b", false, 11),
523            Err(DurableOrchestrationError {
524                code: DurableOrchestrationErrorCode::WaitExpired,
525                ..
526            })
527        ));
528        assert!(matches!(
529            wake_wait(&wait, "worker-c", false, 10),
530            Err(DurableOrchestrationError {
531                code: DurableOrchestrationErrorCode::StaleLease,
532                ..
533            })
534        ));
535        let unauthorized = [CompensationStep {
536            capability_id: String::new(),
537            authorization_digest: "wrong".to_string(),
538            order: 1,
539        }];
540        assert!(matches!(
541            authorized_compensation(&unauthorized, "auth"),
542            Err(DurableOrchestrationError {
543                code: DurableOrchestrationErrorCode::CompensationUnauthorized,
544                ..
545            })
546        ));
547    }
548}
traverse_runtime/durable_orchestration.rs

traverse_runtime/
durable_orchestration.rs
