rete-core 0.3.2

Core format types for the Rete cloud-native RDF graph file.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
//! Front-coded, restart-indexed dictionary sections (SPEC.md §5.1).
//!
//! A section holds the UTF-8 terms of one kind (shared / subjects / objects /
//! predicates / graphs), sorted and assigned dense 1-based IDs. Terms are stored
//! in runs of `R`; each run starts with a full term and front-codes the rest.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::OnceLock;

use crate::varint::{read_uvarint, write_uvarint};

/// Default restart interval: a full term every `R` entries.
pub const DEFAULT_RESTART_INTERVAL: u32 = 16;

/// Reserved ID meaning "no such term".
pub const ABSENT: u32 = 0;

#[derive(Debug, thiserror::Error)]
pub enum DictError {
    #[error("malformed dictionary section: {0}")]
    Malformed(&'static str),
}

/// Build a dictionary section from terms (any order; sorted + deduped here).
#[derive(Default)]
pub struct DictSectionBuilder {
    terms: Vec<String>,
    restart_interval: u32,
}

/// Terms per restart run, overridable via `RETE_DICT_RESTART_INTERVAL`.
///
/// The interval sets random-access granularity as well as front-coding gain:
/// chunks are cut on whole-run boundaries, so with very large literals (e.g.
/// base64-embedded images) a single run dwarfs the 64 KiB chunk budget and one
/// term lookup drags in ~`r` terms' worth of bytes. Setting `1` makes every term
/// its own restart — direct seek, larger restart table, no front-coding (which
/// is worthless for base64 anyway, since such terms share no prefix).
pub fn env_restart_interval() -> u32 {
    static R: std::sync::OnceLock<u32> = std::sync::OnceLock::new();
    *R.get_or_init(|| {
        std::env::var("RETE_DICT_RESTART_INTERVAL")
            .ok()
            .and_then(|v| v.parse::<u32>().ok())
            .filter(|&v| v >= 1)
            .unwrap_or(DEFAULT_RESTART_INTERVAL)
    })
}

impl DictSectionBuilder {
    pub fn new() -> Self {
        Self {
            terms: Vec::new(),
            restart_interval: env_restart_interval(),
        }
    }

    pub fn with_restart_interval(mut self, r: u32) -> Self {
        assert!(r >= 1, "restart interval must be >= 1");
        self.restart_interval = r;
        self
    }

    pub fn push(&mut self, term: impl Into<String>) {
        self.terms.push(term.into());
    }

