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

hyphae_storage/log/
mod.rs
