libdictenstein 4.0.0-rc.1

//! `ARTrie` + `EvictableARTrie` + `Dictionary` + `MappedDictionary` trait
//! implementations for `SharedARTrie<V>`.
//!
//! Split out of byte `dict_impl.rs` (lines ~5675-6074, ~400 LOC) as
//! the tenth Phase-5 byte sub-module. The trait blocks plus the
//! evict-node helpers (`evict_node_at_path` + `find_parent_mut`) move
//! here; the per-method semantics are unchanged.

use std::path::Path;
use std::sync::atomic::Ordering as AtomicOrdering;
use std::sync::Arc;

use crate::artrie_trait::{ARTrie, EvictableARTrie};
use crate::persistent_artrie::core::durability::DurabilityPolicy;
use crate::persistent_artrie::core::eviction::{
    EvictionConfig, EvictionCoordinator, EvictionStats,
};
// F4: the `.read()/.write()` compat shim on the collapsed `Arc<PersistentARTrie>`.
use crate::persistent_artrie::core::shared_access::SharedTrieAccess;
use crate::value::DictionaryValue;
use crate::{Dictionary, MappedDictionary, SyncStrategy};

use super::dict_impl::PersistentARTrie;
use super::error::{PersistentARTrieError, Result};
use super::node_impl::PersistentARTrieNode;
use super::recovery::RecoveryReport;
use super::SharedARTrie;

impl<V: DictionaryValue> ARTrie for SharedARTrie<V> {
    type Unit = u8;
    type Value = V;

    fn create<P: AsRef<Path>>(path: P) -> Result<Self> {
        PersistentARTrie::create(path).map(Arc::new)
    }

    fn create_with_slot_tracking<P: AsRef<Path>>(path: P) -> Result<Self> {
        PersistentARTrie::create_with_slot_tracking(path).map(Arc::new)
    }

    fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
        PersistentARTrie::open(path).map(Arc::new)
    }

    fn open_with_slot_tracking<P: AsRef<Path>>(path: P) -> Result<Self> {
        PersistentARTrie::open_with_slot_tracking(path).map(Arc::new)
    }

    fn open_with_recovery<P: AsRef<Path>>(path: P) -> Result<(Self, RecoveryReport)> {
        PersistentARTrie::open_with_recovery(path).map(|(t, r)| (Arc::new(t), r))
    }

    fn open_with_recovery_and_slot_tracking<P: AsRef<Path>>(
        path: P,
    ) -> Result<(Self, RecoveryReport)> {
        let (trie, report) = PersistentARTrie::open_with_recovery(path)?;
        if let Some(ref am) = trie.arena_manager {
            am.write().enable_slot_tracking();
        }
        Ok((Arc::new(trie), report))
    }

    fn enable_slot_tracking(&self) {
        let guard = self.read();
        if let Some(ref am) = guard.arena_manager {
            am.write().enable_slot_tracking();
        }
    }

    fn flush_sequential(&self) -> Result<()> {
        let guard = self.read();
        if let Some(ref am) = guard.arena_manager {
            am.write().flush_sequential()?;
        }
        Ok(())
    }

    fn insert(&self, term: &str) -> bool
    where
        Self::Value: Default,
    {
        // L3.3c: the overlay is the sole representation; route to the routed inherent
        // `insert` (→ `insert_cas_durable`). The durable membership insert is value-free.
        self.write().insert(term)
    }

    fn insert_with_value(&self, term: &str, value: Self::Value) -> bool {
        // L3.3c: route to the routed inherent `insert_with_value` (overlay upsert).
        self.write().insert_with_value(term, value)
    }

    fn contains(&self, term: &str) -> bool {
        // M3 (C6): delegate to the routed `contains_bytes` (this read `contains_impl`
        // directly, bypassing the overlay route).
        let guard = self.read();
        guard.contains_bytes(term.as_bytes())
    }

    fn get_value(&self, term: &str) -> Option<Self::Value> {
        // M3 (C6): delegate to the routed `get_value_bytes` (value-routes to the
        // overlay incl. the empty-term owned exception), NOT `get_value_impl`.
        let guard = self.read();
        guard.get_value_bytes(term.as_bytes())
    }

    fn remove(&self, term: &str) -> bool {
        // L3.3c: route to the routed inherent `remove` (→ `remove_cas_durable`).
        self.write().remove(term)
    }

    #[inline]
    fn len(&self) -> usize {
        // The lock-free overlay is the SOLE representation, so count its resident finals.
        // (The owned `term_count` is no longer maintained — it was cleared on reopen — and
        // `route_overlay()` is universally true, so the old owned-fallback branch was dead.)
        self.read().overlay_len()
    }

    fn checkpoint(&self) -> Result<()> {
        // **F3 / NF-3 — serialize concurrent checkpoints** (byte twin). Today byte's
        // checkpoint holds the outer `self.write()` for the whole body, so checkpoints
        // are ALREADY serialized and this lock is redundant-but-harmless; but the F4
        // `Arc<RwLock>`→`Arc` collapse drops the write lock, and this `checkpoint_lock`
        // then becomes the sole serializer (forward-correct; same lock the char arm
        // uses). Cloned out of a brief read guard so we don't hold the trie lock while
        // acquiring it. Formally verified (ConcurrentCheckpointSerialization.tla).
        let ckpt_lock = self.read().checkpoint_lock.clone();
        let _ckpt_guard = ckpt_lock.lock();
        let guard = self.write();
        guard.checkpoint()
    }

    #[inline]
    fn is_dirty(&self) -> bool {
        let guard = self.read();
        guard.dirty.load(AtomicOrdering::Acquire)
    }

    fn remove_prefix(&self, prefix: &str) -> usize {
        // L3.3c: the overlay is the sole representation; route to the routed inherent
        // `remove_prefix_batched` (overlay remove-CAS).
        self.write().remove_prefix_batched(prefix.as_bytes(), 1024)
    }

    fn iter_prefix(&self, prefix: &str) -> Option<Box<dyn Iterator<Item = String> + '_>> {
        // M3 (C6): `iter_prefix_with_arena` is routed at its public top, so this trait
        // body is overlay-routed automatically under the flip (the terms come from the
        // overlay; lossy-UTF8 mapping is unchanged).
        let guard = self.read();
        let terms = guard.iter_prefix_with_arena(prefix.as_bytes()).ok()??;
        Some(Box::new(
            terms
                .into_iter()
                .map(|t| String::from_utf8_lossy(&t.term).into_owned()),
        ))
    }

    fn sync(&self) -> Result<()> {
        let guard = self.read();
        guard.sync()
    }

    fn current_lsn(&self) -> u64 {
        let guard = self.read();
        guard.current_lsn()
    }

    fn synced_lsn(&self) -> Option<u64> {
        let guard = self.read();
        guard.synced_lsn()
    }

    fn durability_policy(&self) -> DurabilityPolicy {
        let guard = self.read();
        guard.durability_policy()
    }

    fn upsert(&self, term: &str, value: Self::Value) -> Result<bool> {
        let guard = self.write();
        guard.upsert(term, value)
    }

    // C1: `increment` removed from the `ARTrie` trait (now an inherent `V: Counter`
    // method on PersistentARTrie). Delegation commented out (not deleted) per
    // convention; counter callers use the inner inherent method, e.g.
    // `trie.write().increment(..)` on a `<i64>`/`<u64>` trie.
    // fn increment(&self, term: &str, delta: i64) -> Result<i64> {
    //     let guard = self.write();
    //     guard.increment(term, delta)
    // }
}

