kcode_k1_objects/
lib.rs

1use kcode_k1_peering::K1Peering;
2use kcode_k1_transaction::{REGISTER_AT_TIP, SubsystemId, Transaction};
3pub use kcode_k1_transaction_id::TxId;
4use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem};
5use std::sync::Arc;
6
7const PAYLOAD_VERSION: u8 = 1;
8const PAYLOAD_HEADER_BYTES: usize = 25;
9const OBJECT_SUBSYSTEM_BYTES: [u8; 20] = *b"k1-objects-subsystem";
10
11#[derive(Clone, Debug, Eq, PartialEq)]
12pub struct Object {
13    pub filename: String,
14    pub file_type: String,
15    pub description: String,
16    pub data: Vec<u8>,
17}
18
19pub struct K1Objects {
20    ordering: Arc<K1TxnOrdering>,
21    peering: Arc<K1Peering>,
22}
23
24impl K1Objects {
25    pub fn open(ordering: Arc<K1TxnOrdering>, peering: Arc<K1Peering>) -> Result<Self, String> {
26        ordering.register_subsystem(
27            object_subsystem(),
28            Some(REGISTER_AT_TIP),
29            Arc::new(StatelessObjects),
30        )?;
31        Ok(Self { ordering, peering })
32    }
33
34    pub fn save(
35        &self,
36        filename: &str,
37        file_type: &str,
38        description: &str,
39        data: &[u8],
40    ) -> Result<TxId, String> {
41        let payload = encode_payload(filename, file_type, description, data)?;
42        self.peering.submit_txn(object_subsystem(), &payload)
43    }
44
45    pub fn load(&self, id: TxId) -> Result<Option<Object>, String> {
46        let Some(bytes) = self.ordering.get_txn(id)? else {
47            return Ok(None);
48        };
49        let transaction = Transaction::parse(&bytes)
50            .map_err(|message| format!("canonical transaction is invalid: {message}"))?;
51        if transaction.subsystem() != object_subsystem() {
52            return Ok(None);
53        }
54        decode_payload(transaction.payload()).map(Some)
55    }
56}
57
58struct StatelessObjects;
59
60impl Subsystem for StatelessObjects {
61    fn submit_txn(&self, _id: TxId, _payload: &[u8]) -> Result<(), String> {
62        Ok(())
63    }
64
65    fn reorg(&self) -> Result<(), String> {
66        Ok(())
67    }
68}
69
70fn object_subsystem() -> SubsystemId {
71    SubsystemId::from_bytes(OBJECT_SUBSYSTEM_BYTES)
72        .expect("Objects subsystem ID is valid fixed-width UTF-8")
73}
74
75fn encode_payload(
76    filename: &str,
77    file_type: &str,
78    description: &str,
79    data: &[u8],
80) -> Result<Vec<u8>, String> {
81    let filename_len = u64::try_from(filename.len())
82        .map_err(|_| "filename length exceeds the object wire format".to_owned())?;
83    let file_type_len = u64::try_from(file_type.len())
84        .map_err(|_| "file type length exceeds the object wire format".to_owned())?;
85    let description_len = u64::try_from(description.len())
86        .map_err(|_| "description length exceeds the object wire format".to_owned())?;
87    let total_len = PAYLOAD_HEADER_BYTES
88        .checked_add(filename.len())
89        .and_then(|length| length.checked_add(file_type.len()))
90        .and_then(|length| length.checked_add(description.len()))
91        .and_then(|length| length.checked_add(data.len()))
92        .ok_or_else(|| "object payload length exceeds usize".to_owned())?;
93
94    let mut payload = Vec::new();
95    payload
96        .try_reserve_exact(total_len)
97        .map_err(|error| format!("cannot allocate object payload: {error}"))?;
98    payload.push(PAYLOAD_VERSION);
99    payload.extend_from_slice(&filename_len.to_le_bytes());
100    payload.extend_from_slice(&file_type_len.to_le_bytes());
101    payload.extend_from_slice(&description_len.to_le_bytes());
102    payload.extend_from_slice(filename.as_bytes());
103    payload.extend_from_slice(file_type.as_bytes());
104    payload.extend_from_slice(description.as_bytes());
105    payload.extend_from_slice(data);
106
107    debug_assert_eq!(payload.len(), total_len);
108    Ok(payload)
109}
110
111fn decode_payload(payload: &[u8]) -> Result<Object, String> {
112    if payload.len() < PAYLOAD_HEADER_BYTES {
113        return Err(format!(
114            "object payload must contain at least {PAYLOAD_HEADER_BYTES} bytes"
115        ));
116    }
117    if payload[0] != PAYLOAD_VERSION {
118        return Err(format!("unsupported object payload version {}", payload[0]));
119    }
120
121    let filename_len = decode_length(payload, 1, "filename")?;
122    let file_type_len = decode_length(payload, 9, "file type")?;
123    let description_len = decode_length(payload, 17, "description")?;
124    let filename_end = PAYLOAD_HEADER_BYTES
125        .checked_add(filename_len)
126        .ok_or_else(|| "object metadata length overflow".to_owned())?;
127    let file_type_end = filename_end
128        .checked_add(file_type_len)
129        .ok_or_else(|| "object metadata length overflow".to_owned())?;
130    let description_end = file_type_end
131        .checked_add(description_len)
132        .ok_or_else(|| "object metadata length overflow".to_owned())?;
133
134    if description_end > payload.len() {
135        return Err("object metadata extends past the payload".to_owned());
136    }
137
138    let filename = std::str::from_utf8(&payload[PAYLOAD_HEADER_BYTES..filename_end])
139        .map_err(|error| format!("object filename is not valid UTF-8: {error}"))?
140        .to_owned();
141    let file_type = std::str::from_utf8(&payload[filename_end..file_type_end])
142        .map_err(|error| format!("object file type is not valid UTF-8: {error}"))?
143        .to_owned();
144    let description = std::str::from_utf8(&payload[file_type_end..description_end])
145        .map_err(|error| format!("object description is not valid UTF-8: {error}"))?
146        .to_owned();
147
148    Ok(Object {
149        filename,
150        file_type,
151        description,
152        data: payload[description_end..].to_vec(),
153    })
154}
155
156fn decode_length(payload: &[u8], offset: usize, field: &str) -> Result<usize, String> {
157    let bytes: [u8; 8] = payload[offset..offset + 8]
158        .try_into()
159        .expect("object length fields are inside the fixed header");
160    usize::try_from(u64::from_le_bytes(bytes))
161        .map_err(|_| format!("object {field} length exceeds usize"))
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167    use std::collections::HashSet;
168    use std::fs;
169    use std::path::PathBuf;
170    use std::sync::atomic::{AtomicU64, Ordering};
171    use std::thread;
172
173    static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
174
175    struct TempRoots {
176        base: PathBuf,
177    }
178
179    impl TempRoots {
180        fn new(label: &str) -> Self {
181            let sequence = NEXT_ROOT.fetch_add(1, Ordering::Relaxed);
182            let base = std::env::temp_dir().join(format!(
183                "kcode-k1-objects-{}-{sequence}-{label}",
184                std::process::id()
185            ));
186            let _ = fs::remove_dir_all(&base);
187            Self { base }
188        }
189
190        fn ordering(&self) -> PathBuf {
191            self.base.join("ordering")
192        }
193
194        fn peering(&self) -> PathBuf {
195            self.base.join("peering")
196        }
197    }
198
199    impl Drop for TempRoots {
200        fn drop(&mut self) {
201            let _ = fs::remove_dir_all(&self.base);
202        }
203    }
204
205    struct NoopSubsystem;
206
207    impl Subsystem for NoopSubsystem {
208        fn submit_txn(&self, _id: TxId, _payload: &[u8]) -> Result<(), String> {
209            Ok(())
210        }
211
212        fn reorg(&self) -> Result<(), String> {
213            Ok(())
214        }
215    }
216
217    fn open_stack(roots: &TempRoots) -> (Arc<K1TxnOrdering>, Arc<K1Peering>, K1Objects) {
218        let ordering = Arc::new(K1TxnOrdering::open(&roots.ordering()).unwrap());
219        let peering = Arc::new(K1Peering::open(&roots.peering(), ordering.clone()).unwrap());
220        let objects = K1Objects::open(ordering.clone(), peering.clone()).unwrap();
221        (ordering, peering, objects)
222    }
223
224    fn other_subsystem() -> SubsystemId {
225        SubsystemId::from_bytes([b'o'; 20]).unwrap()
226    }
227
228    fn raw_payload(filename: &[u8], file_type: &[u8], description: &[u8], data: &[u8]) -> Vec<u8> {
229        let mut payload = Vec::new();
230        payload.push(PAYLOAD_VERSION);
231        payload.extend_from_slice(&(filename.len() as u64).to_le_bytes());
232        payload.extend_from_slice(&(file_type.len() as u64).to_le_bytes());
233        payload.extend_from_slice(&(description.len() as u64).to_le_bytes());
234        payload.extend_from_slice(filename);
235        payload.extend_from_slice(file_type);
236        payload.extend_from_slice(description);
237        payload.extend_from_slice(data);
238        payload
239    }
240
241    #[test]
242    fn codec_round_trips_empty_and_opaque_values() {
243        let empty = decode_payload(&encode_payload("", "", "", &[]).unwrap()).unwrap();
244        assert_eq!(
245            empty,
246            Object {
247                filename: String::new(),
248                file_type: String::new(),
249                description: String::new(),
250                data: Vec::new(),
251            }
252        );
253
254        let expected = Object {
255            filename: "résumé-太阳.pdf".to_owned(),
256            file_type: "anything at all".to_owned(),
257            description: "descripción".to_owned(),
258            data: vec![0, 255, 1, 128, 0],
259        };
260        let payload = encode_payload(
261            &expected.filename,
262            &expected.file_type,
263            &expected.description,
264            &expected.data,
265        )
266        .unwrap();
267
268        assert_eq!(payload[0], PAYLOAD_VERSION);
269        assert_eq!(decode_payload(&payload).unwrap(), expected);
270    }
271
272    #[test]
273    fn decoder_rejects_invalid_structure_and_metadata() {
274        assert!(decode_payload(&[]).is_err());
275        assert!(decode_payload(&[PAYLOAD_VERSION; PAYLOAD_HEADER_BYTES - 1]).is_err());
276
277        let mut unsupported = raw_payload(b"", b"", b"", b"");
278        unsupported[0] = 2;
279        assert!(decode_payload(&unsupported).is_err());
280
281        let mut overflow = raw_payload(b"", b"", b"", b"");
282        overflow[1..9].copy_from_slice(&u64::MAX.to_le_bytes());
283        assert!(decode_payload(&overflow).is_err());
284
285        let mut out_of_bounds = raw_payload(b"", b"", b"", b"");
286        out_of_bounds[1..9].copy_from_slice(&1_u64.to_le_bytes());
287        assert!(decode_payload(&out_of_bounds).is_err());
288
289        assert!(decode_payload(&raw_payload(&[0xff], b"", b"", b"")).is_err());
290        assert!(decode_payload(&raw_payload(b"", &[0xff], b"", b"")).is_err());
291        assert!(decode_payload(&raw_payload(b"", b"", &[0xff], b"")).is_err());
292    }
293
294    #[test]
295    fn save_returns_canonical_id_and_loads_exact_object() {
296        let roots = TempRoots::new("save-load");
297        let (ordering, _peering, objects) = open_stack(&roots);
298        let expected = Object {
299            filename: "photo.png".to_owned(),
300            file_type: "image/png".to_owned(),
301            description: "opaque bytes".to_owned(),
302            data: vec![0, 1, 2, 255],
303        };
304
305        let id = objects
306            .save(
307                &expected.filename,
308                &expected.file_type,
309                &expected.description,
310                &expected.data,
311            )
312            .unwrap();
313
314        assert_eq!(ordering.tip(), Some(id));
315        let bytes = ordering.get_txn(id).unwrap().unwrap();
316        assert_eq!(TxId::for_transaction(&bytes), id);
317        assert_eq!(
318            Transaction::parse(&bytes).unwrap().subsystem(),
319            object_subsystem()
320        );
321        assert_eq!(objects.load(id).unwrap(), Some(expected));
322    }
323
324    #[test]
325    fn load_distinguishes_unknown_other_and_malformed_objects() {
326        let roots = TempRoots::new("load-boundaries");
327        let (ordering, peering, objects) = open_stack(&roots);
328
329        assert_eq!(objects.load(TxId::from_bytes([0; 12])).unwrap(), None);
330
331        let other = other_subsystem();
332        ordering
333            .register_subsystem(other, None, Arc::new(NoopSubsystem))
334            .unwrap();
335        let other_id = peering.submit_txn(other, b"not an object").unwrap();
336        assert_eq!(objects.load(other_id).unwrap(), None);
337
338        let (malformed_id, _) = ordering
339            .submit_local_txn(
340                1,
341                [9; 32],
342                object_subsystem(),
343                &[PAYLOAD_VERSION],
344                |_| Ok([9; 64]),
345                |_| Ok(()),
346            )
347            .unwrap();
348        assert!(objects.load(malformed_id).is_err());
349    }
350
351    #[test]
352    fn duplicate_values_create_distinct_loadable_objects() {
353        let roots = TempRoots::new("duplicates");
354        let (_ordering, _peering, objects) = open_stack(&roots);
355
356        let first = objects.save("same", "same/type", "same", b"same").unwrap();
357        let second = objects.save("same", "same/type", "same", b"same").unwrap();
358
359        assert_ne!(first, second);
360        assert_eq!(objects.load(first).unwrap(), objects.load(second).unwrap());
361    }
362
363    #[test]
364    fn concurrent_saves_and_loads_preserve_every_object() {
365        let roots = TempRoots::new("concurrent");
366        let (_ordering, _peering, objects) = open_stack(&roots);
367        let objects = Arc::new(objects);
368        let saves: Vec<_> = (0_u8..12)
369            .map(|value| {
370                let objects = objects.clone();
371                thread::spawn(move || {
372                    let filename = format!("file-{value}");
373                    let data = vec![value; usize::from(value) + 1];
374                    let id = objects
375                        .save(&filename, "arbitrary", "concurrent", &data)
376                        .unwrap();
377                    (value, id)
378                })
379            })
380            .collect();
381
382        let saved: Vec<_> = saves.into_iter().map(|work| work.join().unwrap()).collect();
383        let unique: HashSet<_> = saved.iter().map(|(_, id)| *id).collect();
384        assert_eq!(unique.len(), saved.len());
385
386        let loads: Vec<_> = saved
387            .into_iter()
388            .map(|(value, id)| {
389                let objects = objects.clone();
390                thread::spawn(move || {
391                    let loaded = objects.load(id).unwrap().unwrap();
392                    assert_eq!(loaded.filename, format!("file-{value}"));
393                    assert_eq!(loaded.file_type, "arbitrary");
394                    assert_eq!(loaded.description, "concurrent");
395                    assert_eq!(loaded.data, vec![value; usize::from(value) + 1]);
396                })
397            })
398            .collect();
399
400        for load in loads {
401            load.join().unwrap();
402        }
403    }
404
405    #[test]
406    fn single_node_save_restart_canary_loads_history_without_state() {
407        let roots = TempRoots::new("restart");
408        let first_id;
409        let first = Object {
410            filename: "archive.pdf".to_owned(),
411            file_type: "application/pdf".to_owned(),
412            description: "before restart".to_owned(),
413            data: vec![7; 4096],
414        };
415
416        {
417            let (_ordering, _peering, objects) = open_stack(&roots);
418            first_id = objects
419                .save(
420                    &first.filename,
421                    &first.file_type,
422                    &first.description,
423                    &first.data,
424                )
425                .unwrap();
426            assert_eq!(objects.load(first_id).unwrap(), Some(first.clone()));
427        }
428
429        let (ordering, _peering, objects) = open_stack(&roots);
430        assert_eq!(objects.load(first_id).unwrap(), Some(first));
431        let later_id = objects
432            .save("later.bin", "not-validated", "", &[4, 3, 2, 1])
433            .unwrap();
434        assert_eq!(ordering.tip(), Some(later_id));
435        assert_eq!(
436            objects.load(later_id).unwrap().unwrap().data,
437            vec![4, 3, 2, 1]
438        );
439    }
440
441    #[test]
442    fn peering_and_ordering_roots_are_the_only_persistent_paths() {
443        let roots = TempRoots::new("paths");
444        let (_ordering, _peering, objects) = open_stack(&roots);
445        objects.save("x", "y", "z", b"data").unwrap();
446
447        let mut entries: Vec<_> = fs::read_dir(&roots.base)
448            .unwrap()
449            .map(|entry| entry.unwrap().file_name())
450            .collect();
451        entries.sort();
452
453        assert_eq!(entries, vec!["ordering", "peering"]);
454        assert!(roots.ordering().exists());
455        assert!(roots.peering().exists());
456    }
457}
kcode_k1_objects/lib.rs

kcode_k1_objects/
lib.rs
