kevy_embedded/
ops_atomic_all_index.rs

1//! Index reads for [`AtomicAllShards`].
2//!
3//! A consumer asked to check uniqueness inside a transaction against a
4//! declared index instead of a parallel set of claim keys. Claim keys
5//! are a second source of truth that has to be reconciled at boot; an
6//! index is derived by construction and cannot drift.
7//!
8//! **Only the all-shards context gets these, deliberately.** An index
9//! entry lives on the shard of the key it indexes, so "does any row
10//! have this email" is a question about every shard. `atomic()` holds
11//! one shard's lock and can answer only for its own slice — and a
12//! uniqueness check that silently consults 1/N of the keyspace is worse
13//! than no uniqueness check, because it returns "unique" most of the
14//! time. `atomic_all_shards()` already holds every shard's write lock,
15//! so the same read is both complete and consistent there.
16//!
17//! **These see committed state, not the transaction's own writes.**
18//! Index maintenance runs in `commit_write`, which runs after the
19//! closure returns `Ok`. Checking a value against rows written earlier
20//! in the same closure will not find them; a closure inserting two rows
21//! must compare them to each other itself. The lock makes this the only
22//! gap — nothing another writer does can appear mid-transaction.
23
24use super::AtomicAllShards;
25use crate::KevyResult;
26use crate::ops_index::{IndexPage, merge_page};
27use crate::ops_index_sync::sync_segs;
28use kevy_index::{Cursor, IndexValue};
29
30impl AtomicAllShards<'_> {
31    /// `IDX.QUERY name min max` — range or EQ, merged across every
32    /// shard in `(value, key)` order.
33    ///
34    /// Sees the index as of the last commit; see the module note.
35    pub fn idx_query(
36        &mut self,
37        name: &[u8],
38        min: &IndexValue,
39        max: &IndexValue,
40        cursor: Option<&Cursor>,
41        limit: usize,
42    ) -> KevyResult<IndexPage> {
43        let limit = limit.clamp(1, 100_000);
44        let mut all: Vec<(IndexValue, Vec<u8>)> = Vec::new();
45        self.each_segment(name, |seg| {
46            let (hits, _) = seg.range(min, max, cursor, limit);
47            all.extend(hits.into_iter().map(|(k, v)| (v, k)));
48        })?;
49        Ok(merge_page(all, limit))
50    }
51
52    /// `IDX.COUNT name min max` without materialising keys — the cheap
53    /// form of an existence check.
54    pub fn idx_count(
55        &mut self,
56        name: &[u8],
57        min: &IndexValue,
58        max: &IndexValue,
59    ) -> KevyResult<u64> {
60        let mut total = 0u64;
61        self.each_segment(name, |seg| total += seg.count(min, max))?;
62        Ok(total)
63    }
64
65    /// Visit `name`'s segment on every shard, using the write guards
66    /// this transaction already holds. The `Store` twin takes each
67    /// shard lock itself, which is why it cannot be reused here — this
68    /// context holds them all, and re-locking would deadlock.
69    fn each_segment(
70        &mut self,
71        name: &[u8],
72        mut f: impl FnMut(&kevy_index::Segment),
73    ) -> KevyResult<()> {
74        let reg = std::sync::Arc::clone(&self.indexes);
75        let mut found = false;
76        for g in self.guards.iter_mut() {
77            let inner = &mut **g;
78            sync_segs(&reg, &mut inner.idx_segs, &mut inner.store);
79            if let Some((_, seg)) = inner.idx_segs.segs.iter().find(|(s, _)| s.name == name) {
80                found = true;
81                f(seg);
82            }
83        }
84        if found {
85            Ok(())
86        } else {
87            Err(crate::KevyError::NotFound("no such index".into()))
88        }
89    }
90}
kevy_embedded/ops_atomic_all_index.rs

kevy_embedded/
ops_atomic_all_index.rs
