dialcache/
operation.rs

1//! Operations: the typed description of one cached call and its erased form.
2
3use std::any::TypeId;
4use std::sync::Arc;
5
6use futures::future::BoxFuture;
7use serde::de::DeserializeOwned;
8use serde::Serialize;
9use serde_json::Value;
10
11use crate::codec::{Codec, JsonCodec, Payload};
12use crate::error::{BoxError, Error};
13use crate::identity::Identity;
14use crate::limits::DEFAULT_FALLBACK_TIMEOUT_MS;
15use crate::local::StoredValue;
16use crate::policy::Policy;
17use crate::scope::Scope;
18use crate::shadow_log::{JsonRender, ValueRenderer};
19
20/// The source deadline of one operation.
21#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
22pub enum SourceBudget {
23    /// The library default, 60,000 ms.
24    #[default]
25    Default,
26    /// No deadline: the caller waits for the source however long it takes.
27    /// Shadow work still uses the default budget.
28    Unbounded,
29    /// A deadline in whole milliseconds, `1..=2_147_483_647`.
30    Millis(u64),
31}
32
33impl SourceBudget {
34    pub(crate) fn millis(self) -> Option<u64> {
35        match self {
36            SourceBudget::Default => Some(DEFAULT_FALLBACK_TIMEOUT_MS),
37            SourceBudget::Unbounded => None,
38            SourceBudget::Millis(ms) => Some(ms),
39        }
40    }
41}
42
43/// Decides whether a source failure may be answered from the stale value
44/// retained by the initial remote read.
45pub type RecoveryPredicate = Arc<dyn Fn(&Error) -> bool + Send + Sync>;
46/// Application equality used by shadow validation.
47pub type Comparator<T> = Arc<dyn Fn(&T, &T) -> bool + Send + Sync>;
48/// Renders the text a confirmed-mismatch warning logs for a value when
49/// `mismatch_logging.value` is on, in place of the value's JSON. `None`, or a
50/// panic, logs the field as unavailable with a `render_error` outcome. A
51/// preview you set replaces the JSON for the whole warning, so the built-in
52/// difference computation never reads what it leaves out: difference fields
53/// then come only from a [`MismatchDiffHook`]. Runs through
54/// [`Runtime::spawn_blocking`](crate::Runtime::spawn_blocking) after mismatch
55/// confirmation; callbacks may run on a CPU worker thread.
56pub type Preview<T> = Arc<dyn Fn(&T) -> Option<String> + Send + Sync>;
57/// Replaces the built-in difference computation of confirmed-mismatch
58/// warnings for one operation: receives the compared cached and source values
59/// and returns the differences. Like the comparator it must be deterministic,
60/// side-effect-free and bounded; a panic logs the difference fields as
61/// unavailable with a `hook_error` outcome and never falls back to the
62/// built-in computation. Runs through
63/// [`Runtime::spawn_blocking`](crate::Runtime::spawn_blocking) after mismatch
64/// confirmation, and only when `mismatch_logging.diff` is above `off`.
65pub type MismatchDiffHook<T> = Arc<dyn Fn(&T, &T) -> Vec<MismatchEntry> + Send + Sync>;
66
67/// One difference in a confirmed-mismatch warning: an RFC 6901 JSON Pointer
68/// into the compared value and, optionally, what each side holds there. A
69/// side is `None` when the member exists on the other side only, or when the
70/// hook reports where the values differ but not what they hold.
71#[derive(Debug, Clone, PartialEq, Serialize)]
72pub struct MismatchEntry {
73    /// The pointer, such as `/profile/email`; `""` is the whole value.
74    pub path: String,
75    /// The cached side at the pointer.
76    #[serde(skip_serializing_if = "Option::is_none")]
77    pub cached: Option<Value>,
78    /// The source side at the pointer.
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub source: Option<Value>,
81}
82
83/// One inline cached call: its identity, static policy, budget and codecs.
84///
85/// Use it directly with
86/// [`DialCache::get_or_load`](crate::DialCache::get_or_load) when a stable use
87/// case is declared at the call site instead of being registered.
88pub struct Operation<T> {
89    /// The logical identity; an empty namespace inherits the instance's.
90    pub identity: Identity,
91    /// Static caching policy, validated before execution.
92    pub policy: Policy,
93    /// The source deadline.
94    pub budget: SourceBudget,
95    /// Encodes and decodes values for the remote layer.
96    pub codec: Arc<dyn Codec<T>>,
97    /// Application equality for shadow validation.
98    pub comparator: Comparator<T>,
99    /// Per-operation stale-recovery classifier. `None` defers to the
100    /// instance's, which by default admits only the source deadline error.
101    pub should_recover: Option<RecoveryPredicate>,
102    /// Renders the value fields of mismatch warnings; see [`Preview`].
103    /// [`new`](Operation::new) sets the values' JSON rendering, which also
104    /// feeds the built-in difference computation. A closure you set replaces
105    /// that JSON for the whole warning, so difference fields then come only
106    /// from a [`mismatch_diff`](Self::mismatch_diff) hook. `None` logs no
107    /// values and leaves the operation without a JSON rendering, as one built
108    /// with [`with_codec`](Operation::with_codec) is: a requested value field
109    /// is unavailable with a `render_error` outcome, and so are difference
110    /// fields without a hook.
111    pub preview: Option<Preview<T>>,
112    /// Replaces the built-in difference computation of mismatch warnings.
113    /// `None` runs the built-in walk over the values' JSON, which only an
114    /// operation built with [`new`](Operation::new) that keeps its default
115    /// [`preview`](Self::preview) has.
116    pub mismatch_diff: Option<MismatchDiffHook<T>>,
117    /// The JSON rendering captured where `T: Serialize` is known, and the
118    /// default preview that stands for it.
119    pub(crate) json: Option<JsonDefault<T>>,
120}
121
122/// The JSON rendering of an operation built where `T: Serialize` is known,
123/// with the default [`Preview`] installed for it. The rendering applies while
124/// `Operation::preview` still holds that preview, recognized by `Arc`
125/// identity: a closure set in its place replaces the JSON, and `None` removes
126/// it.
127pub(crate) struct JsonDefault<T> {
128    render: JsonRender,
129    preview: Preview<T>,
130}
131
132impl<T: Serialize + Send + Sync + 'static> JsonDefault<T> {
133    pub(crate) fn new() -> Self {
134        JsonDefault {
135            render: JsonRender::of::<T>(),
136            preview: Arc::new(crate::preview::json_preview::<T>),
137        }
138    }
139}
140
141impl<T> JsonDefault<T> {
142    /// The default preview, to install as `Operation::preview`.
143    pub(crate) fn preview(&self) -> Preview<T> {
144        self.preview.clone()
145    }
146}
147
148impl<T> Clone for JsonDefault<T> {
149    fn clone(&self) -> Self {
150        JsonDefault {
151            render: self.render,
152            preview: self.preview.clone(),
153        }
154    }
155}
156
157impl<T> Clone for Operation<T> {
158    fn clone(&self) -> Self {
159        Self {
160            identity: self.identity.clone(),
161            policy: self.policy.clone(),
162            budget: self.budget,
163            codec: self.codec.clone(),
164            comparator: self.comparator.clone(),
165            should_recover: self.should_recover.clone(),
166            preview: self.preview.clone(),
167            mismatch_diff: self.mismatch_diff.clone(),
168            json: self.json.clone(),
169        }
170    }
171}
172
173impl<T> std::fmt::Debug for Operation<T> {
174    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
175        f.debug_struct("Operation")
176            .field("identity", &self.identity)
177            .field("policy", &self.policy)
178            .field("budget", &self.budget)
179            .finish_non_exhaustive()
180    }
181}
182
183impl<T: Serialize + DeserializeOwned + PartialEq + Send + Sync + 'static> Operation<T> {
184    /// An operation using the JSON codec, `PartialEq` shadow comparison and
185    /// the values' JSON for mismatch warnings: its default
186    /// [`preview`](Self::preview) logs that JSON as the value fields, and the
187    /// built-in difference computation walks it. The JSON rendering stays
188    /// when the codec or comparator is replaced afterwards; a preview you set
189    /// replaces it, and `None` in place of the default removes it.
190    pub fn new(identity: Identity) -> Self {
191        let json = JsonDefault::new();
192        Operation {
193            identity,
194            policy: Policy::default(),
195            budget: SourceBudget::Default,
196            codec: Arc::new(JsonCodec),
197            comparator: Arc::new(|a: &T, b: &T| a == b),
198            should_recover: None,
199            preview: Some(json.preview()),
200            mismatch_diff: None,
201            json: Some(json),
202        }
203    }
204}
205
206impl<T: Send + Sync + 'static> Operation<T> {
207    /// An operation with an explicit codec and comparator, for values that
208    /// are not `serde` JSON values. Mismatch warnings have no JSON rendering:
209    /// their value fields need a [`preview`](Self::preview) and their
210    /// difference fields a [`mismatch_diff`](Self::mismatch_diff) hook, and a
211    /// field logged without one is unavailable with a `render_error` outcome.
212    /// A JSON renderer for such operations is tracked in
213    /// <https://github.com/lan17/DialCache/issues/245>.
214    pub fn with_codec(
215        identity: Identity,
216        codec: Arc<dyn Codec<T>>,
217        comparator: impl Fn(&T, &T) -> bool + Send + Sync + 'static,
218    ) -> Self {
219        Operation {
220            identity,
221            policy: Policy::default(),
222            budget: SourceBudget::Default,
223            codec,
224            comparator: Arc::new(comparator),
225            should_recover: None,
226            preview: None,
227            mismatch_diff: None,
228            json: None,
229        }
230    }
231
232    /// Replace the static policy.
233    pub fn policy(mut self, policy: Policy) -> Self {
234        self.policy = policy;
235        self
236    }
237
238    /// Replace the source deadline.
239    pub fn budget(mut self, budget: SourceBudget) -> Self {
240        self.budget = budget;
241        self
242    }
243
244    /// Set the stale-recovery classifier for this operation, overriding the
245    /// instance's.
246    pub fn should_recover(
247        mut self,
248        predicate: impl Fn(&Error) -> bool + Send + Sync + 'static,
249    ) -> Self {
250        self.should_recover = Some(Arc::new(predicate));
251        self
252    }
253
254    /// Set the text mismatch warnings log for each value in place of its
255    /// JSON; `None` from the closure logs the field as unavailable. Difference
256    /// fields then come only from a [`mismatch_diff`](Self::mismatch_diff)
257    /// hook.
258    pub fn preview(
259        mut self,
260        preview: impl Fn(&T) -> Option<String> + Send + Sync + 'static,
261    ) -> Self {
262        self.preview = Some(Arc::new(preview));
263        self
264    }
265
266    /// Replace the built-in difference computation of mismatch warnings; see
267    /// [`MismatchDiffHook`].
268    pub fn mismatch_diff(
269        mut self,
270        hook: impl Fn(&T, &T) -> Vec<MismatchEntry> + Send + Sync + 'static,
271    ) -> Self {
272        self.mismatch_diff = Some(Arc::new(hook));
273        self
274    }
275
276    pub(crate) fn erase(self) -> (Identity, Arc<OperationMetadata>) {
277        let codec = self.codec;
278        let comparator = self.comparator;
279        // The default preview stands for the JSON rendering. Any other
280        // preview replaces it, and `None` leaves the operation without one.
281        let value_renderer = match (self.preview, self.json) {
282            (Some(preview), Some(json)) if Arc::ptr_eq(&preview, &json.preview) => {
283                Some(ValueRenderer::Json(json.render))
284            }
285            (preview, _) => preview.map(|preview| {
286                ValueRenderer::Preview(Arc::new(move |value: &StoredValue| {
287                    value.downcast_ref::<T>().and_then(|v| preview(v))
288                }))
289            }),
290        };
291        let mismatch_diff = self.mismatch_diff;
292        let metadata = OperationMetadata {
293            value_type: TypeId::of::<T>(),
294            policy: self.policy,
295            budget: self.budget,
296            codec: Arc::new(TypedCodec { codec }),
297            compare: Arc::new(move |a: &StoredValue, b: &StoredValue| {
298                let (Some(a), Some(b)) = (a.downcast_ref::<T>(), b.downcast_ref::<T>()) else {
299                    return Err("shadow comparison received a value of another type".into());
300                };
301                Ok(comparator(a, b))
302            }),
303            should_recover: self.should_recover,
304            value_renderer,
305            mismatch_diff: mismatch_diff.map(|hook| {
306                Arc::new(move |cached: &StoredValue, source: &StoredValue| {
307                    // The comparator already accepted both values as `T`.
308                    let typed = "compared values have the operation's type";
309                    hook(
310                        cached.downcast_ref::<T>().expect(typed),
311                        source.downcast_ref::<T>().expect(typed),
312                    )
313                }) as ErasedMismatchDiff
314            }),
315        };
316        (self.identity, Arc::new(metadata))
317    }
318}
319
320/// Source loader over erased values.
321pub(crate) type ErasedLoad =
322    Arc<dyn Fn(Scope) -> BoxFuture<'static, Result<StoredValue, BoxError>> + Send + Sync>;
323pub(crate) type ErasedCompare =
324    Arc<dyn Fn(&StoredValue, &StoredValue) -> Result<bool, BoxError> + Send + Sync>;
325pub(crate) type ErasedPreview = Arc<dyn Fn(&StoredValue) -> Option<String> + Send + Sync>;
326pub(crate) type ErasedMismatchDiff =
327    Arc<dyn Fn(&StoredValue, &StoredValue) -> Vec<MismatchEntry> + Send + Sync>;
328pub(crate) type IdentityProvider = Arc<dyn Fn() -> Result<Identity, BoxError> + Send + Sync>;
329
330/// Immutable adapters and policy, shared by every registered invocation.
331pub(crate) struct OperationMetadata {
332    pub value_type: TypeId,
333    pub policy: Policy,
334    pub budget: SourceBudget,
335    pub codec: Arc<dyn ErasedCodec>,
336    pub compare: ErasedCompare,
337    pub should_recover: Option<RecoveryPredicate>,
338    /// The values' JSON while the operation keeps its default preview, the
339    /// preview set in its place, or `None` without either.
340    pub value_renderer: Option<ValueRenderer>,
341    pub mismatch_diff: Option<ErasedMismatchDiff>,
342}
343
344/// The type-erased operation the engine executes.
345pub(crate) struct ErasedOperation {
346    pub identity: Identity,
347    pub identity_provider: Option<IdentityProvider>,
348    pub metadata: Arc<OperationMetadata>,
349    pub load: ErasedLoad,
350}
351
352/// A codec over erased values.
353pub(crate) trait ErasedCodec: Send + Sync {
354    fn encode(&self, value: StoredValue) -> BoxFuture<'_, Result<Payload, BoxError>>;
355    fn decode(&self, payload: Payload) -> BoxFuture<'_, Result<StoredValue, BoxError>>;
356}
357
358struct TypedCodec<T> {
359    codec: Arc<dyn Codec<T>>,
360}
361
362impl<T: Send + Sync + 'static> ErasedCodec for TypedCodec<T> {
363    fn encode(&self, value: StoredValue) -> BoxFuture<'_, Result<Payload, BoxError>> {
364        match Arc::downcast::<T>(value) {
365            Ok(value) => self.codec.encode_owned(value),
366            Err(_) => Box::pin(std::future::ready(Err(
367                "codec received a value of another type".into(),
368            ))),
369        }
370    }
371
372    fn decode(&self, payload: Payload) -> BoxFuture<'_, Result<StoredValue, BoxError>> {
373        let decoding = self.codec.decode(payload);
374        Box::pin(async move { decoding.await.map(|value| Arc::new(value) as StoredValue) })
375    }
376}
377
378/// Erase a typed source closure.
379pub(crate) fn erase_load<T, F, Fut>(load: F) -> ErasedLoad
380where
381    T: Send + Sync + 'static,
382    F: Fn(Scope) -> Fut + Send + Sync + 'static,
383    Fut: std::future::Future<Output = Result<T, BoxError>> + Send + 'static,
384{
385    Arc::new(move |scope: Scope| {
386        let future = load(scope);
387        Box::pin(async move { future.await.map(|value| Arc::new(value) as StoredValue) })
388    })
389}
390
391/// Downcast an erased value to its typed form.
392pub(crate) fn downcast_value<T: Send + Sync + 'static>(
393    value: StoredValue,
394) -> Result<Arc<T>, Error> {
395    Arc::downcast::<T>(value).map_err(|_| {
396        Error::Config(crate::error::ConfigError::invalid(format!(
397            "cached value is not a {}",
398            std::any::type_name::<T>()
399        )))
400    })
401}
dialcache/operation.rs

dialcache/
operation.rs
