hyphae_storage/log/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2
3mod frame;
4
5use std::{
6    collections::HashMap,
7    fs::{File, OpenOptions},
8    io::{self, Seek, SeekFrom, Write},
9    marker::PhantomData,
10    path::Path,
11};
12
13use thiserror::Error;
14use uuid::Uuid;
15
16use self::frame::{
17    Frame, FrameKind, HEADER_LENGTH, MAX_PAYLOAD_LENGTH, ReadStatus, payload_length,
18    read_exact_or_tail,
19};
20
21const DESCRIPTOR_LENGTH: usize = 36;
22const TRANSACTION_DOMAIN: &[u8] = b"hyphae-transaction-v1";
23pub(crate) const MAX_OPERATION_BYTES: usize = MAX_PAYLOAD_LENGTH;
24
25/// Failure while opening, validating, or appending to a durable log.
26#[derive(Debug, Error)]
27pub enum LogError {
28    /// A filesystem operation failed.
29    #[error(transparent)]
30    Io(#[from] io::Error),
31
32    /// A frame does not start with the Hyphae magic bytes.
33    #[error("invalid frame magic at byte offset {offset}")]
34    BadMagic {
35        /// Frame offset.
36        offset: u64,
37    },
38
39    /// A frame uses a disk version this binary cannot decode.
40    #[error(
41        "unsupported log version {found} at byte offset {offset}; supported version is {supported}"
42    )]
43    UnsupportedVersion {
44        /// Frame offset.
45        offset: u64,
46        /// Version found on disk.
47        found: u16,
48        /// Version understood by this binary.
49        supported: u16,
50    },
51
52    /// A frame kind is not part of this format version.
53    #[error("unknown frame kind {kind} at byte offset {offset}")]
54    UnknownFrameKind {
55        /// Frame offset.
56        offset: u64,
57        /// Raw kind byte.
58        kind: u8,
59    },
60
61    /// Reserved frame flags are nonzero.
62    #[error("unsupported frame flags {flags:#04x} at byte offset {offset}")]
63    UnsupportedFlags {
64        /// Frame offset.
65        offset: u64,
66        /// Raw flag byte.
67        flags: u8,
68    },
69
70    /// A payload exceeds the per-frame allocation limit.
71    #[error("frame payload is {length} bytes; maximum is {maximum}")]
72    PayloadTooLarge {
73        /// Requested or decoded length.
74        length: usize,
75        /// Configured maximum.
76        maximum: usize,
77    },
78
79    /// A frame sequence is not exactly the previous sequence plus one.
80    #[error("invalid sequence at byte offset {offset}: expected {expected}, found {found}")]
81    InvalidSequence {
82        /// Frame offset.
83        offset: u64,
84        /// Expected sequence.
85        expected: u64,
86        /// Sequence found.
87        found: u64,
88    },
89
90    /// The digest chain does not connect to the prior frame.
91    #[error("previous-frame digest mismatch at sequence {sequence}")]
92    PreviousDigestMismatch {
93        /// Invalid frame sequence.
94        sequence: u64,
95    },
96
97    /// The CRC32C integrity check failed.
98    #[error("CRC32C mismatch at sequence {sequence}")]
99    ChecksumMismatch {
100        /// Invalid frame sequence.
101        sequence: u64,
102    },
103
104    /// The BLAKE3 frame digest failed.
105    #[error("BLAKE3 digest mismatch at sequence {sequence}")]
106    DigestMismatch {
107        /// Invalid frame sequence.
108        sequence: u64,
109    },
110
111    /// A transaction descriptor has the wrong length or contents.
112    #[error("malformed transaction descriptor at sequence {sequence}")]
113    MalformedTransaction {
114        /// Invalid frame sequence.
115        sequence: u64,
116    },
117
118    /// An operation or commit appeared without its matching begin frame.
119    #[error("{kind} frame at sequence {sequence} has no matching transaction begin")]
120    TransactionBoundary {
121        /// Frame kind being validated.
122        kind: &'static str,
123        /// Invalid frame sequence.
124        sequence: u64,
125    },
126
127    /// The committed operation count or digest differs from its descriptor.
128    #[error("transaction content mismatch at commit sequence {sequence}")]
129    TransactionContentMismatch {
130        /// Invalid commit sequence.
131        sequence: u64,
132    },
133
134    /// A transaction identifier was reused for different contents.
135    #[error(
136        "transaction identifier {transaction_id} was already committed with different contents"
137    )]
138    IdempotencyConflict {
139        /// Reused identifier.
140        transaction_id: Uuid,
141    },
142
143    /// A transaction cannot be committed without at least one operation.
144    #[error("a transaction must contain at least one operation")]
145    EmptyTransaction,
146
147    /// The operation count cannot be represented by the disk format.
148    #[error("transaction has too many operations")]
149    TooManyOperations,
150
151    /// The sequence space has been exhausted.
152    #[error("log sequence space is exhausted")]
153    SequenceExhausted,
154
155    /// A segment base sequence and digest do not form a canonical anchor.
156    #[error("invalid log segment anchor")]
157    InvalidAnchor,
158
159    /// The writer observed an uncertain I/O result and must be reopened.
160    #[error("durable log writer is poisoned; reopen it before writing again")]
161    Poisoned,
162}
163
164/// Durable identity of a committed transaction.
165#[derive(Clone, Copy, Debug, Eq, PartialEq)]
166pub struct CommitReceipt {
167    /// Caller-supplied idempotency key.
168    pub transaction_id: Uuid,
169    /// Sequence of the durable commit frame.
170    pub commit_sequence: u64,
171    /// Digest of the commit frame and its chain prefix.
172    pub commit_digest: [u8; 32],
173    /// Digest of the canonical operation list.
174    pub transaction_digest: [u8; 32],
175}
176
177/// Result of an idempotent append request.
178#[derive(Clone, Copy, Debug, Eq, PartialEq)]
179pub enum AppendOutcome {
180    /// New frames were written and synchronized.
181    Committed(CommitReceipt),
182    /// The exact transaction was already durable; no frames were appended.
183    Existing(CommitReceipt),
184}
185
186/// A committed transaction reconstructed from the verified log.
187#[derive(Clone, Debug, Eq, PartialEq)]
188pub struct RecoveredTransaction {
189    /// Durable commit identity.
190    pub receipt: CommitReceipt,
191    /// Opaque operation payloads in original order.
192    pub operations: Vec<Vec<u8>>,
193}
194
195/// Evidence produced while opening and validating a segment.
196#[derive(Clone, Debug, Default, Eq, PartialEq)]
197pub struct RecoveryReport {
198    /// Sequence immediately preceding this segment, or zero for the first segment.
199    pub base_sequence: u64,
200    /// Digest immediately preceding this segment, or all zero for the first segment.
201    pub base_digest: [u8; 32],
202    /// Unique committed transactions in commit order.
203    pub transactions: Vec<RecoveredTransaction>,
204    /// Complete but uncommitted transaction attempts ignored during recovery.
205    pub ignored_uncommitted_transactions: u64,
206    /// Repeated commits with the same id and content, deduplicated during replay.
207    pub duplicate_commits: u64,
208    /// Incomplete bytes removed from the physical tail.
209    pub truncated_tail_bytes: u64,
210    /// Length after any incomplete tail was removed.
211    pub valid_bytes: u64,
212    /// Last complete frame sequence, including uncommitted attempts.
213    pub last_sequence: u64,
214    /// Digest of the last complete frame.
215    pub last_digest: [u8; 32],
216}
217
218/// A newly opened writer together with its recovery evidence.
219#[derive(Debug)]
220pub struct OpenedLog<'directory> {
221    /// Exclusive writer handle.
222    pub log: DurableLog,
223    /// Verified replay and tail-repair report.
224    pub recovery: RecoveryReport,
225    directory_lock: PhantomData<&'directory crate::DataDirectory>,
226}
227
228impl OpenedLog<'_> {
229    pub(crate) fn new(log: DurableLog, recovery: RecoveryReport) -> Self {
230        Self {
231            log,
232            recovery,
233            directory_lock: PhantomData,
234        }
235    }
236}
237
238/// Append-only transaction log with synchronous commit durability.
239#[derive(Debug)]
240pub struct DurableLog {
241    file: File,
242    next_sequence: u64,
243    previous_digest: [u8; 32],
244    committed: HashMap<Uuid, CommitReceipt>,
245    poisoned: bool,
246    #[cfg(test)]
247    fail_next_sync: bool,
248}
249
250impl DurableLog {
251    /// Opens, verifies, and repairs only an incomplete physical tail.
252    ///
253    /// Full frames with invalid checksums, digests, versions, sequences, or
254    /// transaction boundaries are rejected as corruption and never truncated.
255    ///
256    /// # Errors
257    ///
258    /// Returns an error for I/O failures or any complete invalid frame.
259    #[cfg(test)]
260    pub(crate) fn open_file(
261        path: impl AsRef<Path>,
262    ) -> Result<(DurableLog, RecoveryReport), LogError> {
263        Self::open_file_at(path, 0, [0; 32])
264    }
265
266    pub(crate) fn open_file_at(
267        path: impl AsRef<Path>,
268        base_sequence: u64,
269        base_digest: [u8; 32],
270    ) -> Result<(DurableLog, RecoveryReport), LogError> {
271        if (base_sequence == 0) != (base_digest == [0; 32]) {
272            return Err(LogError::InvalidAnchor);
273        }
274        let path = path.as_ref();
275        if let Some(parent) = path.parent() {
276            std::fs::create_dir_all(parent)?;
277        }
278        let existed = path.exists();
279        let mut file = OpenOptions::new()
280            .create(true)
281            .read(true)
282            .write(true)
283            .truncate(false)
284            .open(path)?;
285        if !existed {
286            file.sync_all()?;
287            #[cfg(unix)]
288            if let Some(parent) = path.parent() {
289                File::open(parent)?.sync_all()?;
290            }
291        }
292        let recovery = scan(&mut file, base_sequence, base_digest)?;
293        let physical_length = file.metadata()?.len();
294        if physical_length != recovery.valid_bytes {
295            file.set_len(recovery.valid_bytes)?;
296            file.sync_data()?;
297        }
298        file.seek(SeekFrom::End(0))?;
299
300        let committed = recovery
301            .transactions
302            .iter()
303            .map(|transaction| (transaction.receipt.transaction_id, transaction.receipt))
304            .collect();
305        let next_sequence = recovery
306            .last_sequence
307            .checked_add(1)
308            .ok_or(LogError::SequenceExhausted)?;
309        let log = Self {
310            file,
311            next_sequence,
312            previous_digest: recovery.last_digest,
313            committed,
314            poisoned: false,
315            #[cfg(test)]
316            fail_next_sync: false,
317        };
318        Ok((log, recovery))
319    }
320
321    /// Appends and synchronizes one atomic transaction.
322    ///
323    /// Retrying the same identifier and operation bytes returns the original
324    /// receipt without appending. Reusing it with different bytes fails.
325    ///
326    /// # Errors
327    ///
328    /// Returns an error for invalid bounds, conflicting idempotency keys, or
329    /// filesystem failures. Any append/sync failure poisons the writer so the
330    /// caller must reopen and recover before another write.
331    pub fn append_transaction(
332        &mut self,
333        transaction_id: Uuid,
334        operations: &[Vec<u8>],
335    ) -> Result<AppendOutcome, LogError> {
336        if self.poisoned {
337            return Err(LogError::Poisoned);
338        }
339        if operations.is_empty() {
340            return Err(LogError::EmptyTransaction);
341        }
342        let operation_count =
343            u32::try_from(operations.len()).map_err(|_| LogError::TooManyOperations)?;
344        for operation in operations {
345            if operation.len() > MAX_PAYLOAD_LENGTH {
346                return Err(LogError::PayloadTooLarge {
347                    length: operation.len(),
348                    maximum: MAX_PAYLOAD_LENGTH,
349                });
350            }
351        }
352
353        let transaction_digest = transaction_digest(operations, operation_count)?;
354        if let Some(receipt) = self.committed.get(&transaction_id).copied() {
355            return if receipt.transaction_digest == transaction_digest {
356                Ok(AppendOutcome::Existing(receipt))
357            } else {
358                Err(LogError::IdempotencyConflict { transaction_id })
359            };
360        }
361
362        let descriptor = encode_descriptor(operation_count, transaction_digest);
363        let append_result = self.append_new_transaction(
364            transaction_id,
365            operations,
366            &descriptor,
367            transaction_digest,
368        );
369        if append_result.is_err() {
370            self.poisoned = true;
371        }
372        append_result
373    }
374
375    pub(crate) fn is_poisoned(&self) -> bool {
376        self.poisoned
377    }
378
379    #[cfg(test)]
380    pub(crate) fn inject_sync_failure(&mut self) {
381        self.fail_next_sync = true;
382    }
383
384    fn append_new_transaction(
385        &mut self,
386        transaction_id: Uuid,
387        operations: &[Vec<u8>],
388        descriptor: &[u8; DESCRIPTOR_LENGTH],
389        transaction_digest: [u8; 32],
390    ) -> Result<AppendOutcome, LogError> {
391        self.append_frame(FrameKind::Begin, transaction_id, descriptor)?;
392        for operation in operations {
393            self.append_frame(FrameKind::Operation, transaction_id, operation)?;
394        }
395        let receipt = self.append_frame(FrameKind::Commit, transaction_id, descriptor)?;
396        #[cfg(test)]
397        if std::mem::take(&mut self.fail_next_sync) {
398            return Err(io::Error::other("injected log sync failure").into());
399        }
400        self.file.sync_data()?;
401
402        let receipt = CommitReceipt {
403            transaction_id,
404            commit_sequence: receipt.sequence,
405            commit_digest: receipt.digest,
406            transaction_digest,
407        };
408        self.committed.insert(transaction_id, receipt);
409        Ok(AppendOutcome::Committed(receipt))
410    }
411
412    fn append_frame(
413        &mut self,
414        kind: FrameKind,
415        transaction_id: Uuid,
416        payload: &[u8],
417    ) -> Result<WrittenFrame, LogError> {
418        let frame = Frame {
419            kind,
420            sequence: self.next_sequence,
421            transaction_id,
422            previous_digest: self.previous_digest,
423            digest: [0; 32],
424            payload: payload.to_vec(),
425        };
426        let encoded = frame.encode()?;
427        let digest = copy_array(&encoded[80..112]);
428        self.file.write_all(&encoded)?;
429        let written = WrittenFrame {
430            sequence: self.next_sequence,
431            digest,
432        };
433        self.next_sequence = self
434            .next_sequence
435            .checked_add(1)
436            .ok_or(LogError::SequenceExhausted)?;
437        self.previous_digest = digest;
438        Ok(written)
439    }
440}
441
442#[derive(Clone, Copy, Debug)]
443struct WrittenFrame {
444    sequence: u64,
445    digest: [u8; 32],
446}
447
448#[derive(Debug)]
449struct PendingTransaction {
450    transaction_id: Uuid,
451    operation_count: u32,
452    expected_digest: [u8; 32],
453    operations: Vec<Vec<u8>>,
454}
455
456fn scan(
457    file: &mut File,
458    base_sequence: u64,
459    base_digest: [u8; 32],
460) -> Result<RecoveryReport, LogError> {
461    file.seek(SeekFrom::Start(0))?;
462    let physical_length = file.metadata()?.len();
463    let mut report = RecoveryReport {
464        base_sequence,
465        base_digest,
466        last_sequence: base_sequence,
467        last_digest: base_digest,
468        ..RecoveryReport::default()
469    };
470    let mut offset = 0_u64;
471    let mut expected_sequence = base_sequence
472        .checked_add(1)
473        .ok_or(LogError::SequenceExhausted)?;
474    let mut expected_previous_digest = base_digest;
475    let mut pending: Option<PendingTransaction> = None;
476    let mut committed: HashMap<Uuid, CommitReceipt> = HashMap::new();
477
478    loop {
479        let mut header = [0_u8; HEADER_LENGTH];
480        match read_exact_or_tail(file, &mut header)? {
481            ReadStatus::End => break,
482            ReadStatus::Partial => {
483                report.truncated_tail_bytes = physical_length.saturating_sub(offset);
484                break;
485            }
486            ReadStatus::Complete => {}
487        }
488        let length = payload_length(&header, offset)?;
489        let mut payload = vec![0_u8; length];
490        if read_exact_or_tail(file, &mut payload)? != ReadStatus::Complete {
491            report.truncated_tail_bytes = physical_length.saturating_sub(offset);
492            break;
493        }
494
495        let frame = Frame::decode(&header, payload, offset)?;
496        if frame.sequence != expected_sequence {
497            return Err(LogError::InvalidSequence {
498                offset,
499                expected: expected_sequence,
500                found: frame.sequence,
501            });
502        }
503        if frame.previous_digest != expected_previous_digest {
504            return Err(LogError::PreviousDigestMismatch {
505                sequence: frame.sequence,
506            });
507        }
508
509        apply_frame(&frame, &mut pending, &mut committed, &mut report)?;
510        let frame_length = u64::try_from(HEADER_LENGTH + frame.payload.len()).map_err(|_| {
511            LogError::PayloadTooLarge {
512                length: frame.payload.len(),
513                maximum: MAX_PAYLOAD_LENGTH,
514            }
515        })?;
516        offset = offset
517            .checked_add(frame_length)
518            .ok_or(LogError::SequenceExhausted)?;
519        report.valid_bytes = offset;
520        report.last_sequence = frame.sequence;
521        report.last_digest = frame.digest;
522        expected_sequence = expected_sequence
523            .checked_add(1)
524            .ok_or(LogError::SequenceExhausted)?;
525        expected_previous_digest = frame.digest;
526    }
527
528    if pending.is_some() {
529        report.ignored_uncommitted_transactions =
530            report.ignored_uncommitted_transactions.saturating_add(1);
531    }
532    Ok(report)
533}
534
535fn apply_frame(
536    frame: &Frame,
537    pending: &mut Option<PendingTransaction>,
538    committed: &mut HashMap<Uuid, CommitReceipt>,
539    report: &mut RecoveryReport,
540) -> Result<(), LogError> {
541    match frame.kind {
542        FrameKind::Begin => {
543            if pending.is_some() {
544                report.ignored_uncommitted_transactions =
545                    report.ignored_uncommitted_transactions.saturating_add(1);
546            }
547            let (operation_count, expected_digest) =
548                decode_descriptor(&frame.payload, frame.sequence)?;
549            *pending = Some(PendingTransaction {
550                transaction_id: frame.transaction_id,
551                operation_count,
552                expected_digest,
553                operations: Vec::new(),
554            });
555            Ok(())
556        }
557        FrameKind::Operation => {
558            let Some(current) = pending
559                .as_mut()
560                .filter(|current| current.transaction_id == frame.transaction_id)
561            else {
562                return Err(LogError::TransactionBoundary {
563                    kind: "operation",
564                    sequence: frame.sequence,
565                });
566            };
567            current.operations.push(frame.payload.clone());
568            Ok(())
569        }
570        FrameKind::Commit => {
571            let Some(current) = pending
572                .take()
573                .filter(|current| current.transaction_id == frame.transaction_id)
574            else {
575                return Err(LogError::TransactionBoundary {
576                    kind: "commit",
577                    sequence: frame.sequence,
578                });
579            };
580            let (operation_count, commit_digest) =
581                decode_descriptor(&frame.payload, frame.sequence)?;
582            let actual_count =
583                u32::try_from(current.operations.len()).map_err(|_| LogError::TooManyOperations)?;
584            let actual_digest = transaction_digest(&current.operations, actual_count)?;
585            if operation_count != current.operation_count
586                || commit_digest != current.expected_digest
587                || actual_count != operation_count
588                || actual_digest != commit_digest
589            {
590                return Err(LogError::TransactionContentMismatch {
591                    sequence: frame.sequence,
592                });
593            }
594
595            let receipt = CommitReceipt {
596                transaction_id: frame.transaction_id,
597                commit_sequence: frame.sequence,
598                commit_digest: frame.digest,
599                transaction_digest: actual_digest,
600            };
601            if let Some(existing) = committed.get(&frame.transaction_id) {
602                if existing.transaction_digest != actual_digest {
603                    return Err(LogError::IdempotencyConflict {
604                        transaction_id: frame.transaction_id,
605                    });
606                }
607                report.duplicate_commits = report.duplicate_commits.saturating_add(1);
608            } else {
609                committed.insert(frame.transaction_id, receipt);
610                report.transactions.push(RecoveredTransaction {
611                    receipt,
612                    operations: current.operations,
613                });
614            }
615            Ok(())
616        }
617    }
618}
619
620fn encode_descriptor(operation_count: u32, digest: [u8; 32]) -> [u8; DESCRIPTOR_LENGTH] {
621    let mut descriptor = [0_u8; DESCRIPTOR_LENGTH];
622    descriptor[..4].copy_from_slice(&operation_count.to_le_bytes());
623    descriptor[4..].copy_from_slice(&digest);
624    descriptor
625}
626
627fn decode_descriptor(payload: &[u8], sequence: u64) -> Result<(u32, [u8; 32]), LogError> {
628    if payload.len() != DESCRIPTOR_LENGTH {
629        return Err(LogError::MalformedTransaction { sequence });
630    }
631    let operation_count = u32::from_le_bytes(copy_array(&payload[..4]));
632    if operation_count == 0 {
633        return Err(LogError::MalformedTransaction { sequence });
634    }
635    let digest = copy_array(&payload[4..]);
636    Ok((operation_count, digest))
637}
638
639pub(crate) fn transaction_digest(
640    operations: &[Vec<u8>],
641    operation_count: u32,
642) -> Result<[u8; 32], LogError> {
643    let mut hasher = blake3::Hasher::new();
644    hasher.update(TRANSACTION_DOMAIN);
645    hasher.update(&u64::from(operation_count).to_le_bytes());
646    for operation in operations {
647        let length = u64::try_from(operation.len()).map_err(|_| LogError::PayloadTooLarge {
648            length: operation.len(),
649            maximum: MAX_PAYLOAD_LENGTH,
650        })?;
651        hasher.update(&length.to_le_bytes());
652        hasher.update(operation);
653    }
654    Ok(*hasher.finalize().as_bytes())
655}
656
657fn copy_array<const N: usize>(source: &[u8]) -> [u8; N] {
658    let mut output = [0_u8; N];
659    output.copy_from_slice(source);
660    output
661}
662
663#[cfg(test)]
664mod tests {
665    use std::{
666        error::Error,
667        fs::OpenOptions,
668        io::{Read, Seek, SeekFrom, Write},
669    };
670
671    use uuid::Uuid;
672
673    use super::{AppendOutcome, DurableLog, LogError, OpenedLog, frame::HEADER_LENGTH};
674    use crate::test_support::TestDirectory;
675
676    fn open_for_test(path: &std::path::Path) -> Result<OpenedLog<'static>, LogError> {
677        let (log, recovery) = DurableLog::open_file(path)?;
678        Ok(OpenedLog::new(log, recovery))
679    }
680
681    #[test]
682    fn committed_transaction_recovers_in_order() -> Result<(), Box<dyn Error>> {
683        let temporary = TestDirectory::new("log-recovery")?;
684        let path = temporary.path().join("segment.hylog");
685        let transaction_id = Uuid::now_v7();
686        let mut opened = open_for_test(&path)?;
687        let outcome = opened
688            .log
689            .append_transaction(transaction_id, &[b"put:a=1".to_vec(), b"put:b=2".to_vec()])?;
690        assert!(matches!(outcome, AppendOutcome::Committed(_)));
691        drop(opened);
692
693        let reopened = open_for_test(&path)?;
694        assert_eq!(reopened.recovery.transactions.len(), 1);
695        assert_eq!(
696            reopened.recovery.transactions[0].operations,
697            [b"put:a=1".to_vec(), b"put:b=2".to_vec()]
698        );
699        Ok(())
700    }
701
702    #[test]
703    fn idempotency_survives_reopen() -> Result<(), Box<dyn Error>> {
704        let temporary = TestDirectory::new("log-idempotency")?;
705        let path = temporary.path().join("segment.hylog");
706        let transaction_id = Uuid::now_v7();
707        let operations = [b"same".to_vec()];
708        let mut opened = open_for_test(&path)?;
709        let first = opened.log.append_transaction(transaction_id, &operations)?;
710        drop(opened);
711
712        let mut reopened = open_for_test(&path)?;
713        let second = reopened
714            .log
715            .append_transaction(transaction_id, &operations)?;
716        assert!(matches!(first, AppendOutcome::Committed(_)));
717        assert!(matches!(second, AppendOutcome::Existing(_)));
718
719        let conflict = reopened
720            .log
721            .append_transaction(transaction_id, &[b"different".to_vec()]);
722        assert!(matches!(
723            conflict,
724            Err(LogError::IdempotencyConflict { .. })
725        ));
726        Ok(())
727    }
728
729    #[test]
730    fn truncates_only_an_incomplete_tail() -> Result<(), Box<dyn Error>> {
731        let temporary = TestDirectory::new("log-tail")?;
732        let path = temporary.path().join("segment.hylog");
733        let mut opened = open_for_test(&path)?;
734        opened
735            .log
736            .append_transaction(Uuid::now_v7(), &[b"durable".to_vec()])?;
737        drop(opened);
738        let valid_length = std::fs::metadata(&path)?.len();
739
740        OpenOptions::new()
741            .append(true)
742            .open(&path)?
743            .write_all(b"partial")?;
744        let reopened = open_for_test(&path)?;
745        assert_eq!(reopened.recovery.truncated_tail_bytes, 7);
746        assert_eq!(std::fs::metadata(&path)?.len(), valid_length);
747        assert_eq!(reopened.recovery.transactions.len(), 1);
748        Ok(())
749    }
750
751    #[test]
752    fn rejects_complete_corruption_without_truncating() -> Result<(), Box<dyn Error>> {
753        let temporary = TestDirectory::new("log-corruption")?;
754        let path = temporary.path().join("segment.hylog");
755        let mut opened = open_for_test(&path)?;
756        opened
757            .log
758            .append_transaction(Uuid::now_v7(), &[b"durable".to_vec()])?;
759        drop(opened);
760        let original_length = std::fs::metadata(&path)?.len();
761
762        let payload_offset = u64::try_from(HEADER_LENGTH * 2)? + 36;
763        let mut file = OpenOptions::new().read(true).write(true).open(&path)?;
764        file.seek(SeekFrom::Start(payload_offset))?;
765        let mut byte = [0_u8; 1];
766        file.read_exact(&mut byte)?;
767        byte[0] ^= 0x01;
768        file.seek(SeekFrom::Start(payload_offset))?;
769        file.write_all(&byte)?;
770        file.sync_all()?;
771        drop(file);
772
773        let result = open_for_test(&path);
774        assert!(matches!(result, Err(LogError::ChecksumMismatch { .. })));
775        assert_eq!(std::fs::metadata(&path)?.len(), original_length);
776        Ok(())
777    }
778
779    #[test]
780    fn retry_supersedes_an_uncommitted_attempt() -> Result<(), Box<dyn Error>> {
781        let temporary = TestDirectory::new("log-retry")?;
782        let path = temporary.path().join("segment.hylog");
783        let transaction_id = Uuid::now_v7();
784        let operations = [b"complete".to_vec()];
785
786        let mut opened = open_for_test(&path)?;
787        let digest = super::transaction_digest(&operations, 1)?;
788        let descriptor = super::encode_descriptor(1, digest);
789        opened
790            .log
791            .append_frame(super::FrameKind::Begin, transaction_id, &descriptor)?;
792        opened
793            .log
794            .append_frame(super::FrameKind::Operation, transaction_id, b"incomplete")?;
795        opened.log.file.sync_data()?;
796        drop(opened);
797
798        let mut recovered = open_for_test(&path)?;
799        assert_eq!(recovered.recovery.ignored_uncommitted_transactions, 1);
800        recovered
801            .log
802            .append_transaction(transaction_id, &operations)?;
803        drop(recovered);
804
805        let final_open = open_for_test(&path)?;
806        assert_eq!(final_open.recovery.transactions.len(), 1);
807        assert_eq!(final_open.recovery.transactions[0].operations, operations);
808        Ok(())
809    }
810
811    #[test]
812    fn every_incomplete_transaction_prefix_is_atomic() -> Result<(), Box<dyn Error>> {
813        let temporary = TestDirectory::new("log-byte-cuts")?;
814        let seed_path = temporary.path().join("seed.hylog");
815        let target_path = temporary.path().join("cut.hylog");
816        let mut seed = open_for_test(&seed_path)?;
817        seed.log
818            .append_transaction(Uuid::now_v7(), &[b"first".to_vec(), b"second".to_vec()])?;
819        drop(seed);
820        let complete = std::fs::read(&seed_path)?;
821
822        for cut in 0..complete.len() {
823            std::fs::write(&target_path, &complete[..cut])?;
824            let recovered = open_for_test(&target_path)?;
825            assert!(
826                recovered.recovery.transactions.is_empty(),
827                "cut at byte {cut} exposed an uncommitted transaction"
828            );
829            drop(recovered);
830        }
831
832        std::fs::write(&target_path, &complete)?;
833        let recovered = open_for_test(&target_path)?;
834        assert_eq!(recovered.recovery.transactions.len(), 1);
835        Ok(())
836    }
837
838    #[test]
839    fn future_frame_version_fails_before_payload_allocation() -> Result<(), Box<dyn Error>> {
840        let temporary = TestDirectory::new("log-future-version")?;
841        let path = temporary.path().join("segment.hylog");
842        let mut opened = open_for_test(&path)?;
843        opened
844            .log
845            .append_transaction(Uuid::now_v7(), &[b"durable".to_vec()])?;
846        drop(opened);
847
848        let mut bytes = std::fs::read(&path)?;
849        bytes[8..10].copy_from_slice(&2_u16.to_le_bytes());
850        bytes[36..44].copy_from_slice(&u64::MAX.to_le_bytes());
851        std::fs::write(&path, &bytes)?;
852
853        let result = open_for_test(&path);
854        assert!(matches!(
855            result,
856            Err(LogError::UnsupportedVersion {
857                found: 2,
858                supported: 1,
859                ..
860            })
861        ));
862        Ok(())
863    }
864
865    #[test]
866    fn anchored_segment_continues_the_global_digest_chain() -> Result<(), Box<dyn Error>> {
867        let temporary = TestDirectory::new("log-anchored-segment")?;
868        let first_path = temporary.path().join("first.hylog");
869        let second_path = temporary.path().join("second.hylog");
870        let mut first = open_for_test(&first_path)?;
871        first
872            .log
873            .append_transaction(Uuid::now_v7(), &[b"before-compaction".to_vec()])?;
874        drop(first);
875        let (_, first_recovery) = DurableLog::open_file(&first_path)?;
876
877        let (mut second, empty_recovery) = DurableLog::open_file_at(
878            &second_path,
879            first_recovery.last_sequence,
880            first_recovery.last_digest,
881        )?;
882        assert_eq!(empty_recovery.base_sequence, first_recovery.last_sequence);
883        assert_eq!(empty_recovery.last_digest, first_recovery.last_digest);
884        let outcome = second.append_transaction(Uuid::now_v7(), &[b"after-compaction".to_vec()])?;
885        let AppendOutcome::Committed(receipt) = outcome else {
886            return Err("new anchored transaction was not committed".into());
887        };
888        assert_eq!(receipt.commit_sequence, first_recovery.last_sequence + 3);
889        drop(second);
890
891        let (_, reopened) = DurableLog::open_file_at(
892            &second_path,
893            first_recovery.last_sequence,
894            first_recovery.last_digest,
895        )?;
896        assert_eq!(reopened.transactions.len(), 1);
897
898        let wrong_anchor =
899            DurableLog::open_file_at(&second_path, first_recovery.last_sequence, [9; 32]);
900        assert!(matches!(
901            wrong_anchor,
902            Err(LogError::PreviousDigestMismatch { .. })
903        ));
904        Ok(())
905    }
906}
hyphae_storage/log/mod.rs

hyphae_storage/log/
mod.rs