    /// Serialize to bytes. Terms are sorted and deduped; the resulting IDs are
    /// `1..=n` in sorted order.
    pub fn build(mut self) -> Vec<u8> {
        self.terms.sort_unstable();
        self.terms.dedup();
        let r = self.restart_interval as usize;
        let n = self.terms.len();
        let num_restarts = n.div_ceil(r);

        // Encode body and capture each run's starting offset (relative to body).
        let mut body = Vec::new();
        let mut restart_offsets = Vec::with_capacity(num_restarts);
        let mut prev = "";
        for (i, term) in self.terms.iter().enumerate() {
            if i % r == 0 {
                restart_offsets.push(body.len() as u64);
                // restart entry: shared = 0, full term
                write_uvarint(&mut body, 0);
                write_uvarint(&mut body, term.len() as u64);
                body.extend_from_slice(term.as_bytes());
            } else {
                let shared = common_prefix_len(prev, term);
                let suffix = &term.as_bytes()[shared..];
                write_uvarint(&mut body, shared as u64);
                write_uvarint(&mut body, suffix.len() as u64);
                body.extend_from_slice(suffix);
            }
            prev = term;
        }

        // header || restart-offset table || body
        let mut out = Vec::new();
        write_uvarint(&mut out, n as u64);
        write_uvarint(&mut out, self.restart_interval as u64);
        write_uvarint(&mut out, num_restarts as u64);
        for off in &restart_offsets {
            write_uvarint(&mut out, *off);
        }
        out.extend_from_slice(&body);
        out
    }
}

/// Parsed section metadata (the restart table) — cache this once and reuse it
/// across lookups instead of re-parsing the section header every time.
#[derive(Debug, Clone)]
pub struct SectionMeta {
    pub term_count: u32,
    pub restart_interval: u32,
    /// Absolute offsets into the section bytes for each run start.
    ///
    /// `u64`, not `usize`: a section may exceed 4 GiB and wasm32 is a 32-bit
    /// target, so a `usize` offset silently truncates there (see #70). A
    /// dictionary carrying embedded media is exactly this case — WikiArt's is
    /// 23.4 GB.
    pub restart_offsets: Vec<u64>,
}

/// Parse only the header/restart table of a section.
pub fn parse_meta(bytes: &[u8]) -> Result<SectionMeta, DictError> {
    let mut pos = 0;
    let take = |pos: &mut usize| -> Result<u64, DictError> {
        let (v, n) =
            read_uvarint(&bytes[*pos..]).ok_or(DictError::Malformed("truncated header"))?;
        *pos += n;
        Ok(v)
    };
    let term_count = take(&mut pos)? as u32;
    let restart_interval = take(&mut pos)? as u32;
    let num_restarts = take(&mut pos)? as usize;
    if restart_interval == 0 {
        return Err(DictError::Malformed("zero restart interval"));
    }
    // `num_restarts` is untrusted; each restart is ≥1 byte, so cap the
    // pre-allocation at the buffer length to avoid an OOM on a bogus count.
    let mut rel = Vec::with_capacity(num_restarts.min(bytes.len()));
    for _ in 0..num_restarts {
        rel.push(take(&mut pos)?);
    }
    let body_start = pos as u64;
    Ok(SectionMeta {
        term_count,
        restart_interval,
        restart_offsets: rel.into_iter().map(|o| body_start + o).collect(),
    })
}

/// Decode the front-coded entry at `pos` *into* `buf`: truncate `buf` to the
/// entry's shared-prefix length and append its suffix bytes. For a restart
/// entry the stored shared length is 0, so the same decode works for both
/// entry kinds. Returns the next entry's position; `None` on malformed bytes
/// (including a shared length longer than the previous term). Allocation-free
/// after `buf`'s first growth — this is the hot path of every term resolution.
#[inline]
fn entry_into(bytes: &[u8], pos: usize, buf: &mut Vec<u8>) -> Option<usize> {
    let (shared, n1) = read_uvarint(bytes.get(pos..)?)?;
    let p = pos + n1;
    let (suf, n2) = read_uvarint(bytes.get(p..)?)?;
    let start = p + n2;
    let end = start
        .checked_add(suf as usize)
        .filter(|&e| e <= bytes.len())?;
    if shared as usize > buf.len() {
        return None;
    }
    buf.truncate(shared as usize);
    buf.extend_from_slice(&bytes[start..end]);
    Some(end)
}

/// Decode a run's restart (full-term) entry at `off` into `buf`.
fn run_entry_into(bytes: &[u8], off: usize, buf: &mut Vec<u8>) -> Option<usize> {
    buf.clear(); // a restart entry stands alone; stale prefix bytes must not leak
    entry_into(bytes, off, buf)
}

/// Resolve `id` (1-based) to its term using cached metadata. Returns `None` on
/// any inconsistency in untrusted metadata/bytes rather than panicking.
pub fn section_term(bytes: &[u8], meta: &SectionMeta, id: u32) -> Option<String> {
    if id == ABSENT || id > meta.term_count {
        return None;
    }
    let idx = (id - 1) as usize;
    let run = idx / meta.restart_interval as usize;
    let steps = idx % meta.restart_interval as usize;
    let mut buf = Vec::new();
    let mut pos = run_entry_into(bytes, *meta.restart_offsets.get(run)? as usize, &mut buf)?;
    for _ in 0..steps {
        pos = entry_into(bytes, pos, &mut buf)?;
    }
    Some(String::from_utf8_lossy(&buf).into_owned())
}

/// Resolve `term` to its ID using cached metadata.
pub fn section_id(bytes: &[u8], meta: &SectionMeta, term: &str) -> Option<u32> {
    let mut buf = Vec::new();
    // Binary search restart runs by their first (full) term.
    let mut lo = 0usize;
    let mut hi = meta.restart_offsets.len();
    while lo < hi {
        let mid = (lo + hi) / 2;
        run_entry_into(bytes, meta.restart_offsets[mid] as usize, &mut buf)?;
        if buf.as_slice() <= term.as_bytes() {
            lo = mid + 1;
        } else {
            hi = mid;
        }
    }
    if lo == 0 {
        return None; // smaller than every term
    }
    let run = lo - 1;
    let mut pos = run_entry_into(bytes, meta.restart_offsets[run] as usize, &mut buf)?;
    let base_id = (run * meta.restart_interval as usize) as u32 + 1;
    // saturating_sub: corrupt metadata where run*interval > term_count must not
    // underflow-panic.
    let run_len = meta.restart_interval.min(
        meta.term_count
            .saturating_sub(run as u32 * meta.restart_interval),
    );
    for step in 0..run_len {
        if buf.as_slice() == term.as_bytes() {
            return Some(base_id + step);
        }
        if buf.as_slice() > term.as_bytes() {
            return None;
        }
        if step + 1 < run_len {
            pos = entry_into(bytes, pos, &mut buf)?;
        }
    }
    None
}

// --- chunked sections ---------------------------------------------------------
//
// A section's body is split (on run boundaries) into **chunks** that are
// compressed and fetched independently (format v0.2), so a remote client
// resolves a term with one chunk fault instead of downloading the whole
// section. Local sections are the degenerate case: one pre-set chunk holding
// the entire serialized section — a single code path serves both.

/// Fetches one chunk's decompressed body slice on demand (`None` = the fetch
/// failed; the section records it and the lookup misses — callers over remote
/// data must check [`ChunkedSection::load_incomplete`] after evaluating).
pub type ChunkLoader = Box<dyn Fn(usize) -> Option<Vec<u8>> + Send + Sync>;

/// Fetches **many** chunks in one round trip: given ascending chunk indices,
/// returns each chunk's decompressed body in the same order. The ranged
/// reader implements this by coalescing byte-adjacent chunk ranges into
/// single range reads — a full-dictionary sweep (export, dump) costs a few
/// requests per section instead of one per chunk. `None` = the batch failed;
/// callers fall back to the per-chunk [`ChunkLoader`].
pub type ChunkBulkLoader = Box<dyn Fn(&[usize]) -> Option<Vec<Vec<u8>>> + Send + Sync>;

/// One chunk: a run-aligned slice of the section body. `body_start` is the
/// offset (in the section's coordinate space — the same space
/// [`SectionMeta::restart_offsets`] uses) where `data[0]` sits.
pub struct SectionChunk {
    first_run: usize,
    /// First term of the chunk (for chunk-level binary search in `id`);
    /// unused (empty) for the single local chunk.
    first_term: Vec<u8>,
    body_start: u64,
    data: OnceLock<Vec<u8>>,
    /// This chunk's per-run byte offsets, **relative to its own decompressed
    /// body** — the chunk-local stand-in for the section-wide restart table.
    /// Derived by scanning the body once on first lookup and cached, so a
    /// remote open never materializes the section's millions of restart
    /// offsets (a 50 M-term section's table is ~24 MiB — an iOS-Safari OOM).
    runs: OnceLock<Vec<usize>>,
}

impl SectionChunk {
    /// A remote chunk descriptor (data faults in through the loader).
    pub fn remote(first_run: usize, first_term: Vec<u8>, body_start: u64) -> Self {
        SectionChunk {
            first_run,
            first_term,
            body_start,
            data: OnceLock::new(),
            runs: OnceLock::new(),
        }
    }

