Skip to main content

prolly/prolly/secondary_index/
storage.rs

1use serde::de::{DeserializeOwned, Error as DeError, Visitor};
2use serde::{Deserialize, Deserializer, Serialize, Serializer};
3use std::fmt;
4
5use super::super::cid::Cid;
6use super::super::error::Error;
7use super::super::key::{
8    decode_segments, encode_segment, encode_segment_prefix, prefix_end, KeyBuilder,
9};
10use super::super::versioned_map::{MapVersionId, VERSIONED_MAP_ROOT_PREFIX};
11use super::definition::{IndexProjection, SecondaryIndex};
12
13pub const SECONDARY_INDEX_FORMAT_VERSION: u32 = 1;
14pub const INDEX_PHYSICAL_LAYOUT_VERSION: u32 = 1;
15
16const DESCRIPTOR_MAGIC: &[u8; 4] = b"PSID";
17const DESCRIPTOR_FINGERPRINT_MAGIC: &[u8; 4] = b"PSIF";
18const CHECKPOINT_MAGIC: &[u8; 4] = b"PSIP";
19const HEAD_MAGIC: &[u8; 4] = b"PSIH";
20const CONTROL_MAGIC: &[u8; 4] = b"PSIO";
21const INDEX_VALUE_MAGIC: &[u8; 4] = b"PSIV";
22const MAX_RECORD_BYTES: usize = 16 * 1024 * 1024;
23const NON_UNIQUE_MODE_TAG: u8 = 0;
24
25#[derive(Clone, Debug, PartialEq, Eq)]
26struct ByteString(Vec<u8>);
27
28impl Serialize for ByteString {
29    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
30    where
31        S: Serializer,
32    {
33        serializer.serialize_bytes(&self.0)
34    }
35}
36
37struct ByteStringVisitor;
38
39impl<'de> Visitor<'de> for ByteStringVisitor {
40    type Value = ByteString;
41
42    fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
43        formatter.write_str("a CBOR byte string")
44    }
45
46    fn visit_bytes<E>(self, value: &[u8]) -> Result<Self::Value, E>
47    where
48        E: DeError,
49    {
50        Ok(ByteString(value.to_vec()))
51    }
52
53    fn visit_byte_buf<E>(self, value: Vec<u8>) -> Result<Self::Value, E>
54    where
55        E: DeError,
56    {
57        Ok(ByteString(value))
58    }
59}
60
61impl<'de> Deserialize<'de> for ByteString {
62    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
63    where
64        D: Deserializer<'de>,
65    {
66        deserializer.deserialize_byte_buf(ByteStringVisitor)
67    }
68}
69
70fn encode_record<T: Serialize>(magic: &[u8; 4], wire: &T) -> Result<Vec<u8>, Error> {
71    let payload = serde_cbor::to_vec(wire).map_err(|error| Error::Serialize(error.to_string()))?;
72    let total =
73        8usize
74            .checked_add(payload.len())
75            .ok_or_else(|| Error::IndexResourceLimitExceeded {
76                resource: "record_bytes",
77                limit: MAX_RECORD_BYTES,
78                actual: usize::MAX,
79            })?;
80    if total > MAX_RECORD_BYTES {
81        return Err(Error::IndexResourceLimitExceeded {
82            resource: "record_bytes",
83            limit: MAX_RECORD_BYTES,
84            actual: total,
85        });
86    }
87    let mut bytes = Vec::with_capacity(total);
88    bytes.extend_from_slice(magic);
89    bytes.extend_from_slice(&SECONDARY_INDEX_FORMAT_VERSION.to_be_bytes());
90    bytes.extend_from_slice(&payload);
91    Ok(bytes)
92}
93
94fn decode_record<T: DeserializeOwned>(bytes: &[u8], magic: &[u8; 4]) -> Result<T, Error> {
95    if bytes.len() > MAX_RECORD_BYTES {
96        return Err(Error::IndexResourceLimitExceeded {
97            resource: "record_bytes",
98            limit: MAX_RECORD_BYTES,
99            actual: bytes.len(),
100        });
101    }
102    if bytes.len() < 8 || &bytes[..4] != magic {
103        return Err(Error::Deserialize(
104            "invalid secondary-index record magic".to_string(),
105        ));
106    }
107    let version = u32::from_be_bytes(bytes[4..8].try_into().expect("fixed header length"));
108    if version != SECONDARY_INDEX_FORMAT_VERSION {
109        return Err(Error::Deserialize(format!(
110            "unsupported secondary-index record version {version}"
111        )));
112    }
113    let mut decoder = serde_cbor::Deserializer::from_slice(&bytes[8..]);
114    let wire =
115        T::deserialize(&mut decoder).map_err(|error| Error::Deserialize(error.to_string()))?;
116    decoder
117        .end()
118        .map_err(|error| Error::Deserialize(error.to_string()))?;
119    Ok(wire)
120}
121
122fn cid_from_bytes(bytes: Vec<u8>, field: &str) -> Result<Cid, Error> {
123    let array: [u8; 32] = bytes.try_into().map_err(|bytes: Vec<u8>| {
124        Error::Deserialize(format!(
125            "{field} must contain 32 bytes, got {}",
126            bytes.len()
127        ))
128    })?;
129    Ok(Cid(array))
130}
131
132fn projection_tag(projection: IndexProjection) -> u8 {
133    match projection {
134        IndexProjection::KeysOnly => 0,
135        IndexProjection::Include => 1,
136        IndexProjection::All => 2,
137    }
138}
139
140fn projection_from_tag(tag: u8) -> Result<IndexProjection, Error> {
141    match tag {
142        0 => Ok(IndexProjection::KeysOnly),
143        1 => Ok(IndexProjection::Include),
144        2 => Ok(IndexProjection::All),
145        _ => Err(Error::Deserialize(format!(
146            "unknown index projection tag {tag}"
147        ))),
148    }
149}
150
151#[derive(Serialize)]
152struct DescriptorFingerprintWire(ByteString, ByteString, u64, String, u8, u8, u32);
153
154#[derive(Serialize, Deserialize)]
155struct DescriptorWire(
156    u32,
157    ByteString,
158    ByteString,
159    u64,
160    String,
161    ByteString,
162    u8,
163    u32,
164);
165
166/// Canonical persisted semantic definition for one index generation.
167#[derive(Clone, Debug, PartialEq, Eq)]
168pub struct SecondaryIndexDescriptor {
169    pub format_version: u32,
170    pub source_map_id: Vec<u8>,
171    pub name: Vec<u8>,
172    pub generation: u64,
173    pub extractor_id: String,
174    pub fingerprint: Cid,
175    pub projection: IndexProjection,
176    pub physical_layout_version: u32,
177}
178
179impl SecondaryIndexDescriptor {
180    pub fn from_runtime(
181        source_map_id: impl AsRef<[u8]>,
182        index: &SecondaryIndex,
183    ) -> Result<Self, Error> {
184        let mut descriptor = Self {
185            format_version: SECONDARY_INDEX_FORMAT_VERSION,
186            source_map_id: source_map_id.as_ref().to_vec(),
187            name: index.name().to_vec(),
188            generation: index.generation(),
189            extractor_id: index.extractor_id().to_string(),
190            fingerprint: Cid([0; 32]),
191            projection: index.projection(),
192            physical_layout_version: INDEX_PHYSICAL_LAYOUT_VERSION,
193        };
194        descriptor.fingerprint = descriptor_fingerprint(&descriptor)?;
195        Ok(descriptor)
196    }
197
198    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
199        self.validate()?;
200        encode_record(
201            DESCRIPTOR_MAGIC,
202            &DescriptorWire(
203                self.format_version,
204                ByteString(self.source_map_id.clone()),
205                ByteString(self.name.clone()),
206                self.generation,
207                self.extractor_id.clone(),
208                ByteString(self.fingerprint.as_bytes().to_vec()),
209                projection_tag(self.projection),
210                self.physical_layout_version,
211            ),
212        )
213    }
214
215    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
216        let DescriptorWire(
217            format_version,
218            source_map_id,
219            name,
220            generation,
221            extractor_id,
222            fingerprint,
223            projection,
224            physical_layout_version,
225        ) = decode_record(bytes, DESCRIPTOR_MAGIC)?;
226        let descriptor = Self {
227            format_version,
228            source_map_id: source_map_id.0,
229            name: name.0,
230            generation,
231            extractor_id,
232            fingerprint: cid_from_bytes(fingerprint.0, "descriptor fingerprint")?,
233            projection: projection_from_tag(projection)?,
234            physical_layout_version,
235        };
236        descriptor.validate()?;
237        Ok(descriptor)
238    }
239
240    pub(crate) fn validate(&self) -> Result<(), Error> {
241        if self.format_version != SECONDARY_INDEX_FORMAT_VERSION
242            || self.physical_layout_version != INDEX_PHYSICAL_LAYOUT_VERSION
243            || self.source_map_id.is_empty()
244            || self.name.is_empty()
245            || self.generation == 0
246            || self.extractor_id.is_empty()
247        {
248            return Err(Error::InvalidIndexDefinition {
249                reason: "persisted descriptor contains invalid required fields".to_string(),
250            });
251        }
252        let expected = descriptor_fingerprint(self)?;
253        if expected != self.fingerprint {
254            return Err(Error::IndexDefinitionMismatch {
255                name: self.name.clone(),
256                persisted: self.fingerprint.clone(),
257                runtime: expected,
258            });
259        }
260        Ok(())
261    }
262}
263
264/// Hash the canonical semantic descriptor envelope (excluding its fingerprint field).
265pub fn descriptor_fingerprint(descriptor: &SecondaryIndexDescriptor) -> Result<Cid, Error> {
266    let bytes = encode_record(
267        DESCRIPTOR_FINGERPRINT_MAGIC,
268        &DescriptorFingerprintWire(
269            ByteString(descriptor.source_map_id.clone()),
270            ByteString(descriptor.name.clone()),
271            descriptor.generation,
272            descriptor.extractor_id.clone(),
273            NON_UNIQUE_MODE_TAG,
274            projection_tag(descriptor.projection),
275            descriptor.physical_layout_version,
276        ),
277    )?;
278    Ok(Cid::from_bytes(&bytes))
279}
280
281#[derive(Serialize, Deserialize)]
282struct CheckpointWire(
283    ByteString,
284    ByteString,
285    ByteString,
286    u64,
287    ByteString,
288    ByteString,
289    ByteString,
290);
291
292/// Exact hidden-index version selected for one source version.
293#[derive(Clone, Debug, PartialEq, Eq)]
294pub struct IndexCheckpoint {
295    pub source_map_id: Vec<u8>,
296    pub source_version: MapVersionId,
297    pub index_name: Vec<u8>,
298    pub generation: u64,
299    pub definition_fingerprint: Cid,
300    pub index_map_id: Vec<u8>,
301    pub index_version: MapVersionId,
302}
303
304impl IndexCheckpoint {
305    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
306        self.validate()?;
307        encode_record(
308            CHECKPOINT_MAGIC,
309            &CheckpointWire(
310                ByteString(self.source_map_id.clone()),
311                ByteString(self.source_version.as_cid().as_bytes().to_vec()),
312                ByteString(self.index_name.clone()),
313                self.generation,
314                ByteString(self.definition_fingerprint.as_bytes().to_vec()),
315                ByteString(self.index_map_id.clone()),
316                ByteString(self.index_version.as_cid().as_bytes().to_vec()),
317            ),
318        )
319    }
320
321    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
322        let CheckpointWire(
323            source_map_id,
324            source_version,
325            index_name,
326            generation,
327            fingerprint,
328            index_map_id,
329            index_version,
330        ) = decode_record(bytes, CHECKPOINT_MAGIC)?;
331        let checkpoint = Self {
332            source_map_id: source_map_id.0,
333            source_version: MapVersionId::from_cid(cid_from_bytes(
334                source_version.0,
335                "source version",
336            )?),
337            index_name: index_name.0,
338            generation,
339            definition_fingerprint: cid_from_bytes(fingerprint.0, "definition fingerprint")?,
340            index_map_id: index_map_id.0,
341            index_version: MapVersionId::from_cid(cid_from_bytes(
342                index_version.0,
343                "index version",
344            )?),
345        };
346        checkpoint.validate()?;
347        Ok(checkpoint)
348    }
349
350    fn validate(&self) -> Result<(), Error> {
351        if self.source_map_id.is_empty()
352            || self.index_name.is_empty()
353            || self.generation == 0
354            || self.index_map_id
355                != index_map_id(
356                    &self.source_map_id,
357                    &self.index_name,
358                    &self.definition_fingerprint,
359                )
360        {
361            return Err(Error::Deserialize(
362                "checkpoint contains invalid required fields or index ownership".to_string(),
363            ));
364        }
365        Ok(())
366    }
367}
368
369#[derive(Serialize, Deserialize)]
370struct IndexedHeadWire(ByteString, Vec<ByteString>);
371
372/// Canonical current selection of one source version and all active index checkpoints.
373#[derive(Clone, Debug, PartialEq, Eq)]
374pub struct IndexedHeadRecord {
375    pub source_version: MapVersionId,
376    pub indexes: Vec<IndexCheckpoint>,
377}
378
379impl IndexedHeadRecord {
380    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
381        self.validate()?;
382        let checkpoints = self
383            .indexes
384            .iter()
385            .map(IndexCheckpoint::to_bytes)
386            .collect::<Result<Vec<_>, _>>()?;
387        encode_record(
388            HEAD_MAGIC,
389            &IndexedHeadWire(
390                ByteString(self.source_version.as_cid().as_bytes().to_vec()),
391                checkpoints.into_iter().map(ByteString).collect(),
392            ),
393        )
394    }
395
396    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
397        let IndexedHeadWire(source_version, checkpoints) = decode_record(bytes, HEAD_MAGIC)?;
398        let indexes = checkpoints
399            .into_iter()
400            .map(|bytes| IndexCheckpoint::from_bytes(&bytes.0))
401            .collect::<Result<Vec<_>, _>>()?;
402        let record = Self {
403            source_version: MapVersionId::from_cid(cid_from_bytes(
404                source_version.0,
405                "source version",
406            )?),
407            indexes,
408        };
409        record.validate()?;
410        Ok(record)
411    }
412
413    fn validate(&self) -> Result<(), Error> {
414        validate_checkpoint_order(&self.indexes)?;
415        if self
416            .indexes
417            .iter()
418            .any(|checkpoint| checkpoint.source_version != self.source_version)
419        {
420            return Err(Error::Deserialize(
421                "active checkpoint source version does not match indexed head".to_string(),
422            ));
423        }
424        for checkpoint in &self.indexes {
425            checkpoint.validate()?;
426        }
427        Ok(())
428    }
429}
430
431fn validate_checkpoint_order(checkpoints: &[IndexCheckpoint]) -> Result<(), Error> {
432    if checkpoints
433        .windows(2)
434        .any(|pair| pair[0].index_name >= pair[1].index_name)
435    {
436        return Err(Error::Deserialize(
437            "active checkpoints must be strictly sorted by index name".to_string(),
438        ));
439    }
440    Ok(())
441}
442
443#[derive(Clone, Debug, PartialEq, Eq)]
444pub struct ActiveIndexControl {
445    pub name: Vec<u8>,
446    pub fingerprint: Cid,
447}
448
449#[derive(Serialize, Deserialize)]
450struct ActiveControlWire(ByteString, ByteString);
451
452#[derive(Serialize, Deserialize)]
453struct ControlWire(ByteString, ByteString, Vec<ActiveControlWire>);
454
455/// Small deterministic root that fences all raw source-map mutations.
456#[derive(Clone, Debug, PartialEq, Eq)]
457pub struct IndexControl {
458    pub source_map_id: Vec<u8>,
459    pub catalog_map_id: Vec<u8>,
460    pub active: Vec<ActiveIndexControl>,
461}
462
463impl IndexControl {
464    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
465        self.validate()?;
466        encode_record(
467            CONTROL_MAGIC,
468            &ControlWire(
469                ByteString(self.source_map_id.clone()),
470                ByteString(self.catalog_map_id.clone()),
471                self.active
472                    .iter()
473                    .map(|entry| {
474                        ActiveControlWire(
475                            ByteString(entry.name.clone()),
476                            ByteString(entry.fingerprint.as_bytes().to_vec()),
477                        )
478                    })
479                    .collect(),
480            ),
481        )
482    }
483
484    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
485        let ControlWire(source_map_id, catalog_map_id, active) =
486            decode_record(bytes, CONTROL_MAGIC)?;
487        let control = Self {
488            source_map_id: source_map_id.0,
489            catalog_map_id: catalog_map_id.0,
490            active: active
491                .into_iter()
492                .map(|ActiveControlWire(name, fingerprint)| {
493                    Ok(ActiveIndexControl {
494                        name: name.0,
495                        fingerprint: cid_from_bytes(fingerprint.0, "control fingerprint")?,
496                    })
497                })
498                .collect::<Result<Vec<_>, Error>>()?,
499        };
500        control.validate()?;
501        Ok(control)
502    }
503
504    pub fn fingerprint(&self) -> Result<Cid, Error> {
505        Ok(Cid::from_bytes(&self.to_bytes()?))
506    }
507
508    fn validate(&self) -> Result<(), Error> {
509        if self.source_map_id.is_empty()
510            || self.catalog_map_id != catalog_map_id(&self.source_map_id)
511            || self.active.is_empty()
512        {
513            return Err(Error::Deserialize(
514                "invalid secondary-index control record".to_string(),
515            ));
516        }
517        if self
518            .active
519            .windows(2)
520            .any(|pair| pair[0].name >= pair[1].name)
521            || self.active.iter().any(|entry| entry.name.is_empty())
522        {
523            return Err(Error::Deserialize(
524                "active control entries must be non-empty and strictly sorted".to_string(),
525            ));
526        }
527        Ok(())
528    }
529}
530
531/// Canonical physical value stored in a secondary-index tree.
532#[derive(Clone, Debug, PartialEq, Eq)]
533pub enum IndexValue {
534    KeysOnly,
535    Included(Vec<u8>),
536    FullSource(Vec<u8>),
537}
538
539impl IndexValue {
540    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
541        let (tag, payload) = match self {
542            Self::KeysOnly => return Ok(Vec::new()),
543            Self::Included(payload) => (1, payload),
544            Self::FullSource(payload) => (2, payload),
545        };
546        let payload_len =
547            u32::try_from(payload.len()).map_err(|_| Error::IndexResourceLimitExceeded {
548                resource: "projection_bytes",
549                limit: u32::MAX as usize,
550                actual: payload.len(),
551            })?;
552        let mut bytes = Vec::with_capacity(13usize.saturating_add(payload.len()));
553        bytes.extend_from_slice(INDEX_VALUE_MAGIC);
554        bytes.extend_from_slice(&SECONDARY_INDEX_FORMAT_VERSION.to_be_bytes());
555        bytes.push(tag);
556        bytes.extend_from_slice(&payload_len.to_be_bytes());
557        bytes.extend_from_slice(payload);
558        Ok(bytes)
559    }
560
561    pub fn from_bytes(bytes: &[u8], max_payload_bytes: usize) -> Result<Self, Error> {
562        if bytes.is_empty() {
563            return Ok(Self::KeysOnly);
564        }
565        if bytes.len() < 13 || &bytes[..4] != INDEX_VALUE_MAGIC {
566            return Err(Error::Deserialize(
567                "invalid secondary-index value envelope".to_string(),
568            ));
569        }
570        let version = u32::from_be_bytes(bytes[4..8].try_into().expect("fixed header length"));
571        if version != SECONDARY_INDEX_FORMAT_VERSION {
572            return Err(Error::Deserialize(format!(
573                "unsupported secondary-index value version {version}"
574            )));
575        }
576        let payload_len =
577            u32::from_be_bytes(bytes[9..13].try_into().expect("fixed value header length"))
578                as usize;
579        if payload_len > max_payload_bytes {
580            return Err(Error::IndexResourceLimitExceeded {
581                resource: "projection_bytes",
582                limit: max_payload_bytes,
583                actual: payload_len,
584            });
585        }
586        if bytes.len() != 13usize.saturating_add(payload_len) {
587            return Err(Error::Deserialize(
588                "secondary-index value length mismatch or trailing bytes".to_string(),
589            ));
590        }
591        let payload = bytes[13..].to_vec();
592        match bytes[8] {
593            1 => Ok(Self::Included(payload)),
594            2 => Ok(Self::FullSource(payload)),
595            tag => Err(Error::Deserialize(format!(
596                "unknown secondary-index value tag {tag}"
597            ))),
598        }
599    }
600}
601
602pub fn catalog_map_id(source_map_id: impl AsRef<[u8]>) -> Vec<u8> {
603    KeyBuilder::new()
604        .push_str("system")
605        .push_str("secondary-index-catalog")
606        .push_segment(source_map_id)
607        .finish()
608}
609
610pub fn index_map_id(
611    source_map_id: impl AsRef<[u8]>,
612    index_name: impl AsRef<[u8]>,
613    fingerprint: &Cid,
614) -> Vec<u8> {
615    KeyBuilder::new()
616        .push_str("system")
617        .push_str("secondary-index")
618        .push_segment(source_map_id)
619        .push_segment(index_name)
620        .push_segment(fingerprint.as_bytes())
621        .finish()
622}
623
624pub fn control_root_name(source_map_id: impl AsRef<[u8]>) -> Vec<u8> {
625    let source_map_id = source_map_id.as_ref();
626    let mut name = VERSIONED_MAP_ROOT_PREFIX.to_vec();
627    append_hex(&mut name, source_map_id);
628    name.extend_from_slice(b"/secondary-index-control");
629    name
630}
631
632pub fn control_record_key() -> Vec<u8> {
633    KeyBuilder::new().push_str("control").finish()
634}
635
636fn append_hex(output: &mut Vec<u8>, bytes: &[u8]) {
637    const HEX: &[u8; 16] = b"0123456789abcdef";
638    output.reserve(bytes.len().saturating_mul(2));
639    for byte in bytes {
640        output.push(HEX[(byte >> 4) as usize]);
641        output.push(HEX[(byte & 0x0f) as usize]);
642    }
643}
644
645#[derive(Clone, Debug, PartialEq, Eq)]
646pub struct DecodedPhysicalIndexKey {
647    pub term: Vec<u8>,
648    pub primary_key: Vec<u8>,
649}
650
651pub fn physical_index_key(term: &[u8], primary_key: &[u8]) -> Result<Vec<u8>, Error> {
652    let capacity = term
653        .len()
654        .checked_mul(2)
655        .and_then(|size| size.checked_add(primary_key.len().saturating_mul(2)))
656        .and_then(|size| size.checked_add(4))
657        .ok_or(Error::IndexResourceLimitExceeded {
658            resource: "physical_key_bytes",
659            limit: usize::MAX,
660            actual: usize::MAX,
661        })?;
662    let mut key = Vec::with_capacity(capacity);
663    key.extend_from_slice(&encode_segment(term));
664    key.extend_from_slice(&encode_segment(primary_key));
665    Ok(key)
666}
667
668pub fn decode_physical_index_key(key: &[u8]) -> Result<DecodedPhysicalIndexKey, Error> {
669    let segments = decode_segments(key).map_err(|error| Error::Deserialize(error.to_string()))?;
670    let [term, primary_key]: [Vec<u8>; 2] =
671        segments.try_into().map_err(|segments: Vec<Vec<u8>>| {
672            Error::Deserialize(format!(
673                "physical secondary-index key must contain two segments, got {}",
674                segments.len()
675            ))
676        })?;
677    Ok(DecodedPhysicalIndexKey { term, primary_key })
678}
679
680#[derive(Clone, Debug, PartialEq, Eq)]
681pub struct TermBounds {
682    pub start: Vec<u8>,
683    pub end: Option<Vec<u8>>,
684}
685
686pub fn term_bounds_exact(term: &[u8]) -> TermBounds {
687    let start = encode_segment(term);
688    let end = prefix_end(&start);
689    TermBounds { start, end }
690}
691
692pub fn term_bounds_prefix(prefix: &[u8]) -> TermBounds {
693    let start = encode_segment_prefix(prefix);
694    let end = prefix_end(&start);
695    TermBounds { start, end }
696}
697
698pub fn term_bounds_range(start_term: &[u8], end_term: Option<&[u8]>) -> Result<TermBounds, Error> {
699    if end_term.is_some_and(|end| start_term > end) {
700        return Err(Error::InvalidIndexDefinition {
701            reason: "secondary-index term range start exceeds end".to_string(),
702        });
703    }
704    Ok(TermBounds {
705        start: encode_segment(start_term),
706        end: end_term.map(encode_segment),
707    })
708}
709
710pub fn catalog_format_key() -> Vec<u8> {
711    KeyBuilder::new().push_str("format").finish()
712}
713
714pub fn catalog_current_key() -> Vec<u8> {
715    KeyBuilder::new().push_str("current").finish()
716}
717
718pub fn catalog_descriptor_key(name: &[u8], generation: u64) -> Vec<u8> {
719    KeyBuilder::new()
720        .push_str("definitions")
721        .push_segment(name)
722        .push_u64(generation)
723        .finish()
724}
725
726pub fn catalog_checkpoint_key(
727    source_version: &MapVersionId,
728    name: &[u8],
729    generation: u64,
730) -> Vec<u8> {
731    KeyBuilder::new()
732        .push_str("checkpoints")
733        .push_segment(source_version.as_cid().as_bytes())
734        .push_segment(name)
735        .push_u64(generation)
736        .finish()
737}
738
739/// Prefix containing every source-version checkpoint record.
740pub fn catalog_checkpoints_prefix() -> Vec<u8> {
741    KeyBuilder::new().push_str("checkpoints").finish()
742}
743
744pub fn catalog_retired_key(name: &[u8], generation: u64) -> Vec<u8> {
745    KeyBuilder::new()
746        .push_str("retired")
747        .push_segment(name)
748        .push_u64(generation)
749        .finish()
750}
751
752#[cfg(test)]
753mod tests {
754    use super::*;
755    use crate::prolly::cid::Cid;
756
757    fn hex(bytes: &[u8]) -> String {
758        bytes.iter().map(|byte| format!("{byte:02x}")).collect()
759    }
760
761    #[test]
762    fn physical_keys_round_trip_arbitrary_bytes() {
763        let key = physical_index_key(&[0, b'a'], &[b'u', 0, 0xff]).unwrap();
764        assert_eq!(hex(&key), "00ff6100007500ffff0000");
765        let decoded = decode_physical_index_key(&key).unwrap();
766        assert_eq!(decoded.term, vec![0, b'a']);
767        assert_eq!(decoded.primary_key, vec![b'u', 0, 0xff]);
768        assert!(decode_physical_index_key(&[key, vec![1, 0, 0]].concat()).is_err());
769    }
770
771    #[test]
772    fn projection_values_are_versioned_and_canonical() {
773        assert_eq!(IndexValue::KeysOnly.to_bytes().unwrap(), Vec::<u8>::new());
774        let encoded = IndexValue::Included(b"Ada".to_vec()).to_bytes().unwrap();
775        assert_eq!(hex(&encoded), "50534956000000010100000003416461");
776        assert_eq!(
777            IndexValue::from_bytes(&encoded, 1024).unwrap(),
778            IndexValue::Included(b"Ada".to_vec())
779        );
780        assert!(IndexValue::from_bytes(&[encoded, vec![0]].concat(), 1024).is_err());
781        assert!(
782            IndexValue::from_bytes(&IndexValue::FullSource(vec![7; 5]).to_bytes().unwrap(), 4)
783                .is_err()
784        );
785    }
786
787    #[test]
788    fn hidden_ids_and_term_bounds_are_segment_safe() {
789        let fingerprint = Cid([7; 32]);
790        let source = b"users\0prod";
791        let catalog = catalog_map_id(source);
792        let index = index_map_id(source, b"by\0tag", &fingerprint);
793        assert_ne!(catalog, index);
794        assert_eq!(
795            crate::prolly::key::decode_segments(&catalog).unwrap()[2],
796            source
797        );
798        assert_eq!(
799            crate::prolly::key::decode_segments(&index).unwrap()[3],
800            b"by\0tag"
801        );
802
803        let exact = term_bounds_exact(b"a\0");
804        assert!(physical_index_key(b"a\0", b"pk").unwrap() >= exact.start);
805        assert!(physical_index_key(b"a\0x", b"pk").unwrap() >= exact.end.unwrap());
806        let prefix = term_bounds_prefix(b"a\0");
807        assert!(physical_index_key(b"a\0x", b"pk").unwrap() < prefix.end.unwrap());
808    }
809
810    #[test]
811    fn control_record_has_fixed_canonical_bytes() {
812        let control = IndexControl {
813            source_map_id: b"u".to_vec(),
814            catalog_map_id: catalog_map_id(b"u"),
815            active: vec![ActiveIndexControl {
816                name: b"i".to_vec(),
817                fingerprint: Cid([0; 32]),
818            }],
819        };
820        let bytes = control.to_bytes().unwrap();
821        assert_eq!(
822            hex(&bytes),
823            "5053494f00000001834175582473797374656d00007365636f6e646172792d696e6465782d636174616c6f6700007500008182416958200000000000000000000000000000000000000000000000000000000000000000"
824        );
825        assert_eq!(IndexControl::from_bytes(&bytes).unwrap(), control);
826        assert!(IndexControl::from_bytes(&[bytes, vec![0]].concat()).is_err());
827    }
828
829    #[test]
830    fn descriptor_fingerprint_and_bytes_are_canonical() {
831        let runtime =
832            SecondaryIndex::non_unique("i", 1, "x/v1", |_, _| Ok(Vec::<Vec<u8>>::new())).unwrap();
833        let descriptor = SecondaryIndexDescriptor::from_runtime(b"u", &runtime).unwrap();
834        assert_eq!(
835            descriptor.fingerprint,
836            descriptor_fingerprint(&descriptor).unwrap()
837        );
838        let bytes = descriptor.to_bytes().unwrap();
839        assert_eq!(
840            hex(&bytes),
841            "50534944000000018801417541690164782f76315820000a70517a0f9edfd0338318cb726a310dff9f0e7df0324eebf441f365ded3730001"
842        );
843        assert_eq!(
844            SecondaryIndexDescriptor::from_bytes(&bytes).unwrap(),
845            descriptor
846        );
847        assert!(SecondaryIndexDescriptor::from_bytes(&[bytes, vec![0]].concat()).is_err());
848    }
849
850    #[test]
851    fn checkpoints_and_heads_round_trip_and_require_sorted_names() {
852        let checkpoint = IndexCheckpoint {
853            source_map_id: b"u".to_vec(),
854            source_version: MapVersionId::from_cid(Cid([1; 32])),
855            index_name: b"i".to_vec(),
856            generation: 1,
857            definition_fingerprint: Cid([2; 32]),
858            index_map_id: index_map_id(b"u", b"i", &Cid([2; 32])),
859            index_version: MapVersionId::from_cid(Cid([3; 32])),
860        };
861        let bytes = checkpoint.to_bytes().unwrap();
862        assert_eq!(IndexCheckpoint::from_bytes(&bytes).unwrap(), checkpoint);
863        let mut invalid_checkpoint = checkpoint.clone();
864        invalid_checkpoint.generation = 0;
865        assert!(invalid_checkpoint.to_bytes().is_err());
866
867        let head = IndexedHeadRecord {
868            source_version: checkpoint.source_version.clone(),
869            indexes: vec![checkpoint.clone()],
870        };
871        let bytes = head.to_bytes().unwrap();
872        assert_eq!(IndexedHeadRecord::from_bytes(&bytes).unwrap(), head);
873
874        let mismatched = IndexedHeadRecord {
875            source_version: MapVersionId::from_cid(Cid([8; 32])),
876            indexes: vec![checkpoint.clone()],
877        };
878        assert!(mismatched.to_bytes().is_err());
879
880        let mut duplicate = checkpoint;
881        duplicate.generation = 2;
882        let invalid = IndexedHeadRecord {
883            source_version: duplicate.source_version.clone(),
884            indexes: vec![duplicate.clone(), duplicate],
885        };
886        assert!(invalid.to_bytes().is_err());
887    }
888
889    #[test]
890    fn catalog_keys_are_unambiguous_segments() {
891        let version = MapVersionId::from_cid(Cid([9; 32]));
892        assert_eq!(
893            decode_segments(&catalog_format_key()).unwrap(),
894            vec![b"format".to_vec()]
895        );
896        assert_eq!(
897            decode_segments(&catalog_current_key()).unwrap(),
898            vec![b"current".to_vec()]
899        );
900        assert_eq!(
901            decode_segments(&catalog_descriptor_key(b"i\0", 7)).unwrap()[1],
902            b"i\0"
903        );
904        assert_eq!(
905            decode_segments(&catalog_checkpoint_key(&version, b"i", 7)).unwrap()[0],
906            b"checkpoints"
907        );
908        assert_eq!(
909            decode_segments(&catalog_retired_key(b"i", 7)).unwrap()[0],
910            b"retired"
911        );
912    }
913}