kcode-kweb-db 0.1.0

A convergent signed-DAG store for Kweb nodes and objects
Documentation
use crate::{
    Error, ObjectId, ObjectPayload, Result, TransactionId, TransactionPackage,
    ledger::sync_directory,
    wire::{self, ParsedTransaction},
};
use std::{
    fs::{self, OpenOptions},
    io::Write,
    path::{Path, PathBuf},
};

const HEADER: &[u8] = b"KWOBJECT\0\x01\0\0";
const FIXED_LENGTH: usize = 12 + 6 + 32 + 8 + 32;

pub(crate) fn initialize(root: &Path) -> Result<()> {
    ensure_directory(&root.join("objects"))
}

pub(crate) fn verify_package(
    parsed: &ParsedTransaction,
    package: &TransactionPackage,
) -> Result<()> {
    if package.transaction.len() > crate::model::MAX_TRANSACTION_BYTES {
        return Err(Error::invalid_transaction(
            "package transaction is too large",
        ));
    }
    if TransactionId::for_signed_bytes(&package.transaction) != parsed.id {
        return Err(Error::invalid_transaction(
            "package transaction ID does not match parsed bytes",
        ));
    }
    if package.objects.len() != parsed.unsigned.objects.len() {
        return Err(Error::invalid_transaction(
            "package does not contain exactly the declared objects",
        ));
    }
    for (declaration, payload) in parsed.unsigned.objects.iter().zip(&package.objects) {
        if payload.id != declaration.id
            || payload.bytes.len() as u64 != declaration.length
            || wire::object_hash(&payload.bytes) != declaration.sha256
        {
            return Err(Error::invalid_transaction(
                "package object does not match its signed declaration",
            ));
        }
    }
    Ok(())
}

pub(crate) fn install_package(
    root: &Path,
    parsed: &ParsedTransaction,
    package: &TransactionPackage,
) -> Result<()> {
    for payload in &package.objects {
        install(root, parsed.id, payload)?;
    }
    Ok(())
}

pub(crate) fn load_package(
    root: &Path,
    transaction: Vec<u8>,
    parsed: &ParsedTransaction,
) -> Result<TransactionPackage> {
    let mut payloads = Vec::with_capacity(parsed.unsigned.objects.len());
    for declaration in &parsed.unsigned.objects {
        payloads.push(ObjectPayload {
            id: declaration.id,
            bytes: read_envelope(root, declaration.id, parsed.id)?,
        });
    }
    Ok(TransactionPackage {
        transaction,
        objects: payloads,
    })
}

pub(crate) fn read_visible(
    root: &Path,
    id: ObjectId,
    transaction: TransactionId,
) -> Result<Vec<u8>> {
    read_envelope(root, id, transaction)
}

fn install(root: &Path, transaction: TransactionId, payload: &ObjectPayload) -> Result<()> {
    let path = object_path(root, payload.id, transaction);
    let directory = path
        .parent()
        .ok_or_else(|| Error::corrupt("object path has no parent"))?;
    ensure_directory(directory)?;
    match fs::symlink_metadata(&path) {
        Ok(metadata) => {
            if !metadata.file_type().is_file() || metadata.file_type().is_symlink() {
                return Err(Error::corrupt("object envelope is not a regular file"));
            }
            if read_envelope(root, payload.id, transaction)? != payload.bytes {
                return Err(Error::corrupt(
                    "existing object envelope has different bytes",
                ));
            }
            return Ok(());
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(error.into()),
    }
    let temporary_path = directory.join(format!(".tmp-{}", hex::encode(rand::random::<[u8; 8]>())));
    let bytes = encode_envelope(payload.id, transaction, &payload.bytes);
    let mut temporary = OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&temporary_path)?;
    temporary.write_all(&bytes)?;
    temporary.sync_all()?;
    drop(temporary);
    match fs::hard_link(&temporary_path, &path) {
        Ok(()) => {}
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
            if read_envelope(root, payload.id, transaction)? != payload.bytes {
                let _ = fs::remove_file(&temporary_path);
                return Err(Error::corrupt(
                    "concurrent object envelope has different bytes",
                ));
            }
        }
        Err(error) => {
            let _ = fs::remove_file(&temporary_path);
            return Err(error.into());
        }
    }
    fs::remove_file(&temporary_path)?;
    sync_directory(directory)?;
    Ok(())
}

