feat: add stable WAL session placement and receipt scopes

.github/workflows/rust-test.yml

@@ -63,6 +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 compatibility tests
+        run: cargo test -p lance-context-wal --features placement --lib
       - name: Start etcd for HA scheduler test
         run: |
           ETCD_VERSION=3.7.0

.github/workflows/style.yml

@@ -55,8 +55,8 @@ jobs:
       # cfg-gated breakage is caught here instead of at release time.
       - name: Clippy (lance-context, default features)
         run: cargo clippy --manifest-path crates/lance-context/Cargo.toml --all-targets -- -D warnings
-      - name: Clippy (WAL component experiments)
-        run: cargo clippy -p lance-context-wal --features lab --all-targets -- -D warnings
+      - name: Clippy (WAL experiments)
+        run: cargo clippy -p lance-context-wal --features lab,placement --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]
+placement = ["dep:anyhow", "dep:serde", "dep:serde_json", "dep:sha2", "dep:uuid"]
 lab = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "dep:arrow-schema", "dep:bytes", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
@@ -21,6 +22,9 @@ hdrhistogram = { version = "7", optional = true }
 raft-engine = { version = "=0.4.2", default-features = false, optional = true }
 raft-proto = { version = "=0.7.0", optional = true }
 serde_json = { version = "1", optional = true }
+serde = { version = "1", features = ["derive"], optional = true }
+sha2 = { version = "0.10", optional = true }
+uuid = { version = "1", features = ["serde"], optional = true }
 
 [[bin]]
 name = "context-wal-lab"

crates/lance-context-wal/README.md

@@ -3,6 +3,22 @@
 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.
 
