diff --git a/.github/workflows/rust-publish.yml b/.github/workflows/rust-publish.yml index 652d1865..02f9c3e0 100644 --- a/.github/workflows/rust-publish.yml +++ b/.github/workflows/rust-publish.yml @@ -75,7 +75,9 @@ jobs: - name: Validate Rust crate packages if: github.event_name == 'pull_request' - run: cargo package --workspace --exclude lance-context-python --allow-dirty --no-verify --all-features --locked + # The private ingress adapter depends on the unpublished WAL crate. + # Rust Tests checks this workspace-only integration with its features. + run: cargo package --workspace --exclude lance-context-python --exclude lance-context-ingest --allow-dirty --no-verify --all-features --locked - name: Publish Rust crates if: github.event_name != 'pull_request' diff --git a/Cargo.lock b/Cargo.lock index da498434..83f3d314 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4174,6 +4174,30 @@ dependencies = [ "uuid", ] +[[package]] +name = "lance-context-ingest" +version = "0.6.7" +dependencies = [ + "anyhow", + "arrow-array", + "arrow-ipc", + "arrow-schema", + "base64 0.22.1", + "bytes", + "lance", + "lance-context-core", + "lance-context-wal", + "lance-table", + "rcgen", + "ring", + "serde", + "serde_json", + "tempfile", + "tokio", + "tokio-rustls", + "uuid", +] + [[package]] name = "lance-context-master" version = "0.6.3" diff --git a/Cargo.toml b/Cargo.toml index d5cd296f..62fb5c54 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ members = [ "crates/lance-context-merge", "crates/lance-context-core", + "crates/lance-context-ingest", "crates/lance-context-wal", "crates/lance-context-metrics", "crates/lance-context", diff --git a/crates/lance-context-ingest/Cargo.toml b/crates/lance-context-ingest/Cargo.toml new file mode 100644 index 00000000..83d723b3 --- /dev/null +++ b/crates/lance-context-ingest/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "lance-context-ingest" +version = "0.6.7" +edition = "2021" +license = "Apache-2.0" +publish = false +description = "Durable rollout ingress over the independent lance-context WAL" + +[dependencies] +anyhow = "1" +arrow-array = "58" +arrow-schema = "58" +base64 = "0.22" +bytes = "1" +lance-context-core = { version = "0.6.7", path = "../lance-context-core" } +lance-context-wal = { version = "0.6.7", path = "../lance-context-wal", features = ["peer-transport", "ipc", "arrow-contract"] } +ring = "0.17" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["rt", "sync", "time"] } +uuid = { version = "1.20.0", features = ["v4", "serde"] } + +[dev-dependencies] +arrow-ipc = "58" +lance = "13.0.0" +lance-table = "13.0.0" +rcgen = "0.14" +tempfile = "3" +tokio = { version = "1", features = ["rt-multi-thread", "macros", "net"] } +tokio-rustls = { version = "0.26.4", default-features = false, features = ["ring", "tls12"] } diff --git a/crates/lance-context-ingest/src/lib.rs b/crates/lance-context-ingest/src/lib.rs new file mode 100644 index 00000000..222ae07b --- /dev/null +++ b/crates/lance-context-ingest/src/lib.rs @@ -0,0 +1,566 @@ +//! Durable Arrow rollout ingestion over the independent WAL's authenticated RPCs. +//! +//! Use a newly bound immutable-ID rollout table. Resolve/authorize its namespace, +//! table generation and manifest incarnation before connecting. This does not +//! migrate or replace legacy merge/upsert tables, or authenticate an HTTP caller. +//! One atomic append must fit one virtual bucket; reject mixed buckets before ACK. +use anyhow::{bail, ensure, Context, Result}; +use arrow_array::{Array, StringArray}; +use arrow_schema::SchemaRef; +use bytes::Bytes; +use lance_context_core::{rollout_schema, MergeMemoryBudget, MergeReservation}; +use lance_context_wal::{ + ingest_contract::{arrow_schema_fingerprint, IngestContract, WritePolicy}, + ipc, + placement::{Layout, Receipt}, + record::{Batch, BatchMeta, RecordKind}, + replication::{ + producer, + publication::Destination, + query, + runtime::{ + client::{Client, Reply, Request}, + Command, + }, + Applied, Outcome, + }, +}; +use serde::{Deserialize, Serialize}; +use std::{collections::HashSet, sync::Arc, time::Duration}; +use uuid::Uuid; + +pub mod receipt; + +/// Resolved by the application's authorized catalog, not an uploaded URI/UUID. +#[derive(Clone, Debug)] +pub struct TableBinding { + pub table: Uuid, + pub destination: Destination, + pub incarnation: Uuid, + pub tenant: String, + pub qualified_table: String, + pub max_record_bytes: u32, + pub max_rows: u32, +} +impl TableBinding { + pub fn contract(&self) -> Result { + let value = IngestContract { + tenant: self.tenant.clone(), + qualified_table: self.qualified_table.clone(), + schema: arrow_schema_fingerprint(&rollout_schema())?, + policy: WritePolicy::ImmutableUniqueIds, + max_record_bytes: self.max_record_bytes, + max_rows: self.max_rows, + }; + value.validate()?; + Ok(value) + } +} +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Scope { + cluster: Uuid, + layout: [u8; 32], + table: Uuid, + destination: Destination, + incarnation: Uuid, + contract: IngestContract, +} +#[derive(Clone)] +pub struct Ingress { + client: Arc, + layout: Arc, + scope: Arc, + schema: SchemaRef, + budget: Arc, +} +/// Held before body buffering; retained by preparation and every ambiguous retry. +pub struct Upload { + scope: Arc, + bytes: usize, + reservation: MergeReservation, +} +/// Clone only the opaque Bytes internally; do not re-encode an ambiguous retry. +pub struct PreparedAppend { + scope: Arc, + batch: Batch, + rows: u32, + _reservation: MergeReservation, +} +impl PreparedAppend { + pub fn meta(&self) -> &BatchMeta { + &self.batch.meta + } + pub fn rows(&self) -> u32 { + self.rows + } + pub fn body(&self) -> &Bytes { + &self.batch.body + } +} +/// Constructor and receipt fields are private: a bare index is not membership. +#[derive(Clone, Debug)] +pub struct AcceptedAppend { + scope: Arc, + position: Receipt, + meta: BatchMeta, + rows: u32, +} +impl AcceptedAppend { + pub fn position(&self) -> &Receipt { + &self.position + } + pub fn meta(&self) -> &BatchMeta { + &self.meta + } + pub fn rows(&self) -> u32 { + self.rows + } +} +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProducerOpen { + /// Persist the complete tuple before the first open RPC; retry it unchanged. + pub bucket: u32, + pub slot: u32, + pub expected_generation: u64, + pub operation: Uuid, +} +#[derive(Clone, Debug)] +pub struct Producer { + scope: Arc, + open: ProducerOpen, + id: Uuid, + generation: u64, +} +impl Producer { + pub fn id(&self) -> Uuid { + self.id + } + pub fn bucket(&self) -> u32 { + self.open.bucket + } + pub fn generation(&self) -> u64 { + self.generation + } +} +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AppendFailure { + Sequence { expected: u64 }, + Capacity, + ProducerFenced, + Unknown, +} +impl std::fmt::Display for AppendFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Sequence { expected } => { + write!(f, "producer sequence rejected; expected {expected}") + } + Self::Capacity => f.write_str("WAL capacity exhausted"), + Self::ProducerFenced => f.write_str("producer fenced"), + Self::Unknown => { + f.write_str("append outcome unknown; retain exact producer sequence and bytes") + } + } + } +} +impl std::error::Error for AppendFailure {} +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum PublicationStatus { + Pending, + Published { minimum_read_version: u64 }, +} + +impl Ingress { + pub async fn connect( + client: Arc, + layout: Arc, + binding: TableBinding, + budget: Arc, + ) -> Result { + let fingerprint = layout.fingerprint()?; + ensure!( + client.identity() == (layout.cluster, fingerprint), + "ingress and client layout differ" + ); + ensure!( + !binding.table.is_nil() && !binding.incarnation.is_nil() && client.producer_slots() > 0, + "ingress requires a generation and bounded producer slots" + ); + let contract = binding.contract()?; + ensure!( + contract.max_record_bytes as usize <= client.max_record_bytes(), + "contract exceeds WAL record capacity" + ); + ensure!( + workspace(&contract, contract.max_record_bytes as usize)? <= budget.limit(), + "contract exceeds shared ingress workspace" + ); + let value = Self { + client, + layout: layout.clone(), + scope: Arc::new(Scope { + cluster: layout.cluster, + layout: fingerprint, + table: binding.table, + destination: binding.destination, + incarnation: binding.incarnation, + contract, + }), + schema: Arc::new(rollout_schema()), + budget, + }; + value.visibility().await?; + Ok(value) + } + fn group(&self, bucket: u32) -> Result { + self.layout.group_for_bucket(bucket) + } + pub fn bucket_for_session(&self, session: &str) -> Result { + ensure!(!session.is_empty(), "empty session key"); + self.layout.session_bucket( + &self.scope.contract.tenant, + &self.scope.contract.qualified_table, + session, + ) + } + async fn visibility(&self) -> Result> { + let group = self.client.catalog_group(); + let response = self + .client + .request_routed(Request::Query { + group, + query: query::Query::Visibility { + table: self.scope.table, + }, + }) + .await?; + let Reply::Query(query::Outcome::Observed { + cut, + value: query::Value::Visibility(Some(view)), + }) = response.reply + else { + bail!("table binding unavailable at quorum"); + }; + ensure!( + cut.cluster == self.scope.cluster + && cut.layout == self.scope.layout + && cut.group == group + && cut.term > 0 + && cut.applied >= cut.quorum_index, + "foreign visibility quorum scope" + ); + ensure!( + view.destination == self.scope.destination + && view.ingest == self.scope.contract + && view.incarnation == Some(self.scope.incarnation), + "durable table binding differs from resolved rollout generation" + ); + Ok(view) + } + pub async fn open_producer(&self, open: ProducerOpen) -> Result { + ensure!( + !open.operation.is_nil() + && open.slot < self.client.producer_slots() + && open.expected_generation < u64::MAX, + "invalid producer open identity" + ); + let response = self + .client + .request_routed(Request::Commit(Box::new(Command::Producer( + producer::Command { + cluster: self.scope.cluster, + layout: self.scope.layout, + group: self.group(open.bucket)?, + table: self.scope.table, + slot: open.slot, + action: producer::Action::Open { + expected_generation: open.expected_generation, + operation: open.operation, + bucket: open.bucket, + }, + }, + )))) + .await?; + let Reply::Commit(Outcome::Committed(Applied::Producer(producer::Reply::Opened { + producer: id, + generation, + .. + }))) = response.reply + else { + bail!("producer allocation not confirmed; reconcile the original open tuple"); + }; + ensure!( + generation == open.expected_generation + 1 + && id == producer::producer_id(open.slot, generation), + "foreign producer allocation" + ); + Ok(Producer { + scope: self.scope.clone(), + open, + id, + generation, + }) + } + pub async fn close_producer( + &self, + producer: &Producer, + last: Option<&AcceptedAppend>, + ) -> Result<()> { + ensure!(*producer.scope == *self.scope, "foreign producer scope"); + let expected_last = last + .map(|last| { + self.check_receipt(last)?; + ensure!( + last.meta.producer == producer.id && last.meta.bucket == producer.open.bucket, + "foreign producer receipt" + ); + Ok::<_, anyhow::Error>(producer::Last { + sequence: last.meta.sequence, + index: last.position.index, + }) + }) + .transpose()?; + let response = self + .client + .request_routed(Request::Commit(Box::new(Command::Producer( + producer::Command { + cluster: self.scope.cluster, + layout: self.scope.layout, + group: self.group(producer.open.bucket)?, + table: self.scope.table, + slot: producer.open.slot, + action: producer::Action::Close { + generation: producer.generation, + expected_last, + }, + }, + )))) + .await?; + ensure!( + matches!(response.reply, Reply::Commit(Outcome::Committed(Applied::Producer(producer::Reply::Closed { last }))) if last == expected_last), + "producer close not confirmed; preserve its outstanding state" + ); + Ok(()) + } + /// Reserve before reading/buffering the IPC body. The row allowance also + /// budgets the identity/routing workspace; this is not an allocator quota. + pub async fn reserve_upload(&self, bytes: usize) -> Result { + ensure!( + bytes > 0 && bytes <= self.scope.contract.max_record_bytes as usize, + "upload exceeds registered record limit" + ); + let amount = workspace(&self.scope.contract, bytes)?; + Ok(Upload { + scope: self.scope.clone(), + bytes, + reservation: self.budget.reserve(amount).await, + }) + } + /// Arrow validation and payload hashing stay off the async I/O executor. + /// Cancellation keeps the reservation with the running blocking operation. + pub async fn prepare( + &self, + upload: Upload, + body: Bytes, + producer: &Producer, + sequence: u64, + ) -> Result { + ensure!( + Arc::ptr_eq(&upload.scope, &self.scope) && *producer.scope == *self.scope, + "foreign preparation scope" + ); + ensure!( + sequence > 0 && sequence < u64::MAX && body.len() == upload.bytes, + "invalid sequence or upload length" + ); + let schema = self.schema.clone(); + let layout = self.layout.clone(); + let producer = producer.clone(); + tokio::task::spawn_blocking(move || { + prepare(upload, body, producer, sequence, schema, &layout) + }) + .await + .context("ingress preparation task failed")? + } + pub async fn append(&self, prepared: &PreparedAppend) -> Result { + ensure!( + *prepared.scope == *self.scope, + "prepared append belongs to another table" + ); + let group = self.group(prepared.batch.meta.bucket)?; + let response = self + .client + .request_routed(Request::Append { + group, + batch: prepared.batch.clone(), + }) + .await + .map_err(|error| anyhow::Error::new(AppendFailure::Unknown).context(error))?; + let index = match response.reply { + Reply::Commit(Outcome::Committed(Applied::Accepted { index })) => index, + Reply::Commit(Outcome::Committed(Applied::Rejected { expected_sequence })) => { + return Err(AppendFailure::Sequence { + expected: expected_sequence, + } + .into()) + } + Reply::Commit(Outcome::Committed(Applied::Capacity)) => { + return Err(AppendFailure::Capacity.into()) + } + Reply::Commit(Outcome::Committed(Applied::ProducerFenced)) => { + return Err(AppendFailure::ProducerFenced.into()) + } + _ => return Err(AppendFailure::Unknown.into()), + }; + ensure!(index > 0 && index < u64::MAX, "invalid accepted index"); + Ok(AcceptedAppend { + scope: self.scope.clone(), + position: Receipt { + cluster: self.scope.cluster, + layout: self.scope.layout, + group, + index, + }, + meta: prepared.batch.meta.clone(), + rows: prepared.rows, + }) + } + fn check_receipt(&self, accepted: &AcceptedAppend) -> Result<()> { + ensure!( + *accepted.scope == *self.scope + && accepted.position.cluster == self.scope.cluster + && accepted.position.layout == self.scope.layout + && accepted.position.group == self.group(accepted.meta.bucket)? + && accepted.position.index > 0 + && accepted.position.index < u64::MAX + && accepted.meta.stream == self.scope.table + && accepted.meta.kind == RecordKind::Data + && !accepted.meta.producer.is_nil() + && accepted.meta.sequence > 0 + && accepted.meta.sequence < u64::MAX + && accepted.rows > 0 + && accepted.rows <= self.scope.contract.max_rows, + "foreign or invalid accepted receipt" + ); + accepted.meta.check_body_length( + usize::try_from(accepted.meta.body_bytes)?, + self.scope.contract.max_record_bytes as usize, + )?; + Ok(()) + } + pub async fn publication_status(&self, accepted: &AcceptedAppend) -> Result { + self.check_receipt(accepted)?; + let view = self.visibility().await?; + if view + .checkpoint + .get(&accepted.position.group) + .copied() + .unwrap_or(0) + < accepted.position.index + { + return Ok(PublicationStatus::Pending); + } + ensure!( + view.minimum_read_version > 0, + "published prefix has no Lance version" + ); + Ok(PublicationStatus::Published { + minimum_read_version: view.minimum_read_version, + }) + } + pub async fn wait_published(&self, accepted: &AcceptedAppend, limit: Duration) -> Result { + tokio::time::timeout(limit, async { + loop { + if let PublicationStatus::Published { + minimum_read_version, + } = self.publication_status(accepted).await? + { + return Ok(minimum_read_version); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .context("publication deadline expired; durable append remains accepted")? + } +} +fn workspace(contract: &IngestContract, bytes: usize) -> Result { + bytes + .checked_mul(3) + .and_then(|v| v.checked_add(contract.max_rows as usize * 64)) + .context("ingress workspace overflow") +} +fn prepare( + upload: Upload, + body: Bytes, + producer: Producer, + sequence: u64, + schema: SchemaRef, + layout: &Layout, +) -> Result { + let batches = ipc::decode(body.clone(), schema)?; + let mut ids = HashSet::new(); + let mut rows = 0u32; + let mut previous_session = None; + for batch in &batches { + ensure!( + batch.num_rows() <= (upload.scope.contract.max_rows - rows) as usize, + "upload exceeds registered row limit" + ); + for (field, column) in batch.schema().fields().iter().zip(batch.columns()) { + column.to_data().validate_full()?; + ensure!( + field.is_nullable() || column.logical_null_count() == 0, + "null in a required rollout field" + ); + } + let sessions = batch + .column_by_name("rollout_id") + .and_then(|a| a.as_any().downcast_ref::()) + .context("missing rollout_id column")?; + let keys = batch + .column_by_name("id") + .and_then(|a| a.as_any().downcast_ref::()) + .context("missing id column")?; + for (session, id) in sessions.iter().zip(keys.iter()) { + let session = session + .filter(|s| !s.is_empty()) + .context("empty session key")?; + let id = id.filter(|s| !s.is_empty()).context("empty row id")?; + ensure!(ids.insert(id), "duplicate row id within atomic append"); + // Consecutive rows in a rollout share routing. Borrow the decoded + // string and hash only when the session changes; no string copy. + if previous_session != Some(session) { + let bucket = layout.session_bucket( + &upload.scope.contract.tenant, + &upload.scope.contract.qualified_table, + session, + )?; + ensure!( + bucket == producer.open.bucket, + "atomic append spans producer bucket" + ); + previous_session = Some(session); + } + rows += 1; + } + } + ensure!(rows > 0, "empty append"); + let batch = Batch::new( + upload.scope.table, + producer.open.bucket, + producer.id, + sequence, + body, + ); + Ok(PreparedAppend { + scope: upload.scope, + batch, + rows, + _reservation: upload.reservation, + }) +} + +#[cfg(test)] +mod tests; diff --git a/crates/lance-context-ingest/src/receipt.rs b/crates/lance-context-ingest/src/receipt.rs new file mode 100644 index 00000000..e131e0e9 --- /dev/null +++ b/crates/lance-context-ingest/src/receipt.rs @@ -0,0 +1,138 @@ +//! Persistable membership proof for a durable ACK, without retaining its body. +//! +//! Supply persistent random keys from trusted service configuration. All key +//! holders can issue receipts; tokens neither authenticate users nor authorize +//! table access. Resolve and authorize the table before restoring a receipt. +//! Install a new key on all readers before switching issuance, and retain old +//! keys while their receipts must remain verifiable. Tokens do not expire: +//! losing or retiring a key must never authorize deletion of acknowledged data. + +use super::{AcceptedAppend, Ingress, Scope}; +use anyhow::{ensure, Context, Result}; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; +use lance_context_wal::{placement::Receipt, record::BatchMeta}; +use ring::{digest, hmac}; +use serde::{Deserialize, Serialize}; +use std::{collections::BTreeMap, sync::Arc}; + +const VERSION: u8 = 1; +const HEADER: usize = 5; +const TAG: usize = 32; +const MAX_CLAIMS: usize = 8192; +const MAX_BINARY: usize = HEADER + MAX_CLAIMS + TAG; +const MAX_TOKEN: usize = MAX_BINARY.div_ceil(3) * 4; +const DOMAIN: &[u8] = b"lance-context/durable-ingress-receipt/v1"; + +/// Deliberately not serializable or Debug: key material stays in configuration. +pub struct ReceiptKeys { + active: u32, + keys: BTreeMap, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Claims { + scope: [u8; 32], + position: Receipt, + meta: BatchMeta, + rows: u32, +} + +impl ReceiptKeys { + pub fn new(active: u32, keys: impl IntoIterator) -> Result { + let mut configured = BTreeMap::new(); + for (id, key) in keys { + ensure!( + id != 0 && key != [0; 32], + "invalid receipt key configuration" + ); + ensure!(configured.len() < 16, "too many receipt keys"); + ensure!( + configured + .insert(id, hmac::Key::new(hmac::HMAC_SHA256, &key)) + .is_none(), + "duplicate receipt key identifier" + ); + } + ensure!( + configured.contains_key(&active), + "active receipt key is missing" + ); + Ok(Self { + active, + keys: configured, + }) + } + + pub fn encode(&self, accepted: &AcceptedAppend) -> Result { + let claims = serde_json::to_vec(&Claims { + scope: scope_hash(&accepted.scope)?, + position: accepted.position.clone(), + meta: accepted.meta.clone(), + rows: accepted.rows, + })?; + ensure!( + claims.len() <= MAX_CLAIMS, + "receipt claims exceed protocol limit" + ); + let mut binary = Vec::with_capacity(HEADER + claims.len() + TAG); + binary.push(VERSION); + binary.extend_from_slice(&self.active.to_le_bytes()); + binary.extend_from_slice(&claims); + let mut context = hmac::Context::with_key(&self.keys[&self.active]); + context.update(DOMAIN); + context.update(&binary); + binary.extend_from_slice(context.sign().as_ref()); + Ok(URL_SAFE_NO_PAD.encode(binary)) + } + + fn decode(&self, token: &str, scope: Arc) -> Result { + ensure!( + token.len() <= MAX_TOKEN, + "receipt token exceeds protocol limit" + ); + let binary = URL_SAFE_NO_PAD + .decode(token) + .context("invalid receipt encoding")?; + ensure!( + (HEADER + TAG + 1..=MAX_BINARY).contains(&binary.len()), + "invalid receipt length" + ); + ensure!(binary[0] == VERSION, "unsupported receipt version"); + let id = u32::from_le_bytes(binary[1..HEADER].try_into()?); + let key = self.keys.get(&id).context("receipt key is unavailable")?; + let end = binary.len() - TAG; + // Verification is constant-time and precedes parsing any claims. The + // domain prefix is bounded and contains no caller-controlled length. + let mut authenticated = Vec::with_capacity(DOMAIN.len() + end); + authenticated.extend_from_slice(DOMAIN); + authenticated.extend_from_slice(&binary[..end]); + hmac::verify(key, &authenticated, &binary[end..]) + .map_err(|_| anyhow::anyhow!("invalid receipt authentication"))?; + let claims: Claims = serde_json::from_slice(&binary[HEADER..end])?; + ensure!(claims.scope == scope_hash(&scope)?, "foreign receipt scope"); + Ok(AcceptedAppend { + scope, + position: claims.position, + meta: claims.meta, + rows: claims.rows, + }) + } +} + +fn scope_hash(scope: &Scope) -> Result<[u8; 32]> { + let mut hash = digest::Context::new(&digest::SHA256); + hash.update(b"lance-context/durable-ingress-scope/v1"); + hash.update(&serde_json::to_vec(scope)?); + Ok(hash.finish().as_ref().try_into()?) +} + +impl Ingress { + /// Restore only after the application resolves and authorizes this binding. + /// Complete scope comparison includes the table incarnation and contract. + pub fn restore_receipt(&self, keys: &ReceiptKeys, token: &str) -> Result { + let accepted = keys.decode(token, self.scope.clone())?; + self.check_receipt(&accepted)?; + Ok(accepted) + } +} diff --git a/crates/lance-context-ingest/src/tests.rs b/crates/lance-context-ingest/src/tests.rs new file mode 100644 index 00000000..d6d2473e --- /dev/null +++ b/crates/lance-context-ingest/src/tests.rs @@ -0,0 +1,440 @@ +use super::*; +use arrow_array::{ + new_null_array, types::Int8Type, ArrayRef, DictionaryArray, Int32Array, Int8Array, + LargeBinaryArray, RecordBatch, RecordBatchIterator, TimestampMicrosecondArray, +}; +use arrow_ipc::writer::FileWriter; +use lance::{dataset::WriteParams, Dataset}; +use lance_context_wal::replication::publication as p; +use lance_table::io::commit::{ConditionalPutCommitHandler, ManifestNamingScheme}; +use receipt::ReceiptKeys; +use std::collections::BTreeMap; + +mod network; + +fn layout() -> Arc { + Arc::new(Layout { + cluster: Uuid::from_u128(1), + nodes: (1..=3) + .map(|id| (id, format!("wal-{id}.example.invalid:7000"))) + .collect(), + groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![1, 2, 3])]), + buckets: vec![10, 20, 10, 20], + }) +} +fn binding() -> TableBinding { + TableBinding { + table: Uuid::from_u128(2), + destination: Destination { + namespace: Uuid::from_u128(3), + dataset: "rollouts.lance".into(), + }, + incarnation: Uuid::from_u128(4), + tenant: "tenant".into(), + qualified_table: "rollouts".into(), + max_record_bytes: 1 << 20, + max_rows: 8, + } +} +fn scope(binding: &TableBinding) -> Arc { + Arc::new(Scope { + cluster: layout().cluster, + layout: layout().fingerprint().unwrap(), + table: binding.table, + destination: binding.destination.clone(), + incarnation: binding.incarnation, + contract: binding.contract().unwrap(), + }) +} +fn session(bucket: u32) -> String { + (0..1000) + .map(|i| format!("session-{i}")) + .find(|s| layout().session_bucket("tenant", "rollouts", s).unwrap() == bucket) + .unwrap() +} +fn batch(ids: &[&str], sessions: &[&str]) -> Result { + let schema = Arc::new(rollout_schema()); + let n = ids.len(); + ensure!(n == sessions.len(), "test row mismatch"); + let payload = vec![0x5a; 16 << 10]; + let columns = schema + .fields() + .iter() + .map(|field| -> Result { + Ok(match field.name().as_str() { + "id" => Arc::new(StringArray::from(ids.to_vec())), + "rollout_id" => Arc::new(StringArray::from(sessions.to_vec())), + "problem_id" => Arc::new(StringArray::from(vec!["problem"; n])), + "sequence_order" => Arc::new(Int32Array::from_iter_values(0..n as i32)), + "role" => Arc::new(DictionaryArray::::try_new( + Int8Array::from(vec![0; n]), + Arc::new(StringArray::from(vec!["assistant"])), + )?), + "created_at" => Arc::new(TimestampMicrosecondArray::from(vec![1234; n])), + "content_type" => Arc::new(StringArray::from(vec!["binary"; n])), + "binary_payload" => Arc::new(LargeBinaryArray::from_iter_values( + (0..n).map(|_| payload.as_slice()), + )), + _ => new_null_array(field.data_type(), n), + }) + }) + .collect::>>()?; + Ok(RecordBatch::try_new(schema, columns)?) +} +fn ipc_file(batches: &[RecordBatch]) -> Result { + let mut body = Vec::new(); + let mut writer = FileWriter::try_new(&mut body, &rollout_schema())?; + for batch in batches { + writer.write(batch)?; + } + writer.finish()?; + drop(writer); + Ok(Bytes::from(body)) +} +async fn prepare_local(body: Bytes, binding: &TableBinding, bucket: u32) -> Result { + let scope = scope(binding); + let budget = MergeMemoryBudget::new(8 << 20); + let reservation = budget + .reserve(workspace(&scope.contract, body.len())?) + .await; + let producer = Producer { + scope: scope.clone(), + open: ProducerOpen { + bucket, + slot: 0, + expected_generation: 0, + operation: Uuid::from_u128(5), + }, + id: producer::producer_id(0, 1), + generation: 1, + }; + prepare( + Upload { + scope, + bytes: body.len(), + reservation, + }, + body, + producer, + 1, + Arc::new(rollout_schema()), + &layout(), + ) +} + +#[tokio::test] +async fn validates_whole_atomic_batch_and_preserves_original_buffer() -> Result<()> { + let s = session(0); + let body = ipc_file(&[batch(&["one", "two"], &[&s, &s])?])?; + let prepared = prepare_local(body.clone(), &binding(), 0).await?; + assert_eq!(prepared.body().as_ptr(), body.as_ptr()); + assert_eq!(prepared.rows(), 2); + // Buckets 0 and 2 deliberately share the same physical group. + let other = session(2); + assert!(prepare_local( + ipc_file(&[batch(&["one", "two"], &[&s, &other])?])?, + &binding(), + 0 + ) + .await + .is_err()); + assert!(prepare_local( + ipc_file(&[batch(&["one"], &[&s])?, batch(&["one"], &[&s])?])?, + &binding(), + 0 + ) + .await + .is_err()); + assert!( + prepare_local(ipc_file(&[batch(&[""], &[&s])?])?, &binding(), 0) + .await + .is_err() + ); + assert!( + prepare_local(ipc_file(&[batch(&["one"], &[""])?])?, &binding(), 0) + .await + .is_err() + ); + let mut limited = binding(); + limited.max_rows = 1; + assert!(prepare_local( + ipc_file(&[batch(&["one"], &[&s])?, batch(&["two"], &[&s])?])?, + &limited, + 0 + ) + .await + .is_err()); + assert!(prepare_local(ipc_file(&[batch(&[], &[])?])?, &binding(), 0) + .await + .is_err()); + assert!( + prepare_local(Bytes::from_static(b"invalid IPC"), &binding(), 0) + .await + .is_err() + ); + Ok(()) +} + +async fn bootstrap( + c: &network::Cluster, + path: &std::path::Path, + binding: &TableBinding, +) -> Result<()> { + let dataset = Dataset::write( + RecordBatchIterator::new(vec![Ok(batch(&[], &[])?)], Arc::new(rollout_schema())), + path.to_str().context("test path")?, + Some(WriteParams { + commit_handler: Some(Arc::new(ConditionalPutCommitHandler)), + ..Default::default() + }), + ) + .await?; + let response = c + .client + .request_routed(Request::Commit(Box::new(Command::Acquire(p::Acquire { + table: binding.table, + destination: binding.destination.clone(), + ingest: binding.contract()?, + owner: Uuid::new_v4(), + request: Uuid::new_v4(), + expected_epoch: 0, + })))) + .await?; + let Reply::Commit(Outcome::Committed(Applied::Publisher(p::Reply::Acquired(authority)))) = + response.reply + else { + bail!("publisher allocation failed"); + }; + let location = dataset.manifest_location(); + let response = c + .client + .request_routed(Request::Commit(Box::new(Command::Bootstrap( + p::BootstrapManifest { + authority, + incarnation: binding.incarnation, + request: Uuid::new_v4(), + version: 1, + object: p::ManifestObject { + path: format!( + "_versions/{}", + location.path.filename().context("manifest name")? + ), + bytes: location.size.context("manifest size")?, + v2_names: location.naming_scheme == ManifestNamingScheme::V2, + }, + transaction: Uuid::parse_str( + &dataset + .read_transaction_by_version(1) + .await? + .context("genesis transaction")? + .uuid, + )?, + }, + )))) + .await?; + ensure!( + matches!( + response.reply, + Reply::Commit(Outcome::Committed(Applied::Publisher( + p::Reply::ManifestReserved(_) + ))) + ), + "bootstrap not confirmed" + ); + Ok(()) +} +async fn connect(c: &network::Cluster, binding: TableBinding) -> Result { + Ingress::connect( + c.client.clone(), + layout(), + binding, + MergeMemoryBudget::new(8 << 20), + ) + .await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn durable_core_ack_receipt_and_exact_retry_survive_three_replica_restart() -> Result<()> { + let root = tempfile::tempdir()?; + let c = network::Cluster::open(root.path(), layout()).await?; + let b = binding(); + bootstrap(&c, &root.path().join("rollouts.lance"), &b).await?; + let ingress = connect(&c, b.clone()).await?; + let mut wrong = b.clone(); + wrong.incarnation = Uuid::new_v4(); + assert!(connect(&c, wrong).await.is_err()); + let mut wrong = b.clone(); + wrong.destination.dataset = "other.lance".into(); + assert!(connect(&c, wrong).await.is_err()); + let mut wrong = b.clone(); + wrong.max_rows += 1; + assert!(connect(&c, wrong).await.is_err()); + assert!(Ingress::connect( + c.client.clone(), + layout(), + b.clone(), + MergeMemoryBudget::new(1 << 20) + ) + .await + .is_err()); + assert!(ingress.reserve_upload(0).await.is_err()); + assert!(ingress + .reserve_upload(1 + b.max_record_bytes as usize) + .await + .is_err()); + let full_a = ingress.reserve_upload(b.max_record_bytes as usize).await?; + let full_b = ingress.reserve_upload(b.max_record_bytes as usize).await?; + assert!(tokio::time::timeout( + Duration::from_millis(30), + ingress.reserve_upload(b.max_record_bytes as usize) + ) + .await + .is_err()); + drop(full_a); + let admitted = tokio::time::timeout( + Duration::from_secs(1), + ingress.reserve_upload(b.max_record_bytes as usize), + ) + .await??; + drop((full_b, admitted)); + assert_eq!(ingress.budget.reserved(), 0); + // Acquire before buffering; the same budget remains held through prepare + // and explicit retries, and is released only when the prepared body drops. + let open = ProducerOpen { + bucket: 0, + slot: 0, + expected_generation: 0, + operation: Uuid::new_v4(), + }; + let producer = ingress.open_producer(open.clone()).await?; + let s = session(0); + let body = ipc_file(&[batch(&["one", "two"], &[&s, &s])?])?; + let prepared = ingress + .prepare( + ingress.reserve_upload(body.len()).await?, + body.clone(), + &producer, + 1, + ) + .await?; + assert_eq!(prepared.body().as_ptr(), body.as_ptr()); + assert!(ingress.budget.reserved() >= body.len()); + let accepted = ingress.append(&prepared).await?; + let retry = ingress.append(&prepared).await?; + assert_eq!(retry.position(), accepted.position()); + assert_eq!( + ingress.publication_status(&accepted).await?, + PublicationStatus::Pending + ); + assert!(ingress + .wait_published(&accepted, Duration::from_millis(30)) + .await + .is_err()); + let altered = ipc_file(&[batch(&["changed", "two"], &[&s, &s])?])?; + let altered = ingress + .prepare( + ingress.reserve_upload(altered.len()).await?, + altered, + &producer, + 1, + ) + .await?; + let error = ingress.append(&altered).await.unwrap_err(); + assert!(matches!( + error.downcast_ref::(), + Some(AppendFailure::Sequence { .. }) + )); + drop(altered); + let keys = ReceiptKeys::new(1, [(1, [0x41; 32])])?; + let token = keys.encode(&accepted)?; + let restored = ingress.restore_receipt(&keys, &token)?; + assert_eq!(restored.meta(), accepted.meta()); + let bad_keys = ReceiptKeys::new(1, [(1, [0x42; 32])])?; + assert!(ingress.restore_receipt(&bad_keys, &token).is_err()); + use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; + let mut corrupt = URL_SAFE_NO_PAD.decode(&token)?; + corrupt[10] ^= 1; + assert!(ingress + .restore_receipt(&keys, &URL_SAFE_NO_PAD.encode(corrupt)) + .is_err()); + assert!(ingress + .restore_receipt(&keys, &"a".repeat(64 << 10)) + .is_err()); + let mut foreign = ingress.clone(); + Arc::make_mut(&mut foreign.scope).incarnation = Uuid::new_v4(); + assert!(foreign.restore_receipt(&keys, &token).is_err()); + let position = accepted.position().clone(); + let meta = accepted.meta().clone(); + drop(prepared); + assert_eq!(ingress.budget.reserved(), 0); + drop(ingress); + c.shutdown().await?; + + let c = network::Cluster::open(root.path(), layout()).await?; + let ingress = connect(&c, b).await?; + let rotated = ReceiptKeys::new(2, [(1, [0x41; 32]), (2, [0x43; 32])])?; + let restored = ingress.restore_receipt(&rotated, &token)?; + assert_eq!(restored.position(), &position); + assert_eq!(restored.meta(), &meta); + let new_token = rotated.encode(&restored)?; + assert!(ingress.restore_receipt(&keys, &new_token).is_err()); + assert!(ingress.restore_receipt(&rotated, &new_token).is_ok()); + let retired = ReceiptKeys::new(2, [(2, [0x43; 32])])?; + assert!(ingress.restore_receipt(&retired, &token).is_err()); + let producer = ingress.open_producer(open).await?; + let prepared = ingress + .prepare( + ingress.reserve_upload(body.len()).await?, + body.clone(), + &producer, + 1, + ) + .await?; + let retry = ingress.append(&prepared).await?; + assert_eq!(retry.position(), &position); + for node in 1..=3 { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let read = c + .client + .request( + node, + Request::Fetch { + group: position.group, + index: position.index, + }, + ) + .await?; + if let Reply::Record(Some(found)) = &read.reply { + assert_eq!(found, &meta); + assert_eq!(read.body, body); + break Ok::<_, anyhow::Error>(()); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await??; + } + assert_eq!( + ingress.publication_status(&restored).await?, + PublicationStatus::Pending + ); + ingress.close_producer(&producer, Some(&retry)).await?; + // Closing fences new writes while preserving reconciliation of the last + // accepted tuple. Otherwise a lost final ACK could never be reconciled. + assert_eq!(ingress.append(&prepared).await?.position(), &position); + let next = ingress + .prepare( + ingress.reserve_upload(body.len()).await?, + body, + &producer, + 2, + ) + .await?; + let error = ingress.append(&next).await.unwrap_err(); + assert!(matches!( + error.downcast_ref::(), + Some(AppendFailure::ProducerFenced) + )); + c.shutdown().await +} diff --git a/crates/lance-context-ingest/src/tests/network.rs b/crates/lance-context-ingest/src/tests/network.rs new file mode 100644 index 00000000..083f76f6 --- /dev/null +++ b/crates/lance-context-ingest/src/tests/network.rs @@ -0,0 +1,187 @@ +use super::*; +use lance_context_wal::{ + raft_log::DiskOptions, + replication::{ + runtime::{self, client, peer, Runtime}, + Limits, Node, + }, +}; +use std::{ + collections::{BTreeMap, BTreeSet}, + path::Path, +}; +use tokio::net::TcpListener; +use tokio_rustls::rustls::{ + self, + pki_types::{CertificateDer, PrivatePkcs8KeyDer}, + RootCertStore, +}; + +struct Tls { + client: rustls::ClientConfig, + server: rustls::ServerConfig, + certificate: [u8; 32], + name: String, +} +fn certificates() -> Result> { + let mut identities = Vec::new(); + let mut roots = RootCertStore::empty(); + for id in 1..=4 { + let name = format!("wal-{id}.example.invalid"); + let certified = rcgen::generate_simple_self_signed(vec![name.clone()])?; + let der: CertificateDer<'static> = certified.cert.der().clone(); + roots.add(der.clone())?; + identities.push((id, name, der, certified.signing_key.serialize_der())); + } + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let roots = Arc::new(roots); + let mut configs = BTreeMap::new(); + for (id, name, cert, key) in identities { + let certificate = ring::digest::digest(&ring::digest::SHA256, cert.as_ref()) + .as_ref() + .try_into()?; + let verifier = rustls::server::WebPkiClientVerifier::builder_with_provider( + roots.clone(), + provider.clone(), + ) + .build()?; + let server = rustls::ServerConfig::builder_with_provider(provider.clone()) + .with_safe_default_protocol_versions()? + .with_client_cert_verifier(verifier) + .with_single_cert( + vec![cert.clone()], + PrivatePkcs8KeyDer::from(key.clone()).into(), + )?; + let client = rustls::ClientConfig::builder_with_provider(provider.clone()) + .with_safe_default_protocol_versions()? + .with_root_certificates(roots.clone()) + .with_client_auth_cert(vec![cert], PrivatePkcs8KeyDer::from(key).into())?; + configs.insert( + id, + Tls { + client, + server, + certificate, + name, + }, + ); + } + Ok(configs) +} + +pub(super) struct Cluster { + pub client: Arc, + nodes: Vec<(Runtime, peer::Transport, client::Server)>, +} +impl Cluster { + pub async fn open(root: &Path, layout: Arc) -> Result { + let tls = certificates()?; + let limits = Limits { + producer_slots: 16, + ..Limits::default() + }; + let mut peers = BTreeMap::new(); + let mut listeners = Vec::new(); + let mut endpoints = BTreeMap::new(); + for id in 1..=3 { + let peer = TcpListener::bind("127.0.0.1:0").await?; + peers.insert( + id, + peer::Peer { + address: peer.local_addr()?.to_string(), + server_name: tls[&id].name.clone(), + certificate: tls[&id].certificate, + }, + ); + let data = TcpListener::bind("127.0.0.1:0").await?; + let control = TcpListener::bind("127.0.0.1:0").await?; + endpoints.insert( + id, + client::Endpoint { + data_address: data.local_addr()?.to_string(), + control_address: control.local_addr()?.to_string(), + server_name: tls[&id].name.clone(), + certificate: tls[&id].certificate, + }, + ); + listeners.push((id, peer, data, control)); + } + let mut nodes = Vec::new(); + for (id, listener, data, control) in listeners { + let node = Node::open( + &root.join(id.to_string()), + layout.clone(), + id, + limits.clone(), + )? + .with_disk_budget(DiskOptions { + max_bytes: 512 << 20, + data_bytes: 128 << 20, + headroom_bytes: 256 << 20, + })?; + let (runtime, outbox) = Runtime::spawn(node, runtime::Options::default())?; + let mut server = tls[&id].server.clone(); + server.alpn_protocols = vec![peer::ALPN.to_vec()]; + let mut client = tls[&id].client.clone(); + client.alpn_protocols = vec![peer::ALPN.to_vec()]; + let transport = peer::Transport::spawn( + runtime.handle(), + outbox, + listener, + peers + .iter() + .filter(|(n, _)| **n != id) + .map(|(n, p)| (*n, p.clone())) + .collect(), + Arc::new(client), + Arc::new(server), + peer::Options::default(), + )?; + let mut server = tls[&id].server.clone(); + server.alpn_protocols = vec![client::ALPN.to_vec()]; + let rpc = client::Server::spawn( + runtime.handle(), + data, + control, + BTreeSet::from([tls[&4].certificate]), + Arc::new(server), + client::Options::default(), + )?; + nodes.push((runtime, transport, rpc)); + } + let mut config = tls[&4].client.clone(); + config.alpn_protocols = vec![client::ALPN.to_vec()]; + let client = Arc::new(Client::new( + &layout, + &limits, + endpoints, + Arc::new(config), + client::Options::default(), + )?); + for &group in layout.groups.keys() { + nodes[0].0.handle().campaign(group)?.await?; + tokio::time::timeout(Duration::from_secs(15), async { + loop { + let leaders = nodes + .iter() + .map(|n| n.0.handle().observation(group).unwrap().status.leader) + .collect::>(); + if leaders.len() == 1 && leaders.first().is_some_and(|v| (1..=3).contains(v)) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + } + Ok(Self { client, nodes }) + } + pub async fn shutdown(self) -> Result<()> { + for (runtime, peer, rpc) in self.nodes { + rpc.shutdown().await?; + peer.shutdown().await?; + runtime.shutdown().await?; + } + Ok(()) + } +} diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs index db398f76..40883703 100644 --- a/crates/lance-context-wal/src/replication/query.rs +++ b/crates/lance-context-wal/src/replication/query.rs @@ -48,6 +48,10 @@ pub enum Query { ProducerSlot { slot: u32, }, + /// Compact table binding and publication cursors, without plans/descriptors. + Visibility { + table: Uuid, + }, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -68,6 +72,16 @@ pub enum Value { /// missing-object result nor proof that a supplied staging object lost. ManifestRetired(publication::HistoryBoundary), ProducerSlot(Option), + Visibility(Option>), +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Visibility { + pub destination: publication::Destination, + pub ingest: crate::ingest_contract::IngestContract, + pub incarnation: Option, + pub checkpoint: BTreeMap, + pub minimum_read_version: u64, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -136,7 +150,7 @@ impl Query { | Self::ClosedProof { table } => { ensure!(!table.is_nil(), "nil source table") } - Self::Publisher { table } => ensure!( + Self::Publisher { table } | Self::Visibility { table } => ensure!( !table.is_nil() && group == publication::control_group(layout), "invalid publisher query scope" ), @@ -267,6 +281,17 @@ impl Query { Self::Publisher { table } => { Value::Publisher(state.catalog.tables.get(table).cloned().map(Box::new)) } + Self::Visibility { table } => { + Value::Visibility(state.catalog.tables.get(table).map(|table| { + Box::new(Visibility { + destination: table.destination.clone(), + ingest: table.ingest.clone(), + incarnation: table.manifests.as_ref().map(|m| m.incarnation), + checkpoint: table.progress.checkpoint.clone(), + minimum_read_version: table.progress.minimum_read_version, + }) + })) + } Self::Accepted { after, .. } if *after > state.applied => Value::CursorAhead { applied: state.applied, }, diff --git a/crates/lance-context-wal/src/replication/query/tests.rs b/crates/lance-context-wal/src/replication/query/tests.rs index efda4bfa..2307bfca 100644 --- a/crates/lance-context-wal/src/replication/query/tests.rs +++ b/crates/lance-context-wal/src/replication/query/tests.rs @@ -46,6 +46,67 @@ fn evaluate(query: &Query, cut: &Cut, state: &State) -> Result { ) } +#[test] +fn compact_visibility_uses_catalog_quorum_and_fits_control_reply_budget() -> Result<()> { + let (_root, mut nodes) = nodes(Limits::default())?; + nodes[0].campaign(10)?; + nodes[0].campaign(20)?; + pump(&mut nodes)?; + let acquire = publication::Acquire { + table: Uuid::from_u128(10), + destination: publication::Destination { + namespace: Uuid::from_u128(11), + dataset: "x".repeat(4096), + }, + ingest: crate::ingest_contract::IngestContract { + tenant: "x".repeat(1024), + qualified_table: "x".repeat(1024), + schema: [3; 32], + policy: crate::ingest_contract::WritePolicy::ImmutableUniqueIds, + max_record_bytes: 1024 * 1024, + max_rows: 1024, + }, + owner: Uuid::from_u128(12), + request: Uuid::from_u128(13), + expected_epoch: 0, + }; + let query = Query::Visibility { + table: acquire.table, + }; + assert!(nodes[0].query(20, 1, query.clone()).is_err()); + nodes[0].query(10, 1, query.clone())?; + assert!(matches!( + &pump(&mut nodes)?.reads[0].outcome, + Outcome::Observed { + value: Value::Visibility(None), + .. + } + )); + nodes[0].acquire_publisher(2, acquire.clone())?; + pump(&mut nodes)?; + nodes[0].query(10, 3, query)?; + let observed = pump(&mut nodes)?; + let Outcome::Observed { + cut, + value: Value::Visibility(Some(view)), + } = &observed.reads[0].outcome + else { + anyhow::bail!("missing quorum visibility"); + }; + assert!(cut.quorum_index > 0 && cut.applied >= cut.quorum_index); + assert_eq!(view.destination, acquire.destination); + assert_eq!(view.ingest, acquire.ingest); + assert_eq!(view.incarnation, None); + assert!(view.checkpoint.is_empty()); + // Maximum configured group count, even with maximal binding strings, + // remains below the runtime's generic control reply allowance. + let mut maximum = view.clone(); + maximum.incarnation = Some(Uuid::from_u128(100)); + maximum.checkpoint = (1..=64).map(|group| (group, u64::MAX - 1)).collect(); + assert!(bincode::serialized_size(&maximum)? < 16 * 1024); + Ok(()) +} + #[test] fn new_leader_rejects_reads_until_its_first_current_term_entry_commits() -> Result<()> { let (_root, mut nodes) = nodes(Limits { diff --git a/crates/lance-context-wal/src/replication/runtime/client.rs b/crates/lance-context-wal/src/replication/runtime/client.rs index 534bc1b9..59f5105c 100644 --- a/crates/lance-context-wal/src/replication/runtime/client.rs +++ b/crates/lance-context-wal/src/replication/runtime/client.rs @@ -457,12 +457,25 @@ pub struct Client { layout: [u8; 32], policy: [u8; 32], max_record: usize, + producer_slots: u32, next: AtomicU64, voters: BTreeMap>, catalog: u64, leaders: Mutex>, } impl Client { + pub fn identity(&self) -> (uuid::Uuid, [u8; 32]) { + (self.cluster, self.layout) + } + pub fn max_record_bytes(&self) -> usize { + self.max_record + } + pub fn producer_slots(&self) -> u32 { + self.producer_slots + } + pub fn catalog_group(&self) -> u64 { + self.catalog + } pub fn new( layout: &Layout, limits: &Limits, @@ -498,6 +511,7 @@ impl Client { layout: layout.fingerprint()?, policy: Sha256::digest(limits.identity()?).into(), max_record: limits.max_record_bytes as usize, + producer_slots: limits.producer_slots, connector: TlsConnector::from(tls), budgets: Budgets::new(&options), nodes: endpoints diff --git a/docs/design/wal-rollout-ingress.md b/docs/design/wal-rollout-ingress.md new file mode 100644 index 00000000..579bc6cf --- /dev/null +++ b/docs/design/wal-rollout-ingress.md @@ -0,0 +1,77 @@ +# Durable rollout ingress + +The independent `lance-context-ingest` workspace crate binds core's fixed rollout +schema to the WAL client. An application resolves and authorizes a `TableBinding`, +then supplies a shared memory budget and a configured mutual-TLS client. Connect +checks the entire durable table contract and manifest incarnation through a +quorum read. This is an adapter for newly bound immutable-ID rollout tables; +legacy upserts, existing table migration, and table delete/recreate are separate +protocols. + +This integration crate is not published while the independent WAL remains +unpublished. The existing core crate keeps its current dependency and registry +release boundary. WAL replicas do not depend on core, Arrow decoding or Lance. + +`rollout_id` is the session routing key. An atomic request must belong to one +virtual bucket, including when different buckets share a physical replica +group. The adapter rejects mixed buckets instead of splitting a previously +atomic request. Worker count does not affect placement. Row IDs must be globally +unique and immutable by the caller's contract; validation rejects duplicates +within one atomic request but does not implement a global uniqueness index. + +## Write and retry contract + +1. Persist the entire `ProducerOpen` tuple before opening a bounded producer + slot. Reconcile the same tuple after an uncertain response. Do not allocate a + different identity merely because an RPC times out. +2. Reserve ingress workspace before buffering the body. Prepare uncompressed + Arrow IPC against the actual rollout schema, validating arrays, required + fields, row limits, row IDs and session routing. Decoding and hashing run on + the blocking executor. Consecutive rows from one session reuse its routing + check. The original reference-counted body is retained without re-encoding. +3. Persist the producer, sequence and exact body in the caller's durable retry + journal before append. Only a quorum-committed `Accepted` reply creates an + `AcceptedAppend`. An unknown result retains the same sequence and bytes; + advancing the sequence or changing the body is not reconciliation. +4. Persist the authenticated receipt before discarding the client retry body. + A publication timeout does not cancel an ACK. `publication_status` uses a + compact quorum view and the receipt's source-group checkpoint. A published + response supplies the minimum Lance version needed by a reader. +5. Close with the confirmed final receipt. The WAL preserves exact retry of the + last accepted tuple while fencing new writes. Slot reuse additionally needs + durable source retirement; closing alone does not recycle it. + +The preparation reservation follows the body through retries and cancellation. +The configured maximum request must fit the supplied shared budget. This is +logical admission, not an allocator or cgroup hard limit. Application clones of +exposed `Bytes` require their own lifetime accounting. TLS encryption and Arrow +alignment repair may copy; this is not an end-to-end zero-copy claim. + +## Restartable receipts + +`receipt::ReceiptKeys` issues authenticated, bounded tokens with no payload or +storage URI. Scope hashing includes the cluster, layout, table, destination, +incarnation and ingestion contract. Authentication precedes claim parsing, and +restoration checks all membership fields against the authorized binding. These +are trusted-service membership proofs, not user authentication or authorization. + +Keys must come from persistent trusted configuration. During rotation, install +read keys everywhere before switching issuance and retain previous keys for as +long as old receipts must remain verifiable. Receipts have no automatic expiry. +Losing a key or deliberately revoking it never authorizes deleting acknowledged +data. A token cannot replace the durable producer/body journal when the ACK was +never received. + +## Validation and integration boundary + +Integration tests use three actual TLS-connected replicas, persistent original node +directories, and a real Lance genesis manifest. They exercise durable append, +exact retry, whole-cluster restart, full payload reads on every replica, receipt +restoration/rotation, mismatched bindings, bounded admission and producer close. +Preparation coverage checks same-group/different-bucket rejection, duplicate IDs +across IPC batches, row limits and original body allocation identity. + +This crate does not yet wire the server HTTP route, a standalone node command, +automatic maintenance, or the full materialization/publication loop. The tests +must not be described as a sustained ingestion benchmark or as evidence that +newly appended rows have already become visible in Lance.