1use 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
12pub const NON_DETERMINISM_WORKFLOW_ERROR_PREFIX: &str = "non-determinism violation";
18
19#[derive(Clone, Debug)]
21pub struct Resolver {
22 workflow_id: WorkflowId,
23 cursor: HistoryCursor,
24}
25
26impl Resolver {
27 #[must_use]
32 pub const fn new(workflow_id: WorkflowId, cursor: HistoryCursor) -> Self {
33 Self {
34 workflow_id,
35 cursor,
36 }
37 }
38
39 #[must_use]
41 pub fn history(&self) -> &[Event] {
42 self.cursor.events()
43 }
44
45 pub fn resolve(&mut self, command: Command) -> Result<ResolveOutcome, DurabilityError> {
53 self.resolve_with_consumed(command)
54 .map(ResolvedCommand::into_outcome)
55 }
56
57 #[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 pub fn fast_forward_to(&mut self, key: &CorrelationKey) {
70 self.cursor.fast_forward_to_key(key);
71 }
72
73 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 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#[derive(Clone, Debug, PartialEq)]
150pub enum ResolvedCommand {
151 Recorded {
153 resolution: Resolution,
155 recorded_at: DateTime<Utc>,
157 },
158 ResumeLive {
160 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
174pub 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 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 #[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}