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::trible::Fragment;
12use crate::trible::Trible;
13use crate::trible::TribleSet;
14
15use anybytes::Bytes;
16use anybytes::View;
17
18/// Canonical trible sequence stored as raw 64-byte entries.
19///
20/// The simplest portable archive format — a flat byte array of tribles
21/// in canonical EAV order with no compression. Used for commits,
22/// streaming, hashing, and audit trails where byte-for-byte stability
23/// matters.
24pub struct SimpleArchive;
25
26impl BlobEncoding for SimpleArchive {}
27
28impl MetaDescribe for SimpleArchive {
29    fn describe() -> Fragment {
30        let id: Id = id_hex!("8F4A27C8581DADCBA1ADA8BA228069B6");
31        entity! {
32            ExclusiveId::force_ref(&id) @
33                metadata::name: "simplearchive",
34                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.",
35                metadata::tag: metadata::KIND_BLOB_ENCODING,
36        }
37    }
38}
39
40impl Encodes<TribleSet> for SimpleArchive
41where crate::inline::encodings::hash::Handle<SimpleArchive>: crate::inline::InlineEncoding,
42{
43    type Output = Blob<SimpleArchive>;
44    fn encode(source: TribleSet) -> Blob<SimpleArchive> {
45        let mut tribles: Vec<[u8; 64]> = Vec::with_capacity(source.len());
46        tribles.extend(source.eav.iter_ordered());
47        let bytes: Bytes = tribles.into();
48        Blob::new(bytes)
49    }
50}
51
52impl Encodes<&TribleSet> for SimpleArchive
53where crate::inline::encodings::hash::Handle<SimpleArchive>: crate::inline::InlineEncoding,
54{
55    type Output = Blob<SimpleArchive>;
56    fn encode(source: &TribleSet) -> Blob<SimpleArchive> {
57        let mut tribles: Vec<[u8; 64]> = Vec::with_capacity(source.len());
58        tribles.extend(source.eav.iter_ordered());
59        let bytes: Bytes = tribles.into();
60        Blob::new(bytes)
61    }
62}
63
64/// Error returned when deserializing a [`SimpleArchive`] blob into a [`TribleSet`].
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum UnarchiveError {
67    /// The blob length is not a multiple of 64 bytes.
68    BadArchive,
69    /// A 64-byte entry has a nil entity or attribute.
70    BadTrible,
71    /// The archive contains duplicate tribles.
72    BadCanonicalizationRedundancy,
73    /// The tribles are not in ascending canonical order.
74    BadCanonicalizationOrdering,
75}
76
77impl std::fmt::Display for UnarchiveError {
78    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79        match self {
80            UnarchiveError::BadArchive => write!(f, "The archive is malformed or invalid."),
81            UnarchiveError::BadTrible => write!(f, "A trible in the archive is malformed."),
82            UnarchiveError::BadCanonicalizationRedundancy => {
83                write!(f, "The archive contains redundant tribles.")
84            }
85            UnarchiveError::BadCanonicalizationOrdering => {
86                write!(f, "The tribles in the archive are not in canonical order.")
87            }
88        }
89    }
90}
91
92impl std::error::Error for UnarchiveError {}
93
94/// Below this many tribles, serial unarchive wins (rayon overhead
95/// dominates).
96#[cfg(feature = "parallel")]
97const PARALLEL_UNARCHIVE_THRESHOLD: usize = 4096;
98
99impl TryFromBlob<SimpleArchive> for TribleSet {
100    type Error = UnarchiveError;
101
102    fn try_from_blob(blob: Blob<SimpleArchive>) -> Result<Self, Self::Error> {
103        let Ok(packed_tribles): Result<View<[[u8; 64]]>, _> = blob.bytes.clone().view() else {
104            return Err(UnarchiveError::BadArchive);
105        };
106        let slice: &[[u8; 64]] = &packed_tribles;
107
108        #[cfg(feature = "parallel")]
109        {
110            if slice.len() >= PARALLEL_UNARCHIVE_THRESHOLD {
111                return parallel_unarchive(slice);
112            }
113        }
114
115        serial_unarchive(slice)
116    }
117}
118
119/// Serial fallback. Validates ordering + redundancy inline with
120/// insertion — every byte read once.
121fn serial_unarchive(slice: &[[u8; 64]]) -> Result<TribleSet, UnarchiveError> {
122    let mut tribles = TribleSet::new();
123    let mut prev_trible: Option<&[u8; 64]> = None;
124    for t in slice.iter() {
125        let Some(trible) = Trible::as_transmute_force_raw(t) else {
126            return Err(UnarchiveError::BadTrible);
127        };
128        if let Some(prev) = prev_trible {
129            if prev == t {
130                return Err(UnarchiveError::BadCanonicalizationRedundancy);
131            }
132            if prev > t {
133                return Err(UnarchiveError::BadCanonicalizationOrdering);
134            }
135        }
136        prev_trible = Some(t);
137        tribles.insert(trible);
138    }
139    Ok(tribles)
140}
141
142/// Parallel unarchive: chunk the blob, validate internal ordering
143/// per chunk in parallel, build per-chunk `TribleSet`s, verify
144/// boundary ordering between adjacent chunks, then reduce via
145/// `TribleSet::union` (which itself fans out across the six
146/// indexes — three levels of parallelism stacked).
147#[cfg(feature = "parallel")]
148fn parallel_unarchive(slice: &[[u8; 64]]) -> Result<TribleSet, UnarchiveError> {
149    use rayon::prelude::*;
150
151    let n_threads = rayon::current_num_threads().max(1);
152    // Aim for ~1 chunk per worker so each thread gets a clean slice
153    // to crunch with maximal cache locality. Round up.
154    let chunk_size = slice.len().div_ceil(n_threads).max(1);
155    let chunks: Vec<&[[u8; 64]]> = slice.chunks(chunk_size).collect();
156
157    // Phase 1: validate boundary ordering (sequential, but it's a
158    // tiny O(num_chunks) scan over already-cache-hot slice ends).
159    for w in chunks.windows(2) {
160        let last_a = w[0].last().expect("non-empty chunk");
161        let first_b = w[1].first().expect("non-empty chunk");
162        if last_a == first_b {
163            return Err(UnarchiveError::BadCanonicalizationRedundancy);
164        }
165        if last_a > first_b {
166            return Err(UnarchiveError::BadCanonicalizationOrdering);
167        }
168    }
169
170    // Phase 2: per-chunk serial unarchive in parallel.
171    let chunk_sets: Result<Vec<TribleSet>, UnarchiveError> = chunks
172        .par_iter()
173        .map(|chunk| serial_unarchive(chunk))
174        .collect();
175
176    // Phase 3: reduce the per-chunk sets via TribleSet::union (the
177    // 6-way index fan-out kicks in for any chunk pair above its
178    // own threshold).
179    Ok(chunk_sets?
180        .into_par_iter()
181        .reduce(TribleSet::new, |a, b| a + b))
182}
triblespace_core/blob/encodings/simplearchive.rs

triblespace_core/blob/encodings/
simplearchive.rs
