1use crate::config::Compression;
44use crate::fetch::ObjectEntry;
45use crate::framer::Codec;
46use base64::Engine as _;
47use base64::engine::general_purpose::URL_SAFE_NO_PAD;
48use serde::{Deserialize, Serialize};
49use sha2::{Digest as _, Sha256};
50use spate_core::coordination::{CoordinationError, CoordinationErrorKind, SplitId};
51use std::collections::VecDeque;
52use std::rc::Rc;
53
54pub const DESCRIPTOR_VERSION: u32 = 2;
58
59pub(crate) const PACKING_VERSION: u32 = 2;
63
64pub(crate) const MAX_SUBDIVIDABLE_BYTES: u64 = 50_000 << 30;
68
69pub(crate) const PACKING_LOOKBACK: usize = 10;
74
75pub(crate) const OPEN_COST_DIVISOR: u64 = 16;
78
79#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
84pub struct DescriptorObject {
85 pub key: String,
87 pub size: u64,
89 pub etag: Option<String>,
93 pub last_modified_ms: i64,
95}
96
97#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
105#[non_exhaustive]
106pub struct SplitRange {
107 pub start: u64,
109 pub end: u64,
111 pub delimiter: u8,
114}
115
116impl SplitRange {
117 #[must_use]
119 pub fn new(start: u64, end: u64, delimiter: u8) -> SplitRange {
120 SplitRange {
121 start,
122 end,
123 delimiter,
124 }
125 }
126}
127
128#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
141pub struct SplitDescriptor {
142 pub(crate) v: u32,
146 pub objects: Vec<DescriptorObject>,
148 #[serde(default, skip_serializing_if = "Option::is_none")]
151 pub range: Option<SplitRange>,
152}
153
154#[derive(Deserialize)]
157struct VersionProbe {
158 v: u32,
159}
160
161impl SplitDescriptor {
162 #[must_use]
167 pub fn new(objects: Vec<DescriptorObject>) -> SplitDescriptor {
168 SplitDescriptor {
169 v: DESCRIPTOR_VERSION,
170 objects,
171 range: None,
172 }
173 }
174
175 #[must_use]
181 pub fn with_range(object: DescriptorObject, range: SplitRange) -> SplitDescriptor {
182 SplitDescriptor {
183 v: DESCRIPTOR_VERSION,
184 objects: vec![object],
185 range: Some(range),
186 }
187 }
188
189 #[must_use]
192 pub fn version(&self) -> u32 {
193 self.v
194 }
195
196 pub(crate) fn to_entries(&self) -> Vec<ObjectEntry> {
199 self.objects
200 .iter()
201 .map(|o| ObjectEntry {
202 key: o.key.clone(),
203 size: o.size,
204 etag: o.etag.clone(),
205 last_modified_ms: o.last_modified_ms,
206 })
207 .collect()
208 }
209
210 pub(crate) fn from_entries(entries: &[ObjectEntry]) -> SplitDescriptor {
212 SplitDescriptor::new(
213 entries
214 .iter()
215 .map(|e| DescriptorObject {
216 key: e.key.clone(),
217 size: e.size,
218 etag: e.etag.clone(),
219 last_modified_ms: e.last_modified_ms,
220 })
221 .collect(),
222 )
223 }
224
225 pub fn encode(&self) -> Result<Vec<u8>, CoordinationError> {
235 if self.v != DESCRIPTOR_VERSION {
236 return Err(CoordinationError::new(
237 CoordinationErrorKind::Fatal,
238 format!(
239 "descriptor version {} is not the supported {DESCRIPTOR_VERSION}; \
240 construct via SplitDescriptor::new",
241 self.v
242 ),
243 ));
244 }
245 self.validate_range()?;
246 Ok(serde_json::to_vec(self).expect("descriptor serialization is infallible: no non-string map keys, no fallible Serialize impls"))
247 }
248
249 pub fn decode(bytes: &[u8]) -> Result<SplitDescriptor, CoordinationError> {
258 let fatal = |reason: String| CoordinationError::new(CoordinationErrorKind::Fatal, reason);
259 let probe: VersionProbe = serde_json::from_slice(bytes)
260 .map_err(|e| fatal(format!("split descriptor is not valid JSON: {e}")))?;
261 if probe.v != DESCRIPTOR_VERSION {
262 return Err(fatal(format!(
263 "split descriptor version {} is not this release's version \
264 {DESCRIPTOR_VERSION}; the split was planned by an incompatible release",
265 probe.v
266 )));
267 }
268 let descriptor: SplitDescriptor = serde_json::from_slice(bytes)
269 .map_err(|e| fatal(format!("split descriptor failed to decode: {e}")))?;
270 descriptor.validate_range()?;
271 Ok(descriptor)
272 }
273
274 fn validate_range(&self) -> Result<(), CoordinationError> {
277 let Some(range) = self.range else {
278 return Ok(());
279 };
280 let fatal = |reason: String| CoordinationError::new(CoordinationErrorKind::Fatal, reason);
281 let [object] = self.objects.as_slice() else {
282 return Err(fatal(format!(
283 "a ranged split descriptor has exactly one member, this one has {}",
284 self.objects.len()
285 )));
286 };
287 if object.etag.is_none() {
288 return Err(fatal(format!(
289 "ranged split descriptor over \"{}\" has no ETag to pin its reads to",
290 object.key
291 )));
292 }
293 if range.start >= range.end || range.end > object.size {
294 return Err(fatal(format!(
295 "ranged split descriptor over \"{}\" has range [{}, {}), which is empty or \
296 extends past the object's {} bytes",
297 object.key, range.start, range.end, object.size
298 )));
299 }
300 Ok(())
301 }
302}
303
304pub fn split_id_for<'a, I>(members: I) -> Result<SplitId, CoordinationError>
349where
350 I: IntoIterator<Item = (&'a str, Option<&'a str>)>,
351{
352 split_id_with_version(members, PACKING_VERSION)
353}
354
355fn split_id_with_version<'a, I>(members: I, version: u32) -> Result<SplitId, CoordinationError>
358where
359 I: IntoIterator<Item = (&'a str, Option<&'a str>)>,
360{
361 let mut members: Vec<(&str, Option<&str>)> = members.into_iter().collect();
362 if members.is_empty() {
363 return Err(CoordinationError::new(
364 CoordinationErrorKind::Fatal,
365 "cannot mint a split id for an empty member set",
366 ));
367 }
368 members.sort_unstable_by_key(|(key, _)| *key);
369
370 let mut hasher = Sha256::new();
371 hasher.update(b"spate-s3-split\n");
372 hasher.update(version.to_le_bytes());
373 for (key, etag) in members {
374 digest_member(&mut hasher, key, etag);
375 }
376 id_from_digest(hasher)
377}
378
379pub fn split_id_for_range(
413 key: &str,
414 etag: &str,
415 range: SplitRange,
416) -> Result<SplitId, CoordinationError> {
417 if range.start >= range.end {
418 return Err(CoordinationError::new(
419 CoordinationErrorKind::Fatal,
420 format!(
421 "cannot mint a split id for the empty range [{}, {})",
422 range.start, range.end
423 ),
424 ));
425 }
426 let mut hasher = Sha256::new();
427 hasher.update(b"spate-s3-range\n");
428 hasher.update(PACKING_VERSION.to_le_bytes());
429 digest_member(&mut hasher, key, Some(etag));
430 hasher.update(range.start.to_le_bytes());
431 hasher.update(range.end.to_le_bytes());
432 hasher.update([range.delimiter]);
433 id_from_digest(hasher)
434}
435
436fn digest_member(hasher: &mut Sha256, key: &str, etag: Option<&str>) {
439 hasher.update(u32::try_from(key.len()).unwrap_or(u32::MAX).to_le_bytes());
440 hasher.update(key.as_bytes());
441 match etag {
442 Some(etag) => {
443 hasher.update([0x01]);
444 hasher.update(u32::try_from(etag.len()).unwrap_or(u32::MAX).to_le_bytes());
445 hasher.update(etag.as_bytes());
446 }
447 None => hasher.update([0x00]),
448 }
449}
450
451fn id_from_digest(hasher: Sha256) -> Result<SplitId, CoordinationError> {
453 let digest = hasher.finalize();
454 SplitId::new(format!("s3-{}", URL_SAFE_NO_PAD.encode(&digest[..16])))
455}
456
457#[derive(Clone, Copy, Debug)]
459pub(crate) struct Packing {
460 pub(crate) target_bytes: u64,
462 pub(crate) compression: Compression,
464 pub(crate) delimiter: Option<u8>,
466}
467
468impl Packing {
469 pub(crate) fn delimiter_for(&self, entry: &ObjectEntry) -> Option<u8> {
474 let delimiter = self.delimiter?;
475 let cut = entry.size > self.target_bytes
476 && entry.size <= MAX_SUBDIVIDABLE_BYTES
477 && entry.etag.is_some()
478 && Codec::resolve(self.compression, &entry.key) == Codec::Plain;
479 cut.then_some(delimiter)
480 }
481}
482
483#[derive(Clone, Debug, PartialEq, Eq)]
485pub(crate) enum Packed {
486 Objects(Vec<ObjectEntry>),
488 Range(Rc<ObjectEntry>, SplitRange),
491}
492
493impl Packed {
494 pub(crate) fn id(&self) -> Result<SplitId, CoordinationError> {
496 match self {
497 Packed::Objects(members) => {
498 split_id_for(members.iter().map(|m| (m.key.as_str(), m.etag.as_deref())))
499 }
500 Packed::Range(entry, range) => split_id_for_range(
501 &entry.key,
502 entry
503 .etag
504 .as_deref()
505 .expect("pack cuts only an object with an ETag"),
506 *range,
507 ),
508 }
509 }
510
511 pub(crate) fn descriptor(&self) -> SplitDescriptor {
513 match self {
514 Packed::Objects(members) => SplitDescriptor::from_entries(members),
515 Packed::Range(entry, range) => SplitDescriptor {
516 range: Some(*range),
517 ..SplitDescriptor::from_entries(std::slice::from_ref(&**entry))
518 },
519 }
520 }
521
522 pub(crate) fn weight(&self) -> u64 {
525 match self {
526 Packed::Objects(members) => members
528 .iter()
529 .fold(0u64, |acc, m| acc.saturating_add(m.size)),
530 Packed::Range(_, range) => range.end - range.start,
531 }
532 .max(1)
533 }
534}
535
536fn tile(size: u64, target: u64, delimiter: u8) -> impl Iterator<Item = SplitRange> {
539 let n = size.div_ceil(target);
540 let boundary = move |i: u64| (u128::from(i) * u128::from(size) / u128::from(n)) as u64;
543 (0..n).map(move |i| SplitRange::new(boundary(i), boundary(i + 1), delimiter))
544}
545
546pub(crate) fn pack(entries: Vec<ObjectEntry>, packing: &Packing) -> Vec<Packed> {
560 let target_bytes = packing.target_bytes;
561 debug_assert!(
562 target_bytes > 0,
563 "config validation rejects a zero split target"
564 );
565 struct Bin {
566 members: Vec<ObjectEntry>,
567 cost: u64,
568 range: Option<(Rc<ObjectEntry>, SplitRange)>,
570 }
571 let floor = (target_bytes / OPEN_COST_DIVISOR).max(1);
572 let mut bins: Vec<Bin> = Vec::new();
573 let mut open: VecDeque<usize> = VecDeque::new();
575 for entry in entries {
576 if let Some(delimiter) = packing.delimiter_for(&entry) {
577 if open.len() == PACKING_LOOKBACK {
578 open.pop_front();
579 }
580 let entry = Rc::new(entry);
581 bins.extend(tile(entry.size, target_bytes, delimiter).map(|range| Bin {
582 members: Vec::new(),
583 cost: range.end - range.start,
584 range: Some((Rc::clone(&entry), range)),
585 }));
586 continue;
587 }
588 let cost = entry.size.max(floor);
589 let idx = match open
592 .iter()
593 .position(|&i| bins[i].cost.saturating_add(cost) <= target_bytes)
594 {
595 Some(pos) => open[pos],
596 None => {
597 if open.len() == PACKING_LOOKBACK {
598 open.pop_front();
599 }
600 bins.push(Bin {
601 members: Vec::new(),
602 cost: 0,
603 range: None,
604 });
605 let idx = bins.len() - 1;
606 open.push_back(idx);
607 idx
608 }
609 };
610 bins[idx].members.push(entry);
611 bins[idx].cost = bins[idx].cost.saturating_add(cost);
612 if bins[idx].cost >= target_bytes
613 && let Some(pos) = open.iter().position(|&i| i == idx)
614 {
615 open.remove(pos);
616 }
617 }
618 bins.into_iter()
619 .map(|Bin { members, range, .. }| match range {
620 Some((entry, range)) => Packed::Range(entry, range),
621 None => Packed::Objects(members),
622 })
623 .collect()
624}
625
626#[cfg(test)]
627mod tests {
628 use super::*;
629 use proptest::prelude::*;
630
631 fn entry(key: &str, size: u64) -> ObjectEntry {
632 ObjectEntry {
633 key: key.to_string(),
634 size,
635 etag: Some(format!("\"etag-{key}\"")),
636 last_modified_ms: 1_760_000_000_000,
637 }
638 }
639
640 const MB: u64 = 1024 * 1024;
641
642 #[test]
644 fn the_open_cost_floor_rounds_down() {
645 let target = "1.1MiB".parse::<bytesize::ByteSize>().unwrap().as_u64();
646 assert_ne!(target % 16, 0);
647 let listing: Vec<ObjectEntry> = (0..32).map(|i| entry(&format!("k{i:02}"), 1)).collect();
648 let bins: Vec<usize> = object_bins(&pack(listing, &whole(target)))
649 .iter()
650 .map(Vec::len)
651 .collect();
652 assert_eq!(bins, [16, 16]);
653 }
654
655 #[test]
658 fn whole_object_bins_are_pinned() {
659 let keys = |packed: Vec<Packed>| -> Vec<Vec<String>> {
660 object_bins(&packed)
661 .into_iter()
662 .map(|b| b.into_iter().map(|e| e.key).collect())
663 .collect()
664 };
665 for packing in [whole(64 * MB), cut(64 * MB)] {
666 let first_fit = vec![
668 entry("a", 40 * MB),
669 entry("b", 50 * MB),
670 entry("c", 10 * MB),
671 ];
672 assert_eq!(keys(pack(first_fit, &packing)), [vec!["a", "c"], vec!["b"]]);
673
674 let mut evicted = vec![entry("a", 40 * MB)];
676 evicted.extend((0..10).map(|i| entry(&format!("m{i}"), 60 * MB)));
677 evicted.push(entry("z", 20 * MB));
678 assert_eq!(keys(pack(evicted, &packing)).len(), 12, "{packing:?}");
679
680 let mut exact = vec![entry("a", 40 * MB), entry("b", 64 * MB)];
683 exact.extend((0..9).map(|i| entry(&format!("m{i}"), 60 * MB)));
684 exact.push(entry("z", 20 * MB));
685 let bins = keys(pack(exact, &packing));
686 assert_eq!(bins[0], ["a", "z"], "{packing:?}");
687 }
688 }
689
690 fn reference_pack(entries: Vec<ObjectEntry>, target_bytes: u64) -> Vec<Vec<ObjectEntry>> {
693 struct Bin {
694 members: Vec<ObjectEntry>,
695 cost: u64,
696 }
697 let floor = (target_bytes / 16).max(1);
698 let mut bins: Vec<Bin> = Vec::new();
699 let mut open: VecDeque<usize> = VecDeque::new();
700 for entry in entries {
701 let cost = entry.size.max(floor);
702 let idx = match open
703 .iter()
704 .position(|&i| bins[i].cost.saturating_add(cost) <= target_bytes)
705 {
706 Some(pos) => open[pos],
707 None => {
708 if open.len() == 10 {
709 open.pop_front();
710 }
711 bins.push(Bin {
712 members: Vec::new(),
713 cost: 0,
714 });
715 open.push_back(bins.len() - 1);
716 bins.len() - 1
717 }
718 };
719 bins[idx].members.push(entry);
720 bins[idx].cost = bins[idx].cost.saturating_add(cost);
721 if bins[idx].cost >= target_bytes
722 && let Some(pos) = open.iter().position(|&i| i == idx)
723 {
724 open.remove(pos);
725 }
726 }
727 bins.into_iter().map(|b| b.members).collect()
728 }
729
730 fn whole(target: u64) -> Packing {
732 Packing {
733 target_bytes: target,
734 compression: Compression::Auto,
735 delimiter: None,
736 }
737 }
738
739 fn cut(target: u64) -> Packing {
741 Packing {
742 delimiter: Some(b'\n'),
743 ..whole(target)
744 }
745 }
746
747 fn object_bins(packed: &[Packed]) -> Vec<Vec<ObjectEntry>> {
749 packed
750 .iter()
751 .filter_map(|p| match p {
752 Packed::Objects(members) => Some(members.clone()),
753 Packed::Range(..) => None,
754 })
755 .collect()
756 }
757
758 #[test]
761 fn digest_id_is_pinned() {
762 let id = split_id_for([
766 ("exports/2026/part-000.ndjson", Some("\"9b2cf5\"")),
767 ("exports/2026/part-001.ndjson", None),
768 ])
769 .unwrap();
770 assert_eq!(id.as_str(), "s3-xje_MfL-wSNqMCft5ellCg");
771 }
772
773 #[test]
774 fn digest_is_order_insensitive_but_content_sensitive() {
775 let forward = split_id_for([("a", Some("1")), ("b", Some("2"))]).unwrap();
776 let reversed = split_id_for([("b", Some("2")), ("a", Some("1"))]).unwrap();
777 assert_eq!(forward, reversed);
778
779 let other_key = split_id_for([("a", Some("1")), ("c", Some("2"))]).unwrap();
780 assert_ne!(forward, other_key);
781 }
782
783 #[test]
784 fn digest_changes_when_an_etag_changes() {
785 let before = split_id_for([("a", Some("v1")), ("b", Some("x"))]).unwrap();
786 let overwritten = split_id_for([("a", Some("v2")), ("b", Some("x"))]).unwrap();
787 let dropped = split_id_for([("a", None), ("b", Some("x"))]).unwrap();
788 assert_ne!(before, overwritten);
789 assert_ne!(before, dropped);
790 }
791
792 #[test]
793 fn digest_folds_in_the_packing_version() {
794 let v1 = split_id_with_version([("a", Some("1"))], 1).unwrap();
795 let v2 = split_id_with_version([("a", Some("1"))], 2).unwrap();
796 assert_ne!(v1, v2);
797 }
798
799 #[test]
800 fn empty_member_set_is_rejected() {
801 assert!(split_id_for(std::iter::empty()).is_err());
802 }
803
804 #[test]
805 fn ambiguous_concatenations_do_not_collide() {
806 let a = split_id_for([("ab", Some("c"))]).unwrap();
809 let b = split_id_for([("a", Some("bc"))]).unwrap();
810 let c = split_id_for([("abc", None)]).unwrap();
811 assert_ne!(a, b);
812 assert_ne!(a, c);
813 assert_ne!(b, c);
814 }
815
816 #[test]
819 fn descriptor_round_trips() {
820 let desc = SplitDescriptor::from_entries(&[
821 entry("exports/part-000.ndjson.gz", 52 * MB),
822 ObjectEntry {
823 key: "exports/part-001.ndjson.gz".to_string(),
824 size: 9,
825 etag: None,
826 last_modified_ms: 1,
827 },
828 ]);
829 let decoded = SplitDescriptor::decode(&desc.encode().unwrap()).unwrap();
830 assert_eq!(decoded, desc);
831 assert_eq!(decoded.v, DESCRIPTOR_VERSION);
832 }
833
834 #[test]
835 fn descriptor_encoding_is_pinned() {
836 let desc = SplitDescriptor::from_entries(&[entry("k", 5)]);
840 assert_eq!(
841 String::from_utf8(desc.encode().unwrap()).unwrap(),
842 r#"{"v":2,"objects":[{"key":"k","size":5,"etag":"\"etag-k\"","last_modified_ms":1760000000000}]}"#,
843 );
844 }
845
846 #[test]
847 fn encode_refuses_a_descriptor_not_built_by_new() {
848 let rogue = SplitDescriptor {
851 v: 0,
852 objects: vec![],
853 range: None,
854 };
855 let err = rogue.encode().unwrap_err();
856 assert_eq!(err.kind, CoordinationErrorKind::Fatal);
857 assert!(
858 err.reason.contains("SplitDescriptor::new"),
859 "reason: {}",
860 err.reason
861 );
862 assert_eq!(SplitDescriptor::new(vec![]).version(), DESCRIPTOR_VERSION);
863 }
864
865 #[test]
868 fn a_version_1_descriptor_is_refused() {
869 let err = SplitDescriptor::decode(
870 br#"{"v":1,"objects":[{"key":"k","size":5,"etag":null,"last_modified_ms":1}]}"#,
871 )
872 .unwrap_err();
873 assert_eq!(err.kind, CoordinationErrorKind::Fatal);
874 assert!(err.reason.contains("version 1"), "reason: {}", err.reason);
875 }
876
877 #[test]
878 fn unknown_descriptor_version_is_rejected_actionably() {
879 let err = SplitDescriptor::decode(br#"{"v":999,"objects":[]}"#).unwrap_err();
880 assert_eq!(err.kind, CoordinationErrorKind::Fatal);
881 assert!(err.reason.contains("version 999"), "reason: {}", err.reason);
882 assert!(
883 err.reason.contains("incompatible release"),
884 "reason: {}",
885 err.reason
886 );
887
888 let garbage = SplitDescriptor::decode(b"not json").unwrap_err();
889 assert_eq!(garbage.kind, CoordinationErrorKind::Fatal);
890 }
891
892 fn ranged_object(size: u64, etag: Option<&str>) -> DescriptorObject {
893 DescriptorObject {
894 key: "big.ndjson".to_string(),
895 size,
896 etag: etag.map(str::to_owned),
897 last_modified_ms: 7,
898 }
899 }
900
901 #[test]
902 fn ranged_descriptor_round_trips_and_its_encoding_is_pinned() {
903 let desc = SplitDescriptor::with_range(
904 ranged_object(100, Some("\"e\"")),
905 SplitRange::new(40, 80, b'\n'),
906 );
907 let encoded = desc.encode().unwrap();
908 assert_eq!(
909 String::from_utf8(encoded.clone()).unwrap(),
910 r#"{"v":2,"objects":[{"key":"big.ndjson","size":100,"etag":"\"e\"","last_modified_ms":7}],"range":{"start":40,"end":80,"delimiter":10}}"#,
911 );
912 assert_eq!(SplitDescriptor::decode(&encoded).unwrap(), desc);
913 }
914
915 #[test]
918 fn invalid_ranged_descriptors_are_refused_on_both_sides() {
919 let pinned = || ranged_object(100, Some("\"e\""));
920 let two_members = SplitDescriptor {
921 objects: vec![pinned(), pinned()],
922 ..SplitDescriptor::with_range(pinned(), SplitRange::new(0, 10, b'\n'))
923 };
924 let no_members = SplitDescriptor {
925 objects: vec![],
926 ..SplitDescriptor::with_range(pinned(), SplitRange::new(0, 10, b'\n'))
927 };
928 let cases = [
929 ("member", two_members),
930 ("member", no_members),
931 (
932 "ETag",
933 SplitDescriptor::with_range(
934 ranged_object(100, None),
935 SplitRange::new(0, 10, b'\n'),
936 ),
937 ),
938 (
939 "empty",
940 SplitDescriptor::with_range(pinned(), SplitRange::new(10, 10, b'\n')),
941 ),
942 (
943 "empty",
944 SplitDescriptor::with_range(pinned(), SplitRange::new(11, 10, b'\n')),
945 ),
946 (
947 "past the object",
948 SplitDescriptor::with_range(pinned(), SplitRange::new(0, 101, b'\n')),
949 ),
950 ];
951 for (needle, desc) in cases {
952 let err = desc.encode().unwrap_err();
953 assert_eq!(err.kind, CoordinationErrorKind::Fatal);
954 assert!(err.reason.contains(needle), "encode: {}", err.reason);
955 let raw = serde_json::to_vec(&desc).unwrap();
957 let err = SplitDescriptor::decode(&raw).unwrap_err();
958 assert_eq!(err.kind, CoordinationErrorKind::Fatal);
959 assert!(err.reason.contains(needle), "decode: {}", err.reason);
960 }
961 let whole = SplitDescriptor::with_range(pinned(), SplitRange::new(0, 100, b'\n'));
963 SplitDescriptor::decode(&whole.encode().unwrap()).unwrap();
964 }
965
966 #[test]
967 fn ranged_id_is_pinned() {
968 let id = split_id_for_range(
971 "exports/2026/part-000.ndjson",
972 "\"9b2cf5\"",
973 SplitRange::new(0, 64 * MB, b'\n'),
974 )
975 .unwrap();
976 assert_eq!(id.as_str(), "s3-fA0wNRW2MyDxquwFvNy0JA");
977 }
978
979 #[test]
982 fn ranged_ids_digest_every_input_and_never_equal_member_set_ids() {
983 let base = split_id_for_range("k", "e", SplitRange::new(0, 10, b'\n')).unwrap();
984 for other in [
985 split_id_for_range("k2", "e", SplitRange::new(0, 10, b'\n')),
986 split_id_for_range("k", "e2", SplitRange::new(0, 10, b'\n')),
987 split_id_for_range("k", "e", SplitRange::new(1, 10, b'\n')),
988 split_id_for_range("k", "e", SplitRange::new(0, 11, b'\n')),
989 split_id_for_range("k", "e", SplitRange::new(0, 10, b';')),
990 ] {
991 assert_ne!(other.unwrap(), base);
992 }
993 assert_ne!(split_id_for([("k", Some("e"))]).unwrap(), base);
994 assert!(split_id_for_range("k", "e", SplitRange::new(10, 10, b'\n')).is_err());
995 }
996
997 fn shaped(i: usize, size: u64, kind: u8) -> ObjectEntry {
1002 let key = format!(
1003 "k{i:04}.ndjson{}",
1004 ["", ".gz", ".zst", ""][kind as usize % 4]
1005 );
1006 ObjectEntry {
1007 etag: (kind % 4 != 3).then(|| format!("\"etag-{i}\"")),
1008 ..entry(&key, size)
1009 }
1010 }
1011
1012 #[test]
1013 fn packing_is_deterministic_for_a_fixed_listing() {
1014 let listing: Vec<ObjectEntry> = (0..200)
1015 .map(|i| entry(&format!("k{i:04}"), (i % 90) * MB))
1016 .collect();
1017 let a = pack(listing.clone(), &cut(64 * MB));
1018 let b = pack(listing, &cut(64 * MB));
1019 assert_eq!(a, b);
1020 }
1021
1022 #[test]
1023 fn tiny_objects_coalesce_under_the_open_cost_floor() {
1024 let listing: Vec<ObjectEntry> = (0..64).map(|i| entry(&format!("k{i:02}"), 1)).collect();
1027 let bins = object_bins(&pack(listing, &whole(64 * MB)));
1028 assert_eq!(bins.len(), 4);
1029 assert!(bins.iter().all(|b| b.len() == 16));
1030 }
1031
1032 #[test]
1033 fn adversarial_listing_sizes_do_not_overflow_the_fit_test() {
1034 let listing = vec![
1037 entry("a", 10 * MB),
1038 entry("huge", u64::MAX),
1039 entry("z", 10 * MB),
1040 ];
1041 let packed = pack(listing, &cut(64 * MB));
1042 let bins = object_bins(&packed);
1043 assert_eq!(bins.len(), packed.len(), "nothing is cut");
1044 let huge_bin = bins
1045 .iter()
1046 .find(|b| b.iter().any(|e| e.key == "huge"))
1047 .unwrap();
1048 assert_eq!(huge_bin.len(), 1, "the oversized object lands alone");
1049 let total: usize = bins.iter().map(Vec::len).sum();
1050 assert_eq!(total, 3, "nothing lost, nothing duplicated");
1051 }
1052
1053 #[test]
1056 fn an_object_above_the_subdivision_limit_stays_whole() {
1057 let target = 1 << 30;
1058 let six_tib = pack(vec![entry("6tib", 6 << 40)], &cut(target));
1059 assert_eq!(six_tib.len(), 6 << 10);
1060 let over = entry("over", MAX_SUBDIVIDABLE_BYTES + 1);
1061 assert_eq!(
1062 pack(vec![over.clone()], &cut(target)),
1063 [Packed::Objects(vec![over])]
1064 );
1065 let at = pack(vec![entry("at", MAX_SUBDIVIDABLE_BYTES)], &cut(target));
1066 assert_eq!(at.len() as u64, MAX_SUBDIVIDABLE_BYTES / target);
1067 assert!(at.iter().all(|p| matches!(p, Packed::Range(..))));
1068 }
1069
1070 #[test]
1074 fn objects_that_cannot_be_cut_stay_whole() {
1075 let target = 64 * MB;
1076 let plain = entry("plain.ndjson", 3 * target);
1077 let gzip_config = Packing {
1078 compression: Compression::Gzip,
1079 ..cut(target)
1080 };
1081 let no_etag = ObjectEntry {
1082 etag: None,
1083 ..plain.clone()
1084 };
1085 for (packing, object) in [
1086 (cut(target), entry("gz.ndjson.gz", 3 * target)),
1087 (cut(target), entry("zst.ndjson.zst", 3 * target)),
1088 (gzip_config, plain.clone()),
1089 (cut(target), no_etag),
1090 (whole(target), plain.clone()),
1091 (cut(target), entry("at.ndjson", target)),
1092 ] {
1093 assert_eq!(
1094 pack(vec![object.clone()], &packing),
1095 [Packed::Objects(vec![object.clone()])],
1096 "{packing:?} {object:?}"
1097 );
1098 }
1099 assert_eq!(
1100 pack(vec![entry("over.ndjson", target + 1)], &cut(target)).len(),
1101 2,
1102 "one byte above the target is cut"
1103 );
1104 }
1105
1106 #[test]
1107 fn oversized_object_gets_its_own_split() {
1108 let listing = vec![
1109 entry("a", 10 * MB),
1110 entry("huge", 500 * MB),
1111 entry("b", 10 * MB),
1112 ];
1113 let bins = object_bins(&pack(listing, &whole(64 * MB)));
1114 let huge_bin = bins
1115 .iter()
1116 .find(|b| b.iter().any(|e| e.key == "huge"))
1117 .unwrap();
1118 assert_eq!(
1119 huge_bin.len(),
1120 1,
1121 "an oversized object never shares a split"
1122 );
1123 }
1124
1125 #[test]
1128 fn a_large_plain_object_is_cut_in_place() {
1129 let listing = vec![
1130 entry("a", 60 * MB),
1131 entry("huge", 500 * MB),
1132 entry("z", 60 * MB),
1133 ];
1134 let packed = pack(listing, &cut(64 * MB));
1135 let keys: Vec<&str> = packed
1136 .iter()
1137 .map(|p| match p {
1138 Packed::Objects(members) => members[0].key.as_str(),
1139 Packed::Range(entry, _) => entry.key.as_str(),
1140 })
1141 .collect();
1142 let mut want = vec!["a"];
1143 want.extend(["huge"; 8]);
1144 want.push("z");
1145 assert_eq!(keys, want);
1146 }
1147
1148 #[test]
1152 fn ranges_tile_the_object_within_half_to_one_target() {
1153 for target in 1..=64u64 {
1154 for size in target + 1..=8 * target {
1155 assert_tiles(size, target);
1156 }
1157 }
1158 assert_tiles(MAX_SUBDIVIDABLE_BYTES, MB);
1159 assert_tiles(MAX_SUBDIVIDABLE_BYTES - 1, MB);
1160 }
1161
1162 fn assert_tiles(size: u64, target: u64) {
1163 let mut count = 0;
1164 let mut next = 0;
1165 for range in tile(size, target, b'\n') {
1166 count += 1;
1167 assert_eq!(range.start, next, "size {size} target {target}");
1168 let len = range.end - range.start;
1169 assert!(
1170 target / 2 <= len && len <= target,
1171 "size {size} target {target}: range {range:?} is {len} bytes"
1172 );
1173 next = range.end;
1174 }
1175 assert_eq!(count, size.div_ceil(target), "size {size} target {target}");
1176 assert_eq!(next, size, "size {size} target {target}");
1177 }
1178
1179 #[test]
1180 fn packing_preserves_listing_order_within_and_across_bins() {
1181 let listing: Vec<ObjectEntry> = (0..50)
1182 .map(|i| {
1183 entry(
1184 &format!("k{i:02}"),
1185 if i % 7 == 0 { 60 * MB } else { 3 * MB },
1186 )
1187 })
1188 .collect();
1189 let bins = object_bins(&pack(listing, &whole(64 * MB)));
1190 for bin in &bins {
1191 let keys: Vec<&str> = bin.iter().map(|e| e.key.as_str()).collect();
1192 let mut sorted = keys.clone();
1193 sorted.sort_unstable();
1194 assert_eq!(keys, sorted, "members stay in listing order within a bin");
1195 }
1196 let firsts: Vec<&str> = bins.iter().map(|b| b[0].key.as_str()).collect();
1197 let mut sorted = firsts.clone();
1198 sorted.sort_unstable();
1199 assert_eq!(
1200 firsts, sorted,
1201 "bins emerge in listing order of their first member"
1202 );
1203 }
1204
1205 #[test]
1206 fn lookback_bounds_how_long_a_bin_stays_open() {
1207 let mut listing = vec![entry("a-half-full", 40 * MB)];
1210 for i in 0..PACKING_LOOKBACK + 2 {
1213 listing.push(entry(&format!("b{i:02}"), 60 * MB));
1214 }
1215 listing.sort_unstable_by(|a, b| a.key.cmp(&b.key));
1216 let bins = object_bins(&pack(listing, &whole(64 * MB)));
1217 assert_eq!(bins.len(), PACKING_LOOKBACK + 3);
1220 assert!(bins.iter().all(|b| b.len() == 1));
1221 }
1222
1223 proptest! {
1224 #[test]
1227 fn prop_packing_partitions_the_listing_exactly(
1228 objects in proptest::collection::vec((0u64..300 * MB, 0u8..4), 0..120),
1229 target_mb in 1u64..129,
1230 ) {
1231 let listing: Vec<ObjectEntry> = objects
1232 .iter()
1233 .enumerate()
1234 .map(|(i, &(size, kind))| shaped(i, size, kind))
1235 .collect();
1236 let packed = pack(listing.clone(), &cut(target_mb * MB));
1237 for object in &listing {
1238 let whole_count = object_bins(&packed)
1239 .iter()
1240 .flatten()
1241 .filter(|e| *e == object)
1242 .count();
1243 let ranges: Vec<SplitRange> = packed
1244 .iter()
1245 .filter_map(|p| match p {
1246 Packed::Range(e, range) if **e == *object => Some(*range),
1247 _ => None,
1248 })
1249 .collect();
1250 if ranges.is_empty() {
1251 prop_assert_eq!(whole_count, 1, "{:?}", object);
1252 } else {
1253 prop_assert_eq!(whole_count, 0, "{:?}", object);
1254 let mut next = 0;
1255 for range in ranges {
1256 prop_assert_eq!(range.start, next);
1257 next = range.end;
1258 }
1259 prop_assert_eq!(next, object.size);
1260 }
1261 }
1262 prop_assert!(object_bins(&packed).iter().all(|b| !b.is_empty()));
1263 }
1264
1265 #[test]
1268 fn prop_small_object_bins_do_not_depend_on_cutting(
1269 objects in proptest::collection::vec((0u64..300 * MB, 0u8..4), 0..120),
1270 target_mb in 1u64..129,
1271 ) {
1272 let listing: Vec<ObjectEntry> = objects
1273 .iter()
1274 .enumerate()
1275 .map(|(i, &(size, kind))| shaped(i, size, kind))
1276 .collect();
1277 let packing = cut(target_mb * MB);
1278 let expanded: Vec<Packed> = pack(listing.clone(), &whole(target_mb * MB))
1279 .into_iter()
1280 .flat_map(|split| match &split {
1281 Packed::Objects(members) if members.len() == 1 => {
1282 match packing.delimiter_for(&members[0]) {
1283 Some(delimiter) => tile(members[0].size, packing.target_bytes, delimiter)
1284 .map(|range| Packed::Range(Rc::new(members[0].clone()), range))
1285 .collect(),
1286 None => vec![split],
1287 }
1288 }
1289 _ => vec![split],
1290 })
1291 .collect();
1292 prop_assert_eq!(pack(listing, &packing), expanded);
1293 }
1294
1295 #[test]
1298 fn prop_whole_object_packing_matches_the_reference(
1299 objects in proptest::collection::vec((0u64..300, 0u8..4), 0..120),
1300 target_mb in 1u64..129,
1301 ) {
1302 let listing: Vec<ObjectEntry> = objects
1303 .iter()
1304 .enumerate()
1305 .map(|(i, &(size_mb, kind))| shaped(i, size_mb * MB, kind))
1306 .collect();
1307 let packed = pack(listing.clone(), &whole(target_mb * MB));
1308 prop_assert_eq!(object_bins(&packed).len(), packed.len());
1309 prop_assert_eq!(object_bins(&packed), reference_pack(listing, target_mb * MB));
1310 }
1311
1312 #[test]
1313 fn prop_ranges_tile_large_objects(
1314 target in 1u64..=1 << 32,
1315 n in 2u64..=1 << 12,
1316 extra in any::<u64>(),
1317 ) {
1318 assert_tiles((n - 1) * target + 1 + extra % target, target);
1320 assert_tiles(n * target, target);
1321 }
1322
1323 #[test]
1324 fn prop_member_count_is_bounded_by_the_floor(
1325 sizes in proptest::collection::vec(0u64..300 * MB, 0..120),
1326 ) {
1327 let target = 64 * MB;
1328 let listing: Vec<ObjectEntry> = sizes
1329 .iter()
1330 .enumerate()
1331 .map(|(i, &s)| entry(&format!("k{i:04}"), s))
1332 .collect();
1333 let bins = object_bins(&pack(listing, &cut(target)));
1334 prop_assert!(bins.iter().all(|b| b.len() <= 16));
1337 }
1338
1339 #[test]
1340 fn prop_split_ids_are_valid_for_arbitrary_keys(
1341 keys in proptest::collection::btree_set("[ -~]{1,64}", 1..8),
1342 etag in proptest::option::of("[ -~]{1,16}"),
1343 ) {
1344 let members: Vec<(&str, Option<&str>)> =
1347 keys.iter().map(|k| (k.as_str(), etag.as_deref())).collect();
1348 let id = split_id_for(members).unwrap();
1349 prop_assert!(id.as_str().starts_with("s3-"));
1350 prop_assert_eq!(id.as_str().len(), 25);
1351 }
1352
1353 #[test]
1354 fn prop_packing_and_ids_are_deterministic(
1355 sizes in proptest::collection::vec(0u64..200 * MB, 1..60),
1356 ) {
1357 let listing: Vec<ObjectEntry> = sizes
1358 .iter()
1359 .enumerate()
1360 .map(|(i, &s)| entry(&format!("k{i:04}"), s))
1361 .collect();
1362 let ids = |packed: &[Packed]| -> Vec<SplitId> {
1363 packed.iter().map(|p| p.id().unwrap()).collect()
1364 };
1365 let a = pack(listing.clone(), &cut(32 * MB));
1366 let b = pack(listing, &cut(32 * MB));
1367 prop_assert_eq!(ids(&a), ids(&b));
1368 }
1369 }
1370}