feat: add shared WAL record envelopes and payload validation

.github/workflows/rust-test.yml

@@ -63,8 +63,8 @@ jobs:
         run: |
           cargo run -p lance-context-wal --features lab --bin context-wal-lab -- codec --payload-bytes 1024 --rows 8 --iterations 2
           cargo run -p lance-context-wal --features lab --bin context-wal-lab -- disk --directory "$RUNNER_TEMP/wal-disk-smoke" --payload-bytes 1024 --batch 4 --writers 2 --groups-per-writer 2 --rounds 8
-      - name: WAL placement, admission and IPC decoding tests
-        run: cargo test -p lance-context-wal --features placement,admission,ipc --lib
+      - name: WAL placement, admission, record and IPC decoding tests
+        run: cargo test -p lance-context-wal --features placement,admission,ipc,record --lib
       - name: Start etcd for HA scheduler test
         run: |
           ETCD_VERSION=3.7.0

.github/workflows/style.yml

@@ -56,7 +56,7 @@ jobs:
       - name: Clippy (lance-context, default features)
         run: cargo clippy --manifest-path crates/lance-context/Cargo.toml --all-targets -- -D warnings
       - name: Clippy (WAL experiments)
-        run: cargo clippy -p lance-context-wal --features lab,placement,admission,ipc --all-targets -- -D warnings
+        run: cargo clippy -p lance-context-wal --features lab,placement,admission,ipc,record --all-targets -- -D warnings
 
   typos:
     name: Spell Check

crates/lance-context-wal/Cargo.toml

@@ -7,6 +7,7 @@ publish = false
 description = "Independent Rust WAL experiments owned by lance-context"
 
 [features]
+record = ["dep:bytes", "bytes/serde", "dep:serde", "dep:sha2", "dep:uuid"]
 placement = ["dep:anyhow", "dep:serde", "dep:serde_json", "dep:sha2", "dep:uuid"]
 admission = ["dep:anyhow", "dep:serde"]
 ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "dep:arrow-schema", "dep:bytes"]
@@ -29,6 +30,7 @@ sha2 = { version = "0.10", optional = true }
 uuid = { version = "1", features = ["serde"], optional = true }
 
 [dev-dependencies]
+bincode = "1.3"
 arrow-ipc = { version = "58", features = ["zstd"] }
 tempfile = "3"
 

crates/lance-context-wal/README.md

@@ -3,6 +3,27 @@
 Feature-gated Rust measurements for choosing a WAL payload path and durable
 local log engine. This crate does not change production ingestion or Lance OSS.
 