    /// A resident chunk (data already decoded).
    pub fn resident(first_run: usize, first_term: Vec<u8>, body_start: u64, data: Vec<u8>) -> Self {
        let cell = OnceLock::new();
        let _ = cell.set(data);
        SectionChunk {
            first_run,
            first_term,
            body_start,
            data: cell,
            runs: OnceLock::new(),
        }
    }

    /// The byte offset (into `data`) of each run in this chunk, computed once by
    /// scanning the decompressed body and cached. `data` must be this chunk's
    /// body. Offset 0 is the chunk's first run (chunks are run-aligned); every
    /// `restart_interval` terms starts the next run. An EMPTY body (the
    /// transient-fetch-failure sentinel — never a real chunk) is not cached, or
    /// a failed chunk's empty offset table would outlive the retryable data.
    fn run_offsets(&self, data: &[u8], restart_interval: usize) -> &[usize] {
        if let Some(r) = self.runs.get() {
            return r;
        }
        if data.is_empty() {
            return &[];
        }
        self.runs
            .get_or_init(|| chunk_run_offsets(data, restart_interval))
    }
}

/// Scan a chunk's decompressed body for the byte offset of each run start (see
/// [`SectionChunk::run_offsets`]). O(terms-in-chunk), run once per faulted chunk.
fn chunk_run_offsets(data: &[u8], restart_interval: usize) -> Vec<usize> {
    let mut offs = vec![0usize];
    let mut pos = 0usize;
    let mut buf = Vec::new();
    let mut count = 0usize;
    while pos < data.len() {
        let Some(next) = entry_into(data, pos, &mut buf) else {
            break;
        };
        count += 1;
        pos = next;
        if pos < data.len() && count.is_multiple_of(restart_interval) {
            offs.push(pos);
        }
    }
    offs
}

/// The first (full) term of the run starting at `off`, as raw bytes. `None`
/// on malformed bytes.
pub fn run_first_term(bytes: &[u8], off: usize) -> Option<Vec<u8>> {
    let mut buf = Vec::new();
    run_entry_into(bytes, off, &mut buf)?;
    Some(buf)
}

/// A dictionary section whose body is served chunk-by-chunk: metadata + chunk
/// directory always present, chunk bytes local or faulted in on first touch.
pub struct ChunkedSection {
    meta: SectionMeta,
    chunks: Vec<SectionChunk>,
    loader: Option<ChunkLoader>,
    bulk: Option<ChunkBulkLoader>,
    failed: AtomicBool,
}

impl ChunkedSection {
    /// A local section: the whole serialized section (header + body) as one
    /// pre-set chunk at coordinate 0, so the absolute restart offsets index it
    /// directly. Malformed bytes degrade to an empty section (no panics on
    /// untrusted files), matching the previous reader behavior.
    pub fn local(section_bytes: Vec<u8>) -> Self {
        let meta = parse_meta(&section_bytes).unwrap_or(SectionMeta {
            term_count: 0,
            restart_interval: 1,
            restart_offsets: Vec::new(),
        });
        let data = OnceLock::new();
        let _ = data.set(section_bytes);
        ChunkedSection {
            meta,
            chunks: vec![SectionChunk {
                first_run: 0,
                first_term: Vec::new(),
                body_start: 0,
                data,
                runs: OnceLock::new(),
            }],
            loader: None,
            bulk: None,
            failed: AtomicBool::new(false),
        }
    }

