Skip to main content

armdb/
zero_map.rs

1use std::hash::Hash;
2use std::marker::PhantomData;
3use std::ops::{Bound, ControlFlow};
4use std::sync::Arc;
5
6use crate::Key;
7
8use crate::compaction::CompactionIndex;
9use crate::config::Config;
10use crate::const_map::{ConstMap, ConstMapShard};
11use crate::durability::{Bitcask, Durability, Fixed};
12use crate::error::{DbError, DbResult};
13use crate::fixed::config::FixedConfig;
14use crate::hook::{NoHook, TypedWriteHook, WriteHook, ZeroHookAdapter};
15use crate::map_index::{Dir, KeyPage, MapIterView, Page, ShardSource, TryMapIter, dir_from};
16use crate::projection::IndexKeyed;
17use crate::upsert::Upsert;
18use crate::watch::{self, WatchOptions, WatchReceiver, WatchSpec, ZeroWatchDecoder};
19use crate::zero_tree::{from_value_bytes, to_bytes};
20
21/// A map with fixed-size keys and zerocopy-compatible typed values.
22///
23/// Thin wrapper around [`ConstMap<K, V, D>`] that provides a typed API for values
24/// implementing [`FromBytes`] + [`IntoBytes`] + [`Immutable`]. Conversions are
25/// zero-cost — values are transmuted without serialization.
26///
27/// Generic over `D: Durability` (default: `Bitcask`). Use `Fixed` backend for
28/// frequent updates without compaction: `ZeroMap::<K, V, Fixed, T>::open(...)`.
29///
30/// Requires `size_of::<T>() == V` (compile-time assertion in constructor).
31///
32/// # Write hooks
33///
34/// Uses [`TypedWriteHook<K, T>`] — the hook receives `&T` directly (not raw bytes).
35/// `on_write` fires on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update`/`fetch_update`,
36/// and on each mutation inside `atomic()` (replayed after the lock is released, in application order).
37/// Old value is always provided (it lives in
38/// memory) — `NEEDS_OLD_VALUE` is ignored.
39///
40/// `on_init` fires once per live entry during `migrate()` or `replay_init()`
41/// (enable via `NEEDS_INIT = true`).
42///
43/// # When to use
44///
45/// For `T` with trivial representation (no heap allocations, fixed layout):
46/// structs with `#[derive(FromBytes, IntoBytes, Immutable, KnownLayout)]`.
47/// All values live in memory — reads never touch disk. O(1) HashMap lookup.
48///
49/// Ordered iteration is available when opened with `Config::iterable(true)` via
50/// [`iter_view`](Self::iter_view); otherwise use [`ZeroTree`](crate::ZeroTree) for prefix/range scans.
51///
52/// # Usage
53///
54/// ```ignore
55/// use zerocopy::{FromBytes, IntoBytes, Immutable, KnownLayout};
56///
57/// #[derive(FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy)]
58/// #[repr(C)]
59/// struct Session {
60///     user_id: u64,
61///     expires: u64,
62/// }
63///
64/// let map = ZeroMap::<[u8; 16], { size_of::<Session>() }, Session>::open(
65///     "data/sessions",
66///     Config::balanced().build(),
67/// )?;
68/// map.put(&key, &Session { user_id: 42, expires: 0 })?;
69/// if let Some(s) = map.get(&key) {
70///     println!("user: {}", s.user_id);
71/// }
72/// ```
73///
74/// Facade mutations and `atomic()` forward to the inner Const map, so they ride
75/// Const's `Family` / `Ctx` seam without a separate Zero `Family` impl.
76/// [`ZeroMapShard`] stays `#[repr(transparent)]` over [`ConstMapShard`].
77pub struct ZeroMap<
78    K: Key + Send + Sync + Hash + Eq,
79    const V: usize,
80    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
81    H: TypedWriteHook<K, T> = NoHook,
82    D: Durability = Bitcask,
83> {
84    inner: Arc<ConstMap<K, V, ZeroHookAdapter<K, T, H>, D>>,
85    _marker: PhantomData<T>,
86}
87
88impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> watch::sealed::Spec
89    for ZeroMap<K, V, T, H, D>
90where
91    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
92    H: TypedWriteHook<K, T>,
93    D: Durability,
94{
95}
96impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> WatchSpec for ZeroMap<K, V, T, H, D>
97where
98    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
99    H: TypedWriteHook<K, T>,
100    D: Durability,
101{
102    type Key = K;
103    type Value = T;
104    type Decoder = ZeroWatchDecoder<T>;
105}
106impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> ZeroMap<K, V, T, H, D>
107where
108    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
109    H: TypedWriteHook<K, T> + watch::sealed::Enabled,
110    D: Durability,
111{
112    /// Subscribe to this collection's mutations.
113    ///
114    /// Infallible and repeatable; every receiver is independent and owns a
115    /// bounded ring. Delivery is **at-most-once** — a full ring drops the
116    /// oldest event and reports [`WatchError::Lagged`] at the gap.
117    ///
118    /// The consumer recipe is `watch()` **first**, then scan, and rescan on
119    /// `Lagged`. It is complete because of what this call guarantees on return:
120    /// every write that decided not to publish to this receiver has already
121    /// made its row visible. The scan half is why a watched Map must be opened
122    /// with [`Config::iterable`] — without an iterator view the open itself is
123    /// refused with [`DbError::WatchRequiresIterable`], so that this call can
124    /// stay infallible.
125    ///
126    /// An event is a pointer, not a snapshot — a rebuild re-reads rows by key
127    /// instead of applying payloads. There is no ordering across keys or
128    /// collections, and transaction boundaries are invisible. A `new_only` hook
129    /// reports every non-delete as [`WatchKind::Upsert`].
130    ///
131    /// Subscriptions end with the database: `Db::shutdown` (and `Drop`) closes
132    /// every channel; `Db::close` leaves receivers live. See `docs/watch.md`.
133    pub fn watch(&self, options: WatchOptions) -> WatchReceiver<Self> {
134        WatchReceiver::subscribe(
135            self.inner.watch_channel().expect("enabled watch hook"),
136            ZeroWatchDecoder::new(),
137            options,
138        )
139    }
140}
141
142// ==========================================================================
143// Bitcask-specific impl blocks
144// ==========================================================================
145
146impl<
147    K: Key + Send + Sync + Hash + Eq,
148    const V: usize,
149    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
150> ZeroMap<K, V, T, NoHook, Bitcask>
151{
152    /// Open or create a `ZeroMap` at the given path.
153    /// Recovers the index from existing data files on disk.
154    ///
155    #[doc = include_str!("../docs/exclusive-access.md")]
156    pub fn open(path: impl AsRef<std::path::Path>, config: Config) -> DbResult<Self> {
157        const { assert!(size_of::<T>() == V) }
158        let adapter = ZeroHookAdapter {
159            inner: NoHook,
160            _marker: PhantomData,
161        };
162        Ok(Self {
163            inner: Arc::new(ConstMap::open_inner(path, config, adapter)?),
164            _marker: PhantomData,
165        })
166    }
167}
168
169impl<
170    K: Key + Send + Sync + Hash + Eq,
171    const V: usize,
172    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
173    H: TypedWriteHook<K, T>,
174> ZeroMap<K, V, T, H, Bitcask>
175{
176    /// Open or create a `ZeroMap` with a write hook for secondary index maintenance.
177    ///
178    #[doc = include_str!("../docs/exclusive-access.md")]
179    pub fn open_hooked(
180        path: impl AsRef<std::path::Path>,
181        config: Config,
182        hook: H,
183    ) -> DbResult<Self> {
184        crate::counter_dirty::check_no_counters(&hook)?;
185        crate::unique::check_no_claims(&hook)?;
186        #[cfg(feature = "armour")]
187        if H::EMITS {
188            return Err(DbError::ChangeStreamRequiresManagedOpen);
189        }
190        Self::open_hooked_inner(path, config, hook)
191    }
192
193    /// Skips [`check_no_counters`](crate::counter_dirty::check_no_counters) —
194    /// only `Db::open_managed` may call this, immediately followed by
195    /// `ManagedOpen::bind_counter_dirty` before the collection is published.
196    pub(crate) fn open_hooked_inner(
197        path: impl AsRef<std::path::Path>,
198        config: Config,
199        hook: H,
200    ) -> DbResult<Self> {
201        const { assert!(size_of::<T>() == V) }
202        if H::WATCHES && !config.iterable {
203            return Err(DbError::WatchRequiresIterable);
204        }
205        let adapter = ZeroHookAdapter {
206            inner: hook,
207            _marker: PhantomData,
208        };
209        Ok(Self {
210            inner: Arc::new(ConstMap::open_inner(path, config, adapter)?),
211            _marker: PhantomData,
212        })
213    }
214
215    /// Graceful shutdown: write hint files, flush write buffers + fsync.
216    pub fn clean_shutdown(&self) -> DbResult<()> {
217        self.inner.clean_shutdown()
218    }
219
220    pub fn close(self) -> DbResult<()> {
221        self.clean_shutdown()
222    }
223
224    /// Flush all shard write buffers to disk (without fsync).
225    pub fn flush_buffers(&self) -> DbResult<()> {
226        self.inner.flush_buffers()
227    }
228
229    /// Get the database configuration.
230    pub fn config(&self) -> &Config {
231        self.inner.config()
232    }
233
234    /// Trigger a compaction pass across all shards.
235    pub fn compact(&self) -> DbResult<usize> {
236        self.inner.compact()
237    }
238
239    /// Write hint files for all active shard files. Call during graceful shutdown.
240    pub fn sync_hints(&self) -> DbResult<()> {
241        self.inner.sync_hints()
242    }
243
244    /// Access the underlying `ConstMap`.
245    pub fn as_inner(&self) -> &ConstMap<K, V, ZeroHookAdapter<K, T, H>, Bitcask> {
246        &self.inner
247    }
248}
249
250// ==========================================================================
251// Fixed-specific impl blocks
252// ==========================================================================
253
254impl<
255    K: Key + Send + Sync + Hash + Eq,
256    const V: usize,
257    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
258> ZeroMap<K, V, T, NoHook, Fixed>
259{
260    /// Open or create a `ZeroMap` with Fixed (fixed-slot) backend.
261    /// Recovers the index from existing data files on disk.
262    ///
263    #[doc = include_str!("../docs/exclusive-access.md")]
264    pub fn open(path: impl AsRef<std::path::Path>, config: FixedConfig) -> DbResult<Self> {
265        const { assert!(size_of::<T>() == V) }
266        let adapter = ZeroHookAdapter {
267            inner: NoHook,
268            _marker: PhantomData,
269        };
270        Ok(Self {
271            inner: Arc::new(ConstMap::open_fixed_inner(path, config, adapter)?),
272            _marker: PhantomData,
273        })
274    }
275}
276
277impl<
278    K: Key + Send + Sync + Hash + Eq,
279    const V: usize,
280    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
281    H: TypedWriteHook<K, T>,
282> ZeroMap<K, V, T, H, Fixed>
283{
284    /// Open or create a `ZeroMap` with a write hook, using Fixed (fixed-slot) backend.
285    ///
286    #[doc = include_str!("../docs/exclusive-access.md")]
287    pub fn open_with_hook(
288        path: impl AsRef<std::path::Path>,
289        config: FixedConfig,
290        hook: H,
291    ) -> DbResult<Self> {
292        crate::counter_dirty::check_no_counters(&hook)?;
293        crate::unique::check_no_claims(&hook)?;
294        #[cfg(feature = "armour")]
295        if H::EMITS {
296            return Err(DbError::ChangeStreamRequiresManagedOpen);
297        }
298        Self::open_with_hook_inner(path, config, hook)
299    }
300
301    /// Skips [`check_no_counters`](crate::counter_dirty::check_no_counters) —
302    /// only `Db::open_managed` may call this, immediately followed by
303    /// `ManagedOpen::bind_counter_dirty` before the collection is published.
304    pub(crate) fn open_with_hook_inner(
305        path: impl AsRef<std::path::Path>,
306        config: FixedConfig,
307        hook: H,
308    ) -> DbResult<Self> {
309        const { assert!(size_of::<T>() == V) }
310        if H::WATCHES && !config.iterable {
311            return Err(DbError::WatchRequiresIterable);
312        }
313        let adapter = ZeroHookAdapter {
314            inner: hook,
315            _marker: PhantomData,
316        };
317        Ok(Self {
318            inner: Arc::new(ConstMap::open_fixed_inner(path, config, adapter)?),
319            _marker: PhantomData,
320        })
321    }
322
323    /// Perform a clean shutdown (Fixed backend).
324    pub fn clean_shutdown(&self) -> DbResult<()> {
325        self.inner.clean_shutdown()
326    }
327
328    pub fn close(self) -> DbResult<()> {
329        self.clean_shutdown()
330    }
331}
332
333// ==========================================================================
334// Generic impl block — works with any D: Durability
335// ==========================================================================
336
337impl<
338    K: Key + Send + Sync + Hash + Eq,
339    const V: usize,
340    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
341    H: TypedWriteHook<K, T>,
342    D: Durability,
343> ZeroMap<K, V, T, H, D>
344{
345    pub(crate) fn tx_inner(&self) -> &ConstMap<K, V, ZeroHookAdapter<K, T, H>, D> {
346        self.inner.as_ref()
347    }
348
349    /// Register the `Arc` handle the inner [`ConstMap`] needs for transaction deferred replay.
350    pub(crate) fn bind_tx_self(arc: &Arc<Self>) {
351        ConstMap::bind_tx_self(&arc.inner);
352    }
353
354    pub(crate) fn tx_install_projection_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
355        self.tx_inner().tx_install_projection_metadata(db)
356    }
357
358    pub(crate) fn tx_preplan_projection_targets(
359        &self,
360    ) -> DbResult<Vec<Arc<dyn crate::txn::participant::TxParticipant>>> {
361        self.tx_inner().tx_preplan_projection_targets()
362    }
363
364    pub(crate) fn tx_project_writes(
365        &self,
366        key: &[u8],
367        old: Option<&[u8]>,
368        new: Option<&[u8]>,
369    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
370        self.tx_inner().tx_project_writes(key, old, new)
371    }
372
373    pub(crate) fn tx_counter_deltas_for(
374        &self,
375        key: &[u8],
376        old: Option<&[u8]>,
377        new: Option<&[u8]>,
378    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
379        self.tx_inner().tx_counter_deltas_for(key, old, new)
380    }
381
382    /// Claim ops for one decided write inside a transaction (T10 Task 6) —
383    /// `name` is passed straight through to [`ConstMap::tx_claims_for`],
384    /// which cannot name `T` itself.
385    pub(crate) fn tx_claims_for(
386        &self,
387        name: &'static str,
388        key: &[u8],
389        old: Option<&[u8]>,
390        new: Option<&[u8]>,
391    ) -> DbResult<Vec<crate::txn::participant::ClaimOp>> {
392        self.tx_inner().tx_claims_for(name, key, old, new)
393    }
394
395    /// Build the per-shard TTL indexes by walking live rows.
396    ///
397    /// Called once by `Db::open_managed`, after migrations and before the handle
398    /// is published. Not public API — visible only because the integration tests
399    /// build collections without going through `Db`. To rebuild a live
400    /// collection's index (e.g. after replication promotion) call
401    /// [`rebuild_ttl_index`](Self::rebuild_ttl_index) instead: this one skips the
402    /// `H::EXPIRES` check and does not clear the promotion requirement.
403    #[doc(hidden)]
404    pub fn build_ttl_index(&self) -> DbResult<()> {
405        self.inner.build_ttl_index()
406    }
407
408    #[doc(hidden)]
409    pub fn ttl_debug_len(&self) -> usize {
410        self.inner.ttl_debug_len()
411    }
412
413    #[doc(hidden)]
414    pub fn ttl_debug_deadline(&self, key: &K) -> Option<u64> {
415        self.inner.ttl_debug_deadline(key)
416    }
417
418    pub fn sweep_expired(&self, now_ms: u64, budget: usize) -> DbResult<crate::ttl::SweepOutcome> {
419        self.inner.sweep_expired(now_ms, budget)
420    }
421
422    pub fn set_ttl_sweep(&self, enabled: bool) -> DbResult<()> {
423        self.inner.set_ttl_sweep(enabled)
424    }
425
426    pub fn rebuild_ttl_index(&self) -> DbResult<()> {
427        self.inner.rebuild_ttl_index()
428    }
429
430    // -- Reads ----------------------------------------------------------------
431
432    /// Get a value by key. O(1) lookup, zero disk I/O. Returns a copy of `T`.
433    /// Takes a brief per-shard mutex to read the index (value is inline).
434    pub fn get(&self, key: &K) -> Option<T> {
435        let bytes = self.inner.get(key)?;
436        Some(from_value_bytes::<V, T>(&bytes))
437    }
438
439    /// Get a value by key, returning `Err(KeyNotFound)` if absent.
440    pub fn get_or_err(&self, key: &K) -> DbResult<T> {
441        self.get(key).ok_or(DbError::KeyNotFound)
442    }
443
444    /// Check if a key exists.
445    pub fn contains(&self, key: &K) -> bool {
446        self.inner.contains(key)
447    }
448
449    // -- Writes ---------------------------------------------------------------
450
451    /// Insert or update a key-value pair. Returns the old value if the key existed.
452    pub fn put(&self, key: &K, value: &T) -> DbResult<Option<T>> {
453        let bytes = to_bytes::<V, T>(value);
454        self.inner
455            .put(key, &bytes)
456            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
457    }
458
459    /// Insert a key-value pair only if the key does not exist.
460    /// Returns `Err(KeyExists)` if the key is already present.
461    pub fn insert(&self, key: &K, value: &T) -> DbResult<()> {
462        let bytes = to_bytes::<V, T>(value);
463        self.inner.insert(key, &bytes)
464    }
465
466    /// Delete a key. Returns the old value if the key existed.
467    pub fn delete(&self, key: &K) -> DbResult<Option<T>> {
468        self.inner
469            .delete(key)
470            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
471    }
472
473    /// Compare-and-swap: if current value == expected, replace with new_value.
474    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
475    /// `Err(KeyNotFound)` if key doesn't exist.
476    pub fn cas(&self, key: &K, expected: &T, new_value: &T) -> DbResult<()> {
477        let exp_bytes = to_bytes::<V, T>(expected);
478        let new_bytes = to_bytes::<V, T>(new_value);
479        self.inner.cas(key, &exp_bytes, &new_bytes)
480    }
481
482    /// Compare-and-delete: if the current value == `expected`, delete the key.
483    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
484    /// `Err(KeyNotFound)` if the key doesn't exist.
485    pub fn compare_delete(&self, key: &K, expected: &T) -> DbResult<()> {
486        let exp_bytes = to_bytes::<V, T>(expected);
487        self.inner.compare_delete(key, &exp_bytes)
488    }
489
490    /// Atomically read-modify-write. Returns `Some(T)` (the **new** value)
491    /// if key existed, `None` otherwise.
492    /// The closure must not be heavy (shard lock is held).
493    pub fn update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
494        self.inner
495            .update(key, |bytes| {
496                let val = from_value_bytes::<V, T>(bytes);
497                let new_val = f(&val);
498                to_bytes::<V, T>(&new_val)
499            })
500            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
501    }
502
503    /// Like [`update()`](Self::update), but returns `Some(T)` with the **old** value.
504    pub fn fetch_update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
505        self.inner
506            .fetch_update(key, |bytes| {
507                let val = from_value_bytes::<V, T>(bytes);
508                let new_val = f(&val);
509                to_bytes::<V, T>(&new_val)
510            })
511            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
512    }
513
514    /// Decide-and-write: insert, update, keep, or delete based on the closure's
515    /// [`Upsert`](crate::Upsert) decision. The closure must be short — the shard
516    /// lock is held across it, so it must not panic and must not call back into
517    /// this collection (the shard mutex is not reentrant).
518    pub fn upsert<U: crate::IntoUpsert<T>>(
519        &self,
520        key: &K,
521        f: impl FnOnce(Option<&T>) -> U,
522    ) -> crate::DbResult<U::Out> {
523        self.atomic(key, |s| crate::upsert::upsert_via_held(s, key, f))
524    }
525
526    // -- Atomic ---------------------------------------------------------------
527
528    /// Atomically execute multiple operations on a single shard.
529    /// All keys must route to the same shard as `shard_key`.
530    /// The closure must be short — shard lock is held for its duration.
531    pub fn atomic<R>(
532        &self,
533        shard_key: &K,
534        f: impl FnOnce(&mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>) -> DbResult<R>,
535    ) -> DbResult<R> {
536        self.inner.atomic(shard_key, |const_shard| {
537            // SAFETY: ZeroMapShard is #[repr(transparent)] over ConstMapShard with the SAME
538            // hook type (ZeroHookAdapter<K, T, H>). PhantomData<T> is ZST.
539            let shard = unsafe {
540                &mut *(const_shard as *mut ConstMapShard<'_, K, V, ZeroHookAdapter<K, T, H>, D>
541                    as *mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>)
542            };
543            f(shard)
544        })
545    }
546
547    // -- Info -----------------------------------------------------------------
548
549    pub fn len(&self) -> usize {
550        self.inner.len()
551    }
552
553    pub fn is_empty(&self) -> bool {
554        self.inner.is_empty()
555    }
556
557    pub fn shard_for(&self, key: &K) -> usize {
558        self.inner.shard_for(key)
559    }
560
561    /// Flush all shard data to disk.
562    pub fn flush(&self) -> DbResult<()> {
563        self.inner.flush()
564    }
565
566    // -- Migration ------------------------------------------------------------
567
568    /// Iterate all entries and optionally mutate them. Call once at startup.
569    ///
570    /// The callback receives each (key, &T) and returns `MigrateAction`:
571    /// - `Keep` — no change (fires `on_init` if `NEEDS_INIT`)
572    /// - `Update(new_value)` — replace value (hook-free write, fires `on_init`)
573    /// - `Delete` — remove entry (hook-free tombstone, no `on_init`)
574    ///
575    /// `on_write` is **never** fired during migration.
576    /// Returns the number of mutated entries.
577    pub fn migrate(&self, f: impl Fn(&K, &T) -> crate::MigrateAction<T>) -> DbResult<usize> {
578        self.migrate_inner(f, true)
579    }
580
581    pub(crate) fn migrate_inner(
582        &self,
583        f: impl Fn(&K, &T) -> crate::MigrateAction<T>,
584        fire_init: bool,
585    ) -> DbResult<usize> {
586        self.inner.migrate_inner(
587            |key, bytes| {
588                let val: T = from_value_bytes(bytes);
589                match f(key, &val) {
590                    crate::MigrateAction::Keep => crate::MigrateAction::Keep,
591                    crate::MigrateAction::Update(new) => {
592                        crate::MigrateAction::Update(to_bytes::<V, T>(&new))
593                    }
594                    crate::MigrateAction::Delete => crate::MigrateAction::Delete,
595                }
596            },
597            fire_init,
598        )
599    }
600
601    /// Read-only pass over all live entries (see `ConstMap::for_each`).
602    pub fn for_each(&self, mut f: impl FnMut(&K, &T)) {
603        self.inner.for_each(|key, bytes| {
604            let val: T = from_value_bytes(bytes);
605            f(key, &val);
606        });
607    }
608
609    /// Unordered snapshot visitor for inspection; stops when `f` returns `Break`.
610    pub(crate) fn for_each_while(
611        &self,
612        mut f: impl FnMut(&K, &T) -> ControlFlow<()>,
613    ) -> ControlFlow<()> {
614        self.inner.for_each_while(|key, bytes| {
615            let val: T = from_value_bytes::<V, T>(bytes);
616            f(key, &val)
617        })
618    }
619
620    /// Replay `on_init` for every live entry. Used when no migration runs.
621    pub(crate) fn replay_init(&self) {
622        self.inner.replay_init();
623    }
624
625    pub(crate) fn reconcile_projections(&self) -> DbResult<()> {
626        self.inner.reconcile_projections()
627    }
628
629    pub(crate) fn effective_projection_limits(&self) -> (usize, usize) {
630        self.inner.effective_projection_limits()
631    }
632
633    // -- Ordered iteration ----------------------------------------------------
634
635    /// Ordered scan view. `None` unless the collection was opened with
636    /// `iterable(true)`. Direction follows `Config::reversed` or `FixedConfig::reversed`
637    /// (default DESC).
638    ///
639    /// **Occasional admin/maintenance API, not a hot path.** Use it to browse a
640    /// collection or bulk-clean stale rows (`retain`); for scan-heavy access reach
641    /// for the `*Tree` sibling — a Map's strength is O(1) point lookup, and the
642    /// companion key index costs extra memory plus O(log N) per write. Map and Tree
643    /// scans are both weakly consistent; the Map implementation releases shard locks
644    /// between batches, so concurrent writes may be observed partially.
645    pub fn iter_view(&self) -> Option<ZeroMapIterView<'_, K, V, T, H, D>> {
646        self.inner.iter_view()?;
647        Some(ZeroMapIterView {
648            inner: MapIterView::new(self, dir_from(self.inner.reversed())),
649        })
650    }
651
652    /// Field-granular prefix / range / page builder. `None` unless opened `iterable(true)`.
653    pub fn scope<const N: usize>(
654        &self,
655        key: &K,
656    ) -> Option<crate::scope::Scope<'_, crate::scope::MapScan<&Self>, N>>
657    where
658        T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
659    {
660        crate::scope::map_scope::<_, _, N>(
661            self,
662            key,
663            dir_from(self.reversed()),
664            self.iter_view().is_some(),
665        )
666    }
667
668    /// Arc-owned prefix/range builder for [`crate::Tx::delete_scope`].
669    /// Returns `None` unless this map was opened iterable.
670    #[cfg(feature = "armour")]
671    pub fn tx_scope<const N: usize>(
672        self: &Arc<Self>,
673        key: &K,
674    ) -> Option<crate::TxScope<Self, K, N>> {
675        self.iter_view().is_some().then(|| {
676            crate::txn::scope::TxScope::from_parts(
677                self.clone(),
678                key,
679                dir_from(self.reversed()),
680                Self::tx_scope_keys,
681            )
682        })
683    }
684
685    fn tx_scope_keys<'a>(
686        collection: &'a Self,
687        start: Bound<&'a K>,
688        end: Bound<&'a K>,
689        dir: Dir,
690    ) -> Box<dyn Iterator<Item = K> + 'a> {
691        Box::new(MapIterView::new(collection, dir).keys_range_bounds(start, end))
692    }
693
694    /// First `(key, value)` in configured iteration order, or `None` if empty.
695    ///
696    /// Map scans are weakly consistent: shard locks are released between batches, so
697    /// concurrent writes may be observed partially.
698    pub fn first(&self) -> DbResult<Option<(K, T)>> {
699        let view = self.iter_view().ok_or(DbError::NotImplemented)?;
700        Ok(view.iter().next())
701    }
702
703    /// Last `(key, value)` in configured iteration order, or `None` if empty.
704    ///
705    /// Map scans are weakly consistent: shard locks are released between batches, so
706    /// concurrent writes may be observed partially.
707    pub fn last(&self) -> DbResult<Option<(K, T)>> {
708        self.iter_view().ok_or(DbError::NotImplemented)?;
709        Ok(
710            MapIterView::new(self, dir_from(self.inner.reversed()).flip())
711                .iter()
712                .next(),
713        )
714    }
715
716    /// Whether iterable scan yields descending key order.
717    pub(crate) fn reversed(&self) -> bool {
718        self.inner.reversed()
719    }
720
721    /// Shard count for limit-derived inspection batch sizing.
722    pub(crate) fn inspection_shard_count(&self) -> usize {
723        ShardSource::shard_count(self)
724    }
725
726    /// Strict bounded ordered scan for inspection (not the public `iter_view`).
727    pub(crate) fn inspection_range_bounds(
728        &self,
729        start: Bound<&K>,
730        end: Bound<&K>,
731        batch: usize,
732    ) -> TryMapIter<'_, ZeroMap<K, V, T, H, D>>
733    where
734        D: Durability,
735    {
736        MapIterView::with_batch(self, dir_from(self.inner.reversed()), batch)
737            .try_range_bounds(start, end)
738    }
739}
740
741impl<
742    K: Key + Send + Sync + Hash + Eq,
743    const V: usize,
744    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
745    H: TypedWriteHook<K, T>,
746    D: Durability,
747> ShardSource for ZeroMap<K, V, T, H, D>
748{
749    type Key = K;
750    type Val = T;
751
752    fn shard_count(&self) -> usize {
753        ShardSource::shard_count(self.inner.as_ref())
754    }
755
756    fn next_batch(
757        &self,
758        shard: usize,
759        after: Option<&K>,
760        dir: Dir,
761        n: usize,
762        range_start: Bound<crate::map_index::OrdKey<K>>,
763        range_end: Bound<crate::map_index::OrdKey<K>>,
764    ) -> Vec<(K, T)> {
765        ShardSource::next_batch(
766            self.inner.as_ref(),
767            shard,
768            after,
769            dir,
770            n,
771            range_start,
772            range_end,
773        )
774        .into_iter()
775        .map(|(k, bytes)| (k, from_value_bytes::<V, T>(&bytes)))
776        .collect()
777    }
778
779    fn next_keys_batch(
780        &self,
781        shard: usize,
782        after: Option<&K>,
783        dir: Dir,
784        n: usize,
785        range_start: Bound<crate::map_index::OrdKey<K>>,
786        range_end: Bound<crate::map_index::OrdKey<K>>,
787    ) -> Vec<K> {
788        ShardSource::next_keys_batch(
789            self.inner.as_ref(),
790            shard,
791            after,
792            dir,
793            n,
794            range_start,
795            range_end,
796        )
797    }
798}
799
800/// View over an iterable [`ZeroMap`]; exposes ordered iteration over `(K, T)`
801/// entries and cheap keys-only iteration. Obtained via [`ZeroMap::iter_view`].
802pub struct ZeroMapIterView<'a, K, const V: usize, T, H = NoHook, D = Bitcask>
803where
804    K: Key + Send + Sync + Hash + Eq,
805    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
806    H: TypedWriteHook<K, T>,
807    D: Durability,
808{
809    inner: MapIterView<'a, ZeroMap<K, V, T, H, D>>,
810}
811
812impl<K, const V: usize, T, H, D> ZeroMapIterView<'_, K, V, T, H, D>
813where
814    K: Key + Send + Sync + Hash + Eq,
815    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
816    H: TypedWriteHook<K, T>,
817    D: Durability,
818{
819    /// Full ordered scan in the collection's configured direction.
820    pub fn iter(&self) -> impl Iterator<Item = (K, T)> + '_ {
821        self.inner.iter()
822    }
823
824    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
825    pub fn range(&self, start: &K, end: &K) -> impl Iterator<Item = (K, T)> + '_ {
826        self.inner.range(start, end)
827    }
828
829    /// Iterate entries in range defined by `start` and `end` bounds.
830    pub fn range_bounds(
831        &self,
832        start: Bound<&K>,
833        end: Bound<&K>,
834    ) -> impl Iterator<Item = (K, T)> + '_ {
835        self.inner.range_bounds(start, end)
836    }
837
838    /// Iterate entries whose keys start with `prefix`.
839    pub fn prefix_iter(&self, prefix: &[u8]) -> impl Iterator<Item = (K, T)> + '_ {
840        self.inner.prefix_iter(prefix)
841    }
842
843    /// Keyset pagination in the collection's configured direction.
844    pub fn paginate(&self, after: Option<&K>, limit: usize) -> Page<K, T> {
845        self.inner.paginate(after, limit)
846    }
847
848    /// Full ordered **keys-only** scan — no value materialized.
849    pub fn keys(&self) -> impl Iterator<Item = K> + '_ {
850        self.inner.keys()
851    }
852
853    /// Keys-only iteration over `[start, end)`.
854    pub fn keys_range(&self, start: &K, end: &K) -> impl Iterator<Item = K> + '_ {
855        self.inner.keys_range(start, end)
856    }
857
858    /// Keys-only iteration over a bounded range.
859    pub fn keys_range_bounds(
860        &self,
861        start: Bound<&K>,
862        end: Bound<&K>,
863    ) -> impl Iterator<Item = K> + '_ {
864        self.inner.keys_range_bounds(start, end)
865    }
866
867    /// Keys-only iteration over keys starting with `prefix`.
868    pub fn keys_prefix(&self, prefix: &[u8]) -> impl Iterator<Item = K> + '_ {
869        self.inner.keys_prefix(prefix)
870    }
871
872    /// Keys-only pagination — cheap, materializes no values.
873    pub fn keys_paginate(&self, after: Option<&K>, limit: usize) -> KeyPage<K> {
874        self.inner.keys_paginate(after, limit)
875    }
876
877    /// Keep entries where `f` returns `true`; delete the rest via
878    /// [`ZeroMap::compare_delete`] using the value observed during iteration.
879    ///
880    /// Not atomic across the collection; a genuine storage error aborts the scan
881    /// and is returned. Concurrent value/key changes (`CasMismatch`/`KeyNotFound`)
882    /// are skipped.
883    pub fn retain(&self, mut f: impl FnMut(&K, &T) -> bool) -> DbResult<()> {
884        for (key, value) in self.inner.iter() {
885            if f(&key, &value) {
886                continue;
887            }
888            match self.inner.src().compare_delete(&key, &value) {
889                Ok(()) => {}
890                Err(DbError::CasMismatch) | Err(DbError::KeyNotFound) => {
891                    tracing::debug!("retain: compare_delete skipped (concurrent change)");
892                }
893                Err(e) => return Err(e),
894            }
895        }
896        Ok(())
897    }
898}
899
900impl<
901    K: Key + Send + Sync + Hash + Eq,
902    const V: usize,
903    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
904    H: TypedWriteHook<K, T>,
905> CompactionIndex<K> for ZeroMap<K, V, T, H, Bitcask>
906{
907    fn update_if_match(
908        &self,
909        key: &K,
910        old_loc: crate::disk_loc::DiskLoc,
911        new_loc: crate::disk_loc::DiskLoc,
912    ) -> bool {
913        self.inner.update_if_match(key, old_loc, new_loc)
914    }
915
916    fn contains_key(&self, key: &K) -> bool {
917        self.contains(key)
918    }
919
920    fn is_live(&self, shard_id: u8, key: &K, loc: crate::disk_loc::DiskLoc) -> bool {
921        self.inner.is_live(shard_id, key, loc)
922    }
923}
924
925// ---------------------------------------------------------------------------
926// ZeroMapShard
927// ---------------------------------------------------------------------------
928
929/// Handle for atomic multi-key operations within a single shard.
930/// Obtained via [`ZeroMap::atomic`].
931#[repr(transparent)]
932pub struct ZeroMapShard<
933    'a,
934    K: Key + Send + Sync + Hash + Eq,
935    const V: usize,
936    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
937    D: Durability = Bitcask,
938    WH: WriteHook<K> = NoHook,
939> {
940    inner: ConstMapShard<'a, K, V, WH, D>,
941    _marker: PhantomData<T>,
942}
943
944impl<
945    K: Key + Send + Sync + Hash + Eq,
946    const V: usize,
947    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
948    D: Durability,
949    WH: WriteHook<K>,
950> ZeroMapShard<'_, K, V, T, D, WH>
951{
952    pub fn put(&mut self, key: &K, value: &T) -> DbResult<Option<T>> {
953        let bytes = to_bytes::<V, T>(value);
954        self.inner
955            .put(key, &bytes)
956            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
957    }
958
959    pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
960        let bytes = to_bytes::<V, T>(value);
961        self.inner.insert(key, &bytes)
962    }
963
964    pub fn delete(&mut self, key: &K) -> DbResult<Option<T>> {
965        self.inner
966            .delete(key)
967            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
968    }
969
970    pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
971        Ok(self
972            .inner
973            .try_get(key)?
974            .map(|bytes| from_value_bytes::<V, T>(&bytes)))
975    }
976
977    pub fn get_or_err(&self, key: &K) -> DbResult<T> {
978        self.try_get(key)?.ok_or(DbError::KeyNotFound)
979    }
980
981    pub fn try_contains(&self, key: &K) -> DbResult<bool> {
982        self.inner.try_contains(key)
983    }
984
985    /// Atomically read-modify-write. Returns `Some(T)` (the **new** value)
986    /// if key existed, `None` otherwise.
987    pub fn update(&mut self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
988        self.inner
989            .update(key, |bytes| {
990                let val = from_value_bytes::<V, T>(bytes);
991                to_bytes::<V, T>(&f(&val))
992            })
993            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
994    }
995
996    /// Like [`update()`](Self::update), but returns `Some(T)` with the **old** value.
997    pub fn fetch_update(&mut self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
998        self.inner
999            .fetch_update(key, |bytes| {
1000                let val = from_value_bytes::<V, T>(bytes);
1001                to_bytes::<V, T>(&f(&val))
1002            })
1003            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
1004    }
1005
1006    /// Decide-and-write within the shard. The closure must be short — shard lock is held.
1007    pub fn upsert<U: crate::IntoUpsert<T>>(
1008        &mut self,
1009        key: &K,
1010        f: impl FnOnce(Option<&T>) -> U,
1011    ) -> crate::DbResult<U::Out> {
1012        crate::upsert::upsert_via_held(self, key, f)
1013    }
1014}
1015
1016#[cfg(feature = "armour")]
1017impl<T, const V: usize, H> crate::armour::collection::Collection
1018    for ZeroMap<T::SelfId, V, T, H, Bitcask>
1019where
1020    T: crate::CollectionMeta
1021        + Copy
1022        + zerocopy::IntoBytes
1023        + zerocopy::FromBytes
1024        + zerocopy::Immutable
1025        + Send
1026        + Sync,
1027    H: crate::hook::TypedWriteHook<T::SelfId, T>,
1028    T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
1029{
1030    fn name(&self) -> &str {
1031        T::NAME
1032    }
1033    fn len(&self) -> usize {
1034        self.len()
1035    }
1036    fn compact(&self) -> crate::DbResult<usize> {
1037        self.compact()
1038    }
1039    fn flush(&self) -> crate::DbResult<()> {
1040        crate::projection::collection_flush_fault()?;
1041        self.flush_buffers()?;
1042        if self.config().hints.unwrap_or(false) {
1043            self.sync_hints()?;
1044        }
1045        Ok(())
1046    }
1047    fn periodic_flush(&self) -> crate::DbResult<()> {
1048        self.flush_buffers()
1049    }
1050    fn durable_flush(&self) -> crate::DbResult<()> {
1051        self.inner.durable_flush()
1052    }
1053    fn clean_shutdown(&self) -> crate::DbResult<()> {
1054        ZeroMap::<T::SelfId, V, T, H, Bitcask>::clean_shutdown(self)
1055    }
1056    fn sweep_expired(
1057        &self,
1058        now_ms: u64,
1059        budget: usize,
1060    ) -> crate::DbResult<crate::ttl::SweepOutcome> {
1061        self.sweep_expired(now_ms, budget)
1062    }
1063    fn set_ttl_sweep(&self, enabled: bool) -> crate::DbResult<()> {
1064        self.set_ttl_sweep(enabled)
1065    }
1066    fn rebuild_ttl_index(&self) -> crate::DbResult<()> {
1067        self.rebuild_ttl_index()
1068    }
1069    fn reset_counter_dirty_marker(&self) -> bool {
1070        self.tx_inner()
1071            .counter_dirty
1072            .get()
1073            .is_some_and(|d| d.take_marked())
1074    }
1075    fn reconcile_counters(&self) -> crate::DbResult<u64> {
1076        self.tx_inner().reconcile_counters()
1077    }
1078    fn counter_target_names(&self) -> Vec<&'static str> {
1079        self.tx_inner().counter_target_names()
1080    }
1081    fn close_watches(&self) {
1082        if let Some(channel) = self.inner.watch_channel() {
1083            channel.close_terminal();
1084        }
1085    }
1086}
1087
1088#[cfg(feature = "armour")]
1089impl<T, const V: usize, H> crate::armour::collection::Collection
1090    for ZeroMap<T::SelfId, V, T, H, Fixed>
1091where
1092    T: crate::CollectionMeta
1093        + Copy
1094        + zerocopy::IntoBytes
1095        + zerocopy::FromBytes
1096        + zerocopy::Immutable
1097        + Send
1098        + Sync,
1099    H: crate::hook::TypedWriteHook<T::SelfId, T>,
1100    T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
1101{
1102    fn name(&self) -> &str {
1103        T::NAME
1104    }
1105    fn len(&self) -> usize {
1106        self.len()
1107    }
1108    fn compact(&self) -> crate::DbResult<usize> {
1109        Ok(0)
1110    }
1111    fn flush(&self) -> crate::DbResult<()> {
1112        crate::projection::collection_flush_fault()?;
1113        // FixedStore writes in-place; no hint file. flush() makes pending
1114        // writes durable.
1115        self.flush()
1116    }
1117    fn periodic_flush(&self) -> crate::DbResult<()> {
1118        // FixedStore: fdatasync (data already in the page cache via pwrite).
1119        // Resolves to the inherent ZeroMap<Fixed>::flush(), same as the
1120        // Collection::flush() impl above.
1121        self.flush()
1122    }
1123    fn durable_flush(&self) -> crate::DbResult<()> {
1124        self.flush()
1125    }
1126    fn clean_shutdown(&self) -> crate::DbResult<()> {
1127        ZeroMap::<T::SelfId, V, T, H, Fixed>::clean_shutdown(self)
1128    }
1129    fn sweep_expired(
1130        &self,
1131        now_ms: u64,
1132        budget: usize,
1133    ) -> crate::DbResult<crate::ttl::SweepOutcome> {
1134        self.sweep_expired(now_ms, budget)
1135    }
1136    fn set_ttl_sweep(&self, enabled: bool) -> crate::DbResult<()> {
1137        self.set_ttl_sweep(enabled)
1138    }
1139    fn rebuild_ttl_index(&self) -> crate::DbResult<()> {
1140        self.rebuild_ttl_index()
1141    }
1142    fn reset_counter_dirty_marker(&self) -> bool {
1143        self.tx_inner()
1144            .counter_dirty
1145            .get()
1146            .is_some_and(|d| d.take_marked())
1147    }
1148    fn reconcile_counters(&self) -> crate::DbResult<u64> {
1149        self.tx_inner().reconcile_counters()
1150    }
1151    fn counter_target_names(&self) -> Vec<&'static str> {
1152        self.tx_inner().counter_target_names()
1153    }
1154    fn close_watches(&self) {
1155        if let Some(channel) = self.inner.watch_channel() {
1156            channel.close_terminal();
1157        }
1158    }
1159}
1160
1161#[cfg(feature = "typed-tree")]
1162impl<
1163    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1164    const V: usize,
1165    T: crate::CollectionMeta
1166        + Copy
1167        + zerocopy::IntoBytes
1168        + zerocopy::FromBytes
1169        + zerocopy::Immutable
1170        + Default
1171        + PartialEq
1172        + Send
1173        + Sync
1174        + 'static,
1175    H: TypedWriteHook<K, T> + 'static,
1176> crate::projection::sealed::Sealed for ZeroMap<K, V, T, H, Bitcask>
1177{
1178}
1179
1180impl<
1181    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1182    const V: usize,
1183    T: crate::CollectionMeta
1184        + Copy
1185        + zerocopy::IntoBytes
1186        + zerocopy::FromBytes
1187        + zerocopy::Immutable
1188        + Default
1189        + PartialEq
1190        + Send
1191        + Sync
1192        + 'static,
1193    H: TypedWriteHook<K, T> + 'static,
1194> crate::projection::ProjectableIndex for ZeroMap<K, V, T, H, Bitcask>
1195{
1196    fn collection_name(&self) -> &'static str {
1197        T::NAME
1198    }
1199
1200    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
1201        if key.len() != std::mem::size_of::<K>() {
1202            return Err(DbError::CorruptedEntry { offset: 0 });
1203        }
1204        let k = K::from_bytes(key);
1205        let v: T = zerocopy::FromBytes::read_from_bytes(value)
1206            .map_err(|_| DbError::FormatMismatch("zero index value decode".into()))?;
1207        self.put(&k, &v)?;
1208        Ok(())
1209    }
1210
1211    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
1212        if key.len() != std::mem::size_of::<K>() {
1213            return Err(DbError::CorruptedEntry { offset: 0 });
1214        }
1215        let k = K::from_bytes(key);
1216        self.delete(&k)?;
1217        Ok(())
1218    }
1219
1220    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
1221        Ok(a == b)
1222    }
1223
1224    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
1225        let mut err: Option<DbError> = None;
1226        self.for_each(|k, v| {
1227            if err.is_some() {
1228                return;
1229            }
1230            if let Err(e) = f(k.as_bytes(), zerocopy::IntoBytes::as_bytes(v)) {
1231                err = Some(e);
1232            }
1233        });
1234        match err {
1235            Some(e) => Err(e),
1236            None => Ok(()),
1237        }
1238    }
1239
1240    fn durable_flush(&self) -> DbResult<()> {
1241        ZeroMap::flush(self)
1242    }
1243}
1244
1245#[cfg(feature = "typed-tree")]
1246impl<
1247    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1248    const V: usize,
1249    T: crate::CollectionMeta
1250        + Copy
1251        + zerocopy::IntoBytes
1252        + zerocopy::FromBytes
1253        + zerocopy::Immutable
1254        + Default
1255        + PartialEq
1256        + Send
1257        + Sync
1258        + 'static,
1259    H: TypedWriteHook<K, T> + 'static,
1260> crate::projection::ProjectableIndexTx for ZeroMap<K, V, T, H, Bitcask>
1261{
1262    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
1263        self
1264    }
1265}
1266
1267#[cfg(feature = "typed-tree")]
1268impl<
1269    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1270    const V: usize,
1271    T: crate::CollectionMeta
1272        + Copy
1273        + zerocopy::IntoBytes
1274        + zerocopy::FromBytes
1275        + zerocopy::Immutable
1276        + Default
1277        + PartialEq
1278        + Send
1279        + Sync
1280        + 'static,
1281    H: TypedWriteHook<K, T> + 'static,
1282> crate::projection::sealed::Sealed for ZeroMap<K, V, T, H, Fixed>
1283{
1284}
1285
1286impl<
1287    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1288    const V: usize,
1289    T: crate::CollectionMeta
1290        + Copy
1291        + zerocopy::IntoBytes
1292        + zerocopy::FromBytes
1293        + zerocopy::Immutable
1294        + Default
1295        + PartialEq
1296        + Send
1297        + Sync
1298        + 'static,
1299    H: TypedWriteHook<K, T> + 'static,
1300> crate::projection::ProjectableIndex for ZeroMap<K, V, T, H, Fixed>
1301{
1302    fn collection_name(&self) -> &'static str {
1303        T::NAME
1304    }
1305
1306    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
1307        if key.len() != std::mem::size_of::<K>() {
1308            return Err(DbError::CorruptedEntry { offset: 0 });
1309        }
1310        let k = K::from_bytes(key);
1311        let v: T = zerocopy::FromBytes::read_from_bytes(value)
1312            .map_err(|_| DbError::FormatMismatch("zero index value decode".into()))?;
1313        self.put(&k, &v)?;
1314        Ok(())
1315    }
1316
1317    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
1318        if key.len() != std::mem::size_of::<K>() {
1319            return Err(DbError::CorruptedEntry { offset: 0 });
1320        }
1321        let k = K::from_bytes(key);
1322        self.delete(&k)?;
1323        Ok(())
1324    }
1325
1326    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
1327        Ok(a == b)
1328    }
1329
1330    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
1331        let mut err: Option<DbError> = None;
1332        self.for_each(|k, v| {
1333            if err.is_some() {
1334                return;
1335            }
1336            if let Err(e) = f(k.as_bytes(), zerocopy::IntoBytes::as_bytes(v)) {
1337                err = Some(e);
1338            }
1339        });
1340        match err {
1341            Some(e) => Err(e),
1342            None => Ok(()),
1343        }
1344    }
1345
1346    fn durable_flush(&self) -> DbResult<()> {
1347        ZeroMap::flush(self)
1348    }
1349}
1350
1351#[cfg(feature = "typed-tree")]
1352impl<
1353    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1354    const V: usize,
1355    T: crate::CollectionMeta
1356        + Copy
1357        + zerocopy::IntoBytes
1358        + zerocopy::FromBytes
1359        + zerocopy::Immutable
1360        + Default
1361        + PartialEq
1362        + Send
1363        + Sync
1364        + 'static,
1365    H: TypedWriteHook<K, T> + 'static,
1366> crate::projection::ProjectableIndexTx for ZeroMap<K, V, T, H, Fixed>
1367{
1368    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
1369        self
1370    }
1371}
1372
1373#[cfg(feature = "typed-tree")]
1374impl<
1375    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1376    const V: usize,
1377    T: crate::CollectionMeta
1378        + Copy
1379        + zerocopy::IntoBytes
1380        + zerocopy::FromBytes
1381        + zerocopy::Immutable
1382        + Default
1383        + PartialEq
1384        + Send
1385        + Sync
1386        + 'static,
1387    H: TypedWriteHook<K, T> + 'static,
1388> crate::projection::IndexKeyed for ZeroMap<K, V, T, H, Bitcask>
1389{
1390    type K = K;
1391    type V = T;
1392
1393    fn encode_value(&self, v: &T) -> DbResult<Vec<u8>> {
1394        Ok(zerocopy::IntoBytes::as_bytes(v).to_vec())
1395    }
1396}
1397
1398#[cfg(feature = "typed-tree")]
1399impl<
1400    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1401    const V: usize,
1402    T: crate::CollectionMeta
1403        + Copy
1404        + zerocopy::IntoBytes
1405        + zerocopy::FromBytes
1406        + zerocopy::Immutable
1407        + Default
1408        + PartialEq
1409        + Send
1410        + Sync
1411        + 'static,
1412    H: TypedWriteHook<K, T> + 'static,
1413> crate::counter::CounterCollection for ZeroMap<K, V, T, H, Bitcask>
1414{
1415    fn decode_value(&self, bytes: &[u8]) -> DbResult<T> {
1416        zerocopy::FromBytes::read_from_bytes(bytes)
1417            .map_err(|_| DbError::FormatMismatch("zero counter value decode".into()))
1418    }
1419
1420    /// Read path — no shard write lock: `get` is an O(1) index lookup.
1421    fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
1422        if key.len() != std::mem::size_of::<K>() {
1423            return Err(DbError::CorruptedEntry { offset: 0 });
1424        }
1425        let k = K::from_bytes(key);
1426        self.get(&k).map(|v| self.encode_value(&v)).transpose()
1427    }
1428
1429    fn contains_bytes(&self, key: &[u8]) -> DbResult<bool> {
1430        if key.len() != std::mem::size_of::<K>() {
1431            return Err(DbError::CorruptedEntry { offset: 0 });
1432        }
1433        let k = K::from_bytes(key);
1434        Ok(self.contains(&k))
1435    }
1436
1437    fn upsert_counter(
1438        &self,
1439        key: &[u8],
1440        f: &mut dyn FnMut(Option<&T>) -> Upsert<T>,
1441    ) -> DbResult<()> {
1442        if key.len() != std::mem::size_of::<K>() {
1443            return Err(DbError::CorruptedEntry { offset: 0 });
1444        }
1445        let k = K::from_bytes(key);
1446        self.upsert(&k, f)?;
1447        Ok(())
1448    }
1449}
1450
1451#[cfg(feature = "typed-tree")]
1452impl<
1453    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1454    const V: usize,
1455    T: crate::CollectionMeta
1456        + Copy
1457        + zerocopy::IntoBytes
1458        + zerocopy::FromBytes
1459        + zerocopy::Immutable
1460        + Default
1461        + PartialEq
1462        + Send
1463        + Sync
1464        + 'static,
1465    H: TypedWriteHook<K, T> + 'static,
1466> crate::projection::IndexKeyed for ZeroMap<K, V, T, H, Fixed>
1467{
1468    type K = K;
1469    type V = T;
1470
1471    fn encode_value(&self, v: &T) -> DbResult<Vec<u8>> {
1472        Ok(zerocopy::IntoBytes::as_bytes(v).to_vec())
1473    }
1474}
1475
1476#[cfg(feature = "typed-tree")]
1477impl<
1478    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
1479    const V: usize,
1480    T: crate::CollectionMeta
1481        + Copy
1482        + zerocopy::IntoBytes
1483        + zerocopy::FromBytes
1484        + zerocopy::Immutable
1485        + Default
1486        + PartialEq
1487        + Send
1488        + Sync
1489        + 'static,
1490    H: TypedWriteHook<K, T> + 'static,
1491> crate::counter::CounterCollection for ZeroMap<K, V, T, H, Fixed>
1492{
1493    fn decode_value(&self, bytes: &[u8]) -> DbResult<T> {
1494        zerocopy::FromBytes::read_from_bytes(bytes)
1495            .map_err(|_| DbError::FormatMismatch("zero counter value decode".into()))
1496    }
1497
1498    /// Read path — no shard write lock: `get` is an O(1) index lookup.
1499    fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
1500        if key.len() != std::mem::size_of::<K>() {
1501            return Err(DbError::CorruptedEntry { offset: 0 });
1502        }
1503        let k = K::from_bytes(key);
1504        self.get(&k).map(|v| self.encode_value(&v)).transpose()
1505    }
1506
1507    fn contains_bytes(&self, key: &[u8]) -> DbResult<bool> {
1508        if key.len() != std::mem::size_of::<K>() {
1509            return Err(DbError::CorruptedEntry { offset: 0 });
1510        }
1511        let k = K::from_bytes(key);
1512        Ok(self.contains(&k))
1513    }
1514
1515    fn upsert_counter(
1516        &self,
1517        key: &[u8],
1518        f: &mut dyn FnMut(Option<&T>) -> Upsert<T>,
1519    ) -> DbResult<()> {
1520        if key.len() != std::mem::size_of::<K>() {
1521            return Err(DbError::CorruptedEntry { offset: 0 });
1522        }
1523        let k = K::from_bytes(key);
1524        self.upsert(&k, f)?;
1525        Ok(())
1526    }
1527}
1528
1529#[cfg(test)]
1530mod tests {
1531    use super::*;
1532    use crate::Config;
1533    use crate::hook::TypedWriteHook;
1534    use std::collections::HashMap as StdHashMap;
1535    use std::mem::size_of;
1536    use std::sync::Arc;
1537    use std::sync::atomic::{AtomicUsize, Ordering as AO};
1538    use tempfile::tempdir;
1539    use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
1540
1541    #[derive(
1542        FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy, Debug, PartialEq, Eq, Default,
1543    )]
1544    #[repr(C)]
1545    struct Counters {
1546        views: u64,
1547        likes: u64,
1548    }
1549
1550    type Map = ZeroMap<[u8; 8], { size_of::<Counters>() }, Counters>;
1551
1552    fn open(dir: &std::path::Path) -> Map {
1553        ZeroMap::<[u8; 8], { size_of::<Counters>() }, Counters>::open(dir, Config::test()).unwrap()
1554    }
1555
1556    fn key_on_other_shard(map: &Map, anchor: &[u8; 8]) -> [u8; 8] {
1557        let anchor_shard = map.shard_for(anchor);
1558        (2u64..1000)
1559            .map(|i| i.to_be_bytes())
1560            .find(|k| map.shard_for(k) != anchor_shard)
1561            .expect("key on different shard")
1562    }
1563
1564    #[test]
1565    fn for_each_visits_all_entries() {
1566        let dir = tempdir().unwrap();
1567        let map = open(dir.path());
1568
1569        let key_a = 1u64.to_be_bytes();
1570        let key_b = key_on_other_shard(&map, &key_a);
1571        let key_c = key_on_other_shard(&map, &key_b);
1572
1573        let expected = StdHashMap::from([
1574            (
1575                key_a,
1576                Counters {
1577                    views: 10,
1578                    likes: 1,
1579                },
1580            ),
1581            (
1582                key_b,
1583                Counters {
1584                    views: 20,
1585                    likes: 2,
1586                },
1587            ),
1588            (
1589                key_c,
1590                Counters {
1591                    views: 30,
1592                    likes: 3,
1593                },
1594            ),
1595        ]);
1596        for (key, value) in &expected {
1597            map.put(key, value).unwrap();
1598        }
1599
1600        let mut seen = StdHashMap::new();
1601        map.for_each(|key, value| {
1602            seen.insert(*key, *value);
1603        });
1604
1605        assert_eq!(seen, expected);
1606        assert_ne!(map.shard_for(&key_a), map.shard_for(&key_b));
1607    }
1608
1609    struct ZHook {
1610        writes: Arc<AtomicUsize>,
1611        last_new: Arc<crate::sync::Mutex<Option<u32>>>,
1612    }
1613
1614    impl ZHook {
1615        fn new() -> Self {
1616            Self {
1617                writes: Arc::new(AtomicUsize::new(0)),
1618                last_new: Arc::new(crate::sync::Mutex::new(None)),
1619            }
1620        }
1621    }
1622
1623    impl TypedWriteHook<[u8; 8], u32> for ZHook {
1624        fn on_write(&self, _k: &[u8; 8], _old: Option<&u32>, new: Option<&u32>) {
1625            self.writes.fetch_add(1, AO::Relaxed);
1626            *crate::sync::lock(&self.last_new) = new.copied();
1627        }
1628    }
1629
1630    #[test]
1631    fn zero_map_atomic_fires_typed_hook() {
1632        let dir = tempfile::tempdir().unwrap();
1633        let mut cfg = Config::test();
1634        cfg.shard_count = 1;
1635        let hook = ZHook::new();
1636        let writes = Arc::clone(&hook.writes);
1637        let last_new = Arc::clone(&hook.last_new);
1638        let map = ZeroMap::<[u8; 8], 4, u32, _>::open_hooked(dir.path(), cfg, hook).unwrap();
1639        let k = 1u64.to_be_bytes();
1640        map.atomic(&k, |s| {
1641            s.put(&k, &42u32)?;
1642            Ok(())
1643        })
1644        .unwrap();
1645        assert_eq!(writes.load(AO::Relaxed), 1);
1646        assert_eq!(*crate::sync::lock(&last_new), Some(42));
1647    }
1648
1649    #[test]
1650    fn zero_map_first_last_contract_both_backends() {
1651        let bitcask_dir = tempdir().unwrap();
1652        let bitcask_cfg = Config::balanced()
1653            .shard_count(2)
1654            .hints(true)
1655            .iterable(true)
1656            .reversed(false)
1657            .build();
1658        let bitcask = ZeroMap::<[u8; 8], 8, u64>::open(bitcask_dir.path(), bitcask_cfg).unwrap();
1659        assert!(bitcask.first().unwrap().is_none());
1660        assert!(bitcask.last().unwrap().is_none());
1661        for id in [3_u64, 1, 2] {
1662            bitcask.put(&id.to_be_bytes(), &(id * 10)).unwrap();
1663        }
1664        assert_eq!(
1665            bitcask
1666                .first()
1667                .unwrap()
1668                .map(|(k, v)| (u64::from_be_bytes(k), v)),
1669            Some((1, 10))
1670        );
1671        assert_eq!(
1672            bitcask
1673                .last()
1674                .unwrap()
1675                .map(|(k, v)| (u64::from_be_bytes(k), v)),
1676            Some((3, 30))
1677        );
1678
1679        let fixed_dir = tempdir().unwrap();
1680        let fixed_cfg = crate::FixedConfig {
1681            iterable: true,
1682            reversed: false,
1683            ..crate::FixedConfig::test()
1684        };
1685        let fixed =
1686            ZeroMap::<[u8; 8], 8, u64, NoHook, Fixed>::open(fixed_dir.path(), fixed_cfg).unwrap();
1687        assert!(fixed.first().unwrap().is_none());
1688        assert!(fixed.last().unwrap().is_none());
1689        for id in [3_u64, 1, 2] {
1690            fixed.put(&id.to_be_bytes(), &(id * 10)).unwrap();
1691        }
1692        assert_eq!(
1693            fixed
1694                .first()
1695                .unwrap()
1696                .map(|(k, v)| (u64::from_be_bytes(k), v)),
1697            Some((1, 10))
1698        );
1699        assert_eq!(
1700            fixed
1701                .last()
1702                .unwrap()
1703                .map(|(k, v)| (u64::from_be_bytes(k), v)),
1704            Some((3, 30))
1705        );
1706
1707        let no_index_dir = tempdir().unwrap();
1708        let no_index =
1709            ZeroMap::<[u8; 8], 8, u64>::open(no_index_dir.path(), Config::test()).unwrap();
1710        assert!(matches!(no_index.first(), Err(DbError::NotImplemented)));
1711        assert!(matches!(no_index.last(), Err(DbError::NotImplemented)));
1712    }
1713
1714    #[test]
1715    fn zero_map_fixed_direction() {
1716        #[derive(Clone, Copy, FromBytes, IntoBytes, Immutable)]
1717        #[repr(C)]
1718        struct V {
1719            x: u32,
1720        }
1721        let dir = tempdir().unwrap();
1722        let cfg = crate::FixedConfig {
1723            iterable: true,
1724            reversed: false,
1725            ..crate::FixedConfig::test()
1726        };
1727        let m = ZeroMap::<[u8; 8], { size_of::<V>() }, V, NoHook, Fixed>::open(dir.path(), cfg)
1728            .unwrap();
1729        for i in 1u64..=3 {
1730            m.put(&i.to_be_bytes(), &V { x: i as u32 }).unwrap();
1731        }
1732        let keys: Vec<u64> = m
1733            .iter_view()
1734            .unwrap()
1735            .keys()
1736            .map(u64::from_be_bytes)
1737            .collect();
1738        assert_eq!(keys, [1, 2, 3], "ZeroMap<Fixed> reversed=false ASC");
1739    }
1740
1741    #[test]
1742    fn zero_map_iter_yields_decoded_values_desc() {
1743        #[derive(Clone, Copy, FromBytes, IntoBytes, Immutable)]
1744        #[repr(C)]
1745        struct Val {
1746            x: u32,
1747        }
1748        let dir = tempdir().unwrap();
1749        let cfg = Config::balanced()
1750            .shard_count(2)
1751            .hints(true)
1752            .iterable(true)
1753            .build();
1754        let m = ZeroMap::<[u8; 2], { size_of::<Val>() }, Val>::open(dir.path(), cfg).unwrap();
1755        m.put(&[1, 0], &Val { x: 11 }).unwrap();
1756        m.put(&[2, 0], &Val { x: 22 }).unwrap();
1757        let got: Vec<(u8, u32)> = m
1758            .iter_view()
1759            .unwrap()
1760            .iter()
1761            .map(|(k, v)| (k[0], v.x))
1762            .collect();
1763        assert_eq!(got, [(2, 22), (1, 11)]);
1764        // keys-only: no zerocopy value decode.
1765        let keys: Vec<u8> = m.iter_view().unwrap().keys().map(|k| k[0]).collect();
1766        assert_eq!(keys, [2, 1]);
1767    }
1768
1769    #[test]
1770    fn zero_map_shard_update_and_fetch_update() {
1771        let dir = tempdir().unwrap();
1772        let mut cfg = Config::test();
1773        cfg.shard_count = 1;
1774        let map = ZeroMap::<[u8; 8], 4, u32>::open(dir.path(), cfg).unwrap();
1775        let key = 1u64.to_be_bytes();
1776        map.put(&key, &100u32).unwrap();
1777
1778        let (upd, fetched, missing) = map
1779            .atomic(&key, |s| {
1780                let upd = s.update(&key, |old| old + 1)?; // new -> 101
1781                let fetched = s.fetch_update(&key, |old| old + 1)?; // old -> 101
1782                let missing = s.update(&2u64.to_be_bytes(), |old| *old)?;
1783                Ok((upd, fetched, missing))
1784            })
1785            .expect("atomic");
1786
1787        assert_eq!(upd, Some(101));
1788        assert_eq!(fetched, Some(101));
1789        assert_eq!(missing, None);
1790        assert_eq!(map.get(&key), Some(102));
1791    }
1792
1793    #[test]
1794    fn for_each_while_stops_after_break_without_visiting_later_shards() {
1795        use std::ops::ControlFlow;
1796        use std::sync::atomic::{AtomicUsize, Ordering};
1797
1798        let dir = tempdir().unwrap();
1799        let map = open(dir.path());
1800
1801        let key_a = 1u64.to_be_bytes();
1802        let key_b = key_on_other_shard(&map, &key_a);
1803        assert_ne!(map.shard_for(&key_a), map.shard_for(&key_b));
1804
1805        map.put(&key_a, &Counters { views: 1, likes: 0 }).unwrap();
1806        map.put(&key_b, &Counters { views: 2, likes: 0 }).unwrap();
1807
1808        let seen = AtomicUsize::new(0);
1809        let flow = map.for_each_while(|_k, _v| {
1810            seen.fetch_add(1, Ordering::Relaxed);
1811            ControlFlow::Break(())
1812        });
1813        assert_eq!(seen.load(Ordering::Relaxed), 1);
1814        assert!(matches!(flow, ControlFlow::Break(())));
1815    }
1816}