fn encode_envelope(id: ObjectId, transaction: TransactionId, payload: &[u8]) -> Vec<u8> {
    let mut bytes = Vec::with_capacity(FIXED_LENGTH + payload.len());
    bytes.extend_from_slice(HEADER);
    bytes.extend_from_slice(&id.0);
    bytes.extend_from_slice(&transaction.0);
    bytes.extend_from_slice(&(payload.len() as u64).to_be_bytes());
    bytes.extend_from_slice(&wire::object_hash(payload));
    bytes.extend_from_slice(payload);
    bytes
}

fn read_envelope(root: &Path, id: ObjectId, transaction: TransactionId) -> Result<Vec<u8>> {
    let path = object_path(root, id, transaction);
    let metadata = fs::symlink_metadata(&path).map_err(|error| {
        if error.kind() == std::io::ErrorKind::NotFound {
            Error::corrupt(format!("declared object envelope is missing: {id}"))
        } else {
            error.into()
        }
    })?;
    if !metadata.file_type().is_file() || metadata.file_type().is_symlink() {
        return Err(Error::corrupt("object envelope is not a regular file"));
    }
    let bytes = fs::read(&path)?;
    if bytes.len() < FIXED_LENGTH || &bytes[..HEADER.len()] != HEADER {
        return Err(Error::corrupt(
            "object envelope has unsupported magic or version",
        ));
    }
    let mut offset = HEADER.len();
    if bytes[offset..offset + 6] != id.0 {
        return Err(Error::corrupt("object envelope locator mismatch"));
    }
    offset += 6;
    if bytes[offset..offset + 32] != transaction.0 {
        return Err(Error::corrupt("object envelope transaction mismatch"));
    }
    offset += 32;
    let length = u64::from_be_bytes(bytes[offset..offset + 8].try_into().unwrap());
    offset += 8;
    let hash: [u8; 32] = bytes[offset..offset + 32].try_into().unwrap();
    offset += 32;
    let payload = &bytes[offset..];
    if length != payload.len() as u64 || wire::object_hash(payload) != hash {
        return Err(Error::corrupt("object envelope length or hash mismatch"));
    }
    Ok(payload.to_vec())
}

fn object_path(root: &Path, id: ObjectId, transaction: TransactionId) -> PathBuf {
    let text = id.to_string();
    root.join("objects")
        .join(&text[..2])
        .join(&text[2..])
        .join(format!("{transaction}.kwo"))
}

fn ensure_directory(path: &Path) -> Result<()> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => {
            if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
                return Err(Error::corrupt(format!(
                    "{} is not a real directory",
                    path.display()
                )));
            }
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            fs::create_dir_all(path)?;
        }
        Err(error) => return Err(error.into()),
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn envelope_round_trip_and_corruption() {
        let root = tempfile::tempdir().unwrap();
        initialize(root.path()).unwrap();
        let id = ObjectId([1; 6]);
        let transaction = TransactionId([2; 32]);
        let payload = ObjectPayload {
            id,
            bytes: b"hello".to_vec(),
        };
        install(root.path(), transaction, &payload).unwrap();
        assert_eq!(
            read_envelope(root.path(), id, transaction).unwrap(),
            b"hello"
        );
        let path = object_path(root.path(), id, transaction);
        let mut bytes = fs::read(&path).unwrap();
        let last = bytes.len() - 1;
        bytes[last] ^= 1;
        fs::write(path, bytes).unwrap();
        assert!(read_envelope(root.path(), id, transaction).is_err());
    }
}