    /// A section from parsed parts: metadata + chunk list, with an optional
    /// loader for non-resident chunks (the remote lazy-open path) — resident
    /// chunk lists (a locally-decoded chunked section) pass `None`.
    pub fn from_parts(
        meta: SectionMeta,
        chunks: Vec<SectionChunk>,
        loader: Option<ChunkLoader>,
    ) -> Self {
        ChunkedSection {
            meta,
            chunks,
            loader,
            bulk: None,
            failed: AtomicBool::new(false),
        }
    }

    /// Attach a batched chunk fetcher (see [`ChunkBulkLoader`]): full-section
    /// sweeps ([`prefetch_all`](Self::prefetch_all)) go through it instead of
    /// faulting chunk by chunk.
    pub fn with_bulk_loader(mut self, bulk: ChunkBulkLoader) -> Self {
        self.bulk = Some(bulk);
        self
    }

    /// Batch-fault every unloaded chunk through the bulk loader, if one is
    /// attached and at least two chunks are missing. Callers about to sweep
    /// the whole section (export/dump term resolution) call this once; a
    /// failed batch leaves the chunks unloaded for the per-chunk loader to
    /// retry (and record failures) lookup by lookup.
    pub fn prefetch_all(&self) {
        self.prefetch_chunks(&(0..self.chunks.len()).collect::<Vec<_>>());
    }

