Skip to main content

rete_core/
dict.rs

1//! Front-coded, restart-indexed dictionary sections (SPEC.md §5.1).
2//!
3//! A section holds the UTF-8 terms of one kind (shared / subjects / objects /
4//! predicates / graphs), sorted and assigned dense 1-based IDs. Terms are stored
5//! in runs of `R`; each run starts with a full term and front-codes the rest.
6
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::Arc;
9
10use crate::chunk_cache::{CacheEntry, ChunkCache};
11use crate::varint::{read_uvarint, write_uvarint};
12
13/// Default restart interval: a full term every `R` entries.
14pub const DEFAULT_RESTART_INTERVAL: u32 = 16;
15
16/// Reserved ID meaning "no such term".
17pub const ABSENT: u32 = 0;
18
19/// How many not-yet-resident chunks a windowed resolution
20/// ([`ChunkedSection::decode_and_walk`]) faults concurrently in one group. It
21/// bounds the transient set of decoded bodies held at once (this many `Arc`s,
22/// each ≤ one chunk), so the parallel decode's memory is a small constant on top
23/// of the `--memory-budget-mb` cache, independent of the window size or budget.
24/// Also the group size, so a larger value means fewer (larger) parallel
25/// dispatches — amortizing the per-dispatch cost on the tight-budget path where
26/// real decoding happens.
27#[cfg(feature = "parallel")]
28const PAR_DECODE_FANOUT: usize = 32;
29
30/// A small, dedicated rayon pool for parallel chunk decode. Separate from (and
31/// smaller than) the global pool so the number of decode worker threads — and
32/// thus their allocator arenas, the dominant driver of the resident-set bump
33/// under parallel decode — is a small constant, not the machine's full core
34/// count. Decode is memory-bandwidth/CPU bound and a handful of threads saturate
35/// it, so this bounds peak RSS without materially costing throughput. Built once.
36#[cfg(feature = "parallel")]
37fn decode_pool() -> &'static rayon::ThreadPool {
38    use std::sync::OnceLock;
39    static POOL: OnceLock<rayon::ThreadPool> = OnceLock::new();
40    POOL.get_or_init(|| {
41        let n = std::thread::available_parallelism()
42            .map(|n| n.get())
43            .unwrap_or(4)
44            .clamp(1, 8);
45        rayon::ThreadPoolBuilder::new()
46            .num_threads(n)
47            .thread_name(|i| format!("rete-decode-{i}"))
48            .build()
49            .expect("build dedicated chunk-decode thread pool")
50    })
51}
52
53#[derive(Debug, thiserror::Error)]
54pub enum DictError {
55    #[error("malformed dictionary section: {0}")]
56    Malformed(&'static str),
57}
58
59/// Build a dictionary section from terms (any order; sorted + deduped here).
60#[derive(Default)]
61pub struct DictSectionBuilder {
62    terms: Vec<String>,
63    restart_interval: u32,
64}
65
66/// Terms per restart run, overridable via `RETE_DICT_RESTART_INTERVAL`.
67///
68/// The interval sets random-access granularity as well as front-coding gain:
69/// chunks are cut on whole-run boundaries, so with very large literals (e.g.
70/// base64-embedded images) a single run dwarfs the 64 KiB chunk budget and one
71/// term lookup drags in ~`r` terms' worth of bytes. Setting `1` makes every term
72/// its own restart — direct seek, larger restart table, no front-coding (which
73/// is worthless for base64 anyway, since such terms share no prefix).
74pub fn env_restart_interval() -> u32 {
75    static R: std::sync::OnceLock<u32> = std::sync::OnceLock::new();
76    *R.get_or_init(|| {
77        std::env::var("RETE_DICT_RESTART_INTERVAL")
78            .ok()
79            .and_then(|v| v.parse::<u32>().ok())
80            .filter(|&v| v >= 1)
81            .unwrap_or(DEFAULT_RESTART_INTERVAL)
82    })
83}
84
85impl DictSectionBuilder {
86    pub fn new() -> Self {
87        Self {
88            terms: Vec::new(),
89            restart_interval: env_restart_interval(),
90        }
91    }
92
93    pub fn with_restart_interval(mut self, r: u32) -> Self {
94        assert!(r >= 1, "restart interval must be >= 1");
95        self.restart_interval = r;
96        self
97    }
98
99    pub fn push(&mut self, term: impl Into<String>) {
100        self.terms.push(term.into());
101    }
102
103    /// Serialize to bytes. Terms are sorted and deduped; the resulting IDs are
104    /// `1..=n` in sorted order.
105    pub fn build(mut self) -> Vec<u8> {
106        self.terms.sort_unstable();
107        self.terms.dedup();
108        let r = self.restart_interval as usize;
109        let n = self.terms.len();
110        let num_restarts = n.div_ceil(r);
111
112        // Encode body and capture each run's starting offset (relative to body).
113        let mut body = Vec::new();
114        let mut restart_offsets = Vec::with_capacity(num_restarts);
115        let mut prev = "";
116        for (i, term) in self.terms.iter().enumerate() {
117            if i % r == 0 {
118                restart_offsets.push(body.len() as u64);
119                // restart entry: shared = 0, full term
120                write_uvarint(&mut body, 0);
121                write_uvarint(&mut body, term.len() as u64);
122                body.extend_from_slice(term.as_bytes());
123            } else {
124                let shared = common_prefix_len(prev, term);
125                let suffix = &term.as_bytes()[shared..];
126                write_uvarint(&mut body, shared as u64);
127                write_uvarint(&mut body, suffix.len() as u64);
128                body.extend_from_slice(suffix);
129            }
130            prev = term;
131        }
132
133        // header || restart-offset table || body
134        let mut out = Vec::new();
135        write_uvarint(&mut out, n as u64);
136        write_uvarint(&mut out, self.restart_interval as u64);
137        write_uvarint(&mut out, num_restarts as u64);
138        for off in &restart_offsets {
139            write_uvarint(&mut out, *off);
140        }
141        out.extend_from_slice(&body);
142        out
143    }
144}
145
146/// Parsed section metadata (the restart table) — cache this once and reuse it
147/// across lookups instead of re-parsing the section header every time.
148#[derive(Debug, Clone)]
149pub struct SectionMeta {
150    pub term_count: u32,
151    pub restart_interval: u32,
152    /// Absolute offsets into the section bytes for each run start.
153    ///
154    /// `u64`, not `usize`: a section may exceed 4 GiB and wasm32 is a 32-bit
155    /// target, so a `usize` offset silently truncates there (see #70). A
156    /// dictionary carrying embedded media is exactly this case — WikiArt's is
157    /// 23.4 GB.
158    pub restart_offsets: Vec<u64>,
159}
160
161/// Parse only the header/restart table of a section.
162pub fn parse_meta(bytes: &[u8]) -> Result<SectionMeta, DictError> {
163    let mut pos = 0;
164    let take = |pos: &mut usize| -> Result<u64, DictError> {
165        let (v, n) =
166            read_uvarint(&bytes[*pos..]).ok_or(DictError::Malformed("truncated header"))?;
167        *pos += n;
168        Ok(v)
169    };
170    let term_count = take(&mut pos)? as u32;
171    let restart_interval = take(&mut pos)? as u32;
172    let num_restarts = take(&mut pos)? as usize;
173    if restart_interval == 0 {
174        return Err(DictError::Malformed("zero restart interval"));
175    }
176    // `num_restarts` is untrusted; each restart is ≥1 byte, so cap the
177    // pre-allocation at the buffer length to avoid an OOM on a bogus count.
178    let mut rel = Vec::with_capacity(num_restarts.min(bytes.len()));
179    for _ in 0..num_restarts {
180        rel.push(take(&mut pos)?);
181    }
182    let body_start = pos as u64;
183    Ok(SectionMeta {
184        term_count,
185        restart_interval,
186        restart_offsets: rel.into_iter().map(|o| body_start + o).collect(),
187    })
188}
189
190/// Decode the front-coded entry at `pos` *into* `buf`: truncate `buf` to the
191/// entry's shared-prefix length and append its suffix bytes. For a restart
192/// entry the stored shared length is 0, so the same decode works for both
193/// entry kinds. Returns the next entry's position; `None` on malformed bytes
194/// (including a shared length longer than the previous term). Allocation-free
195/// after `buf`'s first growth — this is the hot path of every term resolution.
196#[inline]
197fn entry_into(bytes: &[u8], pos: usize, buf: &mut Vec<u8>) -> Option<usize> {
198    let (shared, n1) = read_uvarint(bytes.get(pos..)?)?;
199    let p = pos + n1;
200    let (suf, n2) = read_uvarint(bytes.get(p..)?)?;
201    let start = p + n2;
202    let end = start
203        .checked_add(suf as usize)
204        .filter(|&e| e <= bytes.len())?;
205    if shared as usize > buf.len() {
206        return None;
207    }
208    buf.truncate(shared as usize);
209    buf.extend_from_slice(&bytes[start..end]);
210    Some(end)
211}
212
213/// Decode a run's restart (full-term) entry at `off` into `buf`.
214fn run_entry_into(bytes: &[u8], off: usize, buf: &mut Vec<u8>) -> Option<usize> {
215    buf.clear(); // a restart entry stands alone; stale prefix bytes must not leak
216    entry_into(bytes, off, buf)
217}
218
219/// Resolve `id` (1-based) to its term using cached metadata. Returns `None` on
220/// any inconsistency in untrusted metadata/bytes rather than panicking.
221pub fn section_term(bytes: &[u8], meta: &SectionMeta, id: u32) -> Option<String> {
222    if id == ABSENT || id > meta.term_count {
223        return None;
224    }
225    let idx = (id - 1) as usize;
226    let run = idx / meta.restart_interval as usize;
227    let steps = idx % meta.restart_interval as usize;
228    let mut buf = Vec::new();
229    let mut pos = run_entry_into(bytes, *meta.restart_offsets.get(run)? as usize, &mut buf)?;
230    for _ in 0..steps {
231        pos = entry_into(bytes, pos, &mut buf)?;
232    }
233    Some(String::from_utf8_lossy(&buf).into_owned())
234}
235
236/// Resolve `term` to its ID using cached metadata.
237pub fn section_id(bytes: &[u8], meta: &SectionMeta, term: &str) -> Option<u32> {
238    let mut buf = Vec::new();
239    // Binary search restart runs by their first (full) term.
240    let mut lo = 0usize;
241    let mut hi = meta.restart_offsets.len();
242    while lo < hi {
243        let mid = (lo + hi) / 2;
244        run_entry_into(bytes, meta.restart_offsets[mid] as usize, &mut buf)?;
245        if buf.as_slice() <= term.as_bytes() {
246            lo = mid + 1;
247        } else {
248            hi = mid;
249        }
250    }
251    if lo == 0 {
252        return None; // smaller than every term
253    }
254    let run = lo - 1;
255    let mut pos = run_entry_into(bytes, meta.restart_offsets[run] as usize, &mut buf)?;
256    let base_id = (run * meta.restart_interval as usize) as u32 + 1;
257    // saturating_sub: corrupt metadata where run*interval > term_count must not
258    // underflow-panic.
259    let run_len = meta.restart_interval.min(
260        meta.term_count
261            .saturating_sub(run as u32 * meta.restart_interval),
262    );
263    for step in 0..run_len {
264        if buf.as_slice() == term.as_bytes() {
265            return Some(base_id + step);
266        }
267        if buf.as_slice() > term.as_bytes() {
268            return None;
269        }
270        if step + 1 < run_len {
271            pos = entry_into(bytes, pos, &mut buf)?;
272        }
273    }
274    None
275}
276
277// --- chunked sections ---------------------------------------------------------
278//
279// A section's body is split (on run boundaries) into **chunks** that are
280// compressed and fetched independently (format v0.2), so a remote client
281// resolves a term with one chunk fault instead of downloading the whole
282// section. Local sections are the degenerate case: one pre-set chunk holding
283// the entire serialized section — a single code path serves both.
284
285/// Fetches one chunk's decompressed body slice on demand (`None` = the fetch
286/// failed; the section records it and the lookup misses — callers over remote
287/// data must check [`ChunkedSection::load_incomplete`] after evaluating).
288pub type ChunkLoader = Box<dyn Fn(usize) -> Option<Vec<u8>> + Send + Sync>;
289
290/// Fetches **many** chunks in one round trip: given ascending chunk indices,
291/// returns each chunk's decompressed body in the same order. The ranged
292/// reader implements this by coalescing byte-adjacent chunk ranges into
293/// single range reads — a full-dictionary sweep (export, dump) costs a few
294/// requests per section instead of one per chunk. `None` = the batch failed;
295/// callers fall back to the per-chunk [`ChunkLoader`].
296pub type ChunkBulkLoader = Box<dyn Fn(&[usize]) -> Option<Vec<Vec<u8>>> + Send + Sync>;
297
298/// One chunk: a run-aligned slice of the section body. `body_start` is the
299/// offset (in the section's coordinate space — the same space
300/// [`SectionMeta::restart_offsets`] uses) where `data[0]` sits.
301pub struct SectionChunk {
302    first_run: usize,
303    /// The chunk's **routing key** — the only thing the chunk-level binary
304    /// search in [`ChunkedSection::id`] compares against. It is a *separator*,
305    /// **not a term**:
306    ///
307    /// ```text
308    /// last_term(chunk i-1)  <  key(i)  <=  first_term(chunk i)
309    /// ```
310    ///
311    /// and chunk 0's key is empty (`b"" <= anything`, and nothing routes before
312    /// chunk 0 anyway). Writers store the *shortest* such string — see
313    /// [`shortest_separator`] — which is why a file's chunk directory is a few
314    /// bytes per chunk instead of a full copy of every boundary term.
315    ///
316    /// Files written before that change carry the boundary term verbatim; the
317    /// verbatim first term is the degenerate separator, so both read the same
318    /// way and no version check is needed. Nothing may reconstruct a term from
319    /// this field, compare it for equality with a term, or report it as one: a
320    /// key that is *not* a separator (a truncation, a hash) mis-routes silently
321    /// — `term(id)` and every dump keep working because they route by
322    /// `first_run`, and only `id(term)` lies.
323    ///
324    /// Unused (empty) for the single local chunk.
325    key: Vec<u8>,
326    body_start: u64,
327}
328
329impl SectionChunk {
330    /// A chunk descriptor: routing metadata only. The decompressed body (and
331    /// its derived run-offset table) lives in the section's [`ChunkCache`],
332    /// keyed by `(section_index, chunk_index)`, faulted in on first touch and
333    /// evictable — not held here. `key` is the routing separator, **not a
334    /// term** (see the `key` field).
335    pub fn new(first_run: usize, key: Vec<u8>, body_start: u64) -> Self {
336        SectionChunk {
337            first_run,
338            key,
339            body_start,
340        }
341    }
342}
343
344/// Scan a chunk's decompressed body for the byte offset of each run start (see
345/// [`SectionChunk::run_offsets`]). O(terms-in-chunk), run once per faulted chunk.
346fn chunk_run_offsets(data: &[u8], restart_interval: usize) -> Vec<usize> {
347    let mut offs = vec![0usize];
348    let mut pos = 0usize;
349    let mut buf = Vec::new();
350    let mut count = 0usize;
351    while pos < data.len() {
352        let Some(next) = entry_into(data, pos, &mut buf) else {
353            break;
354        };
355        count += 1;
356        pos = next;
357        if pos < data.len() && count.is_multiple_of(restart_interval) {
358            offs.push(pos);
359        }
360    }
361    offs
362}
363
364/// The first (full) term of the run starting at `off`, as raw bytes. `None`
365/// on malformed bytes.
366pub fn run_first_term(bytes: &[u8], off: usize) -> Option<Vec<u8>> {
367    let mut buf = Vec::new();
368    run_entry_into(bytes, off, &mut buf)?;
369    Some(buf)
370}
371
372/// The LAST term encoded in `bytes[run_off..end]`, where `run_off` is a run
373/// (restart) boundary and `end` is the end of that run's chunk. Walks the
374/// front-coded entries forward, so it decodes at most `restart_interval` of
375/// them. `None` on malformed bytes.
376///
377/// Writers use it for one thing: the previous chunk's last term, the lower
378/// bound of the next chunk's [`shortest_separator`].
379pub fn run_last_term(bytes: &[u8], run_off: usize, end: usize) -> Option<Vec<u8>> {
380    let lim = bytes.get(..end)?;
381    let mut buf = Vec::new();
382    let mut pos = run_entry_into(lim, run_off, &mut buf)?;
383    while pos < end {
384        pos = entry_into(lim, pos, &mut buf)?;
385    }
386    Some(buf)
387}
388
389/// The **shortest** byte string `s` with `prev_last < s <= first`: `first`
390/// truncated one byte past where it first differs from `prev_last`.
391///
392/// This is the routing key a chunk directory stores (`SectionChunk::key`).
393/// `first` is the chunk's first term and `prev_last`
394/// the previous chunk's last term, so `prev_last < first` holds for any sorted
395/// section; if it does not (malformed input), the full `first` is returned —
396/// always a valid separator, just not a short one.
397///
398/// It really is the shortest. Let `k` be the length of the common prefix of
399/// `prev_last` and `first`; the result has length `k + 1`. No shorter `s`
400/// exists: any `s` with `|s| <= k` shares that prefix's bytes with both, so it
401/// is either a prefix of `prev_last` (hence `<= prev_last`), or it first
402/// differs from the shared prefix — below it, making `s < prev_last`, or above
403/// it, making `s > first`. Every case is outside `(prev_last, first]`.
404pub fn shortest_separator(prev_last: &[u8], first: &[u8]) -> Vec<u8> {
405    let shared = prev_last
406        .iter()
407        .zip(first)
408        .take_while(|(a, b)| a == b)
409        .count();
410    if shared >= first.len() {
411        // `first` is a prefix of `prev_last`, i.e. `first <= prev_last`: not a
412        // sorted pair. Keep the whole term.
413        return first.to_vec();
414    }
415    first[..shared + 1].to_vec()
416}
417
418/// A dictionary section whose body is served chunk-by-chunk: metadata + chunk
419/// directory always present, chunk bytes local or faulted in on first touch.
420pub struct ChunkedSection {
421    meta: SectionMeta,
422    chunks: Vec<SectionChunk>,
423    loader: Option<ChunkLoader>,
424    bulk: Option<ChunkBulkLoader>,
425    failed: AtomicBool,
426    /// Decompressed chunk bodies (and their derived run tables) live here, keyed
427    /// by `(section_index, chunk_index)`, faulted in on first touch and
428    /// evictable. Shared across a [`Dictionary`]'s four sections. Phase 0 caps
429    /// it at unlimited, so residency is identical to the old per-chunk
430    /// `OnceLock`s.
431    cache: Arc<ChunkCache>,
432    /// This section's discriminant in the shared cache's key space (0 shared,
433    /// 1 subject-only, 2 object-only, 3 predicates).
434    section_index: u8,
435}
436
437impl ChunkedSection {
438    /// A local section: the whole serialized section (header + body) as one
439    /// resident chunk at coordinate 0, so the absolute restart offsets index it
440    /// directly. Malformed bytes degrade to an empty section (no panics on
441    /// untrusted files), matching the previous reader behavior. The body is
442    /// stored in `cache` at `(section_index, 0)` (unlimited cap ⇒ resident for
443    /// the section's lifetime, exactly as before).
444    pub fn local(section_bytes: Vec<u8>, cache: Arc<ChunkCache>, section_index: u8) -> Self {
445        let meta = parse_meta(&section_bytes).unwrap_or(SectionMeta {
446            term_count: 0,
447            restart_interval: 1,
448            restart_offsets: Vec::new(),
449        });
450        cache.insert((section_index, 0), Arc::from(section_bytes));
451        ChunkedSection {
452            meta,
453            chunks: vec![SectionChunk::new(0, Vec::new(), 0)],
454            loader: None,
455            bulk: None,
456            failed: AtomicBool::new(false),
457            cache,
458            section_index,
459        }
460    }
461
462    /// A section from parsed parts: metadata + chunk list, with an optional
463    /// loader for non-resident chunks (the remote lazy-open path) — resident
464    /// chunk lists (a locally-decoded chunked section) supply their bodies
465    /// through `resident` instead.
466    pub fn from_parts(
467        meta: SectionMeta,
468        chunks: Vec<SectionChunk>,
469        loader: Option<ChunkLoader>,
470        cache: Arc<ChunkCache>,
471        section_index: u8,
472    ) -> Self {
473        ChunkedSection {
474            meta,
475            chunks,
476            loader,
477            bulk: None,
478            failed: AtomicBool::new(false),
479            cache,
480            section_index,
481        }
482    }
483
484    /// A section whose chunk bodies are already decoded (the local chunked-file
485    /// open path): the descriptors and their bodies arrive together and the
486    /// bodies are seeded into the cache. No loader — every chunk is resident.
487    pub fn resident(
488        meta: SectionMeta,
489        chunks: Vec<(SectionChunk, Vec<u8>)>,
490        cache: Arc<ChunkCache>,
491        section_index: u8,
492    ) -> Self {
493        let descriptors = chunks
494            .into_iter()
495            .enumerate()
496            .map(|(ci, (chunk, body))| {
497                cache.insert((section_index, ci as u32), Arc::from(body));
498                chunk
499            })
500            .collect();
501        ChunkedSection {
502            meta,
503            chunks: descriptors,
504            loader: None,
505            bulk: None,
506            failed: AtomicBool::new(false),
507            cache,
508            section_index,
509        }
510    }
511
512    /// Attach a batched chunk fetcher (see [`ChunkBulkLoader`]): full-section
513    /// sweeps ([`prefetch_all`](Self::prefetch_all)) go through it instead of
514    /// faulting chunk by chunk.
515    pub fn with_bulk_loader(mut self, bulk: ChunkBulkLoader) -> Self {
516        self.bulk = Some(bulk);
517        self
518    }
519
520    /// Batch-fault every unloaded chunk through the bulk loader, if one is
521    /// attached and at least two chunks are missing. Callers about to sweep
522    /// the whole section (export/dump term resolution) call this once; a
523    /// failed batch leaves the chunks unloaded for the per-chunk loader to
524    /// retry (and record failures) lookup by lookup.
525    pub fn prefetch_all(&self) {
526        self.prefetch_chunks(&(0..self.chunks.len()).collect::<Vec<_>>());
527    }
528
529    /// Batch-fault a *specific* set of chunks (the subset a bounded query's
530    /// output touches) through the bulk loader, coalescing byte-adjacent ranges
531    /// into single reads. `cis` should be ascending and deduped; chunks already
532    /// resident are skipped. Like [`prefetch_all`](Self::prefetch_all) a single
533    /// missing chunk is left for the per-chunk loader, and a failed batch leaves
534    /// the chunks unloaded for that loader to retry (and record failures).
535    pub fn prefetch_chunks(&self, cis: &[usize]) {
536        let Some(bulk) = &self.bulk else { return };
537        let missing: Vec<usize> = cis
538            .iter()
539            .copied()
540            .filter(|&ci| {
541                ci < self.chunks.len() && !self.cache.contains((self.section_index, ci as u32))
542            })
543            .collect();
544        if missing.len() < 2 {
545            return;
546        }
547        if let Some(bodies) = bulk(&missing) {
548            if bodies.len() == missing.len() {
549                for (&ci, body) in missing.iter().zip(bodies) {
550                    self.cache
551                        .insert((self.section_index, ci as u32), Arc::from(body));
552                }
553            }
554        }
555    }
556
557    pub fn meta(&self) -> &SectionMeta {
558        &self.meta
559    }
560
561    pub fn term_count(&self) -> u32 {
562        self.meta.term_count
563    }
564
565    /// The shared chunk cache backing this section (all four sections of a
566    /// [`crate::dictionary::Dictionary`] share one instance) — the handle the
567    /// bounded-export budget (phase 2) sets a byte cap on.
568    pub(crate) fn cache(&self) -> &Arc<ChunkCache> {
569        &self.cache
570    }
571
572    /// Did any chunk fetch fail since this section was opened — or since the
573    /// last [`reset_load_failure`](Self::reset_load_failure)?
574    pub fn load_incomplete(&self) -> bool {
575        self.failed.load(Ordering::Relaxed)
576    }
577
578    /// Forget recorded fetch failures (see `GraphIndex::reset_load_failure` —
579    /// the per-query reset for resident sessions). Failed chunks were never
580    /// cached, so the next resolution retries them.
581    pub fn reset_load_failure(&self) {
582        self.failed.store(false, Ordering::Relaxed);
583    }
584
585    /// The cache entry for chunk `ci`, faulting its body in through the loader
586    /// on a miss. Returns an owned [`CacheEntry`] handle (an `Arc`) the caller
587    /// holds for the duration of one decode — valid even if a later eviction
588    /// drops the map slot (the handout invariant).
589    ///
590    /// A FAILED fetch records the failure and caches nothing (returns `None`),
591    /// so a later resolution retries the chunk — a transient network error must
592    /// not permanently poison a resident session. A local section with no
593    /// loader returns an empty body for a missing chunk, matching the old
594    /// `get_or_init(Vec::new)` behavior.
595    fn chunk_entry(&self, ci: usize) -> Option<Arc<CacheEntry>> {
596        let key = (self.section_index, ci as u32);
597        if let Some(e) = self.cache.get(key) {
598            return Some(e);
599        }
600        match &self.loader {
601            Some(load) => match load(ci) {
602                Some(bytes) => Some(self.cache.insert(key, Arc::from(bytes))),
603                None => {
604                    self.failed.store(true, Ordering::Relaxed);
605                    None
606                }
607            },
608            None => Some(self.cache.insert(key, Arc::from(Vec::new()))),
609        }
610    }
611
612    /// The chunk holding `run` (chunks ascend by `first_run`; the first chunk
613    /// always starts at run 0).
614    fn chunk_of_run(&self, run: usize) -> Option<usize> {
615        let i = self.chunks.partition_point(|c| c.first_run <= run);
616        i.checked_sub(1)
617    }
618
619    /// The chunk index holding term `id` (1-based, section-local), or `None` if
620    /// out of range. Mirrors the run→chunk math in [`term`](Self::term) so a
621    /// caller can group a set of output ids by chunk and batch-prefetch them.
622    pub fn chunk_of_id(&self, id: u32) -> Option<usize> {
623        if id == ABSENT || id > self.meta.term_count {
624            return None;
625        }
626        let run = (id - 1) as usize / self.meta.restart_interval as usize;
627        self.chunk_of_run(run)
628    }
629
630    /// The byte offset (into chunk `ci`'s decompressed body) of `run`.
631    /// Full/local sections use the section-wide restart table (unchanged
632    /// behavior); a *lite* remote section (empty `restart_offsets`) derives it
633    /// from the chunk itself — that per-chunk run table is cached inside the
634    /// `entry` (evicted with the body), so the open never holds the whole table.
635    fn run_off_in_chunk(
636        &self,
637        ci: usize,
638        run: usize,
639        entry: &CacheEntry,
640        ri: usize,
641    ) -> Option<usize> {
642        let chunk = &self.chunks[ci];
643        if self.meta.restart_offsets.is_empty() {
644            entry
645                .runs(|b| chunk_run_offsets(b, ri))
646                .get(run.checked_sub(chunk.first_run)?)
647                .copied()
648        } else {
649            self.meta
650                .restart_offsets
651                .get(run)?
652                .checked_sub(chunk.body_start)
653                .map(|o| o as usize)
654        }
655    }
656
657    /// One past the last run index held by chunk `ci` (its run range is
658    /// `[first_run, run_end)`).
659    fn run_end_of_chunk(&self, ci: usize, entry: &CacheEntry, ri: usize) -> usize {
660        if let Some(next) = self.chunks.get(ci + 1) {
661            return next.first_run;
662        }
663        let chunk = &self.chunks[ci];
664        if self.meta.restart_offsets.is_empty() {
665            chunk.first_run + entry.runs(|b| chunk_run_offsets(b, ri)).len()
666        } else {
667            self.meta.restart_offsets.len()
668        }
669    }
670
671    /// Resolve `id` (1-based) to its term. One chunk fault at most.
672    /// Chunk index holding term `id` — [`term`](Self::term)'s routing without
673    /// faulting the chunk. `None` for an absent/out-of-range id.
674    pub(crate) fn chunk_of_term(&self, id: u32) -> Option<usize> {
675        if id == ABSENT || id > self.meta.term_count {
676            return None;
677        }
678        self.chunk_of_run((id - 1) as usize / self.meta.restart_interval as usize)
679    }
680
681    pub fn term(&self, id: u32) -> Option<String> {
682        if id == ABSENT || id > self.meta.term_count {
683            return None;
684        }
685        let idx = (id - 1) as usize;
686        let ri = self.meta.restart_interval as usize;
687        let run = idx / ri;
688        let steps = idx % ri;
689        let ci = self.chunk_of_run(run)?;
690        let entry = self.chunk_entry(ci)?;
691        let bytes: &[u8] = entry.body();
692        let off = self.run_off_in_chunk(ci, run, &entry, ri)?;
693        let mut buf = Vec::new();
694        let mut pos = run_entry_into(bytes, off, &mut buf)?;
695        for _ in 0..steps {
696            pos = entry_into(bytes, pos, &mut buf)?;
697        }
698        Some(String::from_utf8_lossy(&buf).into_owned())
699    }
700
701    /// Resolve many ids of THIS section in one chunk-ordered, single-pass walk,
702    /// filling `out[slot]` for each `(id, slot)` in `jobs`.
703    ///
704    /// `jobs` is sorted in place by id. Because chunks and runs are *contiguous
705    /// ascending* id ranges, sorting by id also groups the jobs by chunk and by
706    /// run — so each chunk is faulted **once** and each run walked **once**
707    /// forward, picking off every requested id as the decode passes its
708    /// position. This is where the redundant per-id re-walk that
709    /// [`term`](Self::term) does (decode from the run start for *every* id)
710    /// disappears: within a run the walk advances monotonically across all the
711    /// window's ids that live there.
712    ///
713    /// For any id this fills `out[slot]` with exactly what [`term`](Self::term)
714    /// returns for that id (the same decoded bytes, the same
715    /// `String::from_utf8_lossy`), and leaves `out[slot]` untouched where
716    /// `term(id)` is `None` — an absent/out-of-range id, a failed chunk fault,
717    /// or malformed bytes. Duplicate ids each fill their own slot. It is the
718    /// batch twin of `term`, used by the export window
719    /// ([`Dictionary::resolve_window`](crate::dictionary::Dictionary::resolve_window))
720    /// so a window's scattered object ids touch each chunk once instead of
721    /// re-faulting and re-walking per row.
722    ///
723    /// [`crate::dictionary::Dictionary::resolve_window`]: the sole caller.
724    pub fn resolve_into(&self, jobs: &mut [(u32, usize)], out: &mut [Option<String>]) {
725        let ri = self.meta.restart_interval as usize;
726        if ri == 0 {
727            return;
728        }
729        let tc = self.meta.term_count;
730        // Sort by id: chunks and runs are contiguous ascending id ranges, so
731        // this groups jobs by chunk and by run in one pass.
732        jobs.sort_unstable_by_key(|&(id, _)| id);
733        // Partition the sorted jobs into contiguous per-chunk segments
734        // `(chunk, [start, end))`. Every job of a chunk is one contiguous slice
735        // (chunks are ascending id ranges, `jobs` is id-sorted). Absent (id
736        // `ABSENT` = 0) and out-of-range (`id > tc`) ids sort to the ends and
737        // belong to no chunk — they form no segment, so their slots stay None,
738        // exactly as `term(id) == None` leaves them.
739        let mut segments: Vec<(usize, usize, usize)> = Vec::new();
740        let mut i = 0;
741        while i < jobs.len() {
742            let id0 = jobs[i].0;
743            if id0 == ABSENT || id0 > tc {
744                i += 1;
745                continue;
746            }
747            let Some(ci) = self.chunk_of_run((id0 - 1) as usize / ri) else {
748                i += 1;
749                continue;
750            };
751            let start = i;
752            i += 1;
753            while i < jobs.len() {
754                let id = jobs[i].0;
755                if id == ABSENT || id > tc {
756                    i += 1;
757                    continue;
758                }
759                if self.chunk_of_run((id - 1) as usize / ri) != Some(ci) {
760                    break; // first id of the next chunk
761                }
762                i += 1;
763            }
764            segments.push((ci, start, i));
765        }
766        if segments.is_empty() {
767            return;
768        }
769        self.decode_and_walk(&segments, jobs, out, ri, tc);
770    }
771
772    /// Fault each segment's chunk and fill its slots. The chunk decodes are
773    /// independent and CPU-bound (zstd), so with the `parallel` feature a group
774    /// whose chunks are **not yet resident** is decoded concurrently on a
775    /// bounded pool before its (serial) fill; without it, one at a time — the
776    /// exact behavior this refactor extracted. Either way each chunk is faulted
777    /// once and its slots filled in segment order, so the output is byte-for-byte
778    /// identical.
779    ///
780    /// **Parallel decode runs only where there is decode work.** A group whose
781    /// chunks are already resident (a generous `--memory-budget-mb` after the
782    /// window prefetch warmed them, or a chunk re-touched by a later window) is
783    /// walked serially with **no** parallel dispatch — so the single-pass /
784    /// large-budget export keeps its serial speed and the rayon fan-out only
785    /// pays for itself in the tight-budget regime, where the whole dictionary
786    /// does not fit and chunks are genuinely re-decoded. Skipping the dispatch
787    /// on cache-warm windows is what keeps a large budget from regressing.
788    ///
789    /// The bound matters for `--memory-budget-mb`: a decoding group holds at
790    /// most [`PAR_DECODE_FANOUT`] decoded bodies at once (a small constant,
791    /// independent of the window size or the budget) and each is held behind its
792    /// cache `Arc` for the group's fill — so even a budget that evicts a body
793    /// the moment it lands cannot force a re-decode before the fill consumes it.
794    /// The decode runs on a small dedicated pool ([`decode_pool`]), so the number
795    /// of worker threads (and their allocator arenas) is bounded regardless of
796    /// the machine's core count.
797    #[cfg(feature = "parallel")]
798    fn decode_and_walk(
799        &self,
800        segments: &[(usize, usize, usize)],
801        jobs: &[(u32, usize)],
802        out: &mut [Option<String>],
803        ri: usize,
804        tc: u32,
805    ) {
806        use rayon::prelude::*;
807        for group in segments.chunks(PAR_DECODE_FANOUT) {
808            let has_miss = group
809                .iter()
810                .any(|&(ci, _, _)| !self.cache.contains((self.section_index, ci as u32)));
811            if has_miss {
812                // Real decode work: fault this group's chunks in parallel on the
813                // bounded pool (order preserved so each entry lines up with its
814                // segment), then fill serially. A failed fault yields `None`
815                // (recorded in `self.failed`), same as the serial path.
816                let entries: Vec<Option<Arc<CacheEntry>>> = decode_pool().install(|| {
817                    group
818                        .par_iter()
819                        .map(|&(ci, _, _)| self.chunk_entry(ci))
820                        .collect()
821                });
822                for (&(ci, s, e), entry) in group.iter().zip(entries) {
823                    if let Some(entry) = entry {
824                        self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
825                    }
826                }
827            } else {
828                // Cache-warm: no decode to parallelize, so avoid the dispatch.
829                for &(ci, s, e) in group {
830                    if let Some(entry) = self.chunk_entry(ci) {
831                        self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
832                    }
833                }
834            }
835        }
836    }
837
838    /// Serial fallback for builds without the `parallel` feature (wasm, and
839    /// `--no-default-features`): fault and walk one chunk at a time.
840    #[cfg(not(feature = "parallel"))]
841    fn decode_and_walk(
842        &self,
843        segments: &[(usize, usize, usize)],
844        jobs: &[(u32, usize)],
845        out: &mut [Option<String>],
846        ri: usize,
847        tc: u32,
848    ) {
849        for &(ci, s, e) in segments {
850            if let Some(entry) = self.chunk_entry(ci) {
851                self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
852            }
853        }
854    }
855
856    /// Fill the output slots owned by one chunk: `chunk_jobs` is the id-sorted
857    /// slice of `(id, slot)` jobs that all live in chunk `ci`, and `entry` is
858    /// that chunk's already-faulted body. Walks the chunk's front-coded runs
859    /// once each, forward, picking off every requested id as the decode passes
860    /// it — the single-pass twin of `term(id)`, writing exactly the bytes
861    /// `term(id)` would for each id and leaving a slot None where `term(id)` is
862    /// None (a malformed run entry, past which no larger id in that run decodes).
863    fn walk_chunk(
864        &self,
865        ci: usize,
866        entry: &CacheEntry,
867        chunk_jobs: &[(u32, usize)],
868        out: &mut [Option<String>],
869        ri: usize,
870        tc: u32,
871    ) {
872        let bytes: &[u8] = entry.body();
873        let mut k = 0;
874        while k < chunk_jobs.len() {
875            let id = chunk_jobs[k].0;
876            if id == ABSENT || id > tc {
877                k += 1;
878                continue;
879            }
880            let run = (id - 1) as usize / ri;
881            // Decode this run's restart entry. Any failure here means
882            // `term(id)` is None for the whole run — skip every job in it.
883            let run_start = self.run_off_in_chunk(ci, run, entry, ri).and_then(|off| {
884                let mut buf = Vec::new();
885                run_entry_into(bytes, off, &mut buf).map(|pos| (buf, pos))
886            });
887            let Some((mut buf, mut pos)) = run_start else {
888                while k < chunk_jobs.len() {
889                    let id = chunk_jobs[k].0;
890                    if id == ABSENT || id > tc {
891                        k += 1;
892                        continue;
893                    }
894                    if (id - 1) as usize / ri != run {
895                        break;
896                    }
897                    k += 1;
898                }
899                continue;
900            };
901            // `buf` holds the run's term #`cur_step`; advance monotonically.
902            let mut cur_step = 0usize;
903            let mut broken = false;
904            while k < chunk_jobs.len() {
905                let id = chunk_jobs[k].0;
906                if id == ABSENT || id > tc {
907                    k += 1;
908                    continue;
909                }
910                if (id - 1) as usize / ri != run {
911                    break; // next run
912                }
913                let steps = (id - 1) as usize % ri;
914                if !broken {
915                    while cur_step < steps {
916                        match entry_into(bytes, pos, &mut buf) {
917                            Some(next) => {
918                                pos = next;
919                                cur_step += 1;
920                            }
921                            None => {
922                                // Malformed past this point: `term(id)` is None
923                                // here and for every larger id in the run (they
924                                // all decode through this entry).
925                                broken = true;
926                                break;
927                            }
928                        }
929                    }
930                    if !broken && cur_step == steps {
931                        out[chunk_jobs[k].1] = Some(String::from_utf8_lossy(&buf).into_owned());
932                    }
933                }
934                k += 1;
935            }
936        }
937    }
938
939    /// Resolve `term` to its ID. Chunk-level binary search runs on the (local)
940    /// chunk directory, so this also costs at most one chunk fault.
941    pub fn id(&self, term: &str) -> Option<u32> {
942        if self.chunks.is_empty() {
943            return None;
944        }
945        let ri = self.meta.restart_interval as usize;
946        // Pick the chunk: the last one whose ROUTING KEY is <= `term` (the
947        // single local chunk skips the search — its key is unset).
948        //
949        // This is the ONLY read of [`SectionChunk::key`] in the codebase, and
950        // all it needs is the separator invariant `last(i-1) < key(i) <=
951        // first(i)`: `term` lands in chunk `i` exactly when `key(i) <= term <
952        // key(i+1)`, which covers every term of chunk `i` and nothing in any
953        // other chunk. A verbatim first term satisfies it too, which is why
954        // files written either way route identically.
955        //
956        // With separators the search can land on chunk `i` for a `term` in the
957        // gap `key(i) <= term < first(i)`; the run search below then hits its
958        // `lo == first_run` branch and answers `None`, which is correct — such
959        // a term is strictly between two chunks, i.e. absent.
960        let ci = if self.chunks.len() == 1 {
961            0
962        } else {
963            let i = self
964                .chunks
965                .partition_point(|c| c.key.as_slice() <= term.as_bytes());
966            i.checked_sub(1)?
967        };
968        let first_run = self.chunks[ci].first_run;
969        let entry = self.chunk_entry(ci)?;
970        let bytes: &[u8] = entry.body();
971
972        // Binary search this chunk's runs by their first (full) term.
973        let run_end = self.run_end_of_chunk(ci, &entry, ri);
974        let mut buf = Vec::new();
975        let mut lo = first_run;
976        let mut hi = run_end;
977        while lo < hi {
978            let mid = (lo + hi) / 2;
979            let off = self.run_off_in_chunk(ci, mid, &entry, ri)?;
980            run_entry_into(bytes, off, &mut buf)?;
981            if buf.as_slice() <= term.as_bytes() {
982                lo = mid + 1;
983            } else {
984                hi = mid;
985            }
986        }
987        if lo == first_run {
988            // Smaller than every term in (and before) this chunk. Unreachable
989            // when the key is a verbatim first term; reachable, and correct,
990            // when it is a separator: `key(ci) <= term < first(ci)` puts the
991            // term strictly between two chunks.
992            return None;
993        }
994        let run = lo - 1;
995        let off = self.run_off_in_chunk(ci, run, &entry, ri)?;
996        let mut pos = run_entry_into(bytes, off, &mut buf)?;
997        let base_id = (run * ri) as u32 + 1;
998        // saturating_sub: corrupt metadata must not underflow-panic.
999        let run_len = self.meta.restart_interval.min(
1000            self.meta
1001                .term_count
1002                .saturating_sub(run as u32 * self.meta.restart_interval),
1003        );
1004        for step in 0..run_len {
1005            if buf.as_slice() == term.as_bytes() {
1006                return Some(base_id + step);
1007            }
1008            if buf.as_slice() > term.as_bytes() {
1009                return None;
1010            }
1011            if step + 1 < run_len {
1012                pos = entry_into(bytes, pos, &mut buf)?;
1013            }
1014        }
1015        None
1016    }
1017
1018    /// The full serialized section (header + body), for the file writer. The
1019    /// local single-chunk case returns the stored bytes; a chunked section
1020    /// re-assembles them (header re-encoded from the metadata).
1021    pub fn raw_section_bytes(&self) -> Vec<u8> {
1022        if self.chunks.len() == 1 && self.chunks[0].body_start == 0 {
1023            if let Some(entry) = self.cache.get((self.section_index, 0)) {
1024                return entry.body().to_vec();
1025            }
1026        }
1027        let mut out = encode_section_header(&self.meta);
1028        for ci in 0..self.chunks.len() {
1029            if let Some(entry) = self.chunk_entry(ci) {
1030                out.extend_from_slice(entry.body());
1031            }
1032        }
1033        out
1034    }
1035}
1036
1037/// Re-encode a section header (term_count, interval, restart table) from its
1038/// parsed metadata — the inverse of [`parse_meta`]. Restart offsets are stored
1039/// body-relative on disk; `meta` holds them absolute, so the body start is
1040/// re-derived as the first run's offset.
1041pub fn encode_section_header(meta: &SectionMeta) -> Vec<u8> {
1042    let body_start = meta.restart_offsets.first().copied().unwrap_or(0);
1043    let mut out = Vec::new();
1044    write_uvarint(&mut out, meta.term_count as u64);
1045    write_uvarint(&mut out, meta.restart_interval as u64);
1046    write_uvarint(&mut out, meta.restart_offsets.len() as u64);
1047    for off in &meta.restart_offsets {
1048        write_uvarint(&mut out, off.saturating_sub(body_start));
1049    }
1050    out
1051}
1052
1053/// A parsed, read-only dictionary section (parses its metadata on construction).
1054pub struct DictSection<'a> {
1055    bytes: &'a [u8],
1056    meta: SectionMeta,
1057}
1058
1059impl<'a> DictSection<'a> {
1060    pub fn parse(bytes: &'a [u8]) -> Result<Self, DictError> {
1061        Ok(Self {
1062            bytes,
1063            meta: parse_meta(bytes)?,
1064        })
1065    }
1066
1067    pub fn len(&self) -> u32 {
1068        self.meta.term_count
1069    }
1070
1071    pub fn is_empty(&self) -> bool {
1072        self.meta.term_count == 0
1073    }
1074
1075    /// Resolve `id` (1-based) to its term, or `None` if out of range.
1076    pub fn term(&self, id: u32) -> Option<String> {
1077        section_term(self.bytes, &self.meta, id)
1078    }
1079
1080    /// Resolve `term` to its ID, or `None` if absent.
1081    pub fn id(&self, term: &str) -> Option<u32> {
1082        section_id(self.bytes, &self.meta, term)
1083    }
1084}
1085
1086fn common_prefix_len(a: &str, b: &str) -> usize {
1087    a.bytes().zip(b.bytes()).take_while(|(x, y)| x == y).count()
1088}
1089
1090#[cfg(test)]
1091mod tests {
1092    use super::*;
1093
1094    /// Build a *lite* (empty section-wide restart table), genuinely multi-chunk
1095    /// [`ChunkedSection`] from `terms`: the exact shape a ranged open produces,
1096    /// so `resolve_into`'s per-chunk run-table branch and its cross-chunk /
1097    /// cross-run walk are all exercised. `runs_per_chunk` chops the runs into
1098    /// several chunks regardless of size, so a tiny fixture still spans many.
1099    fn lite_multichunk(terms: &[&str], ri: u32, runs_per_chunk: usize) -> (ChunkedSection, u32) {
1100        let mut b = DictSectionBuilder::new().with_restart_interval(ri);
1101        for t in terms {
1102            b.push(*t);
1103        }
1104        let raw = b.build();
1105        let full_meta = parse_meta(&raw).unwrap();
1106        let n_runs = full_meta.restart_offsets.len();
1107        let mut chunks = Vec::new();
1108        let mut bodies: Vec<Vec<u8>> = Vec::new();
1109        let mut r = 0;
1110        while r < n_runs {
1111            let first_run = r;
1112            let start = full_meta.restart_offsets[r] as usize;
1113            let r2 = (r + runs_per_chunk).min(n_runs);
1114            let end = if r2 < n_runs {
1115                full_meta.restart_offsets[r2] as usize
1116            } else {
1117                raw.len()
1118            };
1119            chunks.push(SectionChunk::new(first_run, Vec::new(), start as u64));
1120            bodies.push(raw[start..end].to_vec());
1121            r = r2;
1122        }
1123        assert!(chunks.len() > 1, "fixture must be multi-chunk");
1124        // Lite: no section-wide restart table -> run offsets come from each
1125        // decoded chunk body, cached in its cache entry (the ranged path).
1126        let lite_meta = SectionMeta {
1127            term_count: full_meta.term_count,
1128            restart_interval: full_meta.restart_interval,
1129            restart_offsets: Vec::new(),
1130        };
1131        let loader: ChunkLoader = Box::new(move |ci| bodies.get(ci).cloned());
1132        let sec = ChunkedSection::from_parts(
1133            lite_meta,
1134            chunks,
1135            Some(loader),
1136            ChunkCache::unlimited_arc(),
1137            0,
1138        );
1139        (sec, full_meta.term_count)
1140    }
1141
1142    /// The windowed batch resolver must fill exactly what a per-id `term()`
1143    /// would, for a scrambled, duplicated, out-of-range id set spread across
1144    /// many chunks and runs — the byte-identity guarantee `dump_filtered_each`
1145    /// rests on (phase 1). Also cross-checked against the ground-truth
1146    /// `section_term` decoder.
1147    #[test]
1148    fn resolve_into_matches_term_across_chunks_and_runs() {
1149        // Shared prefixes (front-coding), a long literal (a run that dwarfs its
1150        // neighbors), blank nodes — 240 distinct terms.
1151        let owned: Vec<String> = (0..240)
1152            .map(|i| match i % 4 {
1153                0 => format!("<http://example.org/entity/{i:05}>"),
1154                1 => format!("<http://example.org/entity/{i:05}/sub/leaf>"),
1155                2 => format!("\"literal value {} padded {}\"", i, "x".repeat(i % 50)),
1156                _ => format!("_:b{i:05}"),
1157            })
1158            .collect();
1159        let terms: Vec<&str> = owned.iter().map(String::as_str).collect();
1160
1161        for (ri, rpc) in [(1u32, 5usize), (4, 3), (16, 2)] {
1162            let (sec, count) = lite_multichunk(&terms, ri, rpc);
1163            let raw = {
1164                let mut b = DictSectionBuilder::new().with_restart_interval(ri);
1165                for t in &terms {
1166                    b.push(*t);
1167                }
1168                b.build()
1169            };
1170            let full_meta = parse_meta(&raw).unwrap();
1171
1172            // A scrambled probe set: every id once, a run of duplicates, some
1173            // out-of-range ids (0 and past the end) that must stay `None`.
1174            let mut probe: Vec<u32> = (1..=count).collect();
1175            probe.extend([1, 1, count, count, ABSENT, count + 3, count + 100]);
1176            // Deterministic shuffle so chunk order != probe order.
1177            let mut state = 0xDEAD_BEEF_1234_5678u64;
1178            for k in (1..probe.len()).rev() {
1179                state ^= state << 13;
1180                state ^= state >> 7;
1181                state ^= state << 17;
1182                probe.swap(k, (state % (k as u64 + 1)) as usize);
1183            }
1184
1185            let mut jobs: Vec<(u32, usize)> = probe.iter().map(|&id| (id, 0)).collect();
1186            for (slot, j) in jobs.iter_mut().enumerate() {
1187                j.1 = slot;
1188            }
1189            let mut out: Vec<Option<String>> = vec![None; probe.len()];
1190            sec.resolve_into(&mut jobs, &mut out);
1191
1192            for (slot, &id) in probe.iter().enumerate() {
1193                let via_term = sec.term(id);
1194                assert_eq!(
1195                    out[slot], via_term,
1196                    "resolve_into(id={id}) != term(id) at ri={ri} rpc={rpc}"
1197                );
1198                // Ground truth for in-range ids.
1199                let truth = section_term(&raw, &full_meta, id);
1200                assert_eq!(out[slot], truth, "resolve_into(id={id}) != section_term");
1201            }
1202        }
1203    }
1204
1205    fn sample() -> Vec<String> {
1206        // Deliberately unsorted, with shared prefixes and a duplicate.
1207        [
1208            "http://ex.org/Alice",
1209            "http://ex.org/Bob",
1210            "http://ex.org/Alan",
1211            "http://ex.org/knows",
1212            "http://ex.org/Alice", // dup
1213            "zeta",
1214        ]
1215        .iter()
1216        .map(|s| s.to_string())
1217        .collect()
1218    }
1219
1220    #[test]
1221    fn round_trip_all_ids_and_terms() {
1222        for r in [1u32, 2, 16, 1000] {
1223            let mut b = DictSectionBuilder::new().with_restart_interval(r);
1224            for t in sample() {
1225                b.push(t);
1226            }
1227            let bytes = b.build();
1228            let sec = DictSection::parse(&bytes).unwrap();
1229
1230            // Expected sorted unique set.
1231            let mut expected = sample();
1232            expected.sort();
1233            expected.dedup();
1234            assert_eq!(sec.len() as usize, expected.len());
1235
1236            for (i, term) in expected.iter().enumerate() {
1237                let id = (i + 1) as u32;
1238                assert_eq!(
1239                    sec.term(id).as_deref(),
1240                    Some(term.as_str()),
1241                    "term({id}) r={r}"
1242                );
1243                assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
1244            }
1245        }
1246    }
1247
1248    #[test]
1249    fn lookups_for_absent_terms() {
1250        let mut b = DictSectionBuilder::new();
1251        for t in sample() {
1252            b.push(t);
1253        }
1254        let bytes = b.build();
1255        let sec = DictSection::parse(&bytes).unwrap();
1256        assert_eq!(sec.id("aaa-before-everything"), None);
1257        assert_eq!(sec.id("zzz-after-everything"), None);
1258        assert_eq!(sec.id("http://ex.org/Alic"), None); // prefix, not present
1259        assert_eq!(sec.term(0), None);
1260        assert_eq!(sec.term(9999), None);
1261    }
1262
1263    /// Property-style stress of the front-coded decode paths: a large
1264    /// deterministic-pseudo-random term pool (IRIs with heavy shared prefixes,
1265    /// literals, blank nodes — duplicates likely) must round-trip every id and
1266    /// term across restart intervals, and near-miss probes (a stored term ± a
1267    /// suffix/truncation, including run-boundary terms) must resolve to `None`.
1268    /// Term resolution is the engine's hot path, so this is the test that
1269    /// guards the buffer-reuse decode.
1270    #[test]
1271    fn randomized_round_trip_and_near_misses_across_restart_intervals() {
1272        let mut state = 0x9E37_79B9_7F4A_7C15u64;
1273        let mut next = move || {
1274            state ^= state << 13;
1275            state ^= state >> 7;
1276            state ^= state << 17;
1277            state
1278        };
1279        let pool: Vec<String> = (0..1500)
1280            .map(|i| {
1281                let n = next();
1282                match n % 4 {
1283                    0 => format!("<http://example.org/entity/{n:x}>"),
1284                    1 => format!("\"literal value {} with spaces\"", n % 300), // dups likely
1285                    2 => format!("<http://example.org/entity/{}/sub/{i}>", n % 64), // long shared prefixes
1286                    _ => format!("_:b{}", n % 256),
1287                }
1288            })
1289            .collect();
1290        let mut expected = pool.clone();
1291        expected.sort();
1292        expected.dedup();
1293
1294        for r in [1u32, 3, 16, 64] {
1295            let mut b = DictSectionBuilder::new().with_restart_interval(r);
1296            for t in &pool {
1297                b.push(t.clone());
1298            }
1299            let bytes = b.build();
1300            let sec = DictSection::parse(&bytes).unwrap();
1301            assert_eq!(sec.len() as usize, expected.len(), "r={r}");
1302
1303            for (i, term) in expected.iter().enumerate() {
1304                let id = (i + 1) as u32;
1305                assert_eq!(
1306                    sec.term(id).as_deref(),
1307                    Some(term.as_str()),
1308                    "term({id}) r={r}"
1309                );
1310                assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
1311            }
1312            // Near misses around run boundaries (and a sample of the rest):
1313            // an appended suffix or a truncation is never a stored term unless
1314            // it happens to collide with one.
1315            for (i, term) in expected.iter().enumerate() {
1316                let near_boundary = (i as u32) % r <= 1;
1317                if !near_boundary && i % 37 != 0 {
1318                    continue;
1319                }
1320                let longer = format!("{term}\u{1}");
1321                assert_eq!(sec.id(&longer), None, "near-miss long r={r}");
1322                let mut shorter = term.clone();
1323                shorter.pop();
1324                if !shorter.is_empty() && expected.binary_search(&shorter).is_err() {
1325                    assert_eq!(sec.id(&shorter), None, "near-miss short {shorter:?} r={r}");
1326                }
1327            }
1328            assert_eq!(sec.term(expected.len() as u32 + 1), None, "past-end r={r}");
1329        }
1330    }
1331}