+The independent `placement` feature provides `placement::Layout`, session bucket
+hashing and scoped `Receipt` types. A validated layout assigns each stable virtual
+bucket to one RF3 group, with explicit voters and StatefulSet DNS addresses. Use a
+canonical immutable tenant/table identity in the routing key, not a mutable display
+name. Worker count does not participate in routing. Persist the layout and its
+fingerprint; changing it requires a migration protocol, which is not implemented
+here. A receipt identifies a position and is not authentication or proof of commit.
+
+```bash
+cargo test -p lance-context-wal --features placement --lib
+```
+
+Compatibility tests preserve a layout fingerprint recorded by a real three-node
+experiment and a fixed session-hash vector. Other tests reject ambiguous ownership
+and ensure routes survive JSON serialization and restart.
+
 ```bash
 cargo build --release -p lance-context-wal --features lab --bin context-wal-lab
 target/release/context-wal-lab codec --payload-bytes 262144 --rows 8 --iterations 30

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

@@ -3,3 +3,7 @@
 //! The optional `lab` feature builds Arrow IPC and raft-engine component
 //! benchmarks. It is not connected to production ingestion and makes no
 //! distributed durability or end-to-end Lance throughput claim.
+//! `placement` exposes immutable session routing and scoped receipt types.
+
+#[cfg(feature = "placement")]
+pub mod placement;

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

@@ -0,0 +1,217 @@
+//! Immutable placement and receipt scopes for an independent WAL.
+//!
+//! Validate configuration before routing. Persist the full layout and its
+//! fingerprint with storage: changing addresses, voters or bucket ownership is a
+//! migration, not a restart option. This module provides no consensus or I/O.
+use std::collections::{BTreeMap, BTreeSet};
+
+use anyhow::{ensure, Context, Result};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Layout {
+    pub cluster: Uuid,
+    /// Stable physical identities and addresses, such as StatefulSet service DNS.
+    pub nodes: BTreeMap<u64, String>,
+    /// Three sorted, distinct voters per physical group; at most eight per node.
+    pub groups: BTreeMap<u64, Vec<u64>>,
+    /// Virtual bucket to physical group, independent of client worker count.
+    pub buckets: Vec<u64>,
+}
+
+impl Layout {
+    pub fn validate(&self) -> Result<()> {
+        ensure!(!self.cluster.is_nil(), "nil cluster identity");
+        ensure!(
+            !self.nodes.is_empty() && self.nodes.len() <= 192,
+            "node count out of bounds"
+        );
+        ensure!(
+            !self.groups.is_empty() && self.groups.len() <= 64,
+            "group count out of bounds"
+        );
+        ensure!(
+            self.buckets.len().is_power_of_two() && self.buckets.len() <= 65536,
+            "bucket count must be a power of two <= 65536"
+        );
+        let mut addresses = BTreeSet::new();
+        for (id, address) in &self.nodes {
+            ensure!(
+                *id > 0
+                    && *id <= u32::MAX as u64
+                    && !address.is_empty()
+                    && address.len() <= 1024
+                    && addresses.insert(address),
+                "invalid/duplicate node identity or address"
+            );
+        }
+        let mut hosted = BTreeMap::<u64, usize>::new();
+        for (group, voters) in &self.groups {
+            ensure!(
+                *group > 0 && voters.len() == 3 && voters.windows(2).all(|v| v[0] < v[1]),
+                "groups require three sorted, distinct voters and a positive id"
+            );
+            for voter in voters {
+                ensure!(self.nodes.contains_key(voter), "unknown voter");
+                *hosted.entry(*voter).or_default() += 1;
+            }
+        }
+        ensure!(
+            hosted.len() == self.nodes.len() && hosted.values().all(|n| *n <= 8),
+            "every node must host between one and eight physical groups"
+        );
+        let used: BTreeSet<_> = self.buckets.iter().copied().collect();
+        ensure!(
+            used == self.groups.keys().copied().collect(),
+            "bucket map must use exactly the configured groups"
+        );
+        Ok(())
+    }
+
+    /// Validates and hashes the canonical v1 representation. Cache this at
+    /// admission; do not serialize the layout for every payload on the hot path.
+    pub fn fingerprint(&self) -> Result<[u8; 32]> {
+        self.validate()?;
+        let mut digest = Sha256::new();
+        digest.update(b"lance-context/static-layout/v1");
+        digest.update(serde_json::to_vec(self)?);
+        Ok(digest.finalize().into())
+    }
+
+    pub fn peers(&self, group: u64) -> Result<BTreeMap<u64, String>> {
+        self.groups
+            .get(&group)
+            .context("unknown group")?
+            .iter()
+            .map(|id| Ok((*id, self.nodes.get(id).context("unknown node")?.clone())))
+            .collect()
+    }
+
+    pub fn group_for_bucket(&self, bucket: u32) -> Result<u64> {
+        self.buckets
+            .get(bucket as usize)
+            .copied()
+            .context("unknown virtual bucket")
+    }
+
+    pub fn session_bucket(
+        &self,
+        tenant: &str,
+        qualified_table: &str,
+        session: &str,
+    ) -> Result<u32> {
+        bucket_for(tenant, qualified_table, session, self.buckets.len() as u32)
+    }
+}
+
+/// Stable v1 session routing using length-prefixed UTF-8 fields and SHA-256.
+/// Persist the bucket count; changing it requires migration. This deliberately
+/// avoids Rust's unspecified DefaultHasher algorithm and process-random seeds.
+pub fn bucket_for(tenant: &str, table: &str, session: &str, bucket_count: u32) -> Result<u32> {
+    ensure!(
+        bucket_count.is_power_of_two(),
+        "bucket count must be a power of two"
+    );
+    let mut hash = Sha256::new();
+    hash.update(b"lance-context/session-route/v1");
+    for part in [tenant, table, session] {
+        hash.update((part.len() as u64).to_le_bytes());
+        hash.update(part.as_bytes());
+    }
+    let hash = hash.finalize();
+    Ok(u32::from_le_bytes(hash[..4].try_into().unwrap()) & (bucket_count - 1))
+}
+
+/// A log index is meaningful only within its cluster, layout and group.
+/// This identifies a position; it is not authentication or proof of durability.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Receipt {
+    pub cluster: Uuid,
+    pub layout: [u8; 32],
+    pub group: u64,
+    pub index: u64,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn measured_layout() -> Layout {
+        Layout {
+            cluster: Uuid::parse_str("b2da8f23-3a69-4538-9f86-2c10a61540cc").unwrap(),
+            nodes: (1..=3).map(|id| (id, format!(
+                "context-wal-lab-20261018-{}.context-wal-lab-20261018.souschef.svc.cluster.local:7800", id-1))).collect(),
+            groups: (1..=3).map(|id| (id, vec![1,2,3])).collect(),
+            buckets: (0..1024).map(|bucket| 1 + bucket % 3).collect(),
+        }
+    }
+
+    #[test]
+    fn layout_fingerprint_matches_recorded_multinode_experiment() -> Result<()> {
+        // Fixed compatibility vector from the real PHX protocol10 experiment.
+        // Changing serialization or the hash domain must not reroute old data.
+        let layout = measured_layout();
+        let expected = [
+            32, 110, 106, 153, 114, 83, 142, 215, 88, 81, 79, 126, 93, 225, 129, 65, 94, 127, 167,
+            140, 244, 218, 38, 47, 110, 21, 222, 171, 71, 7, 133, 7,
+        ];
+        assert_eq!(layout.fingerprint()?, expected);
+        let decoded: Layout = serde_json::from_slice(&serde_json::to_vec(&layout)?)?;
+        assert_eq!(decoded.fingerprint()?, expected);
+        let mut changed = layout.clone();
+        changed.buckets.swap(0, 1);
+        assert_ne!(changed.fingerprint()?, expected);
+        let mut changed = layout;
+        changed.nodes.get_mut(&1).unwrap().push_str("-changed");
+        assert_ne!(changed.fingerprint()?, expected);
+        Ok(())
+    }
+
+    #[test]
+    fn placement_rejects_ambiguous_or_incomplete_ownership() {
+        let mut layout = measured_layout();
+        layout.groups.insert(1, vec![1, 1, 3]);
+        assert!(layout.validate().is_err());
+        let mut layout = measured_layout();
+        layout.nodes.insert(2, layout.nodes[&1].clone());
+        assert!(layout.validate().is_err());
+        let mut layout = measured_layout();
+        layout.buckets[0] = 99;
+        assert!(layout.validate().is_err());
+        let mut layout = measured_layout();
+        layout.nodes.insert(4, "unused:7800".into());
+        assert!(layout.validate().is_err());
+        let mut layout = measured_layout();
+        layout.groups = (1..=9).map(|id| (id, vec![1, 2, 3])).collect();
+        layout.buckets = (0..16).map(|id| 1 + id % 9).collect();
+        assert!(layout.validate().is_err());
+        assert!(bucket_for("tenant", "table", "session", 0).is_err());
+        assert!(bucket_for("tenant", "table", "session", 3).is_err());
+    }
+
+    #[test]
+    fn session_routing_survives_configuration_roundtrip() -> Result<()> {
+        // SHA-256 of the literal v1 prefix and three LE-u64 length=1 fields
+        // a/b/c, independently checked with OpenSSL: c5f0901a5b7f96b9... .
+        assert_eq!(bucket_for("a", "b", "c", 65536)?, 0xf0c5);
+        let layout = measured_layout();
+        let restored: Layout = serde_json::from_slice(&serde_json::to_vec(&layout)?)?;
+        for session in ["session-1", "session-2", "训练会话", ""] {
+            let bucket = layout.session_bucket("tenant", "namespace/table", session)?;
+            assert_eq!(
+                restored.session_bucket("tenant", "namespace/table", session)?,
+                bucket
+            );
+            assert_eq!(
+                restored.group_for_bucket(bucket)?,
+                layout.group_for_bucket(bucket)?
+            );
+        }
+        assert!(restored.group_for_bucket(1024).is_err());
+        assert!(restored.peers(99).is_err());
+        Ok(())
+    }
+}