    /// Batch-fault a *specific* set of chunks (the subset a bounded query's
    /// output touches) through the bulk loader, coalescing byte-adjacent ranges
    /// into single reads. `cis` should be ascending and deduped; chunks already
    /// resident are skipped. Like [`prefetch_all`](Self::prefetch_all) a single
    /// missing chunk is left for the per-chunk loader, and a failed batch leaves
    /// the chunks unloaded for that loader to retry (and record failures).
    pub fn prefetch_chunks(&self, cis: &[usize]) {
        let Some(bulk) = &self.bulk else { return };
        let missing: Vec<usize> = cis
            .iter()
            .copied()
            .filter(|&ci| self.chunks.get(ci).is_some_and(|c| c.data.get().is_none()))
            .collect();
        if missing.len() < 2 {
            return;
        }
        if let Some(bodies) = bulk(&missing) {
            if bodies.len() == missing.len() {
                for (&ci, body) in missing.iter().zip(bodies) {
                    let _ = self.chunks[ci].data.set(body);
                }
            }
        }
    }

    pub fn meta(&self) -> &SectionMeta {
        &self.meta
    }

    pub fn term_count(&self) -> u32 {
        self.meta.term_count
    }

    /// Did any chunk fetch fail since this section was opened — or since the
    /// last [`reset_load_failure`](Self::reset_load_failure)?
    pub fn load_incomplete(&self) -> bool {
        self.failed.load(Ordering::Relaxed)
    }

    /// Forget recorded fetch failures (see `GraphIndex::reset_load_failure` —
    /// the per-query reset for resident sessions). Failed chunks were never
    /// cached, so the next resolution retries them.
    pub fn reset_load_failure(&self) {
        self.failed.store(false, Ordering::Relaxed);
    }

    /// A FAILED fetch records the failure and returns an empty slice WITHOUT
    /// caching it, so a later resolution retries the chunk — a transient
    /// network error must not permanently poison a resident session.
    fn chunk_data(&self, ci: usize) -> &[u8] {
        let cell = &self.chunks[ci].data;
        if let Some(d) = cell.get() {
            return d;
        }
        match &self.loader {
            Some(load) => match load(ci) {
                Some(bytes) => cell.get_or_init(|| bytes),
                None => {
                    self.failed.store(true, Ordering::Relaxed);
                    &[]
                }
            },
            None => cell.get_or_init(Vec::new),
        }
    }

    /// The chunk holding `run` (chunks ascend by `first_run`; the first chunk
    /// always starts at run 0).
    fn chunk_of_run(&self, run: usize) -> Option<usize> {
        let i = self.chunks.partition_point(|c| c.first_run <= run);
        i.checked_sub(1)
    }

    /// The chunk index holding term `id` (1-based, section-local), or `None` if
    /// out of range. Mirrors the run→chunk math in [`term`](Self::term) so a
    /// caller can group a set of output ids by chunk and batch-prefetch them.
    pub fn chunk_of_id(&self, id: u32) -> Option<usize> {
        if id == ABSENT || id > self.meta.term_count {
            return None;
        }
        let run = (id - 1) as usize / self.meta.restart_interval as usize;
        self.chunk_of_run(run)
    }

