mod frame;
use std::{
collections::HashMap,
fs::{File, OpenOptions},
io::{self, Seek, SeekFrom, Write},
marker::PhantomData,
path::Path,
};
use thiserror::Error;
use uuid::Uuid;
use self::frame::{
Frame, FrameKind, HEADER_LENGTH, MAX_PAYLOAD_LENGTH, ReadStatus, payload_length,
read_exact_or_tail,
};
const DESCRIPTOR_LENGTH: usize = 36;
const TRANSACTION_DOMAIN: &[u8] = b"hyphae-transaction-v1";
pub(crate) const MAX_OPERATION_BYTES: usize = MAX_PAYLOAD_LENGTH;
#[derive(Debug, Error)]
pub enum LogError {
#[error(transparent)]
Io(#[from] io::Error),
#[error("invalid frame magic at byte offset {offset}")]
BadMagic {
offset: u64,
},
#[error(
"unsupported log version {found} at byte offset {offset}; supported version is {supported}"
)]
UnsupportedVersion {
offset: u64,
found: u16,
supported: u16,
},
#[error("unknown frame kind {kind} at byte offset {offset}")]
UnknownFrameKind {
offset: u64,
kind: u8,
},
#[error("unsupported frame flags {flags:#04x} at byte offset {offset}")]
UnsupportedFlags {
offset: u64,
flags: u8,
},
#[error("frame payload is {length} bytes; maximum is {maximum}")]
PayloadTooLarge {
length: usize,
maximum: usize,
},
#[error("invalid sequence at byte offset {offset}: expected {expected}, found {found}")]
InvalidSequence {
offset: u64,
expected: u64,
found: u64,
},
#[error("previous-frame digest mismatch at sequence {sequence}")]
PreviousDigestMismatch {
sequence: u64,
},
#[error("CRC32C mismatch at sequence {sequence}")]
ChecksumMismatch {
sequence: u64,
},
#[error("BLAKE3 digest mismatch at sequence {sequence}")]
DigestMismatch {
sequence: u64,
},
#[error("malformed transaction descriptor at sequence {sequence}")]
MalformedTransaction {
sequence: u64,
},
#[error("{kind} frame at sequence {sequence} has no matching transaction begin")]
TransactionBoundary {
kind: &'static str,
sequence: u64,
},
#[error("transaction content mismatch at commit sequence {sequence}")]
TransactionContentMismatch {
sequence: u64,
},
#[error(
"transaction identifier {transaction_id} was already committed with different contents"
)]
IdempotencyConflict {
transaction_id: Uuid,
},
#[error("a transaction must contain at least one operation")]
EmptyTransaction,
#[error("transaction has too many operations")]
TooManyOperations,
#[error("log sequence space is exhausted")]
SequenceExhausted,
#[error("invalid log segment anchor")]
InvalidAnchor,
#[error("durable log writer is poisoned; reopen it before writing again")]
Poisoned,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CommitReceipt {
pub transaction_id: Uuid,
pub commit_sequence: u64,
pub commit_digest: [u8; 32],
pub transaction_digest: [u8; 32],
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum AppendOutcome {
Committed(CommitReceipt),
Existing(CommitReceipt),
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RecoveredTransaction {
pub receipt: CommitReceipt,
pub operations: Vec<Vec<u8>>,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct RecoveryReport {
pub base_sequence: u64,
pub base_digest: [u8; 32],
pub transactions: Vec<RecoveredTransaction>,
pub ignored_uncommitted_transactions: u64,
pub duplicate_commits: u64,
pub truncated_tail_bytes: u64,
pub valid_bytes: u64,
pub last_sequence: u64,
pub last_digest: [u8; 32],
}
#[derive(Debug)]
pub struct OpenedLog<'directory> {
pub log: DurableLog,
pub recovery: RecoveryReport,
directory_lock: PhantomData<&'directory crate::DataDirectory>,
}
impl OpenedLog<'_> {
pub(crate) fn new(log: DurableLog, recovery: RecoveryReport) -> Self {
Self {
log,
recovery,
directory_lock: PhantomData,
}
}
}
#[derive(Debug)]
pub struct DurableLog {
file: File,
next_sequence: u64,
previous_digest: [u8; 32],
committed: HashMap<Uuid, CommitReceipt>,
poisoned: bool,
#[cfg(test)]
fail_next_sync: bool,
}
impl DurableLog {
#[cfg(test)]
pub(crate) fn open_file(
path: impl AsRef<Path>,
) -> Result<(DurableLog, RecoveryReport), LogError> {
Self::open_file_at(path, 0, [0; 32])
}
pub(crate) fn open_file_at(
path: impl AsRef<Path>,
base_sequence: u64,
base_digest: [u8; 32],
) -> Result<(DurableLog, RecoveryReport), LogError> {
if (base_sequence == 0) != (base_digest == [0; 32]) {
return Err(LogError::InvalidAnchor);
}
let path = path.as_ref();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let existed = path.exists();
let mut file = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(path)?;
if !existed {
file.sync_all()?;
#[cfg(unix)]
if let Some(parent) = path.parent() {
File::open(parent)?.sync_all()?;
}
}
let recovery = scan(&mut file, base_sequence, base_digest)?;
let physical_length = file.metadata()?.len();
if physical_length != recovery.valid_bytes {
file.set_len(recovery.valid_bytes)?;
file.sync_data()?;
}
file.seek(SeekFrom::End(0))?;
let committed = recovery
.transactions
.iter()
.map(|transaction| (transaction.receipt.transaction_id, transaction.receipt))
.collect();
let next_sequence = recovery
.last_sequence
.checked_add(1)
.ok_or(LogError::SequenceExhausted)?;
let log = Self {
file,
next_sequence,
previous_digest: recovery.last_digest,
committed,
poisoned: false,
#[cfg(test)]
fail_next_sync: false,
};
Ok((log, recovery))
}
pub fn append_transaction(
&mut self,
transaction_id: Uuid,
operations: &[Vec<u8>],
) -> Result<AppendOutcome, LogError> {
if self.poisoned {
return Err(LogError::Poisoned);
}
if operations.is_empty() {
return Err(LogError::EmptyTransaction);
}
let operation_count =
u32::try_from(operations.len()).map_err(|_| LogError::TooManyOperations)?;
for operation in operations {
if operation.len() > MAX_PAYLOAD_LENGTH {
return Err(LogError::PayloadTooLarge {
length: operation.len(),
maximum: MAX_PAYLOAD_LENGTH,
});
}
}
let transaction_digest = transaction_digest(operations, operation_count)?;
if let Some(receipt) = self.committed.get(&transaction_id).copied() {
return if receipt.transaction_digest == transaction_digest {
Ok(AppendOutcome::Existing(receipt))
} else {
Err(LogError::IdempotencyConflict { transaction_id })
};
}
let descriptor = encode_descriptor(operation_count, transaction_digest);
let append_result = self.append_new_transaction(
transaction_id,
operations,
&descriptor,
transaction_digest,
);
if append_result.is_err() {
self.poisoned = true;
}
append_result
}
pub(crate) fn is_poisoned(&self) -> bool {
self.poisoned
}
#[cfg(test)]
pub(crate) fn inject_sync_failure(&mut self) {
self.fail_next_sync = true;
}
fn append_new_transaction(
&mut self,
transaction_id: Uuid,
operations: &[Vec<u8>],
descriptor: &[u8; DESCRIPTOR_LENGTH],
transaction_digest: [u8; 32],
) -> Result<AppendOutcome, LogError> {
self.append_frame(FrameKind::Begin, transaction_id, descriptor)?;
for operation in operations {
self.append_frame(FrameKind::Operation, transaction_id, operation)?;
}
let receipt = self.append_frame(FrameKind::Commit, transaction_id, descriptor)?;
#[cfg(test)]
if std::mem::take(&mut self.fail_next_sync) {
return Err(io::Error::other("injected log sync failure").into());
}
self.file.sync_data()?;
let receipt = CommitReceipt {
transaction_id,
commit_sequence: receipt.sequence,
commit_digest: receipt.digest,
transaction_digest,
};
self.committed.insert(transaction_id, receipt);
Ok(AppendOutcome::Committed(receipt))
}
fn append_frame(
&mut self,
kind: FrameKind,
transaction_id: Uuid,
payload: &[u8],
) -> Result<WrittenFrame, LogError> {
let frame = Frame {
kind,
sequence: self.next_sequence,
transaction_id,
previous_digest: self.previous_digest,
digest: [0; 32],
payload: payload.to_vec(),
};
let encoded = frame.encode()?;
let digest = copy_array(&encoded[80..112]);
self.file.write_all(&encoded)?;
let written = WrittenFrame {
sequence: self.next_sequence,
digest,
};
self.next_sequence = self
.next_sequence
.checked_add(1)
.ok_or(LogError::SequenceExhausted)?;
self.previous_digest = digest;
Ok(written)
}
}
#[derive(Clone, Copy, Debug)]
struct WrittenFrame {
sequence: u64,
digest: [u8; 32],
}
#[derive(Debug)]
struct PendingTransaction {
transaction_id: Uuid,
operation_count: u32,
expected_digest: [u8; 32],
operations: Vec<Vec<u8>>,
}
fn scan(
file: &mut File,
base_sequence: u64,
base_digest: [u8; 32],
) -> Result<RecoveryReport, LogError> {
file.seek(SeekFrom::Start(0))?;
let physical_length = file.metadata()?.len();
let mut report = RecoveryReport {
base_sequence,
base_digest,
last_sequence: base_sequence,
last_digest: base_digest,
..RecoveryReport::default()
};
let mut offset = 0_u64;
let mut expected_sequence = base_sequence
.checked_add(1)
.ok_or(LogError::SequenceExhausted)?;
let mut expected_previous_digest = base_digest;
let mut pending: Option<PendingTransaction> = None;
let mut committed: HashMap<Uuid, CommitReceipt> = HashMap::new();
loop {
let mut header = [0_u8; HEADER_LENGTH];
match read_exact_or_tail(file, &mut header)? {
ReadStatus::End => break,
ReadStatus::Partial => {
report.truncated_tail_bytes = physical_length.saturating_sub(offset);
break;
}
ReadStatus::Complete => {}
}
let length = payload_length(&header, offset)?;
let mut payload = vec![0_u8; length];
if read_exact_or_tail(file, &mut payload)? != ReadStatus::Complete {
report.truncated_tail_bytes = physical_length.saturating_sub(offset);
break;
}
let frame = Frame::decode(&header, payload, offset)?;
if frame.sequence != expected_sequence {
return Err(LogError::InvalidSequence {
offset,
expected: expected_sequence,
found: frame.sequence,
});
}
if frame.previous_digest != expected_previous_digest {
return Err(LogError::PreviousDigestMismatch {
sequence: frame.sequence,
});
}
apply_frame(&frame, &mut pending, &mut committed, &mut report)?;
let frame_length = u64::try_from(HEADER_LENGTH + frame.payload.len()).map_err(|_| {
LogError::PayloadTooLarge {
length: frame.payload.len(),
maximum: MAX_PAYLOAD_LENGTH,
}
})?;
offset = offset
.checked_add(frame_length)
.ok_or(LogError::SequenceExhausted)?;
report.valid_bytes = offset;
report.last_sequence = frame.sequence;
report.last_digest = frame.digest;
expected_sequence = expected_sequence
.checked_add(1)
.ok_or(LogError::SequenceExhausted)?;
expected_previous_digest = frame.digest;
}
if pending.is_some() {
report.ignored_uncommitted_transactions =
report.ignored_uncommitted_transactions.saturating_add(1);
}
Ok(report)
}
fn apply_frame(
frame: &Frame,
pending: &mut Option<PendingTransaction>,
committed: &mut HashMap<Uuid, CommitReceipt>,
report: &mut RecoveryReport,
) -> Result<(), LogError> {
match frame.kind {
FrameKind::Begin => {
if pending.is_some() {
report.ignored_uncommitted_transactions =
report.ignored_uncommitted_transactions.saturating_add(1);
}
let (operation_count, expected_digest) =
decode_descriptor(&frame.payload, frame.sequence)?;
*pending = Some(PendingTransaction {
transaction_id: frame.transaction_id,
operation_count,
expected_digest,
operations: Vec::new(),
});
Ok(())
}
FrameKind::Operation => {
let Some(current) = pending
.as_mut()
.filter(|current| current.transaction_id == frame.transaction_id)
else {
return Err(LogError::TransactionBoundary {
kind: "operation",
sequence: frame.sequence,
});
};
current.operations.push(frame.payload.clone());
Ok(())
}
FrameKind::Commit => {
let Some(current) = pending
.take()
.filter(|current| current.transaction_id == frame.transaction_id)
else {
return Err(LogError::TransactionBoundary {
kind: "commit",
sequence: frame.sequence,
});
};
let (operation_count, commit_digest) =
decode_descriptor(&frame.payload, frame.sequence)?;
let actual_count =
u32::try_from(current.operations.len()).map_err(|_| LogError::TooManyOperations)?;
let actual_digest = transaction_digest(¤t.operations, actual_count)?;
if operation_count != current.operation_count
|| commit_digest != current.expected_digest
|| actual_count != operation_count
|| actual_digest != commit_digest
{
return Err(LogError::TransactionContentMismatch {
sequence: frame.sequence,
});
}
let receipt = CommitReceipt {
transaction_id: frame.transaction_id,
commit_sequence: frame.sequence,
commit_digest: frame.digest,
transaction_digest: actual_digest,
};
if let Some(existing) = committed.get(&frame.transaction_id) {
if existing.transaction_digest != actual_digest {
return Err(LogError::IdempotencyConflict {
transaction_id: frame.transaction_id,
});
}
report.duplicate_commits = report.duplicate_commits.saturating_add(1);
} else {
committed.insert(frame.transaction_id, receipt);
report.transactions.push(RecoveredTransaction {
receipt,
operations: current.operations,
});
}
Ok(())
}
}
}
fn encode_descriptor(operation_count: u32, digest: [u8; 32]) -> [u8; DESCRIPTOR_LENGTH] {
let mut descriptor = [0_u8; DESCRIPTOR_LENGTH];
descriptor[..4].copy_from_slice(&operation_count.to_le_bytes());
descriptor[4..].copy_from_slice(&digest);
descriptor
}
fn decode_descriptor(payload: &[u8], sequence: u64) -> Result<(u32, [u8; 32]), LogError> {
if payload.len() != DESCRIPTOR_LENGTH {
return Err(LogError::MalformedTransaction { sequence });
}
let operation_count = u32::from_le_bytes(copy_array(&payload[..4]));
if operation_count == 0 {
return Err(LogError::MalformedTransaction { sequence });
}
let digest = copy_array(&payload[4..]);
Ok((operation_count, digest))
}
pub(crate) fn transaction_digest(
operations: &[Vec<u8>],
operation_count: u32,
) -> Result<[u8; 32], LogError> {
let mut hasher = blake3::Hasher::new();
hasher.update(TRANSACTION_DOMAIN);
hasher.update(&u64::from(operation_count).to_le_bytes());
for operation in operations {
let length = u64::try_from(operation.len()).map_err(|_| LogError::PayloadTooLarge {
length: operation.len(),
maximum: MAX_PAYLOAD_LENGTH,
})?;
hasher.update(&length.to_le_bytes());
hasher.update(operation);
}
Ok(*hasher.finalize().as_bytes())
}
fn copy_array<const N: usize>(source: &[u8]) -> [u8; N] {
let mut output = [0_u8; N];
output.copy_from_slice(source);
output
}
#[cfg(test)]
mod tests {
use std::{
error::Error,
fs::OpenOptions,
io::{Read, Seek, SeekFrom, Write},
};
use uuid::Uuid;
use super::{AppendOutcome, DurableLog, LogError, OpenedLog, frame::HEADER_LENGTH};
use crate::test_support::TestDirectory;
fn open_for_test(path: &std::path::Path) -> Result<OpenedLog<'static>, LogError> {
let (log, recovery) = DurableLog::open_file(path)?;
Ok(OpenedLog::new(log, recovery))
}
#[test]
fn committed_transaction_recovers_in_order() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-recovery")?;
let path = temporary.path().join("segment.hylog");
let transaction_id = Uuid::now_v7();
let mut opened = open_for_test(&path)?;
let outcome = opened
.log
.append_transaction(transaction_id, &[b"put:a=1".to_vec(), b"put:b=2".to_vec()])?;
assert!(matches!(outcome, AppendOutcome::Committed(_)));
drop(opened);
let reopened = open_for_test(&path)?;
assert_eq!(reopened.recovery.transactions.len(), 1);
assert_eq!(
reopened.recovery.transactions[0].operations,
[b"put:a=1".to_vec(), b"put:b=2".to_vec()]
);
Ok(())
}
#[test]
fn idempotency_survives_reopen() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-idempotency")?;
let path = temporary.path().join("segment.hylog");
let transaction_id = Uuid::now_v7();
let operations = [b"same".to_vec()];
let mut opened = open_for_test(&path)?;
let first = opened.log.append_transaction(transaction_id, &operations)?;
drop(opened);
let mut reopened = open_for_test(&path)?;
let second = reopened
.log
.append_transaction(transaction_id, &operations)?;
assert!(matches!(first, AppendOutcome::Committed(_)));
assert!(matches!(second, AppendOutcome::Existing(_)));
let conflict = reopened
.log
.append_transaction(transaction_id, &[b"different".to_vec()]);
assert!(matches!(
conflict,
Err(LogError::IdempotencyConflict { .. })
));
Ok(())
}
#[test]
fn truncates_only_an_incomplete_tail() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-tail")?;
let path = temporary.path().join("segment.hylog");
let mut opened = open_for_test(&path)?;
opened
.log
.append_transaction(Uuid::now_v7(), &[b"durable".to_vec()])?;
drop(opened);
let valid_length = std::fs::metadata(&path)?.len();
OpenOptions::new()
.append(true)
.open(&path)?
.write_all(b"partial")?;
let reopened = open_for_test(&path)?;
assert_eq!(reopened.recovery.truncated_tail_bytes, 7);
assert_eq!(std::fs::metadata(&path)?.len(), valid_length);
assert_eq!(reopened.recovery.transactions.len(), 1);
Ok(())
}
#[test]
fn rejects_complete_corruption_without_truncating() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-corruption")?;
let path = temporary.path().join("segment.hylog");
let mut opened = open_for_test(&path)?;
opened
.log
.append_transaction(Uuid::now_v7(), &[b"durable".to_vec()])?;
drop(opened);
let original_length = std::fs::metadata(&path)?.len();
let payload_offset = u64::try_from(HEADER_LENGTH * 2)? + 36;
let mut file = OpenOptions::new().read(true).write(true).open(&path)?;
file.seek(SeekFrom::Start(payload_offset))?;
let mut byte = [0_u8; 1];
file.read_exact(&mut byte)?;
byte[0] ^= 0x01;
file.seek(SeekFrom::Start(payload_offset))?;
file.write_all(&byte)?;
file.sync_all()?;
drop(file);
let result = open_for_test(&path);
assert!(matches!(result, Err(LogError::ChecksumMismatch { .. })));
assert_eq!(std::fs::metadata(&path)?.len(), original_length);
Ok(())
}
#[test]
fn retry_supersedes_an_uncommitted_attempt() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-retry")?;
let path = temporary.path().join("segment.hylog");
let transaction_id = Uuid::now_v7();
let operations = [b"complete".to_vec()];
let mut opened = open_for_test(&path)?;
let digest = super::transaction_digest(&operations, 1)?;
let descriptor = super::encode_descriptor(1, digest);
opened
.log
.append_frame(super::FrameKind::Begin, transaction_id, &descriptor)?;
opened
.log
.append_frame(super::FrameKind::Operation, transaction_id, b"incomplete")?;
opened.log.file.sync_data()?;
drop(opened);
let mut recovered = open_for_test(&path)?;
assert_eq!(recovered.recovery.ignored_uncommitted_transactions, 1);
recovered
.log
.append_transaction(transaction_id, &operations)?;
drop(recovered);
let final_open = open_for_test(&path)?;
assert_eq!(final_open.recovery.transactions.len(), 1);
assert_eq!(final_open.recovery.transactions[0].operations, operations);
Ok(())
}
#[test]
fn every_incomplete_transaction_prefix_is_atomic() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-byte-cuts")?;
let seed_path = temporary.path().join("seed.hylog");
let target_path = temporary.path().join("cut.hylog");
let mut seed = open_for_test(&seed_path)?;
seed.log
.append_transaction(Uuid::now_v7(), &[b"first".to_vec(), b"second".to_vec()])?;
drop(seed);
let complete = std::fs::read(&seed_path)?;
for cut in 0..complete.len() {
std::fs::write(&target_path, &complete[..cut])?;
let recovered = open_for_test(&target_path)?;
assert!(
recovered.recovery.transactions.is_empty(),
"cut at byte {cut} exposed an uncommitted transaction"
);
drop(recovered);
}
std::fs::write(&target_path, &complete)?;
let recovered = open_for_test(&target_path)?;
assert_eq!(recovered.recovery.transactions.len(), 1);
Ok(())
}
#[test]
fn future_frame_version_fails_before_payload_allocation() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-future-version")?;
let path = temporary.path().join("segment.hylog");
let mut opened = open_for_test(&path)?;
opened
.log
.append_transaction(Uuid::now_v7(), &[b"durable".to_vec()])?;
drop(opened);
let mut bytes = std::fs::read(&path)?;
bytes[8..10].copy_from_slice(&2_u16.to_le_bytes());
bytes[36..44].copy_from_slice(&u64::MAX.to_le_bytes());
std::fs::write(&path, &bytes)?;
let result = open_for_test(&path);
assert!(matches!(
result,
Err(LogError::UnsupportedVersion {
found: 2,
supported: 1,
..
})
));
Ok(())
}
#[test]
fn anchored_segment_continues_the_global_digest_chain() -> Result<(), Box<dyn Error>> {
let temporary = TestDirectory::new("log-anchored-segment")?;
let first_path = temporary.path().join("first.hylog");
let second_path = temporary.path().join("second.hylog");
let mut first = open_for_test(&first_path)?;
first
.log
.append_transaction(Uuid::now_v7(), &[b"before-compaction".to_vec()])?;
drop(first);
let (_, first_recovery) = DurableLog::open_file(&first_path)?;
let (mut second, empty_recovery) = DurableLog::open_file_at(
&second_path,
first_recovery.last_sequence,
first_recovery.last_digest,
)?;
assert_eq!(empty_recovery.base_sequence, first_recovery.last_sequence);
assert_eq!(empty_recovery.last_digest, first_recovery.last_digest);
let outcome = second.append_transaction(Uuid::now_v7(), &[b"after-compaction".to_vec()])?;
let AppendOutcome::Committed(receipt) = outcome else {
return Err("new anchored transaction was not committed".into());
};
assert_eq!(receipt.commit_sequence, first_recovery.last_sequence + 3);
drop(second);
let (_, reopened) = DurableLog::open_file_at(
&second_path,
first_recovery.last_sequence,
first_recovery.last_digest,
)?;
assert_eq!(reopened.transactions.len(), 1);
let wrong_anchor =
DurableLog::open_file_at(&second_path, first_recovery.last_sequence, [9; 32]);
assert!(matches!(
wrong_anchor,
Err(LogError::PreviousDigestMismatch { .. })
));
Ok(())
}
}