+Implementation and validation are tracked in
+[#378](https://github.com/lance-format/lance-context/issues/378). Related PRs should
+include `Refs #378`; individual steps should leave the tracking issue open.
+
+The optional `record` feature provides opaque `Batch`/`BatchMeta` envelopes with
+shared `Bytes` payload ownership. `BatchMeta::check_body_length` validates frame
+lengths and a caller-supplied data/control limit before payload allocation;
+`Batch::validate` also checks the complete SHA-256 digest. Construction and cloning
+retain the input allocation. Digest computation scans the payload; this is not
+end-to-end zero-copy. Replication can use this feature without Arrow or a log engine.
+
+Routing, producer sequencing/fencing, control commands and durable commit still
+require separate validation. The record domain tags reserve the single-group,
+multi-group and source-control formats used by the ingestion experiments. This
+PR adds the envelope, not their state machines. Serialize metadata separately
+from bulk bytes in network paths; `Debug` excludes the body.
+
+```bash
+cargo test -p lance-context-wal --features record --lib
+```
+
 The optional `ipc` feature exposes `ipc::decode(Bytes, expected_schema)` for
 materializers. It decodes multiple uncompressed record batches and dictionaries
 while sharing binary payload ownership with the input. Numeric alignment repair,

crates/lance-context-wal/src/lib.rs

@@ -13,3 +13,6 @@ pub mod admission;
 
 #[cfg(feature = "ipc")]
 pub mod ipc;
+
+#[cfg(feature = "record")]
+pub mod record;

crates/lance-context-wal/src/record.rs

@@ -0,0 +1,148 @@
+//! Opaque WAL records with shared payload ownership and bounded validation.
+//!
+//! Serialize `BatchMeta` separately from bulk payloads on the network. Cloning
+//! `Batch` shares the input `Bytes`; construction and validation each scan the
+//! payload for its SHA-256 digest. No Arrow decoding is needed here.
+//! A valid envelope establishes neither routing authorization nor a durable ACK.
+
+use bytes::Bytes;
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BatchMeta {
+    #[serde(default)]
+    pub kind: RecordKind,
+    /// Stable qualified table identity, distinct from an RPC connection/session.
+    pub stream: Uuid,
+    pub bucket: u32,
+    pub producer: Uuid,
+    pub sequence: u64,
+    pub digest: [u8; 32],
+    pub body_bytes: u64,
+}
+
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Batch {
+    pub meta: BatchMeta,
+    pub body: Bytes,
+}
+
+/// Record domains used by the ingestion experiments. Control payloads have
+/// their own command validation and smaller limits at the replication layer.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub enum RecordKind {
+    #[default]
+    Data,
+    /// Legacy single-group publication commands.
+    Publication,
+    /// Table-level publication plans and outcomes across source groups.
+    MultiPublication,
+    /// Per-group source pins, release and retention commands.
+    Source,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ValidationError {
+    EmptyBody,
+    ZeroSequence,
+    BodyTooLarge { bytes: usize, limit: usize },
+    LengthMismatch { declared: u64, actual: usize },
+    DigestMismatch,
+}
+
+impl std::fmt::Display for ValidationError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::EmptyBody => f.write_str("empty WAL record body"),
+            Self::ZeroSequence => f.write_str("WAL record sequence must be positive"),
+            Self::BodyTooLarge { bytes, limit } => {
+                write!(
+                    f,
+                    "WAL record body has {bytes} bytes, exceeding limit {limit}"
+                )
+            }
+            Self::LengthMismatch { declared, actual } => {
+                write!(f, "WAL record declares {declared} bytes, received {actual}")
+            }
+            Self::DigestMismatch => f.write_str("WAL record payload digest mismatch"),
+        }
+    }
+}
+impl std::error::Error for ValidationError {}
+
+impl BatchMeta {
+    /// Validate the declared frame length before reserving/reading its body.
+    /// The caller supplies the applicable data/control limit. This does not
+    /// validate table ownership, placement, producer fencing or command contents.
+    pub fn check_body_length(&self, actual: usize, limit: usize) -> Result<(), ValidationError> {
+        if actual > limit {
+            return Err(ValidationError::BodyTooLarge {
+                bytes: actual,
+                limit,
+            });
+        }
+        if actual == 0 {
+            return Err(ValidationError::EmptyBody);
+        }
+        if self.body_bytes != actual as u64 {
+            return Err(ValidationError::LengthMismatch {
+                declared: self.body_bytes,
+                actual,
+            });
+        }
+        if self.sequence == 0 {
+            return Err(ValidationError::ZeroSequence);
+        }
+        Ok(())
+    }
+
+    /// Validate separately framed payload bytes without constructing or copying
+    /// an envelope. Length/admission checks run before the digest scan.
+    pub fn validate_payload(&self, body: &[u8], limit: usize) -> Result<(), ValidationError> {
+        self.check_body_length(body.len(), limit)?;
+        if self.digest != <[u8; 32]>::from(Sha256::digest(body)) {
+            return Err(ValidationError::DigestMismatch);
+        }
+        Ok(())
+    }
+}
+
+impl Batch {
+    /// Retain the caller's allocation and compute its digest once. Routing and
+    /// sequence admission still belong to the caller/replicated state machine.
+    pub fn new(stream: Uuid, bucket: u32, producer: Uuid, sequence: u64, body: Bytes) -> Self {
+        let digest = Sha256::digest(&body).into();
+        let body_bytes = body.len() as u64;
+        Self {
+            meta: BatchMeta {
+                kind: RecordKind::Data,
+                stream,
+                bucket,
+                producer,
+                sequence,
+                digest,
+                body_bytes,
+            },
+            body,
+        }
+    }
+
+    /// Check bounds and the complete opaque payload before admitting an append.
+    pub fn validate(&self, limit: usize) -> Result<(), ValidationError> {
+        self.meta.validate_payload(&self.body, limit)
+    }
+}
+
+impl std::fmt::Debug for Batch {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Batch")
+            .field("meta", &self.meta)
+            .field("bytes", &self.body.len())
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/lance-context-wal/src/record/tests.rs

@@ -0,0 +1,111 @@
+use super::*;
+
+fn record(body: Bytes) -> Batch {
+    Batch::new(
+        Uuid::from_bytes([0x11; 16]),
+        17,
+        Uuid::from_bytes([0x22; 16]),
+        7,
+        body,
+    )
+}
+
+#[test]
+fn payload_owners_share_slices_and_outlive_the_original() {
+    let body = Bytes::from(vec![37; 1024 * 1024]);
+    let batch = record(body.slice(3..body.len() - 5));
+    let copy = batch.clone();
+    assert_eq!(batch.body.as_ptr(), body[3..].as_ptr());
+    assert_eq!(copy.body.as_ptr(), batch.body.as_ptr());
+    drop(batch);
+    drop(body);
+    copy.validate(1024 * 1024).unwrap();
+    assert!(copy.body.iter().all(|byte| *byte == 37));
+}
+
+#[test]
+fn length_limits_are_checked_before_payload_validation() {
+    let mut meta = record(Bytes::from_static(b"abc")).meta;
+    assert!(meta.check_body_length(3, 3).is_ok());
+    assert_eq!(
+        meta.check_body_length(3, 2),
+        Err(ValidationError::BodyTooLarge { bytes: 3, limit: 2 })
+    );
+    assert_eq!(
+        meta.check_body_length(0, 3),
+        Err(ValidationError::EmptyBody)
+    );
+    meta.body_bytes = u64::MAX;
+    assert_eq!(
+        meta.check_body_length(3, 3),
+        Err(ValidationError::LengthMismatch {
+            declared: u64::MAX,
+            actual: 3
+        })
+    );
+    meta.body_bytes = 3;
+    meta.sequence = 0;
+    assert_eq!(
+        meta.check_body_length(3, 3),
+        Err(ValidationError::ZeroSequence)
+    );
+}
+
+#[test]
+fn changed_payloads_and_truncation_are_rejected_without_decoding() {
+    let expected = record(Bytes::from_static(b"opaque payload"));
+    let mut changed = expected.clone();
+    changed.body = Bytes::from_static(b"opaque payloaD");
+    assert_eq!(changed.validate(64), Err(ValidationError::DigestMismatch));
+    changed.body = expected.body.slice(..expected.body.len() - 1);
+    assert!(matches!(
+        changed.validate(64),
+        Err(ValidationError::LengthMismatch { .. })
+    ));
+    changed.body = expected.body;
+    changed.meta.digest[0] ^= 1;
+    assert_eq!(changed.validate(64), Err(ValidationError::DigestMismatch));
+}
+
+#[test]
+fn record_domains_keep_the_metadata_wire_layout() {
+    // Fixed-int bincode 1: domain u32, length-prefixed UUID byte strings, bucket
+    // u32, sequence u64, digest bytes and body length u64, all integers LE.
+    // SHA-256("abc") is the standard published test vector.
+    let suffix = concat!(
+        "100000000000000011111111111111111111111111111111",
+        "11000000",
+        "100000000000000022222222222222222222222222222222",
+        "0700000000000000",
+        "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
+        "0300000000000000",
+    );
+    for (tag, kind) in [
+        RecordKind::Data,
+        RecordKind::Publication,
+        RecordKind::MultiPublication,
+        RecordKind::Source,
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let mut meta = record(Bytes::from_static(b"abc")).meta;
+        meta.kind = kind;
+        let expected = format!("{tag:02x}000000{suffix}");
+        let encoded = bincode::serialize(&meta).unwrap();
+        let hex: String = encoded.iter().map(|byte| format!("{byte:02x}")).collect();
+        assert_eq!(hex, expected);
+        assert_eq!(bincode::deserialize::<BatchMeta>(&encoded).unwrap(), meta);
+    }
+}
+
+#[test]
+fn control_envelopes_allow_their_reserved_producer_and_hide_payload_in_debug() {
+    let mut batch = record(Bytes::from_static(b"private command body"));
+    batch.meta.kind = RecordKind::Source;
+    batch.meta.producer = Uuid::nil();
+    batch.validate(64).unwrap();
+    assert!(!format!("{batch:?}").contains("private command body"));
+    let copy: Batch = bincode::deserialize(&bincode::serialize(&batch).unwrap()).unwrap();
+    assert_eq!(copy, batch);
+}