diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 4ab06986..b8916625 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -30,8 +30,7 @@ env: jobs: test: runs-on: ubuntu-24.04 - # The focused HA suite builds a second feature graph after workspace tests; - # cold native dependency builds can exhaust 30 minutes before HA tests run. + # Keep workspace tests independent of the optional feature build graphs. timeout-minutes: 60 strategy: matrix: @@ -60,18 +59,30 @@ jobs: # regression test was dead weight in CI. `--workspace` covers the crates # that were silently untested (api, server, client, metrics, the facade). run: cargo test --workspace --all-targets + + wal-tests: + runs-on: ubuntu-24.04 + timeout-minutes: 60 + steps: + - uses: actions/checkout@v4 + - name: Setup rust toolchain + run: | + rustup toolchain install stable + rustup default stable + - uses: Swatinem/rust-cache@v2 + with: + shared-key: "lance-context-wal-deps" + workspaces: "." + - name: Install dependencies + run: | + sudo apt update + sudo apt install -y clang libclang-dev protobuf-compiler - name: WAL component correctness smoke tests run: | cargo run -p lance-context-wal --features lab --bin context-wal-lab -- codec --payload-bytes 1024 --rows 8 --iterations 2 cargo test -p lance-context-wal --features lab --bin context-wal-lab cargo test -p lance-context-wal --features lab --test engine_cache 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: Optional Lance object-store metrics - run: | - cargo test -p lance-context-core --features lance-io-metrics --test lance_io_metrics - cargo test -p lance-context-core --all-features --lib latency_histograms_carry_only_bounded_labels - cargo check -p lance-context-server --features lance-io-metrics - cargo check -p lance-context-core --no-default-features - name: WAL placement, admission, record, contract and IPC decoding tests run: | cargo test -p lance-context-wal --features placement,admission,ipc,record,arrow-contract,vectored-io --lib @@ -90,10 +101,6 @@ jobs: run: | cargo clippy -p lance-context-wal --features peer-transport --all-targets -- -D warnings cargo test -p lance-context-wal --features peer-transport --lib --test replication_process - - name: Prepared WAL descriptors with real Lance files and process recovery - run: | - cargo test -p lance-context-wal --features lance-integration --lib replication::publication::lance - cargo test -p lance-context-wal --features lance-integration --test replication_process lance_publication - name: WAL resource observation in a container cgroup namespace run: | cargo test -p lance-context-wal --features runner --test resource --no-run --message-format=json > "$RUNNER_TEMP/wal-resource-build.jsonl" @@ -105,6 +112,52 @@ jobs: mount -t cgroup2 cgroup2 /sys/fs/cgroup exec setpriv --reuid "$1" --regid "$2" --clear-groups --no-new-privs "$3" ' sh "$(id -u)" "$(id -g)" "$wal_resource_test" + + lance-tests: + runs-on: ubuntu-24.04 + timeout-minutes: 60 + steps: + - uses: actions/checkout@v4 + - name: Setup rust toolchain + run: | + rustup toolchain install stable + rustup default stable + - uses: Swatinem/rust-cache@v2 + with: + shared-key: "lance-context-lance-deps" + workspaces: "." + - name: Install dependencies + run: | + sudo apt update + sudo apt install -y clang libclang-dev protobuf-compiler + - name: Optional Lance object-store metrics + run: | + cargo test -p lance-context-core --features lance-io-metrics --test lance_io_metrics + cargo test -p lance-context-core --all-features --lib latency_histograms_carry_only_bounded_labels + cargo check -p lance-context-server --features lance-io-metrics + cargo check -p lance-context-core --no-default-features + - name: Prepared WAL descriptors with real Lance files and process recovery + run: | + cargo test -p lance-context-wal --features lance-integration --lib replication::publication::lance + cargo test -p lance-context-wal --features lance-integration --test replication_process lance_publication + + ha-tests: + runs-on: ubuntu-24.04 + timeout-minutes: 60 + steps: + - uses: actions/checkout@v4 + - name: Setup rust toolchain + run: | + rustup toolchain install stable + rustup default stable + - uses: Swatinem/rust-cache@v2 + with: + shared-key: "lance-context-ha-deps" + workspaces: "." + - name: Install dependencies + run: | + sudo apt update + sudo apt install -y clang libclang-dev protobuf-compiler - name: Start etcd for HA scheduler test run: | ETCD_VERSION=3.7.0 diff --git a/crates/lance-context-wal/src/replication/runtime.rs b/crates/lance-context-wal/src/replication/runtime.rs index 50df6f42..76d0a3d9 100644 --- a/crates/lance-context-wal/src/replication/runtime.rs +++ b/crates/lance-context-wal/src/replication/runtime.rs @@ -13,7 +13,7 @@ use super::{producer, publication as p, query, source, Node, Outcome, Packet, St use crate::record::{Batch, RecordKind}; use anyhow::{ensure, Context, Result}; use protobuf::Message as _; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use std::{ collections::BTreeMap, future::Future, @@ -28,6 +28,8 @@ use std::{ }; use tokio::sync::{mpsc as async_mpsc, oneshot, OwnedSemaphorePermit, Semaphore}; +#[cfg(feature = "peer-transport")] +pub mod client; mod images; #[cfg(feature = "peer-transport")] pub mod peer; @@ -85,7 +87,7 @@ impl Options { /// Typed commands preserve Node's existing validation and durable formats. /// This scheduling enum is not a network or persisted wire format. -#[derive(Debug, Serialize)] +#[derive(Clone, Debug, Serialize, Deserialize)] pub enum Command { Append { group: u64, batch: Batch }, Acquire(p::Acquire), diff --git a/crates/lance-context-wal/src/replication/runtime/client.rs b/crates/lance-context-wal/src/replication/runtime/client.rs new file mode 100644 index 00000000..534bc1b9 --- /dev/null +++ b/crates/lance-context-wal/src/replication/runtime/client.rs @@ -0,0 +1,726 @@ +//! Bounded mutual-TLS RPCs from trusted ingress/materializer services to WAL nodes. +//! +//! Separate data/control listeners and physical-node pools preserve control +//! capacity. Append metadata and opaque Bytes travel separately in one exchange; +//! only the durable runtime completion produces a commit response. This is a +//! trusted service boundary, not end-user table authorization or leader routing. +use super::{query, Command, Handle, Outcome, Pending, Work}; +use crate::{ + placement::Layout, + record::{Batch, BatchMeta}, + replication::{Limits, Status}, +}; +use anyhow::{ensure, Context, Result}; +use bincode::Options as _; +use bytes::Bytes; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::{ + collections::{BTreeMap, BTreeSet}, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, Mutex, + }, + time::Duration, +}; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + sync::{watch, OwnedSemaphorePermit, Semaphore}, + task::{JoinHandle, JoinSet}, + time::timeout, +}; +use tokio_rustls::{ + rustls::{self, pki_types::ServerName}, + TlsAcceptor, TlsConnector, +}; + +pub const ALPN: &[u8] = b"lance-context-wal-client/1"; +const MAGIC: &[u8; 8] = b"LCWCLNT1"; +const MAX_META: usize = 2 << 20; +const PREFIX: usize = 24; + +#[derive(Clone, Debug)] +pub struct Endpoint { + pub data_address: String, + pub control_address: String, + pub server_name: String, + /// SHA-256 of the configured service's DER leaf certificate. + pub certificate: [u8; 32], +} +#[derive(Clone, Debug)] +pub struct Options { + /// Logical live request/response bytes, independent for the two lanes. + /// Caller allocations and allocator/decoded metadata overhead are separate. + pub data_bytes: usize, + pub control_bytes: usize, + /// Per node AND lane in one client's physical connection pool. + pub connections: usize, + /// Per server listener, shared across all independently running clients. + pub server_connections: usize, + /// Entire exchange, including pool/admission waits and durable completion. + pub timeout: Duration, +} +impl Default for Options { + fn default() -> Self { + Self { + data_bytes: 64 << 20, + control_bytes: 32 << 20, + connections: 4, + server_connections: 64, + timeout: Duration::from_secs(30), + } + } +} +impl Options { + fn validate(&self, max_record: usize) -> Result<()> { + ensure!( + (1..=32).contains(&self.connections) + && (1..=256).contains(&self.server_connections) + && (2 * max_record + 4 * MAX_META..=256 << 20).contains(&self.data_bytes) + && (4 * MAX_META..=256 << 20).contains(&self.control_bytes) + && (Duration::from_millis(100)..=Duration::from_secs(120)).contains(&self.timeout), + "invalid client transport limits" + ); + Ok(()) + } +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +struct Scope { + cluster: uuid::Uuid, + layout: [u8; 32], + policy: [u8; 32], + node: u64, +} +#[derive(Clone, Debug)] +pub enum Request { + Append { + group: u64, + batch: Batch, + }, + Commit(Box), + Query { + group: u64, + query: query::Query, + }, + /// Local committed data only. Establish a quorum source pin first; None is + /// not evidence that an uncertain append was rejected or never committed. + Fetch { + group: u64, + index: u64, + }, + /// Cached local leadership hint, never a quorum result. + Status { + group: u64, + }, +} +#[derive(Debug, Serialize, Deserialize)] +enum WireRequest { + Append { group: u64, meta: BatchMeta }, + Control(Box), + Query { group: u64, query: query::Query }, + Fetch { group: u64, index: u64 }, + Status { group: u64 }, +} +impl WireRequest { + fn data(&self) -> bool { + matches!(self, Self::Append { .. } | Self::Fetch { .. }) + } +} +#[derive(Debug, Serialize, Deserialize)] +pub enum Reply { + Commit(Outcome), + Query(query::Outcome), + Record(Option), + Status(Status), + /// Reconcile the exact original identity. This does not prove non-commit. + Unknown, + /// No command was submitted by this handler; leader is only a local hint. + Redirect { + leader: u64, + }, +} +/// Holds receive credit through consumption of metadata and a fetched payload. +/// Cloning body Bytes transfers shared ownership, so callers retaining clones +/// must provide their own application memory budget before dropping this owner. +pub struct Response { + pub reply: Reply, + pub body: Bytes, + _credit: OwnedSemaphorePermit, +} +struct Budgets { + data: Arc, + control: Arc, +} +impl Budgets { + fn new(options: &Options) -> Self { + Self { + data: Arc::new(Semaphore::new(options.data_bytes)), + control: Arc::new(Semaphore::new(options.control_bytes)), + } + } + async fn reserve(&self, data: bool, bytes: usize) -> Result { + Ok(if data { &self.data } else { &self.control } + .clone() + .acquire_many_owned(u32::try_from(bytes)?) + .await?) + } +} +fn codec() -> impl bincode::Options { + bincode::DefaultOptions::new() + .with_fixint_encoding() + .with_limit(MAX_META as u64) + .reject_trailing_bytes() +} +fn certificate(certs: Option<&[rustls::pki_types::CertificateDer<'_>]>) -> Result<[u8; 32]> { + Ok(Sha256::digest( + certs + .and_then(|c| c.first()) + .context("client TLS certificate missing")? + .as_ref(), + ) + .into()) +} +fn scope(handle: &Handle) -> Scope { + Scope { + cluster: handle.cluster, + layout: handle.layout, + policy: handle.policy, + node: handle.id, + } +} +async fn prefix(io: &mut R, max_body: usize) -> Result<(u64, usize, usize)> { + let mut fixed = [0; PREFIX]; + io.read_exact(&mut fixed).await?; + ensure!(&fixed[..8] == MAGIC, "foreign client protocol"); + let id = u64::from_le_bytes(fixed[8..16].try_into()?); + let meta = u32::from_le_bytes(fixed[16..20].try_into()?) as usize; + let body = u32::from_le_bytes(fixed[20..24].try_into()?) as usize; + ensure!( + id > 0 && (1..=MAX_META).contains(&meta) && body <= max_body, + "invalid client frame bounds" + ); + Ok((id, meta, body)) +} +async fn receive(io: &mut R, bytes: usize) -> Result { + let mut body = vec![0; bytes]; + io.read_exact(&mut body).await?; + Ok(body.into()) +} +async fn send( + io: &mut W, + id: u64, + metadata: &[u8], + body: &[u8], +) -> Result<()> { + ensure!( + (1..=MAX_META).contains(&metadata.len()), + "oversized client metadata" + ); + let mut fixed = [0; PREFIX]; + fixed[..8].copy_from_slice(MAGIC); + fixed[8..16].copy_from_slice(&id.to_le_bytes()); + fixed[16..20].copy_from_slice(&u32::try_from(metadata.len())?.to_le_bytes()); + fixed[20..24].copy_from_slice(&u32::try_from(body.len())?.to_le_bytes()); + // TLS owns its encrypted output buffers; no whole-payload concatenation. + io.write_all(&fixed).await?; + io.write_all(metadata).await?; + io.write_all(body).await?; + io.flush().await?; + Ok(()) +} + +impl Handle { + /// Read on the owner thread with data-lane credit, separate from small + /// control replies. Returned bytes describe local committed state only. + pub fn read_accepted(&self, group: u64, index: u64) -> Result>> { + ensure!(index > 0 && index < u64::MAX, "invalid accepted index"); + let route = self.route(group)?; + let credit = self.shared.data.reserve(self.max_record + 1024)?; + let (reply, receive) = tokio::sync::oneshot::channel(); + let completion = super::Completion { + reply, + _credit: credit, + }; + self.enqueue( + route, + Work::Local(Box::new(move |node| { + let result = node.read_accepted(group, index); + Box::new(move || completion.finish(result)) + })), + true, + )?; + Ok(Pending(receive)) + } +} + +pub struct Server { + stop: watch::Sender, + tasks: Vec>>, +} +impl Server { + /// Trusted service certificates may access this node's configured groups. + /// End-user authorization and table binding stay at the application ingress. + pub fn spawn( + handle: Handle, + data: TcpListener, + control: TcpListener, + clients: BTreeSet<[u8; 32]>, + tls: Arc, + options: Options, + ) -> Result { + options.validate(handle.max_record)?; + ensure!( + handle.disk_budget().is_some(), + "client listener requires node disk admission" + ); + ensure!( + !clients.is_empty() && clients.len() <= 256 && !clients.contains(&[0; 32]), + "invalid trusted client pins" + ); + ensure!( + tls.alpn_protocols.iter().any(|p| p == ALPN), + "missing client RPC ALPN" + ); + let (stop, shutdown) = watch::channel(false); + let budgets = Arc::new(Budgets::new(&options)); + let clients = Arc::new(clients); + let mut tasks = Vec::new(); + for (listener, lane) in [(data, true), (control, false)] { + tasks.push(tokio::spawn(listen( + listener, + lane, + handle.clone(), + clients.clone(), + TlsAcceptor::from(tls.clone()), + budgets.clone(), + options.clone(), + shutdown.clone(), + ))); + } + Ok(Self { stop, tasks }) + } + pub async fn shutdown(mut self) -> Result<()> { + let _ = self.stop.send(true); + let mut error = None; + for task in std::mem::take(&mut self.tasks) { + match task.await { + Ok(Ok(())) => {} + Ok(Err(e)) => error = Some(e), + Err(e) => error = Some(e.into()), + } + } + error.map_or(Ok(()), Err) + } +} +impl Drop for Server { + fn drop(&mut self) { + let _ = self.stop.send(true); + } +} +#[allow(clippy::too_many_arguments)] +async fn listen( + listener: TcpListener, + data: bool, + handle: Handle, + clients: Arc>, + acceptor: TlsAcceptor, + budgets: Arc, + options: Options, + mut stop: watch::Receiver, +) -> Result<()> { + let connections = Arc::new(Semaphore::new(options.server_connections)); + let mut tasks = JoinSet::new(); + loop { + tokio::select! { + _ = stop.changed() => break, + result = tasks.join_next(), if !tasks.is_empty() => { if let Some(Err(error)) = result { if error.is_panic() { return Err(error.into()); } } }, + incoming = listener.accept() => { + let (tcp, _) = incoming?; + let Ok(permit) = connections.clone().try_acquire_owned() else { continue; }; + let handle = handle.clone(); let clients = clients.clone(); let acceptor = acceptor.clone(); let budgets = budgets.clone(); let options = options.clone(); + tasks.spawn(async move { + let _permit = permit; + let connection = async { + tcp.set_nodelay(true)?; let stream = acceptor.accept(tcp).await?; + ensure!(stream.get_ref().1.alpn_protocol() == Some(ALPN) && clients.contains(&certificate(stream.get_ref().1.peer_certificates())?), "unconfigured client identity"); + // Handshake is independently bounded; each exchange has its own deadline. + Ok::<_, anyhow::Error>(stream) + }; + if let Ok(Ok(mut stream)) = timeout(options.timeout, connection).await { + loop { if !matches!(timeout(options.timeout, serve_one(&mut stream, data, &handle, &budgets)).await, Ok(Ok(()))) { break; } } + } + }); + } + } + } + tasks.abort_all(); + while tasks.join_next().await.is_some() {} + Ok(()) +} +async fn serve_one( + io: &mut S, + data: bool, + handle: &Handle, + budgets: &Budgets, +) -> Result<()> { + let (id, meta_len, body_len) = prefix(io, if data { handle.max_record } else { 0 }).await?; + // Reserve reply capacity too, before reading any variable-length allocation. + let _credit = budgets + .reserve( + data, + 2 * meta_len + body_len + 2 * MAX_META + if data { handle.max_record } else { 0 }, + ) + .await?; + let metadata = receive(io, meta_len).await?; + let (request_scope, request): (Scope, WireRequest) = codec().deserialize(&metadata)?; + ensure!( + request_scope == scope(handle) && request.data() == data, + "foreign request scope/lane" + ); + if let WireRequest::Append { meta, .. } = &request { + meta.check_body_length(body_len, handle.max_record)?; + } else { + ensure!(body_len == 0, "unexpected request payload"); + } + let body = receive(io, body_len).await?; + let executed = execute(handle, request, body).await; + let (reply, body) = executed.unwrap_or((Reply::Unknown, Bytes::new())); + let metadata = codec().serialize(&(scope(handle), reply))?; + send(io, id, &metadata, &body).await +} +async fn execute(handle: &Handle, request: WireRequest, body: Bytes) -> Result<(Reply, Bytes)> { + let group = match &request { + WireRequest::Append { group, .. } | WireRequest::Query { group, .. } => Some(*group), + WireRequest::Control(command) => Some(command.group(handle.catalog)), + _ => None, + }; + if let Some(group) = group { + let leader = handle.observation(group)?.status.leader; + if leader != handle.id { + return Ok((Reply::Redirect { leader }, Bytes::new())); + } + } + let mut output = Bytes::new(); + let reply = match request { + WireRequest::Append { group, meta } => Reply::Commit( + handle + .commit(Command::Append { + group, + batch: Batch { meta, body }, + })? + .await?, + ), + WireRequest::Control(command) => { + ensure!( + !matches!(*command, Command::Append { .. }), + "append on control lane" + ); + Reply::Commit(handle.commit(*command)?.await?) + } + WireRequest::Query { group, query } => Reply::Query(handle.query(group, query)?.await?), + WireRequest::Fetch { group, index } => { + Reply::Record(handle.read_accepted(group, index)?.await?.map(|batch| { + output = batch.body; + batch.meta + })) + } + WireRequest::Status { group } => Reply::Status(handle.observation(group)?.status), + }; + Ok((reply, output)) +} + +type Stream = tokio_rustls::client::TlsStream; +struct Pool { + idle: Mutex>, + slots: Arc, +} +impl Pool { + fn new(count: usize) -> Self { + Self { + idle: Mutex::new(Vec::new()), + slots: Arc::new(Semaphore::new(count)), + } + } +} +struct NodePools { + endpoint: Endpoint, + data: Pool, + control: Pool, +} +pub struct Client { + nodes: BTreeMap, + connector: TlsConnector, + options: Options, + budgets: Budgets, + cluster: uuid::Uuid, + layout: [u8; 32], + policy: [u8; 32], + max_record: usize, + next: AtomicU64, + voters: BTreeMap>, + catalog: u64, + leaders: Mutex>, +} +impl Client { + pub fn new( + layout: &Layout, + limits: &Limits, + endpoints: BTreeMap, + tls: Arc, + options: Options, + ) -> Result { + options.validate(limits.max_record_bytes as usize)?; + ensure!( + tls.alpn_protocols.iter().any(|p| p == ALPN), + "missing client RPC ALPN" + ); + ensure!( + endpoints.keys().eq(layout.nodes.keys()) + && endpoints.len() * options.connections * 2 <= 1024, + "client endpoints must cover exactly the bounded layout" + ); + let mut pins = BTreeSet::new(); + for endpoint in endpoints.values() { + ensure!( + !endpoint.data_address.is_empty() + && endpoint.data_address.len() <= 1024 + && !endpoint.control_address.is_empty() + && endpoint.control_address.len() <= 1024 + && endpoint.certificate != [0; 32] + && pins.insert(endpoint.certificate), + "invalid/duplicate client endpoint identity" + ); + ServerName::try_from(endpoint.server_name.clone())?; + } + Ok(Self { + cluster: layout.cluster, + layout: layout.fingerprint()?, + policy: Sha256::digest(limits.identity()?).into(), + max_record: limits.max_record_bytes as usize, + connector: TlsConnector::from(tls), + budgets: Budgets::new(&options), + nodes: endpoints + .into_iter() + .map(|(id, endpoint)| { + ( + id, + NodePools { + endpoint, + data: Pool::new(options.connections), + control: Pool::new(options.connections), + }, + ) + }) + .collect(), + options, + next: AtomicU64::new(1), + voters: layout.groups.clone(), + catalog: super::p::control_group(layout), + leaders: Mutex::new(BTreeMap::new()), + }) + } + /// A failure/cancellation is an unknown outcome. Retain and reconcile the + /// original producer sequence/control identity. This method never silently + /// retries, advances a sequence, or treats a cached leader hint as an ACK. + pub async fn request(&self, node: u64, request: Request) -> Result { + timeout(self.options.timeout, self.exchange(node, request)) + .await + .context("client exchange timed out; outcome unknown")? + } + /// Follow configured-voter hints only when the remote confirms it did not + /// submit the request. Unknown outcomes are returned for explicit exact retry. + pub async fn request_routed(&self, request: Request) -> Result { + timeout(self.options.timeout, async { + let group = match &request { + Request::Append { group, .. } + | Request::Query { group, .. } + | Request::Fetch { group, .. } + | Request::Status { group } => *group, + Request::Commit(command) => command.group(self.catalog), + }; + let voters = self.voters.get(&group).context("unknown client group")?; + let mut node = self + .leaders + .lock() + .unwrap() + .get(&group) + .copied() + .unwrap_or(voters[0]); + let mut visited = BTreeSet::new(); + for _ in 0..voters.len() { + visited.insert(node); + let response = match self.exchange(node, request.clone()).await { + Ok(response) => response, + Err(error) => { + let next = voters[(voters.iter().position(|id| *id == node).unwrap() + 1) + % voters.len()]; + self.leaders.lock().unwrap().insert(group, next); + return Err(error); + } + }; + if let Reply::Redirect { leader } = response.reply { + node = if voters.contains(&leader) && !visited.contains(&leader) { + leader + } else { + *voters + .iter() + .find(|n| !visited.contains(n)) + .context("leader hint loop")? + }; + self.leaders.lock().unwrap().insert(group, node); + } else { + if matches!( + response.reply, + Reply::Unknown + | Reply::Commit(Outcome::Unknown) + | Reply::Query(query::Outcome::Unknown) + ) { + let next = voters[(voters.iter().position(|id| *id == node).unwrap() + 1) + % voters.len()]; + self.leaders.lock().unwrap().insert(group, next); + } else if matches!( + request, + Request::Append { .. } | Request::Commit(_) | Request::Query { .. } + ) { + self.leaders.lock().unwrap().insert(group, node); + } + return Ok(response); + } + } + anyhow::bail!("leader not resolved; retain the original request identity") + }) + .await + .context("routed exchange timed out; outcome unknown")? + } + async fn exchange(&self, node: u64, request: Request) -> Result { + let pools = self.nodes.get(&node).context("unknown client target")?; + let (wire, body) = match request { + Request::Append { group, batch } => { + batch + .meta + .check_body_length(batch.body.len(), self.max_record)?; + ( + WireRequest::Append { + group, + meta: batch.meta, + }, + batch.body, + ) + } + Request::Commit(command) => { + ensure!( + !matches!(*command, Command::Append { .. }), + "use the append request for data" + ); + (WireRequest::Control(command), Bytes::new()) + } + Request::Query { group, query } => (WireRequest::Query { group, query }, Bytes::new()), + Request::Fetch { group, index } => (WireRequest::Fetch { group, index }, Bytes::new()), + Request::Status { group } => (WireRequest::Status { group }, Bytes::new()), + }; + let data = wire.data(); + let fetch = matches!(wire, WireRequest::Fetch { .. }); + let reply_kind = match &wire { + WireRequest::Append { .. } | WireRequest::Control(_) => 0, + WireRequest::Query { .. } => 1, + WireRequest::Fetch { .. } => 2, + WireRequest::Status { .. } => 3, + }; + let expected = Scope { + cluster: self.cluster, + layout: self.layout, + policy: self.policy, + node, + }; + let size = usize::try_from(codec().serialized_size(&(&expected, &wire))?)?; + ensure!(size <= MAX_META, "client metadata too large"); + let credit = self + .budgets + .reserve( + data, + 2 * size + body.len() + 2 * MAX_META + if fetch { self.max_record } else { 0 }, + ) + .await?; + let metadata = codec().serialize(&(&expected, wire))?; + let pool = if data { &pools.data } else { &pools.control }; + let _slot = pool.slots.clone().acquire_owned().await?; + let idle = pool.idle.lock().unwrap().pop(); + let mut stream = match idle { + Some(stream) => stream, + None => { + let address = if data { + &pools.endpoint.data_address + } else { + &pools.endpoint.control_address + }; + let tcp = TcpStream::connect(address).await?; + tcp.set_nodelay(true)?; + let stream = self + .connector + .connect( + ServerName::try_from(pools.endpoint.server_name.clone())?, + tcp, + ) + .await?; + ensure!( + stream.get_ref().1.alpn_protocol() == Some(ALPN) + && certificate(stream.get_ref().1.peer_certificates())? + == pools.endpoint.certificate, + "foreign WAL service identity" + ); + stream + } + }; + let id = self + .next + .try_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1)) + .map_err(|_| anyhow::anyhow!("client request ids exhausted"))?; + send(&mut stream, id, &metadata, &body).await?; + let (received_id, meta_len, body_len) = + prefix(&mut stream, if fetch { self.max_record } else { 0 }).await?; + ensure!(received_id == id, "client response correlation mismatch"); + let metadata = receive(&mut stream, meta_len).await?; + let (observed, reply): (Scope, Reply) = codec().deserialize(&metadata)?; + ensure!(observed == expected, "foreign client response scope"); + ensure!( + matches!(reply, Reply::Unknown | Reply::Redirect { .. }) + || matches!( + (&reply, reply_kind), + (Reply::Commit(_), 0) + | (Reply::Query(_), 1) + | (Reply::Record(_), 2) + | (Reply::Status(_), 3) + ), + "unexpected response type" + ); + match &reply { + Reply::Record(Some(meta)) if fetch => { + meta.check_body_length(body_len, self.max_record)? + } + Reply::Record(None) | Reply::Unknown if fetch => { + ensure!(body_len == 0, "unexpected record payload") + } + Reply::Record(_) => anyhow::bail!("record reply for non-fetch request"), + _ => ensure!(body_len == 0 && !fetch, "unexpected response payload/type"), + } + let body = receive(&mut stream, body_len).await?; + // The durable owner already validated fetched bytes; checking corruption + // here belongs off the async executor for large records. TLS authenticates + // this transmission; consumers still match metadata against their pin. + pool.idle.lock().unwrap().push(stream); + Ok(Response { + reply, + body, + _credit: credit, + }) + } +} + +#[cfg(test)] +mod tests; + +#[cfg(feature = "lance-integration")] +mod catalog; +#[cfg(feature = "lance-integration")] +pub use catalog::Catalog; diff --git a/crates/lance-context-wal/src/replication/runtime/client/catalog.rs b/crates/lance-context-wal/src/replication/runtime/client/catalog.rs new file mode 100644 index 00000000..9fa9616e --- /dev/null +++ b/crates/lance-context-wal/src/replication/runtime/client/catalog.rs @@ -0,0 +1,54 @@ +//! Network-backed catalog for the real Lance manifest handler. +use super::*; +use crate::replication::publication::{lance::ManifestCatalog, ReserveManifest}; + +/// Binds ManagedManifest to the same physical-node pools used by ingestion. +/// Resolving/authorizing the table's storage namespace remains the caller's job. +pub struct Catalog { + client: Arc, +} +impl std::fmt::Debug for Catalog { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WalCatalog") + .field("cluster", &self.client.cluster) + .field("group", &self.client.catalog) + .finish_non_exhaustive() + } +} +impl Client { + pub fn manifest_catalog(self: &Arc) -> Catalog { + Catalog { + client: self.clone(), + } + } +} +#[async_trait::async_trait] +impl ManifestCatalog for Catalog { + async fn read_manifest( + &self, + table: uuid::Uuid, + version: Option, + ) -> Result { + let response = self + .client + .request_routed(Request::Query { + group: self.client.catalog, + query: query::Query::Manifest { table, version }, + }) + .await?; + match &response.reply { + Reply::Query(value) => Ok(value.clone()), + _ => anyhow::bail!("manifest query outcome unknown"), + } + } + async fn reserve_manifest(&self, request: ReserveManifest) -> Result { + let response = self + .client + .request_routed(Request::Commit(Box::new(Command::Reserve(request)))) + .await?; + match &response.reply { + Reply::Commit(value) => Ok(value.clone()), + _ => anyhow::bail!("manifest reservation outcome unknown"), + } + } +} diff --git a/crates/lance-context-wal/src/replication/runtime/client/tests.rs b/crates/lance-context-wal/src/replication/runtime/client/tests.rs new file mode 100644 index 00000000..e0039fe5 --- /dev/null +++ b/crates/lance-context-wal/src/replication/runtime/client/tests.rs @@ -0,0 +1,432 @@ +use super::*; +use crate::replication::{ + runtime::peer::tests::Cluster, + tests::{batch, layout}, + Applied, +}; +use tokio::time::sleep; + +struct RpcCluster { + net: Cluster, + servers: Vec, + client: Arc, + endpoints: BTreeMap, +} +impl RpcCluster { + async fn open(root: &std::path::Path, options: Options) -> Result { + let net = Cluster::open(root).await?; + let mut endpoints = BTreeMap::new(); + let mut servers = Vec::new(); + for id in 1..=3 { + let data = TcpListener::bind("127.0.0.1:0").await?; + let control = TcpListener::bind("127.0.0.1:0").await?; + endpoints.insert( + id, + Endpoint { + data_address: data.local_addr()?.to_string(), + control_address: control.local_addr()?.to_string(), + server_name: net.tls[&id].name.clone(), + certificate: net.tls[&id].certificate, + }, + ); + let mut config = (*net.tls[&id].server).clone(); + config.alpn_protocols = vec![ALPN.to_vec()]; + servers.push(Server::spawn( + net.nodes[&id].handle.clone(), + data, + control, + BTreeSet::from([net.tls[&4].certificate]), + Arc::new(config), + options.clone(), + )?); + } + let mut config = (*net.tls[&4].client).clone(); + config.alpn_protocols = vec![ALPN.to_vec()]; + let client = Client::new( + &layout(), + &Limits::default(), + endpoints.clone(), + Arc::new(config), + options, + )?; + net.nodes[&1].handle.campaign(10)?.await?; + net.leader(10).await?; + Ok(Self { + net, + servers, + client: Arc::new(client), + endpoints, + }) + } + async fn leader(&self) -> Result { + self.net.leader(10).await + } + async fn raw(&self, node: u64, data: bool) -> Result { + let endpoint = &self.endpoints[&node]; + let tcp = TcpStream::connect(if data { + &endpoint.data_address + } else { + &endpoint.control_address + }) + .await?; + tcp.set_nodelay(true)?; + Ok(self + .client + .connector + .connect(ServerName::try_from(endpoint.server_name.clone())?, tcp) + .await?) + } + async fn shutdown(self) -> Result<()> { + for server in self.servers { + server.shutdown().await?; + } + self.net.shutdown().await + } +} +fn accepted(response: &Response) -> u64 { + match response.reply { + Reply::Commit(Outcome::Committed(Applied::Accepted { index })) => index, + ref other => panic!("expected durable append, got {other:?}"), + } +} +async fn producer(client: &Client, leader: u64, batch: &Batch) -> Result> { + let response = client + .request( + leader, + Request::Query { + group: 10, + query: query::Query::Producer { + table: batch.meta.stream, + bucket: batch.meta.bucket, + producer: batch.meta.producer, + }, + }, + ) + .await?; + match &response.reply { + Reply::Query(query::Outcome::Observed { + value: query::Value::Producer(value), + .. + }) => Ok(value.as_ref().map(|(_, index)| *index)), + other => anyhow::bail!("expected producer quorum query, got {other:?}"), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn lost_response_retries_original_bytes_after_full_three_node_restart() -> Result<()> { + let dir = tempfile::tempdir()?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + let leader = c.leader().await?; + let base = batch(1); + let original = Batch::new( + base.meta.stream, + 0, + base.meta.producer, + 1, + Bytes::from(vec![0xa7; 2 << 20]), + ); + let mut stream = c.raw(leader, true).await?; + let request = WireRequest::Append { + group: 10, + meta: original.meta.clone(), + }; + let metadata = codec().serialize(&(scope(&c.net.nodes[&leader].handle), request))?; + send(&mut stream, 1, &metadata, &original.body).await?; + // Do not consume the reply. Observe the commit through an independent + // quorum read, then disconnect with the application outcome still unknown. + let index = timeout(Duration::from_secs(10), async { + loop { + if let Some(index) = producer(&c.client, leader, &original).await? { + return Ok::<_, anyhow::Error>(index); + } + sleep(Duration::from_millis(10)).await; + } + }) + .await??; + drop(stream); + let retry = c + .client + .request( + leader, + Request::Append { + group: 10, + batch: original.clone(), + }, + ) + .await?; + assert_eq!(accepted(&retry), index); + drop(retry); + let fetched = c + .client + .request(leader, Request::Fetch { group: 10, index }) + .await?; + assert!(matches!(&fetched.reply, Reply::Record(Some(meta)) if meta == &original.meta)); + assert_eq!(fetched.body, original.body); + drop(fetched); + c.shutdown().await?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + let leader = c.leader().await?; + let retry = c + .client + .request( + leader, + Request::Append { + group: 10, + batch: original.clone(), + }, + ) + .await?; + assert_eq!(accepted(&retry), index); + drop(retry); + for id in 1..=3 { + let fetched = timeout(Duration::from_secs(10), async { + loop { + let value = c + .client + .request(id, Request::Fetch { group: 10, index }) + .await?; + if matches!(value.reply, Reply::Record(Some(_))) { + return Ok::<_, anyhow::Error>(value); + } + sleep(Duration::from_millis(10)).await; + } + }) + .await??; + assert_eq!(fetched.body, original.body); + } + c.shutdown().await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn data_connection_saturation_preserves_control_and_rejects_oversized_frames() -> Result<()> { + let dir = tempfile::tempdir()?; + let c = RpcCluster::open( + dir.path(), + Options { + connections: 1, + server_connections: 1, + ..Options::default() + }, + ) + .await?; + let leader = c.leader().await?; + let mut stalled = c.raw(leader, true).await?; + let response = c + .client + .request( + leader, + Request::Query { + group: 10, + query: query::Query::Accepted { + after: 0, + limit: 16, + }, + }, + ) + .await?; + assert!(matches!( + response.reply, + Reply::Query(query::Outcome::Observed { .. }) + )); + drop(response); + assert!(c + .client + .request( + leader, + Request::Append { + group: 10, + batch: batch(1) + } + ) + .await + .is_err()); + let mut fixed = [0u8; PREFIX]; + fixed[..8].copy_from_slice(MAGIC); + fixed[8..16].copy_from_slice(&1u64.to_le_bytes()); + fixed[16..20].copy_from_slice(&((MAX_META + 1) as u32).to_le_bytes()); + stalled.write_all(&fixed).await?; + stalled.flush().await?; + let mut byte = [0]; + let closed = timeout(Duration::from_secs(5), stalled.read(&mut byte)).await?; + assert!(matches!(closed, Ok(0) | Err(_))); + drop(stalled); + assert_eq!(producer(&c.client, leader, &batch(1)).await?, None); + c.shutdown().await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn configured_tls_identity_and_scope_are_required_before_runtime_admission() -> Result<()> { + let dir = tempfile::tempdir()?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + let leader = c.leader().await?; + // Trusted by the CA but not in the dedicated client service allowlist. + let mut config = (*c.net.tls[&1].client).clone(); + config.alpn_protocols = vec![ALPN.to_vec()]; + let foreign = Client::new( + &layout(), + &Limits::default(), + c.endpoints.clone(), + Arc::new(config), + Options::default(), + )?; + assert!(foreign + .request( + leader, + Request::Append { + group: 10, + batch: batch(1) + } + ) + .await + .is_err()); + let mut stream = c.raw(leader, true).await?; + let mut wrong = scope(&c.net.nodes[&leader].handle); + wrong.cluster = uuid::Uuid::new_v4(); + let original = batch(1); + let metadata = codec().serialize(&( + wrong, + WireRequest::Append { + group: 10, + meta: original.meta.clone(), + }, + ))?; + send(&mut stream, 1, &metadata, &original.body).await?; + assert!(timeout(Duration::from_secs(5), prefix(&mut stream, 0)) + .await? + .is_err()); + assert_eq!(producer(&c.client, leader, &original).await?, None); + c.shutdown().await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn no_quorum_cannot_return_a_durable_append_ack() -> Result<()> { + let dir = tempfile::tempdir()?; + let mut c = RpcCluster::open(dir.path(), Options::default()).await?; + let leader = c.leader().await?; + for id in (1..=3).filter(|id| *id != leader) { + c.net.stop(id).await?; + } + let result = timeout( + Duration::from_millis(500), + c.client.request( + leader, + Request::Append { + group: 10, + batch: batch(1), + }, + ), + ) + .await; + assert!(!matches!( + result, + Ok(Ok(Response { + reply: Reply::Commit(Outcome::Committed(Applied::Accepted { .. })), + .. + })) + )); + c.shutdown().await?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + let leader = c.leader().await?; + let response = c + .client + .request( + leader, + Request::Append { + group: 10, + batch: batch(1), + }, + ) + .await?; + assert!(accepted(&response) > 0); + drop(response); + c.shutdown().await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn routing_uses_only_configured_voters_and_durable_responses_hold_credit() -> Result<()> { + let dir = tempfile::tempdir()?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + let leader = c.leader().await?; + let follower = (1..=3).find(|id| *id != leader).unwrap(); + c.client.leaders.lock().unwrap().insert(10, follower); + let original = batch(1); + let result = c + .client + .request_routed(Request::Append { + group: 10, + batch: original.clone(), + }) + .await?; + let index = accepted(&result); + drop(result); + assert_eq!(c.client.leaders.lock().unwrap().get(&10), Some(&leader)); + let value = c + .client + .request_routed(Request::Fetch { group: 10, index }) + .await?; + assert_eq!(value.body, original.body); + assert!(c.client.budgets.data.available_permits() < c.client.options.data_bytes); + assert_eq!(producer(&c.client, leader, &original).await?, Some(index)); + drop(value); + assert_eq!( + c.client.budgets.data.available_permits(), + c.client.options.data_bytes + ); + c.shutdown().await +} + +#[cfg(feature = "lance-integration")] +mod lance; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn independent_ingress_pools_share_server_capacity_without_request_mixups() -> Result<()> { + let dir = tempfile::tempdir()?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + let leader = c.leader().await?; + let mut tasks = JoinSet::new(); + for _ in 0..4 { + let mut config = (*c.net.tls[&4].client).clone(); + config.alpn_protocols = vec![ALPN.to_vec()]; + let client = Arc::new(Client::new( + &layout(), + &Limits::default(), + c.endpoints.clone(), + Arc::new(config), + Options::default(), + )?); + for _ in 0..4 { + let client = client.clone(); + let id = uuid::Uuid::new_v4(); + tasks.spawn(async move { + let body = Bytes::from(vec![id.as_bytes()[0]; 256 << 10]); + let original = Batch::new(uuid::Uuid::from_u128(2), 0, id, 1, body); + let receipt = client + .request( + leader, + Request::Append { + group: 10, + batch: original.clone(), + }, + ) + .await?; + let index = accepted(&receipt); + drop(receipt); + let fetched = client + .request(leader, Request::Fetch { group: 10, index }) + .await?; + ensure!( + matches!(&fetched.reply, Reply::Record(Some(meta)) if meta == &original.meta) + && fetched.body == original.body, + "response mixed independent producer payloads" + ); + Ok::<_, anyhow::Error>(index) + }); + } + } + let mut indices = BTreeSet::new(); + while let Some(result) = tasks.join_next().await { + assert!(indices.insert(result??)); + } + assert_eq!(indices.len(), 16); + c.shutdown().await +} diff --git a/crates/lance-context-wal/src/replication/runtime/client/tests/lance.rs b/crates/lance-context-wal/src/replication/runtime/client/tests/lance.rs new file mode 100644 index 00000000..2e47eeb2 --- /dev/null +++ b/crates/lance-context-wal/src/replication/runtime/client/tests/lance.rs @@ -0,0 +1,171 @@ +use super::*; +use crate::{ + ingest_contract::{arrow_schema_fingerprint, IngestContract, WritePolicy}, + replication::publication as p, +}; +use ::lance::{ + dataset::{builder::DatasetBuilder, WriteParams}, + Dataset, +}; +use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, StringArray}; +use arrow_schema::{DataType, Field, Schema}; +use futures::TryStreamExt; +use lance_table::io::commit::{ConditionalPutCommitHandler, ManifestNamingScheme}; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn actual_lance_reader_uses_tls_catalog_after_all_wal_nodes_restart() -> Result<()> { + let dir = tempfile::tempdir()?; + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("payload", DataType::Utf8, false), + ])); + let payloads = (0..16) + .map(|id| format!("{id}:{}", "payload".repeat(8192))) + .collect::>(); + let input = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from_iter_values(0..16)), + Arc::new(StringArray::from(payloads.clone())), + ], + )?; + let uri = dir + .path() + .join("dataset.lance") + .to_str() + .unwrap() + .to_owned(); + let dataset = Dataset::write( + RecordBatchIterator::new(vec![Ok(input)], schema.clone()), + &uri, + Some(WriteParams { + commit_handler: Some(Arc::new(ConditionalPutCommitHandler)), + ..Default::default() + }), + ) + .await?; + let location = dataset.manifest_location(); + let base = object_store::path::Path::parse( + location + .path + .as_ref() + .rsplit_once("/_versions/") + .context("dataset manifest base")? + .0, + )?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + let table = uuid::Uuid::new_v4(); + let incarnation = uuid::Uuid::new_v4(); + let response = c + .client + .request_routed(Request::Commit(Box::new(Command::Acquire(p::Acquire { + table, + destination: p::Destination { + namespace: uuid::Uuid::new_v4(), + dataset: "dataset.lance".into(), + }, + ingest: IngestContract { + tenant: String::new(), + qualified_table: "table".into(), + schema: arrow_schema_fingerprint(&schema)?, + policy: WritePolicy::ImmutableUniqueIds, + max_record_bytes: 8 << 20, + max_rows: 1024, + }, + owner: uuid::Uuid::new_v4(), + request: uuid::Uuid::new_v4(), + expected_epoch: 0, + })))) + .await?; + let Reply::Commit(Outcome::Committed(Applied::Publisher(p::Reply::Acquired(authority)))) = + &response.reply + else { + anyhow::bail!("publisher acquisition failed: {:?}", response.reply); + }; + let authority = authority.clone(); + drop(response); + let response = c + .client + .request_routed(Request::Commit(Box::new(Command::Bootstrap( + p::BootstrapManifest { + authority, + incarnation, + request: uuid::Uuid::new_v4(), + version: 1, + object: p::ManifestObject { + path: format!( + "_versions/{}", + location.path.filename().context("manifest filename")? + ), + bytes: location.size.context("written manifest size")?, + v2_names: location.naming_scheme == ManifestNamingScheme::V2, + }, + transaction: uuid::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(_) + ))) + ), + "manifest was not durably registered" + ); + drop(response); + async fn read( + client: &Arc, + uri: &str, + base: &object_store::path::Path, + table: uuid::Uuid, + incarnation: uuid::Uuid, + expected: &[String], + ) -> Result<()> { + let catalog = Arc::new(client.manifest_catalog()); + let handler = p::lance::ManagedManifest::reader( + catalog, + base.clone(), + &layout(), + table, + incarnation, + )?; + let loaded = DatasetBuilder::from_uri(uri) + .with_commit_handler(Arc::new(handler)) + .load() + .await?; + assert_eq!(loaded.version().version, 1); + let batches: Vec<_> = loaded.scan().try_into_stream().await?.try_collect().await?; + let mut rows = BTreeMap::new(); + for batch in batches { + let ids = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for i in 0..batch.num_rows() { + assert!(rows + .insert(ids.value(i) as usize, values.value(i).to_owned()) + .is_none()); + } + } + assert_eq!(rows, expected.iter().cloned().enumerate().collect()); + Ok(()) + } + read(&c.client, &uri, &base, table, incarnation, &payloads).await?; + c.shutdown().await?; + let c = RpcCluster::open(dir.path(), Options::default()).await?; + read(&c.client, &uri, &base, table, incarnation, &payloads).await?; + c.shutdown().await +} diff --git a/crates/lance-context-wal/src/replication/runtime/peer.rs b/crates/lance-context-wal/src/replication/runtime/peer.rs index 9a2d05d4..f4e53fe8 100644 --- a/crates/lance-context-wal/src/replication/runtime/peer.rs +++ b/crates/lance-context-wal/src/replication/runtime/peer.rs @@ -604,4 +604,4 @@ async fn read_packet( } #[cfg(test)] -mod tests; +pub(crate) mod tests; diff --git a/crates/lance-context-wal/src/replication/runtime/peer/tests.rs b/crates/lance-context-wal/src/replication/runtime/peer/tests.rs index f5ab94bb..e3c6ae9b 100644 --- a/crates/lance-context-wal/src/replication/runtime/peer/tests.rs +++ b/crates/lance-context-wal/src/replication/runtime/peer/tests.rs @@ -17,11 +17,11 @@ use std::{net::SocketAddr, path::Path}; mod process; mod snapshots; -struct Tls { - client: Arc, - server: Arc, - certificate: [u8; 32], - name: String, +pub(crate) struct Tls { + pub(crate) client: Arc, + pub(crate) server: Arc, + pub(crate) certificate: [u8; 32], + pub(crate) name: String, } fn certificates() -> Result> { // Fresh keys and ordinary verifying TLS, including client certificates. @@ -71,19 +71,19 @@ fn certificates() -> Result> { Ok(configs) } -struct Running { - runtime: Runtime, - network: Transport, - handle: Handle, +pub(crate) struct Running { + pub(crate) runtime: Runtime, + pub(crate) network: Transport, + pub(crate) handle: Handle, } -struct Cluster { - nodes: BTreeMap, - tls: BTreeMap, - addresses: BTreeMap, +pub(crate) struct Cluster { + pub(crate) nodes: BTreeMap, + pub(crate) tls: BTreeMap, + pub(crate) addresses: BTreeMap, network_options: Options, } impl Cluster { - async fn open(root: &Path) -> Result { + pub(crate) async fn open(root: &Path) -> Result { Self::with_options(root, Options::default()).await } async fn with_options(root: &Path, network_options: Options) -> Result { @@ -144,12 +144,12 @@ impl Cluster { ); Ok(()) } - async fn stop(&mut self, id: u64) -> Result<()> { + pub(crate) async fn stop(&mut self, id: u64) -> Result<()> { let node = self.nodes.remove(&id).context("missing test node")?; node.network.shutdown().await?; node.runtime.shutdown().await } - async fn leader(&self, group: u64) -> Result { + pub(crate) async fn leader(&self, group: u64) -> Result { timeout(Duration::from_secs(10), async { loop { let leaders = self @@ -166,7 +166,7 @@ impl Cluster { .await .context("TLS replica election did not settle") } - async fn shutdown(mut self) -> Result<()> { + pub(crate) async fn shutdown(mut self) -> Result<()> { let ids: Vec<_> = self.nodes.keys().copied().collect(); for id in ids { self.stop(id).await?; diff --git a/crates/lance-context-wal/src/replication/runtime/peer/tests/process.rs b/crates/lance-context-wal/src/replication/runtime/peer/tests/process.rs index d094144a..6af19a6e 100644 --- a/crates/lance-context-wal/src/replication/runtime/peer/tests/process.rs +++ b/crates/lance-context-wal/src/replication/runtime/peer/tests/process.rs @@ -1,6 +1,7 @@ //! Child processes use the actual TLS transport. The parent sends only test //! control requests over loopback; it never relays consensus packets. use super::*; +use crate::replication::runtime::client as rpc; use serde::{Deserialize, Serialize}; use std::{ fs::{self, File}, @@ -16,6 +17,8 @@ struct Address { peer: SocketAddr, image: SocketAddr, control: SocketAddr, + rpc_data: SocketAddr, + rpc_control: SocketAddr, pid: u32, } #[derive(Serialize, Deserialize)] @@ -41,7 +44,7 @@ enum Reply { const FRAME_LIMIT: usize = 16 << 20; fn provision(root: &Path) -> Result<()> { - for id in 1..=3 { + for id in 1..=4 { let certified = rcgen::generate_simple_self_signed(vec![format!("wal-{id}.example.invalid")])?; fs::write(root.join(format!("cert-{id}.der")), certified.cert.der())?; @@ -60,7 +63,7 @@ type TlsConfiguration = ( fn tls(root: &Path, id: u64) -> Result { let mut roots = RootCertStore::empty(); let mut pins = BTreeMap::new(); - for peer in 1..=3 { + for peer in 1..=4 { let cert = CertificateDer::from(fs::read(root.join(format!("cert-{peer}.der")))?); pins.insert(peer, fingerprint(Some(std::slice::from_ref(&cert)))?); roots.add(cert)?; @@ -120,10 +123,14 @@ async fn tls_peer_process_fixture() -> Result<()> { ) .await?; let control = TcpListener::bind("127.0.0.1:0").await?; + let rpc_data = TcpListener::bind("127.0.0.1:0").await?; + let rpc_control = TcpListener::bind("127.0.0.1:0").await?; let address = Address { peer: listener.local_addr()?, image: image_listener.local_addr()?, control: control.local_addr()?, + rpc_data: rpc_data.local_addr()?, + rpc_control: rpc_control.local_addr()?, pid: std::process::id(), }; fs::write( @@ -149,6 +156,16 @@ async fn tls_peer_process_fixture() -> Result<()> { .with_disk_budget(super::disk_options())?; let (runtime, outbox) = Runtime::spawn(node, super::super::super::Options::default())?; let handle = runtime.handle(); + let mut rpc_tls = (*server).clone(); + rpc_tls.alpn_protocols = vec![rpc::ALPN.to_vec()]; + let rpc_service = rpc::Server::spawn( + handle.clone(), + rpc_data, + rpc_control, + BTreeSet::from([pins[&4]]), + Arc::new(rpc_tls), + rpc::Options::default(), + )?; let peers = addresses .into_iter() .filter(|(peer, _)| *peer != id) @@ -210,6 +227,7 @@ async fn tls_peer_process_fixture() -> Result<()> { let request = match receive(&mut stream).await { Ok(request) => request, Err(_) => { + rpc_service.shutdown().await?; images.shutdown().await?; network.shutdown().await?; return runtime.shutdown().await; @@ -257,6 +275,8 @@ struct Process { stream: ControlStream, address: SocketAddr, image_address: SocketAddr, + rpc_data: SocketAddr, + rpc_control: SocketAddr, } impl Process { fn spawn(root: &Path, id: u64, incarnation: u64) -> Result { @@ -302,7 +322,7 @@ impl Process { .open(root.join(format!("stderr-{id}-{incarnation}")))?, )) .spawn()?; - let result = (|| -> Result<(ControlStream, SocketAddr, SocketAddr)> { + let result = (|| -> Result<(ControlStream, Address)> { let deadline = Instant::now() + Duration::from_secs(10); while !address_file.exists() { ensure!(child.try_wait()?.is_none(), "TLS peer exited at startup"); @@ -324,14 +344,16 @@ impl Process { "original peer endpoints changed" ); } - Ok((stream, address.peer, address.image)) + Ok((stream, address)) })(); match result { - Ok((stream, address, image_address)) => Ok(Self { + Ok((stream, address)) => Ok(Self { child, stream, - address, - image_address, + address: address.peer, + image_address: address.image, + rpc_data: address.rpc_data, + rpc_control: address.rpc_control, }), Err(e) => { let _ = child.kill(); @@ -685,3 +707,149 @@ fn restart_follower( )); Ok(()) } + +#[test] +fn client_rpc_survives_leader_sigkill_and_full_original_directory_restart() -> Result<()> { + let directory = tempfile::tempdir()?; + provision(directory.path())?; + let mut processes = Processes { + nodes: BTreeMap::new(), + root: directory.path().to_owned(), + incarnation: 0, + }; + processes.start_all()?; + let executor = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build()?; + let make_client = |processes: &Processes| -> Result { + let (config, _, pins) = tls(&processes.root, 4)?; + let mut config = (*config).clone(); + config.alpn_protocols = vec![rpc::ALPN.to_vec()]; + let endpoints = processes + .nodes + .iter() + .map(|(&id, node)| { + ( + id, + rpc::Endpoint { + data_address: node.rpc_data.to_string(), + control_address: node.rpc_control.to_string(), + server_name: format!("wal-{id}.example.invalid"), + certificate: pins[&id], + }, + ) + }) + .collect(); + rpc::Client::new( + &layout(), + &Limits::default(), + endpoints, + Arc::new(config), + rpc::Options::default(), + ) + }; + let client = make_client(&processes)?; + let mut records = Vec::new(); + for group in [10, 20] { + let leader = processes.leader(group)?; + for sequence in 1..=8 { + let base = batch(sequence); + let body = Bytes::from(vec![sequence as u8; 2 << 20]); + let record = Batch::new( + base.meta.stream, + if group == 10 { 0 } else { 1 }, + base.meta.producer, + sequence, + body, + ); + let response = executor.block_on(client.request( + leader, + rpc::Request::Append { + group, + batch: record.clone(), + }, + ))?; + let rpc::Reply::Commit(Outcome::Committed(Applied::Accepted { index })) = + response.reply + else { + anyhow::bail!("client did not receive durable ACK"); + }; + records.push((group, index, record)); + } + processes.applied(group, records.last().unwrap().1)?; + } + let leader = processes.leader(10)?; + let old = records + .iter() + .rev() + .find(|(group, _, _)| *group == 10) + .unwrap(); + // Populate the actual route cache before killing its current destination. + let receipt = executor.block_on(client.request_routed(rpc::Request::Append { + group: 10, + batch: old.2.clone(), + }))?; + assert!( + matches!(receipt.reply, rpc::Reply::Commit(Outcome::Committed(Applied::Accepted { index })) if index == old.1) + ); + drop(receipt); + processes.nodes.remove(&leader).unwrap().kill()?; + processes.leader(10)?; + let mut recovered = false; + for _ in 0..3 { + // Explicit retry retains identity; the transport does not turn unknown + // outcomes into a fresh sequence or silently replay the operation. + if let Ok(response) = executor.block_on(client.request_routed(rpc::Request::Append { + group: 10, + batch: old.2.clone(), + })) { + if matches!(response.reply, rpc::Reply::Commit(Outcome::Committed(Applied::Accepted { index })) if index == old.1) + { + recovered = true; + break; + } + } + } + ensure!(recovered, "client could not reconcile after leader loss"); + drop(client); + processes.stop_all()?; + processes.start_all()?; + let client = make_client(&processes)?; + for group in [10, 20] { + let leader = processes.leader(group)?; + let (_, index, record) = records.iter().rev().find(|(g, _, _)| *g == group).unwrap(); + let response = executor.block_on(client.request( + leader, + rpc::Request::Append { + group, + batch: record.clone(), + }, + ))?; + ensure!( + matches!(response.reply, rpc::Reply::Commit(Outcome::Committed(Applied::Accepted { index: actual })) if actual == *index), + "retry moved the durable position" + ); + processes.applied(group, *index)?; + } + for node in 1..=3 { + for (group, index, record) in &records { + let response = executor.block_on(client.request( + node, + rpc::Request::Fetch { + group: *group, + index: *index, + }, + ))?; + ensure!( + matches!(&response.reply, rpc::Reply::Record(Some(meta)) if meta == &record.meta), + "recovered record metadata changed" + ); + ensure!( + response.body == record.body, + "recovered complete payload changed" + ); + } + } + processes.stop_all() +} diff --git a/crates/lance-context-wal/tests/raft_log_checkpoint.rs b/crates/lance-context-wal/tests/raft_log_checkpoint.rs index 503a70fb..f3853e7c 100644 --- a/crates/lance-context-wal/tests/raft_log_checkpoint.rs +++ b/crates/lance-context-wal/tests/raft_log_checkpoint.rs @@ -12,6 +12,11 @@ use raft::{GetEntriesContext, Storage, StorageError}; use raft_proto::eraftpb::{Entry, HardState, Snapshot}; use uuid::Uuid; +// A concurrent fork briefly inherits other tests' flock descriptors until exec. +// Exclude that window between independent reopen cases, while preserving the +// real subprocess/crash behavior exercised within each case. +static TEST_CASE: std::sync::Mutex<()> = std::sync::Mutex::new(()); + fn layout() -> Layout { Layout { cluster: Uuid::from_u128(22), @@ -60,6 +65,7 @@ fn read(log: &GroupLog, low: u64, high: u64) -> raft::Result> { #[test] fn checkpoint_preserves_suffix_vote_and_other_group_across_reopen() -> Result<()> { + let _case = TEST_CASE.lock().unwrap(); let directory = tempfile::tempdir()?; let path = directory.path().join("log"); let original: Vec<_> = (1..=6) @@ -143,6 +149,7 @@ fn checkpoint_preserves_suffix_vote_and_other_group_across_reopen() -> Result<() #[test] fn invalid_or_changed_checkpoint_never_changes_usable_log() -> Result<()> { + let _case = TEST_CASE.lock().unwrap(); let directory = tempfile::tempdir()?; let log = PhysicalLog::open(&directory.path().join("log"), &layout(), 1)?; let group = log.group(10)?; @@ -193,6 +200,7 @@ fn invalid_or_changed_checkpoint_never_changes_usable_log() -> Result<()> { #[test] fn missing_corrupt_or_inconsistent_checkpoint_fails_reopen() -> Result<()> { + let _case = TEST_CASE.lock().unwrap(); for corruption in 0..5 { let directory = tempfile::tempdir()?; let path = directory.path().join("log"); @@ -225,10 +233,31 @@ fn missing_corrupt_or_inconsistent_checkpoint_fails_reopen() -> Result<()> { } engine.write(&mut batch, true)?; } - assert!( - PhysicalLog::open(&path, &layout(), 1).is_err(), - "accepted corruption {corruption}" - ); + let error = match PhysicalLog::open(&path, &layout(), 1) { + Ok(_) => anyhow::bail!("accepted corruption {corruption}"), + Err(error) => error, + }; + // A transient lock failure is not evidence that corrupt state was + // rejected. Require the actual validation failure for every case. + match corruption { + 0 | 2 => assert!( + error.to_string().contains("invalid durable term index"), + "{error:#}" + ), + 1 => assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ), + 3 => assert!( + error.to_string().contains("oversized durable snapshot"), + "{error:#}" + ), + 4 => assert!( + error.to_string().contains("unbound checkpoint"), + "{error:#}" + ), + _ => unreachable!(), + } } Ok(()) } @@ -271,6 +300,7 @@ mod process { #[test] fn checkpoint_survives_sigkill() -> Result<()> { + let _case = TEST_CASE.lock().unwrap(); use std::os::unix::process::ExitStatusExt; for phase in ["before", "after"] { let directory = tempfile::tempdir()?; diff --git a/docs/design/wal-client-transport.md b/docs/design/wal-client-transport.md new file mode 100644 index 00000000..db6f4471 --- /dev/null +++ b/docs/design/wal-client-transport.md @@ -0,0 +1,51 @@ +# WAL client transport + +`replication::runtime::client` connects trusted ingress and materializer services +to the public replica runtime. It uses mutual TLS, explicitly pinned service +certificates, a versioned ALPN and cluster/layout/policy/node/request correlation. +A trusted client identity permits access to the node's hosted groups. The HTTP +layer still must resolve and authorize each table; this transport does not accept +end-user credentials or infer permission from a table UUID. + +Each physical node has separate data and control listeners and connection pools. +Every hosted group shares those pools. Appends and source reads use the data lane; +producer allocation, publication, quorum queries and leader hints use control. +Both lanes have bounded connections, logical byte credits and whole-exchange +deadlines. Metadata lengths are checked before variable-sized allocations. +The server connection allowance is independent of each client's pool size, so +several ingress processes can share the listeners without exhausting their +connection capacity merely by retaining healthy idle connections. +Fetched responses retain their byte credit until the response owner is dropped. +These are logical ownership bounds, not a hard allocator/RSS quota; callers must +budget retained clones and input allocations. + +An append sends a bounded metadata frame followed by its original `Bytes` body. +There is no separate upload-grant round trip and no whole-body concatenation or +Arrow decoding in this layer. TLS encryption still copies encrypted buffers. +The server waits for the runtime's durable completion before reporting a commit; +queueing a command or sending bytes cannot produce an accepted ACK. A failed or +cancelled exchange discards its connection. Keep the original producer sequence +and payload for explicit reconciliation; errors do not prove non-commit. + +Routing follows only configured-voter hints returned before command submission. +A cached hint is never treated as an ACK. Unknown outcomes are returned to the +caller instead of automatically advancing/reissuing logical operations. The +source-fetch API returns local committed data; a quorum-confirmed source pin and +matching metadata are required before materialization. Missing local data does +not prove that an uncertain append was rejected. + +With `lance-integration`, `Client::manifest_catalog()` implements the actual +`ManagedManifest` catalog interface over these RPCs. This supplies networked +quorum reads and durable manifest reservations. The complete HTTP/core ingestion +path, standalone node configuration and automatic maintenance remain follow-up +work; this module alone is not a production ingestion deployment. + +Tests use four independent client pools, real TLS and the durable runtime. A +three-process test writes both groups, kills their leader, reconciles the exact +request, restarts all original directories and checks every payload on every +replica. Separate tests abandon a committed reply, exhaust data connections, +reject foreign identities and oversized frames, and verify no-quorum behavior. +A real Lance dataset is opened through the networked catalog before and after +all replicas restart. These checks establish correctness, not throughput. + +Tracking: #378. diff --git a/scripts/ci/install-manylinux-deps.sh b/scripts/ci/install-manylinux-deps.sh index 3bfce4e6..3d2134b3 100644 --- a/scripts/ci/install-manylinux-deps.sh +++ b/scripts/ci/install-manylinux-deps.sh @@ -6,6 +6,8 @@ if command -v yum >/dev/null 2>&1; then if test -f /etc/almalinux-release; then # These build dependencies are all in BaseOS/AppStream. Avoid querying EPEL # and third-party mirrorlists; keep the distribution's GPG and TLS checks. + # yum expands $releasever/$basearch, not the invoking shell. + # shellcheck disable=SC2016 yum_options+=( '--disablerepo=*' --enablerepo=baseos @@ -18,9 +20,50 @@ if command -v yum >/dev/null 2>&1; then fi yum "${yum_options[@]}" install -y curl make openssl-devel perl elif command -v apt-get >/dev/null 2>&1; then - apt-get -o Acquire::Retries=2 -o Acquire::http::Timeout=30 -o Acquire::https::Timeout=30 update - apt-get -o Acquire::Retries=2 -o Acquire::http::Timeout=30 -o Acquire::https::Timeout=30 \ - install -y curl libssl-dev ca-certificates make perl + # The cross-build images also contain unrelated PPAs. These dependencies + # need only official distro repositories; a PPA outage must not block them. + # Keep both metadata and packages authenticated by the distro keyring. + # shellcheck source=/dev/null + source /etc/os-release + [[ ${VERSION_CODENAME:-} =~ ^[a-z]+$ ]] + case "$ID" in + ubuntu) + keyring=/usr/share/keyrings/ubuntu-archive-keyring.gpg + case "$(dpkg --print-architecture)" in + amd64|i386) archive=https://archive.ubuntu.com/ubuntu; security=https://security.ubuntu.com/ubuntu ;; + *) archive=https://ports.ubuntu.com/ubuntu-ports; security=$archive ;; + esac + security_suite=$VERSION_CODENAME-security + ;; + debian) + keyring=/usr/share/keyrings/debian-archive-keyring.gpg + archive=https://deb.debian.org/debian + security=https://security.debian.org/debian-security + security_suite=$VERSION_CODENAME-security + ;; + *) echo "Unsupported apt build distribution: $ID" >&2; exit 1 ;; + esac + test -r "$keyring" + apt_workspace=$(mktemp -d) + trap 'rm -rf -- "$apt_workspace"' EXIT + chmod 755 "$apt_workspace" + mkdir "$apt_workspace/lists" + cat > "$apt_workspace/sources.list" <&2 exit 1