triblespace_core/blob/encodings/
simplearchive.rs

1use crate::inline::Encodes;
2use crate::blob::Blob;
3use crate::blob::BlobEncoding;
4use crate::blob::TryFromBlob;
5use crate::id::ExclusiveId;
6use crate::id::Id;
7use crate::id_hex;
8use crate::macros::entity;
9use crate::metadata;
10use crate::metadata::MetaDescribe;
11use crate::patch::ArchiveEntry;
12use crate::patch::ArchiveOwner;
13use crate::trible::Fragment;
14use crate::trible::Trible;
15use crate::trible::TribleSet;
16
17use anybytes::Bytes;
18use anybytes::View;
19use std::ptr::NonNull;
20use std::sync::Arc;
21
22/// Canonical trible sequence stored as raw 64-byte entries.
23///
24/// The simplest portable archive format — a flat byte array of tribles
25/// in canonical EAV order with no compression. Used for commits,
26/// streaming, hashing, and audit trails where byte-for-byte stability
27/// matters.
28pub struct SimpleArchive;
29
30impl BlobEncoding for SimpleArchive {}
31
32impl MetaDescribe for SimpleArchive {
33    fn describe() -> Fragment {
34        let id: Id = id_hex!("8F4A27C8581DADCBA1ADA8BA228069B6");
35        entity! {
36            ExclusiveId::force_ref(&id) @
37                metadata::name: "simplearchive",
38                metadata::description: "Canonical trible sequence stored as raw 64-byte entries. This is the simplest portable archive format and preserves the exact trible ordering expected by the canonicalization rules.\n\nUse SimpleArchive for export, import, streaming, hashing, or audit trails where you want a byte-for-byte stable representation. Prefer SuccinctArchiveBlob when you need compact indexed storage and fast offline queries, and keep a SimpleArchive around if you want a source of truth that can be re-indexed or validated.",
39                metadata::tag: metadata::KIND_BLOB_ENCODING,
40        }
41    }
42}
43
44impl Encodes<TribleSet> for SimpleArchive
45where crate::inline::encodings::hash::Handle<SimpleArchive>: crate::inline::InlineEncoding,
46{
47    type Output = Blob<SimpleArchive>;
48    fn encode(source: TribleSet) -> Blob<SimpleArchive> {
49        let mut tribles: Vec<[u8; 64]> = Vec::with_capacity(source.len());
50        tribles.extend(source.eav.iter_ordered());
51        let bytes: Bytes = tribles.into();
52        Blob::new(bytes)
53    }
54}
55
56impl Encodes<&TribleSet> for SimpleArchive
57where crate::inline::encodings::hash::Handle<SimpleArchive>: crate::inline::InlineEncoding,
58{
59    type Output = Blob<SimpleArchive>;
60    fn encode(source: &TribleSet) -> Blob<SimpleArchive> {
61        let mut tribles: Vec<[u8; 64]> = Vec::with_capacity(source.len());
62        tribles.extend(source.eav.iter_ordered());
63        let bytes: Bytes = tribles.into();
64        Blob::new(bytes)
65    }
66}
67
68/// Error returned when deserializing a [`SimpleArchive`] blob into a [`TribleSet`].
69#[derive(Debug, Clone, Copy, PartialEq, Eq)]
70pub enum UnarchiveError {
71    /// The blob length is not a multiple of 64 bytes.
72    BadArchive,
73    /// A 64-byte entry has a nil entity or attribute.
74    BadTrible,
75    /// The archive contains duplicate tribles.
76    BadCanonicalizationRedundancy,
77    /// The tribles are not in ascending canonical order.
78    BadCanonicalizationOrdering,
79}
80
81impl std::fmt::Display for UnarchiveError {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        match self {
84            UnarchiveError::BadArchive => write!(f, "The archive is malformed or invalid."),
85            UnarchiveError::BadTrible => write!(f, "A trible in the archive is malformed."),
86            UnarchiveError::BadCanonicalizationRedundancy => {
87                write!(f, "The archive contains redundant tribles.")
88            }
89            UnarchiveError::BadCanonicalizationOrdering => {
90                write!(f, "The tribles in the archive are not in canonical order.")
91            }
92        }
93    }
94}
95
96impl std::error::Error for UnarchiveError {}
97
98/// Below this many tribles, serial unarchive wins (rayon overhead
99/// dominates).
100#[cfg(feature = "parallel")]
101const PARALLEL_UNARCHIVE_THRESHOLD: usize = 4096;
102
103impl TryFromBlob<SimpleArchive> for TribleSet {
104    type Error = UnarchiveError;
105
106    fn try_from_blob(blob: Blob<SimpleArchive>) -> Result<Self, Self::Error> {
107        try_from_blob_inner(blob, /*archive_backed:*/ true)
108    }
109}
110
111/// Decode a [`SimpleArchive`] blob into a [`TribleSet`] forcing the
112/// heap-`Leaf` ingest path (no `LocalLeaf`). Exposed for measurement
113/// so the LocalLeaf path can be compared against the legacy heap
114/// behaviour on identical input.
115pub fn try_from_blob_heap_only(
116    blob: Blob<SimpleArchive>,
117) -> Result<TribleSet, UnarchiveError> {
118    try_from_blob_inner(blob, /*archive_backed:*/ false)
119}
120
121fn try_from_blob_inner(
122    blob: Blob<SimpleArchive>,
123    archive_backed: bool,
124) -> Result<TribleSet, UnarchiveError> {
125    let Ok(packed_tribles): Result<View<[[u8; 64]]>, _> = blob.bytes.clone().view() else {
126        return Err(UnarchiveError::BadArchive);
127    };
128    let slice: &[[u8; 64]] = &packed_tribles;
129
130    // ArchiveEntry / LocalLeaf require the trible pointer to be
131    // 16-byte aligned (the low 4 bits encode `HeadTag::LocalLeaf`).
132    // Every 64-byte stride preserves alignment, so it's enough to
133    // check the slice base. Modern allocators (and mmap'd files)
134    // satisfy this; the heap-Leaf fallback handles the rare miss.
135    let owner: Option<Arc<dyn ArchiveOwner>> =
136        if archive_backed && (slice.as_ptr() as usize) & 0x0f == 0 {
137            Some(Arc::new(blob.bytes.clone()))
138        } else {
139            None
140        };
141
142    #[cfg(feature = "parallel")]
143    {
144        if slice.len() >= PARALLEL_UNARCHIVE_THRESHOLD {
145            return parallel_unarchive(slice, owner);
146        }
147    }
148
149    serial_unarchive(slice, owner.as_ref())
150}
151
152/// Serial fallback. Validates ordering + redundancy inline with
153/// insertion — every byte read once. When `owner` is `Some`, each
154/// trible is inserted as an `ArchiveEntry` (LocalLeaf-backed); when
155/// `None`, the heap-Leaf path is taken.
156fn serial_unarchive(
157    slice: &[[u8; 64]],
158    owner: Option<&Arc<dyn ArchiveOwner>>,
159) -> Result<TribleSet, UnarchiveError> {
160    let mut tribles = TribleSet::new();
161    let mut prev_trible: Option<&[u8; 64]> = None;
162    for t in slice.iter() {
163        let Some(trible) = Trible::as_transmute_force_raw(t) else {
164            return Err(UnarchiveError::BadTrible);
165        };
166        if let Some(prev) = prev_trible {
167            if prev == t {
168                return Err(UnarchiveError::BadCanonicalizationRedundancy);
169            }
170            if prev > t {
171                return Err(UnarchiveError::BadCanonicalizationOrdering);
172            }
173        }
174        prev_trible = Some(t);
175        match owner {
176            Some(owner_arc) => {
177                // SAFETY: `t` points into the archive bytes kept alive
178                // by `owner_arc`, and base-alignment + 64-byte stride
179                // guarantees this element is 16-byte aligned.
180                let ptr = NonNull::from(t);
181                let entry = unsafe { ArchiveEntry::new(ptr, owner_arc) };
182                tribles.insert_archive(&entry);
183            }
184            None => tribles.insert(trible),
185        }
186    }
187    Ok(tribles)
188}
189
190/// Parallel unarchive: chunk the blob, validate internal ordering
191/// per chunk in parallel, build per-chunk `TribleSet`s, verify
192/// boundary ordering between adjacent chunks, then reduce via
193/// `TribleSet::union` (which itself fans out across the six
194/// indexes — three levels of parallelism stacked).
195#[cfg(feature = "parallel")]
196fn parallel_unarchive(
197    slice: &[[u8; 64]],
198    owner: Option<Arc<dyn ArchiveOwner>>,
199) -> Result<TribleSet, UnarchiveError> {
200    use rayon::prelude::*;
201
202    let n_threads = rayon::current_num_threads().max(1);
203    // Aim for ~1 chunk per worker so each thread gets a clean slice
204    // to crunch with maximal cache locality. Round up.
205    let chunk_size = slice.len().div_ceil(n_threads).max(1);
206    let chunks: Vec<&[[u8; 64]]> = slice.chunks(chunk_size).collect();
207
208    // Phase 1: validate boundary ordering (sequential, but it's a
209    // tiny O(num_chunks) scan over already-cache-hot slice ends).
210    for w in chunks.windows(2) {
211        let last_a = w[0].last().expect("non-empty chunk");
212        let first_b = w[1].first().expect("non-empty chunk");
213        if last_a == first_b {
214            return Err(UnarchiveError::BadCanonicalizationRedundancy);
215        }
216        if last_a > first_b {
217            return Err(UnarchiveError::BadCanonicalizationOrdering);
218        }
219    }
220
221    // Phase 2: per-chunk serial unarchive in parallel. Every chunk
222    // shares the same archive owner, so `union` later sees identical
223    // owner Arcs and can adopt LocalLeaves wholesale.
224    let chunk_sets: Result<Vec<TribleSet>, UnarchiveError> = chunks
225        .par_iter()
226        .map(|chunk| serial_unarchive(chunk, owner.as_ref()))
227        .collect();
228
229    // Phase 3: reduce the per-chunk sets via TribleSet::union (the
230    // 6-way index fan-out kicks in for any chunk pair above its
231    // own threshold).
232    Ok(chunk_sets?
233        .into_par_iter()
234        .reduce(TribleSet::new, |a, b| a + b))
235}
triblespace_core/blob/encodings/simplearchive.rs

triblespace_core/blob/encodings/
simplearchive.rs
