a3s_flow/engine/
mod.rs

1use chrono::{DateTime, Utc};
2use std::sync::Arc;
3use uuid::Uuid;
4
5use crate::error::{FlowError, Result};
6use crate::model::{
7    project_run, ActiveHookSnapshot, FlowEvent, FlowEventEnvelope, HookStatus, RuntimeCommand,
8    ScheduledWakeup, ScheduledWakeupKind, StepStatus, WaitStatus, WorkflowRunSnapshot,
9    WorkflowRunStatus, WorkflowRunSummary, WorkflowRunSuspension, WorkflowSpec,
10};
11use crate::observe::{FlowEventObserver, NoopFlowEventObserver};
12use crate::runtime::{FlowRuntime, WorkflowInvocation};
13use crate::store::{scheduled_wakeups_for_snapshot, FlowEventStore, InMemoryEventStore};
14
15mod hooks;
16mod operations;
17mod steps;
18mod validation;
19use steps::{interrupted_retry_exhaustion_event, StepExecutionContext};
20use validation::{
21    ensure_child_operation_matches, ensure_hook_command_matches, ensure_progress_matches,
22    ensure_retry_policy_valid, ensure_same_start, ensure_step_batch_valid,
23    ensure_step_command_matches, ensure_wait_command_matches, is_event_conflict, validate_run_id,
24};
25
26/// Builder for a [`FlowEngine`].
27pub struct FlowEngineBuilder {
28    store: Arc<dyn FlowEventStore>,
29    runtime: Arc<dyn FlowRuntime>,
30    observer: Arc<dyn FlowEventObserver>,
31    max_replay_iterations: usize,
32}
33
34impl FlowEngineBuilder {
35    pub fn new(runtime: Arc<dyn FlowRuntime>) -> Self {
36        Self {
37            store: Arc::new(InMemoryEventStore::new()),
38            runtime,
39            observer: Arc::new(NoopFlowEventObserver),
40            max_replay_iterations: 1024,
41        }
42    }
43
44    pub fn with_store(mut self, store: Arc<dyn FlowEventStore>) -> Self {
45        self.store = store;
46        self
47    }
48
49    pub fn with_observer(mut self, observer: Arc<dyn FlowEventObserver>) -> Self {
50        self.observer = observer;
51        self
52    }
53
54    pub fn with_max_replay_iterations(mut self, max_replay_iterations: usize) -> Self {
55        self.max_replay_iterations = max_replay_iterations.max(1);
56        self
57    }
58
59    pub fn build(self) -> FlowEngine {
60        FlowEngine {
61            store: self.store,
62            runtime: self.runtime,
63            observer: self.observer,
64            max_replay_iterations: self.max_replay_iterations,
65        }
66    }
67}
68
69/// Event-sourced workflow engine.
70#[derive(Clone)]
71pub struct FlowEngine {
72    store: Arc<dyn FlowEventStore>,
73    runtime: Arc<dyn FlowRuntime>,
74    observer: Arc<dyn FlowEventObserver>,
75    max_replay_iterations: usize,
76}
77
78impl FlowEngine {
79    pub fn builder(runtime: Arc<dyn FlowRuntime>) -> FlowEngineBuilder {
80        FlowEngineBuilder::new(runtime)
81    }
82
83    pub fn new(store: Arc<dyn FlowEventStore>, runtime: Arc<dyn FlowRuntime>) -> Self {
84        Self {
85            store,
86            runtime,
87            observer: Arc::new(NoopFlowEventObserver),
88            max_replay_iterations: 1024,
89        }
90    }
91
92    pub fn in_memory(runtime: Arc<dyn FlowRuntime>) -> Self {
93        Self::new(Arc::new(InMemoryEventStore::new()), runtime)
94    }
95
96    pub fn store(&self) -> Arc<dyn FlowEventStore> {
97        Arc::clone(&self.store)
98    }
99
100    pub fn observer(&self) -> Arc<dyn FlowEventObserver> {
101        Arc::clone(&self.observer)
102    }
103
104    /// Start a workflow run and drive it until completion or suspension.
105    pub async fn start(&self, spec: WorkflowSpec, input: serde_json::Value) -> Result<String> {
106        let run_id = Uuid::new_v4().to_string();
107        self.start_with_id(run_id, spec, input).await
108    }
109
110    /// Start a workflow run using a caller-provided durable run id.
111    ///
112    /// Reusing the same `run_id` with the same workflow spec and input is
113    /// idempotent. Reusing it with different spec or input returns a conflict.
114    pub async fn start_with_id(
115        &self,
116        run_id: impl Into<String>,
117        spec: WorkflowSpec,
118        input: serde_json::Value,
119    ) -> Result<String> {
120        spec.validate()?;
121        let run_id = run_id.into();
122        validate_run_id(&run_id)?;
123
124        for _ in 0..self.max_replay_iterations {
125            match self.store.list(&run_id).await {
126                Ok(history) => {
127                    let snapshot = project_run(&run_id, &history)?;
128                    ensure_same_start(&run_id, &snapshot, &spec, &input)?;
129                    if snapshot.status == WorkflowRunStatus::Pending {
130                        match self
131                            .record_event_at(&run_id, snapshot.last_sequence, FlowEvent::RunStarted)
132                            .await
133                        {
134                            Ok(_) => {}
135                            Err(err) if is_event_conflict(&err) => continue,
136                            Err(err) => return Err(err),
137                        }
138                    }
139                    match self.drive(&run_id).await {
140                        Ok(_) => return Ok(run_id),
141                        Err(err) if is_event_conflict(&err) => continue,
142                        Err(err) => return Err(err),
143                    }
144                }
145                Err(FlowError::RunNotFound(_)) => {
146                    let created = match self
147                        .record_event_at(
148                            &run_id,
149                            0,
150                            FlowEvent::RunCreated {
151                                spec: spec.clone(),
152                                input: input.clone(),
153                            },
154                        )
155                        .await
156                    {
157                        Ok(created) => created,
158                        Err(err) if is_event_conflict(&err) => continue,
159                        Err(err) => return Err(err),
160                    };
161                    match self
162                        .record_event_at(&run_id, created.sequence, FlowEvent::RunStarted)
163                        .await
164                    {
165                        Ok(_) => {}
166                        Err(err) if is_event_conflict(&err) => continue,
167                        Err(err) => return Err(err),
168                    }
169                    match self.drive(&run_id).await {
170                        Ok(_) => return Ok(run_id),
171                        Err(err) if is_event_conflict(&err) => continue,
172                        Err(err) => return Err(err),
173                    }
174                }
175                Err(err) => return Err(err),
176            }
177        }
178
179        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
180    }
181
182    /// Resume a wait once its timer has fired.
183    pub async fn resume_wait(&self, run_id: &str, wait_id: &str) -> Result<()> {
184        for _ in 0..self.max_replay_iterations {
185            let snapshot = self.snapshot(run_id).await?;
186            if snapshot.status.is_terminal() {
187                return Err(FlowError::RunTerminal(run_id.to_string()));
188            }
189            match snapshot.waits.get(wait_id) {
190                Some(wait) if wait.status == WaitStatus::Waiting => {
191                    match self
192                        .record_event_at(
193                            run_id,
194                            snapshot.last_sequence,
195                            FlowEvent::WaitCompleted {
196                                wait_id: wait_id.to_string(),
197                            },
198                        )
199                        .await
200                    {
201                        Ok(_) => {}
202                        Err(err) if is_event_conflict(&err) => continue,
203                        Err(err) => return Err(err),
204                    }
205                    match self.drive(run_id).await {
206                        Ok(_) => return Ok(()),
207                        Err(err) if is_event_conflict(&err) => continue,
208                        Err(err) => return Err(err),
209                    }
210                }
211                Some(_) => match self.drive(run_id).await {
212                    Ok(_) => return Ok(()),
213                    Err(err) if is_event_conflict(&err) => continue,
214                    Err(err) => return Err(err),
215                },
216                None => {
217                    return Err(FlowError::InvalidTransition(format!(
218                        "wait {wait_id} does not exist for run {run_id}"
219                    )))
220                }
221            }
222        }
223
224        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
225    }
226
227    /// List active waits whose `resume_at` is at or before `now`.
228    ///
229    /// Scheduler integrations can use this to inspect due timers before
230    /// deciding how aggressively to drive them.
231    pub async fn list_due_waits(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
232        let mut due = self
233            .list_due_wakeups(now)
234            .await?
235            .into_iter()
236            .filter(|wakeup| wakeup.kind == ScheduledWakeupKind::Wait)
237            .map(|wakeup| (wakeup.run_id, wakeup.subject_id))
238            .collect::<Vec<_>>();
239        due.sort();
240        Ok(due)
241    }
242
243    /// Complete every due wait and drive the affected workflows.
244    ///
245    /// Returns the `(run_id, wait_id)` pairs that were resumed. A wait already
246    /// completed by another caller is skipped by [`Self::resume_wait`].
247    pub async fn resume_due_waits(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
248        let due = self.list_due_waits(now).await?;
249        let mut resumed = Vec::with_capacity(due.len());
250        for (run_id, wait_id) in due {
251            self.resume_wait(&run_id, &wait_id).await?;
252            resumed.push((run_id, wait_id));
253        }
254        Ok(resumed)
255    }
256
257    /// List pending step retries whose `retry_after` is at or before `now`.
258    pub async fn list_due_retries(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
259        let mut due = self
260            .list_due_wakeups(now)
261            .await?
262            .into_iter()
263            .filter(|wakeup| wakeup.kind == ScheduledWakeupKind::Retry)
264            .map(|wakeup| (wakeup.run_id, wakeup.subject_id))
265            .collect::<Vec<_>>();
266        due.sort();
267        Ok(due)
268    }
269
270    /// List all due wait timers and delayed retries through the store boundary.
271    pub async fn list_due_wakeups(&self, now: DateTime<Utc>) -> Result<Vec<ScheduledWakeup>> {
272        let mut wakeups = self.store.list_due_wakeups(now).await?;
273        wakeups.sort_by(|left, right| {
274            (left.kind, left.run_id.as_str(), left.subject_id.as_str()).cmp(&(
275                right.kind,
276                right.run_id.as_str(),
277                right.subject_id.as_str(),
278            ))
279        });
280        Ok(wakeups)
281    }
282
283    /// Drive every run with a due step retry.
284    pub async fn resume_due_retries(&self, now: DateTime<Utc>) -> Result<Vec<(String, String)>> {
285        let due = self.list_due_retries(now).await?;
286        let mut run_ids = Vec::new();
287        for (run_id, _) in &due {
288            if !run_ids.contains(run_id) {
289                run_ids.push(run_id.clone());
290            }
291        }
292        for run_id in run_ids {
293            self.drive_at(&run_id, now).await?;
294        }
295        Ok(due)
296    }
297
298    /// Resume the due waits and delayed retries for one targeted run.
299    ///
300    /// Unlike the compatibility-wide `resume_due_*` methods, this path loads
301    /// only `run_id` and never performs another global due-wakeup query. The
302    /// returned records describe the wakeups that were still due when the task
303    /// began handling.
304    pub async fn resume_scheduled_run(
305        &self,
306        run_id: &str,
307        now: DateTime<Utc>,
308    ) -> Result<Vec<ScheduledWakeup>> {
309        let history = self.store.list(run_id).await?;
310        let snapshot = project_run(run_id, &history)?;
311        let due = scheduled_wakeups_for_snapshot(&snapshot)
312            .into_iter()
313            .filter(|wakeup| wakeup.scheduled_at <= now)
314            .collect::<Vec<_>>();
315
316        let due_wait_ids = due
317            .iter()
318            .filter(|wakeup| wakeup.kind == ScheduledWakeupKind::Wait)
319            .map(|wakeup| wakeup.subject_id.clone())
320            .collect::<Vec<_>>();
321        let has_due_retries = due
322            .iter()
323            .any(|wakeup| wakeup.kind == ScheduledWakeupKind::Retry);
324
325        for wait_id in due_wait_ids {
326            self.resume_wait(run_id, &wait_id).await?;
327        }
328        if has_due_retries {
329            self.drive_at(run_id, now).await?;
330        }
331
332        Ok(due)
333    }
334
335    pub async fn snapshot(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
336        let history = self.store.list(run_id).await?;
337        project_run(run_id, &history)
338    }
339
340    pub async fn history(&self, run_id: &str) -> Result<Vec<FlowEventEnvelope>> {
341        self.store.list(run_id).await
342    }
343
344    pub async fn list_run_ids(&self) -> Result<Vec<String>> {
345        self.store.list_run_ids().await
346    }
347
348    pub async fn list_snapshots(&self) -> Result<Vec<WorkflowRunSnapshot>> {
349        let mut snapshots = Vec::new();
350        for run_id in self.store.list_run_ids().await? {
351            snapshots.push(self.snapshot(&run_id).await?);
352        }
353        Ok(snapshots)
354    }
355
356    /// Summarize run state across the active store.
357    ///
358    /// Suspension counters include only non-terminal runs, so a cancelled run
359    /// that still has an old wait or hook in history is not reported as
360    /// actionable work.
361    pub async fn run_summary(&self) -> Result<WorkflowRunSummary> {
362        let snapshots = self.list_snapshots().await?;
363        Ok(WorkflowRunSummary::from_snapshots(&snapshots))
364    }
365
366    /// List open waits, active hooks, and pending delayed retries.
367    ///
368    /// The `due` flag on wait and retry suspensions is computed against `now`.
369    /// Terminal runs are skipped so cancelled histories do not produce
370    /// actionable operator work.
371    pub async fn list_open_suspensions(
372        &self,
373        now: DateTime<Utc>,
374    ) -> Result<Vec<WorkflowRunSuspension>> {
375        let mut suspensions = Vec::new();
376        for run_id in self.store.list_run_ids().await? {
377            let snapshot = self.snapshot(&run_id).await?;
378            if snapshot.status.is_terminal() {
379                continue;
380            }
381            for wait in snapshot.waits.values() {
382                if wait.status == WaitStatus::Waiting {
383                    suspensions.push(WorkflowRunSuspension::Wait {
384                        run_id: run_id.clone(),
385                        wait: wait.clone(),
386                        due: wait.resume_at <= now,
387                    });
388                }
389            }
390            for hook in snapshot.hooks.values() {
391                if hook.status == HookStatus::Active {
392                    suspensions.push(WorkflowRunSuspension::Hook {
393                        run_id: run_id.clone(),
394                        hook: hook.clone(),
395                    });
396                }
397            }
398            for step in snapshot.steps.values() {
399                if step.status == StepStatus::Pending {
400                    if let Some(retry_after) = step.retry_after {
401                        suspensions.push(WorkflowRunSuspension::Retry {
402                            run_id: run_id.clone(),
403                            step: step.clone(),
404                            due: retry_after <= now,
405                        });
406                    }
407                }
408            }
409        }
410        suspensions.sort_by(|left, right| {
411            (left.run_id(), left.kind_order(), left.subject_id()).cmp(&(
412                right.run_id(),
413                right.kind_order(),
414                right.subject_id(),
415            ))
416        });
417        Ok(suspensions)
418    }
419
420    /// Return the earliest open wait or delayed retry across non-terminal runs.
421    ///
422    /// This is useful for hosts that want to sleep until the next scheduler tick
423    /// instead of polling at a fixed interval. Active hooks are intentionally
424    /// ignored because they do not have a scheduled wake-up time.
425    pub async fn next_wakeup(&self, now: DateTime<Utc>) -> Result<Option<WorkflowRunSuspension>> {
426        for _ in 0..2 {
427            let Some(wakeup) = self.store.next_scheduled_wakeup().await? else {
428                return Ok(None);
429            };
430            match self.snapshot(&wakeup.run_id).await {
431                Ok(snapshot) => {
432                    if let Some(suspension) = resolve_scheduled_wakeup(&snapshot, &wakeup, now) {
433                        return Ok(Some(suspension));
434                    }
435                }
436                Err(FlowError::RunNotFound(_)) => {}
437                Err(error) => return Err(error),
438            }
439        }
440
441        self.next_wakeup_by_replay(now).await
442    }
443
444    async fn next_wakeup_by_replay(
445        &self,
446        now: DateTime<Utc>,
447    ) -> Result<Option<WorkflowRunSuspension>> {
448        let mut wakeups = self.list_open_suspensions(now).await?;
449        wakeups.retain(|suspension| suspension.scheduled_at().is_some());
450        wakeups.sort_by(|left, right| {
451            (
452                left.scheduled_at(),
453                left.run_id(),
454                left.kind_order(),
455                left.subject_id(),
456            )
457                .cmp(&(
458                    right.scheduled_at(),
459                    right.run_id(),
460                    right.kind_order(),
461                    right.subject_id(),
462                ))
463        });
464        Ok(wakeups.into_iter().next())
465    }
466
467    /// List active external callback hooks across non-terminal runs.
468    ///
469    /// Callback routers and dashboards can use this to discover public hook
470    /// tokens and their audit metadata without projecting every run manually.
471    /// The result is sorted by run ID and hook ID for stable polling output.
472    pub async fn list_active_hooks(&self) -> Result<Vec<ActiveHookSnapshot>> {
473        self.store.list_active_hooks().await
474    }
475
476    /// Replay and dispatch until the run reaches a terminal state or an open
477    /// wait/hook suspension.
478    pub async fn drive(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
479        self.drive_at(run_id, Utc::now()).await
480    }
481
482    async fn drive_at(&self, run_id: &str, now: DateTime<Utc>) -> Result<WorkflowRunSnapshot> {
483        'replay: for _ in 0..self.max_replay_iterations {
484            let history = self.store.list(run_id).await?;
485            let snapshot = project_run(run_id, &history)?;
486            if let Some(event) = interrupted_retry_exhaustion_event(&snapshot, &history) {
487                match self
488                    .record_event_at(run_id, snapshot.last_sequence, event)
489                    .await
490                {
491                    Ok(_) => continue,
492                    Err(err) if is_event_conflict(&err) => continue,
493                    Err(err) => return Err(err),
494                }
495            }
496            if snapshot.status.is_terminal()
497                || snapshot
498                    .waits
499                    .values()
500                    .any(|wait| wait.status == WaitStatus::Waiting)
501                || snapshot
502                    .hooks
503                    .values()
504                    .any(|hook| hook.status == HookStatus::Active)
505                || (snapshot.has_future_retry(now) && snapshot.due_retries(now).is_empty())
506            {
507                return Ok(snapshot);
508            }
509
510            let command = self
511                .runtime
512                .run_workflow(WorkflowInvocation {
513                    run_id: run_id.to_string(),
514                    spec: snapshot.spec.clone(),
515                    input: snapshot.input.clone(),
516                    history,
517                })
518                .await?;
519
520            match command {
521                RuntimeCommand::Complete { output } => {
522                    if snapshot.status == WorkflowRunStatus::Cancelling {
523                        return Err(FlowError::InvalidTransition(format!(
524                            "workflow run {run_id} completed after cancellation was requested; cleanup-aware cancellation must return cancel or fail"
525                        )));
526                    }
527                    match self
528                        .record_event_at(
529                            run_id,
530                            snapshot.last_sequence,
531                            FlowEvent::RunCompleted { output },
532                        )
533                        .await
534                    {
535                        Ok(_) => {}
536                        Err(err) if is_event_conflict(&err) => continue,
537                        Err(err) => return Err(err),
538                    }
539                    return self.snapshot(run_id).await;
540                }
541                RuntimeCommand::Fail { error } => {
542                    match self
543                        .record_event_at(
544                            run_id,
545                            snapshot.last_sequence,
546                            FlowEvent::RunFailed { error },
547                        )
548                        .await
549                    {
550                        Ok(_) => {}
551                        Err(err) if is_event_conflict(&err) => continue,
552                        Err(err) => return Err(err),
553                    }
554                    return self.snapshot(run_id).await;
555                }
556                RuntimeCommand::Cancel => {
557                    let cancellation = snapshot.cancellation.as_ref().ok_or_else(|| {
558                        FlowError::InvalidTransition(format!(
559                            "workflow run {run_id} returned cancel without a durable cancellation request"
560                        ))
561                    })?;
562                    match self
563                        .record_event_at(
564                            run_id,
565                            snapshot.last_sequence,
566                            FlowEvent::RunCancelled {
567                                reason: cancellation.request.reason.clone(),
568                            },
569                        )
570                        .await
571                    {
572                        Ok(_) => {}
573                        Err(err) if is_event_conflict(&err) => continue,
574                        Err(err) => return Err(err),
575                    }
576                    return self.snapshot(run_id).await;
577                }
578                RuntimeCommand::Timeout { deadline, reason } => {
579                    match self
580                        .record_event_at(
581                            run_id,
582                            snapshot.last_sequence,
583                            FlowEvent::RunTimedOut { deadline, reason },
584                        )
585                        .await
586                    {
587                        Ok(_) => {}
588                        Err(err) if is_event_conflict(&err) => continue,
589                        Err(err) => return Err(err),
590                    }
591                    return self.snapshot(run_id).await;
592                }
593                RuntimeCommand::RecordProgress { progress } => {
594                    progress.validate()?;
595                    if let Some(existing) = snapshot.progress(&progress.progress_id) {
596                        ensure_progress_matches(run_id, existing, &progress)?;
597                        return Err(FlowError::InvalidTransition(format!(
598                            "workflow rescheduled progress {} without progress",
599                            progress.progress_id
600                        )));
601                    }
602                    match self
603                        .record_event_at(
604                            run_id,
605                            snapshot.last_sequence,
606                            FlowEvent::RunProgressRecorded { progress },
607                        )
608                        .await
609                    {
610                        Ok(_) => {}
611                        Err(err) if is_event_conflict(&err) => continue,
612                        Err(err) => return Err(err),
613                    }
614                }
615                RuntimeCommand::LinkChildOperation { child } => {
616                    child.validate()?;
617                    if let Some(existing) = snapshot.child_operation(&child.reference_id) {
618                        ensure_child_operation_matches(run_id, existing, &child)?;
619                        return Err(FlowError::InvalidTransition(format!(
620                            "workflow rescheduled child operation {} without progress",
621                            child.reference_id
622                        )));
623                    }
624                    match self
625                        .record_event_at(
626                            run_id,
627                            snapshot.last_sequence,
628                            FlowEvent::ChildOperationLinked { child },
629                        )
630                        .await
631                    {
632                        Ok(_) => {}
633                        Err(err) if is_event_conflict(&err) => continue,
634                        Err(err) => return Err(err),
635                    }
636                }
637                RuntimeCommand::ScheduleStep {
638                    step_id,
639                    step_name,
640                    input,
641                    retry,
642                } => {
643                    if let Some(step) = snapshot.steps.get(&step_id) {
644                        ensure_step_command_matches(run_id, step, &step_name, &input, retry)?;
645                        if matches!(
646                            step.status,
647                            StepStatus::Completed | StepStatus::Failed | StepStatus::Cancelled
648                        ) {
649                            return Err(FlowError::InvalidTransition(format!(
650                                "workflow rescheduled terminal step {step_id} without progress"
651                            )));
652                        }
653                    }
654                    ensure_retry_policy_valid(retry)?;
655                    match self
656                        .execute_step(
657                            run_id,
658                            &snapshot,
659                            StepExecutionContext {
660                                step_id,
661                                step_name,
662                                input,
663                                retry,
664                                now,
665                            },
666                        )
667                        .await
668                    {
669                        Ok(()) => {}
670                        Err(err) if is_event_conflict(&err) => continue,
671                        Err(err) => return Err(err),
672                    }
673                }
674                RuntimeCommand::ScheduleSteps { steps } => {
675                    ensure_step_batch_valid(&steps)?;
676                    for step in &steps {
677                        if let Some(existing) = snapshot.steps.get(&step.step_id) {
678                            ensure_step_command_matches(
679                                run_id,
680                                existing,
681                                &step.step_name,
682                                &step.input,
683                                step.retry,
684                            )?;
685                        }
686                    }
687                    if steps.iter().all(|step| {
688                        snapshot.steps.get(&step.step_id).is_some_and(|existing| {
689                            matches!(
690                                existing.status,
691                                StepStatus::Completed | StepStatus::Failed | StepStatus::Cancelled
692                            )
693                        })
694                    }) {
695                        let step_ids = steps
696                            .iter()
697                            .map(|step| step.step_id.as_str())
698                            .collect::<Vec<_>>()
699                            .join(", ");
700                        return Err(FlowError::InvalidTransition(format!(
701                            "workflow rescheduled only terminal steps without progress: {step_ids}"
702                        )));
703                    }
704                    for step in &steps {
705                        ensure_retry_policy_valid(step.retry)?;
706                    }
707                    match self.execute_step_batch(run_id, &snapshot, steps, now).await {
708                        Ok(()) => {}
709                        Err(err) if is_event_conflict(&err) => continue 'replay,
710                        Err(err) => return Err(err),
711                    }
712                }
713                RuntimeCommand::WaitUntil { wait_id, resume_at } => {
714                    match snapshot.waits.get(&wait_id) {
715                        Some(wait) => {
716                            ensure_wait_command_matches(run_id, wait, resume_at)?;
717                            match wait.status {
718                                WaitStatus::Completed => continue,
719                                WaitStatus::Waiting => return self.snapshot(run_id).await,
720                                WaitStatus::Cancelled => {
721                                    return Err(FlowError::InvalidTransition(format!(
722                                        "workflow rescheduled cancelled wait {wait_id}; cancellation cleanup must use a distinct stable identity"
723                                    )))
724                                }
725                            }
726                        }
727                        None => {
728                            match self
729                                .record_event_at(
730                                    run_id,
731                                    snapshot.last_sequence,
732                                    FlowEvent::WaitCreated { wait_id, resume_at },
733                                )
734                                .await
735                            {
736                                Ok(_) => {}
737                                Err(err) if is_event_conflict(&err) => continue,
738                                Err(err) => return Err(err),
739                            }
740                            return self.snapshot(run_id).await;
741                        }
742                    }
743                }
744                RuntimeCommand::CreateHook {
745                    hook_id,
746                    token,
747                    metadata,
748                } => match snapshot.hooks.get(&hook_id) {
749                    Some(hook) => {
750                        ensure_hook_command_matches(run_id, hook, &token, &metadata)?;
751                        match hook.status {
752                            HookStatus::Received | HookStatus::Disposed => continue,
753                            HookStatus::Active => return self.snapshot(run_id).await,
754                            HookStatus::Cancelled => {
755                                return Err(FlowError::InvalidTransition(format!(
756                                    "workflow rescheduled cancelled hook {hook_id}; cancellation cleanup must use a distinct stable identity"
757                                )))
758                            }
759                        }
760                    }
761                    None => {
762                        self.ensure_hook_token_available(run_id, &hook_id, &token)
763                            .await?;
764                        match self
765                            .record_event_at(
766                                run_id,
767                                snapshot.last_sequence,
768                                FlowEvent::HookCreated {
769                                    hook_id,
770                                    token,
771                                    metadata,
772                                },
773                            )
774                            .await
775                        {
776                            Ok(_) => {}
777                            Err(err) if is_event_conflict(&err) => continue,
778                            Err(err) => return Err(err),
779                        }
780                        return self.snapshot(run_id).await;
781                    }
782                },
783            }
784        }
785
786        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
787    }
788
789    async fn terminate_run(&self, run_id: &str, event: FlowEvent) -> Result<()> {
790        for _ in 0..self.max_replay_iterations {
791            let snapshot = self.snapshot(run_id).await?;
792            if snapshot.status.is_terminal() {
793                return Ok(());
794            }
795            match self
796                .record_event_at(run_id, snapshot.last_sequence, event.clone())
797                .await
798            {
799                Ok(_) => return Ok(()),
800                Err(err) if is_event_conflict(&err) => continue,
801                Err(err) => return Err(err),
802            }
803        }
804        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
805    }
806
807    async fn record_event_at(
808        &self,
809        run_id: &str,
810        expected_sequence: u64,
811        event: FlowEvent,
812    ) -> Result<FlowEventEnvelope> {
813        let envelope = self
814            .store
815            .append_if_sequence(run_id, expected_sequence, event)
816            .await?;
817        self.observer.observe(envelope.clone()).await;
818        Ok(envelope)
819    }
820
821    async fn ensure_hook_token_available(
822        &self,
823        run_id: &str,
824        hook_id: &str,
825        token: &str,
826    ) -> Result<()> {
827        for active in self.store.find_active_hooks_by_token(token).await? {
828            if active.run_id == run_id && active.hook.hook_id == hook_id {
829                continue;
830            }
831            return Err(FlowError::HookTokenConflict {
832                token: token.to_string(),
833                existing_run_id: active.run_id,
834                existing_hook_id: active.hook.hook_id,
835            });
836        }
837        Ok(())
838    }
839}
840
841fn resolve_scheduled_wakeup(
842    snapshot: &WorkflowRunSnapshot,
843    wakeup: &ScheduledWakeup,
844    now: DateTime<Utc>,
845) -> Option<WorkflowRunSuspension> {
846    if snapshot.run_id != wakeup.run_id || snapshot.status.is_terminal() {
847        return None;
848    }
849    match wakeup.kind {
850        ScheduledWakeupKind::Wait => {
851            let wait = snapshot.waits.get(&wakeup.subject_id)?;
852            if wait.status != WaitStatus::Waiting || wait.resume_at != wakeup.scheduled_at {
853                return None;
854            }
855            Some(WorkflowRunSuspension::Wait {
856                run_id: wakeup.run_id.clone(),
857                wait: wait.clone(),
858                due: wakeup.scheduled_at <= now,
859            })
860        }
861        ScheduledWakeupKind::Retry => {
862            let step = snapshot.steps.get(&wakeup.subject_id)?;
863            if step.status != StepStatus::Pending || step.retry_after != Some(wakeup.scheduled_at) {
864                return None;
865            }
866            Some(WorkflowRunSuspension::Retry {
867                run_id: wakeup.run_id.clone(),
868                step: step.clone(),
869                due: wakeup.scheduled_at <= now,
870            })
871        }
872    }
873}
a3s_flow/engine/mod.rs

a3s_flow/engine/
mod.rs
