dialcache/
engine.rs

1//! The cache instance: construction, scopes, maintenance and call admission.
2
3use std::collections::{HashMap, HashSet};
4use std::future::Future;
5use std::num::NonZeroUsize;
6use std::panic::{catch_unwind, AssertUnwindSafe};
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::Arc;
9
10use futures::future::BoxFuture;
11use parking_lot::Mutex;
12
13use crate::clock::Clock;
14use crate::error::{BoxError, ConfigError, Error};
15use crate::flight::{start_pending, Flight, Settled, ValueResult};
16use crate::identity::{Identity, IntoKeyId};
17use crate::limits::{
18    DEFAULT_LOCAL_CAPACITY, DEFAULT_REMOTE_READ_TIMEOUT_MS, DEFAULT_SHADOW_MAX_IN_FLIGHT,
19    MAX_DEADLINE_MS, MAX_SAFE_INTEGER, MAX_SUPPORTED_DURATION_MS, WATERMARK_USE_CASE,
20};
21use crate::local::{LocalEntry, LocalRead, LocalStore, LruLocalStore, StoredValue};
22use crate::observe::{Event, Labels, Layer, LogEvent, Logger, Observer, OutcomeLabels};
23use crate::operation::{downcast_value, erase_load, ErasedOperation, Operation, RecoveryPredicate};
24use crate::policy::RuntimePolicy;
25use crate::protocol::CompressionConfig;
26use crate::remote::{InvalidateRequest, Remote};
27use crate::runtime::Runtime;
28use crate::scope::{Owner, Scope};
29use crate::shadow::ShadowFlight;
30
31/// Resolves a sparse runtime policy overlay once per enabled call.
32pub type PolicyProvider = Arc<
33    dyn Fn(Identity) -> BoxFuture<'static, Result<Option<RuntimePolicy>, BoxError>> + Send + Sync,
34>;
35
36static NEXT_CACHE_ID: AtomicU64 = AtomicU64::new(1);
37
38pub(crate) struct CoreState {
39    pub(crate) local: Option<Box<dyn LocalStore>>,
40    pub(crate) flights: HashMap<String, Arc<Flight>>,
41    pub(crate) shadows: HashMap<String, Arc<ShadowFlight>>,
42    pub(crate) registered: HashSet<String>,
43}
44
45pub(crate) struct Core {
46    pub(crate) id: u64,
47    pub(crate) namespace: Arc<str>,
48    pub(crate) remote: Option<Arc<dyn Remote>>,
49    pub(crate) observer: Option<Arc<dyn Observer>>,
50    pub(crate) logger: Arc<dyn Logger>,
51    pub(crate) provider: Option<PolicyProvider>,
52    pub(crate) remote_read_timeout_ms: u64,
53    pub(crate) shadow_max_in_flight: usize,
54    pub(crate) should_recover: Option<RecoveryPredicate>,
55    /// `None` disables new compressed writes; reads still decompress.
56    pub(crate) compression: Option<CompressionConfig>,
57    pub(crate) clock: Arc<dyn Clock>,
58    pub(crate) runtime: Arc<dyn Runtime>,
59    pub(crate) state: Mutex<CoreState>,
60}
61
62impl Core {
63    /// Deliver a diagnostic. Observer failures never change results.
64    pub(crate) fn emit(&self, event: Event) {
65        if let Some(observer) = &self.observer {
66            let _ = catch_unwind(AssertUnwindSafe(|| observer.observe(&event)));
67        }
68    }
69
70    /// Deliver a log event; `false` when the logger panicked. Logger
71    /// failures never change results.
72    pub(crate) fn log(&self, event: LogEvent) -> bool {
73        catch_unwind(AssertUnwindSafe(|| self.logger.log(&event))).is_ok()
74    }
75
76    pub(crate) fn shadow_hook_enabled(&self) -> bool {
77        match &self.observer {
78            Some(observer) => {
79                catch_unwind(AssertUnwindSafe(|| observer.observes_shadow_outcomes()))
80                    .unwrap_or(false)
81            }
82            None => false,
83        }
84    }
85
86    /// Read a live local entry, promoting it. `Ok(None)` is a miss.
87    pub(crate) fn local_get(&self, key: &str) -> Result<Option<StoredValue>, BoxError> {
88        // Read the clock before taking the lock: it is the only external code
89        // on this path, and a failure must not poison shared state.
90        let now_ms = self.clock.elapsed_ms();
91        let read = {
92            let mut state = self.state.lock();
93            match state.local.as_mut() {
94                None => return Ok(None),
95                Some(local) => catch_unwind(AssertUnwindSafe(|| local.get(key, now_ms)))
96                    .unwrap_or_else(|payload| {
97                        Err(crate::flight::panic_message(payload).to_string().into())
98                    }),
99            }
100        };
101        // An expired entry drops here, outside the lock: a value's destructor
102        // may call back into the cache.
103        match read? {
104            LocalRead::Live(value) => Ok(Some(value)),
105            LocalRead::Absent | LocalRead::Expired(_) => Ok(None),
106        }
107    }
108
109    pub(crate) fn local_put(
110        &self,
111        key: &str,
112        value: StoredValue,
113        ttl_ms: u64,
114    ) -> Result<(), BoxError> {
115        let now_ms = self.clock.elapsed_ms();
116        let displaced = {
117            let mut state = self.state.lock();
118            match state.local.as_mut() {
119                None => return Ok(()),
120                Some(local) => {
121                    let entry = LocalEntry {
122                        value,
123                        inserted_ms: now_ms,
124                        ttl_ms: ttl_ms.min(i64::MAX as u64) as i64,
125                    };
126                    catch_unwind(AssertUnwindSafe(|| local.put(key.to_string(), entry)))
127                        .unwrap_or_else(|payload| {
128                            Err(crate::flight::panic_message(payload).to_string().into())
129                        })
130                }
131            }
132        };
133        // The displaced entry drops here, outside the lock.
134        displaced.map(|_| ())
135    }
136}
137
138/// Exact process-scoped single-flight state of one instance.
139#[derive(Debug, Clone, PartialEq, Eq, Default)]
140pub struct ProcessCoalescingState {
141    /// Executions currently in flight on this instance, one per logical key.
142    pub active_leaders: usize,
143    /// Callers currently waiting on those executions instead of running their own.
144    pub active_followers: usize,
145    /// Milliseconds since the oldest in-flight execution started; `None`
146    /// when nothing is in flight.
147    pub oldest_leader_age_ms: Option<u64>,
148}
149
150/// A point-in-time snapshot of cache-owned coalescing state.
151#[derive(Debug, Clone, PartialEq, Eq, Default)]
152pub struct CoalescingState {
153    /// Instance-wide single-flight state; request-scoped flights are not counted.
154    pub process: ProcessCoalescingState,
155}
156
157/// Configures a [`DialCache`].
158pub struct DialCacheBuilder {
159    namespace: String,
160    remote: Option<Arc<dyn Remote>>,
161    observer: Option<Arc<dyn Observer>>,
162    logger: Option<Arc<dyn Logger>>,
163    provider: Option<PolicyProvider>,
164    local_capacity: Option<usize>,
165    local_store: Option<Box<dyn LocalStore>>,
166    remote_read_timeout_ms: Option<u64>,
167    shadow_max_in_flight: Option<usize>,
168    should_recover: Option<RecoveryPredicate>,
169    compression: Option<Option<CompressionConfig>>,
170    clock: Option<Arc<dyn Clock>>,
171    runtime: Option<Arc<dyn Runtime>>,
172}
173
174impl std::fmt::Debug for DialCacheBuilder {
175    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
176        f.debug_struct("DialCacheBuilder")
177            .field("namespace", &self.namespace)
178            .finish_non_exhaustive()
179    }
180}
181
182impl DialCacheBuilder {
183    fn new() -> Self {
184        DialCacheBuilder {
185            namespace: "urn".to_string(),
186            remote: None,
187            observer: None,
188            logger: None,
189            provider: None,
190            local_capacity: None,
191            local_store: None,
192            remote_read_timeout_ms: None,
193            shadow_max_in_flight: None,
194            should_recover: None,
195            compression: None,
196            clock: None,
197            runtime: None,
198        }
199    }
200
201    /// Logical namespace used in keys, invalidation identity, cohorts and
202    /// labels. Defaults to `"urn"`; may not contain `{` or `}`.
203    pub fn namespace(mut self, namespace: impl Into<String>) -> Self {
204        self.namespace = namespace.into();
205        self
206    }
207
208    /// The remote layer adapter. Without one, only request and local layers serve.
209    pub fn remote(mut self, remote: impl Remote) -> Self {
210        self.remote = Some(Arc::new(remote));
211        self
212    }
213
214    /// The remote layer adapter, shared.
215    pub fn remote_arc(mut self, remote: Arc<dyn Remote>) -> Self {
216        self.remote = Some(remote);
217        self
218    }
219
220    /// Receives every public diagnostic; also the shadow outcome hook.
221    pub fn observer(mut self, observer: impl Observer) -> Self {
222        self.observer = Some(Arc::new(observer));
223        self
224    }
225
226    /// Receives every public diagnostic, shared.
227    pub fn observer_arc(mut self, observer: Arc<dyn Observer>) -> Self {
228        self.observer = Some(observer);
229        self
230    }
231
232    /// Receives structured log events. Defaults to the `log` crate facade
233    /// under target `dialcache`.
234    pub fn logger(mut self, logger: impl Logger) -> Self {
235        self.logger = Some(Arc::new(logger));
236        self
237    }
238
239    /// Receives structured log events, shared.
240    pub fn logger_arc(mut self, logger: Arc<dyn Logger>) -> Self {
241        self.logger = Some(logger);
242        self
243    }
244
245    /// Runtime policy overlay resolved once per enabled call. `Ok(None)`
246    /// inherits the operation policy; an error bypasses caching for that call.
247    pub fn policy_provider<F, Fut>(mut self, provider: F) -> Self
248    where
249        F: Fn(Identity) -> Fut + Send + Sync + 'static,
250        Fut: Future<Output = Result<Option<RuntimePolicy>, BoxError>> + Send + 'static,
251    {
252        self.provider = Some(Arc::new(move |identity| Box::pin(provider(identity))));
253        self
254    }
255
256    /// Maximum process-local entries across every use case. Zero disables
257    /// local storage while preserving coalescing. Defaults to 10,000.
258    pub fn local_capacity(mut self, capacity: usize) -> Self {
259        self.local_capacity = Some(capacity);
260        self
261    }
262
263    /// Replace the default LRU store. Advanced: the store must preserve the
264    /// documented LRU, expiry and promotion rules.
265    pub fn local_store(mut self, store: Box<dyn LocalStore>) -> Self {
266        self.local_store = Some(store);
267        self
268    }
269
270    /// Instance default remote read budget. Defaults to 50 ms.
271    pub fn remote_read_timeout_ms(mut self, ms: u64) -> Self {
272        self.remote_read_timeout_ms = Some(ms);
273        self
274    }
275
276    /// Concurrent shadow jobs per instance; excess work is dropped. Defaults to 1.
277    pub fn shadow_max_in_flight(mut self, jobs: usize) -> Self {
278        self.shadow_max_in_flight = Some(jobs);
279        self
280    }
281
282    /// Instance default classifier deciding whether a source failure may use
283    /// a retained stale value. Omission admits only the source deadline error.
284    pub fn should_recover(
285        mut self,
286        predicate: impl Fn(&Error) -> bool + Send + Sync + 'static,
287    ) -> Self {
288        self.should_recover = Some(Arc::new(predicate));
289        self
290    }
291
292    /// Write-side compression policy. Enabled by default at a 4,096-byte threshold.
293    pub fn compression(mut self, config: CompressionConfig) -> Self {
294        self.compression = Some(Some(config));
295        self
296    }
297
298    /// Store every payload uncompressed. Reads still accept compressed entries.
299    pub fn disable_compression(mut self) -> Self {
300        self.compression = Some(None);
301        self
302    }
303
304    /// Replace the wall and elapsed clock. Defaults to
305    /// [`SystemClock`](crate::SystemClock).
306    pub fn clock(mut self, clock: impl Clock) -> Self {
307        self.clock = Some(Arc::new(clock));
308        self
309    }
310
311    /// Share a clock between instances (for example a controlled test clock).
312    pub fn clock_arc(mut self, clock: Arc<dyn Clock>) -> Self {
313        self.clock = Some(clock);
314        self
315    }
316
317    /// Share a runtime between instances.
318    pub fn runtime_arc(mut self, runtime: Arc<dyn Runtime>) -> Self {
319        self.runtime = Some(runtime);
320        self
321    }
322
323    /// Replace the executor and timer source. Defaults to tokio.
324    pub fn runtime(mut self, runtime: impl Runtime) -> Self {
325        self.runtime = Some(Arc::new(runtime));
326        self
327    }
328
329    /// Validate the configuration and create the instance. Rejects a
330    /// namespace containing `{` or `}`, a read budget outside
331    /// `1..=2_147_483_647` ms, a zero shadow cap, an invalid compression
332    /// config, or no usable runtime: with the `tokio` feature, no tokio
333    /// runtime is current and none was supplied through `runtime` or
334    /// `runtime_arc`; without it, no runtime was supplied.
335    pub fn build(self) -> Result<DialCache, ConfigError> {
336        if self.namespace.contains(['{', '}']) {
337            return Err(ConfigError::invalid(
338                "DialCache namespace must not contain \"{\" or \"}\"",
339            ));
340        }
341        let compression = match self.compression {
342            None => Some(CompressionConfig::default()),
343            Some(None) => None,
344            Some(Some(config)) => {
345                config
346                    .validate()
347                    .map_err(|e| ConfigError::invalid(e.to_string()))?;
348                Some(config)
349            }
350        };
351        let remote_read_timeout_ms = self
352            .remote_read_timeout_ms
353            .unwrap_or(DEFAULT_REMOTE_READ_TIMEOUT_MS);
354        if !(1..=MAX_DEADLINE_MS).contains(&remote_read_timeout_ms) {
355            return Err(ConfigError::invalid(format!(
356                "DialCache remote read timeout must be a positive integer no greater than {MAX_DEADLINE_MS} ms"
357            )));
358        }
359        let shadow_max_in_flight = self
360            .shadow_max_in_flight
361            .unwrap_or(DEFAULT_SHADOW_MAX_IN_FLIGHT);
362        if shadow_max_in_flight == 0 {
363            return Err(ConfigError::invalid(
364                "DialCache shadow_max_in_flight must be positive",
365            ));
366        }
367        let local: Option<Box<dyn LocalStore>> = match self.local_store {
368            Some(store) => Some(store),
369            None => {
370                let capacity = self.local_capacity.unwrap_or(DEFAULT_LOCAL_CAPACITY);
371                if capacity as u64 > MAX_SAFE_INTEGER {
372                    return Err(ConfigError::invalid(
373                        "DialCache local capacity must be a safe integer",
374                    ));
375                }
376                NonZeroUsize::new(capacity)
377                    .map(|capacity| Box::new(LruLocalStore::new(capacity)) as Box<dyn LocalStore>)
378            }
379        };
380        let clock = match self.clock {
381            Some(clock) => clock,
382            None => default_clock()?,
383        };
384        let runtime = match self.runtime {
385            Some(runtime) => runtime,
386            None => default_runtime()?,
387        };
388        let core = Core {
389            id: NEXT_CACHE_ID.fetch_add(1, Ordering::Relaxed),
390            namespace: Arc::from(self.namespace.as_str()),
391            remote: self.remote,
392            observer: self.observer,
393            logger: self
394                .logger
395                .unwrap_or_else(|| Arc::new(crate::observe::LogFacadeLogger)),
396            provider: self.provider,
397            remote_read_timeout_ms,
398            shadow_max_in_flight,
399            should_recover: self.should_recover,
400            compression,
401            clock,
402            runtime,
403            state: Mutex::new(CoreState {
404                local,
405                flights: HashMap::new(),
406                shadows: HashMap::new(),
407                registered: HashSet::new(),
408            }),
409        };
410        Ok(DialCache {
411            core: Arc::new(core),
412        })
413    }
414}
415
416fn default_clock() -> Result<Arc<dyn Clock>, ConfigError> {
417    Ok(Arc::new(crate::clock::SystemClock::new()))
418}
419
420#[cfg(feature = "tokio")]
421fn default_runtime() -> Result<Arc<dyn Runtime>, ConfigError> {
422    Ok(Arc::new(crate::runtime::TokioRuntime::current()?))
423}
424
425#[cfg(not(feature = "tokio"))]
426fn default_runtime() -> Result<Arc<dyn Runtime>, ConfigError> {
427    Err(ConfigError::invalid("DialCache needs a runtime: enable the tokio feature or supply one with DialCacheBuilder::runtime"))
428}
429
430/// Holds the outermost enabled scope open until dropped.
431#[derive(Debug)]
432pub struct ScopeGuard {
433    scope: Scope,
434    owner: Arc<Owner>,
435}
436
437impl ScopeGuard {
438    /// The enabled scope of this request.
439    pub fn scope(&self) -> &Scope {
440        &self.scope
441    }
442}
443
444impl Drop for ScopeGuard {
445    fn drop(&mut self) {
446        self.owner.close();
447    }
448}
449
450/// A cache instance: request, process-local and remote layers behind explicit enablement.
451///
452/// Instances are cheap to clone and share one state.
453#[derive(Clone)]
454pub struct DialCache {
455    pub(crate) core: Arc<Core>,
456}
457
458impl std::fmt::Debug for DialCache {
459    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
460        f.debug_struct("DialCache")
461            .field("namespace", &self.core.namespace)
462            .finish_non_exhaustive()
463    }
464}
465
466impl DialCache {
467    /// Start configuring an instance with the defaults documented on
468    /// [`DialCacheBuilder`]'s setters.
469    pub fn builder() -> DialCacheBuilder {
470        DialCacheBuilder::new()
471    }
472
473    /// The instance namespace.
474    pub fn namespace(&self) -> &str {
475        &self.core.namespace
476    }
477
478    /// Open the outermost enabled scope and hold it in a guard. The scope
479    /// closes when the guard drops; retained [`Scope`] clones then pass through.
480    ///
481    /// ```ignore
482    /// let request = cache.enable_guard();
483    /// let name = display_name.get(request.scope(), id).await?;
484    /// ```
485    pub fn enable_guard(&self) -> ScopeGuard {
486        let owner = Owner::new();
487        let scope = Scope {
488            cache_id: self.core.id,
489            owner: Some(owner.clone()),
490            enabled: true,
491        };
492        ScopeGuard { scope, owner }
493    }
494
495    /// Open the outermost enabled scope for `f`. The scope closes when the
496    /// returned future completes, is dropped before completion, or unwinds;
497    /// retained clones then pass through.
498    pub async fn enable<F, Fut, R>(&self, f: F) -> R
499    where
500        F: FnOnce(Scope) -> Fut,
501        Fut: Future<Output = R>,
502    {
503        let guard = self.enable_guard();
504        let scope = guard.scope().clone();
505        let result = f(scope).await;
506        drop(guard);
507        result
508    }
509
510    /// Enable caching inside `parent`, reusing its live outer request memo.
511    /// When `parent` has no live outer scope, this opens a new one.
512    pub async fn enable_in<F, Fut, R>(&self, parent: &Scope, f: F) -> R
513    where
514        F: FnOnce(Scope) -> Fut,
515        Fut: Future<Output = R>,
516    {
517        let live = if parent.cache_id == self.core.id {
518            parent.live_owner()
519        } else {
520            None
521        };
522        match live {
523            Some(owner) => {
524                let scope = Scope {
525                    cache_id: self.core.id,
526                    owner: Some(owner),
527                    enabled: true,
528                };
529                f(scope).await
530            }
531            None => self.enable(f).await,
532        }
533    }
534
535    /// Disable caching inside `parent` while preserving its outer request memo
536    /// for nested re-enablement.
537    pub async fn disable_in<F, Fut, R>(&self, parent: &Scope, f: F) -> R
538    where
539        F: FnOnce(Scope) -> Fut,
540        Fut: Future<Output = R>,
541    {
542        let owner = if parent.cache_id == self.core.id {
543            parent.owner.clone()
544        } else {
545            None
546        };
547        let scope = Scope {
548            cache_id: self.core.id,
549            owner,
550            enabled: false,
551        };
552        f(scope).await
553    }
554
555    /// Whether calls made with `scope` on this instance use caching.
556    pub fn is_enabled(&self, scope: &Scope) -> bool {
557        scope.cache_id == self.core.id && scope.is_enabled()
558    }
559
560    /// Process-scoped single-flight state of this instance.
561    pub fn coalescing_state(&self) -> CoalescingState {
562        let now = self.core.clock.elapsed();
563        let state = self.core.state.lock();
564        let mut process = ProcessCoalescingState {
565            active_leaders: state.flights.len(),
566            ..Default::default()
567        };
568        for flight in state.flights.values() {
569            process.active_followers += flight.followers();
570            let age = now
571                .saturating_sub(flight.started)
572                .as_millis()
573                .min(u64::MAX as u128) as u64;
574            if process
575                .oldest_leader_age_ms
576                .is_none_or(|oldest| age > oldest)
577            {
578                process.oldest_leader_age_ms = Some(age);
579            }
580        }
581        CoalescingState { process }
582    }
583
584    /// Advance the invalidation watermark of every tracked variant of one
585    /// entity, in this instance's namespace. Call it after the source
586    /// mutation commits. Requires a remote adapter; failures are returned.
587    /// IDs use the same [`IntoKeyId`] conversion as [`crate::KeySpec::new`].
588    pub async fn invalidate(
589        &self,
590        key_type: &str,
591        id: impl IntoKeyId,
592        future_buffer_ms: u64,
593    ) -> Result<(), Error> {
594        let identity = Identity::new(key_type, id.into_key_id(), WATERMARK_USE_CASE)
595            .tracked(true)
596            .namespace(self.core.namespace.to_string());
597        self.invalidate_identity(identity, future_buffer_ms).await
598    }
599
600    /// Advance the watermark for the entity named by `identity` after its
601    /// source mutation commits. An empty namespace inherits this instance's;
602    /// an explicit namespace is preserved, exactly as in `get_or_load`.
603    ///
604    /// All use cases and argument variants share the entity's watermark.
605    /// `tracked` is ignored: this operation always invalidates tracked remote
606    /// values. Local entries and already acquired snapshots retain their lifetimes.
607    /// Requires a remote adapter; mutation failures are returned to the caller.
608    pub async fn invalidate_identity(
609        &self,
610        mut identity: Identity,
611        future_buffer_ms: u64,
612    ) -> Result<(), Error> {
613        let core = &self.core;
614        if future_buffer_ms > MAX_SUPPORTED_DURATION_MS {
615            return Err(ConfigError::invalid(
616                "DialCache invalidation future buffer must be no greater than 365 days",
617            )
618            .into());
619        }
620        if identity.namespace.is_empty() {
621            identity.namespace = core.namespace.to_string();
622        }
623        identity.tracked = true;
624        let namespace: Arc<str> = Arc::from(identity.namespace.as_str());
625        let key_type: Arc<str> = Arc::from(identity.key_type.as_str());
626        core.emit(Event::Invalidation {
627            namespace: namespace.clone(),
628            key_type: key_type.clone(),
629            layer: Layer::Remote,
630        });
631        let result: Result<(), Error> = async {
632            let remote = core.remote.clone().ok_or(Error::MissingRemote)?;
633            let keys = identity
634                .keys()
635                .map_err(|e| Error::Config(ConfigError::invalid(e.to_string())))?;
636            let watermark_key = keys.watermark.ok_or_else(|| {
637                Error::Config(ConfigError::invalid(
638                    "tracked identity has no watermark key",
639                ))
640            })?;
641            let now = core.clock.wall_ms();
642            if now < 0 || now as u64 > MAX_SAFE_INTEGER - future_buffer_ms {
643                return Err(ConfigError::invalid(
644                    "DialCache invalidation timestamp is outside the safe integer domain",
645                )
646                .into());
647            }
648            let request = InvalidateRequest {
649                watermark_key,
650                invalidated_at_ms: now as u64,
651                future_buffer_ms,
652            };
653            let pending: Settled<Result<(), Error>> = start_pending(
654                core.runtime.as_ref(),
655                async move {
656                    remote
657                        .invalidate(request)
658                        .await
659                        .map_err(|e| Error::Remote(Arc::from(e)))
660                },
661                |message| Err(Error::Panic(message)),
662            );
663            pending.wait().await
664        }
665        .await;
666        if let Err(error) = &result {
667            core.log(LogEvent::InvalidationFailed(error.to_string().into()));
668            core.emit(Event::Error {
669                labels: Labels {
670                    namespace,
671                    use_case: Arc::from(WATERMARK_USE_CASE),
672                    key_type,
673                    layer: Layer::Remote,
674                },
675                error: crate::observe::ErrorKind::Invalidation,
676                in_fallback: false,
677            });
678        }
679        result
680    }
681
682    /// Execute one inline cached call.
683    ///
684    /// `load` is the source of truth; it may be invoked again later by
685    /// served-hit shadow validation, so it must be reusable. The returned
686    /// value is shared by reference: treat it as immutable.
687    pub async fn get_or_load<T, F, Fut>(
688        &self,
689        scope: &Scope,
690        operation: Operation<T>,
691        load: F,
692    ) -> Result<Arc<T>, Error>
693    where
694        T: Send + Sync + 'static,
695        F: Fn(Scope) -> Fut + Send + Sync + 'static,
696        Fut: Future<Output = Result<T, BoxError>> + Send + 'static,
697    {
698        let (identity, metadata) = operation.erase();
699        let erased = ErasedOperation {
700            identity,
701            identity_provider: None,
702            metadata,
703            load: erase_load(load),
704        };
705        let value = self.execute(scope, erased).await?;
706        downcast_value::<T>(value)
707    }
708
709    /// Validate static operation configuration, then run the execution as
710    /// detached work whose result every caller awaits. Dropping the returned
711    /// future does not cancel the execution: sources and publications keep
712    /// their contracts, as in the reference implementations.
713    pub(crate) async fn execute(
714        &self,
715        scope: &Scope,
716        mut operation: ErasedOperation,
717    ) -> ValueResult {
718        validate_operation(&operation)?;
719        if operation.identity.namespace.is_empty() {
720            operation.identity.namespace = self.core.namespace.to_string();
721        }
722        // A scope owned by another instance disables this call but still
723        // reaches the source, so nested calls on its own instance keep caching.
724        let core = self.core.clone();
725        let scope = scope.clone();
726        let pending: Settled<ValueResult> = start_pending(
727            core.runtime.clone().as_ref(),
728            crate::execution::run(core, scope, Arc::new(operation)),
729            |message| Err(Error::Panic(message)),
730        );
731        pending.wait().await
732    }
733
734    pub(crate) fn register_use_case(&self, use_case: &str) -> Result<(), ConfigError> {
735        if use_case == WATERMARK_USE_CASE {
736            return Err(ConfigError::ReservedUseCase(use_case.to_string()));
737        }
738        let mut state = self.core.state.lock();
739        if !state.registered.insert(use_case.to_string()) {
740            return Err(ConfigError::UseCaseAlreadyRegistered(use_case.to_string()));
741        }
742        Ok(())
743    }
744}
745
746pub(crate) fn validate_operation(operation: &ErasedOperation) -> Result<(), Error> {
747    operation
748        .metadata
749        .policy
750        .validate()
751        .map_err(|e| ConfigError::invalid(e.to_string()))?;
752    if let crate::operation::SourceBudget::Millis(ms) = operation.metadata.budget {
753        if !(1..=MAX_DEADLINE_MS).contains(&ms) {
754            return Err(ConfigError::invalid(format!(
755                "DialCache source budget must be a positive integer no greater than {MAX_DEADLINE_MS} ms"
756            ))
757            .into());
758        }
759    }
760    if operation.identity.use_case == WATERMARK_USE_CASE {
761        return Err(ConfigError::ReservedUseCase(operation.identity.use_case.clone()).into());
762    }
763    Ok(())
764}
765
766pub(crate) fn outcome_labels(identity: &Identity) -> OutcomeLabels {
767    OutcomeLabels {
768        namespace: Arc::from(identity.namespace.as_str()),
769        use_case: Arc::from(identity.use_case.as_str()),
770        key_type: Arc::from(identity.key_type.as_str()),
771    }
772}
dialcache/engine.rs

dialcache/
engine.rs
