File size: 4,734 Bytes
30f011f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | //! ledger.rs — WORM append-only audit ledger
//!
//! Every entailment attestation is sealed into an append-only chain.
//! Each record contains:
//! - BLAKE3 hash of the attestation payload
//! - BLAKE3 hash of the previous record (chain link)
//! - The raw bincode payload
//!
//! The chain is stored as a length-prefixed binary flat file.
//! A corrupt or tampered record breaks the chain hash and is immediately detectable.
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::Path;
use std::sync::Arc;
use tokio::sync::mpsc;
/// One sealed record in the WORM chain.
#[derive(Debug)]
pub struct LedgerRecord {
pub sequence: u64,
pub prev_hash: [u8; 32],
pub payload_hash: [u8; 32],
pub payload: Vec<u8>,
}
impl LedgerRecord {
/// Serialise to bytes: [seq 8B][prev 32B][hash 32B][len 8B][payload].
fn to_bytes(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(80 + self.payload.len());
out.extend_from_slice(&self.sequence.to_le_bytes());
out.extend_from_slice(&self.prev_hash);
out.extend_from_slice(&self.payload_hash);
out.extend_from_slice(&(self.payload.len() as u64).to_le_bytes());
out.extend_from_slice(&self.payload);
out
}
}
/// Append-only WORM ledger backed by a flat binary file.
pub struct WormLedger {
file: std::fs::File,
sequence: u64,
last_hash: [u8; 32],
}
impl WormLedger {
pub fn open(path: impl AsRef<Path>) -> io::Result<Self> {
let path = path.as_ref();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.read(true)
.open(path)?;
// Replay existing records to find last hash and sequence number.
let (sequence, last_hash) = Self::replay(&mut file)?;
log::info!(
"[ledger] opened {} — {} existing records",
path.display(),
sequence
);
Ok(Self { file, sequence, last_hash })
}
fn replay(file: &mut std::fs::File) -> io::Result<(u64, [u8; 32])> {
file.seek(SeekFrom::Start(0))?;
let mut seq: u64 = 0;
let mut last: [u8; 32] = [0u8; 32];
loop {
let mut seq_buf = [0u8; 8];
match file.read_exact(&mut seq_buf) {
Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
Err(e) => return Err(e),
Ok(_) => {}
}
let mut prev = [0u8; 32];
let mut hash = [0u8; 32];
file.read_exact(&mut prev)?;
file.read_exact(&mut hash)?;
let mut len_buf = [0u8; 8];
file.read_exact(&mut len_buf)?;
let len = u64::from_le_bytes(len_buf) as usize;
let mut payload = vec![0u8; len];
file.read_exact(&mut payload)?;
seq = u64::from_le_bytes(seq_buf) + 1;
last = hash;
}
Ok((seq, last))
}
/// Append a sealed record. Returns the receipt (sequence + hash).
pub fn append(&mut self, payload_hash: [u8; 32], payload: Vec<u8>) -> io::Result<(u64, [u8; 32])> {
let record = LedgerRecord {
sequence: self.sequence,
prev_hash: self.last_hash,
payload_hash,
payload,
};
let bytes = record.to_bytes();
self.file.write_all(&bytes)?;
self.file.flush()?;
self.last_hash = payload_hash;
let seq = self.sequence;
self.sequence += 1;
Ok((seq, payload_hash))
}
}
// ── Background WORM worker ───────────────────────────────────────────────────
/// Runs as a dedicated Tokio task.
/// Receives (hash, payload) from the inference loop and appends to the ledger.
/// Does not block the GPU batching thread.
pub async fn run_ledger_worker(
mut rx: mpsc::Receiver<([u8; 32], Vec<u8>)>,
ledger_path: String,
) {
let mut ledger = WormLedger::open(&ledger_path)
.unwrap_or_else(|e| panic!("cannot open ledger {}: {}", ledger_path, e));
while let Some((hash, payload)) = rx.recv().await {
match ledger.append(hash, payload) {
Ok((seq, _)) => log::debug!("[ledger] sealed record seq={}", seq),
Err(e) => log::error!("[ledger] write fault: {}", e),
}
}
log::info!("[ledger] worker shutting down");
}
|