1use crate::compression;
35use crate::directory::TileEntry;
36use crate::curve::BlobOrdering;
37use crate::error::{Error, Result};
38use crate::metadata::Metadata;
39use crate::tile::TileId;
40use crate::types::Compression;
41use memmap2::Mmap;
42use rayon::prelude::*;
43use serde::{Deserialize, Serialize};
44use std::collections::HashMap;
45use std::fs::{self, File, OpenOptions};
46use std::io::Write;
47use std::path::{Path, PathBuf};
48
49pub const PACKED_FORMAT: &str = "stt-packed";
51
52pub const PACKED_FORMAT_VERSION: u32 = 1;
55
56pub const DEFAULT_PACK_TARGET_BYTES: u64 = 64 * 1024 * 1024;
59
60fn crc32c_tag(bytes: &[u8]) -> u32 {
62 crc32c::crc32c(bytes)
63}
64
65fn blake3_128_hex(bytes: &[u8]) -> String {
67 let hash = blake3::hash(bytes);
68 hash.as_bytes()[..16]
70 .iter()
71 .map(|b| format!("{b:02x}"))
72 .collect()
73}
74
75pub const DIRECTORY_ENCODING_ZSTD: &str = "zstd";
81
82pub const DIRECTORY_LAYOUT_PAGED: &str = "paged";
85pub const DIRECTORY_LAYOUT_SINGLE: &str = "single";
87
88#[derive(Debug, Clone, Serialize, Deserialize)]
90pub struct DirectoryRef {
91 pub key: String,
93 pub length: u64,
96 #[serde(rename = "directoryVersion")]
100 pub directory_version: u8,
101 #[serde(default, skip_serializing_if = "Option::is_none")]
108 pub encoding: Option<String>,
109 #[serde(default, skip_serializing_if = "Option::is_none")]
114 pub layout: Option<String>,
115 #[serde(default, rename = "rootLength", skip_serializing_if = "Option::is_none")]
119 pub root_length: Option<u64>,
120 #[serde(default, rename = "pageCount", skip_serializing_if = "Option::is_none")]
122 pub page_count: Option<u64>,
123 #[serde(default, rename = "pageEntries", skip_serializing_if = "Option::is_none")]
125 pub page_entries: Option<u64>,
126}
127
128impl DirectoryRef {
129 pub fn is_paged(&self) -> bool {
131 self.layout.as_deref() == Some(DIRECTORY_LAYOUT_PAGED)
132 }
133}
134
135#[derive(Debug, Clone, Serialize, Deserialize)]
138pub struct PackRef {
139 pub key: String,
141 pub length: u64,
143}
144
145#[derive(Debug, Clone, Serialize, Deserialize)]
151pub struct Manifest {
152 pub format: String,
154 #[serde(rename = "formatVersion")]
156 pub format_version: u32,
157 pub compression: String,
159 pub directory: DirectoryRef,
161 pub packs: Vec<PackRef>,
163 pub metadata: Metadata,
165}
166
167impl Manifest {
168 pub fn from_json_bytes(bytes: &[u8]) -> Result<Self> {
170 serde_json::from_slice(bytes)
171 .map_err(|e| Error::InvalidArchive(format!("manifest JSON decode failed: {e}")))
172 }
173
174 pub fn to_json_bytes(&self) -> Result<Vec<u8>> {
176 serde_json::to_vec_pretty(self)
177 .map_err(|e| Error::Other(format!("manifest JSON encode failed: {e}")))
178 }
179}
180
181struct PendingTile {
187 z: u8,
188 x: u32,
189 y: u32,
190 hilbert: u64,
191 time_start: i64,
192 time_end: i64,
193 cover_t_min: Option<i64>,
194 feature_count: u32,
195 temporal_bucket_ms: Option<u64>,
196 payload: Vec<u8>,
197}
198
199pub struct PackWriter {
210 out_dir: PathBuf,
211 ordering: BlobOrdering,
212 pack_target_bytes: u64,
213 pending: Vec<PendingTile>,
214 page_entries: Option<usize>,
218 zstd_level: i32,
223}
224
225impl PackWriter {
226 pub fn create<P: AsRef<Path>>(
234 out_dir: P,
235 ordering: BlobOrdering,
236 pack_target_bytes: u64,
237 ) -> Result<Self> {
238 Ok(Self {
239 out_dir: out_dir.as_ref().to_path_buf(),
240 ordering,
241 pack_target_bytes: pack_target_bytes.max(1),
242 pending: Vec::new(),
243 page_entries: None,
244 zstd_level: compression::ZSTD_LEVEL,
245 })
246 }
247
248 pub fn with_paging(mut self, page_entries: Option<usize>) -> Self {
254 self.page_entries = page_entries.map(|k| k.max(1));
255 self
256 }
257
258 pub fn with_zstd_level(mut self, level: i32) -> Self {
266 self.zstd_level = level.clamp(1, compression::ZSTD_LEVEL_MAX);
267 self
268 }
269
270 #[allow(clippy::too_many_arguments)]
276 pub fn add_tile_full(
277 &mut self,
278 id: &TileId,
279 time_start: i64,
280 time_end: i64,
281 cover_t_min: Option<i64>,
282 feature_count: u32,
283 temporal_bucket_ms: Option<u64>,
284 payload: &[u8],
285 ) -> Result<()> {
286 self.pending.push(PendingTile {
287 z: id.z,
288 x: id.x,
289 y: id.y,
290 hilbert: id.hilbert_index(),
291 time_start,
292 time_end,
293 cover_t_min,
294 feature_count,
295 temporal_bucket_ms,
296 payload: payload.to_vec(),
297 });
298 Ok(())
299 }
300
301 pub fn tile_count(&self) -> usize {
303 self.pending.len()
304 }
305
306 pub fn finalize(self, metadata: &Metadata) -> Result<Manifest> {
315 let PackWriter {
316 out_dir,
317 ordering,
318 pack_target_bytes,
319 mut pending,
320 page_entries,
321 zstd_level,
322 } = self;
323
324 let base_bucket = metadata.temporal_bucket_ms.max(1) as i64;
326 let tb = |p: &PendingTile| {
327 let b = p
328 .temporal_bucket_ms
329 .map(|v| v as i64)
330 .unwrap_or(base_bucket)
331 .max(1);
332 p.time_start.div_euclid(b)
333 };
334 let (tb_min, tb_max) = pending.iter().fold((i64::MAX, i64::MIN), |(lo, hi), p| {
335 let t = tb(p);
336 (lo.min(t), hi.max(t))
337 });
338 let tb_span = if pending.is_empty() { 0 } else { tb_max - tb_min };
339 let ordering = match ordering {
341 BlobOrdering::Auto => {
342 let max_z = pending.iter().map(|p| p.z).max().unwrap_or(0) as u32;
343 let time_bits = crate::curve::bits_for((tb_span.max(0) + 1) as u64);
344 BlobOrdering::choose(max_z, time_bits)
345 }
346 other => other,
347 };
348 pending.sort_by_key(|p| {
356 (
357 crate::curve::space_time_key(
358 ordering,
359 p.z,
360 p.x,
361 p.y,
362 p.hilbert,
363 p.time_start,
364 tb(p),
365 tb_min,
366 tb_span,
367 ),
368 p.z,
369 p.x,
370 p.y,
371 p.time_start,
372 p.temporal_bucket_ms,
373 )
374 });
375
376 struct Blob {
383 compressed: Vec<u8>,
384 uncompressed_size: u32,
385 crc: u32,
386 }
387 let mut blobs: Vec<Blob> = Vec::new();
388 let mut blob_dedup: HashMap<[u8; 32], usize> = HashMap::new();
390 let mut tile_blob: Vec<usize> = Vec::with_capacity(pending.len());
392 const COMPRESS_CHUNK: usize = 8192;
400 for chunk in pending.chunks(COMPRESS_CHUNK) {
401 let compressed_chunk: Vec<Vec<u8>> = chunk
402 .par_iter()
403 .map(|p| compression::compress_zstd_with_dict_level(&p.payload, None, zstd_level))
404 .collect::<Result<Vec<_>>>()?;
405 for (p, compressed) in chunk.iter().zip(compressed_chunk) {
406 let key = *blake3::hash(&compressed).as_bytes();
407 let idx = if let Some(&i) = blob_dedup.get(&key) {
408 i
409 } else {
410 let i = blobs.len();
411 let crc = crc32c_tag(&compressed);
412 let uncompressed_size = p.payload.len() as u32;
413 blobs.push(Blob {
414 compressed,
415 uncompressed_size,
416 crc,
417 });
418 blob_dedup.insert(key, i);
419 i
420 };
421 tile_blob.push(idx);
422 }
423 }
424
425 struct Placement {
432 pack_id: u32,
433 offset: u64,
434 }
435 let mut placements: Vec<Placement> = Vec::with_capacity(blobs.len());
436 let mut packs_blob_ranges: Vec<(usize, usize)> = Vec::new(); if !blobs.is_empty() {
439 let mut pack_start = 0usize;
440 let mut pack_id = 0u32;
441 let mut cur_offset = 0u64;
442 for (i, blob) in blobs.iter().enumerate() {
443 let blen = blob.compressed.len() as u64;
444 if i > pack_start && cur_offset + blen > pack_target_bytes {
448 packs_blob_ranges.push((pack_start, i));
449 pack_start = i;
450 pack_id += 1;
451 cur_offset = 0;
452 }
453 placements.push(Placement {
454 pack_id,
455 offset: cur_offset,
456 });
457 cur_offset += blen;
458 }
459 packs_blob_ranges.push((pack_start, blobs.len()));
460 }
461
462 let mut entries: Vec<TileEntry> = Vec::with_capacity(pending.len());
464 for (p, &bi) in pending.iter().zip(tile_blob.iter()) {
465 let blob = &blobs[bi];
466 let pl = &placements[bi];
467 entries.push(TileEntry {
468 zoom: p.z,
469 x: p.x,
470 y: p.y,
471 time_start: p.time_start,
472 time_end: p.time_end,
473 pack_id: pl.pack_id,
474 offset: pl.offset,
475 length: blob.compressed.len() as u32,
476 uncompressed_size: blob.uncompressed_size,
477 feature_count: p.feature_count,
478 hilbert: p.hilbert,
479 crc32c: blob.crc,
480 temporal_bucket_ms: p.temporal_bucket_ms,
481 cover_t_min: p.cover_t_min,
482 });
483 }
484 entries.sort_by_key(|e| (e.zoom, e.hilbert, e.time_start, e.temporal_bucket_ms));
488
489 let mut metadata = metadata.clone();
496 metadata.tile_count = entries.len() as u64;
497 metadata.feature_count = entries.iter().map(|e| u64::from(e.feature_count)).sum();
498
499 fs::create_dir_all(&out_dir)?;
501 let index_dir = out_dir.join("index");
502 let packs_dir = out_dir.join("packs");
503 fs::create_dir_all(&index_dir)?;
504 fs::create_dir_all(&packs_dir)?;
505
506 let mut pack_refs: Vec<PackRef> = Vec::with_capacity(packs_blob_ranges.len());
509 for (start, end) in &packs_blob_ranges {
510 let mut bytes: Vec<u8> = Vec::new();
511 for blob in &blobs[*start..*end] {
512 bytes.extend_from_slice(&blob.compressed);
513 }
514 let hex = blake3_128_hex(&bytes);
515 let rel = format!("packs/{hex}.sttp");
516 let final_path = out_dir.join(&rel);
517 write_atomic(&packs_dir, &final_path, &bytes)?;
518 pack_refs.push(PackRef {
519 key: rel,
520 length: bytes.len() as u64,
521 });
522 }
523
524 let (index_bytes, directory_ref_fields): (Vec<u8>, _) = if let Some(k) = page_entries {
533 let paged =
534 crate::directory_page::encode_paged_directory_level(&entries, k, true, zstd_level)?;
535 (
536 paged.bytes,
537 (
538 Some(DIRECTORY_LAYOUT_PAGED.to_string()),
539 Some(paged.root_length),
540 Some(paged.page_count as u64),
541 Some(paged.page_entries as u64),
542 ),
543 )
544 } else {
545 let index_plain = crate::directory::encode_directory(&entries);
546 let bytes = compression::compress_zstd_with_dict_level(&index_plain, None, zstd_level)?;
547 (bytes, (None, None, None, None))
548 };
549 let index_hex = blake3_128_hex(&index_bytes);
550 let index_rel = format!("index/{index_hex}.sttd");
551 let index_path = out_dir.join(&index_rel);
552 write_atomic(&index_dir, &index_path, &index_bytes)?;
553
554 let (layout, root_length, page_count, page_entries_field) = directory_ref_fields;
555
556 let manifest = Manifest {
558 format: PACKED_FORMAT.to_string(),
559 format_version: PACKED_FORMAT_VERSION,
560 compression: "zstd".to_string(),
561 directory: DirectoryRef {
562 key: index_rel,
563 length: index_bytes.len() as u64,
564 directory_version: crate::directory::DIRECTORY_VERSION,
565 encoding: Some(DIRECTORY_ENCODING_ZSTD.to_string()),
566 layout,
567 root_length,
568 page_count,
569 page_entries: page_entries_field,
570 },
571 packs: pack_refs,
572 metadata,
573 };
574 let manifest_bytes = manifest.to_json_bytes()?;
575 let manifest_path = out_dir.join("manifest.json");
576 let mut f = File::create(&manifest_path)?;
577 f.write_all(&manifest_bytes)?;
578 f.flush()?;
579
580 Ok(manifest)
581 }
582}
583
584fn decode_directory_object(bytes: &[u8], encoding: Option<&str>) -> Result<Vec<u8>> {
588 match encoding {
589 None => Ok(bytes.to_vec()),
590 Some(DIRECTORY_ENCODING_ZSTD) => compression::decompress_zstd_with_dict(bytes, None),
591 Some(other) => Err(Error::InvalidArchive(format!(
592 "unknown directory encoding {other:?} (this reader supports absent or \"zstd\")"
593 ))),
594 }
595}
596
597fn decode_directory_entries(bytes: &[u8], dref: &DirectoryRef) -> Result<Vec<TileEntry>> {
604 if dref.is_paged() {
605 let root_length = dref.root_length.ok_or_else(|| {
606 Error::InvalidArchive("paged directory: manifest missing rootLength".into())
607 })?;
608 let zstd = dref.encoding.as_deref() == Some(DIRECTORY_ENCODING_ZSTD);
609 crate::directory_page::decode_paged_directory(bytes, root_length, zstd)
610 } else {
611 let raw = decode_directory_object(bytes, dref.encoding.as_deref())?;
612 crate::directory::decode_directory(&raw)
613 }
614}
615
616fn write_atomic(dir: &Path, final_path: &Path, bytes: &[u8]) -> Result<()> {
620 let tmp = dir.join(format!(
622 ".tmp-{}-{}",
623 std::process::id(),
624 blake3_128_hex(bytes)
625 ));
626 {
627 let mut f = OpenOptions::new()
628 .write(true)
629 .create(true)
630 .truncate(true)
631 .open(&tmp)?;
632 f.write_all(bytes)?;
633 f.flush()?;
634 }
635 fs::rename(&tmp, final_path)?;
636 Ok(())
637}
638
639pub fn verify_packed_objects<P: AsRef<Path>>(manifest_path: P) -> Result<Vec<String>> {
655 let manifest_path = manifest_path.as_ref();
656 let root = manifest_path
657 .parent()
658 .ok_or_else(|| Error::InvalidArchive("manifest path has no parent dir".into()))?;
659 let manifest = Manifest::from_json_bytes(&fs::read(manifest_path)?)?;
660
661 let mut issues = Vec::new();
662
663 if manifest.format != PACKED_FORMAT {
664 issues.push(format!(
665 "manifest format is {:?}, expected {PACKED_FORMAT:?}",
666 manifest.format
667 ));
668 }
669 if manifest.format_version != PACKED_FORMAT_VERSION {
670 issues.push(format!(
671 "manifest formatVersion is {}, expected {PACKED_FORMAT_VERSION}",
672 manifest.format_version
673 ));
674 }
675 if manifest.directory.directory_version != crate::directory::DIRECTORY_VERSION {
676 issues.push(format!(
677 "directoryVersion is {}, expected {}",
678 manifest.directory.directory_version,
679 crate::directory::DIRECTORY_VERSION
680 ));
681 }
682
683 fn check_object(
686 root: &Path,
687 key: &str,
688 declared_len: u64,
689 prefix: &str,
690 ext: &str,
691 issues: &mut Vec<String>,
692 ) {
693 match fs::read(root.join(key)) {
694 Ok(bytes) => {
695 if bytes.len() as u64 != declared_len {
696 issues.push(format!(
697 "{key}: on-disk length {} != manifest-declared {declared_len}",
698 bytes.len()
699 ));
700 }
701 let expected = format!("{prefix}/{}.{ext}", blake3_128_hex(&bytes));
702 if key != expected {
703 issues.push(format!(
704 "{key}: content-address mismatch (bytes hash to {expected})"
705 ));
706 }
707 }
708 Err(e) => issues.push(format!("{key}: cannot read object ({e})")),
709 }
710 }
711
712 check_object(
713 root,
714 &manifest.directory.key,
715 manifest.directory.length,
716 "index",
717 "sttd",
718 &mut issues,
719 );
720 for p in &manifest.packs {
721 check_object(root, &p.key, p.length, "packs", "sttp", &mut issues);
722 }
723
724 match fs::read(root.join(&manifest.directory.key)) {
727 Ok(dir_bytes) => match decode_directory_entries(&dir_bytes, &manifest.directory) {
728 Ok(entries) => {
729 if let Some(max_pid) = entries.iter().map(|e| e.pack_id).max() {
730 if max_pid as usize >= manifest.packs.len() {
731 issues.push(format!(
732 "directory references pack_id {max_pid} but the manifest lists only {} pack(s)",
733 manifest.packs.len()
734 ));
735 }
736 }
737 }
738 Err(e) => issues.push(format!("directory failed to decode: {e}")),
739 },
740 Err(_) => {}
742 }
743
744 if manifest.directory.is_paged() {
748 match manifest.directory.root_length {
749 Some(rl) => {
750 if let Ok(dir_bytes) = fs::read(root.join(&manifest.directory.key)) {
751 let zstd =
752 manifest.directory.encoding.as_deref() == Some(DIRECTORY_ENCODING_ZSTD);
753 match crate::directory_page::verify_paged_structure(&dir_bytes, rl, zstd) {
754 Ok(mut more) => issues.append(&mut more),
755 Err(e) => issues.push(format!("paged structure check failed: {e}")),
756 }
757 }
758 }
759 None => issues.push("paged directory: manifest missing rootLength".into()),
760 }
761 }
762
763 Ok(issues)
764}
765
766struct LoadedPack {
772 mmap: Option<Mmap>,
774 path: PathBuf,
776 length: u64,
778}
779
780pub struct PackedReader {
789 entries: Vec<TileEntry>,
790 metadata: Metadata,
791 compression: Compression,
792 packs: Vec<std::cell::RefCell<LoadedPack>>,
793}
794
795impl PackedReader {
796 pub fn open<P: AsRef<Path>>(manifest_path: P) -> Result<Self> {
799 let manifest_path = manifest_path.as_ref();
800 let root = manifest_path
801 .parent()
802 .ok_or_else(|| Error::InvalidArchive("manifest path has no parent dir".into()))?
803 .to_path_buf();
804
805 let manifest_bytes = fs::read(manifest_path)?;
806 let manifest = Manifest::from_json_bytes(&manifest_bytes)?;
807
808 if manifest.format != PACKED_FORMAT {
809 return Err(Error::InvalidArchive(format!(
810 "not a packed manifest: format={:?} (expected {PACKED_FORMAT:?})",
811 manifest.format
812 )));
813 }
814 let compression = match manifest.compression.as_str() {
815 "zstd" => Compression::Zstd,
816 "none" => Compression::None,
817 other => {
818 return Err(Error::InvalidArchive(format!(
819 "unknown packed compression {other:?}"
820 )))
821 }
822 };
823
824 let dir_path = root.join(&manifest.directory.key);
829 let dir_bytes = fs::read(&dir_path)?;
830 let entries = decode_directory_entries(&dir_bytes, &manifest.directory)?;
831
832 let packs = manifest
834 .packs
835 .iter()
836 .map(|p| {
837 std::cell::RefCell::new(LoadedPack {
838 mmap: None,
839 path: root.join(&p.key),
840 length: p.length,
841 })
842 })
843 .collect();
844
845 Ok(Self {
846 entries,
847 metadata: manifest.metadata,
848 compression,
849 packs,
850 })
851 }
852
853 pub fn entries(&self) -> &[TileEntry] {
855 &self.entries
856 }
857
858 pub fn metadata(&self) -> &Metadata {
860 &self.metadata
861 }
862
863 pub fn read_payload(&self, entry: &TileEntry) -> Result<Vec<u8>> {
869 let cell = self.packs.get(entry.pack_id as usize).ok_or_else(|| {
870 Error::InvalidArchive(format!(
871 "tile {:?} references pack {} but only {} packs exist",
872 entry.tile_id(),
873 entry.pack_id,
874 self.packs.len()
875 ))
876 })?;
877
878 let payload = {
879 let mut pack = cell.borrow_mut();
880 if pack.mmap.is_none() {
881 let file = File::open(&pack.path)?;
882 let mmap = unsafe { Mmap::map(&file) }
885 .map_err(|e| Error::Other(format!("mmap failed: {e}")))?;
886 if mmap.len() as u64 != pack.length {
887 return Err(Error::InvalidArchive(format!(
888 "pack {} is {} bytes, manifest declared {}",
889 pack.path.display(),
890 mmap.len(),
891 pack.length
892 )));
893 }
894 pack.mmap = Some(mmap);
895 }
896 let mmap = pack.mmap.as_ref().expect("just loaded");
897
898 let start = entry.offset as usize;
899 let end = start + entry.length as usize;
900 if end > mmap.len() {
901 return Err(Error::InvalidArchive(format!(
902 "tile {:?} blob range {start}..{end} exceeds pack size {}",
903 entry.tile_id(),
904 mmap.len()
905 )));
906 }
907 let compressed = &mmap[start..end];
908
909 if crc32c_tag(compressed) != entry.crc32c {
910 return Err(Error::InvalidArchive(format!(
911 "tile {:?} failed integrity check (corrupt pack)",
912 entry.tile_id()
913 )));
914 }
915
916 if self.compression == Compression::Zstd {
917 compression::decompress_zstd_with_dict(compressed, None)?
918 } else {
919 compression::decompress(compressed, self.compression)?
920 }
921 };
922
923 if payload.len() != entry.uncompressed_size as usize {
924 return Err(Error::InvalidArchive(format!(
925 "tile {:?} decompressed to {} bytes, expected {}",
926 entry.tile_id(),
927 payload.len(),
928 entry.uncompressed_size
929 )));
930 }
931 Ok(payload)
932 }
933
934 pub fn read_layers(&self, entry: &TileEntry) -> Result<Vec<crate::arrow_tile::DecodedLayer>> {
936 let payload = self.read_payload(entry)?;
937 crate::arrow_tile::decode_tile(&payload)
938 }
939}
940
941#[cfg(test)]
942mod tests {
943 use super::*;
944 use crate::arrow_tile::{encode_tile, ColumnarLayer, GeometryColumn};
945
946 fn point_layer(name: &str, ids: Vec<u64>, t0: i64) -> ColumnarLayer {
947 let n = ids.len();
948 ColumnarLayer {
949 name: name.to_string(),
950 feature_ids: ids,
951 start_times: vec![t0; n],
952 end_times: vec![t0 + 100; n],
953 geometry: GeometryColumn::Point(vec![[-122.4, 37.7]; n]),
954 vertex_times: None,
955 vertex_values: None,
956 triangles: None,
957 vertex_value_matrix: None,
958 properties: vec![],
959 }
960 }
961
962 fn distinct_tile(seed: u64) -> Vec<u8> {
965 let ids: Vec<u64> = (0..6).map(|i| seed * 100 + i).collect();
966 encode_tile(&[point_layer("default", ids, (seed as i64) * 7)]).unwrap()
967 }
968
969 #[test]
970 fn blake3_128_hex_is_32_chars() {
971 let h = blake3_128_hex(b"hello");
972 assert_eq!(h.len(), 32);
973 assert!(h.chars().all(|c| c.is_ascii_hexdigit()));
974 }
975
976 #[test]
977 fn manifest_json_roundtrips() {
978 let m = Manifest {
979 format: PACKED_FORMAT.to_string(),
980 format_version: PACKED_FORMAT_VERSION,
981 compression: "zstd".to_string(),
982 directory: DirectoryRef {
983 key: "index/abc.sttd".to_string(),
984 length: 42,
985 directory_version: 5,
986 encoding: Some(DIRECTORY_ENCODING_ZSTD.to_string()),
987 layout: None,
988 root_length: None,
989 page_count: None,
990 page_entries: None,
991 },
992 packs: vec![
993 PackRef { key: "packs/a.sttp".to_string(), length: 100 },
994 PackRef { key: "packs/b.sttp".to_string(), length: 200 },
995 ],
996 metadata: Metadata::new("manifest-test"),
997 };
998 let bytes = m.to_json_bytes().unwrap();
999 let s = String::from_utf8(bytes.clone()).unwrap();
1001 assert!(s.contains("\"formatVersion\""), "{s}");
1002 assert!(s.contains("\"directoryVersion\""), "{s}");
1003 assert!(s.contains("\"stt-packed\""), "{s}");
1004 let back = Manifest::from_json_bytes(&bytes).unwrap();
1005 assert_eq!(back.format, m.format);
1006 assert_eq!(back.format_version, m.format_version);
1007 assert_eq!(back.packs.len(), 2);
1008 assert_eq!(back.directory.directory_version, 5);
1009 assert_eq!(back.directory.encoding.as_deref(), Some("zstd"));
1010 assert_eq!(back.metadata.name, "manifest-test");
1011
1012 let mut legacy_json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
1016 legacy_json["directory"]
1017 .as_object_mut()
1018 .unwrap()
1019 .remove("encoding");
1020 let legacy_back =
1021 Manifest::from_json_bytes(&serde_json::to_vec(&legacy_json).unwrap()).unwrap();
1022 assert_eq!(legacy_back.directory.encoding, None);
1023 let legacy_out =
1024 String::from_utf8(legacy_back.to_json_bytes().unwrap()).unwrap();
1025 assert!(!legacy_out.contains("\"encoding\""), "{legacy_out}");
1026 }
1027
1028 #[test]
1032 fn packwriter_roundtrips_through_multiple_packs() {
1033 let dir = tempfile::tempdir().unwrap();
1034 let out = dir.path().join("dataset");
1035
1036 let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
1037
1038 let static_payload = encode_tile(&[point_layer("default", vec![7, 8, 9], 0)]).unwrap();
1041 let mut expected: Vec<(TileId, i64, i64, Vec<u8>)> = Vec::new();
1042 let bucket = 3_600_000i64;
1043 for k in 0..30u64 {
1044 let zoom = if k % 2 == 0 { 9u8 } else { 10u8 };
1045 let b = (k % 5) as i64; let t = b * bucket;
1047 let payload = if k == 13 || k == 27 {
1048 static_payload.clone() } else {
1050 distinct_tile(k)
1051 };
1052 let id = TileId::new(zoom, (k % 7) as u32, (k / 7) as u32, t as u64);
1053 w.add_tile_full(&id, t, t + bucket - 1, Some(t), 6, Some(bucket as u64), &payload)
1054 .unwrap();
1055 expected.push((id, t, t + bucket - 1, payload));
1056 }
1057 assert_eq!(w.tile_count(), 30);
1058
1059 let meta = Metadata::new("packed-roundtrip").with_temporal_bucket_ms(bucket as u64);
1060 let manifest = w.finalize(&meta).unwrap();
1061
1062 assert!(manifest.packs.len() > 1, "expected multiple packs, got {}", manifest.packs.len());
1064
1065 for p in &manifest.packs {
1067 let bytes = fs::read(out.join(&p.key)).unwrap();
1068 assert_eq!(bytes.len() as u64, p.length);
1069 }
1074
1075 for p in &manifest.packs {
1077 let bytes = fs::read(out.join(&p.key)).unwrap();
1078 let hex = blake3_128_hex(&bytes);
1079 assert_eq!(p.key, format!("packs/{hex}.sttp"));
1080 }
1081 let dir_bytes = fs::read(out.join(&manifest.directory.key)).unwrap();
1082 assert_eq!(
1083 manifest.directory.key,
1084 format!("index/{}.sttd", blake3_128_hex(&dir_bytes))
1085 );
1086
1087 let reader = PackedReader::open(out.join("manifest.json")).unwrap();
1089 let max_pack = reader.entries().iter().map(|e| e.pack_id).max().unwrap();
1090 assert_eq!(max_pack as usize, manifest.packs.len() - 1);
1091 let observed: std::collections::BTreeSet<u32> =
1092 reader.entries().iter().map(|e| e.pack_id).collect();
1093 for pid in 0..manifest.packs.len() as u32 {
1094 assert!(observed.contains(&pid), "pack_id {pid} unused");
1095 }
1096
1097 assert_eq!(reader.metadata().name, "packed-roundtrip");
1099 assert_eq!(reader.metadata().temporal_bucket_ms, bucket as u64);
1100 assert_eq!(reader.entries().len(), 30);
1101
1102 assert_eq!(manifest.metadata.tile_count, 30);
1105 assert_eq!(manifest.metadata.feature_count, 30 * 6);
1106 assert_eq!(reader.metadata().tile_count, 30);
1107 assert_eq!(reader.metadata().feature_count, 30 * 6);
1108
1109 for (id, ts, _te, payload) in &expected {
1111 let e = reader
1112 .entries()
1113 .iter()
1114 .find(|e| {
1115 e.zoom == id.z && e.x == id.x && e.y == id.y && e.time_start == *ts
1116 })
1117 .expect("entry present");
1118 let got = reader.read_payload(e).unwrap();
1119 assert_eq!(&got, payload, "payload mismatch for {id:?}");
1120 }
1121
1122 let static_entries: Vec<&TileEntry> = reader
1124 .entries()
1125 .iter()
1126 .filter(|e| {
1127 let g = reader.read_payload(e).unwrap();
1128 g == static_payload
1129 })
1130 .collect();
1131 assert_eq!(static_entries.len(), 2);
1132 assert_eq!(static_entries[0].pack_id, static_entries[1].pack_id);
1133 assert_eq!(static_entries[0].offset, static_entries[1].offset);
1134 }
1135
1136 #[test]
1145 fn paged_directory_writer_roundtrips_and_matches_single() {
1146 let bucket = 3_600_000i64;
1147 let mut input: Vec<(TileId, i64, i64, Vec<u8>)> = Vec::new();
1149 for k in 0..120u64 {
1150 let zoom = [6u8, 10, 13][(k % 3) as usize];
1151 let b = (k % 4) as i64;
1152 let t = b * bucket;
1153 let id = TileId::new(zoom, (k % 11) as u32, (k / 11) as u32, t as u64);
1154 input.push((id, t, t + bucket - 1, distinct_tile(k)));
1155 }
1156 let build = |out: &Path, page_entries: Option<usize>| -> Manifest {
1157 let mut w = PackWriter::create(out, BlobOrdering::Auto, 16 * 1024)
1158 .unwrap()
1159 .with_paging(page_entries);
1160 for (id, ts, te, payload) in &input {
1161 w.add_tile_full(id, *ts, *te, Some(*ts), 6, Some(bucket as u64), payload)
1162 .unwrap();
1163 }
1164 let meta = Metadata::new("paged-roundtrip").with_temporal_bucket_ms(bucket as u64);
1165 w.finalize(&meta).unwrap()
1166 };
1167
1168 let dir = tempfile::tempdir().unwrap();
1169 let single_out = dir.path().join("single");
1170 let paged_out = dir.path().join("paged");
1171 let single = build(&single_out, None);
1172 let paged = build(&paged_out, Some(16));
1174
1175 assert!(single.directory.layout.is_none());
1177 assert_eq!(paged.directory.layout.as_deref(), Some(DIRECTORY_LAYOUT_PAGED));
1178 assert!(paged.directory.root_length.unwrap() > 0);
1179 assert!(paged.directory.page_count.unwrap() >= 2, "expected multiple leaf pages");
1180 assert_eq!(paged.directory.page_entries, Some(16));
1181 assert_eq!(paged.directory.directory_version, crate::directory::DIRECTORY_VERSION);
1182 assert_eq!(paged.directory.encoding.as_deref(), Some(DIRECTORY_ENCODING_ZSTD));
1183
1184 let r_single = PackedReader::open(single_out.join("manifest.json")).unwrap();
1185 let r_paged = PackedReader::open(paged_out.join("manifest.json")).unwrap();
1186 assert_eq!(r_paged.entries().len(), 120);
1187 assert_eq!(r_single.entries().len(), 120);
1188
1189 let keys = |r: &PackedReader| -> Vec<(u8, u32, u32, i64, i64, u32, Option<u64>, Option<i64>)> {
1192 let mut v: Vec<_> = r
1193 .entries()
1194 .iter()
1195 .map(|e| {
1196 (
1197 e.zoom, e.x, e.y, e.time_start, e.time_end, e.feature_count,
1198 e.temporal_bucket_ms, e.cover_t_min,
1199 )
1200 })
1201 .collect();
1202 v.sort();
1203 v
1204 };
1205 assert_eq!(keys(&r_single), keys(&r_paged));
1206
1207 for (id, ts, _te, payload) in &input {
1209 let e = r_paged
1210 .entries()
1211 .iter()
1212 .find(|x| x.zoom == id.z && x.x == id.x && x.y == id.y && x.time_start == *ts)
1213 .expect("paged entry present");
1214 assert_eq!(&r_paged.read_payload(e).unwrap(), payload, "payload mismatch {id:?}");
1215 }
1216
1217 let issues = verify_packed_objects(paged_out.join("manifest.json")).unwrap();
1219 assert!(issues.is_empty(), "paged verify issues: {issues:?}");
1220 }
1221
1222 #[test]
1224 fn oversized_blob_gets_its_own_pack() {
1225 let dir = tempfile::tempdir().unwrap();
1226 let out = dir.path().join("dataset");
1227
1228 let big_ids: Vec<u64> = (0..4000).collect();
1231 let big = encode_tile(&[ColumnarLayer {
1232 name: "default".to_string(),
1233 feature_ids: big_ids.clone(),
1234 start_times: vec![0; big_ids.len()],
1235 end_times: vec![100; big_ids.len()],
1236 geometry: GeometryColumn::Point(
1237 (0..big_ids.len()).map(|i| [i as f64 * 0.01, i as f64 * 0.013]).collect(),
1238 ),
1239 vertex_times: None,
1240 vertex_values: None,
1241 triangles: None,
1242 vertex_value_matrix: None,
1243 properties: vec![],
1244 }])
1245 .unwrap();
1246 let big_compressed_len = compression::compress_zstd_with_dict(&big, None).unwrap().len();
1247
1248 let mut w = PackWriter::create(&out, BlobOrdering::SpatialMajor, 4 * 1024).unwrap();
1249 assert!(big_compressed_len as u64 > 4 * 1024);
1251 w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 4000, None, &big).unwrap();
1252 for k in 1..4u64 {
1253 let p = distinct_tile(k);
1254 w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p).unwrap();
1255 }
1256 let _manifest = w.finalize(&Metadata::new("big")).unwrap();
1257
1258 let reader = PackedReader::open(out.join("manifest.json")).unwrap();
1260 let big_entry = reader.entries().iter().find(|e| e.x == 0).unwrap();
1261 assert_eq!(reader.read_payload(big_entry).unwrap(), big);
1262 }
1263
1264 #[test]
1269 fn corrupt_pack_blob_is_detected_on_read() {
1270 let dir = tempfile::tempdir().unwrap();
1271 let out = dir.path().join("dataset");
1272
1273 let mut w = PackWriter::create(&out, BlobOrdering::Auto, 64 * 1024 * 1024).unwrap();
1274 for k in 0..4u64 {
1275 let p = distinct_tile(k);
1276 w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p).unwrap();
1277 }
1278 let manifest = w.finalize(&Metadata::new("crc")).unwrap();
1279
1280 let entry = PackedReader::open(out.join("manifest.json")).unwrap().entries()[0].clone();
1282 assert!(PackedReader::open(out.join("manifest.json"))
1283 .unwrap()
1284 .read_payload(&entry)
1285 .is_ok());
1286
1287 let pack_path = out.join(&manifest.packs[entry.pack_id as usize].key);
1289 let mut bytes = fs::read(&pack_path).unwrap();
1290 bytes[entry.offset as usize] ^= 0xff;
1291 fs::write(&pack_path, &bytes).unwrap();
1292
1293 let reader = PackedReader::open(out.join("manifest.json")).unwrap();
1295 assert!(
1296 reader.read_payload(&entry).is_err(),
1297 "corrupt pack blob must fail the read-path CRC32C check"
1298 );
1299 }
1300
1301 #[test]
1307 fn rebuilds_are_byte_reproducible() {
1308 let bucket = 3_600_000i64;
1312 let mut tiles: Vec<(TileId, i64, Option<u64>, Vec<u8>)> = Vec::new();
1313 for k in 0..10u64 {
1314 let t = (k % 4) as i64 * bucket;
1315 tiles.push((
1316 TileId::new(9, (k % 5) as u32, (k / 5) as u32, t as u64),
1317 t,
1318 None,
1319 distinct_tile(k),
1320 ));
1321 }
1322 tiles.push((TileId::new(9, 1, 0, 0), 0, None, distinct_tile(100)));
1324 tiles.push((
1325 TileId::new(9, 1, 0, 0),
1326 0,
1327 Some(24 * bucket as u64),
1328 distinct_tile(101),
1329 ));
1330
1331 let meta = Metadata::new("repro").with_temporal_bucket_ms(bucket as u64);
1332 let build = |order: &[usize]| {
1333 let dir = tempfile::tempdir().unwrap();
1334 let out = dir.path().join("dataset");
1335 let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
1336 for &i in order {
1337 let (id, t, b, payload) = &tiles[i];
1338 w.add_tile_full(id, *t, t + bucket - 1, Some(*t), 6, *b, payload)
1339 .unwrap();
1340 }
1341 let manifest = w.finalize(&meta).unwrap();
1342 (dir, manifest)
1343 };
1344
1345 let forward: Vec<usize> = (0..tiles.len()).collect();
1346 let reverse: Vec<usize> = (0..tiles.len()).rev().collect();
1347 let (_d1, m1) = build(&forward);
1348 let (_d2, m2) = build(&reverse);
1349
1350 assert_eq!(m1.directory.key, m2.directory.key, "directory hash must be stable");
1351 assert_eq!(
1352 m1.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
1353 m2.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
1354 "pack hashes must be stable across rebuilds"
1355 );
1356 assert_eq!(m1.to_json_bytes().unwrap(), m2.to_json_bytes().unwrap());
1357 }
1358
1359 #[test]
1364 fn directory_encoding_compressed_and_raw_both_read() {
1365 let dir = tempfile::tempdir().unwrap();
1366 let out = dir.path().join("dataset");
1367 let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
1368 for k in 0..8u64 {
1369 let p = distinct_tile(k);
1370 w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p)
1371 .unwrap();
1372 }
1373 let manifest = w.finalize(&Metadata::new("dir-enc")).unwrap();
1374 let manifest_path = out.join("manifest.json");
1375
1376 assert_eq!(manifest.directory.encoding.as_deref(), Some("zstd"));
1379 let at_rest = fs::read(out.join(&manifest.directory.key)).unwrap();
1380 assert_eq!(at_rest.len() as u64, manifest.directory.length);
1381 let raw = compression::decompress_zstd_with_dict(&at_rest, None).unwrap();
1382 assert!(crate::directory::decode_directory(&raw).is_ok());
1383 assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
1384 let entries_compressed = PackedReader::open(&manifest_path).unwrap().entries().to_vec();
1385
1386 let raw_hex = blake3_128_hex(&raw);
1389 let raw_rel = format!("index/{raw_hex}.sttd");
1390 fs::write(out.join(&raw_rel), &raw).unwrap();
1391 let mut legacy = manifest.clone();
1392 legacy.directory = DirectoryRef {
1393 key: raw_rel,
1394 length: raw.len() as u64,
1395 directory_version: crate::directory::DIRECTORY_VERSION,
1396 encoding: None,
1397 layout: None,
1398 root_length: None,
1399 page_count: None,
1400 page_entries: None,
1401 };
1402 fs::write(&manifest_path, legacy.to_json_bytes().unwrap()).unwrap();
1403
1404 assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
1405 let entries_raw = PackedReader::open(&manifest_path).unwrap().entries().to_vec();
1406 assert_eq!(entries_raw, entries_compressed);
1407
1408 legacy.directory.encoding = Some("br".to_string());
1410 fs::write(&manifest_path, legacy.to_json_bytes().unwrap()).unwrap();
1411 assert!(PackedReader::open(&manifest_path).is_err());
1412 }
1413
1414 #[test]
1418 fn verify_packed_objects_clean_then_detects_corruption() {
1419 let dir = tempfile::tempdir().unwrap();
1420 let out = dir.path().join("dataset");
1421
1422 let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
1423 for k in 0..12u64 {
1424 let p = distinct_tile(k);
1425 w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p)
1426 .unwrap();
1427 }
1428 let manifest = w.finalize(&Metadata::new("verify")).unwrap();
1429 let manifest_path = out.join("manifest.json");
1430
1431 assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
1433
1434 let pack0 = out.join(&manifest.packs[0].key);
1437 let mut bytes = fs::read(&pack0).unwrap();
1438 bytes[0] ^= 0xff;
1439 fs::write(&pack0, &bytes).unwrap();
1440
1441 let issues = verify_packed_objects(&manifest_path).unwrap();
1442 assert!(
1443 issues.iter().any(|s| s.contains("content-address mismatch")),
1444 "expected a content-address mismatch, got {issues:?}"
1445 );
1446 }
1447}