1#![doc = include_str!("../Documentation.md")]
2#![forbid(unsafe_code)]
3
4use kcode_k1_peering::K1Peering;
5use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem, SubsystemId, TxId};
6use semver::Version;
7use std::sync::{Arc, Mutex, MutexGuard, Weak};
8
9const SUBSYSTEM: &str = "k1-loom-bootstrap";
10
11#[derive(Clone, Debug, Eq, PartialEq)]
12pub struct BeginRecord {
13 archive_sha256: [u8; 32],
14 username: String,
15 full_name: String,
16 public_key: [u8; 32],
17}
18
19impl BeginRecord {
20 pub fn new(
21 archive_sha256: [u8; 32],
22 username: String,
23 full_name: String,
24 public_key: [u8; 32],
25 ) -> Self {
26 Self {
27 archive_sha256,
28 username,
29 full_name,
30 public_key,
31 }
32 }
33
34 pub const fn archive_sha256(&self) -> [u8; 32] {
35 self.archive_sha256
36 }
37
38 pub fn username(&self) -> &str {
39 &self.username
40 }
41
42 pub fn full_name(&self) -> &str {
43 &self.full_name
44 }
45
46 pub const fn public_key(&self) -> [u8; 32] {
47 self.public_key
48 }
49}
50
51#[derive(Clone, Debug, Eq, PartialEq)]
52pub struct ImportedPackage {
53 logical_name: String,
54 version: Version,
55 source_sha256: [u8; 32],
56}
57
58impl ImportedPackage {
59 pub fn new(
60 logical_name: String,
61 version: Version,
62 source_sha256: [u8; 32],
63 ) -> Result<Self, String> {
64 if logical_name.is_empty() {
65 return Err("imported package logical name is empty".to_owned());
66 }
67 if !version.pre.is_empty() || !version.build.is_empty() {
68 return Err("imported package version must be stable".to_owned());
69 }
70 Ok(Self {
71 logical_name,
72 version,
73 source_sha256,
74 })
75 }
76
77 pub fn logical_name(&self) -> &str {
78 &self.logical_name
79 }
80
81 pub fn version(&self) -> &Version {
82 &self.version
83 }
84
85 pub const fn source_sha256(&self) -> [u8; 32] {
86 self.source_sha256
87 }
88}
89
90#[derive(Clone, Debug, Eq, PartialEq)]
91pub struct CompleteRecord {
92 begin: BeginRecord,
93 user_id: [u8; 12],
94 group_id: [u8; 12],
95 profile_id: [u8; 12],
96 root_id: [u8; 12],
97 packages: Vec<ImportedPackage>,
98}
99
100impl CompleteRecord {
101 pub fn new(
102 begin: BeginRecord,
103 user_id: [u8; 12],
104 group_id: [u8; 12],
105 profile_id: [u8; 12],
106 root_id: [u8; 12],
107 mut packages: Vec<ImportedPackage>,
108 ) -> Result<Self, String> {
109 packages.sort_by(|left, right| {
110 left.logical_name
111 .cmp(&right.logical_name)
112 .then_with(|| left.version.cmp(&right.version))
113 });
114 if packages.windows(2).any(|pair| {
115 pair[0].logical_name == pair[1].logical_name && pair[0].version == pair[1].version
116 }) {
117 return Err("duplicate imported package coordinate".to_owned());
118 }
119 Ok(Self {
120 begin,
121 user_id,
122 group_id,
123 profile_id,
124 root_id,
125 packages,
126 })
127 }
128
129 pub fn begin(&self) -> &BeginRecord {
130 &self.begin
131 }
132
133 pub const fn user_id(&self) -> [u8; 12] {
134 self.user_id
135 }
136
137 pub const fn group_id(&self) -> [u8; 12] {
138 self.group_id
139 }
140
141 pub const fn profile_id(&self) -> [u8; 12] {
142 self.profile_id
143 }
144
145 pub const fn root_id(&self) -> [u8; 12] {
146 self.root_id
147 }
148
149 pub fn packages(&self) -> &[ImportedPackage] {
150 &self.packages
151 }
152}
153
154#[derive(Clone, Debug, Eq, PartialEq)]
155pub enum BootstrapStatus {
156 Empty,
157 Begun {
158 record: BeginRecord,
159 transaction: TxId,
160 },
161 Complete {
162 record: CompleteRecord,
163 transaction: TxId,
164 },
165}
166
167#[derive(Clone)]
168pub struct K1BootstrapState {
169 inner: Arc<Inner>,
170}
171
172struct Inner {
173 peering: Arc<K1Peering>,
174 state: Mutex<State>,
175}
176
177struct State {
178 status: BootstrapStatus,
179 available: bool,
180}
181
182struct Handler(Weak<Inner>);
183
184enum Event {
185 Begin(BeginRecord),
186 Complete(CompleteRecord),
187}
188
189impl K1BootstrapState {
190 pub fn open(ordering: Arc<K1TxnOrdering>, peering: Arc<K1Peering>) -> Result<Self, String> {
191 let inner = Arc::new(Inner {
192 peering,
193 state: Mutex::new(State {
194 status: BootstrapStatus::Empty,
195 available: true,
196 }),
197 });
198 let subsystem = SubsystemId::from_str(SUBSYSTEM)?;
199 ordering.register_subsystem(subsystem, None, Arc::new(Handler(Arc::downgrade(&inner))))?;
200 Ok(Self { inner })
201 }
202
203 pub fn status(&self) -> Result<BootstrapStatus, String> {
204 Ok(ready(&self.inner)?.status.clone())
205 }
206
207 pub fn begin(&self, record: BeginRecord) -> Result<TxId, String> {
208 match self.status()? {
209 BootstrapStatus::Begun {
210 record: existing,
211 transaction,
212 } if existing == record => return Ok(transaction),
213 BootstrapStatus::Complete {
214 record: existing,
215 transaction,
216 } if existing.begin == record => return Ok(transaction),
217 BootstrapStatus::Empty => {}
218 _ => return Err("bootstrap Begin conflicts with canonical state".to_owned()),
219 }
220 self.submit(Event::Begin(record.clone()))?;
221 match self.status()? {
222 BootstrapStatus::Begun {
223 record: existing,
224 transaction,
225 } if existing == record => Ok(transaction),
226 BootstrapStatus::Complete {
227 record: existing,
228 transaction,
229 } if existing.begin == record => Ok(transaction),
230 _ => Err("bootstrap Begin lost to conflicting canonical state".to_owned()),
231 }
232 }
233
234 pub fn complete(&self, record: CompleteRecord) -> Result<TxId, String> {
235 match self.status()? {
236 BootstrapStatus::Begun {
237 record: existing, ..
238 } if existing == record.begin => {}
239 BootstrapStatus::Complete {
240 record: existing,
241 transaction,
242 } if existing == record => return Ok(transaction),
243 BootstrapStatus::Empty => return Err("bootstrap has not begun".to_owned()),
244 _ => return Err("bootstrap Complete conflicts with canonical state".to_owned()),
245 }
246 self.submit(Event::Complete(record.clone()))?;
247 match self.status()? {
248 BootstrapStatus::Complete {
249 record: existing,
250 transaction,
251 } if existing == record => Ok(transaction),
252 _ => Err("bootstrap Complete lost to conflicting canonical state".to_owned()),
253 }
254 }
255
256 fn submit(&self, event: Event) -> Result<TxId, String> {
257 let payload = encode(&event)?;
258 self.inner
259 .peering
260 .submit_txn(SubsystemId::from_str(SUBSYSTEM)?, &payload)
261 }
262}
263
264impl Subsystem for Handler {
265 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
266 let inner = self
267 .0
268 .upgrade()
269 .ok_or_else(|| "bootstrap state is unavailable".to_owned())?;
270 let event = decode(payload).map_err(|error| fail(&inner, error))?;
271 let mut state = ready(&inner)?;
272 match (&state.status, event) {
273 (BootstrapStatus::Empty, Event::Begin(record)) => {
274 state.status = BootstrapStatus::Begun {
275 record,
276 transaction: id,
277 };
278 }
279 (BootstrapStatus::Begun { record: begun, .. }, Event::Complete(record))
280 if *begun == record.begin =>
281 {
282 state.status = BootstrapStatus::Complete {
283 record,
284 transaction: id,
285 };
286 }
287 _ => {}
288 }
289 Ok(())
290 }
291
292 fn reorg(&self) -> Result<(), String> {
293 let inner = self
294 .0
295 .upgrade()
296 .ok_or_else(|| "bootstrap state is unavailable".to_owned())?;
297 lock(&inner)?.available = false;
298 Ok(())
299 }
300}
301
302fn lock(inner: &Inner) -> Result<MutexGuard<'_, State>, String> {
303 inner
304 .state
305 .lock()
306 .map_err(|_| "bootstrap state lock poisoned".to_owned())
307}
308
309fn ready(inner: &Inner) -> Result<MutexGuard<'_, State>, String> {
310 let state = lock(inner)?;
311 if state.available {
312 Ok(state)
313 } else {
314 Err("bootstrap state is unavailable".to_owned())
315 }
316}
317
318fn fail(inner: &Inner, error: String) -> String {
319 if let Ok(mut state) = lock(inner) {
320 state.available = false;
321 }
322 error
323}
324
325fn encode(event: &Event) -> Result<Vec<u8>, String> {
326 let mut out = vec![
327 1,
328 match event {
329 Event::Begin(_) => 1,
330 Event::Complete(_) => 2,
331 },
332 ];
333 match event {
334 Event::Begin(record) => put_begin(&mut out, record)?,
335 Event::Complete(record) => {
336 put_begin(&mut out, &record.begin)?;
337 out.extend_from_slice(&record.user_id);
338 out.extend_from_slice(&record.group_id);
339 out.extend_from_slice(&record.profile_id);
340 out.extend_from_slice(&record.root_id);
341 put_u64(&mut out, record.packages.len())?;
342 for package in &record.packages {
343 put_string(&mut out, &package.logical_name)?;
344 put_string(&mut out, &package.version.to_string())?;
345 out.extend_from_slice(&package.source_sha256);
346 }
347 }
348 }
349 Ok(out)
350}
351
352fn put_begin(out: &mut Vec<u8>, record: &BeginRecord) -> Result<(), String> {
353 out.extend_from_slice(&record.archive_sha256);
354 put_string(out, &record.username)?;
355 put_string(out, &record.full_name)?;
356 out.extend_from_slice(&record.public_key);
357 Ok(())
358}
359
360fn put_string(out: &mut Vec<u8>, value: &str) -> Result<(), String> {
361 put_u64(out, value.len())?;
362 out.try_reserve(value.len())
363 .map_err(|_| "bootstrap encoding allocation failed".to_owned())?;
364 out.extend_from_slice(value.as_bytes());
365 Ok(())
366}
367
368fn put_u64(out: &mut Vec<u8>, value: usize) -> Result<(), String> {
369 let value =
370 u64::try_from(value).map_err(|_| "bootstrap encoding length overflow".to_owned())?;
371 out.extend_from_slice(&value.to_le_bytes());
372 Ok(())
373}
374
375fn decode(bytes: &[u8]) -> Result<Event, String> {
376 let mut input = Input { bytes, at: 0 };
377 if input.byte()? != 1 {
378 return Err("unsupported bootstrap state wire version".to_owned());
379 }
380 let kind = input.byte()?;
381 let begin = input.begin()?;
382 let event = match kind {
383 1 => Event::Begin(begin),
384 2 => {
385 let user_id = input.array()?;
386 let group_id = input.array()?;
387 let profile_id = input.array()?;
388 let root_id = input.array()?;
389 let count = input.length()?;
390 let mut packages = Vec::new();
391 packages
392 .try_reserve(count)
393 .map_err(|_| "bootstrap package allocation failed".to_owned())?;
394 for _ in 0..count {
395 packages.push(ImportedPackage::new(
396 input.string()?,
397 input
398 .string()?
399 .parse()
400 .map_err(|_| "invalid bootstrap package version".to_owned())?,
401 input.array()?,
402 )?);
403 }
404 Event::Complete(CompleteRecord::new(
405 begin, user_id, group_id, profile_id, root_id, packages,
406 )?)
407 }
408 _ => return Err("unknown bootstrap state event kind".to_owned()),
409 };
410 if input.at != bytes.len() {
411 return Err("trailing bootstrap state bytes".to_owned());
412 }
413 Ok(event)
414}
415
416struct Input<'a> {
417 bytes: &'a [u8],
418 at: usize,
419}
420
421impl Input<'_> {
422 fn take(&mut self, count: usize) -> Result<&[u8], String> {
423 let end = self
424 .at
425 .checked_add(count)
426 .ok_or_else(|| "bootstrap state length overflow".to_owned())?;
427 let value = self
428 .bytes
429 .get(self.at..end)
430 .ok_or_else(|| "truncated bootstrap state event".to_owned())?;
431 self.at = end;
432 Ok(value)
433 }
434
435 fn byte(&mut self) -> Result<u8, String> {
436 Ok(self.take(1)?[0])
437 }
438
439 fn array<const N: usize>(&mut self) -> Result<[u8; N], String> {
440 self.take(N)?
441 .try_into()
442 .map_err(|_| "invalid bootstrap fixed field".to_owned())
443 }
444
445 fn length(&mut self) -> Result<usize, String> {
446 usize::try_from(u64::from_le_bytes(self.array()?))
447 .map_err(|_| "bootstrap state length overflow".to_owned())
448 }
449
450 fn string(&mut self) -> Result<String, String> {
451 let length = self.length()?;
452 String::from_utf8(self.take(length)?.to_vec())
453 .map_err(|_| "bootstrap state string is not UTF-8".to_owned())
454 }
455
456 fn begin(&mut self) -> Result<BeginRecord, String> {
457 Ok(BeginRecord::new(
458 self.array()?,
459 self.string()?,
460 self.string()?,
461 self.array()?,
462 ))
463 }
464}
465
466#[cfg(test)]
467mod tests {
468 use super::*;
469
470 #[test]
471 fn wire_round_trips_complete_canonically() {
472 let begin = BeginRecord::new([1; 32], "admin".into(), "Admin".into(), [2; 32]);
473 let packages =
474 vec![ImportedPackage::new("loom".into(), Version::new(0, 1, 0), [3; 32]).unwrap()];
475 let complete =
476 CompleteRecord::new(begin, [4; 12], [5; 12], [6; 12], [7; 12], packages).unwrap();
477 let encoded = encode(&Event::Complete(complete.clone())).unwrap();
478 assert!(matches!(decode(&encoded).unwrap(), Event::Complete(found) if found == complete));
479 }
480}