qubit_cas/executor/
cas_executor.rs

1/*******************************************************************************
2 *
3 *    Copyright (c) 2025 - 2026 Haixing Hu.
4 *
5 *    SPDX-License-Identifier: Apache-2.0
6 *
7 *    Licensed under the Apache License, Version 2.0.
8 *
9 ******************************************************************************/
10//! CAS executor implementation.
11
12use std::marker::PhantomData;
13use std::panic::{
14    AssertUnwindSafe,
15    catch_unwind,
16};
17use std::sync::{
18    Arc,
19    Mutex,
20};
21use std::time::Duration;
22
23use qubit_atomic::AtomicRef;
24use qubit_error::BoxError;
25use qubit_function::{
26    Consumer,
27    Function,
28};
29use qubit_retry::{
30    AttemptFailure,
31    AttemptFailureDecision,
32    AttemptTimeoutPolicy,
33    AttemptTimeoutSource,
34    Retry,
35    RetryContext,
36    RetryError,
37    RetryOptions,
38};
39
40use crate::cas_decision::CasDecision;
41use crate::cas_outcome::CasOutcome;
42use crate::cas_success::CasSuccess;
43use crate::error::{
44    CasAttemptFailure,
45    CasError,
46    CasErrorKind,
47};
48use crate::event::{
49    CasContext,
50    CasEvent,
51    CasHooks,
52};
53use crate::observability::{
54    CasAlert,
55    CasObservabilityConfig,
56    CasObservabilityMode,
57    ListenerPanicPolicy,
58};
59use crate::report::{
60    CasExecutionOutcome,
61    CasExecutionReport,
62    CasReportBuilder,
63};
64use crate::strategy::CasStrategy;
65
66use super::cas_builder::CasBuilder;
67
68/// Executor for retry-aware compare-and-swap workflows.
69#[derive(Debug, Clone)]
70pub struct CasExecutor<T, E = BoxError> {
71    /// Immutable retry options shared by every execution.
72    options: RetryOptions,
73    /// Observability settings shared by executions.
74    observability: CasObservabilityConfig,
75    /// Marker preserving `T` and `E`.
76    marker: PhantomData<fn() -> (T, E)>,
77}
78
79/// Success payload produced by one successful attempt before context enrichment.
80enum AttemptSuccess<T, R> {
81    /// One compare-and-swap write succeeded.
82    Updated {
83        previous: Arc<T>,
84        current: Arc<T>,
85        output: R,
86    },
87    /// The operation completed successfully without writing.
88    Finished { current: Arc<T>, output: R },
89}
90
91/// Snapshot of retry-layer limits plus terminal outcome for finalizing [`CasReportBuilder`].
92struct CasReportFinishContext {
93    attempts_total: u32,
94    max_attempts: u32,
95    max_operation_elapsed: Option<Duration>,
96    max_total_elapsed: Option<Duration>,
97    outcome: CasExecutionOutcome,
98}
99
100impl CasReportFinishContext {
101    #[inline]
102    fn new(
103        attempts_total: u32,
104        max_attempts: u32,
105        max_operation_elapsed: Option<Duration>,
106        max_total_elapsed: Option<Duration>,
107        outcome: CasExecutionOutcome,
108    ) -> Self {
109        Self {
110            attempts_total,
111            max_attempts,
112            max_operation_elapsed,
113            max_total_elapsed,
114            outcome,
115        }
116    }
117}
118
119impl<T, E> CasExecutor<T, E> {
120    /// Creates a CAS builder.
121    ///
122    /// # Returns
123    /// A builder configured with default retry settings.
124    #[inline]
125    pub fn builder() -> CasBuilder<T, E> {
126        CasBuilder::new()
127    }
128
129    /// Creates an executor from retry options.
130    ///
131    /// # Parameters
132    /// - `options`: Retry options to validate and install.
133    ///
134    /// # Returns
135    /// A configured executor using the supplied retry options.
136    ///
137    /// # Errors
138    /// Returns the retry-layer validation error when `options` are invalid.
139    pub fn from_options(options: RetryOptions) -> Result<Self, qubit_retry::RetryConfigError> {
140        Self::builder().options(options).build()
141    }
142
143    /// Creates an executor tuned for low-latency workloads.
144    ///
145    /// # Returns
146    /// A configured executor. The built-in strategy is always valid.
147    pub fn latency_first() -> Self {
148        Self::builder()
149            .build_latency_first()
150            .expect("latency-first CAS strategy must be valid")
151    }
152
153    /// Creates an executor tuned for hot-contention workloads.
154    ///
155    /// # Returns
156    /// A configured executor. The built-in strategy is always valid.
157    pub fn contention_adaptive() -> Self {
158        Self::builder()
159            .build_contention_adaptive()
160            .expect("contention-adaptive CAS strategy must be valid")
161    }
162
163    /// Creates an executor tuned for reliability-first workloads.
164    ///
165    /// # Returns
166    /// A configured executor. The built-in strategy is always valid.
167    pub fn reliability_first() -> Self {
168        Self::builder()
169            .build_reliability_first()
170            .expect("reliability-first CAS strategy must be valid")
171    }
172
173    /// Creates an executor from a built-in strategy.
174    ///
175    /// # Parameters
176    /// - `strategy`: Strategy to install.
177    ///
178    /// # Returns
179    /// A configured executor. Built-in strategies are always valid.
180    pub fn with_strategy(strategy: CasStrategy) -> Self {
181        Self::builder()
182            .strategy(strategy)
183            .build()
184            .expect("built-in CAS strategy must be valid")
185    }
186
187    /// Creates one executor from validated parts.
188    ///
189    /// # Parameters
190    /// - `options`: Validated retry options.
191    /// - `observability`: Observability settings shared by executions.
192    ///
193    /// # Returns
194    /// A configured executor.
195    #[inline]
196    pub(crate) fn new(options: RetryOptions, observability: CasObservabilityConfig) -> Self {
197        Self {
198            options,
199            observability,
200            marker: PhantomData,
201        }
202    }
203
204    /// Returns the immutable retry options used by this executor.
205    ///
206    /// # Returns
207    /// Shared retry options.
208    #[inline]
209    pub fn options(&self) -> &RetryOptions {
210        &self.options
211    }
212
213    /// Returns observability settings used by this executor.
214    ///
215    /// # Returns
216    /// Shared observability configuration.
217    #[inline]
218    pub fn observability(&self) -> &CasObservabilityConfig {
219        &self.observability
220    }
221
222    /// Executes one synchronous CAS operation.
223    ///
224    /// # Parameters
225    /// - `state`: Shared atomic state container.
226    /// - `operation`: Pure operation that inspects the current state and returns
227    ///   a CAS decision.
228    ///
229    /// # Returns
230    /// A terminal result together with the execution report.
231    pub fn execute<R, O>(&self, state: &AtomicRef<T>, operation: O) -> CasOutcome<T, R, E>
232    where
233        T: 'static,
234        E: 'static,
235        O: Function<T, CasDecision<T, R, E>>,
236    {
237        self.execute_with_hooks(state, operation, CasHooks::new())
238    }
239
240    /// Executes one synchronous CAS operation with lifecycle hooks.
241    ///
242    /// # Parameters
243    /// - `state`: Shared atomic state container.
244    /// - `operation`: Pure operation that inspects the current state and returns
245    ///   a CAS decision.
246    /// - `hooks`: Per-execution hook registrations.
247    ///
248    /// # Returns
249    /// A terminal result together with the execution report.
250    pub fn execute_with_hooks<R, O>(
251        &self,
252        state: &AtomicRef<T>,
253        operation: O,
254        hooks: CasHooks,
255    ) -> CasOutcome<T, R, E>
256    where
257        T: 'static,
258        E: 'static,
259        O: Function<T, CasDecision<T, R, E>>,
260    {
261        let success_context = Arc::new(Mutex::new(None));
262        let attempt_snapshot = Arc::new(Mutex::new(None));
263        let report_builder = Arc::new(Mutex::new(CasReportBuilder::start()));
264        self.emit_started(&hooks, &report_builder);
265        let retry = self.build_retry(
266            &hooks,
267            Arc::clone(&success_context),
268            Arc::clone(&report_builder),
269        );
270        let attempt = retry.run(|| self.run_sync_attempt(state, &operation));
271        self.finish_execution(
272            attempt,
273            hooks,
274            success_context,
275            attempt_snapshot,
276            report_builder,
277        )
278    }
279
280    /// Executes one asynchronous CAS operation.
281    ///
282    /// # Parameters
283    /// - `state`: Shared atomic state container.
284    /// - `operation`: Async operation factory receiving one state snapshot.
285    ///
286    /// # Returns
287    /// A terminal result together with the execution report.
288    #[cfg(feature = "tokio")]
289    pub async fn execute_async<R, O, Fut>(
290        &self,
291        state: &AtomicRef<T>,
292        operation: O,
293    ) -> CasOutcome<T, R, E>
294    where
295        T: 'static,
296        E: 'static,
297        O: Fn(Arc<T>) -> Fut,
298        Fut: std::future::Future<Output = CasDecision<T, R, E>>,
299    {
300        self.execute_async_with_hooks(state, operation, CasHooks::new())
301            .await
302    }
303
304    /// Executes one asynchronous CAS operation with lifecycle hooks.
305    ///
306    /// # Parameters
307    /// - `state`: Shared atomic state container.
308    /// - `operation`: Async operation factory receiving one state snapshot.
309    /// - `hooks`: Per-execution hook registrations.
310    ///
311    /// # Returns
312    /// A terminal result together with the execution report.
313    #[cfg(feature = "tokio")]
314    pub async fn execute_async_with_hooks<R, O, Fut>(
315        &self,
316        state: &AtomicRef<T>,
317        operation: O,
318        hooks: CasHooks,
319    ) -> CasOutcome<T, R, E>
320    where
321        T: 'static,
322        E: 'static,
323        O: Fn(Arc<T>) -> Fut,
324        Fut: std::future::Future<Output = CasDecision<T, R, E>>,
325    {
326        let success_context = Arc::new(Mutex::new(None));
327        let attempt_snapshot = Arc::new(Mutex::new(None));
328        let report_builder = Arc::new(Mutex::new(CasReportBuilder::start()));
329        self.emit_started(&hooks, &report_builder);
330        let retry = self.build_retry(
331            &hooks,
332            Arc::clone(&success_context),
333            Arc::clone(&report_builder),
334        );
335        let attempt_snapshot_for_attempt = Arc::clone(&attempt_snapshot);
336        let attempt = retry
337            .run_async(|| {
338                self.run_async_attempt(state, &operation, Arc::clone(&attempt_snapshot_for_attempt))
339            })
340            .await;
341        self.finish_execution(
342            attempt,
343            hooks,
344            success_context,
345            attempt_snapshot,
346            report_builder,
347        )
348    }
349
350    /// Builds one retry policy for a single CAS execution.
351    ///
352    /// # Parameters
353    /// - `hooks`: Hook registrations for the current execution.
354    /// - `success_context`: Shared slot used to capture the retry success
355    ///   context.
356    /// # Returns
357    /// A retry policy configured for one CAS execution.
358    fn build_retry(
359        &self,
360        hooks: &CasHooks,
361        success_context: Arc<Mutex<Option<RetryContext>>>,
362        report_builder: Arc<Mutex<CasReportBuilder>>,
363    ) -> Retry<CasAttemptFailure<T, E>>
364    where
365        T: 'static,
366        E: 'static,
367    {
368        let event_hook = hooks.event_hook();
369        let retry_timeout_policy = self
370            .options
371            .attempt_timeout()
372            .map(|attempt_timeout| attempt_timeout.policy());
373        let observability = self.observability.clone();
374
375        let mut builder = Retry::<CasAttemptFailure<T, E>>::builder()
376            .options(self.options.clone())
377            .on_success(move |context: &RetryContext| {
378                *success_context
379                    .lock()
380                    .expect("CAS success context slot should be lockable") = Some(*context);
381            })
382            .on_failure(
383                move |failure: &AttemptFailure<CasAttemptFailure<T, E>>, context: &RetryContext| {
384                    let failure = match failure {
385                        AttemptFailure::Panic(_) | AttemptFailure::Executor(_) => {
386                            return AttemptFailureDecision::UseDefault;
387                        }
388                        AttemptFailure::Error(failure) => failure,
389                        AttemptFailure::Timeout => {
390                            let cas_context = CasContext::new(context);
391                            report_builder
392                                .lock()
393                                .expect("CAS report builder should be lockable")
394                                .record_timeout();
395                            if Self::should_emit_events(&observability, &event_hook) {
396                                let hook = event_hook
397                                    .as_ref()
398                                    .expect("event hook should exist when events are emitted");
399                                Self::dispatch_event(
400                                    &observability,
401                                    hook,
402                                    CasEvent::AttemptFailed {
403                                        context: cas_context,
404                                        kind: crate::error::CasAttemptFailureKind::Timeout,
405                                    },
406                                );
407                                if context.attempt_timeout_source()
408                                    == Some(AttemptTimeoutSource::Configured)
409                                    && retry_timeout_policy == Some(AttemptTimeoutPolicy::Retry)
410                                {
411                                    Self::dispatch_event(
412                                        &observability,
413                                        hook,
414                                        CasEvent::RetryRequested {
415                                            context: cas_context,
416                                        },
417                                    );
418                                }
419                            }
420                            return AttemptFailureDecision::UseDefault;
421                        }
422                    };
423                    let cas_context = CasContext::new(context);
424                    {
425                        let mut report = report_builder
426                            .lock()
427                            .expect("CAS report builder should be lockable");
428                        match failure {
429                            CasAttemptFailure::Conflict { .. } => report.record_conflict(),
430                            CasAttemptFailure::Retry { .. } => report.record_retry_error(),
431                            CasAttemptFailure::Abort { .. } => report.record_abort(),
432                            CasAttemptFailure::Timeout { .. } => report.record_timeout(),
433                        }
434                    }
435                    if Self::should_emit_events(&observability, &event_hook) {
436                        Self::dispatch_event(
437                            &observability,
438                            event_hook
439                                .as_ref()
440                                .expect("event hook should exist when events are emitted"),
441                            CasEvent::AttemptFailed {
442                                context: cas_context,
443                                kind: Self::failure_kind(failure),
444                            },
445                        );
446                    }
447                    match failure {
448                        CasAttemptFailure::Conflict { .. } | CasAttemptFailure::Retry { .. } => {
449                            if Self::should_emit_events(&observability, &event_hook) {
450                                Self::dispatch_event(
451                                    &observability,
452                                    event_hook
453                                        .as_ref()
454                                        .expect("event hook should exist when events are emitted"),
455                                    CasEvent::RetryRequested {
456                                        context: cas_context,
457                                    },
458                                );
459                            }
460                            AttemptFailureDecision::Retry
461                        }
462                        CasAttemptFailure::Abort { .. } => AttemptFailureDecision::Abort,
463                        CasAttemptFailure::Timeout { .. } => AttemptFailureDecision::UseDefault,
464                    }
465                },
466            );
467
468        if self.observability.listener_panic_policy() == ListenerPanicPolicy::Isolate {
469            builder = builder.isolate_listener_panics();
470        }
471        builder
472            .build()
473            .expect("validated CAS executor configuration must build retry policy")
474    }
475
476    /// Runs one synchronous CAS attempt.
477    ///
478    /// # Parameters
479    /// - `state`: Shared atomic state container.
480    /// - `operation`: Pure operation over the current state snapshot.
481    ///
482    /// # Returns
483    /// An attempt success or one attempt failure.
484    fn run_sync_attempt<R, O>(
485        &self,
486        state: &AtomicRef<T>,
487        operation: &O,
488    ) -> Result<AttemptSuccess<T, R>, CasAttemptFailure<T, E>>
489    where
490        O: Function<T, CasDecision<T, R, E>>,
491    {
492        let current = state.load();
493        match operation.apply(current.as_ref()) {
494            CasDecision::Update { next, output } => {
495                match state.compare_set(&current, Arc::clone(&next)) {
496                    Ok(()) => Ok(AttemptSuccess::Updated {
497                        previous: current,
498                        current: next,
499                        output,
500                    }),
501                    Err(actual) => Err(CasAttemptFailure::conflict(actual)),
502                }
503            }
504            CasDecision::Finish { output } => Ok(AttemptSuccess::Finished { current, output }),
505            CasDecision::Retry(error) => Err(CasAttemptFailure::retry(current, error)),
506            CasDecision::Abort(error) => Err(CasAttemptFailure::abort(current, error)),
507        }
508    }
509
510    /// Runs one asynchronous CAS attempt.
511    ///
512    /// # Parameters
513    /// - `state`: Shared atomic state container.
514    /// - `operation`: Async operation factory over one state snapshot.
515    ///
516    /// # Returns
517    /// An attempt success or one attempt failure.
518    #[cfg(feature = "tokio")]
519    async fn run_async_attempt<R, O, Fut>(
520        &self,
521        state: &AtomicRef<T>,
522        operation: &O,
523        attempt_snapshot: Arc<Mutex<Option<Arc<T>>>>,
524    ) -> Result<AttemptSuccess<T, R>, CasAttemptFailure<T, E>>
525    where
526        O: Fn(Arc<T>) -> Fut,
527        Fut: std::future::Future<Output = CasDecision<T, R, E>>,
528    {
529        let current = state.load();
530        *attempt_snapshot
531            .lock()
532            .expect("CAS attempt snapshot slot should be lockable") = Some(Arc::clone(&current));
533        let decision = operation(Arc::clone(&current)).await;
534
535        match decision {
536            CasDecision::Update { next, output } => {
537                match state.compare_set(&current, Arc::clone(&next)) {
538                    Ok(()) => Ok(AttemptSuccess::Updated {
539                        previous: current,
540                        current: next,
541                        output,
542                    }),
543                    Err(actual) => Err(CasAttemptFailure::conflict(actual)),
544                }
545            }
546            CasDecision::Finish { output } => Ok(AttemptSuccess::Finished { current, output }),
547            CasDecision::Retry(error) => Err(CasAttemptFailure::retry(current, error)),
548            CasDecision::Abort(error) => Err(CasAttemptFailure::abort(current, error)),
549        }
550    }
551
552    /// Finalizes one retry execution into the public CAS result type.
553    ///
554    /// # Parameters
555    /// - `attempt`: Retry-layer terminal success or error.
556    /// - `hooks`: Hook registrations for the current execution.
557    /// - `success_context`: Shared slot storing the success context.
558    ///
559    /// # Returns
560    /// Public CAS success or error.
561    fn finish_execution<R>(
562        &self,
563        attempt: Result<AttemptSuccess<T, R>, RetryError<CasAttemptFailure<T, E>>>,
564        hooks: CasHooks,
565        success_context: Arc<Mutex<Option<RetryContext>>>,
566        attempt_snapshot: Arc<Mutex<Option<Arc<T>>>>,
567        report_builder: Arc<Mutex<CasReportBuilder>>,
568    ) -> CasOutcome<T, R, E>
569    where
570        T: 'static,
571        E: 'static,
572    {
573        match attempt {
574            Ok(success) => {
575                let context = success_context
576                    .lock()
577                    .expect("CAS success context slot should be lockable")
578                    .take()
579                    .expect("retry success hook must capture CAS success context");
580                let attempts_total = context.attempt();
581                let max_attempts = context.max_attempts();
582                let max_operation_elapsed = context.max_operation_elapsed();
583                let max_total_elapsed = context.max_total_elapsed();
584                let outcome = match success {
585                    AttemptSuccess::Updated { .. } => CasExecutionOutcome::SuccessUpdated,
586                    AttemptSuccess::Finished { .. } => CasExecutionOutcome::SuccessFinished,
587                };
588                let success = self.enrich_success(success, context);
589                let report = self.finish_report(
590                    &hooks,
591                    report_builder,
592                    CasReportFinishContext::new(
593                        attempts_total,
594                        max_attempts,
595                        max_operation_elapsed,
596                        max_total_elapsed,
597                        outcome,
598                    ),
599                );
600                CasOutcome::new(Ok(success), report)
601            }
602            Err(error) => {
603                let timeout_current = attempt_snapshot
604                    .lock()
605                    .expect("CAS attempt snapshot slot should be lockable")
606                    .clone();
607                let error = CasError::new(error, timeout_current);
608                let context = error.context();
609                let outcome = Self::error_outcome(error.kind());
610                let report = self.finish_report(
611                    &hooks,
612                    report_builder,
613                    CasReportFinishContext::new(
614                        context.attempt(),
615                        context.max_attempts(),
616                        context.max_operation_elapsed(),
617                        context.max_total_elapsed(),
618                        outcome,
619                    ),
620                );
621                CasOutcome::new(Err(error), report)
622            }
623        }
624    }
625
626    /// Enriches one attempt success with the final CAS context.
627    ///
628    /// # Parameters
629    /// - `success`: Attempt success payload.
630    /// - `context`: Retry success context captured by the retry layer.
631    ///
632    /// # Returns
633    /// Public CAS success value with context attached.
634    fn enrich_success<R>(
635        &self,
636        success: AttemptSuccess<T, R>,
637        context: RetryContext,
638    ) -> CasSuccess<T, R> {
639        let context = CasContext::new(&context);
640        match success {
641            AttemptSuccess::Updated {
642                previous,
643                current,
644                output,
645            } => CasSuccess::updated(previous, current, output, context),
646            AttemptSuccess::Finished { current, output } => {
647                CasSuccess::finished(current, output, context)
648            }
649        }
650    }
651
652    /// Emits the execution-started event when event streaming is enabled.
653    ///
654    /// # Parameters
655    /// - `hooks`: Per-execution hooks (checked for event hook presence).
656    /// - `report_builder`: Used to obtain the start instant for the event.
657    fn emit_started(&self, hooks: &CasHooks, report_builder: &Arc<Mutex<CasReportBuilder>>)
658    where
659        T: 'static,
660        E: 'static,
661    {
662        if hooks.event_hook().is_none()
663            || self.observability.mode() == CasObservabilityMode::ReportOnly
664        {
665            return;
666        }
667        let started_at = report_builder
668            .lock()
669            .expect("CAS report builder should be lockable")
670            .started_at();
671        let event_hook = hooks.event_hook();
672        Self::dispatch_event(
673            &self.observability,
674            event_hook
675                .as_ref()
676                .expect("event hook should exist when events are emitted"),
677            CasEvent::ExecutionStarted { started_at },
678        );
679    }
680
681    /// Finishes and emits one execution report (and optional alert).
682    ///
683    /// Locks the report builder, finalizes the report, emits the
684    /// `ExecutionFinished` event if enabled, and dispatches a contention alert
685    /// if the mode and thresholds warrant it.
686    ///
687    /// # Parameters
688    /// - `hooks`: Used for event and alert dispatching.
689    /// - `report_builder`: Accumulator to finalize.
690    /// - `ctx`: Retry limits and terminal outcome for the report.
691    ///
692    /// # Returns
693    /// The finalized [`CasExecutionReport`].
694    fn finish_report(
695        &self,
696        hooks: &CasHooks,
697        report_builder: Arc<Mutex<CasReportBuilder>>,
698        ctx: CasReportFinishContext,
699    ) -> CasExecutionReport
700    where
701        T: 'static,
702        E: 'static,
703    {
704        let report = report_builder
705            .lock()
706            .expect("CAS report builder should be lockable")
707            .finish(
708                ctx.attempts_total,
709                ctx.max_attempts,
710                ctx.max_operation_elapsed,
711                ctx.max_total_elapsed,
712                ctx.outcome,
713            );
714        let event_hook = hooks.event_hook();
715        if Self::should_emit_events(&self.observability, &event_hook) {
716            Self::dispatch_event(
717                &self.observability,
718                event_hook
719                    .as_ref()
720                    .expect("event hook should exist when events are emitted"),
721                CasEvent::ExecutionFinished {
722                    report: report.clone(),
723                },
724            );
725        }
726        if self.observability.mode() == CasObservabilityMode::EventStreamWithAlert
727            && let Some(thresholds) = self.observability.contention_thresholds()
728            && report.is_contention_hot(&thresholds)
729        {
730            Self::dispatch_alert(
731                &self.observability,
732                &hooks.alert_hook(),
733                CasAlert::contention(report.clone(), thresholds),
734            );
735        }
736        report
737    }
738
739    /// Converts a terminal error kind into a report outcome.
740    ///
741    /// # Parameters
742    /// - `kind`: The high-level [`CasErrorKind`].
743    ///
744    /// # Returns
745    /// Corresponding [`CasExecutionOutcome`] variant for the report.
746    #[inline]
747    fn error_outcome(kind: CasErrorKind) -> CasExecutionOutcome {
748        match kind {
749            CasErrorKind::Abort => CasExecutionOutcome::ErrorAbort,
750            CasErrorKind::Conflict => CasExecutionOutcome::ErrorConflictExhausted,
751            CasErrorKind::RetryExhausted => CasExecutionOutcome::ErrorRetryExhausted,
752            CasErrorKind::AttemptTimeout => CasExecutionOutcome::ErrorAttemptTimeout,
753            CasErrorKind::MaxOperationElapsedExceeded => {
754                CasExecutionOutcome::ErrorMaxOperationElapsedExceeded
755            }
756            CasErrorKind::MaxTotalElapsedExceeded => {
757                CasExecutionOutcome::ErrorMaxTotalElapsedExceeded
758            }
759        }
760    }
761
762    /// Converts one attempt failure into its lightweight event kind.
763    ///
764    /// # Parameters
765    /// - `failure`: The [`CasAttemptFailure`] to classify.
766    ///
767    /// # Returns
768    /// The [`CasAttemptFailureKind`] for event emission.
769    #[inline]
770    fn failure_kind(failure: &CasAttemptFailure<T, E>) -> crate::error::CasAttemptFailureKind {
771        failure.kind()
772    }
773
774    /// Dispatches one lifecycle event if event streaming is enabled.
775    fn dispatch_event(
776        observability: &CasObservabilityConfig,
777        hook: &crate::event::CasEventHook,
778        event: CasEvent,
779    ) where
780        T: 'static,
781        E: 'static,
782    {
783        match observability.listener_panic_policy() {
784            ListenerPanicPolicy::Propagate => hook.accept(&event),
785            ListenerPanicPolicy::Isolate => {
786                let _ = catch_unwind(AssertUnwindSafe(|| hook.accept(&event)));
787            }
788        }
789    }
790
791    /// Returns whether lifecycle event construction and dispatch are needed.
792    #[inline]
793    fn should_emit_events(
794        observability: &CasObservabilityConfig,
795        hook: &Option<crate::event::CasEventHook>,
796    ) -> bool {
797        observability.mode() != CasObservabilityMode::ReportOnly && hook.is_some()
798    }
799
800    /// Dispatches one alert if an alert listener is registered.
801    fn dispatch_alert(
802        observability: &CasObservabilityConfig,
803        hook: &Option<crate::event::CasAlertHook>,
804        alert: CasAlert,
805    ) {
806        if let Some(hook) = hook {
807            match observability.listener_panic_policy() {
808                ListenerPanicPolicy::Propagate => hook.accept(&alert),
809                ListenerPanicPolicy::Isolate => {
810                    let _ = catch_unwind(AssertUnwindSafe(|| hook.accept(&alert)));
811                }
812            }
813        }
814    }
815}
qubit_cas/executor/cas_executor.rs

qubit_cas/executor/
cas_executor.rs
