vsh_commit/
journal.rs

1use std::collections::BTreeMap;
2use std::error::Error;
3use std::fmt;
4use std::io::{self, Read, Write};
5
6use cap_std::fs::{Dir, File};
7use vsh_types::{FileStamp, NodeKind, PlatformFileId, TransactionId};
8
9use crate::host::{create_new_file, open_real_file, sync_dir};
10
11const JOURNAL_MAGIC: &[u8; 8] = b"VSHLOG01";
12const MARKER_MAGIC: &[u8; 8] = b"VSHDONE1";
13const MAX_RECORD_PAYLOAD: usize = 64;
14
15pub(crate) const PLAN_FILE: &str = "plan";
16pub(crate) const JOURNAL_FILE: &str = "journal";
17pub(crate) const COMMIT_MARKER: &str = "commit-complete";
18pub(crate) const STAGE_DIRECTORY: &str = "stage";
19pub(crate) const QUARANTINE_DIRECTORY: &str = "quarantine";
20
21#[derive(Clone, Copy, Debug, Eq, PartialEq)]
22pub(crate) struct Witness {
23    pub(crate) kind: NodeKind,
24    pub(crate) file_id: PlatformFileId,
25}
26
27impl From<FileStamp> for Witness {
28    fn from(stamp: FileStamp) -> Self {
29        Self {
30            kind: stamp.kind,
31            file_id: stamp.file_id,
32        }
33    }
34}
35
36#[derive(Debug)]
37pub(crate) struct Journal {
38    file: File,
39}
40
41impl Journal {
42    pub(crate) fn create(transaction_dir: &Dir) -> Result<Self, io::Error> {
43        let mut file = create_new_file(transaction_dir, JOURNAL_FILE)?;
44        file.write_all(JOURNAL_MAGIC)?;
45        file.sync_all()?;
46        sync_dir(transaction_dir)?;
47        Ok(Self { file })
48    }
49
50    pub(crate) fn intent(
51        &mut self,
52        index: u32,
53        source_witness: Option<Witness>,
54        parent_witness: Witness,
55    ) -> Result<(), io::Error> {
56        let mut payload = Vec::with_capacity(40);
57        payload.push(1);
58        payload.extend_from_slice(&index.to_le_bytes());
59        payload.push(u8::from(source_witness.is_some()) | 2);
60        if let Some(witness) = source_witness {
61            encode_witness(witness, &mut payload);
62        }
63        encode_witness(parent_witness, &mut payload);
64        self.append_record(&payload)
65    }
66
67    pub(crate) fn done(&mut self, index: u32, witness: Witness) -> Result<(), io::Error> {
68        let mut payload = Vec::with_capacity(22);
69        payload.push(2);
70        payload.extend_from_slice(&index.to_le_bytes());
71        payload.push(witness.kind.canonical_tag());
72        payload.extend_from_slice(&witness.file_id.high.to_le_bytes());
73        payload.extend_from_slice(&witness.file_id.low.to_le_bytes());
74        self.append_record(&payload)
75    }
76
77    fn append_record(&mut self, payload: &[u8]) -> Result<(), io::Error> {
78        let len = u32::try_from(payload.len())
79            .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "journal record too large"))?;
80        self.file.write_all(&len.to_le_bytes())?;
81        self.file.write_all(payload)?;
82        self.file.write_all(&record_digest(payload))?;
83        self.file.sync_all()
84    }
85}
86
87#[derive(Clone, Debug, Default, Eq, PartialEq)]
88pub(crate) struct JournalState {
89    pub(crate) completed: BTreeMap<u32, Witness>,
90    intents: BTreeMap<u32, IntentWitnesses>,
91    pending: Option<u32>,
92    pub(crate) torn_tail: bool,
93}
94
95#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
96struct IntentWitnesses {
97    source: Option<Witness>,
98    parent: Option<Witness>,
99}
100
101impl JournalState {
102    pub(crate) fn has_intent(&self, index: u32) -> bool {
103        self.intents.contains_key(&index)
104    }
105
106    pub(crate) fn witness(&self, index: u32) -> Option<Witness> {
107        self.completed.get(&index).copied()
108    }
109
110    pub(crate) fn intent_witness(&self, index: u32) -> Option<Witness> {
111        self.intents
112            .get(&index)
113            .and_then(|witnesses| witnesses.source)
114    }
115
116    pub(crate) fn parent_witness(&self, index: u32) -> Option<Witness> {
117        self.intents
118            .get(&index)
119            .and_then(|witnesses| witnesses.parent)
120    }
121}
122
123pub(crate) fn read_journal(
124    transaction_dir: &Dir,
125    maximum_bytes: usize,
126) -> Result<JournalState, JournalError> {
127    let mut file = open_real_file(transaction_dir, JOURNAL_FILE).map_err(JournalError::Io)?;
128    let mut bytes = Vec::new();
129    Read::by_ref(&mut file)
130        .take(
131            u64::try_from(maximum_bytes)
132                .unwrap_or(u64::MAX)
133                .saturating_add(1),
134        )
135        .read_to_end(&mut bytes)
136        .map_err(JournalError::Io)?;
137    if bytes.len() > maximum_bytes {
138        return Err(JournalError::RecordLength);
139    }
140    if bytes.len() < JOURNAL_MAGIC.len() || &bytes[..8] != JOURNAL_MAGIC {
141        return Err(JournalError::Magic);
142    }
143    let mut offset = 8_usize;
144    let mut state = JournalState::default();
145    let mut next_index = 0_u32;
146    while offset < bytes.len() {
147        let Some(length_bytes) = bytes.get(offset..offset.saturating_add(4)) else {
148            state.torn_tail = true;
149            break;
150        };
151        let length = u32::from_le_bytes(
152            length_bytes
153                .try_into()
154                .expect("four-byte journal length slice"),
155        ) as usize;
156        if length > MAX_RECORD_PAYLOAD {
157            return Err(JournalError::RecordLength);
158        }
159        let payload_start = offset + 4;
160        let payload_end = payload_start
161            .checked_add(length)
162            .ok_or(JournalError::RecordLength)?;
163        let checksum_end = payload_end
164            .checked_add(32)
165            .ok_or(JournalError::RecordLength)?;
166        let Some(payload) = bytes.get(payload_start..payload_end) else {
167            state.torn_tail = true;
168            break;
169        };
170        let Some(checksum) = bytes.get(payload_end..checksum_end) else {
171            state.torn_tail = true;
172            break;
173        };
174        if record_digest(payload).as_slice() != checksum {
175            return Err(JournalError::Checksum);
176        }
177        match payload.first().copied() {
178            Some(1) if matches!(payload.len(), 5 | 22 | 23 | 40) => {
179                let index = u32::from_le_bytes(
180                    payload[1..5]
181                        .try_into()
182                        .expect("four-byte intent index slice"),
183                );
184                if state.pending.is_some() || index != next_index {
185                    return Err(JournalError::Sequence);
186                }
187                let witnesses = match payload.len() {
188                    5 => IntentWitnesses::default(),
189                    22 => IntentWitnesses {
190                        source: Some(decode_witness_at(payload, 5)?),
191                        parent: None,
192                    },
193                    23 | 40 => decode_intent_witnesses(payload)?,
194                    _ => unreachable!("guarded intent payload length"),
195                };
196                state.intents.insert(index, witnesses);
197                state.pending = Some(index);
198            }
199            Some(2) if payload.len() == 22 => {
200                let index = u32::from_le_bytes(
201                    payload[1..5]
202                        .try_into()
203                        .expect("four-byte completion index slice"),
204                );
205                if state.pending != Some(index) || index != next_index {
206                    return Err(JournalError::Sequence);
207                }
208                state.completed.insert(index, decode_witness(payload)?);
209                state.pending = None;
210                next_index = next_index.checked_add(1).ok_or(JournalError::Sequence)?;
211            }
212            _ => return Err(JournalError::Tag),
213        }
214        offset = checksum_end;
215    }
216    Ok(state)
217}
218
219fn decode_witness(payload: &[u8]) -> Result<Witness, JournalError> {
220    decode_witness_at(payload, 5)
221}
222
223fn encode_witness(witness: Witness, output: &mut Vec<u8>) {
224    output.push(witness.kind.canonical_tag());
225    output.extend_from_slice(&witness.file_id.high.to_le_bytes());
226    output.extend_from_slice(&witness.file_id.low.to_le_bytes());
227}
228
229fn decode_intent_witnesses(payload: &[u8]) -> Result<IntentWitnesses, JournalError> {
230    let flags = payload[5];
231    if flags & !3 != 0 || flags & 2 == 0 {
232        return Err(JournalError::Tag);
233    }
234    let mut offset = 6;
235    let source = if flags & 1 == 1 {
236        let witness = decode_witness_at(payload, offset)?;
237        offset += 17;
238        Some(witness)
239    } else {
240        None
241    };
242    let parent = Some(decode_witness_at(payload, offset)?);
243    offset += 17;
244    if offset != payload.len() {
245        return Err(JournalError::RecordLength);
246    }
247    Ok(IntentWitnesses { source, parent })
248}
249
250fn decode_witness_at(payload: &[u8], offset: usize) -> Result<Witness, JournalError> {
251    let bytes = payload
252        .get(offset..offset.saturating_add(17))
253        .ok_or(JournalError::RecordLength)?;
254    let kind = match bytes[0] {
255        1 => NodeKind::File,
256        2 => NodeKind::Directory,
257        3 => NodeKind::Symlink,
258        _ => return Err(JournalError::Tag),
259    };
260    let high = u64::from_le_bytes(
261        bytes[1..9]
262            .try_into()
263            .expect("eight-byte witness high slice"),
264    );
265    let low = u64::from_le_bytes(
266        bytes[9..17]
267            .try_into()
268            .expect("eight-byte witness low slice"),
269    );
270    Ok(Witness {
271        kind,
272        file_id: PlatformFileId { high, low },
273    })
274}
275
276pub(crate) fn write_commit_marker(
277    transaction_dir: &Dir,
278    transaction: TransactionId,
279) -> Result<(), io::Error> {
280    let mut payload = Vec::with_capacity(72);
281    payload.extend_from_slice(MARKER_MAGIC);
282    payload.extend_from_slice(transaction.as_bytes());
283    let digest = marker_digest(&payload);
284    payload.extend_from_slice(&digest);
285    let mut marker = create_new_file(transaction_dir, COMMIT_MARKER)?;
286    marker.write_all(&payload)?;
287    marker.sync_all()?;
288    sync_dir(transaction_dir)
289}
290
291pub(crate) fn has_valid_commit_marker(
292    transaction_dir: &Dir,
293    transaction: TransactionId,
294) -> Result<bool, JournalError> {
295    let mut marker = match open_real_file(transaction_dir, COMMIT_MARKER) {
296        Ok(marker) => marker,
297        Err(source) if source.kind() == io::ErrorKind::NotFound => return Ok(false),
298        Err(source) => return Err(JournalError::Io(source)),
299    };
300    let mut bytes = Vec::new();
301    Read::by_ref(&mut marker)
302        .take(73)
303        .read_to_end(&mut bytes)
304        .map_err(JournalError::Io)?;
305    if bytes.len() != 72 || &bytes[..8] != MARKER_MAGIC {
306        return Err(JournalError::Marker);
307    }
308    if &bytes[8..40] != transaction.as_bytes() {
309        return Err(JournalError::Marker);
310    }
311    if marker_digest(&bytes[..40]).as_slice() != &bytes[40..72] {
312        return Err(JournalError::Marker);
313    }
314    Ok(true)
315}
316
317fn record_digest(payload: &[u8]) -> [u8; 32] {
318    let mut hasher = blake3::Hasher::new();
319    hasher.update(b"vsh\0commit-journal-record-v1\0");
320    hasher.update(&(payload.len() as u64).to_le_bytes());
321    hasher.update(payload);
322    *hasher.finalize().as_bytes()
323}
324
325fn marker_digest(payload: &[u8]) -> [u8; 32] {
326    let mut hasher = blake3::Hasher::new();
327    hasher.update(b"vsh\0commit-marker-v1\0");
328    hasher.update(payload);
329    *hasher.finalize().as_bytes()
330}
331
332/// Durable operation-journal or commit-marker validation failure.
333#[derive(Debug)]
334pub enum JournalError {
335    /// Journal I/O failed.
336    Io(io::Error),
337    /// Journal header is unknown.
338    Magic,
339    /// A framed record length is invalid or exceeds its bound.
340    RecordLength,
341    /// A complete record failed its BLAKE3 checksum.
342    Checksum,
343    /// Intent/completion records are not strictly sequential.
344    Sequence,
345    /// A record contains an unknown semantic tag.
346    Tag,
347    /// The durable commit-complete marker is invalid.
348    Marker,
349}
350
351impl fmt::Display for JournalError {
352    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
353        formatter.write_str(match self {
354            Self::Io(_) => "commit journal I/O failed",
355            Self::Magic => "commit journal has an unknown format",
356            Self::RecordLength => "commit journal record length is invalid",
357            Self::Checksum => "commit journal checksum mismatch",
358            Self::Sequence => "commit journal operation sequence is invalid",
359            Self::Tag => "commit journal contains an unknown record tag",
360            Self::Marker => "commit-complete marker is invalid",
361        })
362    }
363}
364
365impl Error for JournalError {
366    fn source(&self) -> Option<&(dyn Error + 'static)> {
367        match self {
368            Self::Io(source) => Some(source),
369            Self::Magic
370            | Self::RecordLength
371            | Self::Checksum
372            | Self::Sequence
373            | Self::Tag
374            | Self::Marker => None,
375        }
376    }
377}
vsh_commit/journal.rs

vsh_commit/
journal.rs
