runifold_effect/
executor.rs

1use std::sync::Arc;
2
3use futures_util::future::{Either, select};
4use runifold_core::{
5    EffectClass, EffectEvent, EffectId, EffectRequest, Instant, RetrySafety, RunContext, RunError,
6    RunEventKind,
7};
8use serde_json::Value;
9
10use crate::{
11    EffectExecutionContext, EffectExecutorError, EffectExecutorErrorKind, EffectHandler,
12    EffectReconciler, EffectReconciliation, EffectRecord, EffectStatus, EffectStore,
13};
14
15/// Recovery behavior for a record whose handler may have executed.
16#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
17#[non_exhaustive]
18pub enum EffectRecoveryPolicy {
19    /// Never retry an ambiguous started effect automatically.
20    #[default]
21    RejectAmbiguous,
22    /// Retry only effects whose class and idempotency contract make it safe.
23    RetrySafe,
24}
25
26/// Controls whether effect outputs are copied into Journal events.
27#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
28#[non_exhaustive]
29pub enum EffectEventPayloadPolicy {
30    /// Keep durable output only in the `EffectStore`.
31    #[default]
32    Redacted,
33    /// Copy the complete output into `EffectEvent::Completed`.
34    Full,
35}
36
37/// Successful coordinated effect result.
38#[derive(Clone, Debug, PartialEq)]
39pub struct EffectOutcome {
40    /// Canonical logical effect identity.
41    pub effect_id: EffectId,
42    /// Canonical output.
43    pub output: Value,
44    /// Whether the output came from a completed durable record.
45    pub replayed: bool,
46    /// Durable record revision containing the output.
47    pub revision: u64,
48}
49
50/// Capability-gated write-ahead external-effect coordinator.
51#[derive(Clone)]
52pub struct EffectExecutor {
53    store: Arc<dyn EffectStore>,
54    event_payload_policy: EffectEventPayloadPolicy,
55}
56
57impl EffectExecutor {
58    /// Creates an executor using the given durable-state boundary.
59    pub fn new(store: Arc<dyn EffectStore>) -> Self {
60        Self {
61            store,
62            event_payload_policy: EffectEventPayloadPolicy::Redacted,
63        }
64    }
65
66    /// Sets explicit Journal output-capture behavior.
67    #[must_use]
68    pub const fn with_event_payload_policy(mut self, policy: EffectEventPayloadPolicy) -> Self {
69        self.event_payload_policy = policy;
70        self
71    }
72
73    /// Executes or recovers one logical effect.
74    ///
75    /// # Errors
76    ///
77    /// Returns [`EffectExecutorError`] for authority, persistence, lifecycle,
78    /// ambiguity, handler, or observability failures.
79    pub async fn execute(
80        &self,
81        request: EffectRequest,
82        run: &RunContext,
83        handler: &dyn EffectHandler,
84        recovery: EffectRecoveryPolicy,
85    ) -> Result<EffectOutcome, EffectExecutorError> {
86        self.execute_inner(request, run, handler, recovery, None)
87            .await
88    }
89
90    /// Executes an effect and consults a remote-state reconciler before
91    /// deciding whether an ambiguously started record may be retried.
92    ///
93    /// This closes the common crash window where the remote side committed but
94    /// the local `Completed` record did not. It does not create a distributed
95    /// transaction: unresolved remote state still returns `Ambiguous`.
96    ///
97    /// # Errors
98    ///
99    /// Returns [`EffectExecutorError`] for authority, persistence, lifecycle,
100    /// reconciliation, ambiguity, handler, or observability failures.
101    pub async fn execute_reconciled(
102        &self,
103        request: EffectRequest,
104        run: &RunContext,
105        handler: &dyn EffectHandler,
106        reconciler: &dyn EffectReconciler,
107        recovery: EffectRecoveryPolicy,
108    ) -> Result<EffectOutcome, EffectExecutorError> {
109        self.execute_inner(request, run, handler, recovery, Some(reconciler))
110            .await
111    }
112
113    async fn execute_inner(
114        &self,
115        request: EffectRequest,
116        run: &RunContext,
117        handler: &dyn EffectHandler,
118        recovery: EffectRecoveryPolicy,
119        reconciler: Option<&dyn EffectReconciler>,
120    ) -> Result<EffectOutcome, EffectExecutorError> {
121        preflight(&request, run)?;
122        let (record, created) = self.resolve(request)?;
123        if created {
124            record_event(
125                run,
126                RunEventKind::Effect(EffectEvent::Requested {
127                    effect_id: record.request.effect_id,
128                }),
129            )?;
130        }
131        match &record.status {
132            EffectStatus::Completed { output } => Ok(EffectOutcome {
133                effect_id: record.request.effect_id,
134                output: output.clone(),
135                replayed: true,
136                revision: record.revision,
137            }),
138            EffectStatus::Failed { error } => Err(EffectExecutorError::handler(error.clone())),
139            EffectStatus::Prepared => self.start(record, run, handler).await,
140            EffectStatus::Started => {
141                if let Some(reconciler) = reconciler {
142                    match self.reconcile(&record, run, reconciler).await? {
143                        EffectReconciliation::Completed(output) => {
144                            let mut outcome = self.complete(&record, output, run)?;
145                            outcome.replayed = true;
146                            return Ok(outcome);
147                        }
148                        EffectReconciliation::NotExecuted => {
149                            return self.start(record, run, handler).await;
150                        }
151                        EffectReconciliation::Ambiguous => {}
152                    }
153                }
154                if recovery != EffectRecoveryPolicy::RetrySafe || !retry_safe(&record.request) {
155                    return Err(EffectExecutorError::new(
156                        EffectExecutorErrorKind::Ambiguous,
157                        format!(
158                            "effect `{}` may already have executed",
159                            record.request.effect_id
160                        ),
161                    ));
162                }
163                self.start(record, run, handler).await
164            }
165        }
166    }
167
168    async fn reconcile(
169        &self,
170        record: &EffectRecord,
171        run: &RunContext,
172        reconciler: &dyn EffectReconciler,
173    ) -> Result<EffectReconciliation, EffectExecutorError> {
174        let context = EffectExecutionContext::for_run(run);
175        let cancellation = context.cancellation().clone();
176        let reconciliation = reconciler.reconcile(&record.request, context);
177        match select(Box::pin(cancellation.cancelled()), Box::pin(reconciliation)).await {
178            Either::Left(_) => Err(EffectExecutorError::new(
179                EffectExecutorErrorKind::Cancelled,
180                "effect reconciliation was cancelled and remains ambiguous",
181            )),
182            Either::Right((result, _)) => result.map_err(EffectExecutorError::reconciliation),
183        }
184    }
185
186    fn resolve(&self, request: EffectRequest) -> Result<(EffectRecord, bool), EffectExecutorError> {
187        if let Some(record) = self.store.load(request.effect_id)? {
188            validate_same_effect(&record.request, &request)?;
189            return Ok((record, false));
190        }
191        if let Some(key) = &request.idempotency_key
192            && let Some(record) = self.store.find_by_idempotency(request.capability_id, key)?
193        {
194            validate_same_effect(&record.request, &request)?;
195            return Ok((record, false));
196        }
197        let record = EffectRecord::prepared(request);
198        self.store.compare_and_swap(&record, None)?;
199        Ok((record, true))
200    }
201
202    async fn start(
203        &self,
204        record: EffectRecord,
205        run: &RunContext,
206        handler: &dyn EffectHandler,
207    ) -> Result<EffectOutcome, EffectExecutorError> {
208        // Reconciliation may have awaited external work since the entry check.
209        preflight(&record.request, run)?;
210        let started = record.next(EffectStatus::Started)?;
211        self.store
212            .compare_and_swap(&started, Some(record.revision))?;
213        record_event(
214            run,
215            RunEventKind::Effect(EffectEvent::Started {
216                effect_id: started.request.effect_id,
217            }),
218        )?;
219
220        preflight(&started.request, run)?;
221        let context = EffectExecutionContext::for_run(run);
222        let cancellation = context.cancellation().clone();
223        let execution = handler.execute(&started.request, context);
224        let result = match select(Box::pin(cancellation.cancelled()), Box::pin(execution)).await {
225            Either::Left(_) => {
226                return Err(EffectExecutorError::new(
227                    EffectExecutorErrorKind::Cancelled,
228                    "effect execution was cancelled and remains ambiguous",
229                ));
230            }
231            Either::Right((result, _)) => result,
232        };
233
234        match result {
235            Ok(output) => self.complete(&started, output, run),
236            Err(error) if outcome_is_ambiguous(&error) => {
237                Err(EffectExecutorError::ambiguous_handler(error))
238            }
239            Err(error) => self.fail(&started, error, run),
240        }
241    }
242
243    fn complete(
244        &self,
245        started: &EffectRecord,
246        output: Value,
247        run: &RunContext,
248    ) -> Result<EffectOutcome, EffectExecutorError> {
249        let completed = started.next(EffectStatus::Completed {
250            output: output.clone(),
251        })?;
252        self.store
253            .compare_and_swap(&completed, Some(started.revision))?;
254        record_event(
255            run,
256            RunEventKind::Effect(EffectEvent::Completed {
257                effect_id: completed.request.effect_id,
258                output: match self.event_payload_policy {
259                    EffectEventPayloadPolicy::Redacted => {
260                        serde_json::json!({"runifold": {"content_recorded": false}})
261                    }
262                    EffectEventPayloadPolicy::Full => output.clone(),
263                },
264            }),
265        )?;
266        Ok(EffectOutcome {
267            effect_id: completed.request.effect_id,
268            output,
269            replayed: false,
270            revision: completed.revision,
271        })
272    }
273
274    fn fail(
275        &self,
276        started: &EffectRecord,
277        error: RunError,
278        run: &RunContext,
279    ) -> Result<EffectOutcome, EffectExecutorError> {
280        let failed = started.next(EffectStatus::Failed {
281            error: error.clone(),
282        })?;
283        self.store
284            .compare_and_swap(&failed, Some(started.revision))?;
285        record_event(
286            run,
287            RunEventKind::Effect(EffectEvent::Failed {
288                effect_id: failed.request.effect_id,
289                error: error.clone(),
290            }),
291        )?;
292        Err(EffectExecutorError::handler(error))
293    }
294}
295
296impl std::fmt::Debug for EffectExecutor {
297    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
298        formatter.write_str("EffectExecutor(..)")
299    }
300}
301
302fn preflight(request: &EffectRequest, run: &RunContext) -> Result<(), EffectExecutorError> {
303    if !run.capabilities().contains(request.capability_id) {
304        return Err(EffectExecutorError::new(
305            EffectExecutorErrorKind::CapabilityDenied,
306            "Run is not granted the effect capability",
307        ));
308    }
309    if run.cancellation().is_cancelled() {
310        return Err(EffectExecutorError::new(
311            EffectExecutorErrorKind::Cancelled,
312            "effect was cancelled before preparation",
313        ));
314    }
315    if run
316        .deadline()
317        .is_some_and(|deadline| deadline <= Instant::now())
318    {
319        return Err(EffectExecutorError::new(
320            EffectExecutorErrorKind::DeadlineExceeded,
321            "effect deadline elapsed before preparation",
322        ));
323    }
324    Ok(())
325}
326
327fn validate_same_effect(
328    existing: &EffectRequest,
329    requested: &EffectRequest,
330) -> Result<(), EffectExecutorError> {
331    let same = existing.kind == requested.kind
332        && existing.capability_id == requested.capability_id
333        && existing.input == requested.input
334        && existing.effect_class == requested.effect_class
335        && existing.idempotency_key == requested.idempotency_key;
336    if !same {
337        return Err(EffectExecutorError::new(
338            EffectExecutorErrorKind::IdempotencyConflict,
339            "effect identity or idempotency key was reused for different work",
340        ));
341    }
342    Ok(())
343}
344
345fn retry_safe(request: &EffectRequest) -> bool {
346    matches!(
347        request.effect_class,
348        EffectClass::Pure | EffectClass::ReadOnly
349    ) || matches!(request.effect_class, EffectClass::IdempotentWrite)
350        && request.idempotency_key.is_some()
351}
352
353fn outcome_is_ambiguous(error: &RunError) -> bool {
354    matches!(
355        error.retry_safety,
356        RetrySafety::RequiresIdempotency
357            | RetrySafety::UnsafeAfterVisibleOutput
358            | RetrySafety::UnsafeAfterSideEffect
359            | RetrySafety::Unknown
360    )
361}
362
363fn record_event(run: &RunContext, kind: RunEventKind) -> Result<(), EffectExecutorError> {
364    run.record(kind, None)?;
365    Ok(())
366}
367
368#[cfg(test)]
369mod tests {
370    use std::{
371        collections::BTreeMap,
372        sync::{
373            Arc,
374            atomic::{AtomicUsize, Ordering},
375        },
376    };
377
378    use runifold_core::{
379        Budget, BudgetTracker, CapabilityDescriptor, CapabilityId, CapabilityKind, CapabilitySet,
380        EffectClass, EffectEvent, EffectId, EffectKind, EffectRequest, InMemoryJournal,
381        InvocationId, RetrySafety, RiskLevel, RunContext, RunError, RunErrorKind, RunEventKind,
382    };
383    use serde_json::{Value, json};
384
385    use crate::{
386        EffectEventPayloadPolicy, EffectExecutionContext, EffectExecutor, EffectExecutorErrorKind,
387        EffectFuture, EffectHandler, EffectReconciler, EffectReconciliation, EffectRecord,
388        EffectRecoveryPolicy, EffectStatus, EffectStore, InMemoryEffectStore,
389    };
390
391    struct CountingHandler {
392        calls: AtomicUsize,
393    }
394
395    impl CountingHandler {
396        fn new() -> Self {
397            Self {
398                calls: AtomicUsize::new(0),
399            }
400        }
401    }
402
403    impl EffectHandler for CountingHandler {
404        fn execute(
405            &self,
406            request: &EffectRequest,
407            _context: EffectExecutionContext,
408        ) -> EffectFuture<'_, Result<Value, RunError>> {
409            self.calls.fetch_add(1, Ordering::SeqCst);
410            let input = request.input.clone();
411            Box::pin(async move { Ok(json!({"echo": input})) })
412        }
413    }
414
415    struct AmbiguousHandler;
416
417    impl EffectHandler for AmbiguousHandler {
418        fn execute(
419            &self,
420            _request: &EffectRequest,
421            _context: EffectExecutionContext,
422        ) -> EffectFuture<'_, Result<Value, RunError>> {
423            Box::pin(async {
424                Err(RunError {
425                    kind: RunErrorKind::Transport,
426                    message: "connection closed after request body was sent".into(),
427                    retry_safety: RetrySafety::UnsafeAfterSideEffect,
428                    metadata: BTreeMap::new(),
429                })
430            })
431        }
432    }
433
434    struct CompletedReconciler;
435
436    impl EffectReconciler for CompletedReconciler {
437        fn reconcile(
438            &self,
439            _request: &EffectRequest,
440            _context: EffectExecutionContext,
441        ) -> EffectFuture<'_, Result<EffectReconciliation, RunError>> {
442            Box::pin(async {
443                Ok(EffectReconciliation::Completed(
444                    json!({"remote": "committed"}),
445                ))
446            })
447        }
448    }
449
450    struct FixedReconciler(EffectReconciliation);
451
452    impl EffectReconciler for FixedReconciler {
453        fn reconcile(
454            &self,
455            _request: &EffectRequest,
456            _context: EffectExecutionContext,
457        ) -> EffectFuture<'_, Result<EffectReconciliation, RunError>> {
458            let outcome = self.0.clone();
459            Box::pin(async move { Ok(outcome) })
460        }
461    }
462
463    struct FailedReconciler;
464
465    impl EffectReconciler for FailedReconciler {
466        fn reconcile(
467            &self,
468            _request: &EffectRequest,
469            _context: EffectExecutionContext,
470        ) -> EffectFuture<'_, Result<EffectReconciliation, RunError>> {
471            Box::pin(async {
472                Err(RunError {
473                    kind: RunErrorKind::Transport,
474                    message: "remote lookup unavailable".into(),
475                    retry_safety: RetrySafety::Unknown,
476                    metadata: BTreeMap::new(),
477                })
478            })
479        }
480    }
481
482    #[test]
483    fn completed_idempotent_effect_is_replayed_without_handler_execution() {
484        let capability = capability(EffectClass::IdempotentWrite);
485        let journal = InMemoryJournal::new();
486        let run = run_with(&capability, Some(journal.clone()));
487        let store = Arc::new(InMemoryEffectStore::new());
488        let executor = EffectExecutor::new(store);
489        let handler = CountingHandler::new();
490        let first_request = request(&capability, Some("order-7"), json!({"value": 7}));
491
492        let first = futures_executor::block_on(executor.execute(
493            first_request.clone(),
494            &run,
495            &handler,
496            EffectRecoveryPolicy::RejectAmbiguous,
497        ))
498        .unwrap();
499        let mut duplicate = first_request;
500        duplicate.effect_id = EffectId::new();
501        duplicate.invocation_id = InvocationId::new();
502        let second = futures_executor::block_on(executor.execute(
503            duplicate,
504            &run,
505            &handler,
506            EffectRecoveryPolicy::RejectAmbiguous,
507        ))
508        .unwrap();
509
510        assert!(!first.replayed);
511        assert!(second.replayed);
512        assert_eq!(second.effect_id, first.effect_id);
513        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
514        let events = journal.events();
515        assert!(matches!(
516            events[0].kind,
517            RunEventKind::Effect(EffectEvent::Requested { .. })
518        ));
519        assert!(matches!(
520            events.last().unwrap().kind,
521            RunEventKind::Effect(EffectEvent::Completed { .. })
522        ));
523        assert!(matches!(
524            &events.last().unwrap().kind,
525            RunEventKind::Effect(EffectEvent::Completed { output, .. })
526                if output == &json!({"runifold": {"content_recorded": false}})
527        ));
528    }
529
530    #[test]
531    fn full_event_payload_capture_requires_explicit_opt_in() {
532        let capability = capability(EffectClass::Pure);
533        let journal = InMemoryJournal::new();
534        let run = run_with(&capability, Some(journal.clone()));
535        let executor = EffectExecutor::new(Arc::new(InMemoryEffectStore::new()))
536            .with_event_payload_policy(EffectEventPayloadPolicy::Full);
537        let handler = CountingHandler::new();
538
539        futures_executor::block_on(executor.execute(
540            request(&capability, None, json!("secret")),
541            &run,
542            &handler,
543            EffectRecoveryPolicy::RejectAmbiguous,
544        ))
545        .unwrap();
546
547        assert!(matches!(
548            &journal.events().last().unwrap().kind,
549            RunEventKind::Effect(EffectEvent::Completed { output, .. })
550                if output == &json!({"echo": "secret"})
551        ));
552    }
553
554    #[test]
555    fn idempotency_key_cannot_be_reused_for_different_input() {
556        let capability = capability(EffectClass::IdempotentWrite);
557        let run = run_with(&capability, None);
558        let executor = EffectExecutor::new(Arc::new(InMemoryEffectStore::new()));
559        let handler = CountingHandler::new();
560        let first = request(&capability, Some("same-key"), json!({"value": 1}));
561        futures_executor::block_on(executor.execute(
562            first,
563            &run,
564            &handler,
565            EffectRecoveryPolicy::RejectAmbiguous,
566        ))
567        .unwrap();
568        let second = request(&capability, Some("same-key"), json!({"value": 2}));
569
570        let error = futures_executor::block_on(executor.execute(
571            second,
572            &run,
573            &handler,
574            EffectRecoveryPolicy::RejectAmbiguous,
575        ))
576        .unwrap_err();
577
578        assert_eq!(error.kind, EffectExecutorErrorKind::IdempotencyConflict);
579        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
580    }
581
582    #[test]
583    fn ambiguous_non_idempotent_effect_is_never_retried() {
584        let capability = capability(EffectClass::NonIdempotentWrite);
585        let run = run_with(&capability, None);
586        let store = Arc::new(InMemoryEffectStore::new());
587        let request = request(&capability, None, json!({"charge": 10}));
588        let prepared = EffectRecord::prepared(request.clone());
589        store.compare_and_swap(&prepared, None).unwrap();
590        let started = prepared.next(EffectStatus::Started).unwrap();
591        store.compare_and_swap(&started, Some(0)).unwrap();
592        let executor = EffectExecutor::new(store);
593        let handler = CountingHandler::new();
594
595        let error = futures_executor::block_on(executor.execute(
596            request,
597            &run,
598            &handler,
599            EffectRecoveryPolicy::RetrySafe,
600        ))
601        .unwrap_err();
602
603        assert_eq!(error.kind, EffectExecutorErrorKind::Ambiguous);
604        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
605    }
606
607    #[test]
608    fn remote_reconciliation_closes_completed_but_unrecorded_crash_window() {
609        let capability = capability(EffectClass::NonIdempotentWrite);
610        let run = run_with(&capability, None);
611        let store = Arc::new(InMemoryEffectStore::new());
612        let request = request(
613            &capability,
614            Some("remote-operation-7"),
615            json!({"charge": 10}),
616        );
617        let prepared = EffectRecord::prepared(request.clone());
618        store.compare_and_swap(&prepared, None).unwrap();
619        let started = prepared.next(EffectStatus::Started).unwrap();
620        store.compare_and_swap(&started, Some(0)).unwrap();
621        let executor = EffectExecutor::new(store.clone());
622        let handler = CountingHandler::new();
623
624        let outcome = futures_executor::block_on(executor.execute_reconciled(
625            request,
626            &run,
627            &handler,
628            &CompletedReconciler,
629            EffectRecoveryPolicy::RejectAmbiguous,
630        ))
631        .unwrap();
632
633        assert!(outcome.replayed);
634        assert_eq!(outcome.output, json!({"remote": "committed"}));
635        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
636        assert!(matches!(
637            store.load(outcome.effect_id).unwrap().unwrap().status,
638            EffectStatus::Completed { .. }
639        ));
640    }
641
642    #[test]
643    fn uncertain_handler_failure_stays_started_until_remote_reconciliation() {
644        let capability = capability(EffectClass::IdempotentWrite);
645        let run = run_with(&capability, None);
646        let store = Arc::new(InMemoryEffectStore::new());
647        let executor = EffectExecutor::new(store.clone());
648        let request = request(
649            &capability,
650            Some("remote-operation-response-loss"),
651            json!({"charge": 10}),
652        );
653
654        let error = futures_executor::block_on(executor.execute(
655            request.clone(),
656            &run,
657            &AmbiguousHandler,
658            EffectRecoveryPolicy::RejectAmbiguous,
659        ))
660        .unwrap_err();
661        assert_eq!(error.kind, EffectExecutorErrorKind::Ambiguous);
662        assert!(matches!(
663            store.load(request.effect_id).unwrap().unwrap().status,
664            EffectStatus::Started
665        ));
666
667        let outcome = futures_executor::block_on(executor.execute_reconciled(
668            request,
669            &run,
670            &CountingHandler::new(),
671            &CompletedReconciler,
672            EffectRecoveryPolicy::RejectAmbiguous,
673        ))
674        .unwrap();
675        assert!(outcome.replayed);
676        assert_eq!(outcome.output, json!({"remote": "committed"}));
677    }
678
679    #[test]
680    fn reconciliation_reexecutes_only_after_remote_not_executed_proof() {
681        let capability = capability(EffectClass::NonIdempotentWrite);
682        let run = run_with(&capability, None);
683        let store = Arc::new(InMemoryEffectStore::new());
684        let request = request(
685            &capability,
686            Some("remote-operation-8"),
687            json!({"charge": 11}),
688        );
689        let prepared = EffectRecord::prepared(request.clone());
690        store.compare_and_swap(&prepared, None).unwrap();
691        let started = prepared.next(EffectStatus::Started).unwrap();
692        store.compare_and_swap(&started, Some(0)).unwrap();
693        let executor = EffectExecutor::new(store);
694        let handler = CountingHandler::new();
695
696        let outcome = futures_executor::block_on(executor.execute_reconciled(
697            request,
698            &run,
699            &handler,
700            &FixedReconciler(EffectReconciliation::NotExecuted),
701            EffectRecoveryPolicy::RejectAmbiguous,
702        ))
703        .unwrap();
704
705        assert!(!outcome.replayed);
706        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
707    }
708
709    #[test]
710    fn cancellation_during_reconciliation_prevents_new_dispatch() {
711        struct CancelThenNotExecuted(runifold_core::CancellationToken);
712        impl EffectReconciler for CancelThenNotExecuted {
713            fn reconcile(
714                &self,
715                _: &EffectRequest,
716                _: EffectExecutionContext,
717            ) -> EffectFuture<'_, Result<EffectReconciliation, RunError>> {
718                Box::pin(async {
719                    self.0.cancel();
720                    Ok(EffectReconciliation::NotExecuted)
721                })
722            }
723        }
724        let capability = capability(EffectClass::NonIdempotentWrite);
725        let run = run_with(&capability, None);
726        let store = Arc::new(InMemoryEffectStore::new());
727        let request = request(
728            &capability,
729            Some("cancel-before-redispatch"),
730            json!({"publish": true}),
731        );
732        let prepared = EffectRecord::prepared(request.clone());
733        store.compare_and_swap(&prepared, None).unwrap();
734        store
735            .compare_and_swap(&prepared.next(EffectStatus::Started).unwrap(), Some(0))
736            .unwrap();
737        let handler = CountingHandler::new();
738        let error =
739            futures_executor::block_on(EffectExecutor::new(store.clone()).execute_reconciled(
740                request.clone(),
741                &run,
742                &handler,
743                &CancelThenNotExecuted(run.cancellation().clone()),
744                EffectRecoveryPolicy::RejectAmbiguous,
745            ))
746            .unwrap_err();
747        assert_eq!(error.kind, EffectExecutorErrorKind::Cancelled);
748        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
749        assert_eq!(store.load(request.effect_id).unwrap().unwrap().revision, 1);
750    }
751
752    #[test]
753    fn unresolved_or_failed_reconciliation_never_runs_the_handler() {
754        let capability = capability(EffectClass::NonIdempotentWrite);
755        let run = run_with(&capability, None);
756        let store = Arc::new(InMemoryEffectStore::new());
757        let request = request(
758            &capability,
759            Some("remote-operation-9"),
760            json!({"charge": 12}),
761        );
762        let prepared = EffectRecord::prepared(request.clone());
763        store.compare_and_swap(&prepared, None).unwrap();
764        let started = prepared.next(EffectStatus::Started).unwrap();
765        store.compare_and_swap(&started, Some(0)).unwrap();
766        let executor = EffectExecutor::new(store);
767        let handler = CountingHandler::new();
768
769        let ambiguous = futures_executor::block_on(executor.execute_reconciled(
770            request.clone(),
771            &run,
772            &handler,
773            &FixedReconciler(EffectReconciliation::Ambiguous),
774            EffectRecoveryPolicy::RejectAmbiguous,
775        ))
776        .unwrap_err();
777        let failed = futures_executor::block_on(executor.execute_reconciled(
778            request,
779            &run,
780            &handler,
781            &FailedReconciler,
782            EffectRecoveryPolicy::RejectAmbiguous,
783        ))
784        .unwrap_err();
785
786        assert_eq!(ambiguous.kind, EffectExecutorErrorKind::Ambiguous);
787        assert_eq!(failed.kind, EffectExecutorErrorKind::Reconciliation);
788        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
789    }
790
791    #[test]
792    fn started_idempotent_effect_can_be_explicitly_reconciled_by_retry() {
793        let capability = capability(EffectClass::IdempotentWrite);
794        let run = run_with(&capability, None);
795        let store = Arc::new(InMemoryEffectStore::new());
796        let request = request(&capability, Some("safe-key"), json!({"write": 1}));
797        let prepared = EffectRecord::prepared(request.clone());
798        store.compare_and_swap(&prepared, None).unwrap();
799        let started = prepared.next(EffectStatus::Started).unwrap();
800        store.compare_and_swap(&started, Some(0)).unwrap();
801        let executor = EffectExecutor::new(store);
802        let handler = CountingHandler::new();
803
804        let outcome = futures_executor::block_on(executor.execute(
805            request,
806            &run,
807            &handler,
808            EffectRecoveryPolicy::RetrySafe,
809        ))
810        .unwrap();
811
812        assert_eq!(outcome.revision, 3);
813        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
814    }
815
816    #[test]
817    fn missing_capability_rejects_before_persistence_or_execution() {
818        let capability = capability(EffectClass::Pure);
819        let run = RunContext::root(BudgetTracker::new(Budget::default()), CapabilitySet::new());
820        let store = Arc::new(InMemoryEffectStore::new());
821        let effect = request(&capability, None, json!({}));
822        let executor = EffectExecutor::new(store.clone());
823        let handler = CountingHandler::new();
824
825        let error = futures_executor::block_on(executor.execute(
826            effect.clone(),
827            &run,
828            &handler,
829            EffectRecoveryPolicy::RejectAmbiguous,
830        ))
831        .unwrap_err();
832
833        assert_eq!(error.kind, EffectExecutorErrorKind::CapabilityDenied);
834        assert!(store.load(effect.effect_id).unwrap().is_none());
835        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
836    }
837
838    fn capability(effect: EffectClass) -> CapabilityDescriptor {
839        CapabilityDescriptor {
840            id: CapabilityId::new(),
841            name: "test-effect".into(),
842            version: "1".into(),
843            kind: CapabilityKind::Resource,
844            input_schema: json!({}),
845            output_schema: json!({}),
846            effect,
847            risk: RiskLevel::Low,
848            metadata: BTreeMap::new(),
849        }
850    }
851
852    fn request(
853        capability: &CapabilityDescriptor,
854        key: Option<&str>,
855        input: Value,
856    ) -> EffectRequest {
857        EffectRequest {
858            effect_id: EffectId::new(),
859            invocation_id: InvocationId::new(),
860            kind: EffectKind::Extension("test".into()),
861            capability_id: capability.id,
862            input,
863            effect_class: capability.effect,
864            idempotency_key: key.map(str::to_owned),
865        }
866    }
867
868    fn run_with(capability: &CapabilityDescriptor, journal: Option<InMemoryJournal>) -> RunContext {
869        let mut capabilities = CapabilitySet::new();
870        capabilities.grant(capability.clone());
871        let run = RunContext::root(BudgetTracker::new(Budget::default()), capabilities);
872        match journal {
873            Some(journal) => run.with_journal(Arc::new(journal)),
874            None => run,
875        }
876    }
877}
runifold_effect/executor.rs

runifold_effect/
executor.rs
