Skip to main content

subetha_cxc/
shared_string_arena.rs

1//! `SharedStringArena` - append-only position-independent string
2//! pool backed by an MMF.
3//!
4//! # Why this exists
5//!
6//! Variable-length strings can't be stored inline in fixed-size
7//! slots (SharedHashMap, SharedVec, etc.) without padding waste or
8//! truncation. The natural cross-process solution is a shared byte
9//! arena: every process maps the same file at (potentially)
10//! different base addresses, and string references are
11//! position-independent `(offset, len)` pairs. Adding
12//! `mmap_base + offset` in any process resolves to the same bytes.
13//!
14//! # Layout
15//!
16//! ```text
17//! +---------------------------+
18//! | ArenaHeader (64B)         |
19//! |   magic, capacity_bytes   |
20//! |   used_bytes: AtomicU64   |
21//! +---------------------------+
22//! | bytes[0 .. capacity]      |
23//! +---------------------------+
24//! ```
25//!
26//! # Protocol
27//!
28//! `intern(s)`:
29//! 1. `offset = used_bytes.fetch_add(len)`.
30//! 2. If `offset + len > capacity`, rollback with
31//!    `fetch_sub(len)` and return `Full`. (Note: the
32//!    rollback is best-effort; if two threads race-overflow
33//!    simultaneously, both fetch_subs leave the counter
34//!    deterministic without "losing" bytes.)
35//! 3. Memcpy `s.bytes()` into `arena[offset..offset+len]`.
36//! 4. Return `StringRef { offset, len }`.
37//!
38//! `get(r)`:
39//! - Bounds-check `r.offset + r.len <= used_bytes` (sanity), then
40//!   return `&arena[r.offset..r.offset+r.len]` as a `&str`.
41//!
42//! # Concurrency
43//!
44//! Concurrent interners get distinct slices via fetch_add. Once
45//! the bytes are written, they are never moved (append-only). A
46//! reader holding a StringRef can always resolve it correctly,
47//! provided their `get` happens AFTER the interner returned the
48//! ref (which is the natural happens-before edge: the interner
49//! does the write, then makes the ref visible to the reader).
50//!
51//! # Deduplication
52//!
53//! Not provided here. For dedup, layer a `SharedHashMap<u64 hash,
54//! StringRef>` over the arena and consult it before each intern.
55//!
56//! # No deletion
57//!
58//! Append-only. The whole arena is reclaimed via `clear` (callers
59//! must ensure no concurrent readers); fine-grained deletion
60//! requires a free-list / compaction protocol that defeats the
61//! point of an arena.
62
63use std::fs::{File, OpenOptions};
64use std::mem::size_of;
65use std::path::Path;
66use std::sync::atomic::{AtomicU64, Ordering};
67
68use memmap2::{Mmap, MmapMut, MmapOptions};
69
70pub const ARENA_MAGIC: u64 = 0x4150_5341_524E_4131;
71
72/// How this process mapped the file. `MmapMut` demands a read+write
73/// file handle, which a consumer holding read access alone cannot get.
74enum Mapping {
75    Writable(MmapMut),
76    ReadOnly(Mmap),
77}
78
79impl Mapping {
80    #[inline]
81    fn as_ptr(&self) -> *const u8 {
82        match self {
83            Mapping::Writable(m) => m.as_ptr(),
84            Mapping::ReadOnly(m) => m.as_ptr(),
85        }
86    }
87
88    #[inline]
89    fn is_writable(&self) -> bool {
90        matches!(self, Mapping::Writable(_))
91    }
92
93    fn flush(&self) -> Result<(), std::io::Error> {
94        match self {
95            Mapping::Writable(m) => m.flush(),
96            Mapping::ReadOnly(_) => Ok(()),
97        }
98    }
99
100    fn flush_async(&self) -> Result<(), std::io::Error> {
101        match self {
102            Mapping::Writable(m) => m.flush_async(),
103            Mapping::ReadOnly(_) => Ok(()),
104        }
105    }
106}
107
108#[repr(C, align(64))]
109pub struct ArenaHeader {
110    pub magic: u64,
111    pub capacity_bytes: u64,
112    pub used_bytes: AtomicU64,
113    _pad: [u8; 40],
114}
115
116const _: () = {
117    assert!(size_of::<ArenaHeader>() == 64);
118};
119
120pub const fn arena_file_size(capacity_bytes: usize) -> usize {
121    size_of::<ArenaHeader>() + capacity_bytes
122}
123
124#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub enum ArenaError {
126    Full,
127    InvalidRef,
128    InvalidUtf8,
129    LayoutMismatch,
130    /// The arena was opened read-only and something tried to write it.
131    ReadOnly,
132    IoError(std::io::ErrorKind),
133}
134
135impl From<std::io::Error> for ArenaError {
136    fn from(e: std::io::Error) -> Self { Self::IoError(e.kind()) }
137}
138
139/// Position-independent reference to a string in a SharedStringArena.
140/// Encoded as a `u64` (offset:u32, len:u32) for stable cross-process
141/// passing (the same u64 resolves to the same bytes in every process
142/// that maps the arena).
143#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
144pub struct StringRef {
145    pub offset: u32,
146    pub len: u32,
147}
148
149impl StringRef {
150    #[inline]
151    pub fn to_u64(self) -> u64 {
152        ((self.offset as u64) << 32) | (self.len as u64)
153    }
154    #[inline]
155    pub fn from_u64(v: u64) -> Self {
156        Self {
157            offset: (v >> 32) as u32,
158            len: v as u32,
159        }
160    }
161}
162
163pub struct SharedStringArena {
164    _file: File,
165    mmap: Mapping,
166    capacity_bytes: usize,
167    header_sidecar: subetha_core::HandshakeHeader,
168    ring_sidecar: Box<subetha_core::ObservationRing>,
169}
170
171unsafe impl Send for SharedStringArena {}
172unsafe impl Sync for SharedStringArena {}
173
174impl subetha_sidecar::AdaptiveInstance for SharedStringArena {
175    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
176    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
177    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
178        Box::new(subetha_sidecar::NoMigrationPolicy)
179    }
180}
181
182impl SharedStringArena {
183    /// Obtain the arena at `path`, initializing an empty one if the
184    /// path does not yet exist and attaching to it if it does.
185    /// Attaching leaves interned strings and `used_bytes` in place, so
186    /// outstanding [`StringRef`]s stay resolvable; a region built with
187    /// a different capacity is a `LayoutMismatch`.
188    /// [`reset`](Self::reset) reinitializes.
189    pub fn create(
190        path: impl AsRef<Path>, capacity_bytes: usize,
191    ) -> Result<Self, ArenaError> {
192        Self::check_capacity(capacity_bytes)?;
193        let (file, mmap) = crate::mmf_attach::create_or_attach(
194            path.as_ref(),
195            arena_file_size(capacity_bytes),
196            |ptr| unsafe { Self::init_region(ptr, capacity_bytes) },
197            |ptr| unsafe { (*(ptr as *const ArenaHeader)).magic == ARENA_MAGIC },
198        )?;
199        let this = Self {
200            _file: file, mmap: Mapping::Writable(mmap), capacity_bytes,
201            header_sidecar: subetha_core::HandshakeHeader::new(),
202            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
203        };
204        this.validate(capacity_bytes)?;
205        Ok(this)
206    }
207
208    /// Truncate the arena at `path` and initialize an empty one,
209    /// invalidating every StringRef live peers hold. For a caller that
210    /// knows it owns the path.
211    pub fn reset(
212        path: impl AsRef<Path>, capacity_bytes: usize,
213    ) -> Result<Self, ArenaError> {
214        Self::check_capacity(capacity_bytes)?;
215        let (file, mmap) = crate::mmf_attach::reset(
216            path.as_ref(),
217            arena_file_size(capacity_bytes),
218            |ptr| unsafe { Self::init_region(ptr, capacity_bytes) },
219        )?;
220        Ok(Self {
221            _file: file, mmap: Mapping::Writable(mmap), capacity_bytes,
222            header_sidecar: subetha_core::HandshakeHeader::new(),
223            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
224        })
225    }
226
227    /// A capacity every constructor must agree on: at least one byte,
228    /// and no larger than a [`StringRef`] offset can address. Reported
229    /// rather than asserted, because a service that panics building its
230    /// arena dies with it.
231    fn check_capacity(capacity_bytes: usize) -> Result<(), ArenaError> {
232        if capacity_bytes < 1 || capacity_bytes > u32::MAX as usize {
233            return Err(ArenaError::LayoutMismatch);
234        }
235        Ok(())
236    }
237
238    /// Lay out an empty arena: capacity first, magic last, because
239    /// attachers spin on it. The zeroed region is already `used_bytes`
240    /// 0 and empty byte space.
241    ///
242    /// # Safety
243    /// `ptr` addresses at least `arena_file_size(capacity_bytes)`
244    /// writable zeroed bytes.
245    unsafe fn init_region(ptr: *mut u8, capacity_bytes: usize) {
246        let hdr = ptr as *mut ArenaHeader;
247        unsafe {
248            (*hdr).capacity_bytes = capacity_bytes as u64;
249            std::ptr::write_volatile(&raw mut (*hdr).magic, ARENA_MAGIC);
250        }
251    }
252
253    pub fn open(
254        path: impl AsRef<Path>, expected_capacity_bytes: usize,
255    ) -> Result<Self, ArenaError> {
256        // A capacity a StringRef offset cannot address is not a layout
257        // this crate ever creates, and interning into it would truncate.
258        if expected_capacity_bytes > u32::MAX as usize {
259            return Err(ArenaError::LayoutMismatch);
260        }
261        let total = arena_file_size(expected_capacity_bytes);
262        let file = OpenOptions::new().read(true).write(true).open(path.as_ref())?;
263        if file.metadata()?.len() < total as u64 {
264            return Err(ArenaError::LayoutMismatch);
265        }
266        let mmap = unsafe { MmapOptions::new().len(total).map_mut(&file)? };
267        let this = Self {
268            _file: file, mmap: Mapping::Writable(mmap),
269            capacity_bytes: expected_capacity_bytes,
270            header_sidecar: subetha_core::HandshakeHeader::new(),
271            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
272        };
273        this.validate(expected_capacity_bytes)?;
274        Ok(this)
275    }
276
277    /// Open an arena this process may only read.
278    ///
279    /// [`open`](Self::open) needs a read+write file handle, which a
280    /// consumer of a privileged producer's arena does not have. Reads
281    /// behave identically; [`intern`](Self::intern) and friends return
282    /// [`ArenaError::ReadOnly`].
283    pub fn open_read_only(
284        path: impl AsRef<Path>, expected_capacity_bytes: usize,
285    ) -> Result<Self, ArenaError> {
286        let total = arena_file_size(expected_capacity_bytes);
287        let file = OpenOptions::new().read(true).open(path.as_ref())?;
288        if file.metadata()?.len() < total as u64 {
289            return Err(ArenaError::LayoutMismatch);
290        }
291        let mmap = unsafe { MmapOptions::new().len(total).map(&file)? };
292        let this = Self {
293            _file: file, mmap: Mapping::ReadOnly(mmap),
294            capacity_bytes: expected_capacity_bytes,
295            header_sidecar: subetha_core::HandshakeHeader::new(),
296            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
297        };
298        this.validate(expected_capacity_bytes)?;
299        Ok(this)
300    }
301
302    /// Whether the header on disk is the one this mapping expects.
303    fn validate(&self, expected_capacity_bytes: usize) -> Result<(), ArenaError> {
304        let hdr = self.header();
305        if hdr.magic != ARENA_MAGIC || hdr.capacity_bytes != expected_capacity_bytes as u64 {
306            return Err(ArenaError::LayoutMismatch);
307        }
308        Ok(())
309    }
310
311    /// Whether this mapping may be written.
312    #[inline]
313    pub fn is_writable(&self) -> bool {
314        self.mmap.is_writable()
315    }
316
317    #[inline]
318    pub fn capacity_bytes(&self) -> usize { self.capacity_bytes }
319
320    #[inline]
321    pub fn used_bytes(&self) -> usize {
322        self.header().used_bytes.load(Ordering::Acquire) as usize
323    }
324
325    #[inline]
326    pub fn remaining_bytes(&self) -> usize {
327        self.capacity_bytes.saturating_sub(self.used_bytes())
328    }
329
330    fn header(&self) -> &ArenaHeader {
331        unsafe { &*(self.mmap.as_ptr() as *const ArenaHeader) }
332    }
333
334    /// Append a string to the arena. Returns a StringRef that
335    /// resolves to the bytes in any mapping of the same file.
336    ///
337    /// Returns `Err(Full)` when the arena has no room. The empty
338    /// string `""` interns at the current offset with `len = 0`.
339    pub fn intern(&self, s: &str) -> Result<StringRef, ArenaError> {
340        self.intern_bytes(s.as_bytes())
341    }
342
343    /// Append arbitrary bytes (not necessarily UTF-8) to the arena.
344    /// Useful for storing binary blobs alongside strings. Retrieve
345    /// with `get_bytes`; `get` will reject non-UTF-8 with
346    /// `InvalidUtf8`.
347    pub fn intern_bytes(&self, bytes: &[u8]) -> Result<StringRef, ArenaError> {
348        if !self.mmap.is_writable() {
349            return Err(ArenaError::ReadOnly);
350        }
351        let len = bytes.len() as u64;
352        if len > self.capacity_bytes as u64 {
353            self.ring_sidecar
354                .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
355            return Err(ArenaError::Full);
356        }
357        let offset = self.header().used_bytes.fetch_add(len, Ordering::AcqRel);
358        if offset.saturating_add(len) > self.capacity_bytes as u64 {
359            self.header().used_bytes.fetch_sub(len, Ordering::AcqRel);
360            self.ring_sidecar
361                .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
362            return Err(ArenaError::Full);
363        }
364        // A StringRef carries the offset as a u32, so an offset past that
365        // is refused here rather than truncated: a truncated offset lands
366        // inside the used region and resolves to another string's bytes,
367        // which every downstream bounds check would accept. `create` and
368        // `reset` refuse such a capacity, so this covers an arena opened
369        // at a capacity they never sanctioned.
370        if offset.saturating_add(len) > u32::MAX as u64 {
371            self.header().used_bytes.fetch_sub(len, Ordering::AcqRel);
372            self.ring_sidecar
373                .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 1);
374            return Err(ArenaError::Full);
375        }
376        let dst = unsafe {
377            self.mmap.as_ptr()
378                .add(size_of::<ArenaHeader>())
379                .add(offset as usize)
380                as *mut u8
381        };
382        unsafe {
383            std::ptr::copy_nonoverlapping(bytes.as_ptr(), dst, bytes.len());
384        }
385        self.ring_sidecar
386            .push_op(crate::sidecar_ops::string_arena::OP_INTERN, 0);
387        Ok(StringRef { offset: offset as u32, len: len as u32 })
388    }
389
390    /// Resolve a StringRef to its `&[u8]`. Returns `Err(InvalidRef)`
391    /// when the ref doesn't fall inside the arena's used region.
392    pub fn get_bytes(&self, r: StringRef) -> Result<&[u8], ArenaError> {
393        let end = (r.offset as u64).saturating_add(r.len as u64);
394        if end > self.header().used_bytes.load(Ordering::Acquire) {
395            self.ring_sidecar
396                .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 1);
397            return Err(ArenaError::InvalidRef);
398        }
399        if end > self.capacity_bytes as u64 {
400            self.ring_sidecar
401                .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 1);
402            return Err(ArenaError::InvalidRef);
403        }
404        self.ring_sidecar
405            .push_op(crate::sidecar_ops::string_arena::OP_GET_BYTES, 0);
406        let base = unsafe {
407            self.mmap.as_ptr()
408                .add(size_of::<ArenaHeader>())
409                .add(r.offset as usize)
410        };
411        Ok(unsafe { std::slice::from_raw_parts(base, r.len as usize) })
412    }
413
414    /// Resolve a StringRef to a `&str`. Returns `Err(InvalidUtf8)`
415    /// when the bytes aren't valid UTF-8 (the arena doesn't enforce
416    /// validity per-segment; it's checked on read).
417    pub fn get(&self, r: StringRef) -> Result<&str, ArenaError> {
418        let bytes = self.get_bytes(r)?;
419        std::str::from_utf8(bytes).map_err(|_| ArenaError::InvalidUtf8)
420    }
421
422    /// Convenience: intern AND return a `&str` view into the
423    /// just-written bytes plus the ref.
424    pub fn intern_and_get(&self, s: &str) -> Result<(StringRef, &str), ArenaError> {
425        let r = self.intern(s)?;
426        let got = self.get(r)?;
427        Ok((r, got))
428    }
429
430    /// Reset the arena to empty. NOT concurrency-safe; callers must
431    /// ensure no other threads/processes are interning or reading.
432    /// Existing StringRefs become invalid (their bytes may be
433    /// overwritten by subsequent interns).
434    pub fn clear(&self) {
435        if !self.mmap.is_writable() {
436            return;
437        }
438        self.header().used_bytes.store(0, Ordering::Release);
439        self.ring_sidecar
440            .push_op(crate::sidecar_ops::string_arena::OP_CLEAR, 0);
441    }
442
443    pub fn flush(&self) -> Result<(), ArenaError> {
444        self.mmap.flush()?;
445        Ok(())
446    }
447
448    /// Non-blocking flush: schedules a writeback via the OS.
449    /// Note: Windows is only partially async (sync to page cache,
450    /// not to disk).
451    pub fn flush_async(&self) -> Result<(), ArenaError> {
452        self.mmap.flush_async()?;
453        Ok(())
454    }
455}
456
457#[cfg(test)]
458mod tests {
459    use super::*;
460    use std::sync::Arc;
461    use std::thread;
462
463    fn tmp(name: &str) -> std::path::PathBuf {
464        let mut p = std::env::temp_dir();
465        let pid = std::process::id();
466        p.push(format!("subetha-arena-{name}-{pid}.bin"));
467        p
468    }
469
470    /// A second create attaches with interned strings in place; reset
471    /// is what strips them.
472    #[test]
473    fn second_create_attaches_and_keeps_strings() {
474        let p = tmp("attach");
475        std::fs::remove_file(&p).ok();
476        let a = SharedStringArena::create(&p, 4096).unwrap();
477        let r = a.intern("held").unwrap();
478
479        let a2 = SharedStringArena::create(&p, 4096).unwrap();
480        assert_eq!(a2.get(r).unwrap(), "held", "attach lost an interned string");
481        assert!(matches!(
482            SharedStringArena::create(&p, 2048),
483            Err(ArenaError::LayoutMismatch),
484        ));
485
486        // Windows refuses to truncate a mapped file, so every handle goes
487        // before the reset.
488        drop(a);
489        drop(a2);
490        let fresh = SharedStringArena::reset(&p, 4096).unwrap();
491        assert_eq!(fresh.used_bytes(), 0, "reset kept interned bytes");
492        drop(fresh);
493        std::fs::remove_file(&p).ok();
494    }
495
496    #[test]
497    fn a_read_only_arena_resolves_refs_and_refuses_interning() {
498        let p = tmp("readonly");
499        let r = {
500            let w = SharedStringArena::create(&p, 4096).unwrap();
501            let r = w.intern("notepad.exe").unwrap();
502            w.flush().unwrap();
503            r
504        };
505        let ro = SharedStringArena::open_read_only(&p, 4096).unwrap();
506        assert!(!ro.is_writable());
507        assert_eq!(ro.get(r), Ok("notepad.exe"));
508        assert_eq!(ro.get_bytes(r), Ok(&b"notepad.exe"[..]));
509        assert_eq!(ro.used_bytes(), 11);
510        assert_eq!(ro.intern("more"), Err(ArenaError::ReadOnly));
511        assert_eq!(ro.intern_bytes(b"more"), Err(ArenaError::ReadOnly));
512        ro.clear();
513        assert_eq!(ro.used_bytes(), 11, "clear on a read-only arena is inert");
514        ro.flush().unwrap();
515        std::fs::remove_file(&p).ok();
516    }
517
518    #[test]
519    fn a_read_only_open_still_validates_the_header() {
520        let p = tmp("readonly-mismatch");
521        {
522            let w = SharedStringArena::create(&p, 4096).unwrap();
523            w.intern("x").unwrap();
524            w.flush().unwrap();
525        }
526        assert_eq!(
527            SharedStringArena::open_read_only(&p, 2048).err(),
528            Some(ArenaError::LayoutMismatch)
529        );
530        std::fs::remove_file(&p).ok();
531    }
532
533    #[test]
534    fn create_initial_state_is_empty() {
535        let p = tmp("init");
536        let a = SharedStringArena::create(&p, 1024).unwrap();
537        assert_eq!(a.capacity_bytes(), 1024);
538        assert_eq!(a.used_bytes(), 0);
539        assert_eq!(a.remaining_bytes(), 1024);
540        std::fs::remove_file(&p).ok();
541    }
542
543    #[test]
544    fn intern_and_get_round_trip() {
545        let p = tmp("rt");
546        let a = SharedStringArena::create(&p, 1024).unwrap();
547        let r1 = a.intern("hello").unwrap();
548        let r2 = a.intern("world").unwrap();
549        assert_eq!(a.get(r1).unwrap(), "hello");
550        assert_eq!(a.get(r2).unwrap(), "world");
551        assert_eq!(a.used_bytes(), 10);
552        std::fs::remove_file(&p).ok();
553    }
554
555    #[test]
556    fn empty_string_interns_with_zero_len() {
557        let p = tmp("empty");
558        let a = SharedStringArena::create(&p, 16).unwrap();
559        let r = a.intern("").unwrap();
560        assert_eq!(r.len, 0);
561        assert_eq!(a.get(r).unwrap(), "");
562        assert_eq!(a.used_bytes(), 0);
563        std::fs::remove_file(&p).ok();
564    }
565
566    #[test]
567    fn full_arena_returns_error() {
568        let p = tmp("full");
569        let a = SharedStringArena::create(&p, 10).unwrap();
570        a.intern("hello").unwrap();
571        a.intern("world").unwrap();
572        assert_eq!(a.intern("more").err(), Some(ArenaError::Full));
573        // Used bytes rolled back, not 14.
574        assert_eq!(a.used_bytes(), 10);
575        std::fs::remove_file(&p).ok();
576    }
577
578    #[test]
579    fn string_too_large_returns_full() {
580        let p = tmp("too-large");
581        let a = SharedStringArena::create(&p, 8).unwrap();
582        let big = "x".repeat(100);
583        assert_eq!(a.intern(&big).err(), Some(ArenaError::Full));
584        assert_eq!(a.used_bytes(), 0);
585        std::fs::remove_file(&p).ok();
586    }
587
588    #[test]
589    fn string_ref_packs_and_unpacks() {
590        let r = StringRef { offset: 0x1234_5678, len: 42 };
591        let packed = r.to_u64();
592        let unpacked = StringRef::from_u64(packed);
593        assert_eq!(unpacked, r);
594    }
595
596    #[test]
597    fn cross_handle_visibility() {
598        let p = tmp("cross-handle");
599        let writer = SharedStringArena::create(&p, 1024).unwrap();
600        let reader = SharedStringArena::open(&p, 1024).unwrap();
601        let r = writer.intern("cross-process").unwrap();
602        assert_eq!(reader.get(r).unwrap(), "cross-process");
603        std::fs::remove_file(&p).ok();
604    }
605
606    #[test]
607    fn invalid_ref_beyond_used_rejected() {
608        let p = tmp("invalid");
609        let a = SharedStringArena::create(&p, 1024).unwrap();
610        a.intern("hi").unwrap();  // used = 2
611        let bad = StringRef { offset: 100, len: 5 };
612        assert_eq!(a.get(bad).err(), Some(ArenaError::InvalidRef));
613        std::fs::remove_file(&p).ok();
614    }
615
616    #[test]
617    fn concurrent_interners_get_distinct_refs() {
618        let p = tmp("concurrent");
619        let a: Arc<SharedStringArena> = Arc::new(SharedStringArena::create(&p, 4096).unwrap());
620        let n_threads = 4;
621        let per_thread = 20;
622        let mut handles = vec![];
623        for t in 0..n_threads {
624            let a = a.clone();
625            handles.push(thread::spawn(move || {
626                let mut refs = vec![];
627                for i in 0..per_thread {
628                    let s = format!("thread-{t}-msg-{i:03}");
629                    let r = a.intern(&s).unwrap();
630                    refs.push((s, r));
631                }
632                refs
633            }));
634        }
635        let all: Vec<(String, StringRef)> = handles.into_iter()
636            .flat_map(|h| h.join().unwrap())
637            .collect();
638        // Every interned string must read back to its original value.
639        for (expected, r) in &all {
640            let got = a.get(*r).unwrap();
641            assert_eq!(got, expected,
642                "ref offset={} len={} should resolve to {expected}",
643                r.offset, r.len);
644        }
645        // No two refs overlap.
646        let mut refs: Vec<StringRef> = all.iter().map(|(_, r)| *r).collect();
647        refs.sort_by_key(|r| r.offset);
648        for w in refs.windows(2) {
649            let r1_end = w[0].offset + w[0].len;
650            assert!(r1_end <= w[1].offset,
651                "ref {:?} overlaps with ref {:?}", w[0], w[1]);
652        }
653        std::fs::remove_file(&p).ok();
654    }
655
656    #[test]
657    fn intern_and_get_helper_returns_both() {
658        let p = tmp("intern-and-get");
659        let a = SharedStringArena::create(&p, 1024).unwrap();
660        let (r, s) = a.intern_and_get("composite").unwrap();
661        assert_eq!(s, "composite");
662        assert_eq!(a.get(r).unwrap(), "composite");
663        std::fs::remove_file(&p).ok();
664    }
665
666    #[test]
667    fn utf8_validation_on_get() {
668        let p = tmp("utf8");
669        let a = SharedStringArena::create(&p, 128).unwrap();
670        // Intern valid UTF-8.
671        let r = a.intern("hello").unwrap();
672        assert!(a.get(r).is_ok());
673        // intern_bytes accepts arbitrary bytes; get() then rejects
674        // non-UTF-8 with InvalidUtf8 while get_bytes returns the raw
675        // bytes without validation.
676        let r2 = a.intern_bytes(&[0xFF, 0xFE, 0xFD]).unwrap();
677        assert_eq!(a.get(r2).err(), Some(ArenaError::InvalidUtf8));
678        assert_eq!(a.get_bytes(r2).unwrap(), &[0xFF, 0xFE, 0xFD]);
679        std::fs::remove_file(&p).ok();
680    }
681
682    #[test]
683    fn clear_resets_used_bytes() {
684        let p = tmp("clear");
685        let a = SharedStringArena::create(&p, 128).unwrap();
686        a.intern("first").unwrap();
687        a.intern("second").unwrap();
688        assert!(a.used_bytes() > 0);
689        a.clear();
690        assert_eq!(a.used_bytes(), 0);
691        // Fresh interns work.
692        let r = a.intern("after-clear").unwrap();
693        assert_eq!(a.get(r).unwrap(), "after-clear");
694        assert_eq!(r.offset, 0);
695        std::fs::remove_file(&p).ok();
696    }
697
698    #[test]
699    fn disk_persistence_survives_reopen() {
700        let p = tmp("disk");
701        let r_persist;
702        {
703            let a = SharedStringArena::create(&p, 1024).unwrap();
704            r_persist = a.intern("persisted-string").unwrap();
705            a.flush().unwrap();
706        }
707        let a2 = SharedStringArena::open(&p, 1024).unwrap();
708        assert_eq!(a2.get(r_persist).unwrap(), "persisted-string");
709        // And it can keep interning.
710        let r2 = a2.intern("more-after-reopen").unwrap();
711        assert_eq!(a2.get(r2).unwrap(), "more-after-reopen");
712        std::fs::remove_file(&p).ok();
713    }
714
715    /// A StringRef addresses its bytes with a u32 offset, so a capacity
716    /// past that is refused at construction rather than silently
717    /// truncating every ref past the 4 GiB mark into another string's
718    /// bytes. No file is touched: the check precedes the mapping.
719    #[test]
720    fn create_refuses_a_capacity_a_ref_cannot_address() {
721        assert!(matches!(
722            SharedStringArena::create(tmp("too-big"), u32::MAX as usize + 1),
723            Err(ArenaError::LayoutMismatch),
724        ));
725        assert!(matches!(
726            SharedStringArena::reset(tmp("too-big-reset"), u32::MAX as usize + 1),
727            Err(ArenaError::LayoutMismatch),
728        ));
729        assert!(matches!(
730            SharedStringArena::create(tmp("zero-cap"), 0),
731            Err(ArenaError::LayoutMismatch),
732        ));
733    }
734
735    /// The same layout refused on the open path, where no assertion
736    /// guards construction.
737    #[test]
738    fn open_refuses_a_capacity_a_ref_cannot_address() {
739        assert!(matches!(
740            SharedStringArena::open(tmp("open-too-big"), u32::MAX as usize + 1),
741            Err(ArenaError::LayoutMismatch),
742        ));
743    }
744
745    #[test]
746    fn deduplication_via_hashmap_composition() {
747        // Demonstrate the dedup pattern: layer SharedHashMap<u64, u64>
748        // (hash -> StringRef.to_u64) over the arena.
749        use crate::SharedHashMap;
750        use crate::shared_hash_map::fnv1a_64;
751
752        let p_arena = tmp("dedup-arena");
753        let p_index = tmp("dedup-index");
754        let arena = SharedStringArena::create(&p_arena, 256).unwrap();
755        let index: SharedHashMap<u64, u64> = SharedHashMap::create(&p_index, 32).unwrap();
756
757        let s = "deduplicate-me";
758        let h = fnv1a_64(s.as_bytes());
759
760        // First intern: check index, miss, intern + insert into index.
761        let r = if let Some(packed) = index.get(&h) {
762            StringRef::from_u64(packed)
763        } else {
764            let r = arena.intern(s).unwrap();
765            index.insert(h, r.to_u64()).unwrap();
766            r
767        };
768        let used_after_first = arena.used_bytes();
769
770        // Second intern of same string: hit in index, no arena append.
771        let r2 = if let Some(packed) = index.get(&h) {
772            StringRef::from_u64(packed)
773        } else {
774            let r = arena.intern(s).unwrap();
775            index.insert(h, r.to_u64()).unwrap();
776            r
777        };
778        assert_eq!(r, r2, "dedup should return the same ref");
779        assert_eq!(arena.used_bytes(), used_after_first,
780            "second intern should not consume more bytes");
781
782        std::fs::remove_file(&p_arena).ok();
783        std::fs::remove_file(&p_index).ok();
784    }
785}