Skip to main content

aion/durability/
resolver.rs

1//! Resolver: recorded/resume-live/violation decision.
2
3use aion_core::{Event, WorkflowError, WorkflowId};
4use chrono::{DateTime, Utc};
5
6use crate::durability::{
7    Command, CorrelationKey, CursorResolveResult, DurabilityError, HistoryCursor,
8    NonDeterminismError, RecordedEventFamily, Recorder, Resolution, ResolveOutcome,
9    cursor::{ChildTerminalResolveResult, FoundEventDescriptor},
10};
11
12/// Stable [`WorkflowError::message`] prefix used when a replay violation fails a workflow.
13///
14/// Until `aion-core` grows a dedicated workflow-error classification enum, AE can surface this
15/// prefix as the non-determinism failure classification for terminal [`Event::WorkflowFailed`]
16/// records produced by [`fail_on_violation`].
17pub const NON_DETERMINISM_WORKFLOW_ERROR_PREFIX: &str = "non-determinism violation";
18
19/// Single durability chokepoint that resolves workflow commands against recorded history.
20#[derive(Clone, Debug)]
21pub struct Resolver {
22    workflow_id: WorkflowId,
23    cursor: HistoryCursor,
24}
25
26impl Resolver {
27    /// Creates a resolver for one workflow history.
28    ///
29    /// The workflow id is retained for typed non-determinism diagnostics; AD-006 wires the
30    /// determinism-context timestamp hook at this same chokepoint.
31    #[must_use]
32    pub const fn new(workflow_id: WorkflowId, cursor: HistoryCursor) -> Self {
33        Self {
34            workflow_id,
35            cursor,
36        }
37    }
38
39    /// Returns the ordered history snapshot backing this resolver.
40    #[must_use]
41    pub fn history(&self) -> &[Event] {
42        self.cursor.events()
43    }
44
45    /// Resolves a command from recorded history or returns [`ResolveOutcome::ResumeLive`].
46    ///
47    /// # Errors
48    ///
49    /// Returns [`DurabilityError::NonDeterminism`] when the cursor reports a command-stream
50    /// mismatch, or [`DurabilityError::HistoryShape`] when matched history lacks one of AD-004's
51    /// recorded terminal outcomes.
52    pub fn resolve(&mut self, command: Command) -> Result<ResolveOutcome, DurabilityError> {
53        self.resolve_with_consumed(command)
54            .map(ResolvedCommand::into_outcome)
55    }
56
57    /// Returns the correlation ordinal at the current replay cursor for the requested family.
58    #[must_use]
59    pub fn next_command_ordinal(&self, family: RecordedEventFamily) -> Option<u64> {
60        self.cursor.next_key(family).and_then(|key| match key {
61            CorrelationKey::Activity(ordinal) | CorrelationKey::Child(ordinal) => Some(ordinal),
62            CorrelationKey::Signal { .. } | CorrelationKey::Timer(_) => None,
63        })
64    }
65
66    /// Advance the cursor past commands consumed by earlier resolver
67    /// instances of the same live execution. See
68    /// [`HistoryCursor::fast_forward_to_key`].
69    pub fn fast_forward_to(&mut self, key: &CorrelationKey) {
70        self.cursor.fast_forward_to_key(key);
71    }
72
73    /// Advance the cursor to the recorded terminal outcome for one awaited
74    /// child workflow. See [`HistoryCursor::fast_forward_to_child_terminal`].
75    pub fn fast_forward_to_child_terminal(&mut self, child_workflow_id: &WorkflowId) {
76        self.cursor
77            .fast_forward_to_child_terminal(child_workflow_id);
78    }
79
80    /// Resolves a command and includes the consumed recorded timestamp for replay bookkeeping.
81    ///
82    /// The returned [`ResolvedCommand`] preserves the existing recorded/resume-live decision while
83    /// exposing the timestamp of the last consumed history event. Replay uses that timestamp as the
84    /// only source for advancing workflow-visible `now`.
85    ///
86    /// # Errors
87    ///
88    /// Returns [`DurabilityError::NonDeterminism`] when the cursor reports a command-stream
89    /// mismatch, or [`DurabilityError::HistoryShape`] when matched history lacks one of AD-004's
90    /// recorded terminal outcomes.
91    pub fn resolve_with_consumed(
92        &mut self,
93        command: Command,
94    ) -> Result<ResolvedCommand, DurabilityError> {
95        if let Command::AwaitChild { child_workflow_id } = command {
96            return match self.cursor.resolve_child_terminal(&child_workflow_id) {
97                ChildTerminalResolveResult::Matched(events) => resolution_from_matched(&events),
98                ChildTerminalResolveResult::Exhausted => {
99                    Ok(ResolvedCommand::ResumeLive { recorded_at: None })
100                }
101                ChildTerminalResolveResult::Mismatch { found } => Err(NonDeterminismError {
102                    workflow_id: self.workflow_id.clone(),
103                    seq: found.seq,
104                    expected: format!(
105                        "Child terminal outcome for child workflow {child_workflow_id}"
106                    ),
107                    found: format!(
108                        "{} family {:?} key {:?}",
109                        found.kind, found.family, found.key
110                    ),
111                }
112                .into()),
113            };
114        }
115
116        let Some((family, key)) = family_and_key(command) else {
117            return Ok(ResolvedCommand::ResumeLive { recorded_at: None });
118        };
119
120        match self.cursor.resolve_next(family, key) {
121            CursorResolveResult::Matched(events) => resolution_from_matched(&events),
122            CursorResolveResult::Exhausted => Ok(ResolvedCommand::ResumeLive { recorded_at: None }),
123            CursorResolveResult::Mismatch {
124                expected_key,
125                found,
126            } => Err(self.mismatch_error(family, &expected_key, &found).into()),
127        }
128    }
129
130    fn mismatch_error(
131        &self,
132        expected_family: RecordedEventFamily,
133        expected_key: &CorrelationKey,
134        found: &FoundEventDescriptor,
135    ) -> NonDeterminismError {
136        NonDeterminismError {
137            workflow_id: self.workflow_id.clone(),
138            seq: found.seq,
139            expected: format!("{expected_family:?} {expected_key:?}"),
140            found: format!(
141                "{} family {:?} key {:?}",
142                found.kind, found.family, found.key
143            ),
144        }
145    }
146}
147
148/// Resolver outcome plus timestamp metadata for replay determinism bookkeeping.
149#[derive(Clone, Debug, PartialEq)]
150pub enum ResolvedCommand {
151    /// The command was satisfied from recorded history without invoking live side effects.
152    Recorded {
153        /// Recorded resolution returned to workflow code.
154        resolution: Resolution,
155        /// Timestamp of the last recorded event consumed for this command.
156        recorded_at: DateTime<Utc>,
157    },
158    /// Recorded history cannot fully satisfy the command; ownership must hand off live.
159    ResumeLive {
160        /// Timestamp of a matched command-issued event consumed before handoff, if any.
161        recorded_at: Option<DateTime<Utc>>,
162    },
163}
164
165impl ResolvedCommand {
166    fn into_outcome(self) -> ResolveOutcome {
167        match self {
168            Self::Recorded { resolution, .. } => ResolveOutcome::Recorded(resolution),
169            Self::ResumeLive { .. } => ResolveOutcome::ResumeLive,
170        }
171    }
172}
173
174/// Records the deterministic terminal failure caused by a replay non-determinism violation.
175///
176/// The supplied [`Recorder`] remains the only append path, preserving the single-writer sequence
177/// discipline. The caller supplies `recorded_at`; this helper does not read the wall clock for a
178/// workflow-visible terminal event. Call this once at the violation handling site so one violation
179/// produces exactly one [`Event::WorkflowFailed`].
180///
181/// # Errors
182///
183/// Returns [`DurabilityError`] if the recorder cannot append the terminal failure event.
184pub async fn fail_on_violation(
185    recorder: &mut Recorder,
186    recorded_at: DateTime<Utc>,
187    violation: &NonDeterminismError,
188) -> Result<(), DurabilityError> {
189    let error = WorkflowError {
190        message: format!("{NON_DETERMINISM_WORKFLOW_ERROR_PREFIX}: {violation}"),
191        details: None,
192    };
193
194    recorder.record_workflow_failed(recorded_at, error).await
195}
196
197fn family_and_key(command: Command) -> Option<(RecordedEventFamily, CorrelationKey)> {
198    match command {
199        Command::RunActivity { key, .. } => Some((RecordedEventFamily::Activity, key)),
200        Command::AwaitSignal { key } | Command::SendSignal { key, .. } => {
201            Some((RecordedEventFamily::Signal, key))
202        }
203        Command::StartTimer { key, .. } => Some((RecordedEventFamily::Timer, key)),
204        Command::SpawnChild { key, .. } => Some((RecordedEventFamily::Child, key)),
205        Command::AwaitChild { .. } | Command::CompleteWorkflow { .. } => None,
206    }
207}
208
209fn resolution_from_matched(events: &[Event]) -> Result<ResolvedCommand, DurabilityError> {
210    let Some(last) = events.last() else {
211        return Err(DurabilityError::HistoryShape {
212            reason: "cursor returned an empty matched event range".to_owned(),
213        });
214    };
215    let recorded_at = *last.recorded_at();
216
217    match last {
218        Event::ActivityCompleted { result, .. } => Ok(recorded(
219            Resolution::ActivityCompleted(result.clone()),
220            recorded_at,
221        )),
222        Event::ActivityFailed { error, .. }
223            if matches!(error.kind, aion_core::ActivityErrorKind::Terminal) =>
224        {
225            Ok(recorded(
226                Resolution::ActivityFailedTerminal(error.clone()),
227                recorded_at,
228            ))
229        }
230        Event::ActivityFailed { error, .. } => Err(DurabilityError::HistoryShape {
231            reason: format!(
232                "matched activity failure is not terminal and is not representable by AD-004 resolution: {:?}",
233                error.kind
234            ),
235        }),
236        Event::TimerFired { .. } => Ok(recorded(Resolution::TimerFired, recorded_at)),
237        Event::TimerCancelled { .. } => Ok(recorded(Resolution::TimerCancelled, recorded_at)),
238        Event::WithTimeoutCompleted {
239            outcome, result, ..
240        } => Ok(recorded(
241            Resolution::WithTimeout {
242                outcome: outcome.clone(),
243                result: result.clone(),
244            },
245            recorded_at,
246        )),
247        Event::TimerStarted { .. } => Ok(recorded(Resolution::TimerStarted, recorded_at)),
248        Event::SignalReceived { payload, .. } => Ok(recorded(
249            Resolution::SignalDelivered(payload.clone()),
250            recorded_at,
251        )),
252        Event::SignalSent { .. } => Ok(recorded(Resolution::SignalSent, recorded_at)),
253        Event::ChildWorkflowCompleted { result, .. } => Ok(recorded(
254            Resolution::ChildCompleted(result.clone()),
255            recorded_at,
256        )),
257        Event::ChildWorkflowFailed { error, .. } => Ok(recorded(
258            Resolution::ChildFailed(error.clone()),
259            recorded_at,
260        )),
261        Event::ChildWorkflowStarted {
262            child_workflow_id, ..
263        } => Ok(recorded(
264            Resolution::ChildStarted(child_workflow_id.clone()),
265            recorded_at,
266        )),
267        Event::ActivityCancelled { .. } | Event::ChildWorkflowCancelled { .. } => {
268            Err(DurabilityError::HistoryShape {
269                reason: format!(
270                    "recorded cancellation outcome is not representable by AD-004 resolution: {}",
271                    event_kind(last)
272                ),
273            })
274        }
275        Event::WorkflowStarted { .. }
276        | Event::WorkflowCompleted { .. }
277        | Event::WorkflowFailed { .. }
278        | Event::WorkflowCancelled { .. }
279        | Event::WorkflowTimedOut { .. }
280        | Event::WorkflowContinuedAsNew { .. }
281        | Event::WorkflowReopened { .. }
282        | Event::WorkflowPaused { .. }
283        | Event::WorkflowResumed { .. }
284        | Event::SearchAttributesUpdated { .. }
285        | Event::ActivityScheduled { .. }
286        | Event::ActivityStarted { .. }
287        | Event::ScheduleCreated { .. }
288        | Event::ScheduleUpdated { .. }
289        | Event::SchedulePaused { .. }
290        | Event::ScheduleResumed { .. }
291        | Event::ScheduleDeleted { .. }
292        | Event::ScheduleTriggered { .. } => Err(DurabilityError::HistoryShape {
293            reason: format!(
294                "matched history ended without a recorded command outcome: {}",
295                event_kind(last)
296            ),
297        }),
298    }
299}
300
301fn recorded(resolution: Resolution, recorded_at: DateTime<Utc>) -> ResolvedCommand {
302    ResolvedCommand::Recorded {
303        resolution,
304        recorded_at,
305    }
306}
307
308fn event_kind(event: &Event) -> &'static str {
309    match event {
310        Event::WorkflowStarted { .. } => "WorkflowStarted",
311        Event::WorkflowCompleted { .. } => "WorkflowCompleted",
312        Event::WorkflowFailed { .. } => "WorkflowFailed",
313        Event::WorkflowCancelled { .. } => "WorkflowCancelled",
314        Event::WorkflowTimedOut { .. } => "WorkflowTimedOut",
315        Event::WorkflowContinuedAsNew { .. } => "WorkflowContinuedAsNew",
316        Event::WorkflowReopened { .. } => "WorkflowReopened",
317        Event::WorkflowPaused { .. } => "WorkflowPaused",
318        Event::WorkflowResumed { .. } => "WorkflowResumed",
319        Event::SearchAttributesUpdated { .. } => "SearchAttributesUpdated",
320        Event::ActivityScheduled { .. } => "ActivityScheduled",
321        Event::ActivityStarted { .. } => "ActivityStarted",
322        Event::ActivityCompleted { .. } => "ActivityCompleted",
323        Event::ActivityFailed { .. } => "ActivityFailed",
324        Event::ActivityCancelled { .. } => "ActivityCancelled",
325        Event::TimerStarted { .. } => "TimerStarted",
326        Event::TimerFired { .. } => "TimerFired",
327        Event::TimerCancelled { .. } => "TimerCancelled",
328        Event::WithTimeoutCompleted { .. } => "WithTimeoutCompleted",
329        Event::SignalReceived { .. } => "SignalReceived",
330        Event::SignalSent { .. } => "SignalSent",
331        Event::ChildWorkflowStarted { .. } => "ChildWorkflowStarted",
332        Event::ChildWorkflowCompleted { .. } => "ChildWorkflowCompleted",
333        Event::ChildWorkflowFailed { .. } => "ChildWorkflowFailed",
334        Event::ChildWorkflowCancelled { .. } => "ChildWorkflowCancelled",
335        Event::ScheduleCreated { .. } => "ScheduleCreated",
336        Event::ScheduleUpdated { .. } => "ScheduleUpdated",
337        Event::SchedulePaused { .. } => "SchedulePaused",
338        Event::ScheduleResumed { .. } => "ScheduleResumed",
339        Event::ScheduleDeleted { .. } => "ScheduleDeleted",
340        Event::ScheduleTriggered { .. } => "ScheduleTriggered",
341    }
342}
343
344#[cfg(test)]
345mod tests {
346    use aion_core::{
347        ActivityError, ActivityErrorKind, ActivityId, Event, EventEnvelope, Payload, TimerId,
348        WorkflowError, WorkflowId,
349    };
350    use chrono::{DateTime, TimeZone, Utc};
351    use serde_json::json;
352    use uuid::Uuid;
353
354    use super::Resolver;
355    use crate::durability::{Command, CorrelationKey, HistoryCursor, Resolution, ResolveOutcome};
356
357    fn workflow_id() -> WorkflowId {
358        WorkflowId::new(Uuid::nil())
359    }
360
361    fn child_workflow_id() -> WorkflowId {
362        WorkflowId::new(Uuid::from_u128(1))
363    }
364
365    fn timestamp() -> Result<DateTime<Utc>, Box<dyn std::error::Error>> {
366        Utc.timestamp_opt(0, 0)
367            .single()
368            .ok_or_else(|| "invalid timestamp".into())
369    }
370
371    fn envelope(seq: u64) -> Result<EventEnvelope, Box<dyn std::error::Error>> {
372        Ok(EventEnvelope {
373            seq,
374            recorded_at: timestamp()?,
375            workflow_id: workflow_id(),
376        })
377    }
378
379    fn payload(label: &str) -> Result<Payload, Box<dyn std::error::Error>> {
380        Ok(Payload::from_json(&json!({ "label": label }))?)
381    }
382
383    fn workflow_error(message: &str) -> WorkflowError {
384        WorkflowError {
385            message: message.to_owned(),
386            details: None,
387        }
388    }
389
390    fn activity_scheduled(seq: u64, ordinal: u64) -> Result<Event, Box<dyn std::error::Error>> {
391        Ok(Event::ActivityScheduled {
392            envelope: envelope(seq)?,
393            activity_id: ActivityId::from_sequence_position(ordinal),
394            activity_type: "activity".to_owned(),
395            input: payload("activity-input")?,
396            task_queue: String::from("default"),
397            node: None,
398        })
399    }
400
401    fn activity_completed(
402        seq: u64,
403        ordinal: u64,
404        result: Payload,
405    ) -> Result<Event, Box<dyn std::error::Error>> {
406        Ok(Event::ActivityCompleted {
407            envelope: envelope(seq)?,
408            activity_id: ActivityId::from_sequence_position(ordinal),
409            result,
410            attempt: 1,
411        })
412    }
413
414    fn timer_started(seq: u64, timer_id: TimerId) -> Result<Event, Box<dyn std::error::Error>> {
415        Ok(Event::TimerStarted {
416            envelope: envelope(seq)?,
417            timer_id,
418            fire_at: timestamp()?,
419        })
420    }
421
422    fn timer_fired(seq: u64, timer_id: TimerId) -> Result<Event, Box<dyn std::error::Error>> {
423        Ok(Event::TimerFired {
424            envelope: envelope(seq)?,
425            timer_id,
426        })
427    }
428
429    fn signal_received(
430        seq: u64,
431        name: &str,
432        payload: Payload,
433    ) -> Result<Event, Box<dyn std::error::Error>> {
434        Ok(Event::SignalReceived {
435            envelope: envelope(seq)?,
436            name: name.to_owned(),
437            payload,
438        })
439    }
440
441    fn child_started(seq: u64) -> Result<Event, Box<dyn std::error::Error>> {
442        Ok(Event::ChildWorkflowStarted {
443            envelope: envelope(seq)?,
444            child_workflow_id: child_workflow_id(),
445            workflow_type: "child".to_owned(),
446            input: payload("child-input")?,
447            package_version: aion_core::PackageVersion::new("a".repeat(64)),
448        })
449    }
450
451    fn child_completed(seq: u64, result: Payload) -> Result<Event, Box<dyn std::error::Error>> {
452        Ok(Event::ChildWorkflowCompleted {
453            envelope: envelope(seq)?,
454            child_workflow_id: child_workflow_id(),
455            result,
456        })
457    }
458
459    fn run_activity_command(ordinal: u64) -> Result<Command, Box<dyn std::error::Error>> {
460        Ok(Command::RunActivity {
461            key: CorrelationKey::Activity(ordinal),
462            activity_type: "activity".to_owned(),
463            input: payload("activity-input")?,
464        })
465    }
466
467    #[test]
468    fn resolves_recorded_activity_then_resumes_live_at_history_end()
469    -> Result<(), Box<dyn std::error::Error>> {
470        let result = payload("activity-result")?;
471        let cursor = HistoryCursor::new(vec![
472            activity_scheduled(1, 0)?,
473            activity_completed(2, 0, result.clone())?,
474        ])?;
475        let mut resolver = Resolver::new(workflow_id(), cursor);
476
477        assert_eq!(
478            resolver.resolve(run_activity_command(0)?)?,
479            ResolveOutcome::Recorded(Resolution::ActivityCompleted(result))
480        );
481        assert_eq!(
482            resolver.resolve(run_activity_command(1)?)?,
483            ResolveOutcome::ResumeLive
484        );
485        Ok(())
486    }
487
488    #[test]
489    fn resolves_all_recorded_families_through_single_entry_point()
490    -> Result<(), Box<dyn std::error::Error>> {
491        let activity_result = payload("activity-result")?;
492        let signal_payload = payload("signal-payload")?;
493        let child_result = payload("child-result")?;
494        let timer_id = TimerId::anonymous(9);
495        let cursor = HistoryCursor::new(vec![
496            activity_scheduled(1, 0)?,
497            activity_completed(2, 0, activity_result.clone())?,
498            timer_started(3, timer_id.clone())?,
499            timer_fired(4, timer_id.clone())?,
500            signal_received(5, "ready", signal_payload.clone())?,
501            child_started(6)?,
502            child_completed(7, child_result.clone())?,
503        ])?;
504        let mut resolver = Resolver::new(workflow_id(), cursor);
505
506        assert_eq!(
507            resolver.resolve(run_activity_command(0)?)?,
508            ResolveOutcome::Recorded(Resolution::ActivityCompleted(activity_result))
509        );
510        assert_eq!(
511            resolver.resolve(Command::StartTimer {
512                key: CorrelationKey::Timer(timer_id),
513                fire_at: timestamp()?,
514            })?,
515            ResolveOutcome::Recorded(Resolution::TimerFired)
516        );
517        assert_eq!(
518            resolver.resolve(Command::AwaitSignal {
519                key: CorrelationKey::Signal {
520                    name: "ready".to_owned(),
521                    index: 0,
522                },
523            })?,
524            ResolveOutcome::Recorded(Resolution::SignalDelivered(signal_payload))
525        );
526        assert_eq!(
527            resolver.resolve(Command::SpawnChild {
528                // The first spawn in the run correlates positionally with the
529                // first recorded ChildWorkflowStarted, never with its seq.
530                key: CorrelationKey::Child(0),
531                workflow_type: "child".to_owned(),
532                input: payload("child-input")?,
533            })?,
534            ResolveOutcome::Recorded(Resolution::ChildStarted(child_workflow_id()))
535        );
536        assert_eq!(
537            resolver.resolve(Command::AwaitChild {
538                child_workflow_id: child_workflow_id(),
539            })?,
540            ResolveOutcome::Recorded(Resolution::ChildCompleted(child_result))
541        );
542        Ok(())
543    }
544
545    #[test]
546    fn maps_terminal_failures_to_recorded_resolutions() -> Result<(), Box<dyn std::error::Error>> {
547        let activity_error = ActivityError {
548            kind: ActivityErrorKind::Terminal,
549            message: "activity failed".to_owned(),
550            details: None,
551        };
552        let child_error = workflow_error("child failed");
553        let cursor = HistoryCursor::new(vec![
554            activity_scheduled(1, 0)?,
555            Event::ActivityFailed {
556                envelope: envelope(2)?,
557                activity_id: ActivityId::from_sequence_position(0),
558                error: activity_error.clone(),
559                attempt: 1,
560            },
561            child_started(3)?,
562            Event::ChildWorkflowFailed {
563                envelope: envelope(4)?,
564                child_workflow_id: child_workflow_id(),
565                error: child_error.clone(),
566            },
567        ])?;
568        let mut resolver = Resolver::new(workflow_id(), cursor);
569
570        assert_eq!(
571            resolver.resolve(run_activity_command(0)?)?,
572            ResolveOutcome::Recorded(Resolution::ActivityFailedTerminal(activity_error))
573        );
574        assert_eq!(
575            resolver.resolve(Command::SpawnChild {
576                key: CorrelationKey::Child(0),
577                workflow_type: "child".to_owned(),
578                input: payload("child-input")?,
579            })?,
580            ResolveOutcome::Recorded(Resolution::ChildStarted(child_workflow_id()))
581        );
582        assert_eq!(
583            resolver.resolve(Command::AwaitChild {
584                child_workflow_id: child_workflow_id(),
585            })?,
586            ResolveOutcome::Recorded(Resolution::ChildFailed(child_error))
587        );
588        Ok(())
589    }
590
591    #[test]
592    fn interleaved_async_arrival_inside_activity_range_replays_clean()
593    -> Result<(), Box<dyn std::error::Error>> {
594        let result = payload("activity-result")?;
595        let signal_payload = payload("signal-payload")?;
596        let cursor = HistoryCursor::new(vec![
597            activity_scheduled(1, 0)?,
598            signal_received(2, "mid", signal_payload.clone())?,
599            activity_completed(3, 0, result.clone())?,
600        ])?;
601        let mut resolver = Resolver::new(workflow_id(), cursor);
602
603        assert_eq!(
604            resolver.resolve(run_activity_command(0)?)?,
605            ResolveOutcome::Recorded(Resolution::ActivityCompleted(result))
606        );
607        assert_eq!(
608            resolver.resolve(Command::AwaitSignal {
609                key: CorrelationKey::Signal {
610                    name: "mid".to_owned(),
611                    index: 0,
612                },
613            })?,
614            ResolveOutcome::Recorded(Resolution::SignalDelivered(signal_payload))
615        );
616        Ok(())
617    }
618
619    #[test]
620    fn genuinely_reordered_activity_anchors_fail_typed() -> Result<(), Box<dyn std::error::Error>> {
621        let cursor = HistoryCursor::new(vec![
622            activity_scheduled(1, 0)?,
623            activity_scheduled(2, 1)?,
624            activity_completed(3, 1, payload("second-result")?)?,
625            activity_completed(4, 0, payload("first-result")?)?,
626        ])?;
627        let mut resolver = Resolver::new(workflow_id(), cursor);
628
629        let error = resolver.resolve(run_activity_command(1)?).err();
630
631        assert!(matches!(
632            error,
633            Some(crate::durability::DurabilityError::NonDeterminism(_))
634        ));
635        Ok(())
636    }
637
638    /// A history ending in a retryable (non-terminal) `ActivityFailed` is a
639    /// retry trail interrupted mid-flight — an engine crash during the
640    /// backoff window (#197). It must resolve `ResumeLive` so the activity
641    /// re-dispatches at the next attempt, never a history-shape fault (the
642    /// pre-#197 behaviour) and never a recorded terminal.
643    #[test]
644    fn dangling_retryable_failure_resumes_live_for_redispatch()
645    -> Result<(), Box<dyn std::error::Error>> {
646        let retryable_error = ActivityError {
647            kind: ActivityErrorKind::Retryable,
648            message: "retryable activity failure without later outcome".to_owned(),
649            details: None,
650        };
651        let cursor = HistoryCursor::new(vec![
652            activity_scheduled(1, 0)?,
653            Event::ActivityFailed {
654                envelope: envelope(2)?,
655                activity_id: ActivityId::from_sequence_position(0),
656                error: retryable_error,
657                attempt: 1,
658            },
659        ])?;
660        let mut resolver = Resolver::new(workflow_id(), cursor);
661
662        assert_eq!(
663            resolver.resolve(run_activity_command(0)?)?,
664            ResolveOutcome::ResumeLive,
665            "a dangling retry trail must hand off to a live re-dispatch"
666        );
667        Ok(())
668    }
669}