mkit_server/store/partition.rs
1//! Storage partitions (D34 shards) and their portable byte encoding.
2
3use bytes::{BufMut, Bytes, BytesMut};
4
5use super::error::StoreError;
6use crate::repo::{NamespaceKey, RepoName};
7
8/// A storage partition: one D34 shard. Everything that must commit
9/// atomically lives in one partition. The core computes the partition of
10/// every operation; a backend maps partitions to whatever it likes (a
11/// Durable Object each, rows keyed by partition in `SQLite`, a qmdb
12/// instance each).
13///
14/// **Catch-all rule for backends.** The enum is `#[non_exhaustive]`: later
15/// work adds kinds. A backend outside this crate never matches on it. It
16/// stores and names partitions by [`Partition::encode`], which is stable
17/// and injective, so a new kind needs no backend change. Existing
18/// encodings never change; a new kind gets a new tag.
19///
20/// **Enumeration.** A store is never asked to list its partitions (a
21/// Durable Object namespace cannot list its instances). Backup and export
22/// enumerate them hierarchically, from bounded or constructible structures
23/// only; nothing lists every ref shard in one place (one ref per file can
24/// mean millions of them):
25/// 1. **Namespaces** (`Namespace` in single-partition mode, `Coordinator`
26/// under D34) come from deployment configuration and the namespace
27/// allowlist. Under `namespace_policy = any` the deployment keeps a
28/// namespace list (reserved key class `nl`, `store::keys`); a backend
29/// MAY keep it in its own metadata instead.
30/// 2. **Repos**: each coordinator keeps a repo registry, one row per repo
31/// of the namespace (reserved key class `rr`; WP-1.22 lays it out).
32/// 3. **Repo index shards** (`RepoIndex`, `RefIndex`) are enumerable by
33/// construction: their object-id-prefix and ref-name-hash fan-outs are
34/// fixed deployment constants.
35/// 4. **Ref shards** are tracked by the coordinator's active-shard table;
36/// WP-5.3a completes GC enumeration, including deleted-ref and ticket-only shards.
37/// 5. **Content shards** are enumerable by construction (`INDEX_FANOUT`).
38#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
39#[non_exhaustive]
40pub enum Partition {
41 /// The whole namespace: single-partition mode. Used by M0 (today's
42 /// single `root` Durable Object) and by the ssh / fs-layout path for
43 /// good; not by D34-sharded deployments.
44 Namespace(NamespaceKey),
45 /// D34 (M1): the namespace coordinator: config, the grant epoch, the
46 /// table of currently epoch-leased shards (bounded by active shards)
47 /// and the repo registry. Rarely written.
48 Coordinator(NamespaceKey),
49 /// D34 (M1): one per (repo, ref). A branch's head and packmap share it
50 /// (`shard_ref` is the `refs/heads/<x>` name). Strongly consistent.
51 Ref {
52 /// Namespace.
53 ns: NamespaceKey,
54 /// Repository.
55 repo: RepoName,
56 /// The ref whose shard this is.
57 shard_ref: String,
58 },
59 /// D34 (M1): repo membership by object-id prefix over the fixed
60 /// `INDEX_FANOUT` (default 4096). Never resharded; eventually
61 /// consistent.
62 RepoIndex {
63 /// Namespace.
64 ns: NamespaceKey,
65 /// Repository.
66 repo: RepoName,
67 /// Object-id prefix bucket.
68 prefix: u16,
69 },
70 /// D34 (M1): the ref-name index `ListRefs` reads, hash-sharded over the
71 /// fixed `REF_INDEX_FANOUT` (default 16). Eventually consistent.
72 RefIndex {
73 /// Namespace.
74 ns: NamespaceKey,
75 /// Repository.
76 repo: RepoName,
77 /// Ref-name hash bucket.
78 bucket: u16,
79 },
80 /// A global `ContentIndex` shard, by object-id prefix over
81 /// `INDEX_FANOUT`.
82 ContentShard(u16),
83}
84
85impl Partition {
86 /// Stable, low-cardinality partition kind for metrics and alerts.
87 #[must_use]
88 pub const fn kind(&self) -> &'static str {
89 match self {
90 Self::Namespace(_) => "namespace",
91 Self::Coordinator(_) => "coordinator",
92 Self::Ref { .. } => "ref",
93 Self::RepoIndex { .. } => "repo_index",
94 Self::RefIndex { .. } => "ref_index",
95 Self::ContentShard(_) => "content",
96 }
97 }
98
99 /// The portable encoding: one kind tag byte (`n` namespace, `c`
100 /// coordinator, `r` ref, `i` repo index, `x` ref index, `s` content
101 /// shard), then each component followed by `0x00`. Strings are their
102 /// UTF-8 bytes; integers are canonical decimal ASCII. Injective, so a
103 /// backend may use it as an opaque name.
104 ///
105 /// # Errors
106 /// [`StoreError::Invalid`] if a component contains `0x00`.
107 pub fn encode(&self) -> Result<Bytes, StoreError> {
108 let (tag, parts): (u8, Vec<String>) = match self {
109 Self::Namespace(ns) => (b'n', vec![ns.as_str().into()]),
110 Self::Coordinator(ns) => (b'c', vec![ns.as_str().into()]),
111 Self::Ref {
112 ns,
113 repo,
114 shard_ref,
115 } => (
116 b'r',
117 vec![ns.as_str().into(), repo.as_str().into(), shard_ref.clone()],
118 ),
119 Self::RepoIndex { ns, repo, prefix } => (
120 b'i',
121 vec![ns.as_str().into(), repo.as_str().into(), prefix.to_string()],
122 ),
123 Self::RefIndex { ns, repo, bucket } => (
124 b'x',
125 vec![ns.as_str().into(), repo.as_str().into(), bucket.to_string()],
126 ),
127 Self::ContentShard(prefix) => (b's', vec![prefix.to_string()]),
128 };
129 let mut buf = BytesMut::new();
130 buf.put_u8(tag);
131 for part in parts {
132 if part.as_bytes().contains(&0) {
133 return Err(StoreError::Invalid(
134 "partition component contains 0x00".into(),
135 ));
136 }
137 buf.put_slice(part.as_bytes());
138 buf.put_u8(0);
139 }
140 Ok(buf.freeze())
141 }
142
143 /// Decode [`Partition::encode`] output.
144 ///
145 /// # Errors
146 /// [`StoreError::Corrupt`] for an unknown tag, a wrong component count,
147 /// a missing terminator, invalid UTF-8, an invalid repo name or a
148 /// non-canonical integer.
149 pub fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
150 let corrupt = || StoreError::Corrupt("malformed partition encoding".into());
151 let (&tag, body) = bytes.split_first().ok_or_else(corrupt)?;
152 let body = body.strip_suffix(&[0]).ok_or_else(corrupt)?;
153 let parts = body
154 .split(|&b| b == 0)
155 .map(|p| String::from_utf8(p.to_vec()).map_err(|_| corrupt()))
156 .collect::<Result<Vec<_>, _>>()?;
157 let ns = |s: &String| NamespaceKey::from_stored(s.clone());
158 let repo = |s: &String| RepoName::new(s.clone()).map_err(|_| corrupt());
159 let int = |s: &String| {
160 s.parse::<u16>()
161 .ok()
162 .filter(|n| n.to_string() == *s)
163 .ok_or_else(corrupt)
164 };
165 Ok(match (tag, parts.as_slice()) {
166 (b'n', [n]) => Self::Namespace(ns(n)),
167 (b'c', [n]) => Self::Coordinator(ns(n)),
168 (b'r', [n, r, shard_ref]) => Self::Ref {
169 ns: ns(n),
170 repo: repo(r)?,
171 shard_ref: shard_ref.clone(),
172 },
173 (b'i', [n, r, p]) => Self::RepoIndex {
174 ns: ns(n),
175 repo: repo(r)?,
176 prefix: int(p)?,
177 },
178 (b'x', [n, r, b]) => Self::RefIndex {
179 ns: ns(n),
180 repo: repo(r)?,
181 bucket: int(b)?,
182 },
183 (b's', [p]) => Self::ContentShard(int(p)?),
184 _ => return Err(corrupt()),
185 })
186 }
187}
188
189#[cfg(test)]
190mod tests {
191 use proptest::prelude::*;
192
193 use super::*;
194
195 fn ns(s: &str) -> NamespaceKey {
196 NamespaceKey::from_stored(s.into())
197 }
198
199 fn repo(s: &str) -> RepoName {
200 RepoName::new(s).unwrap()
201 }
202
203 #[test]
204 fn partition_encoding_golden_bytes() {
205 let cases: [(Partition, &[u8]); 6] = [
206 (Partition::Namespace(ns("root")), b"nroot\0"),
207 (Partition::Coordinator(ns("root")), b"croot\0"),
208 (
209 Partition::Ref {
210 ns: ns("root"),
211 repo: repo("a"),
212 shard_ref: "refs/heads/main".into(),
213 },
214 b"rroot\0a\0refs/heads/main\0",
215 ),
216 (
217 Partition::RepoIndex {
218 ns: ns("root"),
219 repo: repo("a"),
220 prefix: 4095,
221 },
222 b"iroot\0a\x004095\0",
223 ),
224 (
225 Partition::RefIndex {
226 ns: ns("root"),
227 repo: repo("a"),
228 bucket: 0,
229 },
230 b"xroot\0a\x000\0",
231 ),
232 (Partition::ContentShard(7), b"s7\0"),
233 ];
234 for (p, golden) in cases {
235 assert_eq!(p.encode().unwrap().as_ref(), golden);
236 assert_eq!(Partition::decode(golden).unwrap(), p);
237 }
238 let nul = Partition::Ref {
239 ns: ns("root"),
240 repo: repo("a"),
241 shard_ref: "x\0y".into(),
242 };
243 assert!(matches!(nul.encode(), Err(StoreError::Invalid(_))));
244 for bad in [
245 &b""[..],
246 b"nroot",
247 b"s07\0",
248 b"s65536\0",
249 b"q1\0",
250 b"croot\0x\0",
251 b"ir\0 \x001\0",
252 ] {
253 assert!(
254 matches!(Partition::decode(bad), Err(StoreError::Corrupt(_))),
255 "{bad:?}"
256 );
257 }
258 }
259
260 fn partition() -> impl Strategy<Value = Partition> {
261 let s = "[a-z0-9/._-]{0,8}";
262 let r = "[!-~]{1,8}";
263 prop_oneof![
264 s.prop_map(|n| Partition::Namespace(ns(&n))),
265 s.prop_map(|n| Partition::Coordinator(ns(&n))),
266 (s, r, s).prop_map(|(n, rp, sr)| Partition::Ref {
267 ns: ns(&n),
268 repo: repo(&rp),
269 shard_ref: sr,
270 }),
271 (s, r, any::<u16>()).prop_map(|(n, rp, prefix)| Partition::RepoIndex {
272 ns: ns(&n),
273 repo: repo(&rp),
274 prefix,
275 }),
276 (s, r, any::<u16>()).prop_map(|(n, rp, bucket)| Partition::RefIndex {
277 ns: ns(&n),
278 repo: repo(&rp),
279 bucket,
280 }),
281 any::<u16>().prop_map(Partition::ContentShard),
282 ]
283 }
284
285 proptest! {
286 #[test]
287 fn partition_encoding_is_injective(a in partition(), b in partition()) {
288 let (ea, eb) = (a.encode().unwrap(), b.encode().unwrap());
289 prop_assert_eq!(Partition::decode(&ea).unwrap(), a.clone());
290 prop_assert_eq!(ea == eb, a == b);
291 }
292 }
293}