impl<V: DictionaryValue> EvictableARTrie for SharedARTrie<V> {
    fn enable_eviction(&self, config: EvictionConfig) -> Result<()> {
        config
            .validate()
            .map_err(|e| PersistentARTrieError::internal(&e))?;

        // F4 (EC leaf): the coordinator field is a `Mutex<Option<Arc<…>>>`. Check +
        // install under a BRIEF EC lock; the coordinator is fully built + started
        // OUTSIDE the lock so EC is never held across thread spawns or any other
        // lock. Already-enabled ⇒ error (no old Arc to drop, so no re-arm join).
        if self
            .eviction_coordinator
            .lock()
            .expect("eviction_coordinator mutex poisoned")
            .is_some()
        {
            return Err(PersistentARTrieError::internal("Eviction already enabled"));
        }

        // Phase 6 (byte epoch-share, mirror char): SHARE this trie's OWN epoch manager
        // with the coordinator (was a SEPARATE `Arc::new(EpochManager::new())`). The
        // field is now `Arc<EpochManager>`, and `SharedARTrie = Arc<PersistentARTrie>`
        // derefs to it directly. The overlay read/write paths + the lifted overlay
        // evictor (`OverlayEvictable::{find_leaf_faulting, evict_overlay_node_at_path}`)
        // pin THIS same manager via `enter_read`, so the coordinator's quiescence drain
        // genuinely waits on the live overlay readers (honest reader accounting; not a
        // correctness change — overlay reclamation is by `Arc` refcount, not EBR).
        let epoch_manager = Arc::clone(&self.epoch_manager);
        let coordinator = EvictionCoordinator::new(config.clone(), epoch_manager);
        let self_weak = Arc::downgrade(self);

        coordinator
            .start(move |nodes_to_evict| {
                let Some(trie) = self_weak.upgrade() else {
                    return (0, 0);
                };
                // Phase 7.5: route_overlay-GATED. Under the overlay regime reclaim the
                // OVERLAY (the inline evict_node_at_path owned loop below is a no-op on the
                // EMPTY owned tree there); in owned mode keep the proven owned-tree loop
                // (preserves owned + ineligible-V eviction). evict_overlay_nodes locks EC
                // for its LRU remove — safe here (the loop holds no EC, same as the owned
                // loop's EC discipline).
                // L0.1/L3.3: always reclaim the overlay (the owned tree is gone).
                // `evict_overlay_nodes` locks EC for its LRU remove; safe here (this
                // callback holds no EC).
                crate::persistent_artrie::overlay_fault::evict_overlay_nodes(
                    &trie,
                    nodes_to_evict,
                    4,
                )
            })
            .map_err(|e| PersistentARTrieError::internal(&e))?;

        coordinator
            .start_memory_monitor()
            .map_err(|e| PersistentARTrieError::internal(&e))?;

        // Install under a brief EC lock (re-check in case of a concurrent enable —
        // first writer wins; a loser shuts its own coordinator down outside EC).
        let mut slot = self
            .eviction_coordinator
            .lock()
            .expect("eviction_coordinator mutex poisoned");
        if slot.is_some() {
            drop(slot);
            coordinator.shutdown();
            return Err(PersistentARTrieError::internal("Eviction already enabled"));
        }
        *slot = Some(coordinator);
        Ok(())
    }

