graph-explorer-core 0.1.0

//! Cache that lets the synchronous `DataProvider` serve data arriving
//! asynchronously: a miss queues the key and reports `pending`, and the host
//! fills it in later. Pure — no I/O, no wasm — so it is host-testable.

use std::cell::RefCell;
use std::collections::HashMap;
use std::rc::Rc;
use crate::{Cursor, DataProvider, Graph, NeighborResult, NodeId, QueryParams};

pub type CacheKey = (NodeId, Option<Cursor>);

/// `InFlight` is tagged with the epoch of the attempt that installed it, so a
/// settle (`fill`/`fail`) can be matched to the attempt that produced it. Two
/// outstanding attempts for one key are possible after a cancel (the key
/// carries no slot afterwards, so `mark_wanted` queues it again while the
/// first fetch is still outstanding) and are otherwise indistinguishable.
#[derive(Debug, Clone, PartialEq)]
enum Slot { InFlight(u64), Failed(String) }

/// Ready results plus in-flight/failed bookkeeping.
#[derive(Default)]
pub struct NeighborCache {
    ready: HashMap<String, NeighborResult>,
    slots: HashMap<String, Slot>,
    wanted: Vec<CacheKey>,
    dirty: bool,
    /// Structured keys + messages for failures not yet reported to the host.
    /// Kept separately because `key_str` is lossy and cannot be reversed.
    failures: Vec<(CacheKey, String)>,
    /// Keys filled since the host last drained them, so growth is incremental
    /// rather than a re-walk of every ready result on each dirty tick.
    new_arrivals: Vec<CacheKey>,
    /// Monotonic counter; the next `take_wanted` epoch is `next_epoch + 1`.
    next_epoch: u64,
    /// Highest epoch that has settled for a key. The `InFlight` slot is gone
    /// by the time a late settle arrives, so without this there is nothing
    /// left to compare against, and an older attempt's answer would overwrite
    /// a newer one -- or resurrect a key the newer attempt had recorded as
    /// failed.
    settled: HashMap<String, u64>,
    /// Epoch floor installed by `reset`: attempts drawn before it can never
    /// settle. `None` until the first reset, so a cache that has never been
    /// reset imposes no floor at all.
    discard_before: Option<u64>,
}

/// `Cursor` is not reversible from a flattened string, so keys are flattened
/// to one string. Node ids arrive from the network and are not validated, so
/// the id is length-prefixed: that makes the encoding unambiguous even if an
/// id itself contains \u{1}. The cursor half is further tagged with a
/// presence discriminant (`0`/`1`) so an absent cursor (`None`) never
/// collides with a present-but-empty one (`Some(Cursor(String::new()))`) --
/// without it, this flattening would disagree with `CacheKey`'s derived
/// `Hash`/`Eq` (which does distinguish the two), and a host indexing
/// `HashMap<CacheKey, AbortController>` by cache key could hold two entries
/// where this cache holds one slot -- a cancel would then tear down the wrong
/// fetch.
fn key_str(k: &CacheKey) -> String {
    match &k.1 {
        Some(Cursor(c)) => format!("{}\u{1}{}\u{1}1{}", k.0.len(), k.0, c),
        None => format!("{}\u{1}{}\u{1}0", k.0.len(), k.0),
    }
}

impl NeighborCache {
    pub fn new() -> Self { Self::default() }

    pub fn get(&self, k: &CacheKey) -> Option<&NeighborResult> { self.ready.get(&key_str(k)) }

    pub fn error(&self, k: &CacheKey) -> Option<&str> {
        match self.slots.get(&key_str(k)) {
            Some(Slot::Failed(e)) => Some(e.as_str()),
            _ => None,
        }
    }

    /// Queue a miss for fetching. Idempotent: a key already ready, in flight,
    /// failed, or already queued is not queued again — so holding a navigation
    /// key cannot stampede the backend and a failure cannot retry-loop.
    pub fn mark_wanted(&mut self, k: CacheKey) {
        let ks = key_str(&k);
        if self.ready.contains_key(&ks) || self.slots.contains_key(&ks) { return; }
        if self.wanted.iter().any(|w| key_str(w) == ks) { return; }
        self.wanted.push(k);
    }

    /// Drain queued keys, marking each in flight under a fresh epoch. The
    /// epoch identifies THIS attempt: the host must hand it back to
    /// `fill`/`fail` so a settle can be matched to the attempt that produced
    /// it. Without that, two outstanding attempts for one key (possible after
    /// a cancel, since the key carries no slot and re-queues) are
    /// indistinguishable.
    pub fn take_wanted(&mut self) -> Vec<(CacheKey, u64)> {
        let out: Vec<CacheKey> = self.wanted.drain(..).collect();
        out.into_iter()
            .map(|k| {
                self.next_epoch += 1;
                self.slots.insert(key_str(&k), Slot::InFlight(self.next_epoch));
                (k, self.next_epoch)
            })
            .collect()
    }

