feat: define immutable WAL ingestion contracts

.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

.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

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"]

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;

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;

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::<IngestContract>(&fixture).unwrap(),
+        contract()
+    );
+    let mut foreign_policy = fixture;
+    foreign_policy[50] = 1;
+    assert!(bincode::deserialize::<IngestContract>(&foreign_policy).is_err());
+    let mut invalid_limit = fixture;
+    invalid_limit[54..58].fill(0);
+    assert_eq!(
+        bincode::deserialize::<IngestContract>(&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));
+    }
+}

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;