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        let started = record.next(EffectStatus::Started)?;
209        self.store
210            .compare_and_swap(&started, Some(record.revision))?;
211        record_event(
212            run,
213            RunEventKind::Effect(EffectEvent::Started {
214                effect_id: started.request.effect_id,
215            }),
216        )?;
217
218        let context = EffectExecutionContext::for_run(run);
219        let cancellation = context.cancellation().clone();
220        let execution = handler.execute(&started.request, context);
221        let result = match select(Box::pin(cancellation.cancelled()), Box::pin(execution)).await {
222            Either::Left(_) => {
223                return Err(EffectExecutorError::new(
224                    EffectExecutorErrorKind::Cancelled,
225                    "effect execution was cancelled and remains ambiguous",
226                ));
227            }
228            Either::Right((result, _)) => result,
229        };
230
231        match result {
232            Ok(output) => self.complete(&started, output, run),
233            Err(error) if outcome_is_ambiguous(&error) => {
234                Err(EffectExecutorError::ambiguous_handler(error))
235            }
236            Err(error) => self.fail(&started, error, run),
237        }
238    }
239
240    fn complete(
241        &self,
242        started: &EffectRecord,
243        output: Value,
244        run: &RunContext,
245    ) -> Result<EffectOutcome, EffectExecutorError> {
246        let completed = started.next(EffectStatus::Completed {
247            output: output.clone(),
248        })?;
249        self.store
250            .compare_and_swap(&completed, Some(started.revision))?;
251        record_event(
252            run,
253            RunEventKind::Effect(EffectEvent::Completed {
254                effect_id: completed.request.effect_id,
255                output: match self.event_payload_policy {
256                    EffectEventPayloadPolicy::Redacted => {
257                        serde_json::json!({"runifold": {"content_recorded": false}})
258                    }
259                    EffectEventPayloadPolicy::Full => output.clone(),
260                },
261            }),
262        )?;
263        Ok(EffectOutcome {
264            effect_id: completed.request.effect_id,
265            output,
266            replayed: false,
267            revision: completed.revision,
268        })
269    }
270
271    fn fail(
272        &self,
273        started: &EffectRecord,
274        error: RunError,
275        run: &RunContext,
276    ) -> Result<EffectOutcome, EffectExecutorError> {
277        let failed = started.next(EffectStatus::Failed {
278            error: error.clone(),
279        })?;
280        self.store
281            .compare_and_swap(&failed, Some(started.revision))?;
282        record_event(
283            run,
284            RunEventKind::Effect(EffectEvent::Failed {
285                effect_id: failed.request.effect_id,
286                error: error.clone(),
287            }),
288        )?;
289        Err(EffectExecutorError::handler(error))
290    }
291}
292
293impl std::fmt::Debug for EffectExecutor {
294    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
295        formatter.write_str("EffectExecutor(..)")
296    }
297}
298
299fn preflight(request: &EffectRequest, run: &RunContext) -> Result<(), EffectExecutorError> {
300    if !run.capabilities().contains(request.capability_id) {
301        return Err(EffectExecutorError::new(
302            EffectExecutorErrorKind::CapabilityDenied,
303            "Run is not granted the effect capability",
304        ));
305    }
306    if run.cancellation().is_cancelled() {
307        return Err(EffectExecutorError::new(
308            EffectExecutorErrorKind::Cancelled,
309            "effect was cancelled before preparation",
310        ));
311    }
312    if run
313        .deadline()
314        .is_some_and(|deadline| deadline <= Instant::now())
315    {
316        return Err(EffectExecutorError::new(
317            EffectExecutorErrorKind::DeadlineExceeded,
318            "effect deadline elapsed before preparation",
319        ));
320    }
321    Ok(())
322}
323
324fn validate_same_effect(
325    existing: &EffectRequest,
326    requested: &EffectRequest,
327) -> Result<(), EffectExecutorError> {
328    let same = existing.kind == requested.kind
329        && existing.capability_id == requested.capability_id
330        && existing.input == requested.input
331        && existing.effect_class == requested.effect_class
332        && existing.idempotency_key == requested.idempotency_key;
333    if !same {
334        return Err(EffectExecutorError::new(
335            EffectExecutorErrorKind::IdempotencyConflict,
336            "effect identity or idempotency key was reused for different work",
337        ));
338    }
339    Ok(())
340}
341
342fn retry_safe(request: &EffectRequest) -> bool {
343    matches!(
344        request.effect_class,
345        EffectClass::Pure | EffectClass::ReadOnly
346    ) || matches!(request.effect_class, EffectClass::IdempotentWrite)
347        && request.idempotency_key.is_some()
348}
349
350fn outcome_is_ambiguous(error: &RunError) -> bool {
351    matches!(
352        error.retry_safety,
353        RetrySafety::RequiresIdempotency
354            | RetrySafety::UnsafeAfterVisibleOutput
355            | RetrySafety::UnsafeAfterSideEffect
356            | RetrySafety::Unknown
357    )
358}
359
360fn record_event(run: &RunContext, kind: RunEventKind) -> Result<(), EffectExecutorError> {
361    run.record(kind, None)?;
362    Ok(())
363}
364
365#[cfg(test)]
366mod tests {
367    use std::{
368        collections::BTreeMap,
369        sync::{
370            Arc,
371            atomic::{AtomicUsize, Ordering},
372        },
373    };
374
375    use runifold_core::{
376        Budget, BudgetTracker, CapabilityDescriptor, CapabilityId, CapabilityKind, CapabilitySet,
377        EffectClass, EffectEvent, EffectId, EffectKind, EffectRequest, InMemoryJournal,
378        InvocationId, RetrySafety, RiskLevel, RunContext, RunError, RunErrorKind, RunEventKind,
379    };
380    use serde_json::{Value, json};
381
382    use crate::{
383        EffectEventPayloadPolicy, EffectExecutionContext, EffectExecutor, EffectExecutorErrorKind,
384        EffectFuture, EffectHandler, EffectReconciler, EffectReconciliation, EffectRecord,
385        EffectRecoveryPolicy, EffectStatus, EffectStore, InMemoryEffectStore,
386    };
387
388    struct CountingHandler {
389        calls: AtomicUsize,
390    }
391
392    impl CountingHandler {
393        fn new() -> Self {
394            Self {
395                calls: AtomicUsize::new(0),
396            }
397        }
398    }
399
400    impl EffectHandler for CountingHandler {
401        fn execute(
402            &self,
403            request: &EffectRequest,
404            _context: EffectExecutionContext,
405        ) -> EffectFuture<'_, Result<Value, RunError>> {
406            self.calls.fetch_add(1, Ordering::SeqCst);
407            let input = request.input.clone();
408            Box::pin(async move { Ok(json!({"echo": input})) })
409        }
410    }
411
412    struct AmbiguousHandler;
413
414    impl EffectHandler for AmbiguousHandler {
415        fn execute(
416            &self,
417            _request: &EffectRequest,
418            _context: EffectExecutionContext,
419        ) -> EffectFuture<'_, Result<Value, RunError>> {
420            Box::pin(async {
421                Err(RunError {
422                    kind: RunErrorKind::Transport,
423                    message: "connection closed after request body was sent".into(),
424                    retry_safety: RetrySafety::UnsafeAfterSideEffect,
425                    metadata: BTreeMap::new(),
426                })
427            })
428        }
429    }
430
431    struct CompletedReconciler;
432
433    impl EffectReconciler for CompletedReconciler {
434        fn reconcile(
435            &self,
436            _request: &EffectRequest,
437            _context: EffectExecutionContext,
438        ) -> EffectFuture<'_, Result<EffectReconciliation, RunError>> {
439            Box::pin(async {
440                Ok(EffectReconciliation::Completed(
441                    json!({"remote": "committed"}),
442                ))
443            })
444        }
445    }
446
447    struct FixedReconciler(EffectReconciliation);
448
449    impl EffectReconciler for FixedReconciler {
450        fn reconcile(
451            &self,
452            _request: &EffectRequest,
453            _context: EffectExecutionContext,
454        ) -> EffectFuture<'_, Result<EffectReconciliation, RunError>> {
455            let outcome = self.0.clone();
456            Box::pin(async move { Ok(outcome) })
457        }
458    }
459
460    struct FailedReconciler;
461
462    impl EffectReconciler for FailedReconciler {
463        fn reconcile(
464            &self,
465            _request: &EffectRequest,
466            _context: EffectExecutionContext,
467        ) -> EffectFuture<'_, Result<EffectReconciliation, RunError>> {
468            Box::pin(async {
469                Err(RunError {
470                    kind: RunErrorKind::Transport,
471                    message: "remote lookup unavailable".into(),
472                    retry_safety: RetrySafety::Unknown,
473                    metadata: BTreeMap::new(),
474                })
475            })
476        }
477    }
478
479    #[test]
480    fn completed_idempotent_effect_is_replayed_without_handler_execution() {
481        let capability = capability(EffectClass::IdempotentWrite);
482        let journal = InMemoryJournal::new();
483        let run = run_with(&capability, Some(journal.clone()));
484        let store = Arc::new(InMemoryEffectStore::new());
485        let executor = EffectExecutor::new(store);
486        let handler = CountingHandler::new();
487        let first_request = request(&capability, Some("order-7"), json!({"value": 7}));
488
489        let first = futures_executor::block_on(executor.execute(
490            first_request.clone(),
491            &run,
492            &handler,
493            EffectRecoveryPolicy::RejectAmbiguous,
494        ))
495        .unwrap();
496        let mut duplicate = first_request;
497        duplicate.effect_id = EffectId::new();
498        duplicate.invocation_id = InvocationId::new();
499        let second = futures_executor::block_on(executor.execute(
500            duplicate,
501            &run,
502            &handler,
503            EffectRecoveryPolicy::RejectAmbiguous,
504        ))
505        .unwrap();
506
507        assert!(!first.replayed);
508        assert!(second.replayed);
509        assert_eq!(second.effect_id, first.effect_id);
510        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
511        let events = journal.events();
512        assert!(matches!(
513            events[0].kind,
514            RunEventKind::Effect(EffectEvent::Requested { .. })
515        ));
516        assert!(matches!(
517            events.last().unwrap().kind,
518            RunEventKind::Effect(EffectEvent::Completed { .. })
519        ));
520        assert!(matches!(
521            &events.last().unwrap().kind,
522            RunEventKind::Effect(EffectEvent::Completed { output, .. })
523                if output == &json!({"runifold": {"content_recorded": false}})
524        ));
525    }
526
527    #[test]
528    fn full_event_payload_capture_requires_explicit_opt_in() {
529        let capability = capability(EffectClass::Pure);
530        let journal = InMemoryJournal::new();
531        let run = run_with(&capability, Some(journal.clone()));
532        let executor = EffectExecutor::new(Arc::new(InMemoryEffectStore::new()))
533            .with_event_payload_policy(EffectEventPayloadPolicy::Full);
534        let handler = CountingHandler::new();
535
536        futures_executor::block_on(executor.execute(
537            request(&capability, None, json!("secret")),
538            &run,
539            &handler,
540            EffectRecoveryPolicy::RejectAmbiguous,
541        ))
542        .unwrap();
543
544        assert!(matches!(
545            &journal.events().last().unwrap().kind,
546            RunEventKind::Effect(EffectEvent::Completed { output, .. })
547                if output == &json!({"echo": "secret"})
548        ));
549    }
550
551    #[test]
552    fn idempotency_key_cannot_be_reused_for_different_input() {
553        let capability = capability(EffectClass::IdempotentWrite);
554        let run = run_with(&capability, None);
555        let executor = EffectExecutor::new(Arc::new(InMemoryEffectStore::new()));
556        let handler = CountingHandler::new();
557        let first = request(&capability, Some("same-key"), json!({"value": 1}));
558        futures_executor::block_on(executor.execute(
559            first,
560            &run,
561            &handler,
562            EffectRecoveryPolicy::RejectAmbiguous,
563        ))
564        .unwrap();
565        let second = request(&capability, Some("same-key"), json!({"value": 2}));
566
567        let error = futures_executor::block_on(executor.execute(
568            second,
569            &run,
570            &handler,
571            EffectRecoveryPolicy::RejectAmbiguous,
572        ))
573        .unwrap_err();
574
575        assert_eq!(error.kind, EffectExecutorErrorKind::IdempotencyConflict);
576        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
577    }
578
579    #[test]
580    fn ambiguous_non_idempotent_effect_is_never_retried() {
581        let capability = capability(EffectClass::NonIdempotentWrite);
582        let run = run_with(&capability, None);
583        let store = Arc::new(InMemoryEffectStore::new());
584        let request = request(&capability, None, json!({"charge": 10}));
585        let prepared = EffectRecord::prepared(request.clone());
586        store.compare_and_swap(&prepared, None).unwrap();
587        let started = prepared.next(EffectStatus::Started).unwrap();
588        store.compare_and_swap(&started, Some(0)).unwrap();
589        let executor = EffectExecutor::new(store);
590        let handler = CountingHandler::new();
591
592        let error = futures_executor::block_on(executor.execute(
593            request,
594            &run,
595            &handler,
596            EffectRecoveryPolicy::RetrySafe,
597        ))
598        .unwrap_err();
599
600        assert_eq!(error.kind, EffectExecutorErrorKind::Ambiguous);
601        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
602    }
603
604    #[test]
605    fn remote_reconciliation_closes_completed_but_unrecorded_crash_window() {
606        let capability = capability(EffectClass::NonIdempotentWrite);
607        let run = run_with(&capability, None);
608        let store = Arc::new(InMemoryEffectStore::new());
609        let request = request(
610            &capability,
611            Some("remote-operation-7"),
612            json!({"charge": 10}),
613        );
614        let prepared = EffectRecord::prepared(request.clone());
615        store.compare_and_swap(&prepared, None).unwrap();
616        let started = prepared.next(EffectStatus::Started).unwrap();
617        store.compare_and_swap(&started, Some(0)).unwrap();
618        let executor = EffectExecutor::new(store.clone());
619        let handler = CountingHandler::new();
620
621        let outcome = futures_executor::block_on(executor.execute_reconciled(
622            request,
623            &run,
624            &handler,
625            &CompletedReconciler,
626            EffectRecoveryPolicy::RejectAmbiguous,
627        ))
628        .unwrap();
629
630        assert!(outcome.replayed);
631        assert_eq!(outcome.output, json!({"remote": "committed"}));
632        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
633        assert!(matches!(
634            store.load(outcome.effect_id).unwrap().unwrap().status,
635            EffectStatus::Completed { .. }
636        ));
637    }
638
639    #[test]
640    fn uncertain_handler_failure_stays_started_until_remote_reconciliation() {
641        let capability = capability(EffectClass::IdempotentWrite);
642        let run = run_with(&capability, None);
643        let store = Arc::new(InMemoryEffectStore::new());
644        let executor = EffectExecutor::new(store.clone());
645        let request = request(
646            &capability,
647            Some("remote-operation-response-loss"),
648            json!({"charge": 10}),
649        );
650
651        let error = futures_executor::block_on(executor.execute(
652            request.clone(),
653            &run,
654            &AmbiguousHandler,
655            EffectRecoveryPolicy::RejectAmbiguous,
656        ))
657        .unwrap_err();
658        assert_eq!(error.kind, EffectExecutorErrorKind::Ambiguous);
659        assert!(matches!(
660            store.load(request.effect_id).unwrap().unwrap().status,
661            EffectStatus::Started
662        ));
663
664        let outcome = futures_executor::block_on(executor.execute_reconciled(
665            request,
666            &run,
667            &CountingHandler::new(),
668            &CompletedReconciler,
669            EffectRecoveryPolicy::RejectAmbiguous,
670        ))
671        .unwrap();
672        assert!(outcome.replayed);
673        assert_eq!(outcome.output, json!({"remote": "committed"}));
674    }
675
676    #[test]
677    fn reconciliation_reexecutes_only_after_remote_not_executed_proof() {
678        let capability = capability(EffectClass::NonIdempotentWrite);
679        let run = run_with(&capability, None);
680        let store = Arc::new(InMemoryEffectStore::new());
681        let request = request(
682            &capability,
683            Some("remote-operation-8"),
684            json!({"charge": 11}),
685        );
686        let prepared = EffectRecord::prepared(request.clone());
687        store.compare_and_swap(&prepared, None).unwrap();
688        let started = prepared.next(EffectStatus::Started).unwrap();
689        store.compare_and_swap(&started, Some(0)).unwrap();
690        let executor = EffectExecutor::new(store);
691        let handler = CountingHandler::new();
692
693        let outcome = futures_executor::block_on(executor.execute_reconciled(
694            request,
695            &run,
696            &handler,
697            &FixedReconciler(EffectReconciliation::NotExecuted),
698            EffectRecoveryPolicy::RejectAmbiguous,
699        ))
700        .unwrap();
701
702        assert!(!outcome.replayed);
703        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
704    }
705
706    #[test]
707    fn unresolved_or_failed_reconciliation_never_runs_the_handler() {
708        let capability = capability(EffectClass::NonIdempotentWrite);
709        let run = run_with(&capability, None);
710        let store = Arc::new(InMemoryEffectStore::new());
711        let request = request(
712            &capability,
713            Some("remote-operation-9"),
714            json!({"charge": 12}),
715        );
716        let prepared = EffectRecord::prepared(request.clone());
717        store.compare_and_swap(&prepared, None).unwrap();
718        let started = prepared.next(EffectStatus::Started).unwrap();
719        store.compare_and_swap(&started, Some(0)).unwrap();
720        let executor = EffectExecutor::new(store);
721        let handler = CountingHandler::new();
722
723        let ambiguous = futures_executor::block_on(executor.execute_reconciled(
724            request.clone(),
725            &run,
726            &handler,
727            &FixedReconciler(EffectReconciliation::Ambiguous),
728            EffectRecoveryPolicy::RejectAmbiguous,
729        ))
730        .unwrap_err();
731        let failed = futures_executor::block_on(executor.execute_reconciled(
732            request,
733            &run,
734            &handler,
735            &FailedReconciler,
736            EffectRecoveryPolicy::RejectAmbiguous,
737        ))
738        .unwrap_err();
739
740        assert_eq!(ambiguous.kind, EffectExecutorErrorKind::Ambiguous);
741        assert_eq!(failed.kind, EffectExecutorErrorKind::Reconciliation);
742        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
743    }
744
745    #[test]
746    fn started_idempotent_effect_can_be_explicitly_reconciled_by_retry() {
747        let capability = capability(EffectClass::IdempotentWrite);
748        let run = run_with(&capability, None);
749        let store = Arc::new(InMemoryEffectStore::new());
750        let request = request(&capability, Some("safe-key"), json!({"write": 1}));
751        let prepared = EffectRecord::prepared(request.clone());
752        store.compare_and_swap(&prepared, None).unwrap();
753        let started = prepared.next(EffectStatus::Started).unwrap();
754        store.compare_and_swap(&started, Some(0)).unwrap();
755        let executor = EffectExecutor::new(store);
756        let handler = CountingHandler::new();
757
758        let outcome = futures_executor::block_on(executor.execute(
759            request,
760            &run,
761            &handler,
762            EffectRecoveryPolicy::RetrySafe,
763        ))
764        .unwrap();
765
766        assert_eq!(outcome.revision, 3);
767        assert_eq!(handler.calls.load(Ordering::SeqCst), 1);
768    }
769
770    #[test]
771    fn missing_capability_rejects_before_persistence_or_execution() {
772        let capability = capability(EffectClass::Pure);
773        let run = RunContext::root(BudgetTracker::new(Budget::default()), CapabilitySet::new());
774        let store = Arc::new(InMemoryEffectStore::new());
775        let effect = request(&capability, None, json!({}));
776        let executor = EffectExecutor::new(store.clone());
777        let handler = CountingHandler::new();
778
779        let error = futures_executor::block_on(executor.execute(
780            effect.clone(),
781            &run,
782            &handler,
783            EffectRecoveryPolicy::RejectAmbiguous,
784        ))
785        .unwrap_err();
786
787        assert_eq!(error.kind, EffectExecutorErrorKind::CapabilityDenied);
788        assert!(store.load(effect.effect_id).unwrap().is_none());
789        assert_eq!(handler.calls.load(Ordering::SeqCst), 0);
790    }
791
792    fn capability(effect: EffectClass) -> CapabilityDescriptor {
793        CapabilityDescriptor {
794            id: CapabilityId::new(),
795            name: "test-effect".into(),
796            version: "1".into(),
797            kind: CapabilityKind::Resource,
798            input_schema: json!({}),
799            output_schema: json!({}),
800            effect,
801            risk: RiskLevel::Low,
802            metadata: BTreeMap::new(),
803        }
804    }
805
806    fn request(
807        capability: &CapabilityDescriptor,
808        key: Option<&str>,
809        input: Value,
810    ) -> EffectRequest {
811        EffectRequest {
812            effect_id: EffectId::new(),
813            invocation_id: InvocationId::new(),
814            kind: EffectKind::Extension("test".into()),
815            capability_id: capability.id,
816            input,
817            effect_class: capability.effect,
818            idempotency_key: key.map(str::to_owned),
819        }
820    }
821
822    fn run_with(capability: &CapabilityDescriptor, journal: Option<InMemoryJournal>) -> RunContext {
823        let mut capabilities = CapabilitySet::new();
824        capabilities.grant(capability.clone());
825        let run = RunContext::root(BudgetTracker::new(Budget::default()), capabilities);
826        match journal {
827            Some(journal) => run.with_journal(Arc::new(journal)),
828            None => run,
829        }
830    }
831}
runifold_effect/executor.rs

runifold_effect/
executor.rs
