diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 72692e43..678f47c7 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -166,6 +166,11 @@ jobs: run: | cargo clippy -p lance-context-ingest --features materializer --all-targets -- -D warnings cargo test -p lance-context-ingest --features materializer --lib + - name: HTTP ingestion and recoverable producer journal + run: | + cargo clippy -p lance-context-ingest --features http --all-targets -- -D warnings + cargo clippy -p lance-context-ingest --features http-client,materializer --all-targets -- -D warnings + cargo test -p lance-context-ingest --features http-client,materializer --lib - name: Start etcd for HA scheduler test run: | ETCD_VERSION=3.7.0 diff --git a/Cargo.lock b/Cargo.lock index 7a8afd9b..804a7bef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4182,15 +4182,18 @@ dependencies = [ "arrow-array", "arrow-ipc", "arrow-schema", + "axum", "base64 0.22.1", "bytes", "futures", + "http-body-util", "lance", "lance-context-core", "lance-context-wal", "lance-table", "object_store 0.14.2", "rcgen", + "reqwest 0.12.28", "ring", "serde", "serde_json", diff --git a/crates/lance-context-ingest/Cargo.toml b/crates/lance-context-ingest/Cargo.toml index c2575654..1c7e6eae 100644 --- a/crates/lance-context-ingest/Cargo.toml +++ b/crates/lance-context-ingest/Cargo.toml @@ -7,9 +7,14 @@ publish = false description = "Durable rollout ingress over the independent lance-context WAL" [features] +http = ["dep:axum", "dep:http-body-util"] +http-client = ["http", "dep:reqwest", "lance-context-wal/client-journal"] materializer = ["lance-context-wal/lance-integration", "dep:lance", "dep:lance-table", "dep:object_store", "dep:futures"] [dependencies] +axum = { version = "0.8", optional = true, default-features = false, features = ["json", "http1", "tokio"] } +http-body-util = { version = "0.1", optional = true } +reqwest = { version = "0.12", optional = true, default-features = false, features = ["json", "rustls-tls"] } lance = { version = "13.0.0", optional = true } lance-table = { version = "13.0.0", optional = true } object_store = { version = "0.14.2", optional = true } @@ -28,6 +33,7 @@ tokio = { version = "1", features = ["rt", "sync", "time"] } uuid = { version = "1.20.0", features = ["v4", "serde"] } [dev-dependencies] +reqwest = { version = "0.12", default-features = false, features = ["json", "stream"] } arrow-ipc = "58" lance = "13.0.0" lance-table = "13.0.0" diff --git a/crates/lance-context-ingest/src/http.rs b/crates/lance-context-ingest/src/http.rs new file mode 100644 index 00000000..0a93716b --- /dev/null +++ b/crates/lance-context-ingest/src/http.rs @@ -0,0 +1,426 @@ +//! Bounded HTTP ingress over the public TLS WAL client. Install behind the +//! application's authenticated/authorized routing boundary. Credentials prove +//! producer/receipt identity, not permission for an end user to access a table. +use super::*; +use axum::{ + extract::{Path, Request, State}, + http::{header, StatusCode}, + response::{IntoResponse, Response}, + routing::{get, post}, + Json, Router, +}; +use receipt::ReceiptKeys; +use std::{collections::BTreeMap, sync::Mutex}; +use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore}; + +mod body; +#[cfg(feature = "http-client")] +pub mod journal; + +pub const ARROW_FILE: &str = "application/vnd.apache.arrow.file"; +pub const PRODUCER: &str = "x-lance-producer"; +pub const SEQUENCE: &str = "x-lance-sequence"; + +#[derive(Clone, Debug)] +pub struct Options { + pub data_requests: usize, + pub control_requests: usize, + pub admission_timeout: Duration, + pub body_idle_timeout: Duration, + pub body_total_timeout: Duration, +} +impl Default for Options { + fn default() -> Self { + Self { + data_requests: 64, + control_requests: 16, + admission_timeout: Duration::from_secs(5), + body_idle_timeout: Duration::from_secs(10), + body_total_timeout: Duration::from_secs(60), + } + } +} +#[derive(Default)] +struct Lifecycle { + closed: bool, + active: usize, +} +struct Inner { + tables: BTreeMap, + keys: ReceiptKeys, + options: Options, + data: Arc, + control: Arc, + lifecycle: Mutex, + idle: Notify, +} +struct Guard { + inner: Arc, + slot: Option, +} +impl Drop for Guard { + fn drop(&mut self) { + let mut lifecycle = self.inner.lifecycle.lock().unwrap(); + self.slot.take(); + lifecycle.active -= 1; + drop(lifecycle); + self.inner.idle.notify_waiters(); + } +} +#[derive(Clone)] +pub struct Gateway(Arc); +#[derive(Debug, Serialize)] +pub struct Observation { + pub data_inflight: usize, + pub control_inflight: usize, + pub draining: bool, +} +impl Gateway { + /// Table bindings must be resolved and authorized before construction. + /// Cloned ingresses should share their physical WAL client and memory budget. + pub fn new( + tables: BTreeMap, + keys: ReceiptKeys, + options: Options, + ) -> Result { + ensure!( + !tables.is_empty() && tables.len() <= 4096, + "invalid HTTP table count" + ); + ensure!( + tables.iter().all(|(name, table)| !name.is_empty() + && name.len() <= 1024 + && !name.contains('/') + && name == &table.scope.contract.qualified_table), + "invalid HTTP table binding" + ); + ensure!( + (1..=256).contains(&options.data_requests) + && (1..=64).contains(&options.control_requests) + && (Duration::from_millis(1)..=Duration::from_secs(30)) + .contains(&options.admission_timeout) + && (Duration::from_millis(1)..=Duration::from_secs(60)) + .contains(&options.body_idle_timeout) + && (options.body_idle_timeout..=Duration::from_secs(300)) + .contains(&options.body_total_timeout), + "invalid HTTP admission/deadline limits" + ); + Ok(Self(Arc::new(Inner { + tables, + keys, + data: Arc::new(Semaphore::new(options.data_requests)), + control: Arc::new(Semaphore::new(options.control_requests)), + options, + lifecycle: Mutex::new(Lifecycle::default()), + idle: Notify::new(), + }))) + } + pub fn router(&self) -> Router { + Router::new() + .route("/api/v1/health", get(health)) + .route("/api/v1/ingest/{table}/producer", post(open)) + .route("/api/v1/ingest/{table}/append", post(append)) + .route("/api/v1/ingest/{table}/publication", post(publication)) + .route("/api/v1/ingest/{table}/close", post(close)) + .with_state(self.clone()) + } + pub fn observation(&self) -> Observation { + Observation { + data_inflight: self.0.options.data_requests - self.0.data.available_permits(), + control_inflight: self.0.options.control_requests - self.0.control.available_permits(), + draining: self.0.lifecycle.lock().unwrap().closed, + } + } + /// Stop admission and join every admitted handler, including handlers whose + /// HTTP client disconnected. Body reads have total/idle deadlines; WAL RPCs + /// retain their configured exchange deadline. No submitted write is aborted. + pub async fn shutdown(&self) { + self.0.lifecycle.lock().unwrap().closed = true; + loop { + let idle = self.0.idle.notified(); + tokio::pin!(idle); + idle.as_mut().enable(); + if self.0.lifecycle.lock().unwrap().active == 0 { + return; + } + idle.await; + } + } + fn enter(&self, data: bool) -> HttpResult { + let mut state = self.0.lifecycle.lock().unwrap(); + if state.closed { + return Err(Failure::unavailable("draining", "not_submitted")); + } + let lane = if data { &self.0.data } else { &self.0.control }; + let slot = lane + .clone() + .try_acquire_owned() + .map_err(|_| Failure::unavailable("request_capacity", "not_submitted"))?; + state.active += 1; + Ok(Guard { + inner: self.0.clone(), + slot: Some(slot), + }) + } +} +#[derive(Debug)] +struct Failure { + status: StatusCode, + code: &'static str, + outcome: &'static str, +} +type HttpResult = std::result::Result; +impl Failure { + fn bad(code: &'static str) -> Self { + Self { + status: StatusCode::BAD_REQUEST, + code, + outcome: "not_submitted", + } + } + fn unavailable(code: &'static str, outcome: &'static str) -> Self { + Self { + status: StatusCode::SERVICE_UNAVAILABLE, + code, + outcome, + } + } +} +impl IntoResponse for Failure { + fn into_response(self) -> Response { + let mut response = ( + self.status, + Json(serde_json::json!({ + "error": self.code, "operation_outcome": self.outcome, + "retry_same_request": true, + })), + ) + .into_response(); + if self.status == StatusCode::SERVICE_UNAVAILABLE { + response + .headers_mut() + .insert(header::RETRY_AFTER, "1".parse().unwrap()); + } + response + } +} +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Opened { + pub producer: String, + pub generation: u64, + pub scope: [u8; 32], +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct Accepted { + pub receipt: String, + pub rows: u32, + pub sequence: u64, + pub group: u64, + pub index: u64, + pub digest: [u8; 32], +} +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ReceiptRequest { + pub receipt: String, +} +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CloseRequest { + pub producer: String, + pub receipt: Option, +} +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct Visibility { + pub minimum_read_version: Option, +} +#[derive(Clone, Copy)] +enum Action { + Open, + Append, + Publication, + Close, +} +async fn health(State(gateway): State) -> StatusCode { + if gateway.0.lifecycle.lock().unwrap().closed { + StatusCode::SERVICE_UNAVAILABLE + } else { + StatusCode::OK + } +} +async fn open(State(g): State, Path(table): Path, request: Request) -> Response { + dispatch(g, table, request, Action::Open).await +} +async fn append(State(g): State, Path(table): Path, request: Request) -> Response { + dispatch(g, table, request, Action::Append).await +} +async fn publication( + State(g): State, + Path(table): Path, + request: Request, +) -> Response { + dispatch(g, table, request, Action::Publication).await +} +async fn close(State(g): State, Path(table): Path, request: Request) -> Response { + dispatch(g, table, request, Action::Close).await +} +async fn dispatch(gateway: Gateway, table: String, request: Request, action: Action) -> Response { + let guard = match gateway.enter(matches!(action, Action::Append)) { + Ok(guard) => guard, + Err(error) => return error.into_response(), + }; + // The admitted task owns its guard and body independently of the socket. + match tokio::spawn(async move { + let _guard = guard; + process(&gateway.0, &table, request, action).await + }) + .await + { + Ok(Ok(response)) => response, + Ok(Err(error)) => error.into_response(), + Err(_) => Failure::unavailable("handler_failed", "unknown").into_response(), + } +} +async fn process( + inner: &Inner, + table: &str, + request: Request, + action: Action, +) -> HttpResult { + let ingress = inner.tables.get(table).ok_or(Failure { + status: StatusCode::NOT_FOUND, + code: "unknown_table", + outcome: "not_submitted", + })?; + if matches!(action, Action::Append) { + return append_body(inner, ingress, request).await; + } + let data = body::read(request, 32 << 10, "application/json", &inner.options).await?; + match action { + Action::Open => { + let open: ProducerOpen = + serde_json::from_slice(&data).map_err(|_| Failure::bad("invalid_open"))?; + let producer = ingress + .open_producer(open) + .await + .map_err(|_| Failure::unavailable("open_outcome_unknown", "unknown"))?; + let token = inner + .keys + .encode_producer(&producer) + .map_err(|_| Failure::unavailable("credential_unavailable", "unknown"))?; + let scope = super::receipt::scope_hash(&ingress.scope) + .map_err(|_| Failure::unavailable("credential_unavailable", "unknown"))?; + Ok(Json(Opened { + producer: token, + generation: producer.generation(), + scope, + }) + .into_response()) + } + Action::Publication => { + let query: ReceiptRequest = + serde_json::from_slice(&data).map_err(|_| Failure::bad("invalid_receipt_query"))?; + let accepted = ingress + .restore_receipt(&inner.keys, &query.receipt) + .map_err(|_| Failure::bad("invalid_receipt"))?; + let status = ingress + .publication_status(&accepted) + .await + .map_err(|_| Failure::unavailable("publication_unavailable", "unknown"))?; + let version = match status { + PublicationStatus::Pending => None, + PublicationStatus::Published { + minimum_read_version, + } => Some(minimum_read_version), + }; + Ok(Json(Visibility { + minimum_read_version: version, + }) + .into_response()) + } + Action::Close => { + let close: CloseRequest = + serde_json::from_slice(&data).map_err(|_| Failure::bad("invalid_close"))?; + let producer = ingress + .restore_producer(&inner.keys, &close.producer) + .map_err(|_| Failure::bad("invalid_producer"))?; + let receipt = close + .receipt + .as_ref() + .map(|r| ingress.restore_receipt(&inner.keys, r)) + .transpose() + .map_err(|_| Failure::bad("invalid_receipt"))?; + ingress + .close_producer(&producer, receipt.as_ref()) + .await + .map_err(|_| Failure::unavailable("close_outcome_unknown", "unknown"))?; + Ok(StatusCode::NO_CONTENT.into_response()) + } + Action::Append => unreachable!(), + } +} +async fn append_body(inner: &Inner, ingress: &Ingress, request: Request) -> HttpResult { + if request.headers().get_all(PRODUCER).iter().count() != 1 + || request.headers().get_all(SEQUENCE).iter().count() != 1 + { + return Err(Failure::bad("ambiguous_producer_headers")); + } + let producer = request + .headers() + .get(PRODUCER) + .and_then(|v| v.to_str().ok()) + .ok_or_else(|| Failure::bad("missing_producer"))?; + let producer = ingress + .restore_producer(&inner.keys, producer) + .map_err(|_| Failure::bad("invalid_producer"))?; + let sequence = request + .headers() + .get(SEQUENCE) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .filter(|v| *v > 0 && *v < u64::MAX) + .ok_or_else(|| Failure::bad("invalid_sequence"))?; + let length = body::length( + &request, + ingress.scope.contract.max_record_bytes as usize, + ARROW_FILE, + )?; + let upload = tokio::time::timeout( + inner.options.admission_timeout, + ingress.reserve_upload(length), + ) + .await + .map_err(|_| Failure::unavailable("memory_capacity", "not_submitted"))? + .map_err(|_| Failure::bad("invalid_upload_size"))?; + let bytes = body::read_exact(request, length, &inner.options).await?; + let prepared = ingress + .prepare(upload, bytes, &producer, sequence) + .await + .map_err(|_| Failure::bad("invalid_rollout_ipc"))?; + let accepted = ingress.append(&prepared).await.map_err(|error| { + match error.downcast_ref::() { + Some(AppendFailure::Sequence { .. } | AppendFailure::ProducerFenced) => Failure { + status: StatusCode::CONFLICT, + code: "producer_sequence_rejected", + outcome: "rejected", + }, + _ => Failure::unavailable("append_outcome_unknown", "unknown"), + } + })?; + let receipt = inner + .keys + .encode(&accepted) + .map_err(|_| Failure::unavailable("receipt_unavailable", "unknown"))?; + Ok(( + StatusCode::ACCEPTED, + Json(Accepted { + receipt, + rows: accepted.rows(), + sequence, + group: accepted.position().group, + index: accepted.position().index, + digest: accepted.meta().digest, + }), + ) + .into_response()) +} diff --git a/crates/lance-context-ingest/src/http/body.rs b/crates/lance-context-ingest/src/http/body.rs new file mode 100644 index 00000000..a812f7e4 --- /dev/null +++ b/crates/lance-context-ingest/src/http/body.rs @@ -0,0 +1,82 @@ +use super::*; +use bytes::BytesMut; +use http_body_util::BodyExt; +use tokio::time::{timeout_at, Instant}; + +pub(super) fn length(request: &Request, maximum: usize, content_type: &str) -> HttpResult { + let headers = request.headers(); + if headers.contains_key(header::TRANSFER_ENCODING) + || headers.get_all(header::CONTENT_LENGTH).iter().count() != 1 + || headers.get_all(header::CONTENT_TYPE).iter().count() != 1 + || headers + .get(header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + != Some(content_type) + { + return Err(Failure::bad("invalid_body_headers")); + } + headers + .get(header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .filter(|n| *n > 0 && *n <= maximum) + .ok_or_else(|| Failure::bad("invalid_content_length")) +} +pub(super) async fn read( + request: Request, + maximum: usize, + content_type: &str, + options: &Options, +) -> HttpResult { + let length = length(&request, maximum, content_type)?; + read_exact(request, length, options).await +} +pub(super) async fn read_exact( + request: Request, + length: usize, + options: &Options, +) -> HttpResult { + let deadline = Instant::now() + options.body_total_timeout; + let mut body = request.into_body(); + let mut first: Option = None; + let mut combined: Option = None; + let mut received = 0usize; + loop { + let frame = timeout_at( + deadline.min(Instant::now() + options.body_idle_timeout), + body.frame(), + ) + .await + .map_err(|_| Failure::bad("body_deadline"))?; + let Some(frame) = frame else { + break; + }; + let frame = frame.map_err(|_| Failure::bad("body_disconnected"))?; + let data = frame + .into_data() + .map_err(|_| Failure::bad("unexpected_body_trailers"))?; + if data.is_empty() { + continue; + } + if data.len() > length - received { + return Err(Failure::bad("body_length_mismatch")); + } + received += data.len(); + if let Some(buffer) = combined.as_mut() { + buffer.extend_from_slice(&data); + } else if let Some(previous) = first.take() { + let mut buffer = BytesMut::with_capacity(length); + buffer.extend_from_slice(&previous); + buffer.extend_from_slice(&data); + combined = Some(buffer); + } else { + first = Some(data); + } + } + if received != length { + return Err(Failure::bad("body_length_mismatch")); + } + // A single HTTP frame retains its existing buffer. Fragmented bodies are + // coalesced once; subsequent validation and WAL transport share these Bytes. + Ok(combined.map(BytesMut::freeze).or(first).unwrap_or_default()) +} diff --git a/crates/lance-context-ingest/src/http/journal.rs b/crates/lance-context-ingest/src/http/journal.rs new file mode 100644 index 00000000..3d433ae5 --- /dev/null +++ b/crates/lance-context-ingest/src/http/journal.rs @@ -0,0 +1,439 @@ +//! Recoverable HTTP producers sharing one bounded group-commit journal and one +//! HTTP connection pool. Persist an intent before sending, and persist its ACK +//! before releasing the body or advancing the sequence. Local journal ambiguity +//! fences that lane until reopen; network ambiguity retains the exact intent. +use super::*; +use lance_context_wal::client_journal::{Journal, Settings}; +use std::{collections::BTreeSet, path::PathBuf}; +use tokio::sync::Mutex as AsyncMutex; + +const MAX_METADATA: usize = 32 << 10; +const MAX_TOKEN: usize = 12 << 10; + +#[derive(Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Pending { + sequence: u64, + bytes: usize, + digest: [u8; 32], + #[serde(skip)] + body: Bytes, +} +#[derive(Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Saved { + version: u8, + next: u64, + producer: Option, + last: Option, + pending: Option, +} +impl Default for Saved { + fn default() -> Self { + Self { + version: 1, + next: 1, + producer: None, + last: None, + pending: None, + } + } +} +struct Lane { + saved: Saved, + reopen: bool, +} + +pub struct Config { + pub directory: PathBuf, + pub opens: Vec, + pub base: reqwest::Url, + pub request_timeout: Duration, + pub journal: Settings, +} + +pub struct DurableClient { + // Drop/join the journal before releasing the logical workspace reservation. + journal: Journal, + lanes: Vec>, + opens: Vec, + groups: Vec, + scope: Scope, + scope_hash: [u8; 32], + client: reqwest::Client, + base: reqwest::Url, + timeout: Duration, + _workspace: Arc, +} +fn digest(body: &[u8]) -> [u8; 32] { + ring::digest::digest(&ring::digest::SHA256, body) + .as_ref() + .try_into() + .unwrap() +} +impl DurableClient { + /// Reopen with the same layout, binding, ordered producer-open tuples and + /// journal settings. Persist those tuples in configuration; generating new + /// operation UUIDs on restart is not a retry. Endpoint/pool may change for + /// gateway failover. Supply a trusted HTTP client with redirects disabled. + pub async fn open( + config: Config, + layout: &Layout, + binding: TableBinding, + client: reqwest::Client, + budget: Arc, + ) -> Result { + let Config { + directory, + opens, + mut base, + request_timeout: timeout, + journal: settings, + } = config; + settings.validate()?; + layout.validate()?; + let contract = binding.contract()?; + ensure!( + !binding.table.is_nil() + && !binding.incarnation.is_nil() + && (1..=16384).contains(&opens.len()) + && (Duration::from_millis(100)..=Duration::from_secs(120)).contains(&timeout) + && matches!(base.scheme(), "http" | "https") + && base.query().is_none() + && base.fragment().is_none(), + "invalid HTTP client scope" + ); + let scope = Scope { + cluster: layout.cluster, + layout: layout.fingerprint()?, + table: binding.table, + destination: binding.destination, + incarnation: binding.incarnation, + contract, + }; + let scope_hash = super::super::receipt::scope_hash(&scope)?; + let mut used = BTreeSet::new(); + let groups = opens + .iter() + .map(|open| { + let group = layout.group_for_bucket(open.bucket)?; + ensure!( + !open.operation.is_nil() + && open.expected_generation < u64::MAX + && open.slot < producer::MAX_SLOTS + && used.insert((group, open.slot)), + "invalid/duplicate producer slot" + ); + Ok(group) + }) + .collect::>>()?; + let per_lane = scope.contract.max_record_bytes as usize + MAX_METADATA + 4; + ensure!( + per_lane + .checked_mul(opens.len()) + .is_some_and(|n| n <= settings.state_bytes as usize) + && per_lane + 128 <= settings.batch_bytes as usize, + "journal capacity cannot hold the configured producer intents" + ); + let workspace = (settings.state_bytes as usize) + .checked_mul(3) + .and_then(|n| n.checked_add(settings.queue_bytes as usize)) + .and_then(|n| n.checked_add(opens.len() * MAX_METADATA * 2)) + .context("journal workspace overflow")?; + ensure!( + workspace <= budget.limit(), + "journal workspace exceeds client budget" + ); + // Hold before recovery allocates state or the disk worker can encode it. + let reservation = Arc::new(budget.reserve(workspace).await); + base.path_segments_mut() + .map_err(|_| anyhow::anyhow!("HTTP base URL cannot contain path segments"))? + .pop_if_empty() + .extend(["api", "v1", "ingest", &scope.contract.qualified_table]); + tokio::task::spawn_blocking(move || { + let workspace_guard = reservation; + let binding_digest = digest(&serde_json::to_vec(&(&scope, &opens))?); + let count = opens.len() as u32; + let (journal, recovered) = Journal::open(&directory, &binding_digest, count, settings)?; + let mut lanes = Vec::with_capacity(opens.len()); + for (key, value) in recovered.into_iter().enumerate() { + let saved = if let Some(bytes) = value { + ensure!(bytes.len() >= 4, "truncated client journal record"); + let length = u32::from_le_bytes(bytes[..4].try_into()?) as usize; + ensure!( + length > 0 && length <= MAX_METADATA && length <= bytes.len() - 4, + "invalid client journal metadata" + ); + let mut saved: Saved = serde_json::from_slice(&bytes[4..4 + length])?; + let body = bytes.slice(4 + length..); + if let Some(pending) = saved.pending.as_mut() { + ensure!( + pending.bytes == body.len() + && !body.is_empty() + && body.len() <= scope.contract.max_record_bytes as usize + && pending.digest == digest(&body) + && pending.sequence == saved.next, + "invalid recovered append intent" + ); + pending.body = body; + } else { + ensure!(body.is_empty(), "unexpected journal payload"); + } + saved + } else { + Saved::default() + }; + ensure!( + saved.version == 1 + && saved.next > 0 + && saved.next < u64::MAX + && (saved.last.is_none() || saved.producer.is_some()), + "invalid recovered sequence" + ); + ensure!( + saved.last.as_ref().map_or(saved.next == 1, |last| { + last.sequence.checked_add(1) == Some(saved.next) + && last.group == groups[key] + && last.index > 0 + && last.index < u64::MAX + && last.rows > 0 + && last.rows <= scope.contract.max_rows + && !last.receipt.is_empty() + && last.receipt.len() <= MAX_TOKEN + }), + "invalid recovered ACK" + ); + if let Some(credential) = &saved.producer { + ensure!( + credential.scope == scope_hash + && credential.generation == opens[key].expected_generation + 1 + && !credential.producer.is_empty() + && credential.producer.len() <= MAX_TOKEN, + "invalid recovered producer" + ); + } + lanes.push(AsyncMutex::new(Lane { + saved, + reopen: false, + })); + } + Ok(Self { + journal, + lanes, + opens, + groups, + scope, + scope_hash, + client, + base, + timeout, + _workspace: workspace_guard, + }) + }) + .await? + } + fn lane(&self, key: usize) -> Result<&AsyncMutex> { + self.lanes.get(key).context("client lane out of range") + } + fn endpoint(&self, action: &str) -> reqwest::Url { + let mut url = self.base.clone(); + url.path_segments_mut().unwrap().push(action); + url + } + async fn persist(&self, key: usize, lane: &mut Lane, next: Saved) -> Result<()> { + ensure!( + !lane.reopen, + "local journal outcome unknown; reopen before continuing" + ); + let metadata = serde_json::to_vec(&next)?; + ensure!( + metadata.len() <= MAX_METADATA, + "client journal metadata too large" + ); + let mut prefix = Vec::with_capacity(4 + metadata.len()); + prefix.extend_from_slice(&(metadata.len() as u32).to_le_bytes()); + prefix.extend_from_slice(&metadata); + let body = next + .pending + .as_ref() + .map_or_else(Bytes::new, |p| p.body.clone()); + // Cancellation at this await can already have changed disk. Keep the + // fence unless a durable completion and in-memory update both finish. + lane.reopen = true; + self.journal + .put(key as u32, Bytes::from(prefix), body) + .await?; + lane.saved = next; + lane.reopen = false; + Ok(()) + } + async fn stage_locked(&self, key: usize, lane: &mut Lane, body: Bytes) -> Result { + ensure!( + !lane.reopen, + "local journal outcome unknown; reopen before continuing" + ); + ensure!( + !body.is_empty() && body.len() <= self.scope.contract.max_record_bytes as usize, + "append body outside table bounds" + ); + ensure!( + lane.saved.next < u64::MAX - 1, + "producer sequence exhausted" + ); + if let Some(pending) = &lane.saved.pending { + let same = if pending.body.len() != body.len() { + false + } else if pending.body.as_ptr() == body.as_ptr() { + true + } else { + let original = pending.body.clone(); + let workspace = self._workspace.clone(); + tokio::task::spawn_blocking(move || { + let _workspace = workspace; + original == body + }) + .await? + }; + ensure!( + same, + "different payload supplied for an unresolved sequence" + ); + return Ok(pending.sequence); + } + let workspace = self._workspace.clone(); + let body = tokio::task::spawn_blocking(move || { + let _workspace = workspace; + (digest(&body), body) + }) + .await?; + let mut next = lane.saved.clone(); + let sequence = next.next; + next.pending = Some(Pending { + sequence, + bytes: body.1.len(), + digest: body.0, + body: body.1, + }); + self.persist(key, lane, next).await?; + Ok(sequence) + } + /// Persist a complete source intent without sending it yet. Useful when + /// coordinating a durable source checkpoint with later recovery. + pub async fn stage(&self, key: usize, body: Bytes) -> Result { + let mut lane = self.lane(key)?.lock().await; + self.stage_locked(key, &mut lane, body).await + } + pub async fn append(&self, key: usize, body: Bytes) -> Result { + let mut lane = self.lane(key)?.lock().await; + self.stage_locked(key, &mut lane, body).await?; + self.resume_locked(key, &mut lane) + .await? + .context("staged append disappeared") + } + /// Resend only the recovered/staged bytes and sequence. A network error + /// leaves this intent intact; the caller decides when to retry it. + pub async fn resume(&self, key: usize) -> Result> { + let mut lane = self.lane(key)?.lock().await; + self.resume_locked(key, &mut lane).await + } + async fn resume_locked(&self, key: usize, lane: &mut Lane) -> Result> { + ensure!( + !lane.reopen, + "local journal outcome unknown; reopen before continuing" + ); + let Some(pending) = lane.saved.pending.clone() else { + return Ok(None); + }; + if lane.saved.producer.is_none() { + let response = self + .client + .post(self.endpoint("producer")) + .timeout(self.timeout) + .json(&self.opens[key]) + .send() + .await? + .error_for_status()?; + let credential: Opened = bounded_json(response).await?; + ensure!( + credential.scope == self.scope_hash + && credential.generation == self.opens[key].expected_generation + 1 + && !credential.producer.is_empty() + && credential.producer.len() <= MAX_TOKEN, + "foreign producer credential" + ); + let mut next = lane.saved.clone(); + next.producer = Some(credential); + self.persist(key, lane, next).await?; + } + let response = self + .client + .post(self.endpoint("append")) + .timeout(self.timeout) + .header(header::CONTENT_TYPE.as_str(), ARROW_FILE) + .header(PRODUCER, &lane.saved.producer.as_ref().unwrap().producer) + .header(SEQUENCE, pending.sequence.to_string()) + .body(pending.body.clone()) + .send() + .await?; + ensure!( + response.status() == reqwest::StatusCode::ACCEPTED, + "append outcome unresolved (HTTP {}); retain original intent", + response.status() + ); + let accepted: Accepted = bounded_json(response).await?; + ensure!( + accepted.sequence == pending.sequence + && accepted.group == self.groups[key] + && accepted.index > 0 + && accepted.index < u64::MAX + && accepted.digest == pending.digest + && accepted.rows > 0 + && accepted.rows <= self.scope.contract.max_rows + && !accepted.receipt.is_empty() + && accepted.receipt.len() <= MAX_TOKEN, + "foreign append ACK" + ); + let mut next = lane.saved.clone(); + next.next = pending.sequence + 1; + next.last = Some(accepted.clone()); + next.pending = None; + self.persist(key, lane, next).await?; + Ok(Some(accepted)) + } + pub async fn last_accepted(&self, key: usize) -> Result> { + let lane = self.lane(key)?.lock().await; + ensure!( + !lane.reopen, + "local journal outcome unknown; reopen before continuing" + ); + Ok(lane.saved.last.clone()) + } + pub fn statistics(&self) -> lance_context_wal::client_journal::Statistics { + self.journal.statistics() + } + /// Caller first joins its producer tasks. Move the final owner off the I/O + /// executor while the local disk worker joins and releases its engine lock. + pub async fn shutdown(self) -> Result<()> { + tokio::task::spawn_blocking(move || drop(self)).await?; + Ok(()) + } +} +async fn bounded_json( + mut response: reqwest::Response, +) -> Result { + if let Some(length) = response.content_length() { + ensure!( + length <= MAX_METADATA as u64, + "HTTP response metadata too large" + ); + } + let mut bytes = Vec::new(); + while let Some(part) = response.chunk().await? { + ensure!( + part.len() <= MAX_METADATA - bytes.len(), + "HTTP response metadata too large" + ); + bytes.extend_from_slice(&part); + } + Ok(serde_json::from_slice(&bytes)?) +} diff --git a/crates/lance-context-ingest/src/lib.rs b/crates/lance-context-ingest/src/lib.rs index 6de32eca..9b3e9246 100644 --- a/crates/lance-context-ingest/src/lib.rs +++ b/crates/lance-context-ingest/src/lib.rs @@ -29,6 +29,8 @@ use serde::{Deserialize, Serialize}; use std::{collections::HashSet, sync::Arc, time::Duration}; use uuid::Uuid; +#[cfg(feature = "http")] +pub mod http; #[cfg(feature = "materializer")] pub mod materializer; pub mod receipt; diff --git a/crates/lance-context-ingest/src/receipt.rs b/crates/lance-context-ingest/src/receipt.rs index e131e0e9..31e300d3 100644 --- a/crates/lance-context-ingest/src/receipt.rs +++ b/crates/lance-context-ingest/src/receipt.rs @@ -7,12 +7,12 @@ //! 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 super::{AcceptedAppend, Ingress, Producer, ProducerOpen, 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 serde::{de::DeserializeOwned, Deserialize, Serialize}; use std::{collections::BTreeMap, sync::Arc}; const VERSION: u8 = 1; @@ -22,6 +22,7 @@ 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"; +const PRODUCER_DOMAIN: &[u8] = b"lance-context/durable-ingress-producer/v1"; /// Deliberately not serializable or Debug: key material stays in configuration. pub struct ReceiptKeys { @@ -37,6 +38,13 @@ struct Claims { meta: BatchMeta, rows: u32, } +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct ProducerClaims { + scope: [u8; 32], + open: ProducerOpen, + generation: u64, +} impl ReceiptKeys { pub fn new(active: u32, keys: impl IntoIterator) -> Result { @@ -65,12 +73,32 @@ impl ReceiptKeys { } 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, - })?; + self.sign( + DOMAIN, + &Claims { + scope: scope_hash(&accepted.scope)?, + position: accepted.position.clone(), + meta: accepted.meta.clone(), + rows: accepted.rows, + }, + ) + } + + /// Identifies an allocated producer across stateless ingress instances. + /// It is not user authorization; the WAL still fences stale generations. + pub fn encode_producer(&self, producer: &Producer) -> Result { + self.sign( + PRODUCER_DOMAIN, + &ProducerClaims { + scope: scope_hash(&producer.scope)?, + open: producer.open.clone(), + generation: producer.generation, + }, + ) + } + + fn sign(&self, domain: &[u8], claims: &impl Serialize) -> Result { + let claims = serde_json::to_vec(claims)?; ensure!( claims.len() <= MAX_CLAIMS, "receipt claims exceed protocol limit" @@ -80,13 +108,13 @@ impl ReceiptKeys { 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(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 { + fn verify(&self, domain: &[u8], token: &str) -> Result { ensure!( token.len() <= MAX_TOKEN, "receipt token exceeds protocol limit" @@ -104,12 +132,16 @@ impl ReceiptKeys { 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); + 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])?; + Ok(serde_json::from_slice(&binary[HEADER..end])?) + } + + fn decode(&self, token: &str, scope: Arc) -> Result { + let claims: Claims = self.verify(DOMAIN, token)?; ensure!(claims.scope == scope_hash(&scope)?, "foreign receipt scope"); Ok(AcceptedAppend { scope, @@ -120,7 +152,7 @@ impl ReceiptKeys { } } -fn scope_hash(scope: &Scope) -> Result<[u8; 32]> { +pub(super) 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)?); @@ -128,6 +160,29 @@ fn scope_hash(scope: &Scope) -> Result<[u8; 32]> { } impl Ingress { + /// Restore only after resolving and authorizing the configured table. No + /// per-producer gateway cache or additional allocation RPC is required. + pub fn restore_producer(&self, keys: &ReceiptKeys, token: &str) -> Result { + let claims: ProducerClaims = keys.verify(PRODUCER_DOMAIN, token)?; + ensure!( + claims.scope == scope_hash(&self.scope)? + && !claims.open.operation.is_nil() + && claims.open.slot < self.client.producer_slots() + && claims.generation > 0 + && claims.open.expected_generation.checked_add(1) == Some(claims.generation), + "foreign or invalid producer credential" + ); + self.group(claims.open.bucket)?; + Ok(Producer { + scope: self.scope.clone(), + id: lance_context_wal::replication::producer::producer_id( + claims.open.slot, + claims.generation, + ), + generation: claims.generation, + open: claims.open, + }) + } /// 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 { diff --git a/crates/lance-context-ingest/src/tests.rs b/crates/lance-context-ingest/src/tests.rs index 74e53740..9825c6a5 100644 --- a/crates/lance-context-ingest/src/tests.rs +++ b/crates/lance-context-ingest/src/tests.rs @@ -10,6 +10,8 @@ use lance_table::io::commit::{ConditionalPutCommitHandler, ManifestNamingScheme} use receipt::ReceiptKeys; use std::collections::BTreeMap; +#[cfg(all(feature = "http-client", feature = "materializer"))] +mod http; #[cfg(feature = "materializer")] mod materializer; mod network; diff --git a/crates/lance-context-ingest/src/tests/http.rs b/crates/lance-context-ingest/src/tests/http.rs new file mode 100644 index 00000000..1b5539d0 --- /dev/null +++ b/crates/lance-context-ingest/src/tests/http.rs @@ -0,0 +1,398 @@ +use super::*; +use crate::http::{ + self, + journal::{Config, DurableClient}, + Accepted, Gateway, Opened, +}; +use axum::{ + body::to_bytes, + extract::{Request, State}, + middleware::{self, Next}, + response::{IntoResponse, Response}, +}; +use lance_context_wal::client_journal::Settings; +use std::{ + collections::HashMap, + path::Path, + sync::{ + atomic::{AtomicBool, Ordering}, + Mutex, + }, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + sync::oneshot, + task::JoinHandle, +}; + +#[derive(Default)] +struct LostAck { + lost: AtomicBool, + accepted: Mutex>, + producer: Mutex>, +} +async fn lose_one_ack(State(state): State>, request: Request, next: Next) -> Response { + let append = request.uri().path().ends_with("/append"); + let token = request + .headers() + .get(http::PRODUCER) + .and_then(|v| v.to_str().ok()) + .map(str::to_owned); + let response = next.run(request).await; + if append + && response.status() == axum::http::StatusCode::ACCEPTED + && !state.lost.swap(true, Ordering::AcqRel) + { + let bytes = to_bytes(response.into_body(), 16 << 10).await.unwrap(); + *state.accepted.lock().unwrap() = Some(serde_json::from_slice(&bytes).unwrap()); + *state.producer.lock().unwrap() = token; + // The real gateway has durably accepted the append, but the client sees + // an intermediary failure and must retain its journaled original body. + return axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + response +} +struct HttpServer { + gateway: Gateway, + address: std::net::SocketAddr, + stop: oneshot::Sender<()>, + task: JoinHandle>, +} +impl HttpServer { + async fn start( + ingress: Ingress, + lost: Option>, + options: http::Options, + ) -> Result { + let gateway = Gateway::new( + BTreeMap::from([("rollouts".into(), ingress)]), + ReceiptKeys::new(1, [(1, [0x42; 32])])?, + options, + )?; + let mut app = gateway.router(); + if let Some(lost) = lost { + app = app.layer(middleware::from_fn_with_state(lost, lose_one_ack)); + } + let listener = TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let (stop, shutdown) = oneshot::channel(); + let task = tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(async move { + let _ = shutdown.await; + }) + .await + }); + Ok(Self { + gateway, + address, + stop, + task, + }) + } + fn url(&self) -> reqwest::Url { + format!("http://{}/", self.address).parse().unwrap() + } + fn endpoint(&self, action: &str) -> String { + format!("{}api/v1/ingest/rollouts/{action}", self.url()) + } + async fn shutdown(self) -> Result<()> { + self.gateway.shutdown().await; + let _ = self.stop.send(()); + self.task.await??; + Ok(()) + } +} +fn transport() -> Result { + Ok(reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(Duration::from_secs(15)) + .build()?) +} +fn opens() -> Vec { + (0..2) + .map(|bucket| ProducerOpen { + bucket, + slot: bucket, + expected_generation: 0, + operation: Uuid::from_u128(100 + bucket as u128), + }) + .collect() +} +async fn writer( + root: &Path, + server: &HttpServer, + budget: Arc, +) -> Result { + DurableClient::open( + Config { + directory: root.join("journal"), + opens: opens(), + base: server.url(), + request_timeout: Duration::from_secs(15), + journal: Settings { + state_bytes: 16 << 20, + queue_bytes: 8 << 20, + batch_bytes: 2 << 20, + file_bytes: 64 << 10, + log_bytes: 96 << 20, + ..Settings::default() + }, + }, + &layout(), + binding(), + transport()?, + budget, + ) + .await +} +fn data(bucket: u32, round: usize) -> Result { + let session = session(bucket); + ipc_file(&[batch( + &[ + &format!("{bucket}-{round}-a"), + &format!("{bucket}-{round}-b"), + ], + &[&session, &session], + )?]) +} +async fn storage(path: &Path) -> Result { + let base = Dataset::open(path.to_str().unwrap()).await?; + let object_base = base + .manifest_location() + .path + .as_ref() + .rsplit_once("/_versions/") + .context("manifest base")? + .0 + .to_owned(); + Ok(crate::materializer::Storage { + uri: path.to_str().unwrap().into(), + base: object_store::path::Path::parse(object_base)?, + options: HashMap::new(), + session: Arc::new(lance::session::Session::default()), + }) +} +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn lost_http_ack_recovers_shared_journal_across_gateway_and_wal_restart_then_publishes_lance( +) -> Result<()> { + let root = tempfile::tempdir()?; + let path = root.path().join("rollouts.lance"); + let c = network::Cluster::open_signed(root.path(), layout()).await?; + bootstrap(&c, &path, &binding()).await?; + let storage = storage(&path).await?; + let ingress = connect(&c, binding()).await?; + let lost = Arc::new(LostAck::default()); + let server = HttpServer::start( + ingress.clone(), + Some(lost.clone()), + http::Options::default(), + ) + .await?; + let budget = MergeMemoryBudget::new(64 << 20); + let client = writer(root.path(), &server, budget.clone()).await?; + let mut expected = BTreeMap::new(); + let first = data(0, 1)?; + materializer::remember(&mut expected, first.clone())?; + assert!(client.append(0, first.clone()).await.is_err()); + let accepted = lost + .accepted + .lock() + .unwrap() + .clone() + .context("failure was not injected after durable acceptance")?; + assert!(client.last_accepted(0).await?.is_none()); + assert_eq!(client.stage(0, first.clone()).await?, 1); + assert_eq!(client.stage(0, Bytes::copy_from_slice(&first)).await?, 1); + assert!( + client.append(0, data(0, 99)?).await.is_err(), + "unresolved intent was replaced" + ); + let second = data(1, 1)?; + materializer::remember(&mut expected, second.clone())?; + let known = client.append(1, second).await?; + let producer = lost.producer.lock().unwrap().clone().unwrap(); + let keys = ReceiptKeys::new(1, [(1, [0x42; 32])])?; + assert!(ingress.restore_receipt(&keys, &producer).is_err()); + assert!(ingress.restore_producer(&keys, &accepted.receipt).is_err()); + let mut altered = producer.clone().into_bytes(); + altered[20] = if altered[20] == b'A' { b'B' } else { b'A' }; + assert!(ingress + .restore_producer(&keys, std::str::from_utf8(&altered)?) + .is_err()); + client.shutdown().await?; + assert_eq!(budget.reserved(), 0); + server.shutdown().await?; + drop(ingress); + c.shutdown().await?; + + let c = network::Cluster::open_signed(root.path(), layout()).await?; + let ingress = connect(&c, binding()).await?; + let server = HttpServer::start(ingress.clone(), None, http::Options::default()).await?; + let client = writer(root.path(), &server, budget.clone()).await?; + assert_eq!(client.last_accepted(1).await?, Some(known.clone())); + let retried = client + .resume(0) + .await? + .context("lost ACK intent was not recovered")?; + assert_eq!( + retried, accepted, + "gateway failover changed the original ACK" + ); + assert!(client.resume(0).await?.is_none()); + let materializer_budget = MergeMemoryBudget::new(32 << 20); + let mut publisher = + materializer::publisher(&ingress, &storage, &materializer_budget, 1).await?; + let prepared = publisher + .prepare_next() + .await? + .context("missing staged files")?; + // Restart the application endpoints and all original WAL directories with + // a real prepared Lance result but before selecting its manifest. + drop(publisher); + client.shutdown().await?; + server.shutdown().await?; + drop(ingress); + c.shutdown().await?; + + let c = network::Cluster::open_signed(root.path(), layout()).await?; + let ingress = connect(&c, binding()).await?; + let server = HttpServer::start(ingress.clone(), None, http::Options::default()).await?; + let client = writer(root.path(), &server, budget.clone()).await?; + assert_eq!(client.last_accepted(0).await?, Some(accepted.clone())); + let mut publisher = + materializer::publisher(&ingress, &storage, &materializer_budget, 2).await?; + assert_eq!(publisher.prepare_next().await?.unwrap(), prepared); + assert!(publisher.run_once().await?.is_some()); + materializer::verify(&publisher, &expected, 2).await?; + for ack in [&accepted, &known] { + let response: http::Visibility = transport()? + .post(server.endpoint("publication")) + .json(&http::ReceiptRequest { + receipt: ack.receipt.clone(), + }) + .send() + .await? + .error_for_status()? + .json() + .await?; + assert_eq!(response.minimum_read_version, Some(2)); + } + let a = data(0, 2)?; + let b = data(1, 2)?; + materializer::remember(&mut expected, a.clone())?; + materializer::remember(&mut expected, b.clone())?; + let (a, b) = tokio::try_join!(client.append(0, a), client.append(1, b))?; + assert_eq!((a.sequence, b.sequence), (2, 2)); + assert!(client.statistics().requests >= 4); + assert!(publisher.run_once().await?.is_some()); + materializer::verify(&publisher, &expected, 3).await?; + client.shutdown().await?; + let client = writer(root.path(), &server, budget.clone()).await?; + assert_eq!(client.last_accepted(0).await?, Some(a)); + assert_eq!(client.last_accepted(1).await?, Some(b)); + assert!(client.resume(0).await?.is_none()); + assert!(publisher.run_once().await?.is_none()); + materializer::verify(&publisher, &expected, 3).await?; + client.shutdown().await?; + server.shutdown().await?; + drop((publisher, ingress)); + c.shutdown().await?; + assert_eq!(budget.reserved(), 0); + Ok(()) +} +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn saturated_http_data_lane_keeps_control_available_and_drain_joins_disconnected_body( +) -> Result<()> { + let root = tempfile::tempdir()?; + let c = network::Cluster::open_signed(root.path(), layout()).await?; + bootstrap(&c, &root.path().join("rollouts.lance"), &binding()).await?; + let ingress = connect(&c, binding()).await?; + let server = HttpServer::start( + ingress.clone(), + None, + http::Options { + data_requests: 1, + body_idle_timeout: Duration::from_secs(60), + body_total_timeout: Duration::from_secs(60), + ..http::Options::default() + }, + ) + .await?; + let client = transport()?; + let opened: Opened = client + .post(server.endpoint("producer")) + .json(&opens()[0]) + .send() + .await? + .error_for_status()? + .json() + .await?; + let mut slow = TcpStream::connect(server.address).await?; + slow.write_all(format!("POST /api/v1/ingest/rollouts/append HTTP/1.1\r\nHost: localhost\r\nContent-Type: {}\r\nContent-Length: 4096\r\nx-lance-producer: {}\r\nx-lance-sequence: 1\r\n\r\nx", http::ARROW_FILE, opened.producer).as_bytes()).await?; + tokio::time::timeout(Duration::from_secs(5), async { + while server.gateway.observation().data_inflight != 1 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await?; + let overloaded = client + .post(server.endpoint("append")) + .header("content-type", http::ARROW_FILE) + .header(http::PRODUCER, &opened.producer) + .header(http::SEQUENCE, "1") + .body(data(0, 1)?) + .send() + .await?; + assert_eq!( + overloaded.status(), + reqwest::StatusCode::SERVICE_UNAVAILABLE + ); + let opened_again: Opened = client + .post(server.endpoint("producer")) + .json(&opens()[0]) + .send() + .await? + .error_for_status()? + .json() + .await?; + assert_eq!(opened_again.producer, opened.producer); + let gateway = server.gateway.clone(); + let drain = tokio::spawn(async move { gateway.shutdown().await }); + tokio::time::timeout(Duration::from_secs(5), async { + while !server.gateway.observation().draining { + tokio::task::yield_now().await; + } + }) + .await?; + assert!( + !drain.is_finished(), + "shutdown released a still-admitted body reader" + ); + slow.shutdown().await?; + let mut discarded = Vec::new(); + let _ = tokio::time::timeout(Duration::from_secs(5), slow.read_to_end(&mut discarded)).await; + tokio::time::timeout(Duration::from_secs(5), drain).await??; + assert_eq!(server.gateway.observation().data_inflight, 0); + server.shutdown().await?; + let server = HttpServer::start(ingress.clone(), None, http::Options::default()).await?; + let accepted: Accepted = client + .post(server.endpoint("append")) + .header("content-type", http::ARROW_FILE) + .header(http::PRODUCER, &opened.producer) + .header(http::SEQUENCE, "1") + .body(data(0, 1)?) + .send() + .await? + .error_for_status()? + .json() + .await?; + assert_eq!( + accepted.sequence, 1, + "partial body advanced the producer sequence" + ); + server.shutdown().await?; + drop(ingress); + c.shutdown().await +} diff --git a/crates/lance-context-ingest/src/tests/materializer.rs b/crates/lance-context-ingest/src/tests/materializer.rs index 87789de3..7c2808b9 100644 --- a/crates/lance-context-ingest/src/tests/materializer.rs +++ b/crates/lance-context-ingest/src/tests/materializer.rs @@ -4,7 +4,7 @@ use futures::TryStreamExt; use object_store::path::Path as ObjectPath; use std::collections::HashMap; -async fn publisher( +pub(super) async fn publisher( ingress: &Ingress, storage: &Storage, budget: &Arc, @@ -49,7 +49,7 @@ async fn append( let accepted = ingress.append(&prepared).await?; Ok((body, accepted)) } -async fn verify( +pub(super) async fn verify( publisher: &Publisher, expected: &BTreeMap, version: u64, @@ -90,7 +90,7 @@ async fn verify( ); Ok(()) } -fn remember(expected: &mut BTreeMap, body: Bytes) -> Result<()> { +pub(super) fn remember(expected: &mut BTreeMap, body: Bytes) -> Result<()> { for batch in ipc::decode(body, Arc::new(rollout_schema()))? { let ids = batch .column_by_name("id") diff --git a/docs/design/wal-http-ingest.md b/docs/design/wal-http-ingest.md new file mode 100644 index 00000000..421659e4 --- /dev/null +++ b/docs/design/wal-http-ingest.md @@ -0,0 +1,109 @@ +# HTTP rollout ingestion and recoverable Rust producers + +The optional `lance-context-ingest/http` feature supplies an Axum gateway over +the public TLS WAL client. `http-client` adds a Rust producer with a persistent, +shared group-commit journal. Both retain the same rollout schema, session bucket, +producer generation and exact-retry contract as `Ingress`. No Lance OSS changes +are required. + +The application resolves authorized, immutable table bindings before constructing +`Gateway`, and installs its router behind the application's access-control layer. +The configured map is a fixed set of qualified table names, not a namespace +registry or a mechanism for migrating/rebinding live legacy tables. Multiple +gateways can serve the same binding. They share persistent receipt/producer +signing keys, and each process shares physical WAL client pools and a memory +budget across its ingresses. + +## Requests + +All request bodies require a single valid Content-Length and Content-Type. +Control bodies are JSON bounded to 32 KiB. Arrow append bodies use +`application/vnd.apache.arrow.file` and are bounded by the registered contract. +Chunked uploads, conflicting headers, trailers and mismatched lengths fail +before an append is submitted. Proxy request/header/connection limits remain an +application deployment responsibility. + +- `POST /api/v1/ingest/{table}/producer` accepts the persisted `ProducerOpen` + tuple. It returns a generation, scope fingerprint and authenticated producer + credential after durable allocation. Retry an uncertain open with the same + slot, expected generation, operation UUID and bucket. +- `POST /api/v1/ingest/{table}/append` carries the original Arrow IPC body, an + `x-lance-producer` credential and an `x-lance-sequence` header. A 202 response + carries a durable receipt, original WAL position, sequence, row count and body + digest. It does not wait for Lance publication. +- `POST /api/v1/ingest/{table}/publication` accepts a receipt and returns an + optional minimum visible Lance version from a compact quorum query. +- `POST /api/v1/ingest/{table}/close` accepts a producer credential and its last + receipt, if any. The caller retains the original close state on uncertainty. + +Producer and receipt credentials have distinct authenticated domains and include +the complete binding fingerprint. They cannot be interchanged. A producer +credential is portable across gateways; append does not require a gateway-local +producer cache or another allocation round trip. The WAL still fences stale +producer generations. These credentials do not authorize end-user access. + +## Resource and cancellation behavior + +Data and control handlers have independent bounded admission. A saturated data +lane cannot consume the control slots needed for producer/visibility operations. +Append reserves shared workspace before reading its body. Reads have idle and +total deadlines, and memory admission has a bounded wait. A single HTTP frame +retains its Bytes allocation; fragmented bodies are coalesced once. Validation, +digest computation and WAL transport then share the same immutable body. + +Admitted handlers run independently of the HTTP socket and retain their guards +through completion. `Gateway::shutdown` closes admission and joins those +handlers, including disconnected callers. It does not abort a submitted write. +The application also stops/drains its HTTP listener. The gateway reports local +in-flight counts, which are not a quorum or publication-health assertion. + +## Durable client recovery + +`http::journal::DurableClient` opens one bounded local journal for an ordered set +of producer lanes. Those lanes share the journal's group commit, HTTP pool and +memory reservation. Each lane is serialized; different lanes can make progress +concurrently. The parent journal directory must exist on persistent storage. +The client does not make this journal replicated or reassign ownership of its +volume to another process. + +Persist producer-open tuples in application configuration. Reopening requires +the same layout, complete table binding, ordered producer-open tuples and journal +settings. Changing the gateway endpoint is allowed. The supplied HTTP client +must target trusted gateways with redirects disabled. + +`stage` journals the exact body and sequence before any append. `append` combines +staging and transmission. `resume` resends an existing staged/recovered intent; +it never allocates a new sequence after an uncertain response. The client +validates the returned scope/generation and ACK sequence/group/digest before +journaling the ACK, releasing the pending body, or advancing its sequence. +`last_accepted` exposes the durable latest ACK after a restart; calling `append` +again after a known ACK is a new application batch, not a request-ID retry. + +A network failure leaves the original intent available. A failed or cancelled +local journal operation fences that lane until the journal is reopened, because +the disk update may have completed without an observed reply. A different body +cannot replace an unresolved intent. Explicit shutdown joins the disk worker off +the async executor. Recovery parsing and payload verification also run off the +async executor. This is bounded logical accounting, not zero-copy disk encoding +or an allocator/cgroup hard quota. + +## Qualification and remaining integration + +The new qualification connects real HTTP listeners to real TLS replicas and a +real rollout Lance dataset. It injects an intermediary error after a durable +append, restarts the client journal, gateway and original replica directories, +checks the exact original ACK, restarts again with prepared Lance files, then +publishes and verifies complete rows and subsequent journaled sequences. A +second test saturates the data lane with an incomplete body, exercises control +requests, and checks that draining joins the body reader without consuming a +producer sequence. Both integration checks passed with the ingress/materializer +tests, HTTP feature Clippy checks and client-journal recovery regressions. The +application restart checks use joined shutdown/reopen; they are not process +SIGKILL or sustained throughput measurements. + +This supplies application library interfaces, not a complete deployable gateway +or materializer worker. Bootstrap/namespace integration, distributed worker +scheduling and discovery, application readiness and shutdown orchestration, and +sustained Kubernetes/blob performance qualification remain integration work. + +Tracking: #378.