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(§ion_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 /// Heap bytes this section holds for as long as it is open, independent
573 /// of the chunk cache: the chunk directory (one descriptor plus its
574 /// routing key per chunk), the two copies of the per-chunk byte ranges the
575 /// ranged loaders capture, and any eager restart-offset table. Grows with
576 /// the chunk count, never with what a dump touches.
577 pub fn directory_bytes(&self) -> u64 {
578 let n = self.chunks.len() as u64;
579 let keys: u64 = self.chunks.iter().map(|c| c.key.capacity() as u64).sum();
580 let ranges = if self.loader.is_some() { 2 * 16 * n } else { 0 };
581 n * std::mem::size_of::<SectionChunk>() as u64
582 + keys
583 + ranges
584 + self.meta.restart_offsets.capacity() as u64 * 8
585 }
586
587 /// Did any chunk fetch fail since this section was opened — or since the
588 /// last [`reset_load_failure`](Self::reset_load_failure)?
589 pub fn load_incomplete(&self) -> bool {
590 self.failed.load(Ordering::Relaxed)
591 }
592
593 /// Forget recorded fetch failures (see `GraphIndex::reset_load_failure` —
594 /// the per-query reset for resident sessions). Failed chunks were never
595 /// cached, so the next resolution retries them.
596 pub fn reset_load_failure(&self) {
597 self.failed.store(false, Ordering::Relaxed);
598 }
599
600 /// The cache entry for chunk `ci`, faulting its body in through the loader
601 /// on a miss. Returns an owned [`CacheEntry`] handle (an `Arc`) the caller
602 /// holds for the duration of one decode — valid even if a later eviction
603 /// drops the map slot (the handout invariant).
604 ///
605 /// A FAILED fetch records the failure and caches nothing (returns `None`),
606 /// so a later resolution retries the chunk — a transient network error must
607 /// not permanently poison a resident session. A local section with no
608 /// loader returns an empty body for a missing chunk, matching the old
609 /// `get_or_init(Vec::new)` behavior.
610 fn chunk_entry(&self, ci: usize) -> Option<Arc<CacheEntry>> {
611 let key = (self.section_index, ci as u32);
612 if let Some(e) = self.cache.get(key) {
613 return Some(e);
614 }
615 match &self.loader {
616 Some(load) => match load(ci) {
617 Some(bytes) => Some(self.cache.insert(key, Arc::from(bytes))),
618 None => {
619 self.failed.store(true, Ordering::Relaxed);
620 None
621 }
622 },
623 None => Some(self.cache.insert(key, Arc::from(Vec::new()))),
624 }
625 }
626
627 /// The chunk holding `run` (chunks ascend by `first_run`; the first chunk
628 /// always starts at run 0).
629 fn chunk_of_run(&self, run: usize) -> Option<usize> {
630 let i = self.chunks.partition_point(|c| c.first_run <= run);
631 i.checked_sub(1)
632 }
633
634 /// The chunk index holding term `id` (1-based, section-local), or `None` if
635 /// out of range. Mirrors the run→chunk math in [`term`](Self::term) so a
636 /// caller can group a set of output ids by chunk and batch-prefetch them.
637 pub fn chunk_of_id(&self, id: u32) -> Option<usize> {
638 if id == ABSENT || id > self.meta.term_count {
639 return None;
640 }
641 let run = (id - 1) as usize / self.meta.restart_interval as usize;
642 self.chunk_of_run(run)
643 }
644
645 /// The byte offset (into chunk `ci`'s decompressed body) of `run`.
646 /// Full/local sections use the section-wide restart table (unchanged
647 /// behavior); a *lite* remote section (empty `restart_offsets`) derives it
648 /// from the chunk itself — that per-chunk run table is cached inside the
649 /// `entry` (evicted with the body), so the open never holds the whole table.
650 fn run_off_in_chunk(
651 &self,
652 ci: usize,
653 run: usize,
654 entry: &CacheEntry,
655 ri: usize,
656 ) -> Option<usize> {
657 let chunk = &self.chunks[ci];
658 if self.meta.restart_offsets.is_empty() {
659 entry
660 .runs(|b| chunk_run_offsets(b, ri))
661 .get(run.checked_sub(chunk.first_run)?)
662 .copied()
663 } else {
664 self.meta
665 .restart_offsets
666 .get(run)?
667 .checked_sub(chunk.body_start)
668 .map(|o| o as usize)
669 }
670 }
671
672 /// One past the last run index held by chunk `ci` (its run range is
673 /// `[first_run, run_end)`).
674 fn run_end_of_chunk(&self, ci: usize, entry: &CacheEntry, ri: usize) -> usize {
675 if let Some(next) = self.chunks.get(ci + 1) {
676 return next.first_run;
677 }
678 let chunk = &self.chunks[ci];
679 if self.meta.restart_offsets.is_empty() {
680 chunk.first_run + entry.runs(|b| chunk_run_offsets(b, ri)).len()
681 } else {
682 self.meta.restart_offsets.len()
683 }
684 }
685
686 /// Resolve `id` (1-based) to its term. One chunk fault at most.
687 /// Chunk index holding term `id` — [`term`](Self::term)'s routing without
688 /// faulting the chunk. `None` for an absent/out-of-range id.
689 pub(crate) fn chunk_of_term(&self, id: u32) -> Option<usize> {
690 if id == ABSENT || id > self.meta.term_count {
691 return None;
692 }
693 self.chunk_of_run((id - 1) as usize / self.meta.restart_interval as usize)
694 }
695
696 pub fn term(&self, id: u32) -> Option<String> {
697 if id == ABSENT || id > self.meta.term_count {
698 return None;
699 }
700 let idx = (id - 1) as usize;
701 let ri = self.meta.restart_interval as usize;
702 let run = idx / ri;
703 let steps = idx % ri;
704 let ci = self.chunk_of_run(run)?;
705 let entry = self.chunk_entry(ci)?;
706 let bytes: &[u8] = entry.body();
707 let off = self.run_off_in_chunk(ci, run, &entry, ri)?;
708 let mut buf = Vec::new();
709 let mut pos = run_entry_into(bytes, off, &mut buf)?;
710 for _ in 0..steps {
711 pos = entry_into(bytes, pos, &mut buf)?;
712 }
713 Some(String::from_utf8_lossy(&buf).into_owned())
714 }
715
716 /// Resolve many ids of THIS section in one chunk-ordered, single-pass walk,
717 /// filling `out[slot]` for each `(id, slot)` in `jobs`.
718 ///
719 /// `jobs` is sorted in place by id. Because chunks and runs are *contiguous
720 /// ascending* id ranges, sorting by id also groups the jobs by chunk and by
721 /// run — so each chunk is faulted **once** and each run walked **once**
722 /// forward, picking off every requested id as the decode passes its
723 /// position. This is where the redundant per-id re-walk that
724 /// [`term`](Self::term) does (decode from the run start for *every* id)
725 /// disappears: within a run the walk advances monotonically across all the
726 /// window's ids that live there.
727 ///
728 /// For any id this fills `out[slot]` with exactly what [`term`](Self::term)
729 /// returns for that id (the same decoded bytes, the same
730 /// `String::from_utf8_lossy`), and leaves `out[slot]` untouched where
731 /// `term(id)` is `None` — an absent/out-of-range id, a failed chunk fault,
732 /// or malformed bytes. Duplicate ids each fill their own slot. It is the
733 /// batch twin of `term`, used by the export window
734 /// ([`Dictionary::resolve_window`](crate::dictionary::Dictionary::resolve_window))
735 /// so a window's scattered object ids touch each chunk once instead of
736 /// re-faulting and re-walking per row.
737 ///
738 /// [`crate::dictionary::Dictionary::resolve_window`]: the sole caller.
739 pub fn resolve_into(&self, jobs: &mut [(u32, usize)], out: &mut [Option<String>]) {
740 let ri = self.meta.restart_interval as usize;
741 if ri == 0 {
742 return;
743 }
744 let tc = self.meta.term_count;
745 // Sort by id: chunks and runs are contiguous ascending id ranges, so
746 // this groups jobs by chunk and by run in one pass.
747 jobs.sort_unstable_by_key(|&(id, _)| id);
748 // Partition the sorted jobs into contiguous per-chunk segments
749 // `(chunk, [start, end))`. Every job of a chunk is one contiguous slice
750 // (chunks are ascending id ranges, `jobs` is id-sorted). Absent (id
751 // `ABSENT` = 0) and out-of-range (`id > tc`) ids sort to the ends and
752 // belong to no chunk — they form no segment, so their slots stay None,
753 // exactly as `term(id) == None` leaves them.
754 let mut segments: Vec<(usize, usize, usize)> = Vec::new();
755 let mut i = 0;
756 while i < jobs.len() {
757 let id0 = jobs[i].0;
758 if id0 == ABSENT || id0 > tc {
759 i += 1;
760 continue;
761 }
762 let Some(ci) = self.chunk_of_run((id0 - 1) as usize / ri) else {
763 i += 1;
764 continue;
765 };
766 let start = i;
767 i += 1;
768 while i < jobs.len() {
769 let id = jobs[i].0;
770 if id == ABSENT || id > tc {
771 i += 1;
772 continue;
773 }
774 if self.chunk_of_run((id - 1) as usize / ri) != Some(ci) {
775 break; // first id of the next chunk
776 }
777 i += 1;
778 }
779 segments.push((ci, start, i));
780 }
781 if segments.is_empty() {
782 return;
783 }
784 self.decode_and_walk(&segments, jobs, out, ri, tc);
785 }
786
787 /// Fault each segment's chunk and fill its slots. The chunk decodes are
788 /// independent and CPU-bound (zstd), so with the `parallel` feature a group
789 /// whose chunks are **not yet resident** is decoded concurrently on a
790 /// bounded pool before its (serial) fill; without it, one at a time — the
791 /// exact behavior this refactor extracted. Either way each chunk is faulted
792 /// once and its slots filled in segment order, so the output is byte-for-byte
793 /// identical.
794 ///
795 /// **Parallel decode runs only where there is decode work.** A group whose
796 /// chunks are already resident (a generous `--memory-budget-mb` after the
797 /// window prefetch warmed them, or a chunk re-touched by a later window) is
798 /// walked serially with **no** parallel dispatch — so the single-pass /
799 /// large-budget export keeps its serial speed and the rayon fan-out only
800 /// pays for itself in the tight-budget regime, where the whole dictionary
801 /// does not fit and chunks are genuinely re-decoded. Skipping the dispatch
802 /// on cache-warm windows is what keeps a large budget from regressing.
803 ///
804 /// The bound matters for `--memory-budget-mb`: a decoding group holds at
805 /// most [`PAR_DECODE_FANOUT`] decoded bodies at once (a small constant,
806 /// independent of the window size or the budget) and each is held behind its
807 /// cache `Arc` for the group's fill — so even a budget that evicts a body
808 /// the moment it lands cannot force a re-decode before the fill consumes it.
809 /// The decode runs on a small dedicated pool ([`decode_pool`]), so the number
810 /// of worker threads (and their allocator arenas) is bounded regardless of
811 /// the machine's core count.
812 #[cfg(feature = "parallel")]
813 fn decode_and_walk(
814 &self,
815 segments: &[(usize, usize, usize)],
816 jobs: &[(u32, usize)],
817 out: &mut [Option<String>],
818 ri: usize,
819 tc: u32,
820 ) {
821 use rayon::prelude::*;
822 for group in segments.chunks(PAR_DECODE_FANOUT) {
823 let has_miss = group
824 .iter()
825 .any(|&(ci, _, _)| !self.cache.contains((self.section_index, ci as u32)));
826 if has_miss {
827 // Real decode work: fault this group's chunks in parallel on the
828 // bounded pool (order preserved so each entry lines up with its
829 // segment), then fill serially. A failed fault yields `None`
830 // (recorded in `self.failed`), same as the serial path.
831 let entries: Vec<Option<Arc<CacheEntry>>> = decode_pool().install(|| {
832 group
833 .par_iter()
834 .map(|&(ci, _, _)| self.chunk_entry(ci))
835 .collect()
836 });
837 for (&(ci, s, e), entry) in group.iter().zip(entries) {
838 if let Some(entry) = entry {
839 self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
840 }
841 }
842 } else {
843 // Cache-warm: no decode to parallelize, so avoid the dispatch.
844 for &(ci, s, e) in group {
845 if let Some(entry) = self.chunk_entry(ci) {
846 self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
847 }
848 }
849 }
850 }
851 }
852
853 /// Serial fallback for builds without the `parallel` feature (wasm, and
854 /// `--no-default-features`): fault and walk one chunk at a time.
855 #[cfg(not(feature = "parallel"))]
856 fn decode_and_walk(
857 &self,
858 segments: &[(usize, usize, usize)],
859 jobs: &[(u32, usize)],
860 out: &mut [Option<String>],
861 ri: usize,
862 tc: u32,
863 ) {
864 for &(ci, s, e) in segments {
865 if let Some(entry) = self.chunk_entry(ci) {
866 self.walk_chunk(ci, &entry, &jobs[s..e], out, ri, tc);
867 }
868 }
869 }
870
871 /// Fill the output slots owned by one chunk: `chunk_jobs` is the id-sorted
872 /// slice of `(id, slot)` jobs that all live in chunk `ci`, and `entry` is
873 /// that chunk's already-faulted body. Walks the chunk's front-coded runs
874 /// once each, forward, picking off every requested id as the decode passes
875 /// it — the single-pass twin of `term(id)`, writing exactly the bytes
876 /// `term(id)` would for each id and leaving a slot None where `term(id)` is
877 /// None (a malformed run entry, past which no larger id in that run decodes).
878 fn walk_chunk(
879 &self,
880 ci: usize,
881 entry: &CacheEntry,
882 chunk_jobs: &[(u32, usize)],
883 out: &mut [Option<String>],
884 ri: usize,
885 tc: u32,
886 ) {
887 let bytes: &[u8] = entry.body();
888 let mut k = 0;
889 while k < chunk_jobs.len() {
890 let id = chunk_jobs[k].0;
891 if id == ABSENT || id > tc {
892 k += 1;
893 continue;
894 }
895 let run = (id - 1) as usize / ri;
896 // Decode this run's restart entry. Any failure here means
897 // `term(id)` is None for the whole run — skip every job in it.
898 let run_start = self.run_off_in_chunk(ci, run, entry, ri).and_then(|off| {
899 let mut buf = Vec::new();
900 run_entry_into(bytes, off, &mut buf).map(|pos| (buf, pos))
901 });
902 let Some((mut buf, mut pos)) = run_start else {
903 while k < chunk_jobs.len() {
904 let id = chunk_jobs[k].0;
905 if id == ABSENT || id > tc {
906 k += 1;
907 continue;
908 }
909 if (id - 1) as usize / ri != run {
910 break;
911 }
912 k += 1;
913 }
914 continue;
915 };
916 // `buf` holds the run's term #`cur_step`; advance monotonically.
917 let mut cur_step = 0usize;
918 let mut broken = false;
919 while k < chunk_jobs.len() {
920 let id = chunk_jobs[k].0;
921 if id == ABSENT || id > tc {
922 k += 1;
923 continue;
924 }
925 if (id - 1) as usize / ri != run {
926 break; // next run
927 }
928 let steps = (id - 1) as usize % ri;
929 if !broken {
930 while cur_step < steps {
931 match entry_into(bytes, pos, &mut buf) {
932 Some(next) => {
933 pos = next;
934 cur_step += 1;
935 }
936 None => {
937 // Malformed past this point: `term(id)` is None
938 // here and for every larger id in the run (they
939 // all decode through this entry).
940 broken = true;
941 break;
942 }
943 }
944 }
945 if !broken && cur_step == steps {
946 out[chunk_jobs[k].1] = Some(String::from_utf8_lossy(&buf).into_owned());
947 }
948 }
949 k += 1;
950 }
951 }
952 }
953
954 /// Resolve `term` to its ID. Chunk-level binary search runs on the (local)
955 /// chunk directory, so this also costs at most one chunk fault.
956 pub fn id(&self, term: &str) -> Option<u32> {
957 if self.chunks.is_empty() {
958 return None;
959 }
960 let ri = self.meta.restart_interval as usize;
961 // Pick the chunk: the last one whose ROUTING KEY is <= `term` (the
962 // single local chunk skips the search — its key is unset).
963 //
964 // This is the ONLY read of [`SectionChunk::key`] in the codebase, and
965 // all it needs is the separator invariant `last(i-1) < key(i) <=
966 // first(i)`: `term` lands in chunk `i` exactly when `key(i) <= term <
967 // key(i+1)`, which covers every term of chunk `i` and nothing in any
968 // other chunk. A verbatim first term satisfies it too, which is why
969 // files written either way route identically.
970 //
971 // With separators the search can land on chunk `i` for a `term` in the
972 // gap `key(i) <= term < first(i)`; the run search below then hits its
973 // `lo == first_run` branch and answers `None`, which is correct — such
974 // a term is strictly between two chunks, i.e. absent.
975 let ci = if self.chunks.len() == 1 {
976 0
977 } else {
978 let i = self
979 .chunks
980 .partition_point(|c| c.key.as_slice() <= term.as_bytes());
981 i.checked_sub(1)?
982 };
983 let first_run = self.chunks[ci].first_run;
984 let entry = self.chunk_entry(ci)?;
985 let bytes: &[u8] = entry.body();
986
987 // Binary search this chunk's runs by their first (full) term.
988 let run_end = self.run_end_of_chunk(ci, &entry, ri);
989 let mut buf = Vec::new();
990 let mut lo = first_run;
991 let mut hi = run_end;
992 while lo < hi {
993 let mid = (lo + hi) / 2;
994 let off = self.run_off_in_chunk(ci, mid, &entry, ri)?;
995 run_entry_into(bytes, off, &mut buf)?;
996 if buf.as_slice() <= term.as_bytes() {
997 lo = mid + 1;
998 } else {
999 hi = mid;
1000 }
1001 }
1002 if lo == first_run {
1003 // Smaller than every term in (and before) this chunk. Unreachable
1004 // when the key is a verbatim first term; reachable, and correct,
1005 // when it is a separator: `key(ci) <= term < first(ci)` puts the
1006 // term strictly between two chunks.
1007 return None;
1008 }
1009 let run = lo - 1;
1010 let off = self.run_off_in_chunk(ci, run, &entry, ri)?;
1011 let mut pos = run_entry_into(bytes, off, &mut buf)?;
1012 let base_id = (run * ri) as u32 + 1;
1013 // saturating_sub: corrupt metadata must not underflow-panic.
1014 let run_len = self.meta.restart_interval.min(
1015 self.meta
1016 .term_count
1017 .saturating_sub(run as u32 * self.meta.restart_interval),
1018 );
1019 for step in 0..run_len {
1020 if buf.as_slice() == term.as_bytes() {
1021 return Some(base_id + step);
1022 }
1023 if buf.as_slice() > term.as_bytes() {
1024 return None;
1025 }
1026 if step + 1 < run_len {
1027 pos = entry_into(bytes, pos, &mut buf)?;
1028 }
1029 }
1030 None
1031 }
1032
1033 /// The full serialized section (header + body), for the file writer. The
1034 /// local single-chunk case returns the stored bytes; a chunked section
1035 /// re-assembles them (header re-encoded from the metadata).
1036 pub fn raw_section_bytes(&self) -> Vec<u8> {
1037 if self.chunks.len() == 1 && self.chunks[0].body_start == 0 {
1038 if let Some(entry) = self.cache.get((self.section_index, 0)) {
1039 return entry.body().to_vec();
1040 }
1041 }
1042 let mut out = encode_section_header(&self.meta);
1043 for ci in 0..self.chunks.len() {
1044 if let Some(entry) = self.chunk_entry(ci) {
1045 out.extend_from_slice(entry.body());
1046 }
1047 }
1048 out
1049 }
1050}
1051
1052/// Re-encode a section header (term_count, interval, restart table) from its
1053/// parsed metadata — the inverse of [`parse_meta`]. Restart offsets are stored
1054/// body-relative on disk; `meta` holds them absolute, so the body start is
1055/// re-derived as the first run's offset.
1056pub fn encode_section_header(meta: &SectionMeta) -> Vec<u8> {
1057 let body_start = meta.restart_offsets.first().copied().unwrap_or(0);
1058 let mut out = Vec::new();
1059 write_uvarint(&mut out, meta.term_count as u64);
1060 write_uvarint(&mut out, meta.restart_interval as u64);
1061 write_uvarint(&mut out, meta.restart_offsets.len() as u64);
1062 for off in &meta.restart_offsets {
1063 write_uvarint(&mut out, off.saturating_sub(body_start));
1064 }
1065 out
1066}
1067
1068/// A parsed, read-only dictionary section (parses its metadata on construction).
1069pub struct DictSection<'a> {
1070 bytes: &'a [u8],
1071 meta: SectionMeta,
1072}
1073
1074impl<'a> DictSection<'a> {
1075 pub fn parse(bytes: &'a [u8]) -> Result<Self, DictError> {
1076 Ok(Self {
1077 bytes,
1078 meta: parse_meta(bytes)?,
1079 })
1080 }
1081
1082 pub fn len(&self) -> u32 {
1083 self.meta.term_count
1084 }
1085
1086 pub fn is_empty(&self) -> bool {
1087 self.meta.term_count == 0
1088 }
1089
1090 /// Resolve `id` (1-based) to its term, or `None` if out of range.
1091 pub fn term(&self, id: u32) -> Option<String> {
1092 section_term(self.bytes, &self.meta, id)
1093 }
1094
1095 /// Resolve `term` to its ID, or `None` if absent.
1096 pub fn id(&self, term: &str) -> Option<u32> {
1097 section_id(self.bytes, &self.meta, term)
1098 }
1099}
1100
1101fn common_prefix_len(a: &str, b: &str) -> usize {
1102 a.bytes().zip(b.bytes()).take_while(|(x, y)| x == y).count()
1103}
1104
1105#[cfg(test)]
1106mod tests {
1107 use super::*;
1108
1109 /// Build a *lite* (empty section-wide restart table), genuinely multi-chunk
1110 /// [`ChunkedSection`] from `terms`: the exact shape a ranged open produces,
1111 /// so `resolve_into`'s per-chunk run-table branch and its cross-chunk /
1112 /// cross-run walk are all exercised. `runs_per_chunk` chops the runs into
1113 /// several chunks regardless of size, so a tiny fixture still spans many.
1114 fn lite_multichunk(terms: &[&str], ri: u32, runs_per_chunk: usize) -> (ChunkedSection, u32) {
1115 let mut b = DictSectionBuilder::new().with_restart_interval(ri);
1116 for t in terms {
1117 b.push(*t);
1118 }
1119 let raw = b.build();
1120 let full_meta = parse_meta(&raw).unwrap();
1121 let n_runs = full_meta.restart_offsets.len();
1122 let mut chunks = Vec::new();
1123 let mut bodies: Vec<Vec<u8>> = Vec::new();
1124 let mut r = 0;
1125 while r < n_runs {
1126 let first_run = r;
1127 let start = full_meta.restart_offsets[r] as usize;
1128 let r2 = (r + runs_per_chunk).min(n_runs);
1129 let end = if r2 < n_runs {
1130 full_meta.restart_offsets[r2] as usize
1131 } else {
1132 raw.len()
1133 };
1134 chunks.push(SectionChunk::new(first_run, Vec::new(), start as u64));
1135 bodies.push(raw[start..end].to_vec());
1136 r = r2;
1137 }
1138 assert!(chunks.len() > 1, "fixture must be multi-chunk");
1139 // Lite: no section-wide restart table -> run offsets come from each
1140 // decoded chunk body, cached in its cache entry (the ranged path).
1141 let lite_meta = SectionMeta {
1142 term_count: full_meta.term_count,
1143 restart_interval: full_meta.restart_interval,
1144 restart_offsets: Vec::new(),
1145 };
1146 let loader: ChunkLoader = Box::new(move |ci| bodies.get(ci).cloned());
1147 let sec = ChunkedSection::from_parts(
1148 lite_meta,
1149 chunks,
1150 Some(loader),
1151 ChunkCache::unlimited_arc(),
1152 0,
1153 );
1154 (sec, full_meta.term_count)
1155 }
1156
1157 /// The windowed batch resolver must fill exactly what a per-id `term()`
1158 /// would, for a scrambled, duplicated, out-of-range id set spread across
1159 /// many chunks and runs — the byte-identity guarantee `dump_filtered_each`
1160 /// rests on (phase 1). Also cross-checked against the ground-truth
1161 /// `section_term` decoder.
1162 #[test]
1163 fn resolve_into_matches_term_across_chunks_and_runs() {
1164 // Shared prefixes (front-coding), a long literal (a run that dwarfs its
1165 // neighbors), blank nodes — 240 distinct terms.
1166 let owned: Vec<String> = (0..240)
1167 .map(|i| match i % 4 {
1168 0 => format!("<http://example.org/entity/{i:05}>"),
1169 1 => format!("<http://example.org/entity/{i:05}/sub/leaf>"),
1170 2 => format!("\"literal value {} padded {}\"", i, "x".repeat(i % 50)),
1171 _ => format!("_:b{i:05}"),
1172 })
1173 .collect();
1174 let terms: Vec<&str> = owned.iter().map(String::as_str).collect();
1175
1176 for (ri, rpc) in [(1u32, 5usize), (4, 3), (16, 2)] {
1177 let (sec, count) = lite_multichunk(&terms, ri, rpc);
1178 let raw = {
1179 let mut b = DictSectionBuilder::new().with_restart_interval(ri);
1180 for t in &terms {
1181 b.push(*t);
1182 }
1183 b.build()
1184 };
1185 let full_meta = parse_meta(&raw).unwrap();
1186
1187 // A scrambled probe set: every id once, a run of duplicates, some
1188 // out-of-range ids (0 and past the end) that must stay `None`.
1189 let mut probe: Vec<u32> = (1..=count).collect();
1190 probe.extend([1, 1, count, count, ABSENT, count + 3, count + 100]);
1191 // Deterministic shuffle so chunk order != probe order.
1192 let mut state = 0xDEAD_BEEF_1234_5678u64;
1193 for k in (1..probe.len()).rev() {
1194 state ^= state << 13;
1195 state ^= state >> 7;
1196 state ^= state << 17;
1197 probe.swap(k, (state % (k as u64 + 1)) as usize);
1198 }
1199
1200 let mut jobs: Vec<(u32, usize)> = probe.iter().map(|&id| (id, 0)).collect();
1201 for (slot, j) in jobs.iter_mut().enumerate() {
1202 j.1 = slot;
1203 }
1204 let mut out: Vec<Option<String>> = vec![None; probe.len()];
1205 sec.resolve_into(&mut jobs, &mut out);
1206
1207 for (slot, &id) in probe.iter().enumerate() {
1208 let via_term = sec.term(id);
1209 assert_eq!(
1210 out[slot], via_term,
1211 "resolve_into(id={id}) != term(id) at ri={ri} rpc={rpc}"
1212 );
1213 // Ground truth for in-range ids.
1214 let truth = section_term(&raw, &full_meta, id);
1215 assert_eq!(out[slot], truth, "resolve_into(id={id}) != section_term");
1216 }
1217 }
1218 }
1219
1220 fn sample() -> Vec<String> {
1221 // Deliberately unsorted, with shared prefixes and a duplicate.
1222 [
1223 "http://ex.org/Alice",
1224 "http://ex.org/Bob",
1225 "http://ex.org/Alan",
1226 "http://ex.org/knows",
1227 "http://ex.org/Alice", // dup
1228 "zeta",
1229 ]
1230 .iter()
1231 .map(|s| s.to_string())
1232 .collect()
1233 }
1234
1235 #[test]
1236 fn round_trip_all_ids_and_terms() {
1237 for r in [1u32, 2, 16, 1000] {
1238 let mut b = DictSectionBuilder::new().with_restart_interval(r);
1239 for t in sample() {
1240 b.push(t);
1241 }
1242 let bytes = b.build();
1243 let sec = DictSection::parse(&bytes).unwrap();
1244
1245 // Expected sorted unique set.
1246 let mut expected = sample();
1247 expected.sort();
1248 expected.dedup();
1249 assert_eq!(sec.len() as usize, expected.len());
1250
1251 for (i, term) in expected.iter().enumerate() {
1252 let id = (i + 1) as u32;
1253 assert_eq!(
1254 sec.term(id).as_deref(),
1255 Some(term.as_str()),
1256 "term({id}) r={r}"
1257 );
1258 assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
1259 }
1260 }
1261 }
1262
1263 #[test]
1264 fn lookups_for_absent_terms() {
1265 let mut b = DictSectionBuilder::new();
1266 for t in sample() {
1267 b.push(t);
1268 }
1269 let bytes = b.build();
1270 let sec = DictSection::parse(&bytes).unwrap();
1271 assert_eq!(sec.id("aaa-before-everything"), None);
1272 assert_eq!(sec.id("zzz-after-everything"), None);
1273 assert_eq!(sec.id("http://ex.org/Alic"), None); // prefix, not present
1274 assert_eq!(sec.term(0), None);
1275 assert_eq!(sec.term(9999), None);
1276 }
1277
1278 /// Property-style stress of the front-coded decode paths: a large
1279 /// deterministic-pseudo-random term pool (IRIs with heavy shared prefixes,
1280 /// literals, blank nodes — duplicates likely) must round-trip every id and
1281 /// term across restart intervals, and near-miss probes (a stored term ± a
1282 /// suffix/truncation, including run-boundary terms) must resolve to `None`.
1283 /// Term resolution is the engine's hot path, so this is the test that
1284 /// guards the buffer-reuse decode.
1285 #[test]
1286 fn randomized_round_trip_and_near_misses_across_restart_intervals() {
1287 let mut state = 0x9E37_79B9_7F4A_7C15u64;
1288 let mut next = move || {
1289 state ^= state << 13;
1290 state ^= state >> 7;
1291 state ^= state << 17;
1292 state
1293 };
1294 let pool: Vec<String> = (0..1500)
1295 .map(|i| {
1296 let n = next();
1297 match n % 4 {
1298 0 => format!("<http://example.org/entity/{n:x}>"),
1299 1 => format!("\"literal value {} with spaces\"", n % 300), // dups likely
1300 2 => format!("<http://example.org/entity/{}/sub/{i}>", n % 64), // long shared prefixes
1301 _ => format!("_:b{}", n % 256),
1302 }
1303 })
1304 .collect();
1305 let mut expected = pool.clone();
1306 expected.sort();
1307 expected.dedup();
1308
1309 for r in [1u32, 3, 16, 64] {
1310 let mut b = DictSectionBuilder::new().with_restart_interval(r);
1311 for t in &pool {
1312 b.push(t.clone());
1313 }
1314 let bytes = b.build();
1315 let sec = DictSection::parse(&bytes).unwrap();
1316 assert_eq!(sec.len() as usize, expected.len(), "r={r}");
1317
1318 for (i, term) in expected.iter().enumerate() {
1319 let id = (i + 1) as u32;
1320 assert_eq!(
1321 sec.term(id).as_deref(),
1322 Some(term.as_str()),
1323 "term({id}) r={r}"
1324 );
1325 assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
1326 }
1327 // Near misses around run boundaries (and a sample of the rest):
1328 // an appended suffix or a truncation is never a stored term unless
1329 // it happens to collide with one.
1330 for (i, term) in expected.iter().enumerate() {
1331 let near_boundary = (i as u32) % r <= 1;
1332 if !near_boundary && i % 37 != 0 {
1333 continue;
1334 }
1335 let longer = format!("{term}\u{1}");
1336 assert_eq!(sec.id(&longer), None, "near-miss long r={r}");
1337 let mut shorter = term.clone();
1338 shorter.pop();
1339 if !shorter.is_empty() && expected.binary_search(&shorter).is_err() {
1340 assert_eq!(sec.id(&shorter), None, "near-miss short {shorter:?} r={r}");
1341 }
1342 }
1343 assert_eq!(sec.term(expected.len() as u32 + 1), None, "past-end r={r}");
1344 }
1345 }
1346}