From 972234dc93c2024ba4d3e902026ca4eb39c31db1 Mon Sep 17 00:00:00 2001 From: Beinan Date: Mon, 19 Oct 2026 08:32:13 +0000 Subject: [PATCH] feat: define immutable WAL ingestion contracts --- .github/workflows/rust-test.yml | 4 +- .github/workflows/style.yml | 2 +- crates/lance-context-wal/Cargo.toml | 1 + crates/lance-context-wal/README.md | 13 +++ .../lance-context-wal/src/ingest_contract.rs | 92 +++++++++++++++++++ .../src/ingest_contract/tests.rs | 88 ++++++++++++++++++ crates/lance-context-wal/src/lib.rs | 3 + 7 files changed, 200 insertions(+), 3 deletions(-) create mode 100644 crates/lance-context-wal/src/ingest_contract.rs create mode 100644 crates/lance-context-wal/src/ingest_contract/tests.rs diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 2544846..bc87151 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, record and IPC decoding tests - run: cargo test -p lance-context-wal --features placement,admission,ipc,record --lib + - name: WAL placement, admission, record, contract and IPC decoding tests + run: cargo test -p lance-context-wal --features placement,admission,ipc,record,contract --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 5f69a73..7059843 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,record --all-targets -- -D warnings + run: cargo clippy -p lance-context-wal --features lab,placement,admission,ipc,record,contract --all-targets -- -D warnings typos: name: Spell Check diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 7c7fb18..33845e3 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] +contract = ["dep:serde", "dep:sha2"] 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"] diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index e7c27a2..78dd940 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -7,6 +7,19 @@ 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 independent `contract` feature defines ingestion routing keys, the schema +fingerprint, immutable-ID policy and record byte/row limits. Validation checks +the representation after decoding; callers must durably bind and compare the +complete contract for each table generation. The ingress still validates actual +Arrow input, and the materializer must support the registered maximum record. +This component does not enforce persistence, authentication or ID uniqueness. +Schema hashing accepts already canonical bytes; it does not normalize JSON or +parse Arrow. The module requires neither Arrow nor a replication/log engine. + +```bash +cargo test -p lance-context-wal --features contract --lib +``` + 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; diff --git a/crates/lance-context-wal/src/ingest_contract.rs b/crates/lance-context-wal/src/ingest_contract.rs new file mode 100644 index 0000000..6ff7e32 --- /dev/null +++ b/crates/lance-context-wal/src/ingest_contract.rs @@ -0,0 +1,92 @@ +//! Immutable ingestion metadata shared by a table's ingress and publisher. +//! +//! These types validate the representation, not persistence or enforcement. +//! Callers must compare the complete contract when binding a generation and +//! validate actual Arrow input before ACK. This is not authentication, producer +//! deduplication, cross-producer ID uniqueness or a table migration protocol. +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +/// Protocol ceiling. Deployments may set a smaller maximum in each contract; +/// ingress and materializers must both have capacity for that maximum. +pub const MAX_RECORD_BYTES: u32 = 255 * 1024 * 1024; +pub const MAX_ROWS: u32 = 1_048_576; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum WritePolicy { + /// Caller owns unique immutable row IDs; producer dedup is a separate layer. + ImmutableUniqueIds, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct IngestContract { + /// Empty is permitted for a deployment without tenant routing. Otherwise + /// callers supply a canonical tenant key, not an untrusted upload field. + pub tenant: String, + pub qualified_table: String, + /// Fingerprint of a canonical Arrow schema document, including metadata. + pub schema: [u8; 32], + pub policy: WritePolicy, + pub max_record_bytes: u32, + pub max_rows: u32, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ValidationError { + TenantTooLong, + InvalidTable, + MissingSchema, + InvalidRecordLimit, + InvalidRowLimit, +} + +impl std::fmt::Display for ValidationError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::TenantTooLong => "ingestion tenant exceeds 1024 UTF-8 bytes", + Self::InvalidTable => "ingestion table must contain 1..=1024 UTF-8 bytes", + Self::MissingSchema => "ingestion schema fingerprint must be nonzero", + Self::InvalidRecordLimit => "ingestion record limit exceeds protocol bounds", + Self::InvalidRowLimit => "ingestion row limit exceeds protocol bounds", + }) + } +} +impl std::error::Error for ValidationError {} + +impl IngestContract { + /// Validate after decoding and before persisting or comparing a contract. + /// Serde decoding itself does not establish validity. Byte limits do not + /// reserve memory, and a schema fingerprint cannot validate an Arrow body. + pub fn validate(&self) -> Result<(), ValidationError> { + if self.tenant.len() > 1024 { + return Err(ValidationError::TenantTooLong); + } + if self.qualified_table.is_empty() || self.qualified_table.len() > 1024 { + return Err(ValidationError::InvalidTable); + } + if self.schema == [0; 32] { + return Err(ValidationError::MissingSchema); + } + if !(1..=MAX_RECORD_BYTES).contains(&self.max_record_bytes) { + return Err(ValidationError::InvalidRecordLimit); + } + if !(1..=MAX_ROWS).contains(&self.max_rows) { + return Err(ValidationError::InvalidRowLimit); + } + Ok(()) + } +} + +/// Hash a caller-canonicalized schema document. This hashes bytes as provided: +/// object-key order/metadata normalization belong to the caller. Replicas can +/// retain this identity without Arrow dependencies or decoding payloads. +pub fn schema_fingerprint(canonical_document: &[u8]) -> [u8; 32] { + let mut digest = Sha256::new(); + digest.update(b"lance-context/ingest-arrow-schema/v1"); + digest.update((canonical_document.len() as u64).to_le_bytes()); + digest.update(canonical_document); + digest.finalize().into() +} + +#[cfg(test)] +mod tests; diff --git a/crates/lance-context-wal/src/ingest_contract/tests.rs b/crates/lance-context-wal/src/ingest_contract/tests.rs new file mode 100644 index 0000000..68b5ff9 --- /dev/null +++ b/crates/lance-context-wal/src/ingest_contract/tests.rs @@ -0,0 +1,88 @@ +use super::*; + +fn contract() -> IngestContract { + IngestContract { + tenant: "t".into(), + qualified_table: "q".into(), + schema: [1; 32], + policy: WritePolicy::ImmutableUniqueIds, + max_record_bytes: 1024, + max_rows: 1, + } +} + +#[test] +fn binary_contract_fixture_preserves_field_order_policy_and_limits() { + // Independent bincode v1 fixed-width layout: u64 UTF-8 byte lengths, schema + // bytes, u32 enum discriminant, u32 byte limit and u32 row limit. + let fixture = [ + 1, 0, 0, 0, 0, 0, 0, 0, b't', 1, 0, 0, 0, 0, 0, 0, 0, b'q', 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, + 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 4, 0, 0, + 1, 0, 0, 0, + ]; + assert_eq!(bincode::serialize(&contract()).unwrap(), fixture); + assert_eq!( + bincode::deserialize::(&fixture).unwrap(), + contract() + ); + let mut foreign_policy = fixture; + foreign_policy[50] = 1; + assert!(bincode::deserialize::(&foreign_policy).is_err()); + let mut invalid_limit = fixture; + invalid_limit[54..58].fill(0); + assert_eq!( + bincode::deserialize::(&invalid_limit) + .unwrap() + .validate(), + Err(ValidationError::InvalidRecordLimit) + ); +} + +#[test] +fn schema_identity_has_an_independent_domain_and_length_fixture() { + // SHA-256 of the documented domain, 8-byte LE length 2, and "{}", checked + // independently with OpenSSL. This does not imply JSON canonicalization. + assert_eq!( + schema_fingerprint(b"{}"), + [ + 5, 36, 127, 73, 165, 116, 54, 235, 119, 118, 209, 98, 217, 13, 2, 25, 226, 70, 105, + 188, 240, 227, 113, 53, 118, 28, 0, 33, 28, 126, 136, 241, + ] + ); + assert_ne!(schema_fingerprint(b"{}"), schema_fingerprint(b"{ }")); + assert_ne!( + schema_fingerprint(b"{}"), + <[u8; 32]>::from(Sha256::digest(b"{}")) + ); +} + +#[test] +fn validation_uses_utf8_byte_limits_and_rejects_invalid_decoded_metadata() { + let mut value = contract(); + value.tenant = "é".repeat(512); + value.qualified_table = "表".repeat(341) + "x"; + value.max_record_bytes = MAX_RECORD_BYTES; + value.max_rows = MAX_ROWS; + value.validate().unwrap(); + value.tenant.push('x'); + assert_eq!(value.validate(), Err(ValidationError::TenantTooLong)); + value.tenant.clear(); + value.validate().unwrap(); + value.qualified_table.push('x'); + assert_eq!(value.validate(), Err(ValidationError::InvalidTable)); + value.qualified_table.clear(); + assert_eq!(value.validate(), Err(ValidationError::InvalidTable)); + value = contract(); + value.schema = [0; 32]; + assert_eq!(value.validate(), Err(ValidationError::MissingSchema)); + value = contract(); + for bytes in [0, MAX_RECORD_BYTES + 1] { + value.max_record_bytes = bytes; + assert_eq!(value.validate(), Err(ValidationError::InvalidRecordLimit)); + } + value = contract(); + for rows in [0, MAX_ROWS + 1] { + value.max_rows = rows; + assert_eq!(value.validate(), Err(ValidationError::InvalidRowLimit)); + } +} diff --git a/crates/lance-context-wal/src/lib.rs b/crates/lance-context-wal/src/lib.rs index fe59034..04bf18d 100644 --- a/crates/lance-context-wal/src/lib.rs +++ b/crates/lance-context-wal/src/lib.rs @@ -16,3 +16,6 @@ pub mod ipc; #[cfg(feature = "record")] pub mod record; + +#[cfg(feature = "contract")] +pub mod ingest_contract;