    fn disable_eviction(&self) -> Result<()> {
        // **F4 drop-before-join (GAP 1 / V11.3 site 1):** take the coordinator out
        // of the EC `Mutex` into a statement-temporary so the EC guard DROPS before
        // `shutdown()` joins the eviction thread — the eviction callback takes OR
        // (and briefly EC), so joining while holding EC would deadlock (the worker
        // waits on EC; disable holds EC + joins).
        let coordinator = self
            .eviction_coordinator
            .lock()
            .expect("eviction_coordinator mutex poisoned")
            .take();
        // EC guard dropped here.
        if let Some(coordinator) = coordinator {
            coordinator.shutdown();
        }
        Ok(())
    }

    fn eviction_enabled(&self) -> bool {
        self.eviction_coordinator
            .lock()
            .expect("eviction_coordinator mutex poisoned")
            .is_some()
    }

    fn eviction_stats(&self) -> EvictionStats {
        self.eviction_coordinator
            .lock()
            .expect("eviction_coordinator mutex poisoned")
            .as_ref()
            .map(|c| c.stats())
            .unwrap_or_default()
    }

    fn force_eviction(&self, target_bytes: usize) -> Result<(usize, usize)> {
        // Clone the coordinator Arc out under a BRIEF EC lock, then release EC
        // before `force_eviction` (whose reclaim callback takes OR — order OR > EC).
        let coordinator = {
            match self
                .eviction_coordinator
                .lock()
                .expect("eviction_coordinator mutex poisoned")
                .as_ref()
            {
                Some(c) => Arc::clone(c),
                None => return Ok((0, 0)),
            }
        };
        // L0.1: always reclaim the OVERLAY (the owned select-and-count arm was deleted).
        // `force_eviction_bytes` returns the EVICTED count, not the candidate count.
        let trie = Arc::clone(self);
        Ok(
            coordinator.force_eviction_bytes(target_bytes, move |nodes| {
                crate::persistent_artrie::overlay_fault::evict_overlay_nodes(&trie, nodes, 4)
            }),
        )
    }

    fn touch_node(&self, path: &[Self::Unit]) {
        if let Some(coordinator) = self
            .eviction_coordinator
            .lock()
            .expect("eviction_coordinator mutex poisoned")
            .as_ref()
        {
            coordinator.lru_registry().touch(path);
        }
    }
}

// ============================================================================
// B1: `Dictionary` / `MappedDictionary` for the `Arc` handle `SharedARTrie<V>`
// ============================================================================
//
// The bare `PersistentARTrie<V, S>` implements `Dictionary`/`MappedDictionary`
// (see `dictionary_traits.rs`) but is NOT `Clone` — it owns the WAL writer, the
// mmap buffer manager, the arena, and atomic counters. The `Arc` handle
// `SharedARTrie<V>` is `Clone` but, until now, carried only `ARTrie` /
// `EvictableARTrie` (above). Consumers that need BOTH `MappedDictionary` AND
// `Clone` on a single type — e.g. a value-returning `Transducer` driven over an
// `Arc`-shared count store — therefore had no usable byte type (the split is the
// "B1 blocker").
//
// These two impls close the gap by delegating through the no-lock `read()` shim
// to the bare trie's own `Dictionary` / `MappedDictionary` bodies. This is the
// exact analog of the `SharedVocabARTrie` impls (`vocab/mod.rs`, "the impl lives
// on the `Arc` handle, whose `Clone` satisfies the bound") and the
// `DynamicDawgU64` value impls. Method resolution is unambiguous: the guard
// derefs to `&PersistentARTrie` (which does NOT implement `ARTrie` — only the
// `Arc` does), and the bare trie has no inherent `root`/`contains`/`len`/
// `get_value`, so each call binds to the `Dictionary`/`MappedDictionary` method.
impl<V: DictionaryValue> Dictionary for SharedARTrie<V> {
    type Node = PersistentARTrieNode<V>;

