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>(&self, state: &AtomicRef<T>, operation: O, hooks: CasHooks) -> CasOutcome<T, R, E>
251    where
252        T: 'static,
253        E: 'static,
254        O: Function<T, CasDecision<T, R, E>>,
255    {
256        let success_context = Arc::new(Mutex::new(None));
257        let attempt_snapshot = Arc::new(Mutex::new(None));
258        let report_builder = Arc::new(Mutex::new(CasReportBuilder::start()));
259        self.emit_started(&hooks, &report_builder);
260        let retry = self.build_retry(&hooks, Arc::clone(&success_context), Arc::clone(&report_builder));
261        let attempt = retry.run(|| self.run_sync_attempt(state, &operation));
262        self.finish_execution(attempt, hooks, success_context, attempt_snapshot, report_builder)
263    }
264
265    /// Executes one asynchronous CAS operation.
266    ///
267    /// # Parameters
268    /// - `state`: Shared atomic state container.
269    /// - `operation`: Async operation factory receiving one state snapshot.
270    ///
271    /// # Returns
272    /// A terminal result together with the execution report.
273    #[cfg(feature = "tokio")]
274    pub async fn execute_async<R, O, Fut>(&self, state: &AtomicRef<T>, operation: O) -> CasOutcome<T, R, E>
275    where
276        T: 'static,
277        E: 'static,
278        O: Fn(Arc<T>) -> Fut,
279        Fut: std::future::Future<Output = CasDecision<T, R, E>>,
280    {
281        self.execute_async_with_hooks(state, operation, CasHooks::new()).await
282    }
283
284    /// Executes one asynchronous CAS operation with lifecycle hooks.
285    ///
286    /// # Parameters
287    /// - `state`: Shared atomic state container.
288    /// - `operation`: Async operation factory receiving one state snapshot.
289    /// - `hooks`: Per-execution hook registrations.
290    ///
291    /// # Returns
292    /// A terminal result together with the execution report.
293    #[cfg(feature = "tokio")]
294    pub async fn execute_async_with_hooks<R, O, Fut>(
295        &self,
296        state: &AtomicRef<T>,
297        operation: O,
298        hooks: CasHooks,
299    ) -> CasOutcome<T, R, E>
300    where
301        T: 'static,
302        E: 'static,
303        O: Fn(Arc<T>) -> Fut,
304        Fut: std::future::Future<Output = CasDecision<T, R, E>>,
305    {
306        let success_context = Arc::new(Mutex::new(None));
307        let attempt_snapshot = Arc::new(Mutex::new(None));
308        let report_builder = Arc::new(Mutex::new(CasReportBuilder::start()));
309        self.emit_started(&hooks, &report_builder);
310        let retry = self.build_retry(&hooks, Arc::clone(&success_context), Arc::clone(&report_builder));
311        let attempt_snapshot_for_attempt = Arc::clone(&attempt_snapshot);
312        let attempt = retry
313            .run_async(|| self.run_async_attempt(state, &operation, Arc::clone(&attempt_snapshot_for_attempt)))
314            .await;
315        self.finish_execution(attempt, hooks, success_context, attempt_snapshot, report_builder)
316    }
317
318    /// Builds one retry policy for a single CAS execution.
319    ///
320    /// # Parameters
321    /// - `hooks`: Hook registrations for the current execution.
322    /// - `success_context`: Shared slot used to capture the retry success
323    ///   context.
324    /// # Returns
325    /// A retry policy configured for one CAS execution.
326    fn build_retry(
327        &self,
328        hooks: &CasHooks,
329        success_context: Arc<Mutex<Option<RetryContext>>>,
330        report_builder: Arc<Mutex<CasReportBuilder>>,
331    ) -> Retry<CasAttemptFailure<T, E>>
332    where
333        T: 'static,
334        E: 'static,
335    {
336        let event_hook = hooks.event_hook();
337        let retry_timeout_policy = self
338            .options
339            .attempt_timeout()
340            .map(|attempt_timeout| attempt_timeout.policy());
341        let observability = self.observability.clone();
342
343        let mut builder = Retry::<CasAttemptFailure<T, E>>::builder()
344            .options(self.options.clone())
345            .on_success(move |context: &RetryContext| {
346                *success_context
347                    .lock()
348                    .expect("CAS success context slot should be lockable") = Some(*context);
349            })
350            .on_failure(
351                move |failure: &AttemptFailure<CasAttemptFailure<T, E>>, context: &RetryContext| {
352                    let failure = match failure {
353                        AttemptFailure::Panic(_) | AttemptFailure::Executor(_) => {
354                            return AttemptFailureDecision::UseDefault;
355                        }
356                        AttemptFailure::Error(failure) => failure,
357                        AttemptFailure::Timeout => {
358                            let cas_context = CasContext::new(context);
359                            report_builder
360                                .lock()
361                                .expect("CAS report builder should be lockable")
362                                .record_timeout();
363                            if Self::should_emit_events(&observability, &event_hook) {
364                                let hook = event_hook
365                                    .as_ref()
366                                    .expect("event hook should exist when events are emitted");
367                                Self::dispatch_event(
368                                    &observability,
369                                    hook,
370                                    CasEvent::AttemptFailed {
371                                        context: cas_context,
372                                        kind: crate::error::CasAttemptFailureKind::Timeout,
373                                    },
374                                );
375                                if context.attempt_timeout_source() == Some(AttemptTimeoutSource::Configured)
376                                    && retry_timeout_policy == Some(AttemptTimeoutPolicy::Retry)
377                                {
378                                    Self::dispatch_event(
379                                        &observability,
380                                        hook,
381                                        CasEvent::RetryRequested { context: cas_context },
382                                    );
383                                }
384                            }
385                            return AttemptFailureDecision::UseDefault;
386                        }
387                    };
388                    let cas_context = CasContext::new(context);
389                    {
390                        let mut report = report_builder.lock().expect("CAS report builder should be lockable");
391                        match failure {
392                            CasAttemptFailure::Conflict { .. } => report.record_conflict(),
393                            CasAttemptFailure::Retry { .. } => report.record_retry_error(),
394                            CasAttemptFailure::Abort { .. } => report.record_abort(),
395                            CasAttemptFailure::Timeout { .. } => report.record_timeout(),
396                        }
397                    }
398                    if Self::should_emit_events(&observability, &event_hook) {
399                        Self::dispatch_event(
400                            &observability,
401                            event_hook
402                                .as_ref()
403                                .expect("event hook should exist when events are emitted"),
404                            CasEvent::AttemptFailed {
405                                context: cas_context,
406                                kind: Self::failure_kind(failure),
407                            },
408                        );
409                    }
410                    match failure {
411                        CasAttemptFailure::Conflict { .. } | CasAttemptFailure::Retry { .. } => {
412                            if Self::should_emit_events(&observability, &event_hook) {
413                                Self::dispatch_event(
414                                    &observability,
415                                    event_hook
416                                        .as_ref()
417                                        .expect("event hook should exist when events are emitted"),
418                                    CasEvent::RetryRequested { context: cas_context },
419                                );
420                            }
421                            AttemptFailureDecision::Retry
422                        }
423                        CasAttemptFailure::Abort { .. } => AttemptFailureDecision::Abort,
424                        CasAttemptFailure::Timeout { .. } => AttemptFailureDecision::UseDefault,
425                    }
426                },
427            );
428
429        if self.observability.listener_panic_policy() == ListenerPanicPolicy::Isolate {
430            builder = builder.isolate_listener_panics();
431        }
432        builder
433            .build()
434            .expect("validated CAS executor configuration must build retry policy")
435    }
436
437    /// Runs one synchronous CAS attempt.
438    ///
439    /// # Parameters
440    /// - `state`: Shared atomic state container.
441    /// - `operation`: Pure operation over the current state snapshot.
442    ///
443    /// # Returns
444    /// An attempt success or one attempt failure.
445    fn run_sync_attempt<R, O>(
446        &self,
447        state: &AtomicRef<T>,
448        operation: &O,
449    ) -> Result<AttemptSuccess<T, R>, CasAttemptFailure<T, E>>
450    where
451        O: Function<T, CasDecision<T, R, E>>,
452    {
453        let current = state.load();
454        match operation.apply(current.as_ref()) {
455            CasDecision::Update { next, output } => match state.compare_set(&current, Arc::clone(&next)) {
456                Ok(()) => Ok(AttemptSuccess::Updated {
457                    previous: current,
458                    current: next,
459                    output,
460                }),
461                Err(actual) => Err(CasAttemptFailure::conflict(actual)),
462            },
463            CasDecision::Finish { output } => Ok(AttemptSuccess::Finished { current, output }),
464            CasDecision::Retry(error) => Err(CasAttemptFailure::retry(current, error)),
465            CasDecision::Abort(error) => Err(CasAttemptFailure::abort(current, error)),
466        }
467    }
468
469    /// Runs one asynchronous CAS attempt.
470    ///
471    /// # Parameters
472    /// - `state`: Shared atomic state container.
473    /// - `operation`: Async operation factory over one state snapshot.
474    ///
475    /// # Returns
476    /// An attempt success or one attempt failure.
477    #[cfg(feature = "tokio")]
478    async fn run_async_attempt<R, O, Fut>(
479        &self,
480        state: &AtomicRef<T>,
481        operation: &O,
482        attempt_snapshot: Arc<Mutex<Option<Arc<T>>>>,
483    ) -> Result<AttemptSuccess<T, R>, CasAttemptFailure<T, E>>
484    where
485        O: Fn(Arc<T>) -> Fut,
486        Fut: std::future::Future<Output = CasDecision<T, R, E>>,
487    {
488        let current = state.load();
489        *attempt_snapshot
490            .lock()
491            .expect("CAS attempt snapshot slot should be lockable") = Some(Arc::clone(&current));
492        let decision = operation(Arc::clone(&current)).await;
493
494        match decision {
495            CasDecision::Update { next, output } => 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            CasDecision::Finish { output } => Ok(AttemptSuccess::Finished { current, output }),
504            CasDecision::Retry(error) => Err(CasAttemptFailure::retry(current, error)),
505            CasDecision::Abort(error) => Err(CasAttemptFailure::abort(current, error)),
506        }
507    }
508
509    /// Finalizes one retry execution into the public CAS result type.
510    ///
511    /// # Parameters
512    /// - `attempt`: Retry-layer terminal success or error.
513    /// - `hooks`: Hook registrations for the current execution.
514    /// - `success_context`: Shared slot storing the success context.
515    ///
516    /// # Returns
517    /// Public CAS success or error.
518    fn finish_execution<R>(
519        &self,
520        attempt: Result<AttemptSuccess<T, R>, RetryError<CasAttemptFailure<T, E>>>,
521        hooks: CasHooks,
522        success_context: Arc<Mutex<Option<RetryContext>>>,
523        attempt_snapshot: Arc<Mutex<Option<Arc<T>>>>,
524        report_builder: Arc<Mutex<CasReportBuilder>>,
525    ) -> CasOutcome<T, R, E>
526    where
527        T: 'static,
528        E: 'static,
529    {
530        match attempt {
531            Ok(success) => {
532                let context = success_context
533                    .lock()
534                    .expect("CAS success context slot should be lockable")
535                    .take()
536                    .expect("retry success hook must capture CAS success context");
537                let attempts_total = context.attempt();
538                let max_attempts = context.max_attempts();
539                let max_operation_elapsed = context.max_operation_elapsed();
540                let max_total_elapsed = context.max_total_elapsed();
541                let outcome = match success {
542                    AttemptSuccess::Updated { .. } => CasExecutionOutcome::SuccessUpdated,
543                    AttemptSuccess::Finished { .. } => CasExecutionOutcome::SuccessFinished,
544                };
545                let success = self.enrich_success(success, context);
546                let report = self.finish_report(
547                    &hooks,
548                    report_builder,
549                    CasReportFinishContext::new(
550                        attempts_total,
551                        max_attempts,
552                        max_operation_elapsed,
553                        max_total_elapsed,
554                        outcome,
555                    ),
556                );
557                CasOutcome::new(Ok(success), report)
558            }
559            Err(error) => {
560                let timeout_current = attempt_snapshot
561                    .lock()
562                    .expect("CAS attempt snapshot slot should be lockable")
563                    .clone();
564                let error = CasError::new(error, timeout_current);
565                let context = error.context();
566                let outcome = Self::error_outcome(error.kind());
567                let report = self.finish_report(
568                    &hooks,
569                    report_builder,
570                    CasReportFinishContext::new(
571                        context.attempt(),
572                        context.max_attempts(),
573                        context.max_operation_elapsed(),
574                        context.max_total_elapsed(),
575                        outcome,
576                    ),
577                );
578                CasOutcome::new(Err(error), report)
579            }
580        }
581    }
582
583    /// Enriches one attempt success with the final CAS context.
584    ///
585    /// # Parameters
586    /// - `success`: Attempt success payload.
587    /// - `context`: Retry success context captured by the retry layer.
588    ///
589    /// # Returns
590    /// Public CAS success value with context attached.
591    fn enrich_success<R>(&self, success: AttemptSuccess<T, R>, context: RetryContext) -> CasSuccess<T, R> {
592        let context = CasContext::new(&context);
593        match success {
594            AttemptSuccess::Updated {
595                previous,
596                current,
597                output,
598            } => CasSuccess::updated(previous, current, output, context),
599            AttemptSuccess::Finished { current, output } => CasSuccess::finished(current, output, context),
600        }
601    }
602
603    /// Emits the execution-started event when event streaming is enabled.
604    ///
605    /// # Parameters
606    /// - `hooks`: Per-execution hooks (checked for event hook presence).
607    /// - `report_builder`: Used to obtain the start instant for the event.
608    fn emit_started(&self, hooks: &CasHooks, report_builder: &Arc<Mutex<CasReportBuilder>>)
609    where
610        T: 'static,
611        E: 'static,
612    {
613        if hooks.event_hook().is_none() || self.observability.mode() == CasObservabilityMode::ReportOnly {
614            return;
615        }
616        let started_at = report_builder
617            .lock()
618            .expect("CAS report builder should be lockable")
619            .started_at();
620        let event_hook = hooks.event_hook();
621        Self::dispatch_event(
622            &self.observability,
623            event_hook
624                .as_ref()
625                .expect("event hook should exist when events are emitted"),
626            CasEvent::ExecutionStarted { started_at },
627        );
628    }
629
630    /// Finishes and emits one execution report (and optional alert).
631    ///
632    /// Locks the report builder, finalizes the report, emits the
633    /// `ExecutionFinished` event if enabled, and dispatches a contention alert
634    /// if the mode and thresholds warrant it.
635    ///
636    /// # Parameters
637    /// - `hooks`: Used for event and alert dispatching.
638    /// - `report_builder`: Accumulator to finalize.
639    /// - `ctx`: Retry limits and terminal outcome for the report.
640    ///
641    /// # Returns
642    /// The finalized [`CasExecutionReport`].
643    fn finish_report(
644        &self,
645        hooks: &CasHooks,
646        report_builder: Arc<Mutex<CasReportBuilder>>,
647        ctx: CasReportFinishContext,
648    ) -> CasExecutionReport
649    where
650        T: 'static,
651        E: 'static,
652    {
653        let report = report_builder
654            .lock()
655            .expect("CAS report builder should be lockable")
656            .finish(
657                ctx.attempts_total,
658                ctx.max_attempts,
659                ctx.max_operation_elapsed,
660                ctx.max_total_elapsed,
661                ctx.outcome,
662            );
663        let event_hook = hooks.event_hook();
664        if Self::should_emit_events(&self.observability, &event_hook) {
665            Self::dispatch_event(
666                &self.observability,
667                event_hook
668                    .as_ref()
669                    .expect("event hook should exist when events are emitted"),
670                CasEvent::ExecutionFinished { report: report.clone() },
671            );
672        }
673        if self.observability.mode() == CasObservabilityMode::EventStreamWithAlert
674            && let Some(thresholds) = self.observability.contention_thresholds()
675            && report.is_contention_hot(&thresholds)
676        {
677            Self::dispatch_alert(
678                &self.observability,
679                &hooks.alert_hook(),
680                CasAlert::contention(report.clone(), thresholds),
681            );
682        }
683        report
684    }
685
686    /// Converts a terminal error kind into a report outcome.
687    ///
688    /// # Parameters
689    /// - `kind`: The high-level [`CasErrorKind`].
690    ///
691    /// # Returns
692    /// Corresponding [`CasExecutionOutcome`] variant for the report.
693    #[inline]
694    fn error_outcome(kind: CasErrorKind) -> CasExecutionOutcome {
695        match kind {
696            CasErrorKind::Abort => CasExecutionOutcome::ErrorAbort,
697            CasErrorKind::Conflict => CasExecutionOutcome::ErrorConflictExhausted,
698            CasErrorKind::RetryExhausted => CasExecutionOutcome::ErrorRetryExhausted,
699            CasErrorKind::AttemptTimeout => CasExecutionOutcome::ErrorAttemptTimeout,
700            CasErrorKind::MaxOperationElapsedExceeded => CasExecutionOutcome::ErrorMaxOperationElapsedExceeded,
701            CasErrorKind::MaxTotalElapsedExceeded => CasExecutionOutcome::ErrorMaxTotalElapsedExceeded,
702        }
703    }
704
705    /// Converts one attempt failure into its lightweight event kind.
706    ///
707    /// # Parameters
708    /// - `failure`: The [`CasAttemptFailure`] to classify.
709    ///
710    /// # Returns
711    /// The [`CasAttemptFailureKind`] for event emission.
712    #[inline]
713    fn failure_kind(failure: &CasAttemptFailure<T, E>) -> crate::error::CasAttemptFailureKind {
714        failure.kind()
715    }
716
717    /// Dispatches one lifecycle event if event streaming is enabled.
718    fn dispatch_event(observability: &CasObservabilityConfig, hook: &crate::event::CasEventHook, event: CasEvent)
719    where
720        T: 'static,
721        E: 'static,
722    {
723        match observability.listener_panic_policy() {
724            ListenerPanicPolicy::Propagate => hook.accept(&event),
725            ListenerPanicPolicy::Isolate => {
726                let _ = catch_unwind(AssertUnwindSafe(|| hook.accept(&event)));
727            }
728        }
729    }
730
731    /// Returns whether lifecycle event construction and dispatch are needed.
732    #[inline]
733    fn should_emit_events(observability: &CasObservabilityConfig, hook: &Option<crate::event::CasEventHook>) -> bool {
734        observability.mode() != CasObservabilityMode::ReportOnly && hook.is_some()
735    }
736
737    /// Dispatches one alert if an alert listener is registered.
738    fn dispatch_alert(
739        observability: &CasObservabilityConfig,
740        hook: &Option<crate::event::CasAlertHook>,
741        alert: CasAlert,
742    ) {
743        if let Some(hook) = hook {
744            match observability.listener_panic_policy() {
745                ListenerPanicPolicy::Propagate => hook.accept(&alert),
746                ListenerPanicPolicy::Isolate => {
747                    let _ = catch_unwind(AssertUnwindSafe(|| hook.accept(&alert)));
748                }
749            }
750        }
751    }
752}
qubit_cas/executor/cas_executor.rs

qubit_cas/executor/
cas_executor.rs