    /// Whether a settle for `epoch` has been overtaken: either a newer attempt
    /// is still in flight, or an attempt at least as new has already settled.
    /// `>=`, not `>`: a settle whose epoch equals the last settled one is a
    /// duplicate delivery of the very same attempt, not a fresh answer, so it
    /// is stale too -- otherwise a redelivered `fill` would re-store the same
    /// data and re-report it via `take_new_arrivals` as if it were new.
    fn is_stale(&self, ks: &str, epoch: u64) -> bool {
        // Drawn before a `reset`, so it answers a question about a dataset
        // this cache no longer describes. Checked first: `reset` wipes both
        // `slots` and `settled`, so neither of the checks below has anything
        // left to catch such an attempt with.
        if matches!(self.discard_before, Some(floor) if epoch <= floor) {
            return true;
        }
        if let Some(Slot::InFlight(e)) = self.slots.get(ks) {
            if *e > epoch {
                return true;
            }
        }
        matches!(self.settled.get(ks), Some(s) if *s >= epoch)
    }

    /// Store a received result for the attempt tagged `epoch`. `pending` is a
    /// client-side concept — it means "this placeholder stands in for a fetch
    /// we have not received yet" — so a value decoded off the wire is never
    /// trusted: a host or proxy echoing `{"pending":true}` would otherwise
    /// land a permanent cache HIT that reports pending, wedging the view on a
    /// spinner that can never resolve.
    pub fn fill(&mut self, k: CacheKey, epoch: u64, mut res: NeighborResult) {
        let ks = key_str(&k);
        if self.is_stale(&ks, epoch) { return; }
        // A newer attempt for this key is outstanding, or one at least as new
        // has already settled; either way this answer loses. An attempt with
        // NO slot and nothing newer settled (cancelled, nothing since) still
        // stores: the bytes arrived anyway, and data keyed by node id is
        // always valid, so revisiting the node is a hit rather than a
        // refetch.
        res.pending = false;
        self.slots.remove(&ks);
        self.ready.insert(ks.clone(), res);
        self.settled.insert(ks, epoch);
        self.new_arrivals.push(k);
        self.dirty = true;
    }

    /// Record a failure for the attempt tagged `epoch`.
    pub fn fail(&mut self, k: CacheKey, epoch: u64, err: String) {
        let ks = key_str(&k);
        // Only the attempt currently in flight may record a failure. A
        // cancelled attempt has no slot and a superseded one has a newer
        // epoch; either way this rejection is stale. Recording it would
        // install a `Slot::Failed` that `mark_wanted` refuses to re-queue,
        // making the node a permanent dead end for a cancellation we ourselves
        // requested.
        if !matches!(self.slots.get(&ks), Some(Slot::InFlight(e)) if *e == epoch) {
            return;
        }
        self.slots.insert(ks.clone(), Slot::Failed(err.clone()));
        // Recorded so a later, even-more-stale success (from an attempt older
        // than this failure) cannot fall through `fill`'s guard and erase it --
        // that would be a second automatic path around `clear_failures`.
        self.settled.insert(ks, epoch);
        self.failures.push((k, err));
        self.dirty = true;
    }

    /// Cancel a fetch. A key still queued is simply dequeued; a key in flight
    /// loses its slot, so `is_loading()` goes quiet at once and the stale
    /// settle is discarded by the epoch check in `fill`/`fail`. Cancellation
    /// is not sticky: the key carries no slot afterwards, so `mark_wanted`
    /// queues it again.
    pub fn cancel(&mut self, k: &CacheKey) {
        let ks = key_str(k);
        self.wanted.retain(|w| key_str(w) != ks);
        // ONLY an in-flight attempt is cancellable. Removing a `Slot::Failed`
        // would backdoor the "failures are never cleared automatically"
        // invariant and let a down backend be hammered.
        if matches!(self.slots.get(&ks), Some(Slot::InFlight(_))) {
            self.slots.remove(&ks);
        }
    }

    /// Whether a fetch for `k` is currently outstanding. Hosts holding
    /// per-fetch resources (an `AbortController`, say) use this to prune.
    pub fn is_in_flight(&self, k: &CacheKey) -> bool {
        matches!(self.slots.get(&key_str(k)), Some(Slot::InFlight(_)))
    }

    /// True once since the last call; consumed by the renderer.
    pub fn take_dirty(&mut self) -> bool { std::mem::take(&mut self.dirty) }

    /// Drain failures that have not yet been reported. The `Slot::Failed`
    /// record remains, so a failed key still does not auto-retry.
    pub fn take_failures(&mut self) -> Vec<(CacheKey, String)> {
        std::mem::take(&mut self.failures)
    }

    /// Drain the keys filled since the last call, so a host can absorb only
    /// what actually arrived instead of re-walking every ready result.
    pub fn take_new_arrivals(&mut self) -> Vec<CacheKey> {
        std::mem::take(&mut self.new_arrivals)
    }

