1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
//! A checkpoint body larger than one part is stored in parts (DUR-024).
//!
//! The door room's body reached 101 MB on 11 October 2026: one item over
//! the store's 64 MiB bound, refused however the batches around it were
//! laid out. A body over [`BODY_PART_BYTES`] is cut into numbered parts,
//! each stored under its own content address on the conversation's route,
//! and the body key holds a manifest naming the parts in order. The parts
//! are published like history nodes, in the batches before the locator, so
//! the locator still lands last and names a body every part of which is
//! durable. A body of one part is stored whole under the body key, byte for
//! byte as before.
//!
//! The body key stays the digest of the whole body, so a reader that
//! reassembles the parts checks the same digest the locator names.
use liminal::durability::{DurabilityError, DurableStore};
use super::restore_checkpoint_locator::CheckpointBodyKey;
/// The largest part a body is cut into: half the 16 MiB batch target, so
/// two parts pack in one publication batch and every part is far under any
/// store bound.
pub(super) const BODY_PART_BYTES: usize = 8 * 1024 * 1024;
/// What a manifest begins with. A body is never mistaken for a manifest:
/// the reader takes a value as the whole body when its digest matches the
/// body key, and reads it as a manifest only when it does not.
const MANIFEST_MAGIC: &[u8; 20] = b"liminal-body-parts-1";
/// The routed key prefix every body part is written under.
const PART_KEY_PREFIX: &[u8] = b"checkpoint/body/part/v1/";
/// One body, laid out for staging: what goes under the body key, and the
/// parts (key and bytes) a manifest names.
pub(super) struct BodyLayout {
pub(super) head: Vec<u8>,
pub(super) parts: Vec<(Vec<u8>, Vec<u8>)>,
}
/// The routed key of the part with these bytes.
fn part_key(part: &[u8]) -> Vec<u8> {
let mut key = PART_KEY_PREFIX.to_vec();
key.extend_from_slice(blake3::hash(part).as_bytes());
key
}
/// Lays `body` out: whole when it fits one part, otherwise a manifest of
/// its total length and its parts' digests, in order, and the parts.
pub(super) fn lay_out(body: Vec<u8>) -> BodyLayout {
if body.len() <= BODY_PART_BYTES {
return BodyLayout {
head: body,
parts: Vec::new(),
};
}
let parts: Vec<(Vec<u8>, Vec<u8>)> = body
.chunks(BODY_PART_BYTES)
.map(|part| (part_key(part), part.to_vec()))
.collect();
let mut head = MANIFEST_MAGIC.to_vec();
head.extend_from_slice(&u64::try_from(body.len()).unwrap_or(u64::MAX).to_be_bytes());
for (key, _) in &parts {
head.extend_from_slice(&key[PART_KEY_PREFIX.len()..]);
}
BodyLayout { head, parts }
}
/// The total length and the part keys a manifest names, or `None` when the
/// value is not a manifest.
fn manifest(value: &[u8]) -> Option<(usize, Vec<Vec<u8>>)> {
let rest = value.strip_prefix(MANIFEST_MAGIC.as_slice())?;
let (total, digests) = rest.split_at_checked(std::mem::size_of::<u64>())?;
let total = usize::try_from(u64::from_be_bytes(total.try_into().ok()?)).ok()?;
if digests.is_empty() || digests.len() % blake3::OUT_LEN != 0 {
return None;
}
let keys = digests
.chunks(blake3::OUT_LEN)
.map(|digest| {
let mut key = PART_KEY_PREFIX.to_vec();
key.extend_from_slice(digest);
key
})
.collect();
Some((total, keys))
}
/// The body under `body_key` on `route`: the value itself when it is the
/// whole body, the parts reassembled when it is a manifest, and `None` when
/// the body key or any part is absent. A part whose bytes do not match its
/// address, or a reassembly of the wrong length, answers the value as read,
/// so the caller's digest check refuses it by name.
///
/// # Errors
/// The store's own failure.
pub(super) async fn read_body(
store: &dyn DurableStore,
route: &str,
body_key: CheckpointBodyKey,
) -> Result<Option<Vec<u8>>, DurabilityError> {
let Some(head) = store.get_routed(route, &body_key.routed_key()).await? else {
return Ok(None);
};
if body_key.verify_body(&head).is_ok() {
return Ok(Some(head));
}
let Some((total, keys)) = manifest(&head) else {
return Ok(Some(head));
};
let mut body = Vec::with_capacity(total);
for key in keys {
let Some(part) = store.get_routed(route, &key).await? else {
return Ok(None);
};
if part_key(&part) != key {
return Ok(Some(head));
}
body.extend_from_slice(&part);
}
if body.len() != total {
return Ok(Some(head));
}
Ok(Some(body))
}
#[cfg(test)]
#[path = "tests_checkpoint_body_parts.rs"]
mod tests;