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#[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
264pub 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#[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#[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#[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#[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
739pub 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}