    fn root(&self) -> Self::Node {
        // Delegates to the bare trie's overlay-backed `Dictionary::root`. The
        // returned node OWNS its overlay-root `Arc`, so it outlives the transient
        // no-lock `read()` guard (`SharedTrieAccess` hands back `&PersistentARTrie`).
        self.read().root()
    }

    fn contains(&self, term: &str) -> bool {
        self.read().contains(term)
    }

    fn len(&self) -> Option<usize> {
        self.read().len()
    }

    fn sync_strategy(&self) -> SyncStrategy {
        // The persistent ARTrie syncs internally (WAL), so it reports
        // `InternalSync`. Omitting this override would default to `ExternalSync`
        // and silently change the transducer's synchronization contract vs. the
        // single-trie path.
        self.read().sync_strategy()
    }
}

impl<V: DictionaryValue> MappedDictionary for SharedARTrie<V> {
    type Value = V;

    fn get_value(&self, term: &str) -> Option<Self::Value> {
        self.read().get_value(term)
    }
}

#[cfg(test)]
mod b1_shared_dictionary_tests {
    use super::PersistentARTrie;
    use super::SharedARTrie;
    use crate::artrie_trait::ARTrie;
    use crate::{Dictionary, DictionaryNode, MappedDictionary, MappedDictionaryNode};
    use std::sync::Arc;

    /// B1 (functional): a `SharedARTrie<u64>` populated through the `Arc` answers
    /// the full `Dictionary` + `MappedDictionary` surface — `len`, `contains`,
    /// `get_value`, and an overlay-backed `root()` walk. The `transition(*b)` calls
    /// take a `u8` label, so this also operationally witnesses the byte node's
    /// `Unit = u8` projection (the carrier `U64NgramView` needs on the grammstein side).
    #[test]
    fn shared_artrie_u64_is_a_mapped_dictionary_through_the_arc() {
        let dir = tempfile::TempDir::new().expect("temp dir");
        let path = dir.path().join("b1.part");
        let trie: SharedARTrie<u64> =
            Arc::new(PersistentARTrie::<u64>::create(&path).expect("create"));

        assert!(trie.insert_with_value("the", 100));
        assert!(trie.insert_with_value("the quick", 42));
        assert!(trie.insert_with_value("the quick brown", 7));

        // Dictionary surface through the Arc handle.
        assert_eq!(Dictionary::len(&trie), Some(3));
        assert!(Dictionary::contains(&trie, "the quick"));
        assert!(!Dictionary::contains(&trie, "absent"));

        // MappedDictionary surface through the Arc handle.
        assert_eq!(MappedDictionary::get_value(&trie, "the"), Some(100));
        assert_eq!(MappedDictionary::get_value(&trie, "the quick"), Some(42));
        assert_eq!(
            MappedDictionary::get_value(&trie, "the quick brown"),
            Some(7)
        );
        assert_eq!(MappedDictionary::get_value(&trie, "absent"), None);

        // Root walk: byte node projects `Unit = u8`, `Value = u64`.
        let mut node = Dictionary::root(&trie);
        for b in b"the" {
            node = DictionaryNode::transition(&node, *b).expect("edge 't'/'h'/'e' exists");
        }
        assert!(
            DictionaryNode::is_final(&node),
            "'the' is a stored final node"
        );
        assert_eq!(MappedDictionaryNode::value(&node), Some(100));
    }

    /// B1 + M1 (type-level witness): `SharedARTrie<u64>` satisfies EXACTLY the
    /// bound the libgrammstein `GrammarCorrector<D>` / `U64NgramView` chain
    /// requires — `MappedDictionary<Value = u64> + Clone + Send + Sync + 'static`
    /// with the node projecting `Value = u64`. If this compiles, B1 is resolved.
    /// (The `Unit: VarintByteUnit` half is checked on the grammstein side, where
    /// that sealed trait lives; here the functional test above pins `Unit = u8`.)
    #[test]
    fn shared_artrie_u64_satisfies_the_grammar_corrector_bound() {
        fn requires_clone_mapped_dict<D>()
        where
            D: MappedDictionary<Value = u64> + Clone + Send + Sync + 'static,
            <D as Dictionary>::Node: MappedDictionaryNode<Value = u64>,
        {
        }
        requires_clone_mapped_dict::<SharedARTrie<u64>>();
    }
}