    /// Forget every recorded failure, so those keys can be queued again by a
    /// subsequent `mark_wanted`. Ready and in-flight entries are untouched.
    /// Failures are never cleared automatically — that would hammer a down
    /// backend — so this is the hook behind an explicit host "retry".
    pub fn clear_failures(&mut self) {
        self.slots.retain(|_, s| !matches!(s, Slot::Failed(_)));
    }

    /// Forget everything: entries, in-flight slots, the queue, the dirty flag,
    /// undrained failures and arrivals, and the settled high-water map.
    ///
    /// A `load` REPLACES the dataset, so every cached neighborhood now
    /// describes a graph that no longer exists. Keeping them would serve the
    /// old graph's neighbors for any id the new one happens to reuse — and id
    /// reuse is the norm rather than the exception for generated or
    /// synthetically-keyed datasets, where `w0`/`c0` mean something different
    /// in each. Recorded failures are wrong for exactly the same reason, and
    /// worse for being sticky: a node that failed under the old dataset would
    /// be a permanent dead end under the new one.
    ///
    /// `next_epoch` deliberately does NOT restart. Fetches issued for the old
    /// dataset are still on the wire, and rewinding the counter would let one
    /// of them settle under an epoch a post-reset attempt also claims. Instead
    /// the counter's current value becomes an epoch floor: every attempt drawn
    /// before this reset is at or below it, and `is_stale` discards it on
    /// arrival. Without that floor, clearing `slots` and `settled` would
    /// actually WIDEN the hole — both of `fill`'s other guards read those maps,
    /// so an old-dataset answer would land unopposed.
    pub fn reset(&mut self) {
        self.ready.clear();
        self.slots.clear();
        self.wanted.clear();
        self.dirty = false;
        self.failures.clear();
        self.new_arrivals.clear();
        self.settled.clear();
        self.discard_before = Some(self.next_epoch);
    }

    pub fn is_loading(&self) -> bool {
        !self.wanted.is_empty() || self.slots.values().any(|s| matches!(s, Slot::InFlight(_)))
    }

    /// Every result received so far. Used to grow the known graph (the
    /// traditional view starts at the seed and expands as neighborhoods land).
    pub fn ready_results(&self) -> impl Iterator<Item = &NeighborResult> { self.ready.values() }
}

/// A `DataProvider` whose neighbor lookups are served from a `NeighborCache`.
/// A miss returns a `pending` result and queues the key; the host fetches it
/// and calls `NeighborCache::fill` later, at which point the view rebuilds.
pub struct CachingProvider {
    base: Graph,
    cache: Rc<RefCell<NeighborCache>>,
}

impl CachingProvider {
    pub fn new(base: Graph, cache: Rc<RefCell<NeighborCache>>) -> Self {
        Self { base, cache }
    }

    pub fn cache(&self) -> Rc<RefCell<NeighborCache>> { self.cache.clone() }
}

impl DataProvider for CachingProvider {
    fn load(&self) -> Graph { self.base.clone() }

    fn neighbors(&self, focus: &NodeId, params: &QueryParams) -> NeighborResult {
        let k: CacheKey = (focus.clone(), params.cursor.clone());
        if let Some(hit) = self.cache.borrow().get(&k) {
            return hit.clone();
        }
        // A failed key must NOT look pending: `mark_wanted` refuses to re-queue
        // it, so a pending placeholder here would spin forever. Report a
        // resolved-but-empty neighborhood (a dead end) and queue nothing; the
        // host offers a retry, which clears the failure and lets it queue again.
        if self.cache.borrow().error(&k).is_some() {
            return NeighborResult {
                nodes: vec![], edges: vec![], aggregates: vec![], next: None, pending: false,
            };
        }
        self.cache.borrow_mut().mark_wanted(k);
        NeighborResult {
            nodes: vec![], edges: vec![], aggregates: vec![], next: None, pending: true,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Node;

    fn key(id: &str) -> CacheKey { (id.to_string(), None) }

    fn result_with(id: &str) -> NeighborResult {
        NeighborResult {
            nodes: vec![Node { id: id.into(), label: None, attrs: Default::default() }],
            edges: vec![], aggregates: vec![], next: None, pending: false,
        }
    }

    #[test]
    fn miss_queues_the_key_and_reports_loading() {
        let mut c = NeighborCache::new();
        assert!(c.get(&key("a")).is_none());
        c.mark_wanted(key("a"));
        assert!(c.is_loading());
        assert_eq!(c.take_wanted(), vec![(key("a"), 1)]);
    }

    #[test]
    fn in_flight_key_is_not_queued_twice() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        c.mark_wanted(key("a"));            // repeated/held navigation
        assert_eq!(c.take_wanted().len(), 1);
        c.mark_wanted(key("a"));            // still in flight
        assert!(c.take_wanted().is_empty(), "in-flight key must not re-queue");
    }

    #[test]
    fn fill_stores_result_clears_pending_and_sets_dirty() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.fill(key("a"), epoch, result_with("a"));
        assert!(c.take_dirty());
        assert!(!c.take_dirty(), "dirty is consumed");
        assert!(!c.is_loading());
        assert_eq!(c.get(&key("a")).unwrap().nodes[0].id, "a");
    }