    /// The byte offset (into chunk `ci`'s decompressed body `bytes`) of `run`.
    /// Full/local sections use the section-wide restart table (unchanged
    /// behavior); a *lite* remote section (empty `restart_offsets`) derives it
    /// from the chunk itself, so the open never holds the whole table.
    fn run_off_in_chunk(&self, ci: usize, run: usize, bytes: &[u8], ri: usize) -> Option<usize> {
        let chunk = &self.chunks[ci];
        if self.meta.restart_offsets.is_empty() {
            chunk
                .run_offsets(bytes, ri)
                .get(run.checked_sub(chunk.first_run)?)
                .copied()
        } else {
            self.meta
                .restart_offsets
                .get(run)?
                .checked_sub(chunk.body_start)
                .map(|o| o as usize)
        }
    }

    /// One past the last run index held by chunk `ci` (its run range is
    /// `[first_run, run_end)`).
    fn run_end_of_chunk(&self, ci: usize, bytes: &[u8], ri: usize) -> usize {
        if let Some(next) = self.chunks.get(ci + 1) {
            return next.first_run;
        }
        let chunk = &self.chunks[ci];
        if self.meta.restart_offsets.is_empty() {
            chunk.first_run + chunk.run_offsets(bytes, ri).len()
        } else {
            self.meta.restart_offsets.len()
        }
    }

    /// Resolve `id` (1-based) to its term. One chunk fault at most.
    /// Chunk index holding term `id` — [`term`](Self::term)'s routing without
    /// faulting the chunk. `None` for an absent/out-of-range id.
    pub(crate) fn chunk_of_term(&self, id: u32) -> Option<usize> {
        if id == ABSENT || id > self.meta.term_count {
            return None;
        }
        self.chunk_of_run((id - 1) as usize / self.meta.restart_interval as usize)
    }

    pub fn term(&self, id: u32) -> Option<String> {
        if id == ABSENT || id > self.meta.term_count {
            return None;
        }
        let idx = (id - 1) as usize;
        let ri = self.meta.restart_interval as usize;
        let run = idx / ri;
        let steps = idx % ri;
        let ci = self.chunk_of_run(run)?;
        let bytes = self.chunk_data(ci);
        let off = self.run_off_in_chunk(ci, run, bytes, ri)?;
        let mut buf = Vec::new();
        let mut pos = run_entry_into(bytes, off, &mut buf)?;
        for _ in 0..steps {
            pos = entry_into(bytes, pos, &mut buf)?;
        }
        Some(String::from_utf8_lossy(&buf).into_owned())
    }

    /// Resolve `term` to its ID. Chunk-level binary search runs on the (local)
    /// chunk directory, so this also costs at most one chunk fault.
    pub fn id(&self, term: &str) -> Option<u32> {
        if self.chunks.is_empty() {
            return None;
        }
        let ri = self.meta.restart_interval as usize;
        // Pick the chunk: the last one whose first term is <= `term` (the
        // single local chunk skips the search — its `first_term` is unset).
        let ci = if self.chunks.len() == 1 {
            0
        } else {
            let i = self
                .chunks
                .partition_point(|c| c.first_term.as_slice() <= term.as_bytes());
            i.checked_sub(1)?
        };
        let first_run = self.chunks[ci].first_run;
        let bytes = self.chunk_data(ci);

        // Binary search this chunk's runs by their first (full) term.
        let run_end = self.run_end_of_chunk(ci, bytes, ri);
        let mut buf = Vec::new();
        let mut lo = first_run;
        let mut hi = run_end;
        while lo < hi {
            let mid = (lo + hi) / 2;
            let off = self.run_off_in_chunk(ci, mid, bytes, ri)?;
            run_entry_into(bytes, off, &mut buf)?;
            if buf.as_slice() <= term.as_bytes() {
                lo = mid + 1;
            } else {
                hi = mid;
            }
        }
        if lo == first_run {
            return None; // smaller than every term in (and before) this chunk
        }
        let run = lo - 1;
        let off = self.run_off_in_chunk(ci, run, bytes, ri)?;
        let mut pos = run_entry_into(bytes, off, &mut buf)?;
        let base_id = (run * ri) as u32 + 1;
        // saturating_sub: corrupt metadata must not underflow-panic.
        let run_len = self.meta.restart_interval.min(
            self.meta
                .term_count
                .saturating_sub(run as u32 * self.meta.restart_interval),
        );
        for step in 0..run_len {
            if buf.as_slice() == term.as_bytes() {
                return Some(base_id + step);
            }
            if buf.as_slice() > term.as_bytes() {
                return None;
            }
            if step + 1 < run_len {
                pos = entry_into(bytes, pos, &mut buf)?;
            }
        }
        None
    }

