use kcode_k1_peering::K1Peering;
use kcode_k1_transaction::{REGISTER_AT_TIP, SubsystemId, Transaction};
pub use kcode_k1_txn_ordering::TxId;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem};
use std::sync::Arc;
const PAYLOAD_VERSION: u8 = 1;
const PAYLOAD_HEADER_BYTES: usize = 25;
const OBJECT_SUBSYSTEM_BYTES: [u8; 20] = *b"k1-objects-subsystem";
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Object {
pub filename: String,
pub file_type: String,
pub description: String,
pub data: Vec<u8>,
}
pub struct K1Objects {
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
}
impl K1Objects {
pub fn open(ordering: Arc<K1TxnOrdering>, peering: Arc<K1Peering>) -> Result<Self, String> {
ordering.register_subsystem(
object_subsystem(),
Some(REGISTER_AT_TIP),
Arc::new(StatelessObjects),
)?;
Ok(Self { ordering, peering })
}
pub fn save(
&self,
filename: &str,
file_type: &str,
description: &str,
data: &[u8],
) -> Result<TxId, String> {
let payload = encode_payload(filename, file_type, description, data)?;
self.peering.submit_txn(object_subsystem(), &payload)
}
pub fn load(&self, id: TxId) -> Result<Option<Object>, String> {
let Some(bytes) = self.ordering.get_txn(id)? else {
return Ok(None);
};
let transaction = Transaction::parse(&bytes)
.map_err(|message| format!("canonical transaction is invalid: {message}"))?;
if transaction.subsystem() != object_subsystem() {
return Ok(None);
}
decode_payload(transaction.payload()).map(Some)
}
}
struct StatelessObjects;
impl Subsystem for StatelessObjects {
fn submit_txn(&self, _id: TxId, _payload: &[u8]) -> Result<(), String> {
Ok(())
}
fn reorg(&self) -> Result<(), String> {
Ok(())
}
}
fn object_subsystem() -> SubsystemId {
SubsystemId::from_bytes(OBJECT_SUBSYSTEM_BYTES)
.expect("Objects subsystem ID is valid fixed-width UTF-8")
}
fn encode_payload(
filename: &str,
file_type: &str,
description: &str,
data: &[u8],
) -> Result<Vec<u8>, String> {
let filename_len = u64::try_from(filename.len())
.map_err(|_| "filename length exceeds the object wire format".to_owned())?;
let file_type_len = u64::try_from(file_type.len())
.map_err(|_| "file type length exceeds the object wire format".to_owned())?;
let description_len = u64::try_from(description.len())
.map_err(|_| "description length exceeds the object wire format".to_owned())?;
let total_len = PAYLOAD_HEADER_BYTES
.checked_add(filename.len())
.and_then(|length| length.checked_add(file_type.len()))
.and_then(|length| length.checked_add(description.len()))
.and_then(|length| length.checked_add(data.len()))
.ok_or_else(|| "object payload length exceeds usize".to_owned())?;
let mut payload = Vec::new();
payload
.try_reserve_exact(total_len)
.map_err(|error| format!("cannot allocate object payload: {error}"))?;
payload.push(PAYLOAD_VERSION);
payload.extend_from_slice(&filename_len.to_le_bytes());
payload.extend_from_slice(&file_type_len.to_le_bytes());
payload.extend_from_slice(&description_len.to_le_bytes());
payload.extend_from_slice(filename.as_bytes());
payload.extend_from_slice(file_type.as_bytes());
payload.extend_from_slice(description.as_bytes());
payload.extend_from_slice(data);
debug_assert_eq!(payload.len(), total_len);
Ok(payload)
}
fn decode_payload(payload: &[u8]) -> Result<Object, String> {
if payload.len() < PAYLOAD_HEADER_BYTES {
return Err(format!(
"object payload must contain at least {PAYLOAD_HEADER_BYTES} bytes"
));
}
if payload[0] != PAYLOAD_VERSION {
return Err(format!("unsupported object payload version {}", payload[0]));
}
let filename_len = decode_length(payload, 1, "filename")?;
let file_type_len = decode_length(payload, 9, "file type")?;
let description_len = decode_length(payload, 17, "description")?;
let filename_end = PAYLOAD_HEADER_BYTES
.checked_add(filename_len)
.ok_or_else(|| "object metadata length overflow".to_owned())?;
let file_type_end = filename_end
.checked_add(file_type_len)
.ok_or_else(|| "object metadata length overflow".to_owned())?;
let description_end = file_type_end
.checked_add(description_len)
.ok_or_else(|| "object metadata length overflow".to_owned())?;
if description_end > payload.len() {
return Err("object metadata extends past the payload".to_owned());
}
let filename = std::str::from_utf8(&payload[PAYLOAD_HEADER_BYTES..filename_end])
.map_err(|error| format!("object filename is not valid UTF-8: {error}"))?
.to_owned();
let file_type = std::str::from_utf8(&payload[filename_end..file_type_end])
.map_err(|error| format!("object file type is not valid UTF-8: {error}"))?
.to_owned();
let description = std::str::from_utf8(&payload[file_type_end..description_end])
.map_err(|error| format!("object description is not valid UTF-8: {error}"))?
.to_owned();
Ok(Object {
filename,
file_type,
description,
data: payload[description_end..].to_vec(),
})
}
fn decode_length(payload: &[u8], offset: usize, field: &str) -> Result<usize, String> {
let bytes: [u8; 8] = payload[offset..offset + 8]
.try_into()
.expect("object length fields are inside the fixed header");
usize::try_from(u64::from_le_bytes(bytes))
.map_err(|_| format!("object {field} length exceeds usize"))
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashSet;
use std::fs;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;
static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
struct TempRoots {
base: PathBuf,
}
impl TempRoots {
fn new(label: &str) -> Self {
let sequence = NEXT_ROOT.fetch_add(1, Ordering::Relaxed);
let base = std::env::temp_dir().join(format!(
"kcode-k1-objects-{}-{sequence}-{label}",
std::process::id()
));
let _ = fs::remove_dir_all(&base);
Self { base }
}
fn ordering(&self) -> PathBuf {
self.base.join("ordering")
}
fn peering(&self) -> PathBuf {
self.base.join("peering")
}
}
impl Drop for TempRoots {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.base);
}
}
struct NoopSubsystem;
impl Subsystem for NoopSubsystem {
fn submit_txn(&self, _id: TxId, _payload: &[u8]) -> Result<(), String> {
Ok(())
}
fn reorg(&self) -> Result<(), String> {
Ok(())
}
}
fn open_stack(roots: &TempRoots) -> (Arc<K1TxnOrdering>, Arc<K1Peering>, K1Objects) {
let ordering = Arc::new(K1TxnOrdering::open(&roots.ordering()).unwrap());
let peering = Arc::new(K1Peering::open(&roots.peering(), ordering.clone()).unwrap());
let objects = K1Objects::open(ordering.clone(), peering.clone()).unwrap();
(ordering, peering, objects)
}
fn other_subsystem() -> SubsystemId {
SubsystemId::from_bytes([b'o'; 20]).unwrap()
}
fn raw_payload(filename: &[u8], file_type: &[u8], description: &[u8], data: &[u8]) -> Vec<u8> {
let mut payload = Vec::new();
payload.push(PAYLOAD_VERSION);
payload.extend_from_slice(&(filename.len() as u64).to_le_bytes());
payload.extend_from_slice(&(file_type.len() as u64).to_le_bytes());
payload.extend_from_slice(&(description.len() as u64).to_le_bytes());
payload.extend_from_slice(filename);
payload.extend_from_slice(file_type);
payload.extend_from_slice(description);
payload.extend_from_slice(data);
payload
}
#[test]
fn codec_round_trips_empty_and_opaque_values() {
let empty = decode_payload(&encode_payload("", "", "", &[]).unwrap()).unwrap();
assert_eq!(
empty,
Object {
filename: String::new(),
file_type: String::new(),
description: String::new(),
data: Vec::new(),
}
);
let expected = Object {
filename: "résumé-太阳.pdf".to_owned(),
file_type: "anything at all".to_owned(),
description: "descripción".to_owned(),
data: vec![0, 255, 1, 128, 0],
};
let payload = encode_payload(
&expected.filename,
&expected.file_type,
&expected.description,
&expected.data,
)
.unwrap();
assert_eq!(payload[0], PAYLOAD_VERSION);
assert_eq!(decode_payload(&payload).unwrap(), expected);
}
#[test]
fn decoder_rejects_invalid_structure_and_metadata() {
assert!(decode_payload(&[]).is_err());
assert!(decode_payload(&[PAYLOAD_VERSION; PAYLOAD_HEADER_BYTES - 1]).is_err());
let mut unsupported = raw_payload(b"", b"", b"", b"");
unsupported[0] = 2;
assert!(decode_payload(&unsupported).is_err());
let mut overflow = raw_payload(b"", b"", b"", b"");
overflow[1..9].copy_from_slice(&u64::MAX.to_le_bytes());
assert!(decode_payload(&overflow).is_err());
let mut out_of_bounds = raw_payload(b"", b"", b"", b"");
out_of_bounds[1..9].copy_from_slice(&1_u64.to_le_bytes());
assert!(decode_payload(&out_of_bounds).is_err());
assert!(decode_payload(&raw_payload(&[0xff], b"", b"", b"")).is_err());
assert!(decode_payload(&raw_payload(b"", &[0xff], b"", b"")).is_err());
assert!(decode_payload(&raw_payload(b"", b"", &[0xff], b"")).is_err());
}
#[test]
fn save_returns_canonical_id_and_loads_exact_object() {
let roots = TempRoots::new("save-load");
let (ordering, _peering, objects) = open_stack(&roots);
let expected = Object {
filename: "photo.png".to_owned(),
file_type: "image/png".to_owned(),
description: "opaque bytes".to_owned(),
data: vec![0, 1, 2, 255],
};
let id = objects
.save(
&expected.filename,
&expected.file_type,
&expected.description,
&expected.data,
)
.unwrap();
assert_eq!(ordering.tip(), Some(id));
let bytes = ordering.get_txn(id).unwrap().unwrap();
assert_eq!(TxId::for_transaction(&bytes), id);
assert_eq!(
Transaction::parse(&bytes).unwrap().subsystem(),
object_subsystem()
);
assert_eq!(objects.load(id).unwrap(), Some(expected));
}
#[test]
fn load_distinguishes_unknown_other_and_malformed_objects() {
let roots = TempRoots::new("load-boundaries");
let (ordering, peering, objects) = open_stack(&roots);
assert_eq!(objects.load(TxId::from_bytes([0; 12])).unwrap(), None);
let other = other_subsystem();
ordering
.register_subsystem(other, None, Arc::new(NoopSubsystem))
.unwrap();
let other_id = peering.submit_txn(other, b"not an object").unwrap();
assert_eq!(objects.load(other_id).unwrap(), None);
let (malformed_id, _) = ordering
.submit_local_txn(
1,
[9; 32],
object_subsystem(),
&[PAYLOAD_VERSION],
|_| Ok([9; 64]),
|_| Ok(()),
)
.unwrap();
assert!(objects.load(malformed_id).is_err());
}
#[test]
fn duplicate_values_create_distinct_loadable_objects() {
let roots = TempRoots::new("duplicates");
let (_ordering, _peering, objects) = open_stack(&roots);
let first = objects.save("same", "same/type", "same", b"same").unwrap();
let second = objects.save("same", "same/type", "same", b"same").unwrap();
assert_ne!(first, second);
assert_eq!(objects.load(first).unwrap(), objects.load(second).unwrap());
}
#[test]
fn concurrent_saves_and_loads_preserve_every_object() {
let roots = TempRoots::new("concurrent");
let (_ordering, _peering, objects) = open_stack(&roots);
let objects = Arc::new(objects);
let saves: Vec<_> = (0_u8..12)
.map(|value| {
let objects = objects.clone();
thread::spawn(move || {
let filename = format!("file-{value}");
let data = vec![value; usize::from(value) + 1];
let id = objects
.save(&filename, "arbitrary", "concurrent", &data)
.unwrap();
(value, id)
})
})
.collect();
let saved: Vec<_> = saves.into_iter().map(|work| work.join().unwrap()).collect();
let unique: HashSet<_> = saved.iter().map(|(_, id)| *id).collect();
assert_eq!(unique.len(), saved.len());
let loads: Vec<_> = saved
.into_iter()
.map(|(value, id)| {
let objects = objects.clone();
thread::spawn(move || {
let loaded = objects.load(id).unwrap().unwrap();
assert_eq!(loaded.filename, format!("file-{value}"));
assert_eq!(loaded.file_type, "arbitrary");
assert_eq!(loaded.description, "concurrent");
assert_eq!(loaded.data, vec![value; usize::from(value) + 1]);
})
})
.collect();
for load in loads {
load.join().unwrap();
}
}
#[test]
fn single_node_save_restart_canary_loads_history_without_state() {
let roots = TempRoots::new("restart");
let first_id;
let first = Object {
filename: "archive.pdf".to_owned(),
file_type: "application/pdf".to_owned(),
description: "before restart".to_owned(),
data: vec![7; 4096],
};
{
let (_ordering, _peering, objects) = open_stack(&roots);
first_id = objects
.save(
&first.filename,
&first.file_type,
&first.description,
&first.data,
)
.unwrap();
assert_eq!(objects.load(first_id).unwrap(), Some(first.clone()));
}
let (ordering, _peering, objects) = open_stack(&roots);
assert_eq!(objects.load(first_id).unwrap(), Some(first));
let later_id = objects
.save("later.bin", "not-validated", "", &[4, 3, 2, 1])
.unwrap();
assert_eq!(ordering.tip(), Some(later_id));
assert_eq!(
objects.load(later_id).unwrap().unwrap().data,
vec![4, 3, 2, 1]
);
}
#[test]
fn peering_and_ordering_roots_are_the_only_persistent_paths() {
let roots = TempRoots::new("paths");
let (_ordering, _peering, objects) = open_stack(&roots);
objects.save("x", "y", "z", b"data").unwrap();
let mut entries: Vec<_> = fs::read_dir(&roots.base)
.unwrap()
.map(|entry| entry.unwrap().file_name())
.collect();
entries.sort();
assert_eq!(entries, vec!["ordering", "peering"]);
assert!(roots.ordering().exists());
assert!(roots.peering().exists());
}
}