kcode_k1_bootstrap_state/
lib.rs

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

kcode_k1_bootstrap_state/
lib.rs
