saluki_core/cache/
mod.rs

1//! A concurrent cache with optional expiration and telemetry.
2
3use std::{marker::PhantomData, num::NonZeroUsize, sync::Arc, time::Duration};
4
5use async_trait::async_trait;
6use saluki_common::{hash::FastBuildHasher, sync::shutdown::ShutdownHandle};
7use saluki_error::GenericError;
8use saluki_metrics::{static_metrics, Counter, Gauge, Histogram};
9use tokio::{pin, select, time::sleep};
10use tracing::debug;
11
12use crate::runtime::{self, scope, InitializationError, ScopeGuard, Supervisable, SupervisorFuture};
13
14mod expiry;
15use self::expiry::{Expiration, ExpirationBuilder, ExpiryCapableLifecycle};
16
17pub mod weight;
18use self::weight::{ItemCountWeighter, Weighter, WrappedWeighter};
19
20type RawCache<K, V, W, H> = quick_cache::sync::Cache<K, V, WrappedWeighter<W>, H, ExpiryCapableLifecycle<K>>;
21
22#[static_metrics(prefix = cache, labels(cache_id))]
23#[derive(Clone)]
24struct Telemetry {
25    current_items: Gauge,
26    current_weight: Gauge,
27    weight_limit: Gauge,
28    hits_total: Counter,
29    misses_total: Counter,
30    items_evicted_total: Counter,
31    #[metric(level = debug)]
32    items_inserted_total: Counter,
33    #[metric(level = debug)]
34    items_removed_total: Counter,
35    #[metric(level = debug)]
36    items_expired_total: Counter,
37    #[metric(level = trace)]
38    items_expired_batch_size: Histogram,
39}
40
41struct InnerCache<K, V, W, H> {
42    cache: Arc<RawCache<K, V, W, H>>,
43
44    /// The background tasks of the cache, if it has any.
45    ///
46    /// The guard is kept here, next to the raw cache that the tasks serve. Thus, the tasks stop when the last copy of
47    /// the cache is dropped. They stop earlier if the process that owns them stops first. The tasks hold the raw cache
48    /// but never this guard, so when the cache is dropped, the guard is also dropped.
49    _drivers: Option<ScopeGuard>,
50}
51
52/// Builder for creating a [`Cache`].
53pub struct CacheBuilder<K, V, W = ItemCountWeighter, H = FastBuildHasher> {
54    identifier: String,
55    capacity: NonZeroUsize,
56    weighter: W,
57    idle_period: Option<Duration>,
58    expiration_interval: Option<Duration>,
59    telemetry_enabled: bool,
60    _key: PhantomData<K>,
61    _value: PhantomData<V>,
62    _hasher: PhantomData<H>,
63}
64
65impl<K, V> CacheBuilder<K, V> {
66    /// Creates a new `CacheBuilder` with the given cache identifier.
67    ///
68    /// The cache identifier _should_ be unique, but it isn't required to be. Metrics for the cache will be emitted
69    /// using the given identifier, so in cases where the identifier isn't unique, those metrics will be aggregated
70    /// together and it won't be possible to distinguish between the different caches.
71    ///
72    /// # Errors
73    ///
74    /// If the given cache identifier is empty, an error is returned.
75    pub fn from_identifier<N: Into<String>>(identifier: N) -> Result<CacheBuilder<K, V>, GenericError> {
76        let identifier = identifier.into();
77        if identifier.is_empty() {
78            return Err(GenericError::msg("cache identifier must not be empty"));
79        }
80
81        Ok(CacheBuilder {
82            identifier,
83            capacity: NonZeroUsize::MAX,
84            weighter: ItemCountWeighter,
85            idle_period: None,
86            expiration_interval: None,
87            telemetry_enabled: true,
88            _key: PhantomData,
89            _value: PhantomData,
90            _hasher: PhantomData,
91        })
92    }
93
94    /// Configures a [`CacheBuilder`] that's suitable for tests.
95    ///
96    /// This configures the builder with the following defaults:
97    ///
98    /// - cache identifier of "noop"
99    /// - unlimited cache size
100    /// - telemetry disabled
101    ///
102    /// This is generally only useful for testing purposes, and is exposed publicly in order to be used in cross-crate
103    /// testing scenarios.
104    pub fn for_tests() -> CacheBuilder<K, V> {
105        CacheBuilder::from_identifier("noop")
106            .expect("cache identifier not empty")
107            .with_telemetry(false)
108    }
109}
110
111impl<K, V, W, H> CacheBuilder<K, V, W, H> {
112    /// Sets the capacity for the cache.
113    ///
114    /// The capacity is used, in conjunction with the item weighter, to determine how many items can be held in the
115    /// cache and when items should be evicted to make room for new items.
116    ///
117    /// See [`with_item_weighter`][Self::with_item_weighter] for more information on how the item weighter is used.
118    ///
119    /// Defaults to unlimited capacity.
120    pub fn with_capacity(mut self, capacity: NonZeroUsize) -> Self {
121        self.capacity = capacity;
122        self
123    }
124
125    /// Enables expiration of cached items based on how long since they were last accessed.
126    ///
127    /// Items which haven't been accessed within the configured duration will be marked for expiration, and be removed
128    /// from the cache shortly thereafter. For the purposes of expiration, "accessed" is either when the item was first
129    /// inserted or when it was last read.
130    ///
131    /// If the given value is `None`, expiration is disabled.
132    ///
133    /// Defaults to no expiration.
134    pub fn with_time_to_idle(mut self, idle_period: Option<Duration>) -> Self {
135        self.idle_period = idle_period;
136
137        // Make sure we have an expiration interval set if expiration is enabled.
138        if self.idle_period.is_some() {
139            self.expiration_interval = self.expiration_interval.or(Some(Duration::from_secs(1)));
140        }
141
142        self
143    }
144
145    /// Sets the interval at which the expiration process will run.
146    ///
147    /// This controls how often the expiration process will run to check for expired items. While items become
148    /// _eligible_ for expiration after the configured duration, they're not _guaranteed_ to be
149    /// removed immediately: the expiration process must still run to actually find the expired items and remove them.
150    ///
151    /// This means that the rough upper bound for how long an item may be kept alive is the sum of
152    /// both the configured expiration duration and the expiration interval.
153    ///
154    /// This value is only relevant if expiration is enabled.
155    ///
156    /// Defaults to 1 second.
157    pub fn with_expiration_interval(mut self, expiration_interval: Duration) -> Self {
158        self.expiration_interval = Some(expiration_interval);
159        self
160    }
161
162    /// Sets the item weighter for the cache.
163    ///
164    /// The item weighter is used to determine the "weight" of each item in the cache, which is used during
165    /// insertion/eviction to determine if an item can be held in the cache without first having to evict other items to
166    /// stay within the configured capacity.
167    ///
168    /// For example, if the configured capacity is set to 10,000, and the "item count" weighter is used, then the cache
169    /// will operate in a way that aims to simply ensure that no more than 10,000 items are held in the cache at any given
170    /// time. This allows defining custom weighters that can be used to track other aspects of the items in the cache,
171    /// such as their size in bytes, or some other metric that's relevant to the intended caching behavior.
172    ///
173    /// Defaults to "item count" weighter.
174    pub fn with_item_weighter<W2>(self, weighter: W2) -> CacheBuilder<K, V, W2, H> {
175        CacheBuilder {
176            identifier: self.identifier,
177            capacity: self.capacity,
178            weighter,
179            idle_period: self.idle_period,
180            expiration_interval: self.expiration_interval,
181            telemetry_enabled: self.telemetry_enabled,
182            _key: PhantomData,
183            _value: PhantomData,
184            _hasher: PhantomData,
185        }
186    }
187
188    /// Sets the item hasher for the cache.
189    ///
190    /// As cache keys are hashed before performing any reads or writes, the chosen hasher can potentially impact the
191    /// performance of those operations. In some scenarios, it may be desirable to use a different hasher than the
192    /// default one in order to optimize for specific key types or access patterns.
193    ///
194    /// Defaults to a fast, non-cryptographic hasher: [`FastBuildHasher`].
195    pub fn with_hasher<H2>(self) -> CacheBuilder<K, V, W, H2> {
196        CacheBuilder {
197            identifier: self.identifier,
198            capacity: self.capacity,
199            weighter: self.weighter,
200            idle_period: self.idle_period,
201            expiration_interval: self.expiration_interval,
202            telemetry_enabled: self.telemetry_enabled,
203            _key: PhantomData,
204            _value: PhantomData,
205            _hasher: PhantomData,
206        }
207    }
208
209    /// Sets whether or not to enable telemetry for this cache.
210    ///
211    /// Reporting the telemetry of the cache requires running an asynchronous task to override adding additional
212    /// overhead in the hot path of reading or writing to the cache. In some cases, it may be cumbersome to always
213    /// create the cache in an asynchronous context so that the telemetry task can be spawned. This method allows
214    /// disabling telemetry reporting in those cases.
215    ///
216    /// Defaults to telemetry enabled.
217    pub fn with_telemetry(mut self, enabled: bool) -> Self {
218        self.telemetry_enabled = enabled;
219        self
220    }
221}
222
223impl<K, V, W, H> CacheBuilder<K, V, W, H>
224where
225    K: Eq + std::hash::Hash + Clone + Send + Sync + 'static,
226    V: Clone + Send + Sync + 'static,
227    W: Weighter<K, V> + Clone + Send + Sync + 'static,
228    H: std::hash::BuildHasher + Clone + Default + Send + Sync + 'static,
229{
230    /// Builds a [`Cache`] from the current configuration.
231    ///
232    /// Background tasks drive expiration and telemetry. If a supervised process builds the cache, the tasks run as
233    /// supervised children of that process, and they are restarted if they fail. If other code builds the cache, the
234    /// tasks run as detached tasks. In both cases, the tasks stop when the last copy of the cache is dropped.
235    pub fn build(self) -> Cache<K, V, W, H> {
236        let capacity = self.capacity.get();
237
238        let drivers_name = format!("cache_{}", self.identifier);
239        let telemetry = Telemetry::new(self.identifier);
240        telemetry.weight_limit().set(capacity as f64);
241
242        // Configure expiration if enabled.
243        let eviction_counter = if self.telemetry_enabled {
244            telemetry.items_evicted_total().clone()
245        } else {
246            Counter::noop()
247        };
248        let mut expiration_builder = ExpirationBuilder::new(eviction_counter);
249        if let Some(time_to_idle) = self.idle_period {
250            expiration_builder = expiration_builder.with_time_to_idle(time_to_idle);
251        }
252        let (expiration, expiry_lifecycle) = expiration_builder.build();
253
254        // Create the underlying cache.
255        let raw_cache = Arc::new(RawCache::with(
256            capacity,
257            capacity as u64,
258            WrappedWeighter::from(self.weighter),
259            H::default(),
260            expiry_lifecycle,
261        ));
262
263        // Start the background tasks that the cache needs, in a scope that the cache owns.
264        let needs_drivers = self.expiration_interval.is_some() || self.telemetry_enabled;
265        let mut drivers = needs_drivers.then(|| scope::nested_or_detached(drivers_name));
266
267        if let (Some(drivers), Some(interval)) = (drivers.as_mut(), self.expiration_interval) {
268            drivers.spawn(
269                runtime::supervisable(ExpirationDriver {
270                    cache: Arc::clone(&raw_cache),
271                    telemetry: telemetry.clone(),
272                    expiration: expiration.clone(),
273                    interval,
274                })
275                .build(),
276            );
277        }
278
279        if let (Some(drivers), true) = (drivers.as_mut(), self.telemetry_enabled) {
280            drivers.spawn(
281                runtime::supervisable(TelemetryDriver {
282                    cache: Arc::clone(&raw_cache),
283                    telemetry: telemetry.clone(),
284                })
285                .build(),
286            );
287        }
288
289        Cache {
290            inner: Arc::new(InnerCache {
291                cache: raw_cache,
292                _drivers: drivers,
293            }),
294            expiration,
295            telemetry,
296        }
297    }
298}
299
300/// A simple concurrent cache.
301#[derive(Clone)]
302pub struct Cache<K, V, W = ItemCountWeighter, H = FastBuildHasher> {
303    inner: Arc<InnerCache<K, V, W, H>>,
304    expiration: Expiration<K>,
305    telemetry: Telemetry,
306}
307
308impl<K, V, W, H> Cache<K, V, W, H>
309where
310    K: Eq + std::hash::Hash + Clone,
311    V: Clone,
312    W: Weighter<K, V> + Clone,
313    H: std::hash::BuildHasher + Clone,
314{
315    /// Returns `true` if the cache is empty.
316    pub fn is_empty(&self) -> bool {
317        self.inner.cache.is_empty()
318    }
319
320    /// Returns the number of items currently in the cache.
321    pub fn len(&self) -> usize {
322        self.inner.cache.len()
323    }
324
325    /// Returns the total weight of all items in the cache.
326    pub fn weight(&self) -> u64 {
327        self.inner.cache.weight()
328    }
329
330    /// Inserts an item into the cache with the given key and value.
331    ///
332    /// If an item with the same key already exists, it will be overwritten and the old value will be dropped. If the
333    /// cache is full, one or more items will be evicted to make room for the new item, based on the configured item
334    /// weighter and the weight of the new item.
335    pub fn insert(&self, key: K, value: V) {
336        self.inner.cache.insert(key.clone(), value);
337        self.expiration.mark_entry_accessed(key);
338        self.telemetry.items_inserted_total().increment(1);
339    }
340
341    /// Gets an item from the cache by its key.
342    ///
343    /// If the item is found, it's cloned and `Some(value)` is returned. Otherwise, `None` is returned.
344    pub fn get(&self, key: &K) -> Option<V> {
345        let value = self.inner.cache.get(key);
346        if value.is_some() {
347            self.expiration.mark_entry_accessed(key.clone());
348            self.telemetry.hits_total().increment(1);
349        } else {
350            self.telemetry.misses_total().increment(1);
351        }
352        value
353    }
354
355    /// Removes an item from the cache by its key.
356    pub fn remove(&self, key: &K) {
357        self.inner.cache.remove(key);
358        self.expiration.mark_entry_removed(key.clone());
359        self.telemetry.items_removed_total().increment(1);
360    }
361}
362
363/// Drives expiration for a cache, and removes the entries that are idle.
364struct ExpirationDriver<K, V, W, H> {
365    cache: Arc<RawCache<K, V, W, H>>,
366    telemetry: Telemetry,
367    expiration: Expiration<K>,
368    interval: Duration,
369}
370
371#[async_trait]
372impl<K, V, W, H> Supervisable for ExpirationDriver<K, V, W, H>
373where
374    K: Eq + std::hash::Hash + Clone + Send + Sync + 'static,
375    V: Clone + Send + Sync + 'static,
376    W: Weighter<K, V> + Clone + Send + Sync + 'static,
377    H: std::hash::BuildHasher + Clone + Send + Sync + 'static,
378{
379    fn name(&self) -> &str {
380        "expiration"
381    }
382
383    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
384        let cache = Arc::clone(&self.cache);
385        let telemetry = self.telemetry.clone();
386        let expiration = self.expiration.clone();
387        let interval = self.interval;
388
389        Ok(Box::pin(async move {
390            drive_expiration(cache, telemetry, expiration, interval, process_shutdown).await;
391            Ok(())
392        }))
393    }
394}
395
396/// Drives the telemetry reports for a cache.
397struct TelemetryDriver<K, V, W, H> {
398    cache: Arc<RawCache<K, V, W, H>>,
399    telemetry: Telemetry,
400}
401
402#[async_trait]
403impl<K, V, W, H> Supervisable for TelemetryDriver<K, V, W, H>
404where
405    K: Eq + std::hash::Hash + Clone + Send + Sync + 'static,
406    V: Clone + Send + Sync + 'static,
407    W: Weighter<K, V> + Clone + Send + Sync + 'static,
408    H: std::hash::BuildHasher + Clone + Send + Sync + 'static,
409{
410    fn name(&self) -> &str {
411        "telemetry"
412    }
413
414    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
415        let cache = Arc::clone(&self.cache);
416        let telemetry = self.telemetry.clone();
417
418        Ok(Box::pin(async move {
419            drive_telemetry(cache, telemetry, process_shutdown).await;
420            Ok(())
421        }))
422    }
423}
424
425async fn drive_expiration<K, V, W, H>(
426    cache: Arc<RawCache<K, V, W, H>>, telemetry: Telemetry, expiration: Expiration<K>, expiration_interval: Duration,
427    shutdown: ShutdownHandle,
428) where
429    K: Eq + std::hash::Hash + Clone,
430    V: Clone,
431    W: Weighter<K, V> + Clone,
432    H: std::hash::BuildHasher + Clone,
433{
434    let mut expired_item_keys = Vec::new();
435    pin!(shutdown);
436
437    loop {
438        select! {
439            _ = &mut shutdown => break,
440            _ = sleep(expiration_interval) => {}
441        }
442
443        // Drain all expired items that have been queued up for the cache.
444        expiration.drain_expired_items(&mut expired_item_keys);
445
446        let num_expired_items = expired_item_keys.len();
447        if num_expired_items != 0 {
448            telemetry.items_expired_total().increment(num_expired_items as u64);
449            telemetry.items_expired_batch_size().record(num_expired_items as f64);
450        }
451
452        debug!(num_expired_items, "Found expired items.");
453
454        for item_key in expired_item_keys.drain(..) {
455            cache.remove(&item_key);
456            telemetry.items_removed_total().increment(1);
457            expiration.mark_entry_removed(item_key);
458        }
459
460        debug!(num_expired_items, "Removed expired items.");
461    }
462}
463
464async fn drive_telemetry<K, V, W, H>(cache: Arc<RawCache<K, V, W, H>>, telemetry: Telemetry, shutdown: ShutdownHandle)
465where
466    K: Eq + std::hash::Hash + Clone,
467    V: Clone,
468    W: Weighter<K, V> + Clone,
469    H: std::hash::BuildHasher + Clone,
470{
471    pin!(shutdown);
472
473    loop {
474        select! {
475            _ = &mut shutdown => break,
476            _ = sleep(Duration::from_secs(1)) => {}
477        }
478
479        telemetry.current_items().set(cache.len() as f64);
480        telemetry.current_weight().set(cache.weight() as f64);
481    }
482}
483
484#[cfg(test)]
485mod tests {
486    use super::*;
487
488    #[derive(Clone)]
489    pub struct ItemValueWeighter;
490
491    impl<K> Weighter<K, usize> for ItemValueWeighter {
492        fn item_weight(&self, _key: &K, value: &usize) -> u64 {
493            *value as u64
494        }
495    }
496
497    #[test]
498    fn empty_cache_identifier() {
499        let result = CacheBuilder::<u64, u64>::from_identifier("");
500        assert!(result.is_err(), "expected error for empty cache identifier");
501    }
502
503    #[test]
504    fn basic() {
505        const CACHE_KEY: usize = 42;
506        const CACHE_VALUE: &str = "value1";
507
508        let cache = CacheBuilder::for_tests().build();
509
510        assert_eq!(cache.len(), 0);
511        assert_eq!(cache.weight(), 0);
512
513        cache.insert(CACHE_KEY, CACHE_VALUE);
514        assert_eq!(cache.len(), 1);
515        assert_eq!(cache.weight(), 1);
516
517        assert_eq!(cache.get(&CACHE_KEY), Some(CACHE_VALUE));
518
519        cache.remove(&CACHE_KEY);
520        assert_eq!(cache.len(), 0);
521        assert_eq!(cache.weight(), 0);
522    }
523
524    #[test]
525    fn evict_at_capacity() {
526        const CAPACITY: usize = 3;
527
528        let cache = CacheBuilder::for_tests()
529            .with_capacity(NonZeroUsize::new(CAPACITY).unwrap())
530            .build();
531
532        // Insert items up to the capacity.
533        for i in 0..CAPACITY {
534            cache.insert(i, "value");
535        }
536
537        assert_eq!(cache.len(), CAPACITY);
538        assert_eq!(cache.weight(), CAPACITY as u64);
539
540        // Inserting another item should evict something else to make room, leaving it such that the cache still has the
541        // same number of items.
542        cache.insert(CAPACITY, "new_value");
543        assert_eq!(cache.len(), CAPACITY);
544        assert_eq!(cache.weight(), CAPACITY as u64);
545
546        let mut evicted = false;
547        for i in 0..CAPACITY {
548            if cache.get(&i).is_none() {
549                evicted = true;
550                break;
551            }
552        }
553        assert!(evicted, "expected at least one original item to be evicted");
554    }
555
556    #[test]
557    fn overweight_item() {
558        const CAPACITY: usize = 10;
559
560        // Create our cache using an "item value" weighter, which uses the item value itself as the weight.
561        let cache = CacheBuilder::for_tests()
562            .with_capacity(NonZeroUsize::new(CAPACITY).unwrap())
563            .with_item_weighter(ItemValueWeighter)
564            .build();
565
566        // We should fail to insert an item that is too heavy for the cache by itself.
567        assert_eq!(cache.len(), 0);
568        assert_eq!(cache.weight(), 0);
569
570        cache.insert(1, CAPACITY + 1);
571        assert_eq!(cache.len(), 0);
572        assert_eq!(cache.weight(), 0);
573        assert_eq!(cache.get(&1), None);
574    }
575
576    #[test]
577    fn evict_on_insert_by_weight() {
578        const CAPACITY: usize = 10;
579
580        // Create our cache using an "item value" weighter, which uses the item value itself as the weight.
581        let cache = CacheBuilder::for_tests()
582            .with_capacity(NonZeroUsize::new(CAPACITY).unwrap())
583            .with_item_weighter(ItemValueWeighter)
584            .build();
585
586        // Insert three items which together have a weight equal to the cache capacity.
587        assert_eq!(cache.len(), 0);
588        assert_eq!(cache.weight(), 0);
589
590        cache.insert(1, 3);
591        cache.insert(2, 4);
592        cache.insert(3, 3);
593        assert_eq!(cache.len(), 3);
594        assert_eq!(cache.weight(), CAPACITY as u64);
595
596        // Now try to insert an item that has a weight that is smaller than the cache capacity, but larger than all
597        // prior items combined, which should evict all prior items to make room for the new item.
598        cache.insert(4, CAPACITY - 1);
599        assert_eq!(cache.len(), 1);
600        assert_eq!(cache.weight(), (CAPACITY - 1) as u64);
601
602        assert_eq!(cache.get(&1), None);
603        assert_eq!(cache.get(&2), None);
604        assert_eq!(cache.get(&3), None);
605        assert_eq!(cache.get(&4), Some(CAPACITY - 1));
606    }
607
608    #[tokio::test]
609    async fn tasks_stop_when_cache_dropped() {
610        let cache = CacheBuilder::<u64, u64>::from_identifier("test-drop")
611            .expect("valid identifier")
612            .with_time_to_idle(Some(Duration::from_secs(60)))
613            .with_expiration_interval(Duration::from_millis(50))
614            .build();
615
616        // Grab a weak reference to the raw cache data held by the background tasks.
617        let weak_cache = Arc::downgrade(&cache.inner.cache);
618
619        drop(cache);
620
621        // The cache was built outside supervision, so the tasks run detached. When `InnerCache` is dropped, the guard
622        // that holds the tasks is also dropped, and this signals the tasks. Both tasks then wake immediately, exit, and
623        // release their Arc<RawCache> references.
624        //
625        // TODO: There's no good way to assert the tasks have shutdown besides sleeping and checking the weak cache is
626        // gone. It would be nice if there was a way to asynchronously _and_ fallibly shutdown the runtime with a
627        // timeout, such that we could detect if they shutdown cleanly... but alas.
628        sleep(Duration::from_millis(100)).await;
629
630        assert!(
631            weak_cache.upgrade().is_none(),
632            "raw cache should be released after background tasks exit"
633        );
634    }
635
636    #[tokio::test]
637    async fn tasks_run_under_the_owning_process_and_stop_when_cache_dropped() {
638        use tokio::sync::oneshot;
639
640        use crate::runtime::{NodeSnapshot, Supervisor};
641        use crate::test_support::wait_until;
642
643        fn find<'a>(node: &'a NodeSnapshot, name: &str) -> Option<&'a NodeSnapshot> {
644            if node.name == name {
645                return Some(node);
646            }
647            node.children.iter().find_map(|child| find(child, name))
648        }
649
650        let (weak_tx, weak_rx) = oneshot::channel();
651        let (drop_tx, drop_rx) = oneshot::channel::<()>();
652
653        let mut supervisor = Supervisor::new("cache-test").expect("valid name");
654        let tree = supervisor.tree_handle();
655        supervisor.add_worker(
656            runtime::worker_with_shutdown("owner", move |shutdown| async move {
657                let cache = CacheBuilder::<u64, u64>::from_identifier("test-drop")
658                    .expect("valid identifier")
659                    .with_time_to_idle(Some(Duration::from_secs(60)))
660                    .with_expiration_interval(Duration::from_millis(50))
661                    .build();
662                let _ = weak_tx.send(Arc::downgrade(&cache.inner.cache));
663
664                let _ = drop_rx.await;
665                drop(cache);
666                shutdown.await;
667            })
668            .build(),
669        );
670
671        let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
672        let run = tokio::spawn(async move { supervisor.run_with_shutdown(shutdown_rx).await });
673        let weak_cache = weak_rx.await.expect("owner should build the cache");
674
675        // The cache was built in a supervised process, so the tasks of the cache are children of that process. A scope
676        // that the cache owns holds these tasks.
677        wait_until("the cache's tasks run beneath the owner", || {
678            let root = tree.snapshot().root;
679            find(&root, "cache_test-drop")
680                .is_some_and(|drivers| find(drivers, "expiration").is_some() && find(drivers, "telemetry").is_some())
681        })
682        .await;
683
684        drop_tx.send(()).expect("owner should be waiting");
685        wait_until("dropping the cache stops its tasks", || {
686            find(&tree.snapshot().root, "cache_test-drop").is_none()
687        })
688        .await;
689        wait_until("the raw cache is released", || weak_cache.upgrade().is_none()).await;
690
691        shutdown_tx.send(()).expect("supervisor should be running");
692        let result = tokio::time::timeout(Duration::from_secs(5), run)
693            .await
694            .expect("supervisor should stop")
695            .expect("supervisor task should not panic");
696        assert!(result.is_ok(), "the cache's tasks stopped on their own: {result:?}");
697    }
698}
saluki_core/cache/mod.rs

saluki_core/cache/
mod.rs