    #[test]
    fn failure_is_recorded_and_does_not_retry_loop() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.fail(key("a"), epoch, "boom".into());
        assert!(c.take_dirty());
        assert!(!c.is_loading());
        assert_eq!(c.error(&key("a")), Some("boom"));
        c.mark_wanted(key("a"));
        assert!(c.take_wanted().is_empty(), "failed key must not auto-retry");
    }

    #[test]
    fn distinct_cursors_are_distinct_cache_entries() {
        let mut c = NeighborCache::new();
        let k2 = ("a".to_string(), Some(Cursor("off:8".into())));
        c.fill(key("a"), 0, result_with("a"));
        assert!(c.get(&k2).is_none(), "cursor is part of the key");
    }

    #[test]
    fn ready_results_yields_every_filled_entry() {
        let mut c = NeighborCache::new();
        c.fill(key("a"), 0, result_with("n1"));
        c.fill(("b".to_string(), Some(Cursor("off:8".into()))), 0, result_with("n2"));
        let mut ids: Vec<String> =
            c.ready_results().map(|r| r.nodes[0].id.clone()).collect();
        ids.sort();
        assert_eq!(ids, vec!["n1".to_string(), "n2".to_string()]);
    }

    #[test]
    fn provider_returns_pending_on_miss_and_queues_it() {
        let cache = Rc::new(RefCell::new(NeighborCache::new()));
        let p = CachingProvider::new(Graph::default(), cache.clone());
        let r = p.neighbors(&"a".to_string(), &QueryParams { limit: 8, cursor: None });
        assert!(r.pending);
        assert!(r.nodes.is_empty());
        assert_eq!(cache.borrow_mut().take_wanted(), vec![(key("a"), 1)]);
    }

    #[test]
    fn provider_returns_cached_result_once_filled() {
        let cache = Rc::new(RefCell::new(NeighborCache::new()));
        let p = CachingProvider::new(Graph::default(), cache.clone());
        cache.borrow_mut().fill(key("a"), 0, result_with("n1"));
        let r = p.neighbors(&"a".to_string(), &QueryParams { limit: 8, cursor: None });
        assert!(!r.pending);
        assert_eq!(r.nodes[0].id, "n1");
    }

    #[test]
    fn provider_does_not_queue_a_hit() {
        let cache = Rc::new(RefCell::new(NeighborCache::new()));
        let p = CachingProvider::new(Graph::default(), cache.clone());
        cache.borrow_mut().fill(key("a"), 0, result_with("n1"));
        let _ = p.neighbors(&"a".to_string(), &QueryParams { limit: 8, cursor: None });
        assert!(cache.borrow_mut().take_wanted().is_empty());
    }

    #[test]
    fn provider_reports_a_failed_key_as_resolved_and_empty_not_pending() {
        // A pending placeholder here would spin forever: `mark_wanted` refuses
        // to re-queue a failed key, so nothing would ever resolve it.
        let cache = Rc::new(RefCell::new(NeighborCache::new()));
        let p = CachingProvider::new(Graph::default(), cache.clone());
        cache.borrow_mut().mark_wanted(key("a"));
        let epoch = cache.borrow_mut().take_wanted()[0].1;
        cache.borrow_mut().fail(key("a"), epoch, "boom".into());

        let r = p.neighbors(&"a".to_string(), &QueryParams { limit: 8, cursor: None });
        assert!(!r.pending, "a failed key must not look pending");
        assert!(r.nodes.is_empty());
        assert!(r.next.is_none());
        assert!(cache.borrow_mut().take_wanted().is_empty(), "must not queue a failed key");
    }

    #[test]
    fn clear_failures_lets_a_failed_key_be_queued_again() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.fail(key("a"), epoch, "boom".into());
        c.mark_wanted(key("a"));
        assert!(c.take_wanted().is_empty(), "no auto-retry before an explicit clear");

        c.clear_failures();
        assert_eq!(c.error(&key("a")), None);
        c.mark_wanted(key("a"));
        let requeued = c.take_wanted();
        assert_eq!(requeued.len(), 1, "retry re-queues the key");
        assert_eq!(requeued[0].0, key("a"));
    }

    #[test]
    fn clear_failures_leaves_ready_and_in_flight_alone() {
        let mut c = NeighborCache::new();
        c.fill(key("ready"), 0, result_with("n1"));

        c.mark_wanted(key("inflight"));
        let _ = c.take_wanted();

        c.mark_wanted(key("bad"));
        let epoch = c.take_wanted()[0].1;
        c.fail(key("bad"), epoch, "boom".into());

        c.clear_failures();
        assert!(c.get(&key("ready")).is_some(), "ready entries survive");
        assert!(c.is_loading(), "the in-flight slot survives");
        c.mark_wanted(key("inflight"));
        assert!(c.take_wanted().is_empty(), "still in flight, so still not re-queued");
    }

    #[test]
    fn fill_never_trusts_pending_from_the_wire() {
        let cache = Rc::new(RefCell::new(NeighborCache::new()));
        let mut res = result_with("n1");
        res.pending = true; // a proxy/host echoing `{"pending":true}`
        cache.borrow_mut().fill(key("a"), 0, res);
        assert!(!cache.borrow().get(&key("a")).unwrap().pending, "pending is client-side only");

        // otherwise this hit would report pending forever, and `mark_wanted`
        // would skip the key because it IS ready.
        let p = CachingProvider::new(Graph::default(), cache.clone());
        let r = p.neighbors(&"a".to_string(), &QueryParams { limit: 8, cursor: None });
        assert!(!r.pending);
        assert_eq!(r.nodes[0].id, "n1");
    }

    #[test]
    fn new_arrivals_drain_once_and_report_each_filled_key() {
        let mut c = NeighborCache::new();
        let k2 = ("b".to_string(), Some(Cursor("off:8".into())));
        c.fill(key("a"), 0, result_with("n1"));
        c.fill(k2.clone(), 0, result_with("n2"));

        let mut drained = c.take_new_arrivals();
        drained.sort_by(|x, y| x.0.cmp(&y.0));
        assert_eq!(drained, vec![key("a"), k2]);
        assert!(c.take_new_arrivals().is_empty(), "draining consumes the arrivals");
    }

    #[test]
    fn ids_containing_the_separator_do_not_collide_with_cursor_keys() {
        // ("a\u{1}b", None) and ("a", Some("b")) both flattened to the same
        // string before the key encoding was length-prefixed.
        let mut c = NeighborCache::new();
        let k1 = ("a\u{1}b".to_string(), None);
        let k2 = ("a".to_string(), Some(Cursor("b".into())));
        c.fill(k1.clone(), 0, result_with("from-k1"));
        c.fill(k2.clone(), 0, result_with("from-k2"));
        assert_eq!(c.get(&k1).unwrap().nodes[0].id, "from-k1");
        assert_eq!(c.get(&k2).unwrap().nodes[0].id, "from-k2");
    }

    #[test]
    fn failures_are_drainable_once() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.fail(key("a"), epoch, "boom".into());

        let drained = c.take_failures();
        assert_eq!(drained.len(), 1);
        assert_eq!(drained[0].0, key("a"), "structured key is preserved");
        assert_eq!(drained[0].1, "boom");

        assert!(c.take_failures().is_empty(), "draining consumes the failures");
        // the failure is still recorded, so the key does not auto-retry
        assert_eq!(c.error(&key("a")), Some("boom"));
        c.mark_wanted(key("a"));
        assert!(c.take_wanted().is_empty());
    }

    #[test]
    fn failure_with_a_cursor_keeps_its_cursor_in_the_drained_key() {
        let mut c = NeighborCache::new();
        let k = ("a".to_string(), Some(Cursor("grp:X:0".into())));
        c.mark_wanted(k.clone());
        let epoch = c.take_wanted()[0].1;
        c.fail(k.clone(), epoch, "nope".into());
        let drained = c.take_failures();
        assert_eq!(drained[0].0, k);
    }

    #[test]
    fn cancel_clears_the_in_flight_slot_and_stops_reporting_loading() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let _ = c.take_wanted();
        assert!(c.is_loading());

        c.cancel(&key("a"));
        assert!(!c.is_loading(), "a cancelled fetch must not keep the spinner up");
        assert!(!c.is_in_flight(&key("a")));
    }

    #[test]
    fn fail_after_cancel_records_nothing() {
        // An aborted fetch rejects its promise. Recording that as a failure
        // would install a Slot::Failed, which mark_wanted refuses to re-queue —
        // the node would be a permanent dead end for a cancellation we asked for.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.cancel(&key("a"));

        c.fail(key("a"), epoch, "AbortError".into());
        assert_eq!(c.error(&key("a")), None, "no failure is recorded");
        assert!(c.take_failures().is_empty(), "nothing is reported to the host");
        assert!(!c.take_dirty(), "a cancellation is not a repaint-worthy change");
    }

    #[test]
    fn a_cancelled_key_can_be_requested_again() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.cancel(&key("a"));
        c.fail(key("a"), epoch, "AbortError".into());

        c.mark_wanted(key("a"));
        let requeued = c.take_wanted();
        assert_eq!(requeued.len(), 1, "cancellation is not sticky");
        assert_eq!(requeued[0].0, key("a"));
    }

    #[test]
    fn fill_after_cancel_still_stores_the_result() {
        // The bytes arrived anyway. Data keyed by node id is always valid, so
        // keep it — revisiting the node is then a hit with no refetch. There
        // is no separate "cancelled" flag left lying around to swallow
        // anything later (see `genuine_failure_on_a_re_requested_attempt_
        // after_cancel_is_recorded` for that guarantee on a re-fetched key).
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.cancel(&key("a"));

        c.fill(key("a"), epoch, result_with("n1"));
        assert_eq!(c.get(&key("a")).unwrap().nodes[0].id, "n1");
        assert!(!c.is_in_flight(&key("a")));
    }

    #[test]
    fn cancel_drops_a_key_queued_but_not_yet_drained() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        c.cancel(&key("a"));
        assert!(c.take_wanted().is_empty(), "a queued key is dequeued, not fetched");
        assert!(!c.is_loading());
    }

    #[test]
    fn cancel_of_an_unknown_key_is_a_no_op() {
        let mut c = NeighborCache::new();
        c.cancel(&key("ghost"));
        assert!(!c.is_loading());
        // and it must not arm the swallow-the-next-failure behaviour
        c.mark_wanted(key("ghost"));
        let epoch = c.take_wanted()[0].1;
        c.fail(key("ghost"), epoch, "boom".into());
        assert_eq!(c.error(&key("ghost")), Some("boom"));
    }

    #[test]
    fn is_in_flight_is_true_only_between_take_wanted_and_settle() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        assert!(!c.is_in_flight(&key("a")), "queued is not yet in flight");
        let epoch = c.take_wanted()[0].1;
        assert!(c.is_in_flight(&key("a")));
        c.fill(key("a"), epoch, result_with("n1"));
        assert!(!c.is_in_flight(&key("a")));
    }

    #[test]
    fn cache_keys_are_hashable_so_hosts_can_index_by_them() {
        // graph-explorer-wasm keys its AbortController map by CacheKey.
        use std::collections::HashMap;
        let mut m: HashMap<CacheKey, u32> = HashMap::new();
        m.insert(key("a"), 1);
        m.insert(("a".to_string(), Some(Cursor("off:8".into()))), 2);
        assert_eq!(m.get(&key("a")), Some(&1));
        assert_eq!(m.len(), 2, "the cursor is part of the identity");
    }

    // --- epoch-tagged slots: bugs found in code review of the first cut ---

    #[test]
    fn cancel_requeue_cancel_does_not_poison_the_key() {
        // Two outstanding attempts for the same key must not collide: a
        // `HashSet<String>` "cancelled" mark could only remember one, so the
        // second attempt's own (also-discarded) rejection would wrongly
        // record a Slot::Failed and permanently dead-end the node. The epoch
        // tag distinguishes them.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch1 = c.take_wanted()[0].1; // P1
        c.cancel(&key("a"));

        c.mark_wanted(key("a"));
        let epoch2 = c.take_wanted()[0].1; // P2, a fresh attempt
        c.cancel(&key("a"));

        c.fail(key("a"), epoch1, "AbortError".into()); // P1's rejection: stale
        c.fail(key("a"), epoch2, "AbortError".into()); // P2's rejection: also stale

        assert_eq!(c.error(&key("a")), None, "neither cancelled attempt records a failure");
        c.mark_wanted(key("a"));
        assert_eq!(c.take_wanted().len(), 1, "the key is not poisoned and can be requested again");
    }

    #[test]
    fn genuine_failure_on_a_re_requested_attempt_after_cancel_is_recorded() {
        // The swallow-path must not blindly return before touching `slots`:
        // that would orphan an InFlight slot from an earlier cancelled
        // attempt and wedge `is_loading()` true forever once the live
        // attempt's own failure arrives.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch1 = c.take_wanted()[0].1; // P1
        c.cancel(&key("a"));

        c.mark_wanted(key("a"));
        let epoch2 = c.take_wanted()[0].1; // P2, the live attempt
        assert!(c.is_loading());

        c.fail(key("a"), epoch1, "boom".into()); // P1's rejection: stale, discarded
        assert!(c.is_loading(), "P2 is still outstanding");
        assert_eq!(c.error(&key("a")), None);

        c.fail(key("a"), epoch2, "boom".into()); // P2's own, genuine failure
        assert!(!c.is_loading(), "no orphaned slot: is_loading must go quiet");
        assert_eq!(c.error(&key("a")), Some("boom"));
    }

    #[test]
    fn cancel_of_an_already_failed_key_leaves_the_failure_intact() {
        // Cancelling must be gated on `Slot::InFlight`, not on "any slot at
        // all" — otherwise cancel is a backdoor around "failures are never
        // cleared automatically", letting a down backend be hammered.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.fail(key("a"), epoch, "boom".into());

        c.cancel(&key("a")); // nothing in flight to cancel

        assert_eq!(c.error(&key("a")), Some("boom"), "cancel must not erase a recorded failure");
        c.mark_wanted(key("a"));
        assert!(c.take_wanted().is_empty(), "must not re-queue a failed key via cancel's back door");
    }

    #[test]
    fn stale_fill_from_a_superseded_attempt_is_discarded() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch1 = c.take_wanted()[0].1; // P1
        c.cancel(&key("a"));

        c.mark_wanted(key("a"));
        let epoch2 = c.take_wanted()[0].1; // P2, now the attempt of record

        c.fill(key("a"), epoch1, result_with("stale")); // P1's late answer
        assert!(c.get(&key("a")).is_none(), "a stale fill must not overwrite the live attempt");
        assert!(c.is_in_flight(&key("a")), "P2 is still outstanding");

        c.fill(key("a"), epoch2, result_with("live"));
        assert_eq!(c.get(&key("a")).unwrap().nodes[0].id, "live");
    }

    #[test]
    fn cancel_of_a_ready_key_is_a_no_op() {
        let mut c = NeighborCache::new();
        c.fill(key("a"), 0, result_with("n1"));
        c.cancel(&key("a"));
        assert_eq!(c.get(&key("a")).unwrap().nodes[0].id, "n1", "a ready result is untouched by cancel");
        c.mark_wanted(key("a"));
        assert!(c.take_wanted().is_empty(), "a ready key is still a hit, not re-queued");
    }

    #[test]
    fn absent_cursor_and_empty_cursor_are_distinct_cache_entries() {
        let mut c = NeighborCache::new();
        let empty_cursor_key = ("a".to_string(), Some(Cursor(String::new())));
        c.fill(key("a"), 0, result_with("no-cursor"));
        c.fill(empty_cursor_key.clone(), 0, result_with("empty-cursor"));

        assert_eq!(c.get(&key("a")).unwrap().nodes[0].id, "no-cursor");
        assert_eq!(c.get(&empty_cursor_key).unwrap().nodes[0].id, "empty-cursor");
    }

    // --- `settled`: fill's guard has no memory once the slot is gone ---

    #[test]
    fn late_stale_success_does_not_overwrite_a_newer_settled_success() {
        // `fill`'s guard used to compare only against a live `InFlight` slot.
        // Once P2 settles and removes the slot, P1's late success fell
        // through unguarded and stomped the fresher answer.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("b"));
        let epoch1 = c.take_wanted()[0].1; // P1
        c.cancel(&key("b"));

        c.mark_wanted(key("b"));
        let epoch2 = c.take_wanted()[0].1; // P2

        c.fill(key("b"), epoch2, result_with("new")); // the live attempt settles first
        let _ = c.take_new_arrivals();

        c.fill(key("b"), epoch1, result_with("old")); // P1's late, stale success
        assert_eq!(
            c.get(&key("b")).unwrap().nodes[0].id,
            "new",
            "a stale success must not overwrite a newer one"
        );
        assert!(
            c.take_new_arrivals().is_empty(),
            "a discarded stale fill must not be reported as a new arrival"
        );
    }

    #[test]
    fn late_stale_success_does_not_erase_a_recorded_failure() {
        // Same gap, different victim: a stale `fill` falling through the
        // guard would remove a `Slot::Failed` that a newer attempt recorded
        // and store data in its place -- a second automatic path around
        // `clear_failures`, resurrecting a key the host was already told
        // had failed.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch1 = c.take_wanted()[0].1; // P1
        c.cancel(&key("a"));

        c.mark_wanted(key("a"));
        let epoch2 = c.take_wanted()[0].1; // P2, the live attempt

        c.fail(key("a"), epoch2, "boom".into()); // P2's genuine failure
        let _ = c.take_failures();
        assert_eq!(c.error(&key("a")), Some("boom"));

        c.fill(key("a"), epoch1, result_with("late")); // P1's late, stale success
        assert_eq!(
            c.error(&key("a")),
            Some("boom"),
            "a stale success must not erase a recorded failure"
        );
        assert!(c.get(&key("a")).is_none(), "the failed key must not also look like a hit");
    }

    #[test]
    fn fill_after_cancel_with_nothing_settled_since_still_stores() {
        // The deliberate "keep the bytes" case must survive the new guard:
        // nothing newer is in flight and nothing has settled since, so this
        // settle is not stale even though its slot is long gone.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.cancel(&key("a"));

        c.fill(key("a"), epoch, result_with("n1"));
        assert_eq!(c.get(&key("a")).unwrap().nodes[0].id, "n1");
    }

    // --- `reset`: a load replaces the dataset ---

    #[test]
    fn reset_clears_everything_the_cache_holds() {
        let mut c = NeighborCache::new();

        // a ready entry
        c.mark_wanted(key("ready"));
        let e_ready = c.take_wanted()[0].1;
        c.fill(key("ready"), e_ready, result_with("n1"));
        // a failure
        c.mark_wanted(key("bad"));
        let e_bad = c.take_wanted()[0].1;
        c.fail(key("bad"), e_bad, "boom".into());
        // an in-flight slot, and a key still only queued
        c.mark_wanted(key("inflight"));
        let _ = c.take_wanted();
        c.mark_wanted(key("queued"));

        c.reset();

        assert!(c.get(&key("ready")).is_none(), "ready entries describe the old graph");
        assert_eq!(c.error(&key("bad")), None, "failures are not carried across a load");
        assert!(!c.is_loading(), "no in-flight slots and nothing queued survive");
        assert!(!c.take_dirty(), "the dirty flag is cleared");
        assert!(c.take_failures().is_empty(), "undrained failures are dropped");
        assert!(c.take_new_arrivals().is_empty(), "undrained arrivals are dropped");
        assert_eq!(c.ready_results().count(), 0);
    }

    #[test]
    fn a_key_settled_before_a_reset_can_be_requested_again() {
        // The whole point: the new dataset reuses the id, so the node must be
        // a miss that fetches — not a hit serving the old graph's neighbors,
        // and not a `settled` high-water mark blocking the fresh attempt.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("w0"));
        let epoch = c.take_wanted()[0].1;
        c.fill(key("w0"), epoch, result_with("old-neighbor"));
        assert!(c.get(&key("w0")).is_some());

        c.reset();

        c.mark_wanted(key("w0"));
        let requeued = c.take_wanted();
        assert_eq!(requeued.len(), 1, "a reused id must fetch again, not hit");
        assert_eq!(requeued[0].0, key("w0"));

        c.fill(key("w0"), requeued[0].1, result_with("new-neighbor"));
        assert_eq!(c.get(&key("w0")).unwrap().nodes[0].id, "new-neighbor");
    }

    #[test]
    fn a_key_that_failed_before_a_reset_is_not_a_permanent_dead_end() {
        let mut c = NeighborCache::new();
        c.mark_wanted(key("w0"));
        let epoch = c.take_wanted()[0].1;
        c.fail(key("w0"), epoch, "boom".into());
        c.mark_wanted(key("w0"));
        assert!(c.take_wanted().is_empty(), "failed keys do not auto-retry pre-reset");

        c.reset();

        c.mark_wanted(key("w0"));
        assert_eq!(c.take_wanted().len(), 1, "the new dataset's node is fetchable");
    }

    #[test]
    fn a_fetch_outstanding_across_a_reset_cannot_settle_into_the_new_dataset() {
        // The abort is best-effort — a request already past the wire settles
        // anyway. Its answer describes the OLD graph, so it must be discarded
        // even though `reset` wiped the slot and `settled` entry that `fill`'s
        // other guards would have caught it with.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("w0"));
        let old_epoch = c.take_wanted()[0].1;

        c.reset();

        c.fill(key("w0"), old_epoch, result_with("old-neighbor"));
        assert!(c.get(&key("w0")).is_none(), "an old-dataset answer must not land");
        assert!(c.take_new_arrivals().is_empty(), "nor be reported as an arrival");

        c.fail(key("w0"), old_epoch, "boom".into());
        assert_eq!(c.error(&key("w0")), None, "nor poison the reused id with its failure");
    }

    #[test]
    fn epochs_keep_climbing_across_a_reset() {
        // Rewinding the counter would let a pre-reset attempt settle under an
        // epoch a post-reset attempt also claims, making the two
        // indistinguishable.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let before = c.take_wanted()[0].1;

        c.reset();

        c.mark_wanted(key("a"));
        let after = c.take_wanted()[0].1;
        assert!(after > before, "epochs are monotonic across a reset ({after} > {before})");
    }

    #[test]
    fn reset_on_a_fresh_cache_imposes_no_floor_on_later_attempts() {
        // `reset` records the counter's CURRENT value, so a reset before any
        // fetch has been drawn must not disqualify the fetches that follow.
        let mut c = NeighborCache::new();
        c.reset();

        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.fill(key("a"), epoch, result_with("n1"));
        assert_eq!(c.get(&key("a")).unwrap().nodes[0].id, "n1");
    }

    #[test]
    fn duplicate_fill_with_the_same_epoch_is_discarded() {
        // A redelivered settle of the very same attempt is not a fresh
        // answer: it must not re-store (harmless here, since the data is
        // identical) or re-report via `take_new_arrivals` as if new.
        let mut c = NeighborCache::new();
        c.mark_wanted(key("a"));
        let epoch = c.take_wanted()[0].1;
        c.fill(key("a"), epoch, result_with("first"));
        let _ = c.take_new_arrivals();

        c.fill(key("a"), epoch, result_with("duplicate"));
        assert_eq!(
            c.get(&key("a")).unwrap().nodes[0].id,
            "first",
            "a duplicate delivery of the same attempt does not overwrite"
        );
        assert!(
            c.take_new_arrivals().is_empty(),
            "a duplicate settle is not reported as a new arrival"
        );
    }
}