    /// The full serialized section (header + body), for the file writer. The
    /// local single-chunk case returns the stored bytes; a chunked section
    /// re-assembles them (header re-encoded from the metadata).
    pub fn raw_section_bytes(&self) -> Vec<u8> {
        if self.chunks.len() == 1 && self.chunks[0].body_start == 0 {
            if let Some(bytes) = self.chunks[0].data.get() {
                return bytes.clone();
            }
        }
        let mut out = encode_section_header(&self.meta);
        for ci in 0..self.chunks.len() {
            out.extend_from_slice(self.chunk_data(ci));
        }
        out
    }
}

/// Re-encode a section header (term_count, interval, restart table) from its
/// parsed metadata — the inverse of [`parse_meta`]. Restart offsets are stored
/// body-relative on disk; `meta` holds them absolute, so the body start is
/// re-derived as the first run's offset.
pub fn encode_section_header(meta: &SectionMeta) -> Vec<u8> {
    let body_start = meta.restart_offsets.first().copied().unwrap_or(0);
    let mut out = Vec::new();
    write_uvarint(&mut out, meta.term_count as u64);
    write_uvarint(&mut out, meta.restart_interval as u64);
    write_uvarint(&mut out, meta.restart_offsets.len() as u64);
    for off in &meta.restart_offsets {
        write_uvarint(&mut out, off.saturating_sub(body_start));
    }
    out
}

/// A parsed, read-only dictionary section (parses its metadata on construction).
pub struct DictSection<'a> {
    bytes: &'a [u8],
    meta: SectionMeta,
}

impl<'a> DictSection<'a> {
    pub fn parse(bytes: &'a [u8]) -> Result<Self, DictError> {
        Ok(Self {
            bytes,
            meta: parse_meta(bytes)?,
        })
    }

    pub fn len(&self) -> u32 {
        self.meta.term_count
    }

    pub fn is_empty(&self) -> bool {
        self.meta.term_count == 0
    }

    /// Resolve `id` (1-based) to its term, or `None` if out of range.
    pub fn term(&self, id: u32) -> Option<String> {
        section_term(self.bytes, &self.meta, id)
    }

    /// Resolve `term` to its ID, or `None` if absent.
    pub fn id(&self, term: &str) -> Option<u32> {
        section_id(self.bytes, &self.meta, term)
    }
}

fn common_prefix_len(a: &str, b: &str) -> usize {
    a.bytes().zip(b.bytes()).take_while(|(x, y)| x == y).count()
}

#[cfg(test)]
mod tests {
    use super::*;

    fn sample() -> Vec<String> {
        // Deliberately unsorted, with shared prefixes and a duplicate.
        [
            "http://ex.org/Alice",
            "http://ex.org/Bob",
            "http://ex.org/Alan",
            "http://ex.org/knows",
            "http://ex.org/Alice", // dup
            "zeta",
        ]
        .iter()
        .map(|s| s.to_string())
        .collect()
    }

