use crate::config::Compression;
use crate::fetch::ObjectEntry;
use crate::framer::Codec;
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use serde::{Deserialize, Serialize};
use sha2::{Digest as _, Sha256};
use spate_core::coordination::{CoordinationError, CoordinationErrorKind, SplitId};
use std::collections::VecDeque;
use std::rc::Rc;
pub const DESCRIPTOR_VERSION: u32 = 2;
pub(crate) const PACKING_VERSION: u32 = 2;
pub(crate) const MAX_SUBDIVIDABLE_BYTES: u64 = 50_000 << 30;
pub(crate) const PACKING_LOOKBACK: usize = 10;
pub(crate) const OPEN_COST_DIVISOR: u64 = 16;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DescriptorObject {
pub key: String,
pub size: u64,
pub etag: Option<String>,
pub last_modified_ms: i64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct SplitRange {
pub start: u64,
pub end: u64,
pub delimiter: u8,
}
impl SplitRange {
#[must_use]
pub fn new(start: u64, end: u64, delimiter: u8) -> SplitRange {
SplitRange {
start,
end,
delimiter,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SplitDescriptor {
pub(crate) v: u32,
pub objects: Vec<DescriptorObject>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub range: Option<SplitRange>,
}
#[derive(Deserialize)]
struct VersionProbe {
v: u32,
}
impl SplitDescriptor {
#[must_use]
pub fn new(objects: Vec<DescriptorObject>) -> SplitDescriptor {
SplitDescriptor {
v: DESCRIPTOR_VERSION,
objects,
range: None,
}
}
#[must_use]
pub fn with_range(object: DescriptorObject, range: SplitRange) -> SplitDescriptor {
SplitDescriptor {
v: DESCRIPTOR_VERSION,
objects: vec![object],
range: Some(range),
}
}
#[must_use]
pub fn version(&self) -> u32 {
self.v
}
pub(crate) fn to_entries(&self) -> Vec<ObjectEntry> {
self.objects
.iter()
.map(|o| ObjectEntry {
key: o.key.clone(),
size: o.size,
etag: o.etag.clone(),
last_modified_ms: o.last_modified_ms,
})
.collect()
}
pub(crate) fn from_entries(entries: &[ObjectEntry]) -> SplitDescriptor {
SplitDescriptor::new(
entries
.iter()
.map(|e| DescriptorObject {
key: e.key.clone(),
size: e.size,
etag: e.etag.clone(),
last_modified_ms: e.last_modified_ms,
})
.collect(),
)
}
pub fn encode(&self) -> Result<Vec<u8>, CoordinationError> {
if self.v != DESCRIPTOR_VERSION {
return Err(CoordinationError::new(
CoordinationErrorKind::Fatal,
format!(
"descriptor version {} is not the supported {DESCRIPTOR_VERSION}; \
construct via SplitDescriptor::new",
self.v
),
));
}
self.validate_range()?;
Ok(serde_json::to_vec(self).expect("descriptor serialization is infallible: no non-string map keys, no fallible Serialize impls"))
}
pub fn decode(bytes: &[u8]) -> Result<SplitDescriptor, CoordinationError> {
let fatal = |reason: String| CoordinationError::new(CoordinationErrorKind::Fatal, reason);
let probe: VersionProbe = serde_json::from_slice(bytes)
.map_err(|e| fatal(format!("split descriptor is not valid JSON: {e}")))?;
if probe.v != DESCRIPTOR_VERSION {
return Err(fatal(format!(
"split descriptor version {} is not this release's version \
{DESCRIPTOR_VERSION}; the split was planned by an incompatible release",
probe.v
)));
}
let descriptor: SplitDescriptor = serde_json::from_slice(bytes)
.map_err(|e| fatal(format!("split descriptor failed to decode: {e}")))?;
descriptor.validate_range()?;
Ok(descriptor)
}
fn validate_range(&self) -> Result<(), CoordinationError> {
let Some(range) = self.range else {
return Ok(());
};
let fatal = |reason: String| CoordinationError::new(CoordinationErrorKind::Fatal, reason);
let [object] = self.objects.as_slice() else {
return Err(fatal(format!(
"a ranged split descriptor has exactly one member, this one has {}",
self.objects.len()
)));
};
if object.etag.is_none() {
return Err(fatal(format!(
"ranged split descriptor over \"{}\" has no ETag to pin its reads to",
object.key
)));
}
if range.start >= range.end || range.end > object.size {
return Err(fatal(format!(
"ranged split descriptor over \"{}\" has range [{}, {}), which is empty or \
extends past the object's {} bytes",
object.key, range.start, range.end, object.size
)));
}
Ok(())
}
}
pub fn split_id_for<'a, I>(members: I) -> Result<SplitId, CoordinationError>
where
I: IntoIterator<Item = (&'a str, Option<&'a str>)>,
{
split_id_with_version(members, PACKING_VERSION)
}
fn split_id_with_version<'a, I>(members: I, version: u32) -> Result<SplitId, CoordinationError>
where
I: IntoIterator<Item = (&'a str, Option<&'a str>)>,
{
let mut members: Vec<(&str, Option<&str>)> = members.into_iter().collect();
if members.is_empty() {
return Err(CoordinationError::new(
CoordinationErrorKind::Fatal,
"cannot mint a split id for an empty member set",
));
}
members.sort_unstable_by_key(|(key, _)| *key);
let mut hasher = Sha256::new();
hasher.update(b"spate-s3-split\n");
hasher.update(version.to_le_bytes());
for (key, etag) in members {
digest_member(&mut hasher, key, etag);
}
id_from_digest(hasher)
}
pub fn split_id_for_range(
key: &str,
etag: &str,
range: SplitRange,
) -> Result<SplitId, CoordinationError> {
if range.start >= range.end {
return Err(CoordinationError::new(
CoordinationErrorKind::Fatal,
format!(
"cannot mint a split id for the empty range [{}, {})",
range.start, range.end
),
));
}
let mut hasher = Sha256::new();
hasher.update(b"spate-s3-range\n");
hasher.update(PACKING_VERSION.to_le_bytes());
digest_member(&mut hasher, key, Some(etag));
hasher.update(range.start.to_le_bytes());
hasher.update(range.end.to_le_bytes());
hasher.update([range.delimiter]);
id_from_digest(hasher)
}
fn digest_member(hasher: &mut Sha256, key: &str, etag: Option<&str>) {
hasher.update(u32::try_from(key.len()).unwrap_or(u32::MAX).to_le_bytes());
hasher.update(key.as_bytes());
match etag {
Some(etag) => {
hasher.update([0x01]);
hasher.update(u32::try_from(etag.len()).unwrap_or(u32::MAX).to_le_bytes());
hasher.update(etag.as_bytes());
}
None => hasher.update([0x00]),
}
}
fn id_from_digest(hasher: Sha256) -> Result<SplitId, CoordinationError> {
let digest = hasher.finalize();
SplitId::new(format!("s3-{}", URL_SAFE_NO_PAD.encode(&digest[..16])))
}
#[derive(Clone, Copy, Debug)]
pub(crate) struct Packing {
pub(crate) target_bytes: u64,
pub(crate) compression: Compression,
pub(crate) delimiter: Option<u8>,
}
impl Packing {
pub(crate) fn delimiter_for(&self, entry: &ObjectEntry) -> Option<u8> {
let delimiter = self.delimiter?;
let cut = entry.size > self.target_bytes
&& entry.size <= MAX_SUBDIVIDABLE_BYTES
&& entry.etag.is_some()
&& Codec::resolve(self.compression, &entry.key) == Codec::Plain;
cut.then_some(delimiter)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum Packed {
Objects(Vec<ObjectEntry>),
Range(Rc<ObjectEntry>, SplitRange),
}
impl Packed {
pub(crate) fn id(&self) -> Result<SplitId, CoordinationError> {
match self {
Packed::Objects(members) => {
split_id_for(members.iter().map(|m| (m.key.as_str(), m.etag.as_deref())))
}
Packed::Range(entry, range) => split_id_for_range(
&entry.key,
entry
.etag
.as_deref()
.expect("pack cuts only an object with an ETag"),
*range,
),
}
}
pub(crate) fn descriptor(&self) -> SplitDescriptor {
match self {
Packed::Objects(members) => SplitDescriptor::from_entries(members),
Packed::Range(entry, range) => SplitDescriptor {
range: Some(*range),
..SplitDescriptor::from_entries(std::slice::from_ref(&**entry))
},
}
}
pub(crate) fn weight(&self) -> u64 {
match self {
Packed::Objects(members) => members
.iter()
.fold(0u64, |acc, m| acc.saturating_add(m.size)),
Packed::Range(_, range) => range.end - range.start,
}
.max(1)
}
}
fn tile(size: u64, target: u64, delimiter: u8) -> impl Iterator<Item = SplitRange> {
let n = size.div_ceil(target);
let boundary = move |i: u64| (u128::from(i) * u128::from(size) / u128::from(n)) as u64;
(0..n).map(move |i| SplitRange::new(boundary(i), boundary(i + 1), delimiter))
}
pub(crate) fn pack(entries: Vec<ObjectEntry>, packing: &Packing) -> Vec<Packed> {
let target_bytes = packing.target_bytes;
debug_assert!(
target_bytes > 0,
"config validation rejects a zero split target"
);
struct Bin {
members: Vec<ObjectEntry>,
cost: u64,
range: Option<(Rc<ObjectEntry>, SplitRange)>,
}
let floor = (target_bytes / OPEN_COST_DIVISOR).max(1);
let mut bins: Vec<Bin> = Vec::new();
let mut open: VecDeque<usize> = VecDeque::new();
for entry in entries {
if let Some(delimiter) = packing.delimiter_for(&entry) {
if open.len() == PACKING_LOOKBACK {
open.pop_front();
}
let entry = Rc::new(entry);
bins.extend(tile(entry.size, target_bytes, delimiter).map(|range| Bin {
members: Vec::new(),
cost: range.end - range.start,
range: Some((Rc::clone(&entry), range)),
}));
continue;
}
let cost = entry.size.max(floor);
let idx = match open
.iter()
.position(|&i| bins[i].cost.saturating_add(cost) <= target_bytes)
{
Some(pos) => open[pos],
None => {
if open.len() == PACKING_LOOKBACK {
open.pop_front();
}
bins.push(Bin {
members: Vec::new(),
cost: 0,
range: None,
});
let idx = bins.len() - 1;
open.push_back(idx);
idx
}
};
bins[idx].members.push(entry);
bins[idx].cost = bins[idx].cost.saturating_add(cost);
if bins[idx].cost >= target_bytes
&& let Some(pos) = open.iter().position(|&i| i == idx)
{
open.remove(pos);
}
}
bins.into_iter()
.map(|Bin { members, range, .. }| match range {
Some((entry, range)) => Packed::Range(entry, range),
None => Packed::Objects(members),
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use proptest::prelude::*;
fn entry(key: &str, size: u64) -> ObjectEntry {
ObjectEntry {
key: key.to_string(),
size,
etag: Some(format!("\"etag-{key}\"")),
last_modified_ms: 1_760_000_000_000,
}
}
const MB: u64 = 1024 * 1024;
#[test]
fn the_open_cost_floor_rounds_down() {
let target = "1.1MiB".parse::<bytesize::ByteSize>().unwrap().as_u64();
assert_ne!(target % 16, 0);
let listing: Vec<ObjectEntry> = (0..32).map(|i| entry(&format!("k{i:02}"), 1)).collect();
let bins: Vec<usize> = object_bins(&pack(listing, &whole(target)))
.iter()
.map(Vec::len)
.collect();
assert_eq!(bins, [16, 16]);
}
#[test]
fn whole_object_bins_are_pinned() {
let keys = |packed: Vec<Packed>| -> Vec<Vec<String>> {
object_bins(&packed)
.into_iter()
.map(|b| b.into_iter().map(|e| e.key).collect())
.collect()
};
for packing in [whole(64 * MB), cut(64 * MB)] {
let first_fit = vec![
entry("a", 40 * MB),
entry("b", 50 * MB),
entry("c", 10 * MB),
];
assert_eq!(keys(pack(first_fit, &packing)), [vec!["a", "c"], vec!["b"]]);
let mut evicted = vec![entry("a", 40 * MB)];
evicted.extend((0..10).map(|i| entry(&format!("m{i}"), 60 * MB)));
evicted.push(entry("z", 20 * MB));
assert_eq!(keys(pack(evicted, &packing)).len(), 12, "{packing:?}");
let mut exact = vec![entry("a", 40 * MB), entry("b", 64 * MB)];
exact.extend((0..9).map(|i| entry(&format!("m{i}"), 60 * MB)));
exact.push(entry("z", 20 * MB));
let bins = keys(pack(exact, &packing));
assert_eq!(bins[0], ["a", "z"], "{packing:?}");
}
}
fn reference_pack(entries: Vec<ObjectEntry>, target_bytes: u64) -> Vec<Vec<ObjectEntry>> {
struct Bin {
members: Vec<ObjectEntry>,
cost: u64,
}
let floor = (target_bytes / 16).max(1);
let mut bins: Vec<Bin> = Vec::new();
let mut open: VecDeque<usize> = VecDeque::new();
for entry in entries {
let cost = entry.size.max(floor);
let idx = match open
.iter()
.position(|&i| bins[i].cost.saturating_add(cost) <= target_bytes)
{
Some(pos) => open[pos],
None => {
if open.len() == 10 {
open.pop_front();
}
bins.push(Bin {
members: Vec::new(),
cost: 0,
});
open.push_back(bins.len() - 1);
bins.len() - 1
}
};
bins[idx].members.push(entry);
bins[idx].cost = bins[idx].cost.saturating_add(cost);
if bins[idx].cost >= target_bytes
&& let Some(pos) = open.iter().position(|&i| i == idx)
{
open.remove(pos);
}
}
bins.into_iter().map(|b| b.members).collect()
}
fn whole(target: u64) -> Packing {
Packing {
target_bytes: target,
compression: Compression::Auto,
delimiter: None,
}
}
fn cut(target: u64) -> Packing {
Packing {
delimiter: Some(b'\n'),
..whole(target)
}
}
fn object_bins(packed: &[Packed]) -> Vec<Vec<ObjectEntry>> {
packed
.iter()
.filter_map(|p| match p {
Packed::Objects(members) => Some(members.clone()),
Packed::Range(..) => None,
})
.collect()
}
#[test]
fn digest_id_is_pinned() {
let id = split_id_for([
("exports/2026/part-000.ndjson", Some("\"9b2cf5\"")),
("exports/2026/part-001.ndjson", None),
])
.unwrap();
assert_eq!(id.as_str(), "s3-xje_MfL-wSNqMCft5ellCg");
}
#[test]
fn digest_is_order_insensitive_but_content_sensitive() {
let forward = split_id_for([("a", Some("1")), ("b", Some("2"))]).unwrap();
let reversed = split_id_for([("b", Some("2")), ("a", Some("1"))]).unwrap();
assert_eq!(forward, reversed);
let other_key = split_id_for([("a", Some("1")), ("c", Some("2"))]).unwrap();
assert_ne!(forward, other_key);
}
#[test]
fn digest_changes_when_an_etag_changes() {
let before = split_id_for([("a", Some("v1")), ("b", Some("x"))]).unwrap();
let overwritten = split_id_for([("a", Some("v2")), ("b", Some("x"))]).unwrap();
let dropped = split_id_for([("a", None), ("b", Some("x"))]).unwrap();
assert_ne!(before, overwritten);
assert_ne!(before, dropped);
}
#[test]
fn digest_folds_in_the_packing_version() {
let v1 = split_id_with_version([("a", Some("1"))], 1).unwrap();
let v2 = split_id_with_version([("a", Some("1"))], 2).unwrap();
assert_ne!(v1, v2);
}
#[test]
fn empty_member_set_is_rejected() {
assert!(split_id_for(std::iter::empty()).is_err());
}
#[test]
fn ambiguous_concatenations_do_not_collide() {
let a = split_id_for([("ab", Some("c"))]).unwrap();
let b = split_id_for([("a", Some("bc"))]).unwrap();
let c = split_id_for([("abc", None)]).unwrap();
assert_ne!(a, b);
assert_ne!(a, c);
assert_ne!(b, c);
}
#[test]
fn descriptor_round_trips() {
let desc = SplitDescriptor::from_entries(&[
entry("exports/part-000.ndjson.gz", 52 * MB),
ObjectEntry {
key: "exports/part-001.ndjson.gz".to_string(),
size: 9,
etag: None,
last_modified_ms: 1,
},
]);
let decoded = SplitDescriptor::decode(&desc.encode().unwrap()).unwrap();
assert_eq!(decoded, desc);
assert_eq!(decoded.v, DESCRIPTOR_VERSION);
}
#[test]
fn descriptor_encoding_is_pinned() {
let desc = SplitDescriptor::from_entries(&[entry("k", 5)]);
assert_eq!(
String::from_utf8(desc.encode().unwrap()).unwrap(),
r#"{"v":2,"objects":[{"key":"k","size":5,"etag":"\"etag-k\"","last_modified_ms":1760000000000}]}"#,
);
}
#[test]
fn encode_refuses_a_descriptor_not_built_by_new() {
let rogue = SplitDescriptor {
v: 0,
objects: vec![],
range: None,
};
let err = rogue.encode().unwrap_err();
assert_eq!(err.kind, CoordinationErrorKind::Fatal);
assert!(
err.reason.contains("SplitDescriptor::new"),
"reason: {}",
err.reason
);
assert_eq!(SplitDescriptor::new(vec![]).version(), DESCRIPTOR_VERSION);
}
#[test]
fn a_version_1_descriptor_is_refused() {
let err = SplitDescriptor::decode(
br#"{"v":1,"objects":[{"key":"k","size":5,"etag":null,"last_modified_ms":1}]}"#,
)
.unwrap_err();
assert_eq!(err.kind, CoordinationErrorKind::Fatal);
assert!(err.reason.contains("version 1"), "reason: {}", err.reason);
}
#[test]
fn unknown_descriptor_version_is_rejected_actionably() {
let err = SplitDescriptor::decode(br#"{"v":999,"objects":[]}"#).unwrap_err();
assert_eq!(err.kind, CoordinationErrorKind::Fatal);
assert!(err.reason.contains("version 999"), "reason: {}", err.reason);
assert!(
err.reason.contains("incompatible release"),
"reason: {}",
err.reason
);
let garbage = SplitDescriptor::decode(b"not json").unwrap_err();
assert_eq!(garbage.kind, CoordinationErrorKind::Fatal);
}
fn ranged_object(size: u64, etag: Option<&str>) -> DescriptorObject {
DescriptorObject {
key: "big.ndjson".to_string(),
size,
etag: etag.map(str::to_owned),
last_modified_ms: 7,
}
}
#[test]
fn ranged_descriptor_round_trips_and_its_encoding_is_pinned() {
let desc = SplitDescriptor::with_range(
ranged_object(100, Some("\"e\"")),
SplitRange::new(40, 80, b'\n'),
);
let encoded = desc.encode().unwrap();
assert_eq!(
String::from_utf8(encoded.clone()).unwrap(),
r#"{"v":2,"objects":[{"key":"big.ndjson","size":100,"etag":"\"e\"","last_modified_ms":7}],"range":{"start":40,"end":80,"delimiter":10}}"#,
);
assert_eq!(SplitDescriptor::decode(&encoded).unwrap(), desc);
}
#[test]
fn invalid_ranged_descriptors_are_refused_on_both_sides() {
let pinned = || ranged_object(100, Some("\"e\""));
let two_members = SplitDescriptor {
objects: vec![pinned(), pinned()],
..SplitDescriptor::with_range(pinned(), SplitRange::new(0, 10, b'\n'))
};
let no_members = SplitDescriptor {
objects: vec![],
..SplitDescriptor::with_range(pinned(), SplitRange::new(0, 10, b'\n'))
};
let cases = [
("member", two_members),
("member", no_members),
(
"ETag",
SplitDescriptor::with_range(
ranged_object(100, None),
SplitRange::new(0, 10, b'\n'),
),
),
(
"empty",
SplitDescriptor::with_range(pinned(), SplitRange::new(10, 10, b'\n')),
),
(
"empty",
SplitDescriptor::with_range(pinned(), SplitRange::new(11, 10, b'\n')),
),
(
"past the object",
SplitDescriptor::with_range(pinned(), SplitRange::new(0, 101, b'\n')),
),
];
for (needle, desc) in cases {
let err = desc.encode().unwrap_err();
assert_eq!(err.kind, CoordinationErrorKind::Fatal);
assert!(err.reason.contains(needle), "encode: {}", err.reason);
let raw = serde_json::to_vec(&desc).unwrap();
let err = SplitDescriptor::decode(&raw).unwrap_err();
assert_eq!(err.kind, CoordinationErrorKind::Fatal);
assert!(err.reason.contains(needle), "decode: {}", err.reason);
}
let whole = SplitDescriptor::with_range(pinned(), SplitRange::new(0, 100, b'\n'));
SplitDescriptor::decode(&whole.encode().unwrap()).unwrap();
}
#[test]
fn ranged_id_is_pinned() {
let id = split_id_for_range(
"exports/2026/part-000.ndjson",
"\"9b2cf5\"",
SplitRange::new(0, 64 * MB, b'\n'),
)
.unwrap();
assert_eq!(id.as_str(), "s3-fA0wNRW2MyDxquwFvNy0JA");
}
#[test]
fn ranged_ids_digest_every_input_and_never_equal_member_set_ids() {
let base = split_id_for_range("k", "e", SplitRange::new(0, 10, b'\n')).unwrap();
for other in [
split_id_for_range("k2", "e", SplitRange::new(0, 10, b'\n')),
split_id_for_range("k", "e2", SplitRange::new(0, 10, b'\n')),
split_id_for_range("k", "e", SplitRange::new(1, 10, b'\n')),
split_id_for_range("k", "e", SplitRange::new(0, 11, b'\n')),
split_id_for_range("k", "e", SplitRange::new(0, 10, b';')),
] {
assert_ne!(other.unwrap(), base);
}
assert_ne!(split_id_for([("k", Some("e"))]).unwrap(), base);
assert!(split_id_for_range("k", "e", SplitRange::new(10, 10, b'\n')).is_err());
}
fn shaped(i: usize, size: u64, kind: u8) -> ObjectEntry {
let key = format!(
"k{i:04}.ndjson{}",
["", ".gz", ".zst", ""][kind as usize % 4]
);
ObjectEntry {
etag: (kind % 4 != 3).then(|| format!("\"etag-{i}\"")),
..entry(&key, size)
}
}
#[test]
fn packing_is_deterministic_for_a_fixed_listing() {
let listing: Vec<ObjectEntry> = (0..200)
.map(|i| entry(&format!("k{i:04}"), (i % 90) * MB))
.collect();
let a = pack(listing.clone(), &cut(64 * MB));
let b = pack(listing, &cut(64 * MB));
assert_eq!(a, b);
}
#[test]
fn tiny_objects_coalesce_under_the_open_cost_floor() {
let listing: Vec<ObjectEntry> = (0..64).map(|i| entry(&format!("k{i:02}"), 1)).collect();
let bins = object_bins(&pack(listing, &whole(64 * MB)));
assert_eq!(bins.len(), 4);
assert!(bins.iter().all(|b| b.len() == 16));
}
#[test]
fn adversarial_listing_sizes_do_not_overflow_the_fit_test() {
let listing = vec![
entry("a", 10 * MB),
entry("huge", u64::MAX),
entry("z", 10 * MB),
];
let packed = pack(listing, &cut(64 * MB));
let bins = object_bins(&packed);
assert_eq!(bins.len(), packed.len(), "nothing is cut");
let huge_bin = bins
.iter()
.find(|b| b.iter().any(|e| e.key == "huge"))
.unwrap();
assert_eq!(huge_bin.len(), 1, "the oversized object lands alone");
let total: usize = bins.iter().map(Vec::len).sum();
assert_eq!(total, 3, "nothing lost, nothing duplicated");
}
#[test]
fn an_object_above_the_subdivision_limit_stays_whole() {
let target = 1 << 30;
let six_tib = pack(vec![entry("6tib", 6 << 40)], &cut(target));
assert_eq!(six_tib.len(), 6 << 10);
let over = entry("over", MAX_SUBDIVIDABLE_BYTES + 1);
assert_eq!(
pack(vec![over.clone()], &cut(target)),
[Packed::Objects(vec![over])]
);
let at = pack(vec![entry("at", MAX_SUBDIVIDABLE_BYTES)], &cut(target));
assert_eq!(at.len() as u64, MAX_SUBDIVIDABLE_BYTES / target);
assert!(at.iter().all(|p| matches!(p, Packed::Range(..))));
}
#[test]
fn objects_that_cannot_be_cut_stay_whole() {
let target = 64 * MB;
let plain = entry("plain.ndjson", 3 * target);
let gzip_config = Packing {
compression: Compression::Gzip,
..cut(target)
};
let no_etag = ObjectEntry {
etag: None,
..plain.clone()
};
for (packing, object) in [
(cut(target), entry("gz.ndjson.gz", 3 * target)),
(cut(target), entry("zst.ndjson.zst", 3 * target)),
(gzip_config, plain.clone()),
(cut(target), no_etag),
(whole(target), plain.clone()),
(cut(target), entry("at.ndjson", target)),
] {
assert_eq!(
pack(vec![object.clone()], &packing),
[Packed::Objects(vec![object.clone()])],
"{packing:?} {object:?}"
);
}
assert_eq!(
pack(vec![entry("over.ndjson", target + 1)], &cut(target)).len(),
2,
"one byte above the target is cut"
);
}
#[test]
fn oversized_object_gets_its_own_split() {
let listing = vec![
entry("a", 10 * MB),
entry("huge", 500 * MB),
entry("b", 10 * MB),
];
let bins = object_bins(&pack(listing, &whole(64 * MB)));
let huge_bin = bins
.iter()
.find(|b| b.iter().any(|e| e.key == "huge"))
.unwrap();
assert_eq!(
huge_bin.len(),
1,
"an oversized object never shares a split"
);
}
#[test]
fn a_large_plain_object_is_cut_in_place() {
let listing = vec![
entry("a", 60 * MB),
entry("huge", 500 * MB),
entry("z", 60 * MB),
];
let packed = pack(listing, &cut(64 * MB));
let keys: Vec<&str> = packed
.iter()
.map(|p| match p {
Packed::Objects(members) => members[0].key.as_str(),
Packed::Range(entry, _) => entry.key.as_str(),
})
.collect();
let mut want = vec!["a"];
want.extend(["huge"; 8]);
want.push("z");
assert_eq!(keys, want);
}
#[test]
fn ranges_tile_the_object_within_half_to_one_target() {
for target in 1..=64u64 {
for size in target + 1..=8 * target {
assert_tiles(size, target);
}
}
assert_tiles(MAX_SUBDIVIDABLE_BYTES, MB);
assert_tiles(MAX_SUBDIVIDABLE_BYTES - 1, MB);
}
fn assert_tiles(size: u64, target: u64) {
let mut count = 0;
let mut next = 0;
for range in tile(size, target, b'\n') {
count += 1;
assert_eq!(range.start, next, "size {size} target {target}");
let len = range.end - range.start;
assert!(
target / 2 <= len && len <= target,
"size {size} target {target}: range {range:?} is {len} bytes"
);
next = range.end;
}
assert_eq!(count, size.div_ceil(target), "size {size} target {target}");
assert_eq!(next, size, "size {size} target {target}");
}
#[test]
fn packing_preserves_listing_order_within_and_across_bins() {
let listing: Vec<ObjectEntry> = (0..50)
.map(|i| {
entry(
&format!("k{i:02}"),
if i % 7 == 0 { 60 * MB } else { 3 * MB },
)
})
.collect();
let bins = object_bins(&pack(listing, &whole(64 * MB)));
for bin in &bins {
let keys: Vec<&str> = bin.iter().map(|e| e.key.as_str()).collect();
let mut sorted = keys.clone();
sorted.sort_unstable();
assert_eq!(keys, sorted, "members stay in listing order within a bin");
}
let firsts: Vec<&str> = bins.iter().map(|b| b[0].key.as_str()).collect();
let mut sorted = firsts.clone();
sorted.sort_unstable();
assert_eq!(
firsts, sorted,
"bins emerge in listing order of their first member"
);
}
#[test]
fn lookback_bounds_how_long_a_bin_stays_open() {
let mut listing = vec![entry("a-half-full", 40 * MB)];
for i in 0..PACKING_LOOKBACK + 2 {
listing.push(entry(&format!("b{i:02}"), 60 * MB));
}
listing.sort_unstable_by(|a, b| a.key.cmp(&b.key));
let bins = object_bins(&pack(listing, &whole(64 * MB)));
assert_eq!(bins.len(), PACKING_LOOKBACK + 3);
assert!(bins.iter().all(|b| b.len() == 1));
}
proptest! {
#[test]
fn prop_packing_partitions_the_listing_exactly(
objects in proptest::collection::vec((0u64..300 * MB, 0u8..4), 0..120),
target_mb in 1u64..129,
) {
let listing: Vec<ObjectEntry> = objects
.iter()
.enumerate()
.map(|(i, &(size, kind))| shaped(i, size, kind))
.collect();
let packed = pack(listing.clone(), &cut(target_mb * MB));
for object in &listing {
let whole_count = object_bins(&packed)
.iter()
.flatten()
.filter(|e| *e == object)
.count();
let ranges: Vec<SplitRange> = packed
.iter()
.filter_map(|p| match p {
Packed::Range(e, range) if **e == *object => Some(*range),
_ => None,
})
.collect();
if ranges.is_empty() {
prop_assert_eq!(whole_count, 1, "{:?}", object);
} else {
prop_assert_eq!(whole_count, 0, "{:?}", object);
let mut next = 0;
for range in ranges {
prop_assert_eq!(range.start, next);
next = range.end;
}
prop_assert_eq!(next, object.size);
}
}
prop_assert!(object_bins(&packed).iter().all(|b| !b.is_empty()));
}
#[test]
fn prop_small_object_bins_do_not_depend_on_cutting(
objects in proptest::collection::vec((0u64..300 * MB, 0u8..4), 0..120),
target_mb in 1u64..129,
) {
let listing: Vec<ObjectEntry> = objects
.iter()
.enumerate()
.map(|(i, &(size, kind))| shaped(i, size, kind))
.collect();
let packing = cut(target_mb * MB);
let expanded: Vec<Packed> = pack(listing.clone(), &whole(target_mb * MB))
.into_iter()
.flat_map(|split| match &split {
Packed::Objects(members) if members.len() == 1 => {
match packing.delimiter_for(&members[0]) {
Some(delimiter) => tile(members[0].size, packing.target_bytes, delimiter)
.map(|range| Packed::Range(Rc::new(members[0].clone()), range))
.collect(),
None => vec![split],
}
}
_ => vec![split],
})
.collect();
prop_assert_eq!(pack(listing, &packing), expanded);
}
#[test]
fn prop_whole_object_packing_matches_the_reference(
objects in proptest::collection::vec((0u64..300, 0u8..4), 0..120),
target_mb in 1u64..129,
) {
let listing: Vec<ObjectEntry> = objects
.iter()
.enumerate()
.map(|(i, &(size_mb, kind))| shaped(i, size_mb * MB, kind))
.collect();
let packed = pack(listing.clone(), &whole(target_mb * MB));
prop_assert_eq!(object_bins(&packed).len(), packed.len());
prop_assert_eq!(object_bins(&packed), reference_pack(listing, target_mb * MB));
}
#[test]
fn prop_ranges_tile_large_objects(
target in 1u64..=1 << 32,
n in 2u64..=1 << 12,
extra in any::<u64>(),
) {
assert_tiles((n - 1) * target + 1 + extra % target, target);
assert_tiles(n * target, target);
}
#[test]
fn prop_member_count_is_bounded_by_the_floor(
sizes in proptest::collection::vec(0u64..300 * MB, 0..120),
) {
let target = 64 * MB;
let listing: Vec<ObjectEntry> = sizes
.iter()
.enumerate()
.map(|(i, &s)| entry(&format!("k{i:04}"), s))
.collect();
let bins = object_bins(&pack(listing, &cut(target)));
prop_assert!(bins.iter().all(|b| b.len() <= 16));
}
#[test]
fn prop_split_ids_are_valid_for_arbitrary_keys(
keys in proptest::collection::btree_set("[ -~]{1,64}", 1..8),
etag in proptest::option::of("[ -~]{1,16}"),
) {
let members: Vec<(&str, Option<&str>)> =
keys.iter().map(|k| (k.as_str(), etag.as_deref())).collect();
let id = split_id_for(members).unwrap();
prop_assert!(id.as_str().starts_with("s3-"));
prop_assert_eq!(id.as_str().len(), 25);
}
#[test]
fn prop_packing_and_ids_are_deterministic(
sizes in proptest::collection::vec(0u64..200 * MB, 1..60),
) {
let listing: Vec<ObjectEntry> = sizes
.iter()
.enumerate()
.map(|(i, &s)| entry(&format!("k{i:04}"), s))
.collect();
let ids = |packed: &[Packed]| -> Vec<SplitId> {
packed.iter().map(|p| p.id().unwrap()).collect()
};
let a = pack(listing.clone(), &cut(32 * MB));
let b = pack(listing, &cut(32 * MB));
prop_assert_eq!(ids(&a), ids(&b));
}
}
}