loonfs_api/
wal.rs

1//! The WAL segment format: envelopes, commit payloads, and the delta
2//! records replay applies (format spec, "WAL segments").
3
4use crate::digest::sha256_digest;
5use crate::envelope::{self, EnvelopeCodecError, EnvelopeProbe};
6use crate::manifest::{DeletedDirentry, TombstoneGeneration};
7use crate::{
8    AccessGrants, AccessRevisionNo, AttributeRevisionNo, Attributes, ChangeSeq, CommitFingerprint,
9    CommitId, ContentId, ContentRef, DisplayName, InodeId, InodeKind, NameKey, NamespaceId,
10    RevisionNo, WalNo, WriterEpoch,
11};
12use ciborium::{de::from_reader, ser::into_writer};
13use serde::{Deserialize, Serialize};
14use std::collections::{BTreeMap, BTreeSet};
15use std::io::Read;
16
17/// Version 1: a zstd-compressed CBOR envelope document carrying the payload
18/// as an opaque CBOR byte string. `payload_checksum` covers exactly those
19/// bytes, and delta/precondition tags use the snake_case names the format
20/// spec fixes ("Standard mutation operations" and "Preconditions").
21pub const WAL_FORMAT_VERSION: u32 = 1;
22
23/// Largest decompressed WAL document allowed by [Appendix A.5](../../../docs/specs/format.md#a5-wal-records).
24pub const MAX_WAL_SEGMENT_BYTES: usize = 512 * 1024 * 1024;
25
26/// Reader limit per inline value in
27/// [Appendix A.5](../../../docs/specs/format.md#a5-wal-records);
28/// writer thresholds are policy at or below this limit.
29pub const MAX_WAL_INLINE_CONTENT_BYTES: usize = 256 * 1024;
30
31/// Reader limit for total inline bytes in one WAL segment in
32/// [Appendix A.5](../../../docs/specs/format.md#a5-wal-records);
33/// writer thresholds are policy at or below this limit.
34pub const MAX_WAL_SEGMENT_INLINE_CONTENT_BYTES: usize = 4 * 1024 * 1024;
35
36/// Upper bound for the document and payload fields outside commit records.
37pub const WAL_SEGMENT_OVERHEAD_BYTES: usize = cbor_map_bytes(&[
38    ("kind", cbor_string_bytes("namespace_wal_segment".len())),
39    ("format_version", 5),
40    ("payload_checksum", cbor_string_bytes(64)),
41    ("payload", 9),
42]) + cbor_map_bytes(&[
43    ("namespace_id", cbor_string_bytes(crate::ids::MAX_ID_BYTES)),
44    ("wal_no", 9),
45    ("next_inode_id", 9),
46    ("writer_epoch", 9),
47    ("base_head_seq", 9),
48    ("start_seq", 9),
49    ("end_seq", 9),
50    ("records", 9),
51]);
52
53// CBOR strings, collections, and u64 values need at most nine header bytes.
54const fn cbor_string_bytes(length: usize) -> usize {
55    9 + length
56}
57
58const fn cbor_map_bytes(fields: &[(&str, usize)]) -> usize {
59    let mut bytes = 9;
60    let mut index = 0;
61    while index < fields.len() {
62        bytes += cbor_string_bytes(fields[index].0.len()) + fields[index].1;
63        index += 1;
64    }
65    bytes
66}
67
68/// Identifies the durable payload family carried by a WAL envelope.
69///
70/// See [WAL segment rules](../../../docs/specs/format.md#a5-wal-records).
71#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
72#[serde(rename_all = "snake_case")]
73pub enum WalEnvelopeKind {
74    /// Marks an immutable segment in one namespace's numbered WAL.
75    NamespaceWalSegment,
76}
77
78impl WalEnvelopeKind {
79    /// Returns the frozen envelope discriminator written to durable storage.
80    pub const fn as_str(self) -> &'static str {
81        match self {
82            Self::NamespaceWalSegment => "namespace_wal_segment",
83        }
84    }
85}
86
87/// Records one replayable metadata mutation materialized from a semantic commit operation.
88///
89/// See [standard mutation operations](../../../docs/specs/format.md#66-operations-and-wal-deltas).
90#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
91#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
92pub enum WalDelta {
93    /// Introduces an inode whose identity and kind remain fixed for its lifetime.
94    CreateInode {
95        /// Stable position of this delta within its commit, used in row ordering and identity.
96        delta_index: u32,
97        /// Newly allocated durable inode identity.
98        inode_id: InodeId,
99        /// File-or-directory classification established at creation.
100        inode_kind: InodeKind,
101    },
102    /// Makes a child reachable under one canonical name in a directory.
103    BindDirentry {
104        /// Stable position of this delta within its commit, used to identify the binding.
105        delta_index: u32,
106        /// Directory receiving the new name binding.
107        parent_inode_id: InodeId,
108        /// Policy-derived lookup key on which directory uniqueness is enforced.
109        name_key: NameKey,
110        /// User-facing spelling preserved independently of `name_key`.
111        display_name: DisplayName,
112        /// Inode made reachable by the binding.
113        child_inode_id: InodeId,
114    },
115    /// Removes one exact historical directory binding without affecting a later rebind.
116    UnbindDirentry {
117        /// Stable position of this unbind within its commit.
118        delta_index: u32,
119        /// Directory from which the binding is removed.
120        parent_inode_id: InodeId,
121        /// Canonical lookup key of the binding being removed.
122        name_key: NameKey,
123        /// User-facing spelling the removed binding carried, so feed
124        /// consumers see the name a person typed without a second lookup.
125        display_name: DisplayName,
126        /// Child identity expected on the targeted binding.
127        child_inode_id: InodeId,
128        /// Commit sequence that created the exact binding being removed.
129        bind_seq: ChangeSeq,
130        /// Delta position that disambiguates the binding within `bind_seq`.
131        bind_delta_index: u32,
132    },
133    /// Publishes the next immutable content revision of a file inode.
134    AppendFileRevision {
135        /// Stable position of this revision delta within its commit.
136        delta_index: u32,
137        /// File inode receiving the revision.
138        inode_id: InodeId,
139        /// Monotonic per-file revision number validated against visible history.
140        revision_no: RevisionNo,
141        /// Immutable content that must already be durable before publication.
142        content_ref: ContentRef,
143    },
144    /// Hides a rooted subtree from snapshots at this delta's sequence or later.
145    TombstoneSubtree {
146        /// Stable position that identifies this tombstone within its commit.
147        delta_index: u32,
148        /// Inode at the root of the newly hidden subtree.
149        root_inode_id: InodeId,
150        /// The binding the delete removed, carried so the deleted name
151        /// survives on the immortal tombstone row after unbind rows age out.
152        deleted_direntry: DeletedDirentry,
153    },
154    /// Revokes exactly one subtree tombstone — the one recorded at `target`
155    /// — making the subtree eligible for visibility again once re-bound. An
156    /// immutable compensating event, not an in-place row deletion: a later
157    /// `TombstoneSubtree` for the same root supersedes the revoke.
158    RevokeSubtreeTombstone {
159        /// Stable position of this compensating delta within its commit.
160        delta_index: u32,
161        /// Root inode whose selected tombstone is being revoked.
162        root_inode_id: InodeId,
163        /// The exact tombstone generation this delta compensates.
164        target: TombstoneGeneration,
165    },
166    /// Publishes the next attribute revision of one inode, as complete state.
167    ///
168    /// The delta carries the whole resulting map rather than the changes that
169    /// produced it, so replay never needs an earlier revision to answer what
170    /// an inode holds. An empty map is a real revision: it is the cleared
171    /// state, and it hides every earlier map.
172    AppendAttributesRevision {
173        /// Stable position of this attribute delta within its commit.
174        delta_index: u32,
175        /// Inode whose attributes this revision replaces.
176        inode_id: InodeId,
177        /// Monotonic per-inode attribute revision, exactly one past the
178        /// revision the update was validated against.
179        attributes_revision_no: AttributeRevisionNo,
180        /// The inode's complete attribute map after this update.
181        attributes: Attributes,
182    },
183    /// Publishes the next access revision of one inode, as complete state.
184    ///
185    /// Like an attribute revision, the delta carries the whole resulting
186    /// state. A row with no boundary and no grants is a real revision: it is
187    /// the cleared state, and it hides every earlier row.
188    AppendAccessRevision {
189        /// Stable position of this access delta within its commit.
190        delta_index: u32,
191        /// Inode whose access state this revision replaces.
192        inode_id: InodeId,
193        /// Monotonic per-inode access revision, exactly one past the
194        /// revision the update was validated against.
195        access_revision_no: AccessRevisionNo,
196        /// Whether the directory stops inheritance after this update.
197        boundary: bool,
198        /// The inode's complete direct grants after this update.
199        grants: AccessGrants,
200    },
201}
202
203/// Associates a materialized WAL delta with the semantic operation that produced it.
204///
205/// See [logical commits](../../../docs/specs/format.md#12-commits-and-revisions).
206#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
207#[serde(deny_unknown_fields)]
208pub struct WalCommitDelta {
209    /// Zero-based request-operation position used to attribute one or more resulting deltas.
210    pub semantic_op_index: u32,
211    /// Replay mutation produced for that semantic operation.
212    pub delta: WalDelta,
213}
214
215/// Carries bytes named by a revision delta in the same commit.
216#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
217#[serde(deny_unknown_fields)]
218pub struct WalInlineContent {
219    /// Identity shared with the accompanying `blob_v1` reference.
220    pub content_id: ContentId,
221    /// Complete content encoded as a CBOR byte string.
222    #[serde(with = "serde_bytes")]
223    pub bytes: Vec<u8>,
224}
225
226/// Carries one accepted logical commit inside a WAL segment.
227///
228/// See [WAL segment rules](../../../docs/specs/format.md#a5-wal-records).
229#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
230#[serde(deny_unknown_fields)]
231pub struct WalCommitPayload {
232    /// Namespace-wide commit position; segment records must cover their range contiguously.
233    pub seq: ChangeSeq,
234    /// Caller idempotency key whose reuse must retain the same semantic fingerprint.
235    pub commit_id: CommitId,
236    /// Actor that committed the change, as supplied by the application.
237    pub committed_by: crate::ActorId,
238    /// Digest of semantic request content used to reject conflicting `commit_id` reuse.
239    pub semantic_commit_fingerprint: CommitFingerprint,
240    /// Wall-clock stamp from the publishing writer's request context, in
241    /// Unix milliseconds. Observational only: never a validity or ordering
242    /// input — `seq` is the order — and excluded from the semantic commit
243    /// fingerprint, so replay identity is untouched by clocks.
244    pub committed_at_ms: u64,
245    /// Caller-supplied annotation, omitted when absent and excluded from filesystem semantics.
246    #[serde(default, skip_serializing_if = "Option::is_none")]
247    pub message: Option<String>,
248    /// Materialized mutations in their authoritative `delta_index` order.
249    pub deltas: Vec<WalCommitDelta>,
250    /// Content values governed by [Appendix A.5](../../../docs/specs/format.md#a5-wal-records).
251    #[serde(default, skip_serializing_if = "Vec::is_empty")]
252    pub inline_content: Vec<WalInlineContent>,
253}
254
255/// Carries the namespace identity, numbered range, and commits stored in one WAL object.
256///
257/// See [WAL segment rules](../../../docs/specs/format.md#a5-wal-records).
258#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
259#[serde(deny_unknown_fields)]
260pub struct WalSegmentPayload {
261    /// Namespace this segment belongs to; recovery rejects cross-namespace content.
262    pub namespace_id: NamespaceId,
263    /// Contiguous object number checked against the key.
264    pub wal_no: WalNo,
265    /// Allocation high-water mark after this segment.
266    pub next_inode_id: InodeId,
267    /// Fencing epoch of the writer that proposed this segment.
268    pub writer_epoch: WriterEpoch,
269    /// Head sequence the writer materialized against before adding these records.
270    pub base_head_seq: ChangeSeq,
271    /// Sequence of the first record, or the unchanged head sequence for a fence.
272    pub start_seq: ChangeSeq,
273    /// Visible sequence after this segment, unchanged for a fence.
274    pub end_seq: ChangeSeq,
275    /// Logical commits in contiguous ascending sequence order.
276    pub records: Vec<WalCommitPayload>,
277}
278
279/// A WAL segment decoded through its checked durable codec.
280pub type WalSegmentEnvelope = crate::envelope::VerifiedEnvelope<WalSegmentPayload>;
281
282/// Durable layout of a WAL segment object (before zstd compression): the
283/// envelope fields plus the payload as an opaque CBOR byte string.
284/// `payload_checksum` covers exactly those bytes, so integrity verification
285/// never depends on re-encoding the payload with this build's schema. Unknown
286/// payload fields are rejected after checksum verification.
287#[derive(Serialize, Deserialize)]
288#[serde(deny_unknown_fields)]
289struct WalSegmentDocument {
290    kind: String,
291    format_version: u32,
292    payload_checksum: String,
293    #[serde(with = "serde_bytes")]
294    payload: Vec<u8>,
295}
296
297pub(crate) fn encode_wal_payload_cbor(
298    payload: &WalSegmentPayload,
299) -> Result<Vec<u8>, EnvelopeCodecError> {
300    validate_wal_inline_content(payload)?;
301    let mut encoded = Vec::new();
302    into_writer(payload, &mut encoded)
303        .map_err(|err| EnvelopeCodecError::PayloadEncode(err.to_string()))?;
304    Ok(encoded)
305}
306
307/// Encodes a WAL payload once, then checksums and compresses its durable document.
308pub fn encode_wal_segment_envelope_zstd(
309    payload: WalSegmentPayload,
310) -> Result<crate::envelope::EncodedEnvelope<WalSegmentPayload>, EnvelopeCodecError> {
311    let payload_bytes = encode_wal_payload_cbor(&payload)?;
312    let payload_checksum = sha256_digest(&payload_bytes);
313    let document = WalSegmentDocument {
314        kind: WalEnvelopeKind::NamespaceWalSegment.as_str().to_owned(),
315        format_version: WAL_FORMAT_VERSION,
316        payload_checksum: payload_checksum.clone(),
317        payload: payload_bytes,
318    };
319    let mut encoded = Vec::new();
320    into_writer(&document, &mut encoded)
321        .map_err(|err| EnvelopeCodecError::EnvelopeEncode(err.to_string()))?;
322    let bytes = zstd::stream::encode_all(encoded.as_slice(), crate::sst_blocks::ZSTD_LEVEL)
323        .map_err(|err| EnvelopeCodecError::Compress(err.to_string()))?;
324    Ok(crate::envelope::EncodedEnvelope {
325        envelope: crate::envelope::VerifiedEnvelope {
326            payload_checksum,
327            payload,
328        },
329        bytes,
330        document_len: encoded.len(),
331    })
332}
333
334/// Decodes and verifies a durable zstd-compressed WAL segment envelope.
335///
336/// Decoding fails for invalid compression or CBOR, the wrong kind or version,
337/// a checksum mismatch, or an invalid payload. See
338/// [WAL segment rules](../../../docs/specs/format.md#a5-wal-records).
339pub fn decode_wal_segment_envelope_zstd(
340    bytes: &[u8],
341) -> Result<WalSegmentEnvelope, EnvelopeCodecError> {
342    decode_wal_segment_envelope_zstd_with_limit(bytes, MAX_WAL_SEGMENT_BYTES)
343}
344
345fn decode_wal_segment_envelope_zstd_with_limit(
346    bytes: &[u8],
347    limit: usize,
348) -> Result<WalSegmentEnvelope, EnvelopeCodecError> {
349    let decoder = zstd::stream::read::Decoder::new(bytes)
350        .map_err(|err| EnvelopeCodecError::Decompress(err.to_string()))?;
351    let mut decompressed = Vec::new();
352    decoder
353        .take(limit as u64 + 1)
354        .read_to_end(&mut decompressed)
355        .map_err(|err| EnvelopeCodecError::Decompress(err.to_string()))?;
356    if decompressed.len() > limit {
357        return Err(EnvelopeCodecError::WalSegmentTooLarge { max_bytes: limit });
358    }
359    let probe: EnvelopeProbe = from_reader(decompressed.as_slice())
360        .map_err(|err| EnvelopeCodecError::EnvelopeDecode(err.to_string()))?;
361    let expected_kind = WalEnvelopeKind::NamespaceWalSegment;
362    envelope::verify_kind(expected_kind.as_str(), &probe.kind)?;
363    envelope::verify_version(&probe.kind, probe.format_version, WAL_FORMAT_VERSION)?;
364
365    let document: WalSegmentDocument = from_reader(decompressed.as_slice())
366        .map_err(|err| EnvelopeCodecError::EnvelopeDecode(err.to_string()))?;
367    envelope::verify_payload_checksum(&document.payload_checksum, &document.payload)?;
368    let payload: WalSegmentPayload = from_reader(document.payload.as_slice())
369        .map_err(|err| EnvelopeCodecError::PayloadDecode(err.to_string()))?;
370    validate_wal_inline_content(&payload)?;
371
372    Ok(WalSegmentEnvelope {
373        payload_checksum: document.payload_checksum,
374        payload,
375    })
376}
377
378fn validate_wal_inline_content(payload: &WalSegmentPayload) -> Result<(), EnvelopeCodecError> {
379    let mut total_bytes = 0;
380    for record in &payload.records {
381        if record.inline_content.is_empty() {
382            continue;
383        }
384        let mut reference_sizes: BTreeMap<&ContentId, Vec<u64>> = BTreeMap::new();
385        for delta in &record.deltas {
386            if let WalDelta::AppendFileRevision { content_ref, .. } = &delta.delta {
387                if content_ref.owner_namespace_id == payload.namespace_id {
388                    reference_sizes
389                        .entry(&content_ref.content_id)
390                        .or_default()
391                        .push(content_ref.size_bytes);
392                }
393            }
394        }
395        let mut content_ids = BTreeSet::new();
396        for entry in &record.inline_content {
397            let invalid = |reason| EnvelopeCodecError::InvalidWalInlineContent {
398                seq: record.seq,
399                content_id: entry.content_id.clone(),
400                reason,
401            };
402            if !content_ids.insert(&entry.content_id) {
403                return Err(invalid("duplicate `content_id` in commit"));
404            }
405            if entry.bytes.len() > MAX_WAL_INLINE_CONTENT_BYTES {
406                return Err(invalid("value exceeds `MAX_WAL_INLINE_CONTENT_BYTES`"));
407            }
408            let sizes = reference_sizes.get(&entry.content_id).ok_or_else(|| {
409                invalid("no `append_file_revision` reference in the same commit owned by the segment's `namespace_id`")
410            })?;
411            if !sizes.iter().all(|&size| size == entry.bytes.len() as u64) {
412                return Err(invalid("length does not match reference `size_bytes`"));
413            }
414            total_bytes += entry.bytes.len();
415            if total_bytes > MAX_WAL_SEGMENT_INLINE_CONTENT_BYTES {
416                return Err(invalid(
417                    "segment inline total exceeds `MAX_WAL_SEGMENT_INLINE_CONTENT_BYTES`",
418                ));
419            }
420        }
421    }
422    Ok(())
423}
424
425#[cfg(test)]
426mod tests {
427    #![allow(clippy::panic)]
428
429    use super::*;
430
431    fn inline_segment(lengths: &[usize]) -> WalSegmentPayload {
432        let namespace_id = NamespaceId::parse("bounded").expect("namespace");
433        let records = lengths
434            .iter()
435            .enumerate()
436            .map(|(index, &length)| {
437                let bytes = vec![42; length];
438                let content_id =
439                    ContentId::parse("con_0123456789abcdef0123456789abcdef").expect("content id");
440                WalCommitPayload {
441                    seq: ChangeSeq(index as u64 + 1),
442                    commit_id: CommitId::parse(format!("c_{index:032x}")).expect("commit id"),
443                    committed_by: crate::ActorId::parse("test").expect("actor"),
444                    semantic_commit_fingerprint: serde_json::from_str(r#""v1:sha256:test""#)
445                        .expect("fingerprint"),
446                    committed_at_ms: 0,
447                    message: None,
448                    deltas: vec![WalCommitDelta {
449                        semantic_op_index: 0,
450                        delta: WalDelta::AppendFileRevision {
451                            delta_index: 0,
452                            inode_id: InodeId(2),
453                            revision_no: RevisionNo(index as u64 + 1),
454                            content_ref: ContentRef::blob_v1(
455                                namespace_id.clone(),
456                                content_id.clone(),
457                                &bytes,
458                            ),
459                        },
460                    }],
461                    inline_content: vec![WalInlineContent { content_id, bytes }],
462                }
463            })
464            .collect();
465        WalSegmentPayload {
466            namespace_id,
467            wal_no: WalNo(1),
468            next_inode_id: InodeId(3),
469            writer_epoch: WriterEpoch(1),
470            base_head_seq: ChangeSeq(0),
471            start_seq: ChangeSeq(1),
472            end_seq: ChangeSeq(lengths.len() as u64),
473            records,
474        }
475    }
476
477    fn unchecked_segment_bytes(payload: &WalSegmentPayload) -> Vec<u8> {
478        let mut payload_bytes = Vec::new();
479        into_writer(payload, &mut payload_bytes).expect("encode payload directly");
480        let document = WalSegmentDocument {
481            kind: WalEnvelopeKind::NamespaceWalSegment.as_str().to_owned(),
482            format_version: WAL_FORMAT_VERSION,
483            payload_checksum: sha256_digest(&payload_bytes),
484            payload: payload_bytes,
485        };
486        let mut document_bytes = Vec::new();
487        into_writer(&document, &mut document_bytes).expect("encode document directly");
488        zstd::stream::encode_all(document_bytes.as_slice(), 0).expect("compress document")
489    }
490
491    fn assert_inline_content_rejected(
492        payload: WalSegmentPayload,
493        record_index: usize,
494        expected_reason: &str,
495    ) {
496        let expected_seq = payload.records[record_index].seq;
497        let expected_content_id = payload.records[record_index].inline_content[0]
498            .content_id
499            .clone();
500        let decoded_error = decode_wal_segment_envelope_zstd(&unchecked_segment_bytes(&payload))
501            .expect_err("invalid inline content should not decode");
502        let encoded_error = encode_wal_segment_envelope_zstd(payload)
503            .expect_err("invalid inline content should not encode");
504        for error in [decoded_error, encoded_error] {
505            assert_eq!(
506                error.to_string(),
507                format!(
508                    "invalid wal inline content in commit `{expected_seq}` for `content_id` `{expected_content_id}`: {expected_reason}"
509                ),
510            );
511            match error {
512                EnvelopeCodecError::InvalidWalInlineContent {
513                    seq,
514                    content_id,
515                    reason,
516                } => {
517                    assert_eq!(seq, expected_seq);
518                    assert_eq!(content_id, expected_content_id);
519                    assert_eq!(reason, expected_reason);
520                }
521                other => panic!("expected invalid inline content, got {other:?}"),
522            }
523        }
524    }
525
526    fn assert_inline_content_accepted(payload: WalSegmentPayload) {
527        let encoded = encode_wal_segment_envelope_zstd(payload.clone()).expect("encode segment");
528        let decoded = decode_wal_segment_envelope_zstd(encoded.as_bytes()).expect("decode segment");
529        assert_eq!(decoded.into_payload(), payload);
530    }
531
532    #[test]
533    fn inline_content_requires_a_local_revision_reference_in_the_same_commit() {
534        let expected_reason = "no `append_file_revision` reference in the same commit owned by the segment's `namespace_id`";
535        let mut missing = inline_segment(&[3, 3]);
536        missing.records[0].deltas.clear();
537        assert_inline_content_rejected(missing, 0, expected_reason);
538
539        let mut wrong_id = inline_segment(&[3]);
540        wrong_id.records[0].inline_content[0].content_id =
541            ContentId::parse("con_fedcba9876543210fedcba9876543210").expect("content id");
542        assert_inline_content_rejected(wrong_id, 0, expected_reason);
543
544        let mut foreign = inline_segment(&[3]);
545        foreign.namespace_id = NamespaceId::parse("other").expect("namespace");
546        assert_inline_content_rejected(foreign, 0, expected_reason);
547    }
548
549    #[test]
550    fn inline_content_length_must_match_the_reference() {
551        let mut payload = inline_segment(&[3]);
552        payload.records[0].inline_content[0].bytes.push(42);
553        assert_inline_content_rejected(payload, 0, "length does not match reference `size_bytes`");
554
555        let mut payload = inline_segment(&[3]);
556        let mut other_reference = payload.records[0].deltas[0].clone();
557        other_reference.semantic_op_index = 1;
558        match &mut other_reference.delta {
559            WalDelta::AppendFileRevision {
560                delta_index,
561                revision_no,
562                content_ref,
563                ..
564            } => {
565                *delta_index = 1;
566                *revision_no = RevisionNo(2);
567                content_ref.size_bytes += 1;
568            }
569            other => panic!("expected file revision, got {other:?}"),
570        }
571        payload.records[0].deltas.push(other_reference);
572        assert_inline_content_rejected(payload, 0, "length does not match reference `size_bytes`");
573    }
574
575    #[test]
576    fn inline_content_ids_must_be_unique_within_each_commit() {
577        let mut payload = inline_segment(&[0]);
578        let entry = payload.records[0].inline_content[0].clone();
579        payload.records[0].inline_content.push(entry);
580        assert_inline_content_rejected(payload, 0, "duplicate `content_id` in commit");
581    }
582
583    #[test]
584    fn inline_content_accepts_the_value_limit_and_rejects_one_byte_more() {
585        assert_eq!(MAX_WAL_INLINE_CONTENT_BYTES, 262_144);
586        assert_inline_content_accepted(inline_segment(&[MAX_WAL_INLINE_CONTENT_BYTES]));
587        assert_inline_content_rejected(
588            inline_segment(&[MAX_WAL_INLINE_CONTENT_BYTES + 1]),
589            0,
590            "value exceeds `MAX_WAL_INLINE_CONTENT_BYTES`",
591        );
592    }
593
594    #[test]
595    fn inline_content_accepts_the_segment_limit_and_rejects_one_byte_more() {
596        assert_eq!(MAX_WAL_SEGMENT_INLINE_CONTENT_BYTES, 4_194_304);
597        let mut lengths = vec![
598            MAX_WAL_INLINE_CONTENT_BYTES;
599            MAX_WAL_SEGMENT_INLINE_CONTENT_BYTES / MAX_WAL_INLINE_CONTENT_BYTES
600        ];
601        assert_inline_content_accepted(inline_segment(&lengths));
602        lengths.push(1);
603        assert_inline_content_rejected(
604            inline_segment(&lengths),
605            lengths.len() - 1,
606            "segment inline total exceeds `MAX_WAL_SEGMENT_INLINE_CONTENT_BYTES`",
607        );
608    }
609
610    #[test]
611    fn inline_content_is_not_hashed_against_the_reference_checksum() {
612        let mut payload = inline_segment(&[3]);
613        payload.records[0].inline_content[0].bytes[0] = 43;
614        assert_inline_content_accepted(payload);
615    }
616
617    #[test]
618    fn decoder_accepts_the_limit_and_rejects_the_next_byte_before_decoding() {
619        let encoded = encode_wal_segment_envelope_zstd(WalSegmentPayload {
620            namespace_id: NamespaceId::parse("bounded").expect("namespace"),
621            wal_no: WalNo(1),
622            next_inode_id: InodeId(2),
623            writer_epoch: WriterEpoch(1),
624            base_head_seq: ChangeSeq(0),
625            start_seq: ChangeSeq(0),
626            end_seq: ChangeSeq(0),
627            records: Vec::new(),
628        })
629        .expect("encode");
630        let document = zstd::stream::decode_all(encoded.as_bytes()).expect("decompress");
631        assert_eq!(document.len(), encoded.document_len());
632        assert!(document.len() <= WAL_SEGMENT_OVERHEAD_BYTES);
633        assert_eq!(
634            &decode_wal_segment_envelope_zstd_with_limit(encoded.as_bytes(), document.len())
635                .expect("at limit"),
636            encoded.envelope(),
637        );
638        assert!(matches!(
639            decode_wal_segment_envelope_zstd_with_limit(encoded.as_bytes(), document.len() - 1),
640            Err(EnvelopeCodecError::WalSegmentTooLarge { max_bytes }) if max_bytes == document.len() - 1
641        ));
642        let invalid = zstd::stream::encode_all(&[0xff; 64][..], 0).expect("compress");
643        assert!(matches!(
644            decode_wal_segment_envelope_zstd_with_limit(&invalid, 8),
645            Err(EnvelopeCodecError::WalSegmentTooLarge { max_bytes: 8 })
646        ));
647        assert!(matches!(
648            decode_wal_segment_envelope_zstd_with_limit(&invalid, 64),
649            Err(EnvelopeCodecError::EnvelopeDecode(_))
650        ));
651    }
652}
loonfs_api/wal.rs

loonfs_api/
wal.rs