    #[test]
    fn round_trip_all_ids_and_terms() {
        for r in [1u32, 2, 16, 1000] {
            let mut b = DictSectionBuilder::new().with_restart_interval(r);
            for t in sample() {
                b.push(t);
            }
            let bytes = b.build();
            let sec = DictSection::parse(&bytes).unwrap();

            // Expected sorted unique set.
            let mut expected = sample();
            expected.sort();
            expected.dedup();
            assert_eq!(sec.len() as usize, expected.len());

            for (i, term) in expected.iter().enumerate() {
                let id = (i + 1) as u32;
                assert_eq!(
                    sec.term(id).as_deref(),
                    Some(term.as_str()),
                    "term({id}) r={r}"
                );
                assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
            }
        }
    }

    #[test]
    fn lookups_for_absent_terms() {
        let mut b = DictSectionBuilder::new();
        for t in sample() {
            b.push(t);
        }
        let bytes = b.build();
        let sec = DictSection::parse(&bytes).unwrap();
        assert_eq!(sec.id("aaa-before-everything"), None);
        assert_eq!(sec.id("zzz-after-everything"), None);
        assert_eq!(sec.id("http://ex.org/Alic"), None); // prefix, not present
        assert_eq!(sec.term(0), None);
        assert_eq!(sec.term(9999), None);
    }

    /// Property-style stress of the front-coded decode paths: a large
    /// deterministic-pseudo-random term pool (IRIs with heavy shared prefixes,
    /// literals, blank nodes — duplicates likely) must round-trip every id and
    /// term across restart intervals, and near-miss probes (a stored term ± a
    /// suffix/truncation, including run-boundary terms) must resolve to `None`.
    /// Term resolution is the engine's hot path, so this is the test that
    /// guards the buffer-reuse decode.
    #[test]
    fn randomized_round_trip_and_near_misses_across_restart_intervals() {
        let mut state = 0x9E37_79B9_7F4A_7C15u64;
        let mut next = move || {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            state
        };
        let pool: Vec<String> = (0..1500)
            .map(|i| {
                let n = next();
                match n % 4 {
                    0 => format!("<http://example.org/entity/{n:x}>"),
                    1 => format!("\"literal value {} with spaces\"", n % 300), // dups likely
                    2 => format!("<http://example.org/entity/{}/sub/{i}>", n % 64), // long shared prefixes
                    _ => format!("_:b{}", n % 256),
                }
            })
            .collect();
        let mut expected = pool.clone();
        expected.sort();
        expected.dedup();

        for r in [1u32, 3, 16, 64] {
            let mut b = DictSectionBuilder::new().with_restart_interval(r);
            for t in &pool {
                b.push(t.clone());
            }
            let bytes = b.build();
            let sec = DictSection::parse(&bytes).unwrap();
            assert_eq!(sec.len() as usize, expected.len(), "r={r}");

            for (i, term) in expected.iter().enumerate() {
                let id = (i + 1) as u32;
                assert_eq!(
                    sec.term(id).as_deref(),
                    Some(term.as_str()),
                    "term({id}) r={r}"
                );
                assert_eq!(sec.id(term), Some(id), "id({term}) r={r}");
            }
            // Near misses around run boundaries (and a sample of the rest):
            // an appended suffix or a truncation is never a stored term unless
            // it happens to collide with one.
            for (i, term) in expected.iter().enumerate() {
                let near_boundary = (i as u32) % r <= 1;
                if !near_boundary && i % 37 != 0 {
                    continue;
                }
                let longer = format!("{term}\u{1}");
                assert_eq!(sec.id(&longer), None, "near-miss long r={r}");
                let mut shorter = term.clone();
                shorter.pop();
                if !shorter.is_empty() && expected.binary_search(&shorter).is_err() {
                    assert_eq!(sec.id(&shorter), None, "near-miss short {shorter:?} r={r}");
                }
            }
            assert_eq!(sec.term(expected.len() as u32 + 1), None, "past-end r={r}");
        }
    }
}