use crate::arrow_tile::{TemplateCollector, TemplateRegistry};
use crate::compression;
use crate::directory::TileEntry;
use crate::curve::BlobOrdering;
use crate::error::{Error, Result};
use crate::metadata::Metadata;
use crate::tile::TileId;
use crate::types::Compression;
use memmap2::Mmap;
use rayon::prelude::*;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
pub const PACKED_FORMAT: &str = "stt-packed";
pub const PACKED_FORMAT_VERSION_V1: u32 = 1;
pub const PACKED_FORMAT_VERSION_V2: u32 = 2;
pub const PACKED_FORMAT_VERSION: u32 = PACKED_FORMAT_VERSION_V2;
pub const SUPPORTED_PACKED_FORMAT_VERSIONS: [u32; 2] =
[PACKED_FORMAT_VERSION_V1, PACKED_FORMAT_VERSION_V2];
pub const PACK_MAGIC: [u8; 4] = *b"STTP";
pub const DIRECTORY_MAGIC: [u8; 4] = *b"STTD";
pub const OBJECT_MAGIC_LEN: usize = 8;
const OBJECT_MAGIC_VERSION: u8 = 2;
fn object_magic(kind: [u8; 4]) -> [u8; OBJECT_MAGIC_LEN] {
[kind[0], kind[1], kind[2], kind[3], OBJECT_MAGIC_VERSION, 0, 0, 0]
}
fn strip_object_magic<'a>(bytes: &'a [u8], kind: [u8; 4], what: &str) -> Result<&'a [u8]> {
let tag = std::str::from_utf8(&kind).expect("magic tags are ASCII");
if bytes.len() < OBJECT_MAGIC_LEN || bytes[0..4] != kind {
return Err(Error::InvalidArchive(format!(
"{what}: missing {tag:?} magic (formatVersion-2 objects start with an \
8-byte magic prelude)"
)));
}
if bytes[4] != OBJECT_MAGIC_VERSION {
return Err(Error::InvalidArchive(format!(
"{what}: unsupported {tag} object version {} (this reader supports \
{OBJECT_MAGIC_VERSION})",
bytes[4]
)));
}
if bytes[5..OBJECT_MAGIC_LEN] != [0, 0, 0] {
return Err(Error::InvalidArchive(format!(
"{what}: reserved {tag} magic bytes must be zero"
)));
}
Ok(&bytes[OBJECT_MAGIC_LEN..])
}
pub fn directory_codec_bytes(bytes: &[u8], format_version: u32) -> Result<&[u8]> {
if format_version == PACKED_FORMAT_VERSION_V2 {
strip_object_magic(bytes, DIRECTORY_MAGIC, "directory object")
} else {
Ok(bytes)
}
}
pub const CAPABILITY_COORD_QUANT: &str = "coord-quant";
pub const CAPABILITY_ATTR_QUANT: &str = "attr-quant";
pub const CAPABILITY_ELEVATION_FOLD: &str = "elevation-fold";
pub const KNOWN_CAPABILITIES: &[&str] = &[
CAPABILITY_COORD_QUANT,
CAPABILITY_ATTR_QUANT,
CAPABILITY_ELEVATION_FOLD,
];
pub const DEFAULT_PACK_TARGET_BYTES: u64 = 64 * 1024 * 1024;
fn crc32c_tag(bytes: &[u8]) -> u32 {
crc32c::crc32c(bytes)
}
fn blake3_128_hex(bytes: &[u8]) -> String {
let hash = blake3::hash(bytes);
hash.as_bytes()[..16]
.iter()
.map(|b| format!("{b:02x}"))
.collect()
}
pub const DIRECTORY_ENCODING_ZSTD: &str = "zstd";
pub const DIRECTORY_LAYOUT_PAGED: &str = "paged";
pub const DIRECTORY_LAYOUT_SINGLE: &str = "single";
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DirectoryRef {
pub key: String,
pub length: u64,
#[serde(rename = "directoryVersion")]
pub directory_version: u8,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub encoding: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub layout: Option<String>,
#[serde(default, rename = "rootLength", skip_serializing_if = "Option::is_none")]
pub root_length: Option<u64>,
#[serde(default, rename = "pageCount", skip_serializing_if = "Option::is_none")]
pub page_count: Option<u64>,
#[serde(default, rename = "pageEntries", skip_serializing_if = "Option::is_none")]
pub page_entries: Option<u64>,
}
impl DirectoryRef {
pub fn is_paged(&self) -> bool {
self.layout.as_deref() == Some(DIRECTORY_LAYOUT_PAGED)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PackRef {
pub key: String,
pub length: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaTemplateRef {
pub hash: String,
pub data: String,
}
fn build_template_registry(schemas: &[SchemaTemplateRef]) -> Result<TemplateRegistry> {
use base64::Engine as _;
let mut registry = TemplateRegistry::new();
for (i, s) in schemas.iter().enumerate() {
let data = base64::engine::general_purpose::STANDARD
.decode(&s.data)
.map_err(|e| {
Error::InvalidArchive(format!(
"manifest schemas[{i}] ({}): base64 decode failed: {e}",
s.hash
))
})?;
let actual = blake3_128_hex(&data);
if actual != s.hash {
return Err(Error::InvalidArchive(format!(
"manifest schemas[{i}]: template bytes hash to {actual}, declared {}",
s.hash
)));
}
registry.insert(data);
}
Ok(registry)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Manifest {
pub format: String,
#[serde(rename = "formatVersion")]
pub format_version: u32,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub capabilities: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub schemas: Vec<SchemaTemplateRef>,
pub compression: String,
#[serde(default, rename = "blobOrdering", skip_serializing_if = "Option::is_none")]
pub blob_ordering: Option<String>,
pub directory: DirectoryRef,
pub packs: Vec<PackRef>,
pub metadata: Metadata,
}
impl Manifest {
pub fn from_json_bytes(bytes: &[u8]) -> Result<Self> {
serde_json::from_slice(bytes)
.map_err(|e| Error::InvalidArchive(format!("manifest JSON decode failed: {e}")))
}
pub fn to_json_bytes(&self) -> Result<Vec<u8>> {
serde_json::to_vec_pretty(self)
.map_err(|e| Error::Other(format!("manifest JSON encode failed: {e}")))
}
}
struct PendingTile {
z: u8,
x: u32,
y: u32,
hilbert: u64,
time_start: i64,
time_end: i64,
cover_t_min: Option<i64>,
feature_count: u32,
temporal_bucket_ms: Option<u64>,
payload: PendingPayload,
}
enum PendingPayload {
Mem(Vec<u8>),
Spilled { offset: u64, len: u64 },
}
impl PendingPayload {
fn len(&self) -> usize {
match self {
PendingPayload::Mem(v) => v.len(),
PendingPayload::Spilled { len, .. } => *len as usize,
}
}
}
fn check_payload_len(id: &TileId, len: u64) -> Result<()> {
if len > u32::MAX as u64 {
return Err(Error::Other(format!(
"tile {id:?}: payload is {len} bytes, exceeding the directory's u32 \
uncompressed_size field (the packed format cannot represent tiles of \
4 GiB or larger) — split the tile"
)));
}
Ok(())
}
struct SpillFile {
file: File,
path: PathBuf,
len: u64,
}
impl SpillFile {
fn create(dir: &Path) -> Result<Self> {
fs::create_dir_all(dir)?;
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let path = dir.join(format!(".spill-{}-{nanos:x}", std::process::id()));
let file = OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&path)?;
Ok(Self { file, path, len: 0 })
}
fn append(&mut self, payload: &[u8]) -> Result<u64> {
use std::io::{Seek, SeekFrom};
let offset = self.len;
let mut f = &self.file;
f.seek(SeekFrom::Start(offset))?;
f.write_all(payload)?;
self.len += payload.len() as u64;
Ok(offset)
}
fn read(&self, offset: u64, len: usize) -> Result<Vec<u8>> {
use std::io::{Read, Seek, SeekFrom};
let mut f = &self.file;
f.seek(SeekFrom::Start(offset))?;
let mut buf = vec![0u8; len];
f.read_exact(&mut buf)?;
Ok(buf)
}
}
impl Drop for SpillFile {
fn drop(&mut self) {
let _ = fs::remove_file(&self.path);
}
}
pub struct PackWriter {
out_dir: PathBuf,
ordering: BlobOrdering,
pack_target_bytes: u64,
pending: Vec<PendingTile>,
page_entries: Option<usize>,
zstd_level: i32,
capabilities: Vec<String>,
measured_ordering: bool,
format_version: u32,
templates: Arc<TemplateCollector>,
memory_budget: u64,
pending_payload_bytes: u64,
spill: Option<SpillFile>,
}
impl PackWriter {
pub fn create<P: AsRef<Path>>(
out_dir: P,
ordering: BlobOrdering,
pack_target_bytes: u64,
) -> Result<Self> {
Ok(Self {
out_dir: out_dir.as_ref().to_path_buf(),
ordering,
pack_target_bytes: pack_target_bytes.max(1),
pending: Vec::new(),
page_entries: None,
zstd_level: compression::ZSTD_LEVEL,
capabilities: Vec::new(),
measured_ordering: false,
format_version: PACKED_FORMAT_VERSION,
templates: Arc::new(TemplateCollector::new()),
memory_budget: 0,
pending_payload_bytes: 0,
spill: None,
})
}
pub fn with_memory_budget(mut self, bytes: u64) -> Self {
self.memory_budget = bytes;
self
}
pub fn memory_budget(&self) -> u64 {
self.memory_budget
}
pub fn with_seeded_templates(self, templates: &TemplateRegistry) -> Self {
for (_, bytes) in templates.iter() {
self.templates.record(bytes);
}
self
}
pub fn with_format_version(mut self, format_version: u32) -> Self {
self.format_version = format_version;
self
}
pub fn format_version(&self) -> u32 {
self.format_version
}
pub fn template_collector(&self) -> Arc<TemplateCollector> {
Arc::clone(&self.templates)
}
pub fn with_paging(mut self, page_entries: Option<usize>) -> Self {
self.page_entries = page_entries.map(|k| k.max(1));
self
}
pub fn with_zstd_level(mut self, level: i32) -> Self {
self.zstd_level = level.clamp(1, compression::ZSTD_LEVEL_MAX);
self
}
pub fn with_capabilities(mut self, mut capabilities: Vec<String>) -> Self {
capabilities.sort_unstable();
capabilities.dedup();
self.capabilities = capabilities;
self
}
pub fn with_measured_ordering(mut self, on: bool) -> Self {
self.measured_ordering = on;
self
}
#[allow(clippy::too_many_arguments)]
pub fn add_tile_full(
&mut self,
id: &TileId,
time_start: i64,
time_end: i64,
cover_t_min: Option<i64>,
feature_count: u32,
temporal_bucket_ms: Option<u64>,
payload: &[u8],
) -> Result<()> {
check_payload_len(id, payload.len() as u64)?;
let v2_frame = crate::arrow_tile::is_frame_v2(payload);
if v2_frame != (self.format_version == PACKED_FORMAT_VERSION_V2) {
return Err(Error::Other(format!(
"tile {id:?}: {} layer frame fed to a formatVersion-{} writer \
(mixed-version datasets are unreadable; encode payloads with \
the writer's PackWriter::format_version)",
if v2_frame { "v2" } else { "v1" },
self.format_version,
)));
}
let payload = if self.memory_budget > 0
&& self.pending_payload_bytes + payload.len() as u64 > self.memory_budget
{
let spill = match &mut self.spill {
Some(s) => s,
None => self.spill.insert(SpillFile::create(&self.out_dir)?),
};
let offset = spill.append(payload)?;
PendingPayload::Spilled {
offset,
len: payload.len() as u64,
}
} else {
self.pending_payload_bytes += payload.len() as u64;
PendingPayload::Mem(payload.to_vec())
};
self.pending.push(PendingTile {
z: id.z,
x: id.x,
y: id.y,
hilbert: id.hilbert_index(),
time_start,
time_end,
cover_t_min,
feature_count,
temporal_bucket_ms,
payload,
});
Ok(())
}
pub fn tile_count(&self) -> usize {
self.pending.len()
}
pub fn finalize(self, metadata: &Metadata) -> Result<Manifest> {
let PackWriter {
out_dir,
ordering,
pack_target_bytes,
mut pending,
page_entries,
zstd_level,
capabilities,
measured_ordering,
format_version,
templates,
memory_budget,
pending_payload_bytes: _,
spill,
} = self;
if !SUPPORTED_PACKED_FORMAT_VERSIONS.contains(&format_version) {
return Err(Error::Other(format!(
"unsupported packed formatVersion {format_version} (this writer emits 1 or 2)"
)));
}
let base_bucket = metadata.temporal_bucket_ms.max(1) as i64;
let tb = |p: &PendingTile| {
let b = p
.temporal_bucket_ms
.map(|v| v as i64)
.unwrap_or(base_bucket)
.max(1);
p.time_start.div_euclid(b)
};
let (tb_min, tb_max) = pending.iter().fold((i64::MAX, i64::MIN), |(lo, hi), p| {
let t = tb(p);
(lo.min(t), hi.max(t))
});
let tb_span = if pending.is_empty() { 0 } else { tb_max - tb_min };
let native_z = pending.iter().map(|p| p.z).max().unwrap_or(0);
let (mut x_min, mut x_max, mut y_min, mut y_max) = (u32::MAX, 0u32, u32::MAX, 0u32);
for p in pending.iter().filter(|p| p.z == native_z) {
x_min = x_min.min(p.x);
x_max = x_max.max(p.x);
y_min = y_min.min(p.y);
y_max = y_max.max(p.y);
}
let x_span = x_max.saturating_sub(x_min) as u64;
let y_span = y_max.saturating_sub(y_min) as u64;
let ordering = if measured_ordering {
let samples: Vec<crate::ordering_sim::TileSample> = pending
.iter()
.filter(|p| p.z == native_z)
.map(|p| crate::ordering_sim::TileSample {
z: p.z,
x: p.x,
y: p.y,
hilbert: p.hilbert,
time_start: p.time_start,
tb: tb(p),
len: p.payload.len() as u64,
})
.collect();
crate::ordering_sim::measured_ordering(
&samples,
crate::ordering_sim::SimOptions::default(),
)
} else {
match ordering {
BlobOrdering::Auto => {
let space_bits = crate::curve::bits_for(x_span.max(y_span) + 1);
let time_bits = crate::curve::bits_for((tb_span.max(0) + 1) as u64);
BlobOrdering::choose(space_bits, time_bits)
}
other => other,
}
};
pending.sort_by_key(|p| {
(
crate::curve::space_time_key(
ordering,
p.z,
p.x,
p.y,
p.hilbert,
p.time_start,
tb(p),
tb_min,
tb_span,
),
p.z,
p.x,
p.y,
p.time_start,
p.temporal_bucket_ms,
)
});
fs::create_dir_all(&out_dir)?;
let index_dir = out_dir.join("index");
let packs_dir = out_dir.join("packs");
fs::create_dir_all(&index_dir)?;
fs::create_dir_all(&packs_dir)?;
struct Blob {
length: u32,
uncompressed_size: u32,
crc: u32,
}
struct Placement {
pack_id: u32,
offset: u64,
}
let mut stream =
PackStreamWriter::new(&out_dir, &packs_dir, format_version, pack_target_bytes);
let mut blobs: Vec<Blob> = Vec::new();
let mut placements: Vec<Placement> = Vec::new();
let mut blob_dedup: HashMap<[u8; 32], usize> = HashMap::new();
let mut tile_blob: Vec<usize> = Vec::with_capacity(pending.len());
const COMPRESS_CHUNK: usize = 8192;
let chunk_byte_cap: u64 = if spill.is_some() && memory_budget > 0 {
memory_budget
} else {
u64::MAX
};
let mut chunk_ranges: Vec<(usize, usize)> = Vec::new();
{
let mut start = 0usize;
let mut bytes = 0u64;
for (i, p) in pending.iter().enumerate() {
let len = p.payload.len() as u64;
if i > start && (i - start >= COMPRESS_CHUNK || bytes + len > chunk_byte_cap) {
chunk_ranges.push((start, i));
start = i;
bytes = 0;
}
bytes += len;
}
if start < pending.len() {
chunk_ranges.push((start, pending.len()));
}
}
for (chunk_start, chunk_end) in chunk_ranges {
let chunk = &pending[chunk_start..chunk_end];
let materialized: Vec<Option<Vec<u8>>> = chunk
.iter()
.map(|p| match &p.payload {
PendingPayload::Mem(_) => Ok(None),
PendingPayload::Spilled { offset, len } => {
let s = spill
.as_ref()
.expect("spilled payload without a spill file");
s.read(*offset, *len as usize).map(Some)
}
})
.collect::<Result<_>>()?;
let compressed_chunk: Vec<Vec<u8>> = chunk
.par_iter()
.zip(materialized.par_iter())
.map(|(p, m)| {
let payload: &[u8] = match (&p.payload, m) {
(PendingPayload::Mem(v), _) => v,
(PendingPayload::Spilled { .. }, Some(v)) => v,
(PendingPayload::Spilled { .. }, None) => {
unreachable!("spilled payload was not materialised")
}
};
compression::compress_zstd_with_dict_level(payload, None, zstd_level)
})
.collect::<Result<Vec<_>>>()?;
for (p, compressed) in chunk.iter().zip(compressed_chunk) {
let key = *blake3::hash(&compressed).as_bytes();
let idx = if let Some(&i) = blob_dedup.get(&key) {
i
} else {
let i = blobs.len();
let crc = crc32c_tag(&compressed);
let (pack_id, offset) = stream.append(&compressed)?;
blobs.push(Blob {
length: compressed.len() as u32,
uncompressed_size: p.payload.len() as u32,
crc,
});
placements.push(Placement { pack_id, offset });
blob_dedup.insert(key, i);
i
};
tile_blob.push(idx);
}
}
drop(spill);
let pack_refs: Vec<PackRef> = stream.finish()?;
let mut entries: Vec<TileEntry> = Vec::with_capacity(pending.len());
for (p, &bi) in pending.iter().zip(tile_blob.iter()) {
let blob = &blobs[bi];
let pl = &placements[bi];
entries.push(TileEntry {
zoom: p.z,
x: p.x,
y: p.y,
time_start: p.time_start,
time_end: p.time_end,
pack_id: pl.pack_id,
offset: pl.offset,
length: blob.length,
uncompressed_size: blob.uncompressed_size,
feature_count: p.feature_count,
hilbert: p.hilbert,
crc32c: blob.crc,
temporal_bucket_ms: p.temporal_bucket_ms,
cover_t_min: p.cover_t_min,
});
}
entries.sort_by_key(|e| (e.zoom, e.hilbert, e.time_start, e.temporal_bucket_ms));
let mut metadata = metadata.clone();
metadata.tile_count = entries.len() as u64;
metadata.feature_count = entries.iter().map(|e| u64::from(e.feature_count)).sum();
let (index_bytes, directory_ref_fields): (Vec<u8>, _) = if let Some(k) = page_entries {
let paged =
crate::directory_page::encode_paged_directory_level(&entries, k, true, zstd_level)?;
(
paged.bytes,
(
Some(DIRECTORY_LAYOUT_PAGED.to_string()),
Some(paged.root_length),
Some(paged.page_count as u64),
Some(paged.page_entries as u64),
),
)
} else {
let index_plain = crate::directory::encode_directory(&entries);
let bytes = compression::compress_zstd_with_dict_level(&index_plain, None, zstd_level)?;
(bytes, (None, None, None, None))
};
let index_bytes: Vec<u8> = if format_version == PACKED_FORMAT_VERSION_V2 {
let mut with_magic = Vec::with_capacity(OBJECT_MAGIC_LEN + index_bytes.len());
with_magic.extend_from_slice(&object_magic(DIRECTORY_MAGIC));
with_magic.extend_from_slice(&index_bytes);
with_magic
} else {
index_bytes
};
let index_hex = blake3_128_hex(&index_bytes);
let index_rel = format!("index/{index_hex}.sttd");
let index_path = out_dir.join(&index_rel);
write_atomic(&index_dir, &index_path, &index_bytes)?;
let (layout, root_length, page_count, page_entries_field) = directory_ref_fields;
let schemas: Vec<SchemaTemplateRef> = if format_version == PACKED_FORMAT_VERSION_V2 {
use base64::Engine as _;
templates
.snapshot()
.into_iter()
.map(|(hash, data)| SchemaTemplateRef {
hash: hash.iter().map(|b| format!("{b:02x}")).collect(),
data: base64::engine::general_purpose::STANDARD.encode(&data),
})
.collect()
} else {
if !templates.is_empty() {
return Err(Error::Other(
"template collector is non-empty on a formatVersion-1 build: v2-encoded \
payloads were fed to a v1 writer (mixed-version dataset)"
.into(),
));
}
Vec::new()
};
let manifest = Manifest {
format: PACKED_FORMAT.to_string(),
format_version,
capabilities,
schemas,
compression: "zstd".to_string(),
blob_ordering: Some(ordering.as_str().to_string()),
directory: DirectoryRef {
key: index_rel,
length: index_bytes.len() as u64,
directory_version: crate::directory::DIRECTORY_VERSION,
encoding: Some(DIRECTORY_ENCODING_ZSTD.to_string()),
layout,
root_length,
page_count,
page_entries: page_entries_field,
},
packs: pack_refs,
metadata,
};
let manifest_bytes = manifest.to_json_bytes()?;
let manifest_path = out_dir.join("manifest.json");
let mut f = File::create(&manifest_path)?;
f.write_all(&manifest_bytes)?;
f.flush()?;
Ok(manifest)
}
}
fn decode_directory_object(bytes: &[u8], encoding: Option<&str>) -> Result<Vec<u8>> {
match encoding {
None => Ok(bytes.to_vec()),
Some(DIRECTORY_ENCODING_ZSTD) => compression::decompress_zstd_with_dict(bytes, None),
Some(other) => Err(Error::InvalidArchive(format!(
"unknown directory encoding {other:?} (this reader supports absent or \"zstd\")"
))),
}
}
fn decode_directory_entries(bytes: &[u8], dref: &DirectoryRef) -> Result<Vec<TileEntry>> {
if dref.is_paged() {
let root_length = dref.root_length.ok_or_else(|| {
Error::InvalidArchive("paged directory: manifest missing rootLength".into())
})?;
let zstd = dref.encoding.as_deref() == Some(DIRECTORY_ENCODING_ZSTD);
crate::directory_page::decode_paged_directory(bytes, root_length, zstd)
} else {
let raw = decode_directory_object(bytes, dref.encoding.as_deref())?;
crate::directory::decode_directory(&raw)
}
}
fn write_atomic(dir: &Path, final_path: &Path, bytes: &[u8]) -> Result<()> {
let tmp = dir.join(format!(
".tmp-{}-{}",
std::process::id(),
blake3_128_hex(bytes)
));
{
let mut f = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(&tmp)?;
f.write_all(bytes)?;
f.flush()?;
}
fs::rename(&tmp, final_path)?;
Ok(())
}
struct PackStreamWriter {
out_dir: PathBuf,
packs_dir: PathBuf,
format_version: u32,
pack_target_bytes: u64,
data_start: u64,
current: Option<OpenPack>,
next_pack_id: u32,
refs: Vec<PackRef>,
}
struct OpenPack {
file: File,
tmp_path: PathBuf,
hasher: blake3::Hasher,
len: u64,
}
impl Drop for OpenPack {
fn drop(&mut self) {
let _ = fs::remove_file(&self.tmp_path);
}
}
impl PackStreamWriter {
fn new(
out_dir: &Path,
packs_dir: &Path,
format_version: u32,
pack_target_bytes: u64,
) -> Self {
Self {
out_dir: out_dir.to_path_buf(),
packs_dir: packs_dir.to_path_buf(),
format_version,
pack_target_bytes,
data_start: if format_version == PACKED_FORMAT_VERSION_V2 {
OBJECT_MAGIC_LEN as u64
} else {
0
},
current: None,
next_pack_id: 0,
refs: Vec::new(),
}
}
fn append(&mut self, compressed: &[u8]) -> Result<(u32, u64)> {
let blen = compressed.len() as u64;
if let Some(cur) = &self.current {
if cur.len > self.data_start && cur.len + blen > self.pack_target_bytes {
self.seal()?;
}
}
if self.current.is_none() {
self.open()?;
}
let cur = self.current.as_mut().expect("just opened");
let offset = cur.len;
cur.file.write_all(compressed)?;
cur.hasher.update(compressed);
cur.len += blen;
Ok((self.next_pack_id, offset))
}
fn open(&mut self) -> Result<()> {
let tmp_path = self.packs_dir.join(format!(
".tmp-{}-pack-{}",
std::process::id(),
self.next_pack_id
));
let mut file = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(&tmp_path)?;
let mut hasher = blake3::Hasher::new();
let mut len = 0u64;
if self.format_version == PACKED_FORMAT_VERSION_V2 {
let magic = object_magic(PACK_MAGIC);
file.write_all(&magic)?;
hasher.update(&magic);
len = OBJECT_MAGIC_LEN as u64;
}
self.current = Some(OpenPack {
file,
tmp_path,
hasher,
len,
});
Ok(())
}
fn seal(&mut self) -> Result<()> {
let Some(mut cur) = self.current.take() else {
return Ok(());
};
cur.file.flush()?;
let hash = cur.hasher.finalize();
let hex: String = hash.as_bytes()[..16]
.iter()
.map(|b| format!("{b:02x}"))
.collect();
let rel = format!("packs/{hex}.sttp");
fs::rename(&cur.tmp_path, self.out_dir.join(&rel))?;
self.refs.push(PackRef {
key: rel,
length: cur.len,
});
self.next_pack_id += 1;
Ok(())
}
fn finish(mut self) -> Result<Vec<PackRef>> {
self.seal()?;
Ok(std::mem::take(&mut self.refs))
}
}
pub fn verify_packed_objects<P: AsRef<Path>>(manifest_path: P) -> Result<Vec<String>> {
let manifest_path = manifest_path.as_ref();
let root = manifest_path
.parent()
.ok_or_else(|| Error::InvalidArchive("manifest path has no parent dir".into()))?;
let manifest = Manifest::from_json_bytes(&fs::read(manifest_path)?)?;
let mut issues = Vec::new();
if manifest.format != PACKED_FORMAT {
issues.push(format!(
"manifest format is {:?}, expected {PACKED_FORMAT:?}",
manifest.format
));
}
if !SUPPORTED_PACKED_FORMAT_VERSIONS.contains(&manifest.format_version) {
issues.push(format!(
"manifest formatVersion is {}, expected one of {SUPPORTED_PACKED_FORMAT_VERSIONS:?}",
manifest.format_version
));
}
if manifest.directory.directory_version != crate::directory::DIRECTORY_VERSION {
issues.push(format!(
"directoryVersion is {}, expected {}",
manifest.directory.directory_version,
crate::directory::DIRECTORY_VERSION
));
}
verify_manifest_schemas(&manifest, &mut issues);
fn check_object(
root: &Path,
key: &str,
declared_len: u64,
prefix: &str,
ext: &str,
issues: &mut Vec<String>,
) {
match fs::read(root.join(key)) {
Ok(bytes) => {
if bytes.len() as u64 != declared_len {
issues.push(format!(
"{key}: on-disk length {} != manifest-declared {declared_len}",
bytes.len()
));
}
let expected = format!("{prefix}/{}.{ext}", blake3_128_hex(&bytes));
if key != expected {
issues.push(format!(
"{key}: content-address mismatch (bytes hash to {expected})"
));
}
}
Err(e) => issues.push(format!("{key}: cannot read object ({e})")),
}
}
check_object(
root,
&manifest.directory.key,
manifest.directory.length,
"index",
"sttd",
&mut issues,
);
for p in &manifest.packs {
check_object(root, &p.key, p.length, "packs", "sttp", &mut issues);
}
if manifest.format_version == PACKED_FORMAT_VERSION_V2 {
for (key, kind) in std::iter::once((&manifest.directory.key, DIRECTORY_MAGIC))
.chain(manifest.packs.iter().map(|p| (&p.key, PACK_MAGIC)))
{
if let Ok(bytes) = fs::read(root.join(key)) {
if let Err(e) = strip_object_magic(&bytes, kind, key) {
issues.push(e.to_string());
}
}
}
}
match fs::read(root.join(&manifest.directory.key)) {
Ok(dir_bytes) => {
match directory_codec_bytes(&dir_bytes, manifest.format_version)
.and_then(|codec| decode_directory_entries(codec, &manifest.directory))
{
Ok(entries) => {
if let Some(max_pid) = entries.iter().map(|e| e.pack_id).max() {
if max_pid as usize >= manifest.packs.len() {
issues.push(format!(
"directory references pack_id {max_pid} but the manifest lists only {} pack(s)",
manifest.packs.len()
));
}
}
verify_v2_frame_template_refs(
&manifest,
&entries,
|pid| {
manifest
.packs
.get(pid)
.and_then(|p| fs::read(root.join(&p.key)).ok())
},
&mut issues,
);
}
Err(e) => issues.push(format!("directory failed to decode: {e}")),
}
}
Err(_) => {}
}
if manifest.directory.is_paged() {
match manifest.directory.root_length {
Some(rl) => {
if let Ok(dir_bytes) = fs::read(root.join(&manifest.directory.key)) {
if let Ok(codec) = directory_codec_bytes(&dir_bytes, manifest.format_version) {
let zstd = manifest.directory.encoding.as_deref()
== Some(DIRECTORY_ENCODING_ZSTD);
match crate::directory_page::verify_paged_structure(codec, rl, zstd) {
Ok(mut more) => issues.append(&mut more),
Err(e) => issues.push(format!("paged structure check failed: {e}")),
}
}
}
}
None => issues.push("paged directory: manifest missing rootLength".into()),
}
}
Ok(issues)
}
fn verify_manifest_schemas(manifest: &Manifest, issues: &mut Vec<String>) {
if manifest.format_version == PACKED_FORMAT_VERSION_V1 {
if !manifest.schemas.is_empty() {
issues.push("formatVersion-1 manifest carries a schemas table (v2-only)".into());
}
return;
}
if let Err(e) = build_template_registry(&manifest.schemas) {
issues.push(e.to_string());
}
let hashes: Vec<&str> = manifest.schemas.iter().map(|s| s.hash.as_str()).collect();
if hashes.windows(2).any(|w| w[0] >= w[1]) {
issues.push("manifest schemas are not sorted by hash (or contain duplicates)".into());
}
}
fn parse_schema_hash_hex(s: &str) -> Option<[u8; 16]> {
if s.len() != 32 || !s.is_ascii() {
return None;
}
let mut out = [0u8; 16];
for (i, byte) in out.iter_mut().enumerate() {
*byte = u8::from_str_radix(&s[2 * i..2 * i + 2], 16).ok()?;
}
Some(out)
}
fn verify_v2_frame_template_refs<F>(
manifest: &Manifest,
entries: &[TileEntry],
mut pack_object: F,
issues: &mut Vec<String>,
) where
F: FnMut(usize) -> Option<Vec<u8>>,
{
if manifest.format_version != PACKED_FORMAT_VERSION_V2 {
return;
}
let known: std::collections::HashSet<[u8; 16]> = manifest
.schemas
.iter()
.filter_map(|s| parse_schema_hash_hex(&s.hash))
.collect();
let mut seen_blobs: std::collections::HashSet<(u32, u64)> = std::collections::HashSet::new();
let mut pack_cache: HashMap<u32, Option<Vec<u8>>> = HashMap::new();
let mut missing: std::collections::BTreeMap<String, String> = std::collections::BTreeMap::new();
for e in entries {
if !seen_blobs.insert((e.pack_id, e.offset)) {
continue;
}
let Some(pack) = pack_cache
.entry(e.pack_id)
.or_insert_with(|| pack_object(e.pack_id as usize))
.as_deref()
else {
continue;
};
let (start, end) = (e.offset as usize, e.offset as usize + e.length as usize);
if end > pack.len() {
issues.push(format!(
"tile {:?} blob range {start}..{end} exceeds pack {} size {}",
e.tile_id(),
e.pack_id,
pack.len()
));
continue;
}
let payload = match manifest.compression.as_str() {
"zstd" => match compression::decompress_zstd_with_dict(&pack[start..end], None) {
Ok(p) => p,
Err(err) => {
issues.push(format!(
"tile {:?}: blob failed to decompress during the schema-reference \
check: {err}",
e.tile_id()
));
continue;
}
},
_ => pack[start..end].to_vec(),
};
if !crate::arrow_tile::is_frame_v2(&payload) {
issues.push(format!(
"tile {:?}: v1 layer frame inside a formatVersion-2 dataset",
e.tile_id()
));
continue;
}
match crate::arrow_tile::frame_v2_template_refs(&payload) {
Ok(refs) => {
for hash in refs {
if !known.contains(&hash) {
let hex: String = hash.iter().map(|b| format!("{b:02x}")).collect();
missing
.entry(hex)
.or_insert_with(|| format!("{:?}", e.tile_id()));
}
}
}
Err(err) => issues.push(format!(
"tile {:?}: v2 frame header parse failed: {err}",
e.tile_id()
)),
}
}
for (hex, tile) in missing {
issues.push(format!(
"tile frames reference schema template {hex}, which is absent from \
manifest.schemas (e.g. tile {tile}) — no such tile can decode"
));
}
}
struct LoadedPack {
mmap: Option<Mmap>,
path: PathBuf,
length: u64,
}
pub struct PackedReader {
entries: Vec<TileEntry>,
metadata: Metadata,
compression: Compression,
packs: Vec<std::cell::RefCell<LoadedPack>>,
capabilities: Vec<String>,
format_version: u32,
templates: Option<TemplateRegistry>,
bundle: Option<BundleBacking>,
}
struct BundleBacking {
mmap: Mmap,
windows: Vec<(u64, u64)>,
}
fn manifest_open_checks(manifest: &Manifest) -> Result<Compression> {
if manifest.format != PACKED_FORMAT {
return Err(Error::InvalidArchive(format!(
"not a packed manifest: format={:?} (expected {PACKED_FORMAT:?})",
manifest.format
)));
}
if !SUPPORTED_PACKED_FORMAT_VERSIONS.contains(&manifest.format_version) {
return Err(Error::InvalidArchive(format!(
"unsupported packed formatVersion {} (this reader supports 1 and 2)",
manifest.format_version
)));
}
if manifest.format_version == PACKED_FORMAT_VERSION_V1 && !manifest.schemas.is_empty() {
return Err(Error::InvalidArchive(
"formatVersion-1 manifest carries a schemas table (v2-only)".into(),
));
}
let unknown: Vec<&str> = manifest
.capabilities
.iter()
.map(String::as_str)
.filter(|c| !KNOWN_CAPABILITIES.contains(c))
.collect();
if !unknown.is_empty() {
return Err(Error::InvalidArchive(format!(
"dataset requires capabilities this reader does not implement: {} \
(implemented: {})",
unknown.join(", "),
KNOWN_CAPABILITIES.join(", ")
)));
}
match manifest.compression.as_str() {
"zstd" => Ok(Compression::Zstd),
"none" => Ok(Compression::None),
other => Err(Error::InvalidArchive(format!(
"unknown packed compression {other:?}"
))),
}
}
impl PackedReader {
pub fn open<P: AsRef<Path>>(manifest_path: P) -> Result<Self> {
let manifest_path = manifest_path.as_ref();
let root = manifest_path
.parent()
.ok_or_else(|| Error::InvalidArchive("manifest path has no parent dir".into()))?
.to_path_buf();
let manifest_bytes = fs::read(manifest_path)?;
let manifest = Manifest::from_json_bytes(&manifest_bytes)?;
let compression = manifest_open_checks(&manifest)?;
let templates = (manifest.format_version == PACKED_FORMAT_VERSION_V2)
.then(|| build_template_registry(&manifest.schemas))
.transpose()?;
let dir_path = root.join(&manifest.directory.key);
let dir_bytes = fs::read(&dir_path)?;
let codec_bytes = directory_codec_bytes(&dir_bytes, manifest.format_version)?;
let entries = decode_directory_entries(codec_bytes, &manifest.directory)?;
let packs = manifest
.packs
.iter()
.map(|p| {
std::cell::RefCell::new(LoadedPack {
mmap: None,
path: root.join(&p.key),
length: p.length,
})
})
.collect();
Ok(Self {
entries,
metadata: manifest.metadata,
compression,
packs,
capabilities: manifest.capabilities,
format_version: manifest.format_version,
templates,
bundle: None,
})
}
pub fn open_bundle<P: AsRef<Path>>(path: P) -> Result<Self> {
let file = File::open(path.as_ref())?;
let mmap =
unsafe { Mmap::map(&file) }.map_err(|e| Error::Other(format!("mmap failed: {e}")))?;
let (header, header_end) = parse_bundle_header(&mmap)?;
let manifest = Manifest::from_json_bytes(header.manifest.get().as_bytes())?;
let compression = manifest_open_checks(&manifest)?;
let mut by_key: HashMap<&str, (u64, u64)> = HashMap::with_capacity(header.objects.len());
for o in &header.objects {
let end = o.offset.checked_add(o.length).ok_or_else(|| {
Error::InvalidArchive(format!("bundle object {:?}: offset+length overflows", o.key))
})?;
if o.offset < header_end || end > mmap.len() as u64 {
return Err(Error::InvalidArchive(format!(
"bundle object {:?} window {}..{end} outside the data region ({header_end}..{})",
o.key,
o.offset,
mmap.len()
)));
}
if by_key.insert(o.key.as_str(), (o.offset, o.length)).is_some() {
return Err(Error::InvalidArchive(format!(
"bundle header lists object key {:?} twice",
o.key
)));
}
}
let lookup = |key: &str, declared: u64, what: &str| -> Result<(u64, u64)> {
let &(offset, length) = by_key.get(key).ok_or_else(|| {
Error::InvalidArchive(format!("bundle header lists no object for {what} {key:?}"))
})?;
if length != declared {
return Err(Error::InvalidArchive(format!(
"{what} {key:?} is {length} bytes in the bundle, manifest declared {declared}"
)));
}
Ok((offset, length))
};
let templates = (manifest.format_version == PACKED_FORMAT_VERSION_V2)
.then(|| build_template_registry(&manifest.schemas))
.transpose()?;
let (dir_off, dir_len) =
lookup(&manifest.directory.key, manifest.directory.length, "directory")?;
let dir_object = &mmap[dir_off as usize..(dir_off + dir_len) as usize];
let codec_bytes = directory_codec_bytes(dir_object, manifest.format_version)?;
let entries = decode_directory_entries(codec_bytes, &manifest.directory)?;
let windows = manifest
.packs
.iter()
.map(|p| lookup(&p.key, p.length, "pack"))
.collect::<Result<Vec<_>>>()?;
if manifest.format_version == PACKED_FORMAT_VERSION_V2 {
for (&(off, len), p) in windows.iter().zip(&manifest.packs) {
strip_object_magic(
&mmap[off as usize..(off + len) as usize],
PACK_MAGIC,
&format!("bundle pack {:?}", p.key),
)?;
}
}
Ok(Self {
entries,
metadata: manifest.metadata,
compression,
packs: Vec::new(),
capabilities: manifest.capabilities,
format_version: manifest.format_version,
templates,
bundle: Some(BundleBacking { mmap, windows }),
})
}
pub fn entries(&self) -> &[TileEntry] {
&self.entries
}
pub fn metadata(&self) -> &Metadata {
&self.metadata
}
pub fn capabilities(&self) -> &[String] {
&self.capabilities
}
pub fn read_payload(&self, entry: &TileEntry) -> Result<Vec<u8>> {
if let Some(bundle) = &self.bundle {
let &(pack_offset, pack_length) =
bundle.windows.get(entry.pack_id as usize).ok_or_else(|| {
Error::InvalidArchive(format!(
"tile {:?} references pack {} but the bundle holds {} pack(s)",
entry.tile_id(),
entry.pack_id,
bundle.windows.len()
))
})?;
let end_in_pack = entry.offset.checked_add(entry.length as u64).ok_or_else(|| {
Error::InvalidArchive(format!(
"tile {:?}: blob offset+length overflows",
entry.tile_id()
))
})?;
if end_in_pack > pack_length {
return Err(Error::InvalidArchive(format!(
"tile {:?} blob range {}..{end_in_pack} exceeds pack size {pack_length}",
entry.tile_id(),
entry.offset
)));
}
let start = (pack_offset + entry.offset) as usize;
let end = start + entry.length as usize;
return self.decode_blob(entry, &bundle.mmap[start..end]);
}
let cell = self.packs.get(entry.pack_id as usize).ok_or_else(|| {
Error::InvalidArchive(format!(
"tile {:?} references pack {} but only {} packs exist",
entry.tile_id(),
entry.pack_id,
self.packs.len()
))
})?;
let mut pack = cell.borrow_mut();
if pack.mmap.is_none() {
let file = File::open(&pack.path)?;
let mmap = unsafe { Mmap::map(&file) }
.map_err(|e| Error::Other(format!("mmap failed: {e}")))?;
if mmap.len() as u64 != pack.length {
return Err(Error::InvalidArchive(format!(
"pack {} is {} bytes, manifest declared {}",
pack.path.display(),
mmap.len(),
pack.length
)));
}
if self.format_version == PACKED_FORMAT_VERSION_V2 {
strip_object_magic(
&mmap,
PACK_MAGIC,
&format!("pack {}", pack.path.display()),
)?;
}
pack.mmap = Some(mmap);
}
let mmap = pack.mmap.as_ref().expect("just loaded");
let start = entry.offset as usize;
let end = start + entry.length as usize;
if end > mmap.len() {
return Err(Error::InvalidArchive(format!(
"tile {:?} blob range {start}..{end} exceeds pack size {}",
entry.tile_id(),
mmap.len()
)));
}
self.decode_blob(entry, &mmap[start..end])
}
fn decode_blob(&self, entry: &TileEntry, compressed: &[u8]) -> Result<Vec<u8>> {
if crc32c_tag(compressed) != entry.crc32c {
return Err(Error::InvalidArchive(format!(
"tile {:?} failed integrity check (corrupt pack)",
entry.tile_id()
)));
}
let payload = if self.compression == Compression::Zstd {
compression::decompress_zstd_with_dict(compressed, None)?
} else {
compression::decompress(compressed, self.compression)?
};
if payload.len() != entry.uncompressed_size as usize {
return Err(Error::InvalidArchive(format!(
"tile {:?} decompressed to {} bytes, expected {}",
entry.tile_id(),
payload.len(),
entry.uncompressed_size
)));
}
Ok(payload)
}
pub fn read_layers(&self, entry: &TileEntry) -> Result<Vec<crate::arrow_tile::DecodedLayer>> {
let payload = self.read_payload(entry)?;
self.decode_payload(&payload)
}
pub fn decode_payload(&self, payload: &[u8]) -> Result<Vec<crate::arrow_tile::DecodedLayer>> {
let v2_frame = crate::arrow_tile::is_frame_v2(payload);
if self.format_version == PACKED_FORMAT_VERSION_V2 {
if !v2_frame {
return Err(Error::InvalidArchive(
"v1 layer frame inside a formatVersion-2 dataset (the manifest is \
authoritative; mixed-version datasets are invalid)"
.into(),
));
}
let templates = self
.templates
.as_ref()
.expect("v2 open always builds the template registry");
crate::arrow_tile::decode_tile_with_templates(payload, templates)
} else {
if v2_frame {
return Err(Error::InvalidArchive(
"v2 layer frame inside a formatVersion-1 dataset (the manifest is \
authoritative; mixed-version datasets are invalid)"
.into(),
));
}
crate::arrow_tile::decode_tile(payload)
}
}
pub fn format_version(&self) -> u32 {
self.format_version
}
pub fn templates(&self) -> Option<&TemplateRegistry> {
self.templates.as_ref()
}
}
pub const BUNDLE_MAGIC: [u8; 4] = *b"STTB";
pub const BUNDLE_VERSION: u8 = 1;
const BUNDLE_PRELUDE_LEN: usize = 12;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BundleObject {
pub key: String,
pub offset: u64,
pub length: u64,
}
#[derive(Serialize, Deserialize)]
struct BundleHeader {
manifest: Box<serde_json::value::RawValue>,
objects: Vec<BundleObject>,
}
#[derive(Debug, Clone, Copy)]
pub struct BundleSummary {
pub objects: usize,
pub bytes: u64,
}
fn align8_u64(n: u64) -> u64 {
n.div_ceil(8) * 8
}
fn validate_bundle_key(key: &str) -> Result<()> {
let bad = key.is_empty()
|| key.starts_with('/')
|| key.contains('\\')
|| key.contains(':')
|| key.split('/').any(|seg| seg.is_empty() || seg == "." || seg == "..");
if bad {
return Err(Error::InvalidArchive(format!(
"unsafe bundle object key {key:?} (keys must be relative paths \
with no empty/'.'/'..' segments)"
)));
}
Ok(())
}
fn content_address_stem(key: &str) -> Option<&str> {
let name = key.rsplit('/').next()?;
let stem = name.split('.').next()?;
(stem.len() == 32 && stem.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')))
.then_some(stem)
}
struct ManifestObjectKey {
key: String,
declared_length: Option<u64>,
}
fn manifest_object_keys(manifest: &serde_json::Value) -> Result<Vec<ManifestObjectKey>> {
let obj = manifest
.as_object()
.ok_or_else(|| Error::InvalidArchive("manifest is not a JSON object".into()))?;
if obj.get("format").and_then(|v| v.as_str()) != Some(PACKED_FORMAT) {
return Err(Error::InvalidArchive(format!(
"not a packed manifest: format={:?} (expected {PACKED_FORMAT:?})",
obj.get("format")
)));
}
let entry = |v: &serde_json::Value, what: &str| -> Result<ManifestObjectKey> {
let o = v
.as_object()
.ok_or_else(|| Error::InvalidArchive(format!("manifest {what} is not an object")))?;
let key = o
.get("key")
.and_then(|k| k.as_str())
.ok_or_else(|| Error::InvalidArchive(format!("manifest {what} has no string `key`")))?;
Ok(ManifestObjectKey {
key: key.to_string(),
declared_length: o.get("length").and_then(|l| l.as_u64()),
})
};
let mut out = Vec::new();
out.push(entry(
obj.get("directory")
.ok_or_else(|| Error::InvalidArchive("manifest has no `directory`".into()))?,
"directory",
)?);
let packs = obj
.get("packs")
.and_then(|v| v.as_array())
.ok_or_else(|| Error::InvalidArchive("manifest has no `packs` array".into()))?;
for (i, p) in packs.iter().enumerate() {
out.push(entry(p, &format!("packs[{i}]"))?);
}
let mut seen = std::collections::HashSet::with_capacity(out.len());
for k in &out {
validate_bundle_key(&k.key)?;
if !seen.insert(k.key.as_str()) {
return Err(Error::InvalidArchive(format!(
"manifest lists object key {:?} twice",
k.key
)));
}
}
Ok(out)
}
fn parse_bundle_header(bytes: &[u8]) -> Result<(BundleHeader, u64)> {
if bytes.len() < BUNDLE_PRELUDE_LEN {
return Err(Error::InvalidArchive(format!(
"not an STT bundle: {} bytes is shorter than the {BUNDLE_PRELUDE_LEN}-byte prelude",
bytes.len()
)));
}
if bytes[0..4] != BUNDLE_MAGIC {
return Err(Error::InvalidArchive(
"not an STT bundle (bad magic; expected \"STTB\")".into(),
));
}
if bytes[4] != BUNDLE_VERSION {
return Err(Error::InvalidArchive(format!(
"unsupported bundle version {} (this reader supports {BUNDLE_VERSION})",
bytes[4]
)));
}
if bytes[5..8] != [0, 0, 0] {
return Err(Error::InvalidArchive(
"malformed bundle: reserved prelude bytes must be zero".into(),
));
}
let header_len = u32::from_le_bytes(bytes[8..12].try_into().expect("4 bytes")) as u64;
let header_end = BUNDLE_PRELUDE_LEN as u64 + header_len;
if header_end > bytes.len() as u64 {
return Err(Error::InvalidArchive(format!(
"truncated bundle: header claims {header_len} bytes but only {} remain",
bytes.len() - BUNDLE_PRELUDE_LEN
)));
}
let header: BundleHeader =
serde_json::from_slice(&bytes[BUNDLE_PRELUDE_LEN..header_end as usize])
.map_err(|e| Error::InvalidArchive(format!("bundle header JSON decode failed: {e}")))?;
Ok((header, header_end))
}
pub fn write_bundle<P: AsRef<Path>, Q: AsRef<Path>>(
manifest_path: P,
out_path: Q,
) -> Result<BundleSummary> {
let manifest_path = manifest_path.as_ref();
let out_path = out_path.as_ref();
let root = manifest_path
.parent()
.ok_or_else(|| Error::InvalidArchive("manifest path has no parent dir".into()))?;
let manifest_raw = fs::read(manifest_path)?;
let manifest_text = std::str::from_utf8(&manifest_raw)
.map_err(|_| Error::InvalidArchive("manifest.json is not UTF-8".into()))?
.trim()
.to_string();
let manifest_value: serde_json::Value = serde_json::from_str(&manifest_text)
.map_err(|e| Error::InvalidArchive(format!("manifest JSON decode failed: {e}")))?;
let keys = manifest_object_keys(&manifest_value)?;
let manifest_rv = serde_json::value::RawValue::from_string(manifest_text)
.map_err(|e| Error::InvalidArchive(format!("manifest is not a JSON value: {e}")))?;
let mut lengths: Vec<u64> = Vec::with_capacity(keys.len());
for k in &keys {
let len = fs::metadata(root.join(&k.key))
.map_err(|e| Error::InvalidArchive(format!("{}: cannot stat object ({e})", k.key)))?
.len();
if let Some(declared) = k.declared_length {
if len != declared {
return Err(Error::InvalidArchive(format!(
"{}: on-disk length {len} != manifest-declared {declared}",
k.key
)));
}
}
lengths.push(len);
}
let mut rel_offsets: Vec<u64> = Vec::with_capacity(lengths.len());
let mut data_len = 0u64;
for len in &lengths {
rel_offsets.push(data_len);
data_len = align8_u64(data_len + len);
}
let mut data_start = align8_u64(BUNDLE_PRELUDE_LEN as u64);
let header_bytes = loop {
let objects: Vec<BundleObject> = keys
.iter()
.zip(&rel_offsets)
.zip(&lengths)
.map(|((k, rel), len)| BundleObject {
key: k.key.clone(),
offset: data_start + rel,
length: *len,
})
.collect();
let header = BundleHeader {
manifest: manifest_rv.clone(),
objects,
};
let bytes = serde_json::to_vec(&header)
.map_err(|e| Error::Other(format!("bundle header encode failed: {e}")))?;
if bytes.len() as u64 > u32::MAX as u64 {
return Err(Error::InvalidArchive(format!(
"bundle header is {} bytes, exceeding the u32 header_len cap",
bytes.len()
)));
}
let next = align8_u64(BUNDLE_PRELUDE_LEN as u64 + bytes.len() as u64);
if next == data_start {
break bytes;
}
data_start = next;
};
let out_dir = match out_path.parent() {
Some(p) if p != Path::new("") => p.to_path_buf(),
_ => PathBuf::from("."),
};
fs::create_dir_all(&out_dir)?;
let tmp = out_dir.join(format!(".tmp-sttb-{}", std::process::id()));
let total = {
use std::io::BufWriter;
let file = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(&tmp)?;
let mut w = BufWriter::new(file);
const ZEROS: [u8; 8] = [0u8; 8];
w.write_all(&BUNDLE_MAGIC)?;
w.write_all(&[BUNDLE_VERSION, 0, 0, 0])?;
w.write_all(&(header_bytes.len() as u32).to_le_bytes())?;
w.write_all(&header_bytes)?;
let mut pos = BUNDLE_PRELUDE_LEN as u64 + header_bytes.len() as u64;
w.write_all(&ZEROS[..(data_start - pos) as usize])?;
pos = data_start;
for ((k, rel), len) in keys.iter().zip(&rel_offsets).zip(&lengths) {
debug_assert_eq!(pos, data_start + rel);
let bytes = fs::read(root.join(&k.key))?;
if bytes.len() as u64 != *len {
return Err(Error::InvalidArchive(format!(
"{}: object changed size during bundling ({} != planned {len})",
k.key,
bytes.len()
)));
}
if let Some(stem) = content_address_stem(&k.key) {
let hex = blake3_128_hex(&bytes);
if hex != stem {
return Err(Error::InvalidArchive(format!(
"{}: content-address mismatch (bytes hash to {hex})",
k.key
)));
}
}
w.write_all(&bytes)?;
pos += *len;
let padded = align8_u64(pos);
w.write_all(&ZEROS[..(padded - pos) as usize])?;
pos = padded;
}
w.flush()?;
pos
};
fs::rename(&tmp, out_path)?;
Ok(BundleSummary {
objects: keys.len(),
bytes: total,
})
}
pub fn unpack_bundle<P: AsRef<Path>, Q: AsRef<Path>>(
bundle_path: P,
out_dir: Q,
) -> Result<BundleSummary> {
let out_dir = out_dir.as_ref();
let file = File::open(bundle_path.as_ref())?;
let mmap =
unsafe { Mmap::map(&file) }.map_err(|e| Error::Other(format!("mmap failed: {e}")))?;
let (header, header_end) = parse_bundle_header(&mmap)?;
let mut seen = std::collections::HashSet::with_capacity(header.objects.len());
for o in &header.objects {
validate_bundle_key(&o.key)?;
if !seen.insert(o.key.as_str()) {
return Err(Error::InvalidArchive(format!(
"bundle header lists object key {:?} twice",
o.key
)));
}
let end = o.offset.checked_add(o.length).ok_or_else(|| {
Error::InvalidArchive(format!("bundle object {:?}: offset+length overflows", o.key))
})?;
if o.offset < header_end || end > mmap.len() as u64 {
return Err(Error::InvalidArchive(format!(
"bundle object {:?} window {}..{end} outside the data region ({header_end}..{})",
o.key,
o.offset,
mmap.len()
)));
}
}
let manifest_value: serde_json::Value = serde_json::from_str(header.manifest.get())
.map_err(|e| Error::InvalidArchive(format!("bundle manifest JSON decode failed: {e}")))?;
for k in manifest_object_keys(&manifest_value)? {
if !seen.contains(k.key.as_str()) {
return Err(Error::InvalidArchive(format!(
"bundle header carries no object for manifest key {:?}",
k.key
)));
}
}
fs::create_dir_all(out_dir)?;
let mut bytes_written = 0u64;
for o in &header.objects {
let path = out_dir.join(&o.key);
let parent = path.parent().map(Path::to_path_buf).unwrap_or_else(|| out_dir.to_path_buf());
fs::create_dir_all(&parent)?;
write_atomic(
&parent,
&path,
&mmap[o.offset as usize..(o.offset + o.length) as usize],
)?;
bytes_written += o.length;
}
let manifest_bytes = header.manifest.get().as_bytes();
write_atomic(out_dir, &out_dir.join("manifest.json"), manifest_bytes)?;
bytes_written += manifest_bytes.len() as u64;
Ok(BundleSummary {
objects: header.objects.len(),
bytes: bytes_written,
})
}
pub fn read_bundle_manifest<P: AsRef<Path>>(bundle_path: P) -> Result<Manifest> {
let file = File::open(bundle_path.as_ref())?;
let mmap =
unsafe { Mmap::map(&file) }.map_err(|e| Error::Other(format!("mmap failed: {e}")))?;
let (header, _) = parse_bundle_header(&mmap)?;
Manifest::from_json_bytes(header.manifest.get().as_bytes())
}
pub fn verify_bundle_objects<P: AsRef<Path>>(bundle_path: P) -> Result<Vec<String>> {
let file = File::open(bundle_path.as_ref())?;
let mmap =
unsafe { Mmap::map(&file) }.map_err(|e| Error::Other(format!("mmap failed: {e}")))?;
let (header, header_end) = parse_bundle_header(&mmap)?;
let manifest = Manifest::from_json_bytes(header.manifest.get().as_bytes())?;
let mut issues = Vec::new();
if manifest.format != PACKED_FORMAT {
issues.push(format!(
"manifest format is {:?}, expected {PACKED_FORMAT:?}",
manifest.format
));
}
if !SUPPORTED_PACKED_FORMAT_VERSIONS.contains(&manifest.format_version) {
issues.push(format!(
"manifest formatVersion is {}, expected one of {SUPPORTED_PACKED_FORMAT_VERSIONS:?}",
manifest.format_version
));
}
if manifest.directory.directory_version != crate::directory::DIRECTORY_VERSION {
issues.push(format!(
"directoryVersion is {}, expected {}",
manifest.directory.directory_version,
crate::directory::DIRECTORY_VERSION
));
}
verify_manifest_schemas(&manifest, &mut issues);
let mut by_key: HashMap<&str, &[u8]> = HashMap::with_capacity(header.objects.len());
for o in &header.objects {
if let Err(e) = validate_bundle_key(&o.key) {
issues.push(e.to_string());
continue;
}
let end = match o.offset.checked_add(o.length) {
Some(end) if o.offset >= header_end && end <= mmap.len() as u64 => end,
_ => {
issues.push(format!(
"bundle object {:?} window {}+{} outside the data region ({header_end}..{})",
o.key,
o.offset,
o.length,
mmap.len()
));
continue;
}
};
if by_key
.insert(o.key.as_str(), &mmap[o.offset as usize..end as usize])
.is_some()
{
issues.push(format!("bundle header lists object key {:?} twice", o.key));
}
}
fn check_object(
by_key: &HashMap<&str, &[u8]>,
key: &str,
declared_len: u64,
prefix: &str,
ext: &str,
issues: &mut Vec<String>,
) {
match by_key.get(key) {
Some(bytes) => {
if bytes.len() as u64 != declared_len {
issues.push(format!(
"{key}: in-bundle length {} != manifest-declared {declared_len}",
bytes.len()
));
}
let expected = format!("{prefix}/{}.{ext}", blake3_128_hex(bytes));
if key != expected {
issues.push(format!(
"{key}: content-address mismatch (bytes hash to {expected})"
));
}
}
None => issues.push(format!("{key}: bundle header carries no such object")),
}
}
check_object(
&by_key,
&manifest.directory.key,
manifest.directory.length,
"index",
"sttd",
&mut issues,
);
for p in &manifest.packs {
check_object(&by_key, &p.key, p.length, "packs", "sttp", &mut issues);
}
if manifest.format_version == PACKED_FORMAT_VERSION_V2 {
for (key, kind) in std::iter::once((&manifest.directory.key, DIRECTORY_MAGIC))
.chain(manifest.packs.iter().map(|p| (&p.key, PACK_MAGIC)))
{
if let Some(bytes) = by_key.get(key.as_str()) {
if let Err(e) = strip_object_magic(bytes, kind, key) {
issues.push(e.to_string());
}
}
}
}
if let Some(dir_bytes) = by_key.get(manifest.directory.key.as_str()) {
match directory_codec_bytes(dir_bytes, manifest.format_version) {
Ok(codec) => {
match decode_directory_entries(codec, &manifest.directory) {
Ok(entries) => {
if let Some(max_pid) = entries.iter().map(|e| e.pack_id).max() {
if max_pid as usize >= manifest.packs.len() {
issues.push(format!(
"directory references pack_id {max_pid} but the manifest lists only {} pack(s)",
manifest.packs.len()
));
}
}
verify_v2_frame_template_refs(
&manifest,
&entries,
|pid| {
manifest
.packs
.get(pid)
.and_then(|p| by_key.get(p.key.as_str()))
.map(|b| b.to_vec())
},
&mut issues,
);
}
Err(e) => issues.push(format!("directory failed to decode: {e}")),
}
if manifest.directory.is_paged() {
match manifest.directory.root_length {
Some(rl) => {
let zstd = manifest.directory.encoding.as_deref()
== Some(DIRECTORY_ENCODING_ZSTD);
match crate::directory_page::verify_paged_structure(codec, rl, zstd) {
Ok(mut more) => issues.append(&mut more),
Err(e) => {
issues.push(format!("paged structure check failed: {e}"))
}
}
}
None => {
issues.push("paged directory: manifest missing rootLength".into())
}
}
}
}
Err(_) => {}
}
}
Ok(issues)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::arrow_tile::{
encode_tile, encode_tile_with, ColumnarLayer, EncoderConfig, GeometryColumn,
PropertyColumn, FORMAT_VERSION_V2,
};
fn point_layer(name: &str, ids: Vec<u64>, t0: i64) -> ColumnarLayer {
let n = ids.len();
ColumnarLayer {
name: name.to_string(),
feature_ids: ids,
start_times: vec![t0; n],
end_times: vec![t0 + 100; n],
geometry: GeometryColumn::Point(vec![[-122.4, 37.7]; n]),
vertex_times: None,
vertex_values: None,
triangles: None,
vertex_value_matrix: None,
properties: vec![],
}
}
fn distinct_tile(seed: u64) -> Vec<u8> {
let ids: Vec<u64> = (0..6).map(|i| seed * 100 + i).collect();
encode_tile(&[point_layer("default", ids, (seed as i64) * 7)]).unwrap()
}
fn v1_writer(out: &Path, ordering: BlobOrdering, pack_target_bytes: u64) -> PackWriter {
PackWriter::create(out, ordering, pack_target_bytes)
.unwrap()
.with_format_version(PACKED_FORMAT_VERSION_V1)
}
#[test]
fn blake3_128_hex_is_32_chars() {
let h = blake3_128_hex(b"hello");
assert_eq!(h.len(), 32);
assert!(h.chars().all(|c| c.is_ascii_hexdigit()));
}
#[test]
fn manifest_json_roundtrips() {
let m = Manifest {
format: PACKED_FORMAT.to_string(),
format_version: PACKED_FORMAT_VERSION,
capabilities: Vec::new(),
schemas: Vec::new(),
compression: "zstd".to_string(),
blob_ordering: None,
directory: DirectoryRef {
key: "index/abc.sttd".to_string(),
length: 42,
directory_version: 5,
encoding: Some(DIRECTORY_ENCODING_ZSTD.to_string()),
layout: None,
root_length: None,
page_count: None,
page_entries: None,
},
packs: vec![
PackRef { key: "packs/a.sttp".to_string(), length: 100 },
PackRef { key: "packs/b.sttp".to_string(), length: 200 },
],
metadata: Metadata::new("manifest-test"),
};
let bytes = m.to_json_bytes().unwrap();
let s = String::from_utf8(bytes.clone()).unwrap();
assert!(s.contains("\"formatVersion\""), "{s}");
assert!(s.contains("\"directoryVersion\""), "{s}");
assert!(s.contains("\"stt-packed\""), "{s}");
assert!(!s.contains("\"capabilities\""), "{s}");
assert!(!s.contains("\"blobOrdering\""), "{s}");
let back = Manifest::from_json_bytes(&bytes).unwrap();
assert_eq!(back.format, m.format);
assert_eq!(back.format_version, m.format_version);
assert_eq!(back.packs.len(), 2);
assert_eq!(back.directory.directory_version, 5);
assert_eq!(back.directory.encoding.as_deref(), Some("zstd"));
assert_eq!(back.metadata.name, "manifest-test");
let mut legacy_json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
legacy_json["directory"]
.as_object_mut()
.unwrap()
.remove("encoding");
let legacy_back =
Manifest::from_json_bytes(&serde_json::to_vec(&legacy_json).unwrap()).unwrap();
assert_eq!(legacy_back.directory.encoding, None);
assert!(legacy_back.capabilities.is_empty());
let legacy_out =
String::from_utf8(legacy_back.to_json_bytes().unwrap()).unwrap();
assert!(!legacy_out.contains("\"encoding\""), "{legacy_out}");
}
#[test]
fn packwriter_roundtrips_through_multiple_packs() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024);
let static_payload = encode_tile(&[point_layer("default", vec![7, 8, 9], 0)]).unwrap();
let mut expected: Vec<(TileId, i64, i64, Vec<u8>)> = Vec::new();
let bucket = 3_600_000i64;
for k in 0..30u64 {
let zoom = if k % 2 == 0 { 9u8 } else { 10u8 };
let b = (k % 5) as i64; let t = b * bucket;
let payload = if k == 13 || k == 27 {
static_payload.clone() } else {
distinct_tile(k)
};
let id = TileId::new(zoom, (k % 7) as u32, (k / 7) as u32, t as u64);
w.add_tile_full(&id, t, t + bucket - 1, Some(t), 6, Some(bucket as u64), &payload)
.unwrap();
expected.push((id, t, t + bucket - 1, payload));
}
assert_eq!(w.tile_count(), 30);
let meta = Metadata::new("packed-roundtrip").with_temporal_bucket_ms(bucket as u64);
let manifest = w.finalize(&meta).unwrap();
assert!(manifest.packs.len() > 1, "expected multiple packs, got {}", manifest.packs.len());
for p in &manifest.packs {
let bytes = fs::read(out.join(&p.key)).unwrap();
assert_eq!(bytes.len() as u64, p.length);
}
for p in &manifest.packs {
let bytes = fs::read(out.join(&p.key)).unwrap();
let hex = blake3_128_hex(&bytes);
assert_eq!(p.key, format!("packs/{hex}.sttp"));
}
let dir_bytes = fs::read(out.join(&manifest.directory.key)).unwrap();
assert_eq!(
manifest.directory.key,
format!("index/{}.sttd", blake3_128_hex(&dir_bytes))
);
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
let max_pack = reader.entries().iter().map(|e| e.pack_id).max().unwrap();
assert_eq!(max_pack as usize, manifest.packs.len() - 1);
let observed: std::collections::BTreeSet<u32> =
reader.entries().iter().map(|e| e.pack_id).collect();
for pid in 0..manifest.packs.len() as u32 {
assert!(observed.contains(&pid), "pack_id {pid} unused");
}
assert_eq!(reader.metadata().name, "packed-roundtrip");
assert_eq!(reader.metadata().temporal_bucket_ms, bucket as u64);
assert_eq!(reader.entries().len(), 30);
assert_eq!(manifest.metadata.tile_count, 30);
assert_eq!(manifest.metadata.feature_count, 30 * 6);
assert_eq!(reader.metadata().tile_count, 30);
assert_eq!(reader.metadata().feature_count, 30 * 6);
for (id, ts, _te, payload) in &expected {
let e = reader
.entries()
.iter()
.find(|e| {
e.zoom == id.z && e.x == id.x && e.y == id.y && e.time_start == *ts
})
.expect("entry present");
let got = reader.read_payload(e).unwrap();
assert_eq!(&got, payload, "payload mismatch for {id:?}");
}
let static_entries: Vec<&TileEntry> = reader
.entries()
.iter()
.filter(|e| {
let g = reader.read_payload(e).unwrap();
g == static_payload
})
.collect();
assert_eq!(static_entries.len(), 2);
assert_eq!(static_entries[0].pack_id, static_entries[1].pack_id);
assert_eq!(static_entries[0].offset, static_entries[1].offset);
}
#[test]
fn auto_resolves_from_occupied_extent_not_max_zoom() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("f1");
let mut w = v1_writer(&out, BlobOrdering::Auto, 1 << 20);
let bucket = 3_600_000i64;
for x in 0..4u32 {
for b in 0..64i64 {
let t = b * bucket;
let id = TileId::new(14, 10_000 + x, 20_000, t as u64);
let payload = format!("f1-{x}-{b}").into_bytes();
w.add_tile_full(&id, t, t + bucket - 1, Some(t), 1, None, &payload).unwrap();
}
}
let meta = Metadata::new("f1").with_temporal_bucket_ms(bucket as u64);
let manifest = w.finalize(&meta).unwrap();
assert_eq!(manifest.blob_ordering.as_deref(), Some("spatial"));
}
#[test]
fn measured_ordering_records_concrete_order() {
let build = || {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("m");
let mut w = v1_writer(&out, BlobOrdering::Auto, 1 << 20)
.with_measured_ordering(true);
let bucket = 3_600_000i64;
for x in 0..2u32 {
for b in 0..24i64 {
let t = b * bucket;
let id = TileId::new(10, 5_000 + x, 6_000, t as u64);
let payload = format!("m-{x}-{b}").into_bytes();
w.add_tile_full(&id, t, t + bucket - 1, Some(t), 1, None, &payload).unwrap();
}
}
let meta = Metadata::new("m").with_temporal_bucket_ms(bucket as u64);
w.finalize(&meta).unwrap().blob_ordering
};
let a = build();
let b = build();
assert_eq!(a, b, "measured resolution must be deterministic");
let o = a.expect("measured records a concrete order");
assert!(matches!(o.as_str(), "spatial" | "time-major" | "hilbert3" | "morton3"));
}
#[test]
fn paged_directory_writer_roundtrips_and_matches_single() {
let bucket = 3_600_000i64;
let mut input: Vec<(TileId, i64, i64, Vec<u8>)> = Vec::new();
for k in 0..120u64 {
let zoom = [6u8, 10, 13][(k % 3) as usize];
let b = (k % 4) as i64;
let t = b * bucket;
let id = TileId::new(zoom, (k % 11) as u32, (k / 11) as u32, t as u64);
input.push((id, t, t + bucket - 1, distinct_tile(k)));
}
let build = |out: &Path, page_entries: Option<usize>| -> Manifest {
let mut w = v1_writer(out, BlobOrdering::Auto, 16 * 1024)
.with_paging(page_entries);
for (id, ts, te, payload) in &input {
w.add_tile_full(id, *ts, *te, Some(*ts), 6, Some(bucket as u64), payload)
.unwrap();
}
let meta = Metadata::new("paged-roundtrip").with_temporal_bucket_ms(bucket as u64);
w.finalize(&meta).unwrap()
};
let dir = tempfile::tempdir().unwrap();
let single_out = dir.path().join("single");
let paged_out = dir.path().join("paged");
let single = build(&single_out, None);
let paged = build(&paged_out, Some(16));
assert!(single.directory.layout.is_none());
assert_eq!(paged.directory.layout.as_deref(), Some(DIRECTORY_LAYOUT_PAGED));
assert!(paged.directory.root_length.unwrap() > 0);
assert!(paged.directory.page_count.unwrap() >= 2, "expected multiple leaf pages");
assert_eq!(paged.directory.page_entries, Some(16));
assert_eq!(paged.directory.directory_version, crate::directory::DIRECTORY_VERSION);
assert_eq!(paged.directory.encoding.as_deref(), Some(DIRECTORY_ENCODING_ZSTD));
let r_single = PackedReader::open(single_out.join("manifest.json")).unwrap();
let r_paged = PackedReader::open(paged_out.join("manifest.json")).unwrap();
assert_eq!(r_paged.entries().len(), 120);
assert_eq!(r_single.entries().len(), 120);
assert_eq!(
single.packs.iter().map(|p| (&p.key, p.length)).collect::<Vec<_>>(),
paged.packs.iter().map(|p| (&p.key, p.length)).collect::<Vec<_>>(),
"pack content addresses must match across single vs paged builds"
);
assert_eq!(r_single.entries(), r_paged.entries());
for (id, ts, _te, payload) in &input {
let e = r_paged
.entries()
.iter()
.find(|x| x.zoom == id.z && x.x == id.x && x.y == id.y && x.time_start == *ts)
.expect("paged entry present");
assert_eq!(&r_paged.read_payload(e).unwrap(), payload, "payload mismatch {id:?}");
}
let issues = verify_packed_objects(paged_out.join("manifest.json")).unwrap();
assert!(issues.is_empty(), "paged verify issues: {issues:?}");
}
#[test]
fn oversized_blob_gets_its_own_pack() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let big_ids: Vec<u64> = (0..4000).collect();
let big = encode_tile(&[ColumnarLayer {
name: "default".to_string(),
feature_ids: big_ids.clone(),
start_times: vec![0; big_ids.len()],
end_times: vec![100; big_ids.len()],
geometry: GeometryColumn::Point(
(0..big_ids.len()).map(|i| [i as f64 * 0.01, i as f64 * 0.013]).collect(),
),
vertex_times: None,
vertex_values: None,
triangles: None,
vertex_value_matrix: None,
properties: vec![],
}])
.unwrap();
let big_compressed_len = compression::compress_zstd_with_dict(&big, None).unwrap().len();
let mut w = v1_writer(&out, BlobOrdering::SpatialMajor, 4 * 1024);
assert!(big_compressed_len as u64 > 4 * 1024);
w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 4000, None, &big).unwrap();
for k in 1..4u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p).unwrap();
}
let _manifest = w.finalize(&Metadata::new("big")).unwrap();
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
let big_entry = reader.entries().iter().find(|e| e.x == 0).unwrap();
assert_eq!(reader.read_payload(big_entry).unwrap(), big);
}
#[test]
fn corrupt_pack_blob_is_detected_on_read() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 64 * 1024 * 1024);
for k in 0..4u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p).unwrap();
}
let manifest = w.finalize(&Metadata::new("crc")).unwrap();
let entry = PackedReader::open(out.join("manifest.json")).unwrap().entries()[0].clone();
assert!(PackedReader::open(out.join("manifest.json"))
.unwrap()
.read_payload(&entry)
.is_ok());
let pack_path = out.join(&manifest.packs[entry.pack_id as usize].key);
let mut bytes = fs::read(&pack_path).unwrap();
bytes[entry.offset as usize] ^= 0xff;
fs::write(&pack_path, &bytes).unwrap();
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
assert!(
reader.read_payload(&entry).is_err(),
"corrupt pack blob must fail the read-path CRC32C check"
);
}
#[test]
fn rebuilds_are_byte_reproducible() {
let bucket = 3_600_000i64;
let mut tiles: Vec<(TileId, i64, Option<u64>, Vec<u8>)> = Vec::new();
for k in 0..10u64 {
let t = (k % 4) as i64 * bucket;
tiles.push((
TileId::new(9, (k % 5) as u32, (k / 5) as u32, t as u64),
t,
None,
distinct_tile(k),
));
}
tiles.push((TileId::new(9, 1, 0, 0), 0, None, distinct_tile(100)));
tiles.push((
TileId::new(9, 1, 0, 0),
0,
Some(24 * bucket as u64),
distinct_tile(101),
));
let meta = Metadata::new("repro").with_temporal_bucket_ms(bucket as u64);
let build = |order: &[usize]| {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024);
for &i in order {
let (id, t, b, payload) = &tiles[i];
w.add_tile_full(id, *t, t + bucket - 1, Some(*t), 6, *b, payload)
.unwrap();
}
let manifest = w.finalize(&meta).unwrap();
(dir, manifest)
};
let forward: Vec<usize> = (0..tiles.len()).collect();
let reverse: Vec<usize> = (0..tiles.len()).rev().collect();
let (_d1, m1) = build(&forward);
let (_d2, m2) = build(&reverse);
assert_eq!(m1.directory.key, m2.directory.key, "directory hash must be stable");
assert_eq!(
m1.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
m2.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
"pack hashes must be stable across rebuilds"
);
assert_eq!(m1.to_json_bytes().unwrap(), m2.to_json_bytes().unwrap());
}
#[test]
fn spilled_build_is_byte_identical_to_in_memory() {
let bucket = 3_600_000i64;
let mut tiles: Vec<(TileId, i64, Option<u64>, Vec<u8>)> = Vec::new();
for k in 0..40u64 {
let t = (k % 4) as i64 * bucket;
tiles.push((
TileId::new(9, (k % 8) as u32, (k / 8) as u32, t as u64),
t,
None,
distinct_tile(k),
));
}
tiles.push((TileId::new(9, 7, 7, 0), 0, None, distinct_tile(3)));
let meta = Metadata::new("spill").with_temporal_bucket_ms(bucket as u64);
let build = |budget: u64| {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024).with_memory_budget(budget);
for (id, t, b, payload) in &tiles {
w.add_tile_full(id, *t, t + bucket - 1, Some(*t), 6, *b, payload)
.unwrap();
}
let manifest = w.finalize(&meta).unwrap();
let residue: Vec<_> = fs::read_dir(&out)
.unwrap()
.filter_map(|e| e.ok())
.filter(|e| e.file_name().to_string_lossy().starts_with(".spill-"))
.collect();
assert!(residue.is_empty(), "spill file left behind: {residue:?}");
(dir, manifest)
};
let (d1, spilled) = build(1024);
let (_d2, in_mem) = build(0);
assert_eq!(
spilled.directory.key, in_mem.directory.key,
"directory hash must not depend on the payload storage medium"
);
assert_eq!(
spilled.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
in_mem.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
"pack hashes must not depend on the payload storage medium"
);
assert_eq!(spilled.to_json_bytes().unwrap(), in_mem.to_json_bytes().unwrap());
assert!(spilled.packs.len() > 1, "fixture must span multiple packs");
let spilled_out = d1.path().join("dataset");
assert!(verify_packed_objects(spilled_out.join("manifest.json")).unwrap().is_empty());
let tmp_residue: Vec<_> = fs::read_dir(spilled_out.join("packs"))
.unwrap()
.filter_map(|e| e.ok())
.filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
.collect();
assert!(tmp_residue.is_empty(), "streaming pack temp left behind: {tmp_residue:?}");
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
{
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024).with_memory_budget(1);
for (id, t, b, payload) in tiles.iter().take(4) {
w.add_tile_full(id, *t, t + bucket - 1, Some(*t), 6, *b, payload)
.unwrap();
}
assert!(fs::read_dir(&out)
.unwrap()
.filter_map(|e| e.ok())
.any(|e| e.file_name().to_string_lossy().starts_with(".spill-")));
}
assert!(!fs::read_dir(&out)
.unwrap()
.filter_map(|e| e.ok())
.any(|e| e.file_name().to_string_lossy().starts_with(".spill-")));
}
#[test]
fn payload_len_guard_rejects_4gib_payloads() {
let id = TileId::new(5, 1, 2, 0);
assert!(check_payload_len(&id, 0).is_ok());
assert!(check_payload_len(&id, u32::MAX as u64).is_ok(), "exact u32::MAX still fits");
let err = check_payload_len(&id, u32::MAX as u64 + 1).unwrap_err();
assert!(
err.to_string().contains("uncompressed_size") && err.to_string().contains("4 GiB"),
"got: {err}"
);
assert!(check_payload_len(&id, u64::MAX).is_err());
}
#[test]
fn add_tile_full_rejects_mixed_frame_and_writer_versions() {
let dir = tempfile::tempdir().unwrap();
let id = TileId::new(10, 0, 0, 0);
let v1_payload = encode_tile_with(
&[point_layer("default", vec![1, 2], 0)],
&EncoderConfig::default(),
)
.unwrap();
let v2_payload = encode_tile_with(
&[point_layer("default", vec![1, 2], 0)],
&EncoderConfig {
format_version: FORMAT_VERSION_V2,
..EncoderConfig::default()
},
)
.unwrap();
let mut w2 = PackWriter::create(dir.path().join("v2"), BlobOrdering::Auto, 8 * 1024)
.unwrap();
let err = w2
.add_tile_full(&id, 0, 100, None, 2, None, &v1_payload)
.unwrap_err();
assert!(
err.to_string().contains("v1 layer frame")
&& err.to_string().contains("formatVersion-2"),
"got: {err}"
);
let mut w1 = v1_writer(&dir.path().join("v1"), BlobOrdering::Auto, 8 * 1024);
let err = w1
.add_tile_full(&id, 0, 100, None, 2, None, &v2_payload)
.unwrap_err();
assert!(
err.to_string().contains("v2 layer frame")
&& err.to_string().contains("formatVersion-1"),
"got: {err}"
);
w2.add_tile_full(&id, 0, 100, None, 2, None, &v2_payload).unwrap();
w1.add_tile_full(&id, 0, 100, None, 2, None, &v1_payload).unwrap();
}
#[test]
fn directory_encoding_compressed_and_raw_both_read() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024)
.unwrap()
.with_format_version(PACKED_FORMAT_VERSION_V1);
for k in 0..8u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p)
.unwrap();
}
let manifest = w.finalize(&Metadata::new("dir-enc")).unwrap();
let manifest_path = out.join("manifest.json");
assert_eq!(manifest.directory.encoding.as_deref(), Some("zstd"));
let at_rest = fs::read(out.join(&manifest.directory.key)).unwrap();
assert_eq!(at_rest.len() as u64, manifest.directory.length);
let raw = compression::decompress_zstd_with_dict(&at_rest, None).unwrap();
assert!(crate::directory::decode_directory(&raw).is_ok());
assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
let entries_compressed = PackedReader::open(&manifest_path).unwrap().entries().to_vec();
let raw_hex = blake3_128_hex(&raw);
let raw_rel = format!("index/{raw_hex}.sttd");
fs::write(out.join(&raw_rel), &raw).unwrap();
let mut legacy = manifest.clone();
legacy.directory = DirectoryRef {
key: raw_rel,
length: raw.len() as u64,
directory_version: crate::directory::DIRECTORY_VERSION,
encoding: None,
layout: None,
root_length: None,
page_count: None,
page_entries: None,
};
fs::write(&manifest_path, legacy.to_json_bytes().unwrap()).unwrap();
assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
let entries_raw = PackedReader::open(&manifest_path).unwrap().entries().to_vec();
assert_eq!(entries_raw, entries_compressed);
legacy.directory.encoding = Some("br".to_string());
fs::write(&manifest_path, legacy.to_json_bytes().unwrap()).unwrap();
assert!(PackedReader::open(&manifest_path).is_err());
}
#[test]
fn unknown_format_version_is_rejected_at_open() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024);
w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 6, None, &distinct_tile(1))
.unwrap();
w.finalize(&Metadata::new("ver")).unwrap();
let manifest_path = out.join("manifest.json");
assert!(PackedReader::open(&manifest_path).is_ok());
let mut v: serde_json::Value =
serde_json::from_slice(&fs::read(&manifest_path).unwrap()).unwrap();
v["formatVersion"] = serde_json::json!(3);
fs::write(&manifest_path, serde_json::to_vec(&v).unwrap()).unwrap();
let err = match PackedReader::open(&manifest_path) {
Ok(_) => panic!("formatVersion 3 must be rejected at open"),
Err(e) => e,
};
assert!(err.to_string().contains("formatVersion"), "unexpected error: {err}");
}
#[test]
fn v1_dataset_doctored_to_claim_v2_is_rejected_at_open() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024)
.unwrap()
.with_format_version(PACKED_FORMAT_VERSION_V1);
w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 6, None, &distinct_tile(1))
.unwrap();
w.finalize(&Metadata::new("ver")).unwrap();
let manifest_path = out.join("manifest.json");
let mut v: serde_json::Value =
serde_json::from_slice(&fs::read(&manifest_path).unwrap()).unwrap();
v["formatVersion"] = serde_json::json!(2);
fs::write(&manifest_path, serde_json::to_vec(&v).unwrap()).unwrap();
let err = match PackedReader::open(&manifest_path) {
Ok(_) => panic!("v2-declared manifest over magic-less v1 objects must be rejected"),
Err(e) => e,
};
assert!(err.to_string().contains("STTD"), "unexpected error: {err}");
}
#[test]
fn writer_declares_capabilities_canonically() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024)
.with_capabilities(vec![
CAPABILITY_ELEVATION_FOLD.to_string(),
CAPABILITY_COORD_QUANT.to_string(),
CAPABILITY_COORD_QUANT.to_string(),
]);
w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 6, None, &distinct_tile(1))
.unwrap();
let manifest = w.finalize(&Metadata::new("caps")).unwrap();
assert_eq!(
manifest.capabilities,
vec![CAPABILITY_COORD_QUANT.to_string(), CAPABILITY_ELEVATION_FOLD.to_string()]
);
let manifest_path = out.join("manifest.json");
let s = fs::read_to_string(&manifest_path).unwrap();
assert!(s.contains("\"capabilities\""), "{s}");
assert!(PackedReader::open(&manifest_path).is_ok());
}
#[test]
fn unknown_capability_is_rejected_at_open() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024);
w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 6, None, &distinct_tile(1))
.unwrap();
w.finalize(&Metadata::new("caps")).unwrap();
let manifest_path = out.join("manifest.json");
let mut v: serde_json::Value =
serde_json::from_slice(&fs::read(&manifest_path).unwrap()).unwrap();
v["capabilities"] = serde_json::json!([CAPABILITY_COORD_QUANT, "from-the-future"]);
fs::write(&manifest_path, serde_json::to_vec(&v).unwrap()).unwrap();
let err = match PackedReader::open(&manifest_path) {
Ok(_) => panic!("unknown capability must be rejected at open"),
Err(e) => e,
};
let msg = err.to_string();
assert!(
msg.contains("does not implement: from-the-future ("),
"unexpected error: {msg}"
);
}
#[test]
fn verify_packed_objects_clean_then_detects_corruption() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = v1_writer(&out, BlobOrdering::Auto, 8 * 1024);
for k in 0..12u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p)
.unwrap();
}
let manifest = w.finalize(&Metadata::new("verify")).unwrap();
let manifest_path = out.join("manifest.json");
assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
let pack0 = out.join(&manifest.packs[0].key);
let mut bytes = fs::read(&pack0).unwrap();
bytes[0] ^= 0xff;
fs::write(&pack0, &bytes).unwrap();
let issues = verify_packed_objects(&manifest_path).unwrap();
assert!(
issues.iter().any(|s| s.contains("content-address mismatch")),
"expected a content-address mismatch, got {issues:?}"
);
}
fn build_bundle_fixture(out: &Path) -> Manifest {
let mut w = PackWriter::create(out, BlobOrdering::Auto, 8 * 1024)
.unwrap()
.with_format_version(PACKED_FORMAT_VERSION_V1);
let bucket = 3_600_000i64;
let static_payload = encode_tile(&[point_layer("default", vec![7, 8, 9], 0)]).unwrap();
for k in 0..20u64 {
let t = (k % 4) as i64 * bucket;
let payload = if k == 3 || k == 17 {
static_payload.clone() } else {
distinct_tile(k)
};
let id = TileId::new(10, (k % 5) as u32, (k / 5) as u32, t as u64);
w.add_tile_full(&id, t, t + bucket - 1, Some(t), 6, Some(bucket as u64), &payload)
.unwrap();
}
let meta = Metadata::new("bundle-fixture").with_temporal_bucket_ms(bucket as u64);
w.finalize(&meta).unwrap()
}
#[test]
fn bundle_pack_unpack_roundtrips_byte_identical() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("dataset");
let manifest = build_bundle_fixture(&src);
assert!(manifest.packs.len() > 1, "fixture should span multiple packs");
let manifest_path = src.join("manifest.json");
let bundle_a = dir.path().join("a.sttb");
let bundle_b = dir.path().join("b.sttb");
let summary = write_bundle(&manifest_path, &bundle_a).unwrap();
write_bundle(&manifest_path, &bundle_b).unwrap();
let bytes_a = fs::read(&bundle_a).unwrap();
assert_eq!(bytes_a, fs::read(&bundle_b).unwrap(), "bundling must be deterministic");
assert_eq!(bytes_a.len() as u64, summary.bytes);
assert_eq!(summary.objects, 1 + manifest.packs.len());
assert_eq!(&bytes_a[0..4], b"STTB");
assert_eq!(bytes_a[4], BUNDLE_VERSION);
assert_eq!(&bytes_a[5..8], &[0, 0, 0]);
let (header, header_end) = parse_bundle_header(&bytes_a).unwrap();
assert_eq!(header.objects.len(), summary.objects);
assert_eq!(header.objects[0].key, manifest.directory.key, "directory first");
for (o, p) in header.objects[1..].iter().zip(&manifest.packs) {
assert_eq!(o.key, p.key, "packs in pack_id order");
}
for o in &header.objects {
assert_eq!(o.offset % 8, 0, "object {} not 8-aligned", o.key);
assert!(o.offset >= header_end);
}
assert_eq!(
header.manifest.get().as_bytes(),
fs::read(&manifest_path).unwrap(),
"header manifest must be the manifest.json bytes verbatim"
);
let out = dir.path().join("unpacked");
let back = unpack_bundle(&bundle_a, &out).unwrap();
assert_eq!(back.objects, summary.objects);
assert!(
verify_packed_objects(out.join("manifest.json")).unwrap().is_empty(),
"unpacked dataset must re-verify its content addresses"
);
assert_eq!(
fs::read(out.join("manifest.json")).unwrap(),
fs::read(&manifest_path).unwrap()
);
for key in std::iter::once(&manifest.directory.key)
.chain(manifest.packs.iter().map(|p| &p.key))
{
assert_eq!(
fs::read(out.join(key)).unwrap(),
fs::read(src.join(key)).unwrap(),
"object {key} must round-trip byte-identical"
);
}
assert!(verify_bundle_objects(&bundle_a).unwrap().is_empty());
}
#[test]
fn open_bundle_reads_tiles_byte_identical_to_exploded_dir() {
for paged in [false, true] {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("dataset");
let mut w = v1_writer(&src, BlobOrdering::Auto, 8 * 1024)
.with_paging(paged.then_some(4));
for k in 0..20u64 {
w.add_tile_full(
&TileId::new(10, (k % 5) as u32, (k / 5) as u32, 0),
0,
100,
None,
6,
None,
&distinct_tile(k),
)
.unwrap();
}
w.finalize(&Metadata::new("bundle-read")).unwrap();
let bundle = dir.path().join("dataset.sttb");
write_bundle(src.join("manifest.json"), &bundle).unwrap();
let exploded = PackedReader::open(src.join("manifest.json")).unwrap();
let bundled = PackedReader::open_bundle(&bundle).unwrap();
assert_eq!(exploded.entries(), bundled.entries(), "paged={paged}");
assert_eq!(exploded.metadata().name, bundled.metadata().name);
for (e_dir, e_bun) in exploded.entries().iter().zip(bundled.entries()) {
assert_eq!(
exploded.read_payload(e_dir).unwrap(),
bundled.read_payload(e_bun).unwrap(),
"payload mismatch (paged={paged}) for {:?}",
e_dir.tile_id()
);
}
}
}
#[test]
fn corrupt_bundle_blob_is_detected() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("dataset");
build_bundle_fixture(&src);
let bundle = dir.path().join("dataset.sttb");
write_bundle(src.join("manifest.json"), &bundle).unwrap();
assert!(verify_bundle_objects(&bundle).unwrap().is_empty());
let bytes = fs::read(&bundle).unwrap();
let (header, _) = parse_bundle_header(&bytes).unwrap();
let pack0 = header
.objects
.iter()
.find(|o| o.key.starts_with("packs/"))
.expect("a pack object");
let mut corrupt = bytes.clone();
corrupt[pack0.offset as usize] ^= 0xff;
fs::write(&bundle, &corrupt).unwrap();
let issues = verify_bundle_objects(&bundle).unwrap();
assert!(
issues.iter().any(|s| s.contains("content-address mismatch")),
"expected a content-address mismatch, got {issues:?}"
);
let reader = PackedReader::open_bundle(&bundle).unwrap();
let bad = reader
.entries()
.iter()
.find(|e| reader.read_payload(e).is_err());
assert!(bad.is_some(), "some tile in the corrupted pack must fail its CRC");
}
#[test]
fn truncated_and_doctored_bundles_error_loudly() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("dataset");
build_bundle_fixture(&src);
let bundle = dir.path().join("dataset.sttb");
write_bundle(src.join("manifest.json"), &bundle).unwrap();
let good = fs::read(&bundle).unwrap();
let open_err = |bytes: &[u8]| {
let p = dir.path().join("bad.sttb");
fs::write(&p, bytes).unwrap();
assert!(PackedReader::open_bundle(&p).is_err(), "must reject {} bytes", bytes.len());
assert!(unpack_bundle(&p, dir.path().join("bad-out")).is_err());
};
open_err(&good[..5]);
let mut bad = good.clone();
bad[0] = b'X';
open_err(&bad);
let mut bad = good.clone();
bad[4] = 9;
open_err(&bad);
let mut bad = good.clone();
bad[5] = 1;
open_err(&bad);
let mut bad = good.clone();
bad[8..12].copy_from_slice(&u32::MAX.to_le_bytes());
open_err(&bad);
open_err(&good[..good.len() - 32]);
let mut bad = good.clone();
let hl = u32::from_le_bytes(bad[8..12].try_into().unwrap()) as usize;
bad[12..12 + hl].fill(b'!');
open_err(&bad);
let evil_header = r#"{"manifest":{"format":"stt-packed"},"objects":[{"key":"../evil","offset":128,"length":0}]}"#;
let mut evil = Vec::new();
evil.extend_from_slice(&BUNDLE_MAGIC);
evil.extend_from_slice(&[BUNDLE_VERSION, 0, 0, 0]);
evil.extend_from_slice(&(evil_header.len() as u32).to_le_bytes());
evil.extend_from_slice(evil_header.as_bytes());
evil.resize(256, 0);
let p = dir.path().join("evil.sttb");
fs::write(&p, &evil).unwrap();
let out = dir.path().join("evil-out");
let err = unpack_bundle(&p, &out).unwrap_err();
assert!(err.to_string().contains("unsafe bundle object key"), "got: {err}");
assert!(!dir.path().join("evil-out").exists(), "nothing may be written");
assert!(!dir.path().join("evil").exists(), "traversal target must not exist");
}
fn v2_point_layer(seed: u64, n: usize) -> ColumnarLayer {
let base = 1_700_000_000_000i64 + seed as i64 * 60_000;
ColumnarLayer {
name: "default".to_string(),
feature_ids: (0..n as u64).map(|i| seed * 100 + i).collect(),
start_times: (0..n as i64).map(|i| base + i * 1000).collect(),
end_times: (0..n as i64).map(|i| base + i * 1000 + 500).collect(),
geometry: GeometryColumn::Point(
(0..n).map(|i| [-122.4 + seed as f64 * 0.01, 37.7 + i as f64 * 1e-4]).collect(),
),
vertex_times: None,
vertex_values: None,
triangles: None,
vertex_value_matrix: None,
properties: vec![
(
"speed".to_string(),
PropertyColumn::Numeric(
(0..n).map(|i| Some(seed as f64 * 3.0 + i as f64)).collect(),
),
),
(
"kind".to_string(),
PropertyColumn::Categorical(
(0..n).map(|i| Some(["car", "bus"][i % 2].to_string())).collect(),
),
),
],
}
}
fn v2_fixture_layers() -> Vec<ColumnarLayer> {
let mut layers: Vec<ColumnarLayer> = (0..5).map(|k| v2_point_layer(k, 4)).collect();
layers.push(v2_point_layer(99, 0)); layers
}
fn v2_encoder_cfg(w: &PackWriter) -> EncoderConfig {
EncoderConfig {
quantize_coords_m: Some(1.0),
quantize_attrs_auto: true,
format_version: FORMAT_VERSION_V2,
template_collector: Some(w.template_collector()),
..EncoderConfig::default()
}
}
fn build_v2_dataset(out: &Path, paging: Option<usize>) -> (Manifest, Vec<ColumnarLayer>) {
let mut w = PackWriter::create(out, BlobOrdering::Auto, 8 * 1024)
.unwrap()
.with_paging(paging)
.with_capabilities(vec![
CAPABILITY_COORD_QUANT.to_string(),
CAPABILITY_ATTR_QUANT.to_string(),
]);
let cfg = v2_encoder_cfg(&w);
let layers = v2_fixture_layers();
for (k, layer) in layers.iter().enumerate() {
let payload = encode_tile_with(std::slice::from_ref(layer), &cfg).unwrap();
w.add_tile_full(
&TileId::new(10, k as u32, 0, 0),
0,
100,
None,
layer.feature_count() as u32,
None,
&payload,
)
.unwrap();
}
let manifest = w.finalize(&Metadata::new("v2-fixture")).unwrap();
(manifest, layers)
}
#[test]
fn v2_dataset_roundtrips_with_templates_magic_and_v1_equivalence() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let (manifest, layers) = build_v2_dataset(&out, None);
assert_eq!(manifest.format_version, PACKED_FORMAT_VERSION_V2);
assert_eq!(manifest.schemas.len(), 2, "expected core+props templates");
let hashes: Vec<&str> = manifest.schemas.iter().map(|s| s.hash.as_str()).collect();
assert!(hashes.windows(2).all(|w| w[0] < w[1]), "schemas sorted by hash");
let dir_bytes = fs::read(out.join(&manifest.directory.key)).unwrap();
assert_eq!(&dir_bytes[0..8], &object_magic(DIRECTORY_MAGIC));
assert_eq!(
manifest.directory.key,
format!("index/{}.sttd", blake3_128_hex(&dir_bytes))
);
for p in &manifest.packs {
let bytes = fs::read(out.join(&p.key)).unwrap();
assert_eq!(&bytes[0..8], &object_magic(PACK_MAGIC));
assert_eq!(p.key, format!("packs/{}.sttp", blake3_128_hex(&bytes)));
}
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
assert_eq!(reader.format_version(), PACKED_FORMAT_VERSION_V2);
assert!(reader.templates().is_some());
for e in reader.entries() {
assert!(e.offset >= OBJECT_MAGIC_LEN as u64, "offset {} inside magic", e.offset);
}
assert!(verify_packed_objects(out.join("manifest.json")).unwrap().is_empty());
for (k, layer) in layers.iter().enumerate() {
let e = reader
.entries()
.iter()
.find(|e| e.x == k as u32)
.expect("entry present");
let got = reader.read_layers(e).unwrap();
let v1 = crate::arrow_tile::decode_tile(
&encode_tile_with(
std::slice::from_ref(layer),
&EncoderConfig {
quantize_coords_m: Some(1.0),
quantize_attrs_auto: true,
..EncoderConfig::default()
},
)
.unwrap(),
)
.unwrap();
assert_eq!(got.len(), v1.len());
assert_eq!(got[0].name, v1[0].name);
assert_eq!(
got[0].batch, v1[0].batch,
"tile {k}: v2 decode must equal the v1 decode after re-injection"
);
}
}
#[test]
fn mixed_frame_and_manifest_versions_hard_error() {
let dir = tempfile::tempdir().unwrap();
let v2_out = dir.path().join("v2");
build_v2_dataset(&v2_out, None);
let v2_reader = PackedReader::open(v2_out.join("manifest.json")).unwrap();
let v1_payload = encode_tile(&[point_layer("default", vec![1, 2], 0)]).unwrap();
let err = v2_reader.decode_payload(&v1_payload).unwrap_err();
assert!(err.to_string().contains("v1 layer frame"), "got: {err}");
let v1_out = dir.path().join("v1");
let mut w = PackWriter::create(&v1_out, BlobOrdering::Auto, 8 * 1024)
.unwrap()
.with_format_version(PACKED_FORMAT_VERSION_V1);
w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 6, None, &distinct_tile(1))
.unwrap();
w.finalize(&Metadata::new("v1")).unwrap();
let v1_reader = PackedReader::open(v1_out.join("manifest.json")).unwrap();
let v2_payload = encode_tile_with(
&[v2_point_layer(0, 3)],
&EncoderConfig {
format_version: FORMAT_VERSION_V2,
..EncoderConfig::default()
},
)
.unwrap();
let err = v1_reader.decode_payload(&v2_payload).unwrap_err();
assert!(err.to_string().contains("v2 layer frame"), "got: {err}");
}
#[test]
fn corrupt_manifest_schema_template_fails_open() {
use base64::Engine as _;
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
build_v2_dataset(&out, None);
let manifest_path = out.join("manifest.json");
let mut v: serde_json::Value =
serde_json::from_slice(&fs::read(&manifest_path).unwrap()).unwrap();
let tampered = base64::engine::general_purpose::STANDARD.encode(b"not a template");
v["schemas"][0]["data"] = serde_json::json!(tampered);
fs::write(&manifest_path, serde_json::to_vec(&v).unwrap()).unwrap();
let err = match PackedReader::open(&manifest_path) {
Ok(_) => panic!("corrupt template must fail the open"),
Err(e) => e,
};
assert!(err.to_string().contains("hash"), "got: {err}");
let issues = verify_packed_objects(&manifest_path).unwrap();
assert!(
issues.iter().any(|i| i.contains("hash")),
"verify must report the schema corruption: {issues:?}"
);
}
#[test]
fn corrupt_v2_object_magic_is_detected() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let (manifest, _) = build_v2_dataset(&out, None);
let manifest_path = out.join("manifest.json");
let pack0 = out.join(&manifest.packs[0].key);
let mut bytes = fs::read(&pack0).unwrap();
bytes[0] ^= 0xff;
fs::write(&pack0, &bytes).unwrap();
let reader = PackedReader::open(&manifest_path).unwrap();
let entry = reader.entries().iter().find(|e| e.pack_id == 0).unwrap();
let err = reader.read_payload(entry).unwrap_err();
assert!(err.to_string().contains("STTP"), "got: {err}");
let issues = verify_packed_objects(&manifest_path).unwrap();
assert!(issues.iter().any(|i| i.contains("STTP")), "{issues:?}");
assert!(
issues.iter().any(|i| i.contains("content-address mismatch")),
"{issues:?}"
);
fs::write(&pack0, {
let mut b = fs::read(&pack0).unwrap();
b[0] ^= 0xff;
b
})
.unwrap();
let dir_path = out.join(&manifest.directory.key);
let mut bytes = fs::read(&dir_path).unwrap();
bytes[0] ^= 0xff;
fs::write(&dir_path, &bytes).unwrap();
let err = match PackedReader::open(&manifest_path) {
Ok(_) => panic!("corrupt directory magic must fail the open"),
Err(e) => e,
};
assert!(err.to_string().contains("STTD"), "got: {err}");
}
#[test]
fn v2_paged_directory_roundtrips() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let (manifest, layers) = build_v2_dataset(&out, Some(2));
assert_eq!(manifest.directory.layout.as_deref(), Some(DIRECTORY_LAYOUT_PAGED));
assert!(verify_packed_objects(out.join("manifest.json")).unwrap().is_empty());
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
assert_eq!(reader.entries().len(), layers.len());
for e in reader.entries() {
assert!(reader.read_layers(e).is_ok());
}
}
#[test]
fn v2_bundle_roundtrips_and_verifies() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("dataset");
let (_, layers) = build_v2_dataset(&src, None);
let bundle = dir.path().join("dataset.sttb");
write_bundle(src.join("manifest.json"), &bundle).unwrap();
assert!(verify_bundle_objects(&bundle).unwrap().is_empty());
let exploded = PackedReader::open(src.join("manifest.json")).unwrap();
let bundled = PackedReader::open_bundle(&bundle).unwrap();
assert_eq!(bundled.format_version(), PACKED_FORMAT_VERSION_V2);
assert_eq!(exploded.entries(), bundled.entries());
for (e_dir, e_bun) in exploded.entries().iter().zip(bundled.entries()) {
let a = exploded.read_layers(e_dir).unwrap();
let b = bundled.read_layers(e_bun).unwrap();
assert_eq!(a.len(), b.len());
for (x, y) in a.iter().zip(&b) {
assert_eq!(x.batch, y.batch);
}
}
assert_eq!(bundled.entries().len(), layers.len());
let unpacked = dir.path().join("unpacked");
unpack_bundle(&bundle, &unpacked).unwrap();
assert!(
verify_packed_objects(unpacked.join("manifest.json")).unwrap().is_empty(),
"unpacked v2 dataset must re-verify"
);
}
#[test]
fn v2_verbatim_repack_roundtrips_readable() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("src");
let (src_manifest, layers) = build_v2_dataset(&src, None);
let reader = PackedReader::open(src.join("manifest.json")).unwrap();
let out = dir.path().join("repacked");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024)
.unwrap()
.with_format_version(reader.format_version())
.with_capabilities(reader.capabilities().to_vec())
.with_seeded_templates(reader.templates().expect("v2 source has a registry"));
for e in reader.entries() {
let payload = reader.read_payload(e).unwrap();
w.add_tile_full(
&TileId::new(e.zoom, e.x, e.y, e.time_start.max(0) as u64),
e.time_start,
e.time_end,
e.cover_t_min,
e.feature_count,
e.temporal_bucket_ms,
&payload,
)
.unwrap();
}
let manifest = w.finalize(reader.metadata()).unwrap();
assert_eq!(manifest.format_version, PACKED_FORMAT_VERSION_V2);
assert_eq!(
manifest.schemas.iter().map(|s| &s.hash).collect::<Vec<_>>(),
src_manifest.schemas.iter().map(|s| &s.hash).collect::<Vec<_>>(),
"seeded schemas must match the source's table"
);
assert!(verify_packed_objects(out.join("manifest.json")).unwrap().is_empty());
let re = PackedReader::open(out.join("manifest.json")).unwrap();
assert_eq!(re.entries().len(), layers.len());
for e in re.entries() {
re.read_layers(e)
.unwrap_or_else(|err| panic!("repacked tile {:?} must decode: {err}", e.tile_id()));
}
}
#[test]
fn verify_flags_missing_frame_referenced_schema_template() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let (manifest, _) = build_v2_dataset(&out, None);
assert_eq!(manifest.schemas.len(), 2, "core + props templates");
let manifest_path = out.join("manifest.json");
assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
let mut v: serde_json::Value =
serde_json::from_slice(&fs::read(&manifest_path).unwrap()).unwrap();
let removed = v["schemas"].as_array_mut().unwrap().remove(0);
let removed_hash = removed["hash"].as_str().unwrap().to_string();
fs::write(&manifest_path, serde_json::to_vec(&v).unwrap()).unwrap();
let issues = verify_packed_objects(&manifest_path).unwrap();
assert!(
issues
.iter()
.any(|i| i.contains(&removed_hash) && i.contains("manifest.schemas")),
"expected the missing template hash {removed_hash} to be reported: {issues:?}"
);
let bundle = dir.path().join("dataset.sttb");
write_bundle(&manifest_path, &bundle).unwrap();
let issues = verify_bundle_objects(&bundle).unwrap();
assert!(
issues.iter().any(|i| i.contains(&removed_hash)),
"bundle verify must report the missing template: {issues:?}"
);
}
}