graphrefly/
cascading_cache.rs

1//! Graph-layer reactive cascading cache helper (D104/D105/D107/D123).
2//!
3//! The passive read-through algorithm stays in [`crate::storage`]. This module
4//! wraps it as visible graph topology: request/policy/invalidate deps drive an
5//! events node, and status/value are derived from those events.
6
7use std::collections::BTreeMap;
8use std::panic::{catch_unwind, AssertUnwindSafe};
9use std::rc::Rc;
10
11use crate::graph::{Graph, GraphNodeOpts};
12use crate::node::{Node, NodeOpts};
13use crate::protocol::{AnyValue, Message};
14use crate::storage::{
15    tiered_read_through, KvStorageTier, PromotionPolicy, ReadThroughErrorStage,
16    ReadThroughLookupTier, ReadThroughOutcome, StorageError, StorageResult,
17    TieredReadThroughOptions, TieredReadThroughResult, TieredReadThroughStatus,
18};
19
20/// Dynamic promotion policy for [`reactive_cascading_cache`].
21#[derive(Debug, Clone, PartialEq, Eq, Default)]
22pub struct CascadingCachePolicy {
23    /// `promote_to` field for promote to.
24    pub promote_to: Option<PromotionPolicy>,
25}
26
27/// Visible cache status emitted by [`ReactiveCascadingCache::status`].
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub enum CascadingCacheStatus {
30    /// `Idle` variant.
31    Idle,
32    /// `Loading` variant.
33    Loading {
34        /// `key` field for key.
35        key: String,
36        /// `request_seq` field for request seq.
37        request_seq: u64,
38    },
39    /// `Hit` variant.
40    Hit {
41        /// `key` field for key.
42        key: String,
43        /// `request_seq` field for request seq.
44        request_seq: u64,
45        /// `tier` field for tier.
46        tier: Option<ReadThroughLookupTier>,
47    },
48    /// `Miss` variant.
49    Miss {
50        /// `key` field for key.
51        key: String,
52        /// `request_seq` field for request seq.
53        request_seq: u64,
54    },
55    /// `Error` variant.
56    Error {
57        /// `key` field for key.
58        key: String,
59        /// `request_seq` field for request seq.
60        request_seq: u64,
61        /// `error` field for error.
62        error: StorageError,
63    },
64}
65
66/// Visible cache facts emitted by [`ReactiveCascadingCache::events`].
67#[derive(Debug, Clone, PartialEq)]
68pub enum CascadingCacheEvent<V> {
69    /// `Request` variant.
70    Request {
71        /// `key` field for key.
72        key: String,
73        /// `request_seq` field for request seq.
74        request_seq: u64,
75    },
76    /// `Invalidate` variant.
77    Invalidate {
78        /// `key` field for key.
79        key: String,
80        /// `request_seq` field for request seq.
81        request_seq: u64,
82    },
83    /// `Lookup` variant.
84    Lookup {
85        /// `key` field for key.
86        key: String,
87        /// `request_seq` field for request seq.
88        request_seq: u64,
89        /// `outcome` field for outcome.
90        outcome: ReadThroughOutcome,
91        /// `tier` field for tier.
92        tier: ReadThroughLookupTier,
93        /// `value` field for value.
94        value: Option<V>,
95        /// `error` field for error.
96        error: Option<StorageError>,
97    },
98    /// `Promotion` variant.
99    Promotion {
100        /// `key` field for key.
101        key: String,
102        /// `request_seq` field for request seq.
103        request_seq: u64,
104        /// `tier` field for tier.
105        tier: ReadThroughLookupTier,
106        /// `ok` field for ok.
107        ok: bool,
108        /// `error` field for error.
109        error: Option<StorageError>,
110    },
111    /// `Fill` variant.
112    Fill {
113        /// `key` field for key.
114        key: String,
115        /// `request_seq` field for request seq.
116        request_seq: u64,
117        /// `status` field for status.
118        status: TieredReadThroughStatus,
119        /// `value` field for value.
120        value: Option<V>,
121        /// `tier` field for tier.
122        tier: Option<ReadThroughLookupTier>,
123        /// `error` field for error.
124        error: Option<StorageError>,
125    },
126    /// `Error` variant.
127    Error {
128        /// `key` field for key.
129        key: String,
130        /// `request_seq` field for request seq.
131        request_seq: u64,
132        /// `stage` field for stage.
133        stage: ReadThroughErrorStage,
134        /// `tier` field for tier.
135        tier: Option<ReadThroughLookupTier>,
136        /// `error` field for error.
137        error: StorageError,
138    },
139}
140
141/// `ReactiveCascadingCacheLoadFn` type alias.
142pub type ReactiveCascadingCacheLoadFn<V> = dyn Fn(&str) -> StorageResult<Option<V>>;
143
144/// Options for the Rust graph-layer cache factory.
145pub struct ReactiveCascadingCacheOptions<V: Clone + 'static> {
146    /// `request` field for request.
147    pub request: Node<String>,
148    /// `policy` field for policy.
149    pub policy: Option<Node<CascadingCachePolicy>>,
150    /// `invalidate` field for invalidate.
151    pub invalidate: Option<Node<String>>,
152    /// `tiers` field for tiers.
153    pub tiers: Vec<Rc<dyn KvStorageTier<V>>>,
154    /// `load` field for load.
155    pub load: Option<Rc<ReactiveCascadingCacheLoadFn<V>>>,
156    /// `tier_names` field for tier names.
157    pub tier_names: Vec<String>,
158    /// `promote_to` field for promote to.
159    pub promote_to: PromotionPolicy,
160    /// `name` field for name.
161    pub name: Option<String>,
162    /// `meta` field for meta.
163    pub meta: BTreeMap<String, String>,
164}
165
166impl<V: Clone + 'static> ReactiveCascadingCacheOptions<V> {
167    /// Creates or computes `new`.
168    pub fn new(request: Node<String>, tiers: Vec<Rc<dyn KvStorageTier<V>>>) -> Self {
169        Self {
170            request,
171            policy: None,
172            invalidate: None,
173            tiers,
174            load: None,
175            tier_names: Vec::new(),
176            promote_to: PromotionPolicy::Disabled,
177            name: None,
178            meta: BTreeMap::new(),
179        }
180    }
181}
182
183/// Graph-visible bundle returned by [`reactive_cascading_cache`].
184pub struct ReactiveCascadingCache<V: Clone + 'static> {
185    /// `value` field for value.
186    pub value: Node<V>,
187    /// `status` field for status.
188    pub status: Node<CascadingCacheStatus>,
189    /// `events` field for events.
190    pub events: Node<CascadingCacheEvent<V>>,
191}
192
193#[derive(Debug, Clone, Default)]
194struct DriverState {
195    seq: u64,
196    latest_key: Option<String>,
197}
198
199#[derive(Debug, Clone, Default)]
200struct SeqState {
201    seq: u64,
202}
203
204/// Free-standing graph-layer cascading cache factory (D104/D105/D107/D123).
205///
206/// This is not a [`Graph`] method and not an imperative cache object. Request,
207/// optional policy, and optional invalidation deps are declared topology. Storage
208/// promotion defaults to disabled at this graph layer; opt in through
209/// [`ReactiveCascadingCacheOptions::promote_to`] or a policy node.
210pub fn reactive_cascading_cache<V: Clone + 'static>(
211    graph: &Graph,
212    opts: ReactiveCascadingCacheOptions<V>,
213) -> ReactiveCascadingCache<V> {
214    let ReactiveCascadingCacheOptions {
215        request,
216        policy,
217        invalidate,
218        tiers,
219        load,
220        tier_names,
221        promote_to,
222        name,
223        meta,
224    } = opts;
225
226    let base_name = name.unwrap_or_else(|| "reactiveCascadingCache".to_owned());
227    let mut event_deps = vec![request.erased()];
228    let policy_index = policy.as_ref().map(|node| {
229        event_deps.push(node.erased());
230        event_deps.len() - 1
231    });
232    let invalidate_index = invalidate.as_ref().map(|node| {
233        event_deps.push(node.erased());
234        event_deps.len() - 1
235    });
236
237    let events = graph.node_opts::<CascadingCacheEvent<V>, _>(
238        event_deps,
239        {
240            let tiers = tiers.clone();
241            let tier_names = tier_names.clone();
242            let load = load.clone();
243            let base_promote_to = promote_to.clone();
244            move |ctx| {
245                let mut st = ctx
246                    .state_get::<DriverState>()
247                    .map(|s| (*s).clone())
248                    .unwrap_or_default();
249                let current_policy = policy_index
250                    .and_then(|idx| ctx.data::<CascadingCachePolicy>(idx))
251                    .map(|p| (*p).clone());
252                let effective_promote_to = current_policy
253                    .as_ref()
254                    .and_then(|p| p.promote_to.clone())
255                    .unwrap_or_else(|| base_promote_to.clone());
256
257                for key in ctx.batch::<String>(0) {
258                    start_lookup(
259                        ctx,
260                        &mut st,
261                        LookupCause::Request,
262                        (*key).clone(),
263                        LookupInputs {
264                            tiers: &tiers,
265                            tier_names: &tier_names,
266                            load: load.as_ref(),
267                            promote_to: effective_promote_to.clone(),
268                        },
269                    );
270                }
271
272                if let Some(idx) = invalidate_index {
273                    for key in ctx.batch::<String>(idx) {
274                        start_lookup(
275                            ctx,
276                            &mut st,
277                            LookupCause::Invalidate,
278                            (*key).clone(),
279                            LookupInputs {
280                                tiers: &tiers,
281                                tier_names: &tier_names,
282                                load: load.as_ref(),
283                                promote_to: effective_promote_to.clone(),
284                            },
285                        );
286                    }
287                }
288
289                if ctx.batch::<String>(0).is_empty()
290                    && invalidate_index
291                        .map(|idx| ctx.batch::<String>(idx).is_empty())
292                        .unwrap_or(true)
293                    && policy_index
294                        .map(|idx| !ctx.batch::<CascadingCachePolicy>(idx).is_empty())
295                        .unwrap_or(false)
296                {
297                    if let Some(key) = st.latest_key.clone() {
298                        start_lookup(
299                            ctx,
300                            &mut st,
301                            LookupCause::Request,
302                            key,
303                            LookupInputs {
304                                tiers: &tiers,
305                                tier_names: &tier_names,
306                                load: load.as_ref(),
307                                promote_to: effective_promote_to,
308                            },
309                        );
310                    }
311                }
312
313                ctx.state_set(st);
314            }
315        },
316        GraphNodeOpts {
317            name: Some(format!("{base_name}.events")),
318            meta: meta.clone(),
319            node: NodeOpts {
320                partial: true,
321                ..NodeOpts::default()
322            },
323            ..GraphNodeOpts::default()
324        },
325    );
326
327    let status = graph.node_opts_initial::<CascadingCacheStatus, _>(
328        vec![events.erased()],
329        |ctx| {
330            let mut st = ctx
331                .state_get::<SeqState>()
332                .map(|s| (*s).clone())
333                .unwrap_or_default();
334            for event in ctx.batch::<CascadingCacheEvent<V>>(0) {
335                match event.as_ref() {
336                    CascadingCacheEvent::Request { key, request_seq }
337                    | CascadingCacheEvent::Invalidate { key, request_seq } => {
338                        st.seq = *request_seq;
339                        ctx.emit(CascadingCacheStatus::Loading {
340                            key: key.clone(),
341                            request_seq: *request_seq,
342                        });
343                    }
344                    CascadingCacheEvent::Fill {
345                        key,
346                        request_seq,
347                        status,
348                        tier,
349                        error,
350                        ..
351                    } if *request_seq == st.seq => match status {
352                        TieredReadThroughStatus::Hit => ctx.emit(CascadingCacheStatus::Hit {
353                            key: key.clone(),
354                            request_seq: *request_seq,
355                            tier: tier.clone(),
356                        }),
357                        TieredReadThroughStatus::Miss => ctx.emit(CascadingCacheStatus::Miss {
358                            key: key.clone(),
359                            request_seq: *request_seq,
360                        }),
361                        TieredReadThroughStatus::Error => ctx.emit(CascadingCacheStatus::Error {
362                            key: key.clone(),
363                            request_seq: *request_seq,
364                            error: error
365                                .clone()
366                                .unwrap_or_else(|| StorageError::backend("read-through failed")),
367                        }),
368                    },
369                    CascadingCacheEvent::Error {
370                        key,
371                        request_seq,
372                        error,
373                        ..
374                    } if *request_seq == st.seq => ctx.emit(CascadingCacheStatus::Error {
375                        key: key.clone(),
376                        request_seq: *request_seq,
377                        error: error.clone(),
378                    }),
379                    _ => {}
380                }
381            }
382            ctx.state_set(st);
383        },
384        GraphNodeOpts {
385            name: Some(format!("{base_name}.status")),
386            meta: meta.clone(),
387            node: NodeOpts {
388                partial: true,
389                ..NodeOpts::default()
390            },
391            ..GraphNodeOpts::default()
392        },
393        Some(CascadingCacheStatus::Idle),
394    );
395
396    let value = graph.node_opts::<V, _>(
397        vec![events.erased()],
398        |ctx| {
399            let mut st = ctx
400                .state_get::<SeqState>()
401                .map(|s| (*s).clone())
402                .unwrap_or_default();
403            for event in ctx.batch::<CascadingCacheEvent<V>>(0) {
404                match event.as_ref() {
405                    CascadingCacheEvent::Request { request_seq, .. } => {
406                        st.seq = *request_seq;
407                    }
408                    CascadingCacheEvent::Invalidate { request_seq, .. } => {
409                        st.seq = *request_seq;
410                        ctx.down(vec![Message::Invalidate]);
411                    }
412                    CascadingCacheEvent::Fill {
413                        request_seq,
414                        status,
415                        value,
416                        ..
417                    } if *request_seq == st.seq => {
418                        if *status == TieredReadThroughStatus::Hit {
419                            if let Some(value) = value.clone() {
420                                ctx.emit(value);
421                            } else {
422                                ctx.down(vec![Message::Invalidate]);
423                            }
424                        } else {
425                            ctx.down(vec![Message::Invalidate]);
426                        }
427                    }
428                    _ => {}
429                }
430            }
431            ctx.state_set(st);
432        },
433        GraphNodeOpts {
434            name: Some(format!("{base_name}.value")),
435            meta,
436            node: NodeOpts {
437                partial: true,
438                ..NodeOpts::default()
439            },
440            ..GraphNodeOpts::default()
441        },
442    );
443
444    ReactiveCascadingCache {
445        value,
446        status,
447        events,
448    }
449}
450
451#[derive(Clone, Copy)]
452enum LookupCause {
453    Request,
454    Invalidate,
455}
456
457struct LookupInputs<'a, V: Clone + 'static> {
458    tiers: &'a [Rc<dyn KvStorageTier<V>>],
459    tier_names: &'a [String],
460    load: Option<&'a Rc<ReactiveCascadingCacheLoadFn<V>>>,
461    promote_to: PromotionPolicy,
462}
463
464fn start_lookup<V: Clone + 'static>(
465    ctx: &crate::ctx::Ctx,
466    st: &mut DriverState,
467    cause: LookupCause,
468    key: String,
469    inputs: LookupInputs<'_, V>,
470) {
471    st.seq += 1;
472    let request_seq = st.seq;
473    st.latest_key = Some(key.clone());
474    ctx.state_set(st.clone());
475    let start_event: CascadingCacheEvent<V> = match cause {
476        LookupCause::Request => CascadingCacheEvent::Request {
477            key: key.clone(),
478            request_seq,
479        },
480        LookupCause::Invalidate => CascadingCacheEvent::Invalidate {
481            key: key.clone(),
482            request_seq,
483        },
484    };
485    emit_event(ctx, start_event);
486
487    let tier_refs = inputs
488        .tiers
489        .iter()
490        .map(|tier| tier.as_ref())
491        .collect::<Vec<&dyn KvStorageTier<V>>>();
492    let mut read_opts = TieredReadThroughOptions::new(key.clone(), tier_refs);
493    read_opts.tier_names = inputs.tier_names.to_vec();
494    read_opts.promote_to = inputs.promote_to;
495    if let Some(load) = inputs.load {
496        let load = load.clone();
497        read_opts.load = Some(Box::new(move |key| load(key)));
498    }
499    let messages = match catch_unwind(AssertUnwindSafe(|| tiered_read_through(read_opts))) {
500        Ok(result) => events_from_result(key, request_seq, result),
501        Err(payload) => events_from_error::<V>(
502            key,
503            request_seq,
504            StorageError::backend(panic_payload(payload)),
505        ),
506    };
507    let current = ctx
508        .state_get::<DriverState>()
509        .map(|state| (*state).clone())
510        .unwrap_or_else(|| st.clone());
511    if current.seq != request_seq {
512        *st = current;
513        return;
514    }
515    if !messages.is_empty() {
516        ctx.down(messages);
517    }
518    if let Some(current) = ctx.state_get::<DriverState>().map(|state| (*state).clone()) {
519        *st = current;
520    }
521}
522
523fn emit_event<V: Clone + 'static>(ctx: &crate::ctx::Ctx, event: CascadingCacheEvent<V>) {
524    let value: AnyValue = Rc::new(event);
525    ctx.down(vec![Message::Data(value)]);
526}
527
528fn events_from_result<V: Clone + 'static>(
529    key: String,
530    request_seq: u64,
531    result: TieredReadThroughResult<V>,
532) -> Vec<Message<AnyValue>> {
533    let mut events = Vec::new();
534    let mut first_error = None;
535    for fact in result.facts {
536        let error = fact.error.clone();
537        let lookup_event = CascadingCacheEvent::Lookup {
538            key: key.clone(),
539            request_seq,
540            outcome: fact.outcome.clone(),
541            tier: fact.tier.clone(),
542            value: fact.value.clone(),
543            error: error.clone(),
544        };
545        events.push(data_msg(lookup_event));
546        if fact.outcome == ReadThroughOutcome::Error {
547            if let Some(error) = error {
548                first_error.get_or_insert_with(|| error.clone());
549                events.push(data_msg::<V>(CascadingCacheEvent::Error {
550                    key: key.clone(),
551                    request_seq,
552                    stage: ReadThroughErrorStage::Lookup,
553                    tier: Some(fact.tier),
554                    error,
555                }));
556            }
557        }
558    }
559    for promotion in result.promotions {
560        let error = promotion.error.clone();
561        events.push(data_msg::<V>(CascadingCacheEvent::Promotion {
562            key: key.clone(),
563            request_seq,
564            tier: promotion.tier.clone(),
565            ok: promotion.ok,
566            error: error.clone(),
567        }));
568        if !promotion.ok {
569            if let Some(error) = error {
570                first_error.get_or_insert_with(|| error.clone());
571                events.push(data_msg::<V>(CascadingCacheEvent::Error {
572                    key: key.clone(),
573                    request_seq,
574                    stage: ReadThroughErrorStage::Promotion,
575                    tier: Some(promotion.tier),
576                    error,
577                }));
578            }
579        }
580    }
581    events.push(data_msg(CascadingCacheEvent::Fill {
582        key,
583        request_seq,
584        status: result.status,
585        value: result.value,
586        tier: result.hit_tier,
587        error: first_error,
588    }));
589    events
590}
591
592fn events_from_error<V: Clone + 'static>(
593    key: String,
594    request_seq: u64,
595    error: StorageError,
596) -> Vec<Message<AnyValue>> {
597    vec![
598        data_msg::<V>(CascadingCacheEvent::Error {
599            key: key.clone(),
600            request_seq,
601            stage: ReadThroughErrorStage::Lookup,
602            tier: None,
603            error: error.clone(),
604        }),
605        data_msg::<V>(CascadingCacheEvent::Fill {
606            key,
607            request_seq,
608            status: TieredReadThroughStatus::Error,
609            value: None,
610            tier: None,
611            error: Some(error),
612        }),
613    ]
614}
615
616fn data_msg<V: Clone + 'static>(event: CascadingCacheEvent<V>) -> Message<AnyValue> {
617    Message::Data(Rc::new(event))
618}
619
620fn panic_payload(payload: Box<dyn std::any::Any + Send>) -> String {
621    if let Some(s) = payload.downcast_ref::<&str>() {
622        (*s).to_owned()
623    } else if let Some(s) = payload.downcast_ref::<String>() {
624        s.clone()
625    } else {
626        "reactive_cascading_cache read-through panicked".to_owned()
627    }
628}
graphrefly/cascading_cache.rs

graphrefly/
cascading_cache.rs
