Skip to main content

amethystate_core/primitives/
map_core.rs

1use crate::SignalSubscription;
2use crate::change::MapChange;
3use crate::path::{PathRef, StorePath, Under};
4use crate::primitives::intercept::{InterceptDisposer, InterceptGuard};
5use crate::primitives::signal::{SubscriptionMeta, forget, held, label};
6use arc_swap::ArcSwap;
7use dashmap::DashMap;
8use rpds::RedBlackTreeMapSync;
9use serde::de::DeserializeOwned;
10use serde::{Deserialize, Serialize};
11use smol_str::{SmolStr, SmolStrBuilder};
12use std::any;
13use std::borrow::Borrow;
14use std::fmt::{self, Debug, Display, Write as _};
15use std::hash::Hash;
16use std::panic::Location;
17use std::str::FromStr;
18use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
19use std::sync::{Arc, Mutex};
20
21/// Where a map's entries live, relative to the map itself.
22pub trait MapEntryPath {
23    /// The path of the entry `name` sits at.
24    fn entry(&self, name: &str) -> StorePath;
25}
26
27impl MapEntryPath for StorePath {
28    fn entry(&self, name: &str) -> StorePath {
29        self.push(name)
30    }
31}
32
33pub type InterceptorAny<K, V> =
34    Arc<dyn Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static>;
35pub type InterceptorKey<K, V> =
36    Arc<dyn Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static>;
37pub type SubscriberAny<K, V> = Arc<dyn Fn(&MapChange<K, V>) + Send + Sync + 'static>;
38pub type SubscriberKey<K, V> = Arc<dyn Fn(&MapChange<K, V>) + Send + Sync + 'static>;
39
40/// What a map may be keyed by: a key that already *is* the name its entry sits
41/// at.
42///
43/// An entry lives at one level under the map, and that level is named by the
44/// key. A key that borrows its own name needs no spelling and no parsing: the
45/// store is addressed with what the caller already holds, a listing hands back
46/// what is already there, and neither can disagree with the other, because
47/// there is only one of them.
48///
49/// A key that is not a string is spelled by [`Id`], which renders it once and
50/// holds the rendering.
51/// What one scanned key turned out to be, to the map it was scanned from.
52///
53/// A map is loaded from two places - a store that reads now and a backend that
54/// reads later - and each has its own vocabulary for what went wrong. What they
55/// may not have is their own opinion about *which* of these four a key is, and
56/// each working it out from the path by hand is how they came to differ.
57pub enum EntryOf<K> {
58    /// One level below the map, and the name reads as `K`.
59    Took(K),
60
61    /// The map's own path. Not an entry, and not a fault: a scan of a prefix
62    /// hands the prefix back.
63    ThePathItself,
64
65    /// One level below, and the name does not read as `K`.
66    NameWillNotRead,
67
68    /// Deeper than the level the map owns, or not under it at all. A map owns
69    /// the level below it and nothing further, so a key past that belongs to
70    /// whatever claimed that level.
71    NotThisMaps,
72}
73
74/// Which of the four `stored` is, to a map at `prefix`.
75pub fn entry_of<K>(prefix: &StorePath, stored: PathRef<'_>) -> EntryOf<K>
76where
77    K: ReactiveMapKey,
78{
79    match stored.level_under(prefix) {
80        Under::Prefix => EntryOf::ThePathItself,
81        Under::Deeper(_) | Under::Outside => EntryOf::NotThisMaps,
82        Under::Entry(name) => match K::read(name.as_str()) {
83            Some(key) => EntryOf::Took(key),
84            None => EntryOf::NameWillNotRead,
85        },
86    }
87}
88
89pub trait ReactiveMapKey: AsRef<str> + Clone + Hash + Eq + Send + Sync + 'static {
90    /// The key a stored name stands for, or `None` where that name is not one.
91    ///
92    /// The only direction that can fail, and the only one that allocates. It is
93    /// walked once per entry when a map is built from the store, and never on a
94    /// read.
95    fn read(name: &str) -> Option<Self>;
96}
97
98impl ReactiveMapKey for String {
99    fn read(name: &str) -> Option<Self> {
100        Some(name.to_string())
101    }
102}
103
104impl ReactiveMapKey for SmolStr {
105    fn read(name: &str) -> Option<Self> {
106        Some(SmolStr::new(name))
107    }
108}
109
110impl<T> ReactiveMapKey for Id<T>
111where
112    T: Display + FromStr + Clone + Send + Sync + 'static,
113{
114    fn read(name: &str) -> Option<Self> {
115        T::from_str(name).ok().map(Id::new)
116    }
117}
118
119/// A key that is not a string, spelled once and kept that way.
120///
121/// `Id::new(9u16)` names its entry `9`, `Id::new(some_uuid)` names it by the
122/// uuid's own spelling. The rendering is done when the `Id` is built rather
123/// than at every lookup, which is what a `Uuid` wants most: 36 characters do
124/// not fit beside the value, so spelling one costs the heap, and this pays it
125/// once for the life of the key instead of once per read.
126///
127/// Entries are listed in the order of the *spelling*, so `Id<u16>` lists `10`
128/// before `9`. A number's decimal spelling does not order like the number, and
129/// nothing here pretends otherwise - padding it would make the name in the file
130/// something nobody typed.
131#[derive(Clone)]
132pub struct Id<T> {
133    spelled: SmolStr,
134    value: T,
135}
136
137impl<T: Display> Id<T> {
138    pub fn new(value: T) -> Self {
139        let mut out = SmolStrBuilder::new();
140        let _ = write!(out, "{value}");
141
142        Self {
143            spelled: out.finish(),
144            value,
145        }
146    }
147}
148
149impl<T> Id<T> {
150    /// What was spelled, for a caller that wants the number back.
151    pub fn get(&self) -> &T {
152        &self.value
153    }
154
155    pub fn into_inner(self) -> T {
156        self.value
157    }
158}
159
160impl<T> AsRef<str> for Id<T> {
161    fn as_ref(&self) -> &str {
162        &self.spelled
163    }
164}
165
166/// So a map keyed by an id can be looked up by the spelling alone, without
167/// building one. Sound because `Eq` and `Hash` below answer from the same
168/// spelling, which is what `Borrow` asks.
169impl<T> Borrow<str> for Id<T> {
170    fn borrow(&self) -> &str {
171        &self.spelled
172    }
173}
174
175/// By the spelling, which is what the store is addressed by and what a listing
176/// is ordered by. Two ids that spell alike are one entry whatever they hold.
177impl<T> PartialEq for Id<T> {
178    fn eq(&self, other: &Self) -> bool {
179        self.spelled == other.spelled
180    }
181}
182
183impl<T> Eq for Id<T> {}
184
185impl<T> Hash for Id<T> {
186    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
187        self.spelled.hash(state);
188    }
189}
190
191impl<T> Display for Id<T> {
192    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
193        f.write_str(&self.spelled)
194    }
195}
196
197impl<T: Debug> Debug for Id<T> {
198    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
199        f.debug_tuple("Id").field(&self.value).finish()
200    }
201}
202
203/// As the spelling, so a snapshot holding one for a key is a document with
204/// string keys - which is what every format this library writes needs a map's
205/// keys to be.
206impl<T> Serialize for Id<T> {
207    fn serialize<S: serde::Serializer>(&self, out: S) -> Result<S::Ok, S::Error> {
208        out.serialize_str(&self.spelled)
209    }
210}
211
212impl<'de, T: Display + FromStr> Deserialize<'de> for Id<T> {
213    fn deserialize<D: serde::Deserializer<'de>>(from: D) -> Result<Self, D::Error> {
214        let spelled = <std::borrow::Cow<'de, str>>::deserialize(from)?;
215
216        T::from_str(&spelled).map(Id::new).map_err(|_| {
217            serde::de::Error::custom(format!("{spelled:?} is not a {}", any::type_name::<T>()))
218        })
219    }
220}
221
222pub trait ReactiveMapValue:
223    Serialize + DeserializeOwned + Clone + Send + Sync + 'static + Default
224{
225}
226impl<T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static + Default> ReactiveMapValue
227    for T
228{
229}
230
231/// A map's entries, held in the order the store lists them.
232///
233/// Keyed by the key itself, ordered by the name it borrows. A store orders its
234/// keys by their levels, and an entry is one level, so the name's own order is
235/// the store's - the two cannot come apart, because the cache is not holding a
236/// second thing to compare.
237///
238/// A read takes a version and holds nothing, so a walk neither blocks a writer
239/// nor waits for one, whatever thread either is on. A write publishes a new
240/// version that shares every node it did not touch, so it stays O(log n)
241/// rather than copying the map.
242///
243/// It is a trade, not a free win. Writes and memory both cost more than a
244/// `RwLock<BTreeMap>` would, and
245/// `crates/main/amethystate/benches/map_snapshot_bench.rs` measures against
246/// one.
247///
248/// What it buys is that writing during a walk cannot deadlock. It does not
249/// make it correct: the walk goes on yielding its own version, so a write made
250/// inside the loop is invisible to the rest of it. `for k in keys { remove(k) }`
251/// wants exactly that; a loop that writes and reads the same key back gets a
252/// stale answer and nothing says so. Loud failure was traded for quiet
253/// staleness.
254///
255/// The shape is not selectable per map, though it could be - the backing type
256/// is private either way. Nothing would pick the locking one: it wins only
257/// where a map is written in bulk and never walked while it is written, and a
258/// map nobody observes is the store's job.
259pub struct MapCache<K, V> {
260    entries: ArcSwap<Snapshot<K, V>>,
261}
262
263impl<K: AsRef<str>, V> Default for MapCache<K, V> {
264    fn default() -> Self {
265        Self {
266            entries: ArcSwap::from_pointee(RedBlackTreeMapSync::new_sync()),
267        }
268    }
269}
270
271/// A key held as the cache orders it: by the name it borrows, never by anything
272/// the key type decides for itself.
273///
274/// The invariant the cache rests on is that its order is the store's. Requiring
275/// `K: Ord` would ask the key type to agree and have no way to check; this
276/// takes the ordering away from it instead, so a key whose own `Ord` says
277/// something else cannot make a listing disagree with a scan.
278#[derive(Clone)]
279struct ByName<K>(K);
280
281impl<K: AsRef<str>> Borrow<str> for ByName<K> {
282    fn borrow(&self) -> &str {
283        self.0.as_ref()
284    }
285}
286
287impl<K: AsRef<str>> PartialEq for ByName<K> {
288    fn eq(&self, other: &Self) -> bool {
289        self.0.as_ref() == other.0.as_ref()
290    }
291}
292
293impl<K: AsRef<str>> Eq for ByName<K> {}
294
295impl<K: AsRef<str>> Ord for ByName<K> {
296    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
297        self.0.as_ref().cmp(other.0.as_ref())
298    }
299}
300
301impl<K: AsRef<str>> PartialOrd for ByName<K> {
302    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
303        Some(self.cmp(other))
304    }
305}
306
307impl<K: AsRef<str> + Clone, V: Clone> MapCache<K, V> {
308    pub fn new() -> Self {
309        Self::default()
310    }
311
312    pub fn get(&self, name: &str) -> Option<V> {
313        self.entries.load().get(name).cloned()
314    }
315
316    pub fn contains_key(&self, name: &str) -> bool {
317        self.entries.load().contains_key(name)
318    }
319
320    /// The key as the map holds it, for a caller that looked one up by the name
321    /// it borrows and needs the owned form back.
322    pub fn owned_key(&self, name: &str) -> Option<K> {
323        self.entries
324            .load()
325            .get_key_value(name)
326            .map(|(key, _)| key.0.clone())
327    }
328
329    pub fn len(&self) -> usize {
330        self.entries.load().size()
331    }
332
333    pub fn is_empty(&self) -> bool {
334        self.entries.load().is_empty()
335    }
336
337    pub fn clear(&self) {
338        self.entries
339            .store(Arc::new(RedBlackTreeMapSync::new_sync()));
340    }
341
342    /// The entries themselves, in order, without copying any of them.
343    ///
344    /// The version is taken when the view is, so writes that land afterwards
345    /// are not in it however long it is held.
346    pub fn view(&self) -> Entries<K, V> {
347        Entries {
348            held: self.entries.load_full(),
349        }
350    }
351
352    /// Every key, in the order the contract promises, one at a time.
353    pub fn keys(&self) -> Walk<K, V, K> {
354        Walk::new(self.entries.load_full(), |key, _| key.0.clone())
355    }
356
357    /// Every entry, in that order, one at a time.
358    pub fn entries(&self) -> Walk<K, V, (K, V)> {
359        Walk::new(self.entries.load_full(), |key, value| {
360            (key.0.clone(), value.clone())
361        })
362    }
363}
364
365type Snapshot<K, V> = RedBlackTreeMapSync<ByName<K>, V>;
366type Held<K, V> = Arc<Snapshot<K, V>>;
367type Pairs<'a, K, V> = <&'a Snapshot<K, V> as IntoIterator>::IntoIter;
368/// A borrowed walk of one version, handing back the key beside its value.
369///
370/// Its own type rather than a mapped iterator, so how the cache holds a key
371/// stays inside this module.
372pub struct Values<'a, K: AsRef<str>, V> {
373    pairs: Pairs<'a, K, V>,
374}
375
376impl<'a, K: AsRef<str>, V> Iterator for Values<'a, K, V> {
377    type Item = (&'a K, &'a V);
378
379    fn next(&mut self) -> Option<Self::Item> {
380        self.pairs.next().map(|(key, value)| (&key.0, value))
381    }
382
383    fn size_hint(&self) -> (usize, Option<usize>) {
384        self.pairs.size_hint()
385    }
386}
387
388impl<K: AsRef<str>, V> DoubleEndedIterator for Values<'_, K, V> {
389    fn next_back(&mut self) -> Option<Self::Item> {
390        self.pairs.next_back().map(|(key, value)| (&key.0, value))
391    }
392}
393
394/// A walk of a [`MapCache`], in order, taking nothing it is not asked for.
395///
396/// The position is the iterator's own, so a whole pass costs one descent and
397/// `take(n)` costs `n`.
398///
399/// The walk owns the version it started on. Writing to the same map while it is
400/// alive is allowed from any thread, the walk included, and the walk keeps
401/// handing back what its own version holds.
402pub struct Walk<K: AsRef<str> + 'static, V: 'static, T> {
403    pairs: Pairs<'static, K, V>,
404    take: fn(&ByName<K>, &V) -> T,
405    _held: Held<K, V>,
406}
407
408impl<K: AsRef<str>, V, T> Walk<K, V, T> {
409    fn new(held: Held<K, V>, take: fn(&ByName<K>, &V) -> T) -> Self {
410        // SAFETY: two invariants, both of which this module has to keep.
411        //
412        // The map lives in the `Arc`'s allocation and `_held` owns a strong
413        // reference to it, so it outlives every borrow taken here however the
414        // walk is moved. That is what lets the type carry no lifetime.
415        //
416        // The `'static` never reaches a caller: `new` is private, and `take` is
417        // `for<'x> fn(&'x ByName<K>, &'x V) -> T`, which cannot return either
418        // argument. So no `T` can be a reference into the map, and the only
419        // `&'static` that exists is the temporary inside `next` and `next_back`.
420        let pairs =
421            unsafe { std::mem::transmute::<Pairs<'_, K, V>, Pairs<'static, K, V>>(held.iter()) };
422
423        Self {
424            pairs,
425            take,
426            _held: held,
427        }
428    }
429}
430
431impl<K: AsRef<str>, V, T> Iterator for Walk<K, V, T> {
432    type Item = T;
433
434    fn next(&mut self) -> Option<T> {
435        self.pairs
436            .next()
437            .map(|(key, value)| (self.take)(key, value))
438    }
439
440    fn size_hint(&self) -> (usize, Option<usize>) {
441        self.pairs.size_hint()
442    }
443}
444
445impl<K: AsRef<str>, V, T> DoubleEndedIterator for Walk<K, V, T> {
446    fn next_back(&mut self) -> Option<T> {
447        self.pairs
448            .next_back()
449            .map(|(key, value)| (self.take)(key, value))
450    }
451}
452
453impl<K: AsRef<str>, V, T> ExactSizeIterator for Walk<K, V, T> {}
454
455/// One version of a [`MapCache`], in order, borrowed rather than copied.
456pub struct Entries<K, V> {
457    held: Held<K, V>,
458}
459
460impl<'e, K: AsRef<str>, V> IntoIterator for &'e Entries<K, V> {
461    type Item = (&'e K, &'e V);
462    type IntoIter = Values<'e, K, V>;
463
464    fn into_iter(self) -> Self::IntoIter {
465        Values {
466            pairs: self.held.iter(),
467        }
468    }
469}
470
471impl<K: AsRef<str>, V> Entries<K, V> {
472    pub fn iter(&self) -> impl DoubleEndedIterator<Item = (&K, &V)> {
473        self.held.iter().map(|(key, value)| (&key.0, value))
474    }
475
476    pub fn len(&self) -> usize {
477        self.held.size()
478    }
479
480    pub fn is_empty(&self) -> bool {
481        self.held.is_empty()
482    }
483}
484
485impl<K: AsRef<str> + Clone, V: Clone> MapCache<K, V> {
486    pub fn insert(&self, key: K, value: V) -> Option<V> {
487        let replaced = self
488            .entries
489            .rcu(|current| current.insert(ByName(key.clone()), value.clone()));
490
491        replaced.get(key.as_ref()).cloned()
492    }
493
494    pub fn remove(&self, name: &str) -> Option<(K, V)> {
495        let previous = self.entries.rcu(|current| current.remove(name));
496
497        previous
498            .get_key_value(name)
499            .map(|(key, value)| (key.0.clone(), value.clone()))
500    }
501}
502
503impl<K: AsRef<str> + Debug, V: Debug> Debug for MapCache<K, V> {
504    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
505        let held = self.entries.load();
506
507        f.debug_map()
508            .entries(held.iter().map(|(key, value)| (&key.0, value)))
509            .finish()
510    }
511}
512
513pub struct ReactiveMapCore<K, V> {
514    pub interceptors_any: Arc<Mutex<Vec<(u64, InterceptorAny<K, V>)>>>,
515    pub interceptors_key: Arc<DashMap<K, Vec<(u64, InterceptorKey<K, V>)>>>,
516    pub subscribers_any: Arc<Mutex<Vec<(u64, SubscriberAny<K, V>, SubscriptionMeta)>>>,
517    pub subscribers_key: Arc<DashMap<K, Vec<(u64, SubscriberKey<K, V>, SubscriptionMeta)>>>,
518    pub next_id: Arc<AtomicU64>,
519    pub intercept_depth: Arc<AtomicUsize>,
520    pub cache: Arc<MapCache<K, V>>,
521}
522
523impl<K, V> Clone for ReactiveMapCore<K, V> {
524    fn clone(&self) -> Self {
525        Self {
526            interceptors_any: self.interceptors_any.clone(),
527            interceptors_key: self.interceptors_key.clone(),
528            subscribers_any: self.subscribers_any.clone(),
529            subscribers_key: self.subscribers_key.clone(),
530            next_id: self.next_id.clone(),
531            intercept_depth: self.intercept_depth.clone(),
532            cache: self.cache.clone(),
533        }
534    }
535}
536
537struct Counted<'a, T>(&'a Mutex<Vec<T>>);
538
539impl<T> Debug for Counted<'_, T> {
540    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
541        match self.0.try_lock() {
542            Ok(list) => write!(f, "{}", list.len()),
543            Err(_) => f.write_str("<locked>"),
544        }
545    }
546}
547
548impl<K: AsRef<str> + Debug + Hash + Eq, V: Debug> Debug for ReactiveMapCore<K, V> {
549    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
550        f.debug_struct("ReactiveMapCore")
551            .field("cache", &self.cache)
552            .field("interceptors_any", &Counted(&self.interceptors_any))
553            .field("subscribers_any", &Counted(&self.subscribers_any))
554            .finish()
555    }
556}
557
558impl<K: ReactiveMapKey, V: ReactiveMapValue> Default for ReactiveMapCore<K, V> {
559    fn default() -> Self {
560        Self::new()
561    }
562}
563
564impl<K: ReactiveMapKey, V: ReactiveMapValue> ReactiveMapCore<K, V> {
565    pub fn new() -> Self {
566        Self::with_capacity(0)
567    }
568
569    /// The same. A tree has nothing to reserve, so `entries` is ignored; the
570    /// signature stays because a caller loading a store knows the size and
571    /// should not have to know that.
572    pub fn with_capacity(_entries: usize) -> Self {
573        Self {
574            interceptors_any: Arc::new(Mutex::new(Vec::new())),
575            interceptors_key: Arc::new(DashMap::new()),
576            subscribers_any: Arc::new(Mutex::new(Vec::new())),
577            subscribers_key: Arc::new(DashMap::new()),
578            next_id: Arc::new(AtomicU64::new(0)),
579            intercept_depth: Arc::new(AtomicUsize::new(0)),
580            cache: Arc::new(MapCache::new()),
581        }
582    }
583
584    #[track_caller]
585    pub fn subscribe_any<F>(&self, callback: F) -> SignalSubscription
586    where
587        F: Fn(&MapChange<K, V>) + Send + Sync + 'static,
588    {
589        let location = Location::caller();
590        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
591        let meta = SubscriptionMeta {
592            id,
593            location,
594            name: None,
595        };
596        held(&self.subscribers_any).push((id, Arc::new(callback), meta));
597
598        let subs_for_name = self.subscribers_any.clone();
599        let set_name = Arc::new(move |name: &'static str| {
600            label(&mut held(&subs_for_name), id, name);
601        });
602        let subs_for_cleanup = self.subscribers_any.clone();
603        SignalSubscription::new(
604            id,
605            location,
606            set_name,
607            Arc::new(move |id| {
608                forget(&mut held(&subs_for_cleanup), id);
609            }),
610        )
611    }
612
613    #[track_caller]
614    pub fn subscribe_key<F>(&self, key: K, callback: F) -> SignalSubscription
615    where
616        F: Fn(&MapChange<K, V>) + Send + Sync + 'static,
617    {
618        let location = Location::caller();
619        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
620        let meta = SubscriptionMeta {
621            id,
622            location,
623            name: None,
624        };
625        self.subscribers_key
626            .entry(key.clone())
627            .or_default()
628            .push((id, Arc::new(callback), meta));
629
630        let subs_for_name = self.subscribers_key.clone();
631        let key_for_name = key.clone();
632        let set_name = Arc::new(move |name: &'static str| {
633            if let Some(mut list) = subs_for_name.get_mut(&key_for_name) {
634                label(&mut list, id, name);
635            }
636        });
637        let subs_for_cleanup = self.subscribers_key.clone();
638        SignalSubscription::new(
639            id,
640            location,
641            set_name,
642            Arc::new(move |id| {
643                if let Some(mut list) = subs_for_cleanup.get_mut(&key) {
644                    forget(&mut list, id);
645                }
646                subs_for_cleanup.remove_if(&key, |_, list| list.is_empty());
647            }),
648        )
649    }
650
651    pub fn intercept<F>(&self, path: StorePath, callback: F) -> InterceptDisposer
652    where
653        F: Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static,
654    {
655        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
656        held(&self.interceptors_any).push((id, Arc::new(callback)));
657        let subs = self.interceptors_any.clone();
658        InterceptDisposer {
659            id,
660            path,
661            cleanup: Arc::new(move |id| {
662                held(&subs).retain(|(i, _)| *i != id);
663            }),
664        }
665    }
666
667    pub fn intercept_key<F>(&self, key: K, callback: F) -> InterceptDisposer
668    where
669        F: Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static,
670    {
671        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
672        self.interceptors_key
673            .entry(key.clone())
674            .or_default()
675            .push((id, Arc::new(callback)));
676        let subs = self.interceptors_key.clone();
677        InterceptDisposer {
678            id,
679            path: StorePath::root(),
680            cleanup: Arc::new(move |id| {
681                if let Some(mut list) = subs.get_mut(&key) {
682                    list.retain(|(i, _)| *i != id);
683                }
684                subs.remove_if(&key, |_, list| list.is_empty());
685            }),
686        }
687    }
688
689    pub fn run_interceptors(
690        &self,
691        path: StorePath,
692        mut change: MapChange<K, V>,
693    ) -> Result<MapChange<K, V>, crate::primitives::intercept::Refusal> {
694        use crate::primitives::intercept::Refusal;
695
696        let Some(_guard) = InterceptGuard::enter(&self.intercept_depth, path) else {
697            return Err(Refusal::Recursed);
698        };
699
700        if let Some(key) = change.key().cloned() {
701            let interceptors = self
702                .interceptors_key
703                .get(&key)
704                .map(|entry| entry.clone())
705                .unwrap_or_default();
706            for (_, interceptor) in interceptors {
707                if let Some(new_change) = interceptor(change.clone()) {
708                    change = new_change;
709                } else {
710                    return Err(Refusal::Said(
711                        "refused by an interceptor on that key".to_string(),
712                    ));
713                }
714            }
715        }
716
717        let interceptors_any = held(&self.interceptors_any).clone();
718        for (_, interceptor) in interceptors_any {
719            if let Some(new_change) = interceptor(change.clone()) {
720                change = new_change;
721            } else {
722                return Err(Refusal::Said(
723                    "refused by an interceptor on the map".to_string(),
724                ));
725            }
726        }
727
728        Ok(change)
729    }
730
731    /// Fires every subscriber interested in `change`.
732    ///
733    /// The callbacks are collected before any of them runs, and every guard is
734    /// released first. A subscriber reacting to a change by writing to the same
735    /// map is ordinary, and neither a `Mutex` nor a `DashMap` shard is
736    /// reentrant, so holding either across the calls deadlocks the thread.
737    pub fn notify(&self, change: &MapChange<K, V>) {
738        let keyed: Vec<_> = match change.key() {
739            Some(k) => self
740                .subscribers_key
741                .get(k)
742                .map(|entries| {
743                    entries
744                        .iter()
745                        .map(|(_, cb, meta)| (cb.clone(), *meta))
746                        .collect()
747                })
748                .unwrap_or_default(),
749
750            // A clear is about every key at once, so everyone watching a key
751            // hears it. Told about nothing, an entry cell goes on reporting a
752            // value the map no longer has, and has no later change to correct
753            // it.
754            None => self
755                .subscribers_key
756                .iter()
757                .flat_map(|entries| {
758                    entries
759                        .value()
760                        .iter()
761                        .map(|(_, cb, meta)| (cb.clone(), *meta))
762                        .collect::<Vec<_>>()
763                })
764                .collect(),
765        };
766
767        let any: Vec<_> = held(&self.subscribers_any)
768            .iter()
769            .map(|(_, cb, meta)| (cb.clone(), *meta))
770            .collect();
771
772        for (cb, meta) in keyed {
773            tracing::trace!(
774                target: "amethystate",
775                subscription_id = meta.id,
776                name = meta.name,
777                location = format!("{}:{}", meta.location.file(), meta.location.line()),
778                "map signal emit → key subscription fire",
779            );
780            cb(change);
781        }
782
783        for (cb, meta) in any {
784            tracing::trace!(
785                target: "amethystate",
786                subscription_id = meta.id,
787                name = meta.name,
788                location = format!("{}:{}", meta.location.file(), meta.location.line()),
789                "map signal emit → any subscription fire",
790            );
791            cb(change);
792        }
793    }
794}
795
796#[cfg(test)]
797mod tests {
798    use super::*;
799
800    #[test]
801    fn the_cache_lists_entries_in_the_order_the_store_keys_them() {
802        let cache = MapCache::<String, u8>::new();
803        let names = ["a1b", "a.b", "10", "9", "100", "a\\b", "cpu", "Cpu"];
804
805        for (n, name) in names.iter().enumerate() {
806            cache.insert((*name).to_string(), n as u8);
807        }
808
809        let under = StorePath::segment("m");
810        let mut by_store = names.to_vec();
811        by_store.sort_by_key(|name| under.push(name).key().as_bytes().to_vec());
812
813        assert_eq!(cache.keys().collect::<Vec<_>>(), by_store);
814    }
815
816    #[test]
817    fn a_name_holding_the_separator_is_one_entry() {
818        let cache = MapCache::<String, u8>::new();
819
820        cache.insert("dark.mode".to_string(), 1);
821
822        assert_eq!(cache.len(), 1);
823        assert_eq!(cache.get("dark.mode"), Some(1));
824        assert_eq!(cache.get("dark"), None);
825        assert_eq!(cache.get("dark\\.mode"), None);
826        assert_eq!(cache.keys().collect::<Vec<_>>(), ["dark.mode"]);
827    }
828
829    #[test]
830    fn a_dropped_key_subscription_leaves_no_entry_behind() {
831        let core = ReactiveMapCore::<String, u64>::new();
832
833        for i in 0..64u64 {
834            let key = format!("col{i}");
835            let sub = core.subscribe_key(key.clone(), |_| {});
836            let held = core.intercept_key(key, Some);
837            drop(sub);
838            held.remove();
839        }
840
841        assert_eq!(core.subscribers_key.len(), 0);
842        assert_eq!(core.interceptors_key.len(), 0);
843    }
844
845    #[test]
846    fn a_key_still_watched_by_someone_keeps_its_entry() {
847        let core = ReactiveMapCore::<String, u64>::new();
848
849        let first = core.subscribe_key("cpu".to_string(), |_| {});
850        let second = core.subscribe_key("cpu".to_string(), |_| {});
851
852        drop(first);
853        assert_eq!(core.subscribers_key.len(), 1);
854
855        drop(second);
856        assert_eq!(core.subscribers_key.len(), 0);
857    }
858}