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
//! DUR-028 R2: one namespace per conversation for the keys it writes on its
//! checkpoint route.
//!
//! Haematite routes a routed key to a shard by its route and stores the key
//! exactly as given (haematite 0.21.0 `api/kv.rs` `delete_routed_range`,
//! `range_routed`): the route names the shard, never a namespace. A key that
//! carries no conversation id is therefore one key for every conversation on
//! the shard. The families that carried none (the checkpoint locator, bodies,
//! body parts and prune receipts) are written under
//! [`conversation_namespace`]: `conv/`, the id as 8 big-endian bytes, `/`.
//! The id's width is fixed, so no conversation's namespace is a prefix of
//! another's, and [`conversation_window`] holds exactly one conversation's
//! keys.
/// A half-open `[from, to)` window of physical keys on a route.
pub(super) type KeyWindow = (Vec<u8>, Vec<u8>);
/// What every namespace begins with.
const NAMESPACE: &[u8] = b"conv/";
/// The byte that closes a namespace.
const NAMESPACE_END: u8 = b'/';
/// `conv/`, the conversation id as 8 big-endian bytes, `/`.
pub(super) fn conversation_namespace(conversation_id: u64) -> Vec<u8> {
let mut key = Vec::with_capacity(NAMESPACE.len() + 9);
key.extend_from_slice(NAMESPACE);
key.extend_from_slice(&conversation_id.to_be_bytes());
key.push(NAMESPACE_END);
key
}
/// `key` under the conversation's namespace.
pub(super) fn namespaced(conversation_id: u64, key: &[u8]) -> Vec<u8> {
let mut namespaced = conversation_namespace(conversation_id);
namespaced.extend_from_slice(key);
namespaced
}
/// The half-open window `[from, to)` that holds exactly the keys under the
/// conversation's namespace: its closing byte raised by one bounds it above.
pub(super) fn conversation_window(conversation_id: u64) -> KeyWindow {
let from = conversation_namespace(conversation_id);
let mut to = from.clone();
to.pop();
to.push(NAMESPACE_END + 1);
(from, to)
}
/// Every window of the conversation's checkpoint route that holds its keys
/// and only its keys (DUR-028 R5): its namespace, the head and rows of each
/// of `streams` (its routed streams, by their full names), its version-2 and
/// version-1 history nodes, and its verified history records.
///
/// # Errors
/// A stream name the routed stream codec refuses.
pub(super) fn conversation_windows(
conversation_id: u64,
streams: &[String],
) -> Result<Vec<KeyWindow>, super::routed_stream_codec::CodecError> {
let mut windows = vec![
conversation_window(conversation_id),
super::restore_checkpoint_history::node_window(conversation_id),
super::restore_checkpoint_history::v1_node_window(conversation_id),
super::restore_checkpoint_history_capture::verified_entry_window(conversation_id),
];
let root = super::restore_checkpoint_history_capture::verified_root_key(conversation_id);
let mut past_root = root.clone();
past_root.push(0);
windows.push((root, past_root));
for stream in streams {
windows.push(super::routed_stream_codec::stream_window(stream)?);
}
Ok(windows)
}