From 0f262a7b5b7c366f9b94cbaa6ce2dc0e7771a28b Mon Sep 17 00:00:00 2001 From: Beinan Date: Mon, 19 Oct 2026 06:11:55 +0000 Subject: [PATCH] feat: add shared WAL record envelopes and payload validation --- .github/workflows/rust-test.yml | 4 +- .github/workflows/style.yml | 2 +- Cargo.lock | 13 ++ crates/lance-context-wal/Cargo.toml | 2 + crates/lance-context-wal/README.md | 21 +++ crates/lance-context-wal/src/lib.rs | 3 + crates/lance-context-wal/src/record.rs | 148 +++++++++++++++++++ crates/lance-context-wal/src/record/tests.rs | 111 ++++++++++++++ 8 files changed, 301 insertions(+), 3 deletions(-) create mode 100644 crates/lance-context-wal/src/record.rs create mode 100644 crates/lance-context-wal/src/record/tests.rs diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 88ae8770..25448466 100644 --- a/.github/workflows/rust-test.yml +++ b/.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 diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml index 70b94a0f..5f69a735 100644 --- a/.github/workflows/style.yml +++ b/.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 diff --git a/Cargo.lock b/Cargo.lock index 62dba986..f497b537 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -955,6 +955,15 @@ dependencies = [ "num-traits", ] +[[package]] +name = "bincode" +version = "1.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad" +dependencies = [ + "serde", +] + [[package]] name = "bindgen" version = "0.72.1" @@ -1104,6 +1113,9 @@ name = "bytes" version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +dependencies = [ + "serde", +] [[package]] name = "bytes-utils" @@ -4176,6 +4188,7 @@ dependencies = [ "arrow-buffer", "arrow-ipc", "arrow-schema", + "bincode", "bytes", "clap", "hdrhistogram", diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 340fa6c7..7c7fb18a 100644 --- a/crates/lance-context-wal/Cargo.toml +++ b/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" diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index c6621cd4..e7c27a21 100644 --- a/crates/lance-context-wal/README.md +++ b/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, diff --git a/crates/lance-context-wal/src/lib.rs b/crates/lance-context-wal/src/lib.rs index 7ea0eee9..fe590340 100644 --- a/crates/lance-context-wal/src/lib.rs +++ b/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; diff --git a/crates/lance-context-wal/src/record.rs b/crates/lance-context-wal/src/record.rs new file mode 100644 index 00000000..eeaa9f61 --- /dev/null +++ b/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; diff --git a/crates/lance-context-wal/src/record/tests.rs b/crates/lance-context-wal/src/record/tests.rs new file mode 100644 index 00000000..fd71231b --- /dev/null +++ b/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::(&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); +}