|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| use std::io::{self, Read, Seek, SeekFrom, Write};
|
| use std::path::Path;
|
| use std::sync::Arc;
|
|
|
| use tokio::sync::mpsc;
|
|
|
|
|
| #[derive(Debug)]
|
| pub struct LedgerRecord {
|
| pub sequence: u64,
|
| pub prev_hash: [u8; 32],
|
| pub payload_hash: [u8; 32],
|
| pub payload: Vec<u8>,
|
| }
|
|
|
| impl LedgerRecord {
|
|
|
| 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
|
| }
|
| }
|
|
|
|
|
| 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)?;
|
|
|
|
|
| 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))
|
| }
|
|
|
|
|
| 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))
|
| }
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
| 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");
|
| }
|
|
|