diff --git a/crates/lance-context-core/src/merge_write_scope.rs b/crates/lance-context-core/src/merge_write_scope.rs index 1308af8..5d5d19a 100644 --- a/crates/lance-context-core/src/merge_write_scope.rs +++ b/crates/lance-context-core/src/merge_write_scope.rs @@ -141,11 +141,22 @@ impl Drop for Active { } } +#[cfg(test)] pub(crate) async fn shield(write: F) -> std::result::Result where F: Future> + Send + 'static, T: Send + 'static, E: From + Send + 'static, +{ + shield_observed(write, |result| result.is_ok()).await +} + +async fn shield_observed(write: F, progressed: P) -> std::result::Result +where + F: Future> + Send + 'static, + T: Send + 'static, + E: From + Send + 'static, + P: FnOnce(&std::result::Result) -> bool + Send + 'static, { let Ok(scope) = CURRENT.try_with(Arc::clone) else { return write.await; @@ -166,7 +177,7 @@ where let leaf = tokio::spawn(async move { let mut active = active; let result = write.await; - if result.is_ok() { + if progressed(&result) { active.scope.completed_steps.fetch_add(1, Ordering::Relaxed); } if result.is_err() { @@ -209,18 +220,21 @@ pub(crate) async fn drain_generations( authorize(&resource, next.version).await?; let writer = store.clone(); let scope = CURRENT.try_with(Arc::clone).ok(); - let (next, written) = shield(async move { - let written = writer.write(&next).await; - if written - .as_ref() - .is_err_and(|e| !e.to_string().contains("already exists")) - { - if let Some(scope) = scope { - scope.mark_uncertain(); + let (next, written) = shield_observed( + async move { + let written = writer.write(&next).await; + if written + .as_ref() + .is_err_and(|e| !e.to_string().contains("already exists")) + { + if let Some(scope) = scope { + scope.mark_uncertain(); + } } - } - Ok::<_, Error>((next, written)) - }) + Ok::<_, Error>((next, written)) + }, + |result| result.as_ref().is_ok_and(|(_, written)| written.is_ok()), + ) .await?; match written { Ok(_) => return Ok(next), @@ -384,28 +398,31 @@ impl CommitHandler for GuardedCommit { let path = base_path.clone(); let store = object_store.clone(); let scope = CURRENT.try_with(Arc::clone).ok(); - let (location, committed) = shield(async move { - let location = inner - .commit( - &mut owned_manifest, - indices, - &path, - &store, - manifest_writer, - naming_scheme, - transaction, - ) - .await; - // Only a definite conflict proves this conditional commit did - // not happen. A transport/storage error can arrive after the - // remote service accepted it, even though the Rust future ended. - if matches!(&location, Err(CommitError::OtherError(_))) { - if let Some(scope) = scope { - scope.mark_uncertain(); + let (location, committed) = shield_observed( + async move { + let location = inner + .commit( + &mut owned_manifest, + indices, + &path, + &store, + manifest_writer, + naming_scheme, + transaction, + ) + .await; + // Only a definite conflict proves this conditional commit did + // not happen. A transport/storage error can arrive after the + // remote service accepted it, even though the Rust future ended. + if matches!(&location, Err(CommitError::OtherError(_))) { + if let Some(scope) = scope { + scope.mark_uncertain(); + } } - } - Ok::<_, CommitError>((location, owned_manifest)) - }) + Ok::<_, CommitError>((location, owned_manifest)) + }, + |result| result.as_ref().is_ok_and(|(location, _)| location.is_ok()), + ) .await?; *manifest = committed; location @@ -503,6 +520,21 @@ mod tests { } } + #[tokio::test] + async fn conditional_conflicts_and_errors_do_not_count_as_progress() { + let scope = MergeWriteScope::new(); + for committed in [false, false, true] { + scope + .run(shield_observed( + async move { Ok::<_, Error>(committed) }, + |result| matches!(result, Ok(true)), + )) + .await + .unwrap(); + } + assert_eq!(scope.completed_steps(), 1); + } + #[tokio::test] async fn confirmed_fence_can_abort_a_permanently_stalled_leaf() { struct Writing(Arc); diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 44736a7..3568079 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -1209,7 +1209,6 @@ impl StorageBase { let mut current_batches = Vec::new(); let mut stream = gen_dataset.scan().try_into_stream().await?; while let Some(batch) = stream.try_next().await? { - crate::merge_write_scope::checkpoint(); if batch.num_rows() > 0 { let batch = align_batch_to_schema(batch, merge_schema.clone())?; buffered_bytes = buffered_bytes.saturating_add(batch.get_array_memory_size()); @@ -1229,6 +1228,7 @@ impl StorageBase { } } current_batches.push(batch); + crate::merge_write_scope::checkpoint(); } } generation_batches.push((flushed.generation, current_batches)); diff --git a/crates/lance-context-master/src/catchup/executor.rs b/crates/lance-context-master/src/catchup/executor.rs new file mode 100644 index 0000000..6cf4ca9 --- /dev/null +++ b/crates/lance-context-master/src/catchup/executor.rs @@ -0,0 +1,162 @@ +use super::Result; +use crate::{config::MasterConfig, maintenance_execution, state::MasterState}; +use lance_context_api::TaskKind; +use lance_context_core::{ + GenericStore, GenericStoreOptions, MergeMemoryBudget, RolloutStore, RolloutStoreOptions, +}; +use lance_context_merge::MaintenanceKind; +use std::{sync::Arc, time::Duration}; + +/// One dedicated process, one table, serial shards, bounded passes. No server, +/// scanner, ingestion writer, or fleet task consumer starts in this mode. +pub async fn execute(mut config: MasterConfig, target: &str) -> Result<()> { + config.catchup.enabled = false; + config.catchup.validate()?; + if !config.merge_rollout.owned(target) || config.merge_rollout.draining(target) { + return Err("catch-up executor requires a non-draining owned target".into()); + } + let job = config + .catchup + .job_name + .clone() + .ok_or("missing catch-up Job identity")?; + let state = MasterState::new(config).await.map_err(|e| e.to_string())?; + // Enqueue through the normal durable queue, then claim only this table. + state + .task_store + .enqueue(TaskKind::MergeWal, target, Vec::new()) + .await + .map_err(|e| e.to_string())?; + let claim = tokio::time::timeout(Duration::from_secs(30), async { + loop { + if let Some(claim) = state + .task_store + .claim_merge_target(target, &job) + .await + .map_err(|e| e.to_string())? + { + return Ok::<_, String>(claim); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + }) + .await + .map_err(|_| "catch-up admission remained busy for 30 seconds")??; + let result = async { + // The claim CAS has excluded any live reconciler. A previous process + // may still have a storage PUT in flight: fence before opening writers. + let coordinator = state.task_store.merge_coordinator(); + if let Some(old) = coordinator.get(target).await? { + crate::merge_execution::ensure_recovery_due(&coordinator, &old).await?; + crate::merge_execution::recover_execution( + &state, + &coordinator, + &state.task_store.merge_claim(&claim), + old, + ) + .await?; + } + maintenance_execution::run_as( + &state, + &claim, + MaintenanceKind::Catchup, + merge_passes(&state, target), + ) + .await + } + .await; + let error = result.as_ref().err().cloned(); + state + .task_store + .finish(claim, result) + .await + .map_err(|e| e.to_string())?; + if let Some(error) = error { + return Err(error); + } + Ok(()) +} + +async fn merge_passes(state: &Arc, target: &str) -> Result { + let config = &state.config.catchup; + let session = RolloutStore::build_session(96 * 1024 * 1024, 32 * 1024 * 1024); + let budget = MergeMemoryBudget::new(config.merge_memory_bytes); + let mut total = 0usize; + let started = tokio::time::Instant::now(); + // This is only an admission time slice. A progressing operation already + // admitted may run beyond it; cancellation is based on lack of progress. + let admit_for = Duration::from_secs(config.slice_secs); + // Bound each slice even under continuous ingestion; later fresh stats may + // request another Job. Visit every shard before repeating any hot shard. + for _ in 0..16 { + let before = total; + for shard in &config.shards { + if started.elapsed() >= admit_for { + return Ok(format!( + "dedicated catch-up slice merged {total} generations" + )); + } + let reclaimed = if let Some(name) = target.strip_prefix("generic:") { + let options = GenericStoreOptions { + shard_id: Some(shard.clone()), + merge_after_generations: Some(0), + merge_max_generations: Some(8), + merge_max_bytes: Some(config.merge_max_bytes), + merge_budget: Some(budget.clone()), + session: Some(session.clone()), + pending_generations_max: Some(0), + ..Default::default() + }; + let mut store = GenericStore::open_existing(&state.generic_uri(name), options) + .await + .map_err(|e| e.to_string())?; + // No resident writer was opened, so cleanup only sees sealed WAL. + match store + .prepare_cleanup_merge() + .await + .map_err(|e| e.to_string())? + { + Some((manifest_store, manifest, prepared)) => store + .commit_prepared_merge(&manifest_store, &manifest, prepared) + .await + .map_err(|e| e.to_string())?, + None => 0, + } + } else { + let options = RolloutStoreOptions { + shard_id: Some(shard.clone()), + merge_after_generations: Some(0), + merge_max_generations: Some(8), + merge_max_bytes: Some(config.merge_max_bytes), + merge_budget: Some(budget.clone()), + session: Some(session.clone()), + pending_generations_max: Some(0), + ..Default::default() + }; + let mut store = + RolloutStore::open_existing_with_options(&state.rollout_uri(target), options) + .await + .map_err(|e| e.to_string())?; + match store + .prepare_merge_if_ready(1) + .await + .map_err(|e| e.to_string())? + { + Some((manifest_store, manifest, prepared)) => store + .commit_prepared_merge(&manifest_store, &manifest, prepared) + .await + .map_err(|e| e.to_string())?, + None => 0, + } + }; + total += reclaimed; + metrics::counter!("master_catchup_generations_reclaimed_total") + .increment(reclaimed as u64); + tracing::info!(%target, %shard, reclaimed, total, "dedicated catch-up committed"); + } + if total == before { + break; + } + } + Ok(format!("dedicated catch-up merged {total} generations")) +} diff --git a/crates/lance-context-master/src/catchup/kubernetes.rs b/crates/lance-context-master/src/catchup/kubernetes.rs new file mode 100644 index 0000000..003c5ca --- /dev/null +++ b/crates/lance-context-master/src/catchup/kubernetes.rs @@ -0,0 +1,427 @@ +use super::{store::Record, CatchupConfig, Result}; +use crate::config::MasterConfig; +use reqwest::{Client, Method, StatusCode}; +use serde_json::{json, Value}; +use std::{path::Path, time::Duration}; + +const SERVICE_ACCOUNT: &str = "/var/run/secrets/kubernetes.io/serviceaccount"; + +pub(super) fn read_template(config: &CatchupConfig) -> Result { + let path = config + .pod_template + .as_ref() + .ok_or("missing catch-up PodSpec template")?; + let bytes = std::fs::read(path).map_err(|e| format!("read catch-up template: {e}"))?; + if bytes.len() > 65_536 { + return Err("catch-up template exceeds 64 KiB".into()); + } + let spec: Value = serde_json::from_slice(&bytes).map_err(|e| e.to_string())?; + validate_template(&spec)?; + Ok(spec) +} +fn validate_template(spec: &Value) -> Result<()> { + let containers = spec["containers"] + .as_array() + .ok_or("template requires containers")?; + if containers.len() != 1 + || spec.get("initContainers").is_some() + || spec.get("ephemeralContainers").is_some() + || spec.get("nodeName").is_some() + { + return Err("catch-up requires one container, no init containers or nodeName".into()); + } + let c = &containers[0]; + let image = c["image"].as_str().ok_or("missing image")?; + if !image.rsplit_once("@sha256:").is_some_and(|(_, digest)| { + digest.len() == 64 && digest.bytes().all(|b| b.is_ascii_hexdigit()) + }) { + return Err("catch-up image must be pinned by SHA-256 digest".into()); + } + if !c["command"].as_array().is_some_and(|a| { + a.len() == 1 + && a[0] + .as_str() + .is_some_and(|s| s.ends_with("lance-context-master")) + }) { + return Err("catch-up command must run the native lance-context-master binary".into()); + } + for resource in ["cpu", "memory"] { + let request = &c["resources"]["requests"][resource]; + let limit = &c["resources"]["limits"][resource]; + if request.as_str().is_none_or(|s| s.is_empty() || s == "0") || request != limit { + return Err("catch-up requires equal, nonzero CPU/memory requests and limits".into()); + } + } + Ok(()) +} + +pub(super) struct Kubernetes { + client: Client, + base: String, + token_file: String, + pods_base: String, +} +impl Kubernetes { + pub fn in_cluster(config: &CatchupConfig) -> Result { + let ca = std::fs::read(Path::new(SERVICE_ACCOUNT).join("ca.crt")) + .map_err(|e| format!("read Kubernetes CA: {e}"))?; + let client = Client::builder() + .timeout(Duration::from_secs(10)) + .redirect(reqwest::redirect::Policy::none()) + .add_root_certificate(reqwest::Certificate::from_pem(&ca).map_err(|e| e.to_string())?) + .build() + .map_err(|e| e.to_string())?; + Ok(Self { + client, + base: format!( + "https://kubernetes.default.svc/apis/batch/v1/namespaces/{}/jobs", + config.namespace + ), + token_file: format!("{SERVICE_ACCOUNT}/token"), + pods_base: format!( + "https://kubernetes.default.svc/api/v1/namespaces/{}/pods", + config.namespace + ), + }) + } + async fn request( + &self, + method: Method, + url: &str, + body: Option<&Value>, + ) -> Result<(StatusCode, Value)> { + // Re-read projected tokens so long-lived masters survive token rotation. + let token = std::fs::read_to_string(&self.token_file) + .map_err(|e| format!("read Kubernetes token: {e}"))?; + let patch = method == Method::PATCH; + let mut request = self.client.request(method, url).bearer_auth(token.trim()); + if patch { + request = request.header("Content-Type", "application/json-patch+json"); + } + if let Some(body) = body { + request = request.json(body); + } + let response = request.send().await.map_err(|e| e.to_string())?; + let status = response.status(); + let body = response.json().await.map_err(|e| e.to_string())?; + Ok((status, body)) + } + pub async fn reconcile( + &self, + config: &MasterConfig, + record: &Record, + stop: bool, + ) -> Result<(Option, Option)> { + let (status, mut job) = self + .request(Method::GET, &format!("{}/{}", self.base, record.job), None) + .await?; + if status == StatusCode::NOT_FOUND { + if stop || record.job_uid.is_some() { + return Err( + "catch-up Job missing after confirmation or startup stall; reservation retained" + .into(), + ); + } + let desired = render_job(config, record, read_template(&config.catchup)?); + let (created, body) = self + .request(Method::POST, &self.base, Some(&desired)) + .await?; + // Ambiguous response and conflict both reconcile by the same name. + if created == StatusCode::CONFLICT { + return Ok((None, None)); + } + if !created.is_success() { + return Err(format!("Kubernetes Job creation returned {created}")); + } + job = body; + } else if !status.is_success() { + return Err(format!("Kubernetes Job lookup returned {status}")); + } + if job["metadata"]["annotations"]["lance-context/target"] != record.target + || job["metadata"]["labels"]["lance-context/catchup"] != "native-v1" + { + return Err("Job identity mismatch; refusing adoption".into()); + } + let uid = job["metadata"]["uid"] + .as_str() + .ok_or("Job has no UID")? + .to_string(); + if record + .job_uid + .as_ref() + .is_some_and(|expected| expected != &uid) + { + return Err("Job UID changed; refusing adoption or termination".into()); + } + let terminal = job["status"]["conditions"] + .as_array() + .and_then(|conditions| { + conditions.iter().find_map(|c| { + if c["status"] != "True" { + return None; + } + match c["type"].as_str() { + Some("Complete") => Some(true), + Some("Failed") => Some(false), + _ => None, + } + }) + }); + let Some(success) = terminal else { + if stop && record.job_uid.is_some() { + let uid = job["metadata"]["uid"].as_str().ok_or("Job has no UID")?; + let version = job["metadata"]["resourceVersion"] + .as_str() + .ok_or("Job has no resourceVersion")?; + // Only a confirmed stall enables a deadline. CAS prevents a + // delayed controller from touching a replaced/changed Job. + let patch = json!([ + {"op":"test","path":"/metadata/uid","value":uid}, + {"op":"test","path":"/metadata/resourceVersion","value":version}, + {"op":"add","path":"/spec/activeDeadlineSeconds","value":1} + ]); + let (status, _) = self + .request( + Method::PATCH, + &format!("{}/{}", self.base, record.job), + Some(&patch), + ) + .await?; + if !status.is_success() { + return Err(format!("stalled Job termination returned {status}")); + } + tracing::warn!(target = %record.target, job = %record.job, + "confirmed no-progress catch-up; requested Job termination, retaining ownership"); + } + return Ok((Some(uid), None)); + }; + // Older Kubernetes versions publish Failed while Pods still terminate. + // Check Job UID and every Pod phase before giving the resource slot back. + if !uid.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-') { + return Err("invalid Job UID".into()); + } + let url = format!( + "{}?labelSelector=batch.kubernetes.io%2Fcontroller-uid%3D{uid}&limit=16", + self.pods_base + ); + let (status, pods) = self.request(Method::GET, &url, None).await?; + if !status.is_success() + || pods["metadata"]["continue"] + .as_str() + .is_some_and(|s| !s.is_empty()) + { + return Err("cannot confirm catch-up Pods terminated".into()); + } + let items = pods["items"].as_array().ok_or("invalid Pod list")?; + if items + .iter() + .any(|p| !matches!(p["status"]["phase"].as_str(), Some("Succeeded" | "Failed"))) + { + return Ok((Some(uid), None)); + } + Ok((Some(uid), Some(success))) + } +} + +pub(super) fn render_job(config: &MasterConfig, record: &Record, mut spec: Value) -> Value { + let mut env = spec["containers"][0]["env"] + .as_array() + .cloned() + .unwrap_or_default(); + let mut shards = config.catchup.shards.clone(); + if !shards.is_empty() { + let start = record.attempt.saturating_sub(1) as usize % shards.len(); + shards.rotate_left(start); + } + let overrides = [ + ("DATA_DIR", config.data_dir.clone()), + ("ETCD_ENDPOINTS", config.etcd.etcd_endpoints.join(",")), + ("ETCD_PREFIX", config.etcd.etcd_prefix.clone()), + ("CATCHUP_ENABLED", "false".into()), + ("CATCHUP_TARGET", record.target.clone()), + ("CATCHUP_JOB_NAME", record.job.clone()), + ("CATCHUP_SHARDS", shards.join(",")), + ( + "CATCHUP_MERGE_MAX_BYTES", + config.catchup.merge_max_bytes.to_string(), + ), + ( + "CATCHUP_MERGE_MEMORY_BYTES", + config.catchup.merge_memory_bytes.to_string(), + ), + ("MERGE_OWNED_TARGETS", record.target.clone()), + ("MERGE_DRAIN_TARGETS", String::new()), + ("CATCHUP_SLICE_SECS", config.catchup.slice_secs.to_string()), + ( + "MAINTENANCE_IDLE_TIMEOUT_SECS", + config.maintenance.maintenance_idle_timeout_secs.to_string(), + ), + ("ROLLOUT_CACHE_BYTES", "134217728".into()), + ]; + for (name, value) in overrides { + env.retain(|e| e["name"] != name); + env.push(json!({"name": name, "value": value})); + } + spec["containers"][0]["env"] = json!(env); + spec["containers"][0]["args"] = json!([]); + spec["restartPolicy"] = json!("Never"); + spec["preemptionPolicy"] = json!("Never"); + spec["terminationGracePeriodSeconds"] = json!(30); + spec.as_object_mut() + .expect("validated PodSpec") + .remove("activeDeadlineSeconds"); + // Dedicated executors need storage/etcd, never Kubernetes permissions. + spec["automountServiceAccountToken"] = json!(false); + json!({"apiVersion":"batch/v1","kind":"Job", "metadata":{"name":record.job, + "namespace":config.catchup.namespace,"labels":{"lance-context/catchup":"native-v1"}, + "annotations":{"lance-context/target":record.target}}, + "spec":{"parallelism":1,"completions":1,"backoffLimit":0, + "ttlSecondsAfterFinished":86400, + "template":{"metadata":{"labels":{"lance-context/catchup":"native-v1"}},"spec":spec}}}) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{extract::State, routing::get, Json, Router}; + use clap::Parser; + use std::sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, + }; + struct Mock { + creates: AtomicUsize, + terminal: AtomicBool, + patches: AtomicUsize, + pods_stopped: AtomicBool, + job: std::sync::Mutex>, + } + async fn create( + State(mock): State>, + Json(mut job): Json, + ) -> (StatusCode, Json) { + mock.creates.fetch_add(1, Ordering::SeqCst); + job["metadata"]["uid"] = json!("test-uid"); + job["metadata"]["resourceVersion"] = json!("1"); + *mock.job.lock().unwrap() = Some(job); + // The API committed the Job but the caller received an error. + (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({}))) + } + async fn lookup(State(mock): State>) -> (StatusCode, Json) { + let Some(mut job) = mock.job.lock().unwrap().clone() else { + return (StatusCode::NOT_FOUND, Json(json!({}))); + }; + if mock.terminal.load(Ordering::SeqCst) { + job["status"] = json!({"conditions":[{"type":"Failed","status":"True"}]}); + } + (StatusCode::OK, Json(job)) + } + async fn patch( + State(mock): State>, + Json(patch): Json, + ) -> (StatusCode, Json) { + let mut guard = mock.job.lock().unwrap(); + let job = guard.as_mut().unwrap(); + assert_eq!(patch[0]["path"], "/metadata/uid"); + assert_eq!(patch[1]["path"], "/metadata/resourceVersion"); + if patch[0]["value"] != job["metadata"]["uid"] + || patch[1]["value"] != job["metadata"]["resourceVersion"] + { + return (StatusCode::CONFLICT, Json(json!({}))); + } + assert_eq!(patch[2]["value"], 1); + mock.patches.fetch_add(1, Ordering::SeqCst); + job["spec"]["activeDeadlineSeconds"] = json!(1); + (StatusCode::OK, Json(job.clone())) + } + async fn pods(State(mock): State>) -> Json { + Json( + json!({"items":[{"status":{"phase":if mock.pods_stopped.load(Ordering::SeqCst) {"Failed"} else {"Running"}}}]}), + ) + } + #[tokio::test] + async fn ambiguous_create_is_adopted_and_terminating_pods_retain_capacity() { + let mock = Arc::new(Mock { + creates: AtomicUsize::new(0), + terminal: AtomicBool::new(false), + patches: AtomicUsize::new(0), + pods_stopped: AtomicBool::new(false), + job: std::sync::Mutex::new(None), + }); + let app = Router::new() + .route("/jobs", axum::routing::post(create)) + .route("/jobs/{name}", get(lookup).patch(patch)) + .route("/pods", get(pods)) + .with_state(mock.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let dir = tempfile::tempdir().unwrap(); + let token = dir.path().join("token"); + std::fs::write(&token, "test-token").unwrap(); + let template = dir.path().join("pod.json"); + std::fs::write(&template, json!({"containers":[{"name":"executor","command":["lance-context-master"],"image":format!("native@sha256:{}","a".repeat(64)),"resources":{"requests":{"cpu":"1","memory":"8Gi"},"limits":{"cpu":"1","memory":"8Gi"}}}]}).to_string()).unwrap(); + let mut config = MasterConfig::parse_from(["master"]); + config.catchup.pod_template = Some(template.to_string_lossy().into()); + let client = Kubernetes { + client: Client::new(), + base: format!("http://{addr}/jobs"), + pods_base: format!("http://{addr}/pods"), + token_file: token.to_string_lossy().into(), + }; + let mut record = Record { + target: "hot".into(), + job: "lc-catchup-test".into(), + job_uid: None, + slot: 0, + attempt: 1, + active: true, + termination_requested: false, + reason: "test".into(), + pending_at_admission: 1000, + created_at_ms: 0, + finished_at_ms: None, + consecutive_failures: 0, + needs_attention: false, + next_retry_ms: 0, + outcome: None, + progress: None, + }; + assert!(client.reconcile(&config, &record, false).await.is_err()); + assert_eq!( + client.reconcile(&config, &record, false).await.unwrap().1, + None + ); + assert_eq!(mock.creates.load(Ordering::SeqCst), 1); + record.job_uid = Some("test-uid".into()); + assert_eq!(mock.patches.load(Ordering::SeqCst), 0); + client.reconcile(&config, &record, true).await.unwrap(); + assert_eq!(mock.patches.load(Ordering::SeqCst), 1); + record.job_uid = Some("replaced-uid".into()); + assert!(client + .reconcile(&config, &record, true) + .await + .unwrap_err() + .contains("UID changed")); + assert_eq!(mock.patches.load(Ordering::SeqCst), 1); + record.job_uid = Some("test-uid".into()); + mock.terminal.store(true, Ordering::SeqCst); + assert_eq!( + client.reconcile(&config, &record, false).await.unwrap().1, + None + ); + mock.pods_stopped.store(true, Ordering::SeqCst); + assert_eq!( + client.reconcile(&config, &record, false).await.unwrap().1, + Some(false) + ); + server.abort(); + } + #[test] + fn invalid_or_unbounded_templates_are_rejected() { + assert!(validate_template(&json!({"containers":[]})).is_err()); + assert!(validate_template(&json!({"containers":[{"image":"native:latest"}]})).is_err()); + } +} diff --git a/crates/lance-context-master/src/catchup/mod.rs b/crates/lance-context-master/src/catchup/mod.rs new file mode 100644 index 0000000..c6d91df --- /dev/null +++ b/crates/lance-context-master/src/catchup/mod.rs @@ -0,0 +1,442 @@ +//! Master-owned, bounded Kubernetes catch-up jobs. No payload reads in admission. +mod executor; +mod kubernetes; +mod progress; +pub(crate) mod store; +#[cfg(test)] +mod tests; + +use crate::{config::MasterConfig, error::MasterError, state::MasterState, stats_store::StatRow}; +use axum::{ + extract::{Query, State}, + Json, +}; +pub use executor::execute; +use kubernetes::Kubernetes; +use serde::{Deserialize, Serialize}; +use std::{sync::Arc, time::Duration}; +use store::{Inventory, Record}; + +type Result = std::result::Result; + +#[derive(Clone, Debug, clap::Args)] +pub struct CatchupConfig { + /// Enable autonomous admission. Manual requests use the same policy. + #[arg(long, env = "CATCHUP_ENABLED", default_value_t = false)] + pub enabled: bool, + /// Trusted PodSpec JSON. One container, pinned image, explicit resource bounds. + #[arg(long, env = "CATCHUP_POD_TEMPLATE")] + pub pod_template: Option, + #[arg(long, env = "CATCHUP_NAMESPACE", default_value = "default")] + pub namespace: String, + #[arg(long, env = "CATCHUP_MAX_JOBS", default_value_t = 4)] + pub max_jobs: usize, + #[arg(long, env = "CATCHUP_MIN_PENDING", default_value_t = 256)] + pub min_pending: i64, + #[arg(long, env = "CATCHUP_STATS_MAX_AGE_SECS", default_value_t = 900)] + pub stats_max_age_secs: u64, + #[arg(long, env = "CATCHUP_INTERVAL_SECS", default_value_t = 30)] + pub interval_secs: u64, + #[arg(long, env = "CATCHUP_SLICE_SECS", default_value_t = 1800)] + pub slice_secs: u64, + /// Maximum time without an execution starting, separate from merge progress. + #[arg(long, env = "CATCHUP_STARTUP_TIMEOUT_SECS", default_value_t = 1800)] + pub startup_timeout_secs: u64, + /// Stable writer identities, not Pod IPs. Only sealed generations are merged. + #[arg(long, env = "CATCHUP_SHARDS", value_delimiter = ',')] + pub shards: Vec, + #[arg(long, env = "CATCHUP_MERGE_MAX_BYTES", default_value_t = 67_108_864)] + pub merge_max_bytes: usize, + #[arg( + long, + env = "CATCHUP_MERGE_MEMORY_BYTES", + default_value_t = 1_073_741_824 + )] + pub merge_memory_bytes: usize, + /// Native one-table executor mode; does not start the admin server or scanner. + #[arg(long, env = "CATCHUP_TARGET", hide = true)] + pub target: Option, + #[arg(long, env = "CATCHUP_JOB_NAME", hide = true)] + pub job_name: Option, +} +impl Default for CatchupConfig { + fn default() -> Self { + Self { + enabled: false, + pod_template: None, + namespace: "default".into(), + max_jobs: 4, + min_pending: 256, + stats_max_age_secs: 900, + interval_secs: 30, + slice_secs: 1800, + startup_timeout_secs: 1800, + shards: Vec::new(), + merge_max_bytes: 67_108_864, + merge_memory_bytes: 1_073_741_824, + target: None, + job_name: None, + } + } +} +impl CatchupConfig { + pub fn validate(&self) -> Result<()> { + if !self.enabled && self.target.is_none() && self.pod_template.is_none() { + return Ok(()); + } + if self.max_jobs == 0 + || self.max_jobs > 256 + || self.min_pending < 1 + || self.interval_secs == 0 + || self.stats_max_age_secs == 0 + || self.startup_timeout_secs < 120 + || self.startup_timeout_secs > 86_400 + || self.slice_secs < 120 + || self.slice_secs > 86_400 + || self.merge_max_bytes == 0 + || self.merge_memory_bytes < self.merge_max_bytes + || self.shards.is_empty() + || self.shards.len() > 256 + || self.shards.iter().any(|s| s.is_empty()) + { + return Err("invalid catch-up limits or shard identities".into()); + } + if self.enabled && self.pod_template.is_none() { + return Err("catch-up requires CATCHUP_POD_TEMPLATE".into()); + } + if self.target.is_some() && self.job_name.is_none() { + return Err("executor requires CATCHUP_JOB_NAME".into()); + } + if self.namespace.is_empty() + || self.namespace.len() > 63 + || !self + .namespace + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') + { + return Err("invalid catch-up namespace".into()); + } + if self.enabled { + kubernetes::read_template(self)?; + } + Ok(()) + } +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Trigger { + pub target: String, + pub reason: String, + #[serde(default)] + pub dry_run: bool, +} +#[derive(Debug, Serialize)] +pub struct Decision { + pub target: String, + pub decision: String, + pub job: Option, +} +impl Decision { + fn new(target: &str, decision: impl Into) -> Self { + Self { + target: target.into(), + decision: decision.into(), + job: None, + } + } +} +#[derive(Deserialize)] +pub struct StatusQuery { + pub target: String, +} + +pub async fn status( + State(state): State>, + Query(query): Query, +) -> std::result::Result>, MasterError> { + Inventory::new(&state) + .get(&query.target) + .await + .map(Json) + .map_err(MasterError::Internal) +} +pub async fn trigger( + State(state): State>, + Json(request): Json, +) -> std::result::Result, MasterError> { + if request.target.is_empty() + || request.target.len() > 512 + || request.reason.trim().is_empty() + || request.reason.len() > 512 + { + return Err(MasterError::InvalidRequest( + "target and reason are required (maximum 512 bytes each)".into(), + )); + } + // Followers may not have run the coordinated scanner. Refresh only this + // small stats row, with a strict bound; never open the target dataset. + let mut observed = state + .stats_cache + .read() + .await + .iter() + .find(|r| r.name == request.target) + .cloned(); + if let Ok(mut stats) = state.stats.try_lock() { + if let Ok(Ok(row)) = + tokio::time::timeout(Duration::from_secs(2), stats.get(&request.target)).await + { + observed = row; + } + } + admit(&state, observed.as_ref(), &request) + .await + .map(Json) + .map_err(MasterError::Internal) +} + +fn eligibility( + config: &MasterConfig, + row: Option<&StatRow>, + target: &str, + now: i64, +) -> Option<&'static str> { + if !config.catchup.enabled { + return Some("disabled"); + } + if !config.merge_rollout.owned(target) || config.merge_rollout.draining(target) { + return Some("requires_owned_target"); + } + let Some(row) = row else { + return Some("no_stats"); + }; + if row.scanned_at > now + || now.saturating_sub(row.scanned_at) + > config.catchup.stats_max_age_secs.saturating_mul(1000) as i64 + { + return Some("stale_stats"); + } + if row.pending_wal_generations < config.catchup.min_pending { + return Some("below_threshold"); + } + None +} + +async fn admit( + state: &Arc, + row: Option<&StatRow>, + request: &Trigger, +) -> Result { + let now = chrono::Utc::now().timestamp_millis(); + let target = &request.target; + if let Some(reason) = eligibility(&state.config, row, target, now) { + return Ok(Decision::new(target, reason)); + } + let inventory = Inventory::new(state); + if let Some(record) = inventory.get(target).await? { + if record.outcome.as_deref() == Some("succeeded") + && row.is_some_and(|r| { + record + .finished_at_ms + .is_some_and(|finished| r.scanned_at <= finished) + }) + { + return Ok(Decision::new(target, "awaiting_fresh_stats")); + } + if record.active || record.next_retry_ms > now { + let mut decision = Decision::new( + target, + if record.active { + "already_active" + } else { + "cooling_down" + }, + ); + decision.job = Some(record.job); + return Ok(decision); + } + } + let coordinator = state.task_store.merge_coordinator(); + if coordinator.get(target).await?.is_some() { + // The existing reconciler owns recovery. Do not provision idle waiters. + return Ok(Decision::new(target, "awaiting_execution_recovery")); + } + if state + .task_store + .is_cooling_down(lance_context_api::TaskKind::MergeWal, target) + .await + .map_err(|e| e.to_string())? + { + return Ok(Decision::new(target, "cooling_down")); + } + if let Some(failure) = coordinator.failure(target, "master:catchup").await? { + if failure.next_retry_ms > now as u64 { + return Ok(Decision::new(target, "cooling_down")); + } + } + if let Some(reason) = inventory.blocked(target).await? { + return Ok(Decision::new(target, reason)); + } + if request.dry_run { + return Ok(Decision::new(target, "eligible_dry_run")); + } + inventory.ensure_policy(&state.config).await?; + let result = inventory + .reserve( + target, + &request.reason, + row.unwrap().pending_wal_generations, + now, + ) + .await?; + tracing::info!(%target, decision = %result.decision, job = ?result.job, reason = %request.reason, "catch-up admission"); + metrics::counter!("master_catchup_admissions_total", "decision" => result.decision.clone()) + .increment(1); + Ok(result) +} + +/// Independent from scheduler slots, scanners and master jobs. Only metadata. +pub fn spawn(state: &Arc) -> Option> { + if !state.config.catchup.enabled && state.config.catchup.pod_template.is_none() { + return None; + } + let weak = Arc::downgrade(state); + Some(tokio::spawn(async move { + let mut cursor = 0usize; + loop { + let Some(state) = weak.upgrade() else { + return; + }; + let interval = state.config.catchup.interval_secs; + if let Err(error) = tick(&state, &mut cursor).await { + tracing::error!(%error, "catch-up reconciliation failed; scheduler continues"); + metrics::counter!("master_catchup_reconcile_errors_total").increment(1); + } + drop(state); + tokio::time::sleep(Duration::from_secs(interval)).await; + } + })) +} +async fn tick(state: &Arc, cursor: &mut usize) -> Result<()> { + let inventory = Inventory::new(state); + inventory.ensure_policy(&state.config).await?; + let kube = Kubernetes::in_cluster(&state.config.catchup)?; + // The lease reduces duplicate Kubernetes traffic; CAS inventory updates + // remain correct even if a delayed replica outlives this coordination lease. + if let Some(guard) = state + .task_store + .try_coordination_lock("catchup-jobs") + .await + .map_err(|e| e.to_string())? + { + use futures::{stream, StreamExt, TryStreamExt}; + let result = async { + stream::iter(inventory.active().await?).map(|record| { + let inventory = &inventory; + let kube = &kube; + async move { + let mut record = record; + let mut stop = record.termination_requested; + if !stop { + match progress::sample(state, &record).await { + Ok(sample) => { + let now = chrono::Utc::now().timestamp_millis(); + let observed = progress::observe(record.progress.as_ref(), sample.clone(), now, + &state.config.catchup, state.config.maintenance.maintenance_idle_timeout_secs); + let mut updated = record.clone(); + updated.progress = Some(observed); + if !inventory.update(&record, &updated).await? { return Ok(()); } + record = updated; + // Recheck immediately before acting: a stale sample is not + // permission to terminate a now-progressing execution. + if record.progress.as_ref().is_some_and(|p| p.stalled) { + stop = progress::revoke(state, &record, &sample).await?; + if stop { + let mut updated = record.clone(); + updated.termination_requested = true; + if !inventory.update(&record, &updated).await? { return Ok(()); } + record = updated; + } + } + } + Err(error) => { + inventory.note_error(&record, &error).await?; + return Ok(()); + } + } + } + match kube.reconcile(&state.config, &record, stop).await { + Ok((uid, terminal)) => { + if record.job_uid.is_none() && uid.is_some() { + let mut updated = record.clone(); + updated.job_uid = uid; + if !inventory.update(&record, &updated).await? { return Ok(()); } + record = updated; + } + if let Some(success) = terminal { + inventory.complete(&record, success, chrono::Utc::now().timestamp_millis()).await?; + } + }, + Err(error) => { + tracing::warn!(target = %record.target, job = %record.job, %error, "catch-up job unresolved; reservation retained"); + metrics::counter!("master_catchup_job_errors_total").increment(1); + inventory.note_error(&record, &error).await?; + } + } + Ok::<_,String>(()) + } + }).buffer_unordered(8).try_collect::>().await?; + Ok::<_,String>(()) + }.await; + let released = state + .task_store + .release_coordination_lock(guard) + .await + .map_err(|e| e.to_string()); + result?; + released?; + } + // Never wait behind a stats scan or perform fresh table/payload scans here. + let snapshot = state.stats_cache.read().await.clone(); + let mut candidates: Vec<_> = snapshot + .iter() + .filter(|row| { + eligibility( + &state.config, + Some(row), + &row.name, + chrono::Utc::now().timestamp_millis(), + ) + .is_none() + }) + .collect(); + candidates.sort_unstable_by_key(|r| std::cmp::Reverse(r.pending_wal_generations)); + let count = candidates.len(); + if count > 0 { + candidates.rotate_left(*cursor % count); + } + *cursor = if count == 0 { + 0 + } else { + (*cursor + 128) % count + }; + for row in candidates.into_iter().take(128) { + let decision = admit( + state, + Some(row), + &Trigger { + target: row.name.clone(), + reason: "pending WAL exceeds threshold".into(), + dry_run: false, + }, + ) + .await?; + if decision.decision == "capacity_exhausted" { + break; + } + } + metrics::gauge!("master_catchup_last_success_timestamp_seconds") + .set(chrono::Utc::now().timestamp_millis() as f64 / 1000.0); + Ok(()) +} diff --git a/crates/lance-context-master/src/catchup/progress.rs b/crates/lance-context-master/src/catchup/progress.rs new file mode 100644 index 0000000..adf4542 --- /dev/null +++ b/crates/lance-context-master/src/catchup/progress.rs @@ -0,0 +1,172 @@ +//! Observe completed work across master failover. Heartbeats never reset idle. +use super::{store::Record, CatchupConfig, Result}; +use crate::state::MasterState; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Sample { + pub execution: Option, + pub phase: String, + pub sequence: Option, +} +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Observation { + pub sample: Sample, + pub last_sample_ms: i64, + pub last_progress_ms: i64, + pub suspect_since_ms: Option, + pub stalled: bool, +} + +pub(super) async fn sample(state: &MasterState, record: &Record) -> Result { + let coordinator = state.task_store.merge_coordinator(); + let Some(execution) = coordinator.get(&record.target).await? else { + return Ok(Sample { + execution: None, + phase: "starting_or_finished".into(), + sequence: None, + }); + }; + if execution.instance != record.job + || execution.maintenance != Some(lance_context_merge::MaintenanceKind::Catchup) + { + return Err("catch-up progress belongs to another executor; refusing termination".into()); + } + let progress = coordinator.progress(&execution).await?; + Ok(Sample { + execution: Some(execution.id), + phase: format!("{:?}", execution.phase), + sequence: progress.map(|p| p.sequence), + }) +} + +/// Atomically revoke the exact running execution before asking Kubernetes to +/// terminate it. Slow master replicas cannot act on an obsolete progress sample. +pub(super) async fn revoke( + state: &MasterState, + record: &Record, + observed: &Sample, +) -> Result { + let coordinator = state.task_store.merge_coordinator(); + let current = coordinator.get(&record.target).await?; + match current { + Some(execution) + if observed.execution.as_deref() == Some(&execution.id) + && execution.instance == record.job + && format!("{:?}", execution.phase) == observed.phase => + { + if execution.phase == lance_context_merge::Phase::Running { + coordinator + .revoke_stalled(&execution, observed.sequence) + .await + } else if execution.phase == lance_context_merge::Phase::Reserved { + coordinator.cancel_reserved(&execution).await + } else { + // Already cancelled/recovering/terminal work has no ongoing + // admission to revoke. Recheck phase and progress once more. + Ok(sample(state, record).await? == *observed) + } + } + None => Ok(observed.execution.is_none()), + _ => Ok(false), + } +} + +pub(super) fn observe( + previous: Option<&Observation>, + sample: Sample, + now: i64, + config: &CatchupConfig, + idle_secs: u64, +) -> Observation { + // A sampling outage or clock step cannot prove continuous lack of progress. + let max_gap = config + .interval_secs + .saturating_mul(3) + .max(60) + .saturating_mul(1000) as i64; + let previous = + previous.filter(|p| now >= p.last_sample_ms && now - p.last_sample_ms <= max_gap); + let advanced = previous.is_none_or(|p| { + p.sample.execution != sample.execution + || p.sample.phase != sample.phase + || sample + .sequence + .is_some_and(|current| p.sample.sequence.is_none_or(|old| current > old)) + }); + let last_progress_ms = if advanced { + now + } else { + previous.unwrap().last_progress_ms + }; + let limit = if sample.execution.is_none() { + config.startup_timeout_secs + } else { + idle_secs + }; + let idle = now.saturating_sub(last_progress_ms) >= limit.saturating_mul(1000) as i64; + let suspect_since_ms = if idle { + Some(previous.and_then(|p| p.suspect_since_ms).unwrap_or(now)) + } else { + None + }; + // Give a cooperative executor time to cancel and drain; confirm in a later + // observation before asking Kubernetes to terminate an unresponsive process. + let stalled = suspect_since_ms.is_some_and(|since| now.saturating_sub(since) >= 60_000); + Observation { + sample, + last_sample_ms: now, + last_progress_ms, + suspect_since_ms, + stalled, + } +} + +#[cfg(test)] +mod tests { + use super::*; + fn running(sequence: u64) -> Sample { + Sample { + execution: Some("execution-1".into()), + phase: "Running".into(), + sequence: Some(sequence), + } + } + #[test] + fn progress_outlives_slice_but_heartbeats_eventually_stall() { + let config = CatchupConfig::default(); + let mut observed = observe(None, running(0), 0, &config, 120); + for n in 1..=200 { + observed = observe(Some(&observed), running(n), n as i64 * 30_000, &config, 120); + assert!(!observed.stalled); + } + // Far beyond the 1800-second slice; unchanged publications do not help. + for n in 1..=6 { + observed = observe( + Some(&observed), + running(200), + 6_000_000 + n * 30_000, + &config, + 120, + ); + assert_eq!(observed.stalled, n == 6); + } + observed = observe(Some(&observed), running(201), 6_210_000, &config, 120); + assert!(!observed.stalled); + } + #[test] + fn missing_observations_and_new_executors_receive_a_fresh_grace() { + let config = CatchupConfig::default(); + let initial = observe(None, running(0), 0, &config, 120); + let late = observe(Some(&initial), running(0), 1_000_000, &config, 120); + assert!(!late.stalled); + assert_eq!(late.last_progress_ms, 1_000_000); + let mut replacement = running(0); + replacement.execution = Some("execution-2".into()); + let fresh = observe(Some(&late), replacement, 1_030_000, &config, 120); + assert_eq!(fresh.last_progress_ms, 1_030_000); + assert!(!fresh.stalled); + let backwards = observe(Some(&fresh), running(0), 1, &config, 120); + assert_eq!(backwards.last_progress_ms, 1); + } +} diff --git a/crates/lance-context-master/src/catchup/store.rs b/crates/lance-context-master/src/catchup/store.rs new file mode 100644 index 0000000..dc3062b --- /dev/null +++ b/crates/lance-context-master/src/catchup/store.rs @@ -0,0 +1,309 @@ +use super::{Decision, Result}; +use crate::state::MasterState; +use etcd_client::{Client, Compare, CompareOp, GetOptions, Txn, TxnOp}; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Record { + pub target: String, + pub job: String, + #[serde(default)] + pub job_uid: Option, + pub slot: usize, + pub attempt: u64, + pub active: bool, + #[serde(default)] + pub termination_requested: bool, + pub reason: String, + pub pending_at_admission: i64, + pub created_at_ms: i64, + pub finished_at_ms: Option, + pub consecutive_failures: u32, + pub needs_attention: bool, + pub next_retry_ms: i64, + pub outcome: Option, + #[serde(default)] + pub progress: Option, +} +pub(super) struct Inventory { + pub client: Client, + pub prefix: String, + max_jobs: usize, +} +pub(crate) fn active_key(prefix: &str, target: &str) -> String { + lance_context_merge::execution_key(prefix, target) + .replace("/merge-executions/", "/catchup-active/") +} +impl Inventory { + pub fn new(state: &MasterState) -> Self { + Self { + client: state.task_store.etcd_client().clone(), + prefix: state.config.etcd.etcd_prefix.trim_end_matches('/').into(), + max_jobs: state.config.catchup.max_jobs, + } + } + fn record_key(&self, target: &str) -> String { + lance_context_merge::execution_key(&self.prefix, target) + .replace("/merge-executions/", "/catchup-records/") + } + fn slot_key(&self, slot: usize) -> String { + format!("{}/catchup-slots/{slot:03}", self.prefix) + } + pub async fn get(&self, target: &str) -> Result> { + let response = self + .client + .clone() + .get(self.record_key(target), None) + .await + .map_err(|e| e.to_string())?; + response + .kvs() + .first() + .map(|kv| serde_json::from_slice(kv.value()).map_err(|e| e.to_string())) + .transpose() + } + pub async fn ensure_policy(&self, master: &crate::config::MasterConfig) -> Result<()> { + let config = &master.catchup; + let template = super::kubernetes::read_template(config)?; + let policy = serde_json::json!({"namespace":config.namespace,"max_jobs":config.max_jobs,"template":template, + "shards":config.shards,"merge_max_bytes":config.merge_max_bytes,"merge_memory_bytes":config.merge_memory_bytes,"slice":config.slice_secs,"startup_timeout":config.startup_timeout_secs,"idle_timeout":master.maintenance.maintenance_idle_timeout_secs}).to_string(); + let key = format!("{}/catchup-policy", self.prefix); + let mut client = self.client.clone(); + client + .txn( + Txn::new() + .when([Compare::version(key.as_str(), CompareOp::Equal, 0)]) + .and_then([TxnOp::put(key.as_str(), policy.clone(), None)]), + ) + .await + .map_err(|e| e.to_string())?; + let response = client.get(key, None).await.map_err(|e| e.to_string())?; + if response + .kvs() + .first() + .is_none_or(|kv| kv.value() != policy.as_bytes()) + { + return Err( + "catch-up policy differs across masters; keep the existing policy until jobs drain" + .into(), + ); + } + Ok(()) + } + pub async fn active(&self) -> Result> { + let response = self + .client + .clone() + .get( + format!("{}/catchup-slots/", self.prefix), + Some(GetOptions::new().with_prefix().with_limit(257)), + ) + .await + .map_err(|e| e.to_string())?; + if response.kvs().len() > 256 || response.more() { + return Err("catch-up inventory exceeds configured bounds".into()); + } + response + .kvs() + .iter() + .map(|kv| serde_json::from_slice(kv.value()).map_err(|e| e.to_string())) + .collect() + } + pub async fn blocked(&self, target: &str) -> Result> { + let key = lance_context_merge::target_lock_key(&self.prefix, target); + if !self + .client + .clone() + .get(key, None) + .await + .map_err(|e| e.to_string())? + .kvs() + .is_empty() + { + return Ok(Some("awaiting_current_task")); + } + if self.active().await?.len() >= self.max_jobs { + return Ok(Some("capacity_exhausted")); + } + Ok(None) + } + pub async fn note_error(&self, record: &Record, error: &str) -> Result<()> { + let mut updated = record.clone(); + updated.outcome = Some(error.chars().take(1024).collect()); + updated.needs_attention = true; + updated.progress = None; + self.update(record, &updated).await?; + Ok(()) + } + pub async fn update(&self, record: &Record, updated: &Record) -> Result { + let old = serde_json::to_vec(record).map_err(|e| e.to_string())?; + let new = serde_json::to_vec(&updated).map_err(|e| e.to_string())?; + Ok(self + .client + .clone() + .txn( + Txn::new() + .when([ + Compare::value( + self.record_key(&record.target), + CompareOp::Equal, + old.clone(), + ), + Compare::value(self.slot_key(record.slot), CompareOp::Equal, old), + ]) + .and_then([ + TxnOp::put(self.record_key(&record.target), new.clone(), None), + TxnOp::put(self.slot_key(record.slot), new, None), + ]), + ) + .await + .map_err(|e| e.to_string())? + .succeeded()) + } + pub async fn reserve( + &self, + target: &str, + reason: &str, + pending: i64, + now: i64, + ) -> Result { + let old = self.get(target).await?; + if old + .as_ref() + .is_some_and(|r| r.active || r.next_retry_ms > now) + { + return Ok(Decision::new(target, "already_active_or_cooling_down")); + } + let key = self.record_key(target); + let active = active_key(&self.prefix, target); + let execution = lance_context_merge::execution_key(&self.prefix, target); + let merge_claim = execution.replace("/merge-executions/", "/merge-claims/"); + let target_lock = lance_context_merge::target_lock_key(&self.prefix, target); + let job = format!( + "lc-catchup-{}", + lance_context_core::generate_id().to_lowercase() + ); + for slot in 0..self.max_jobs { + let record = Record { + target: target.into(), + job: job.clone(), + job_uid: None, + slot, + attempt: old.as_ref().map_or(1, |r| r.attempt.saturating_add(1)), + active: true, + termination_requested: false, + reason: reason.into(), + pending_at_admission: pending, + created_at_ms: now, + finished_at_ms: None, + consecutive_failures: old.as_ref().map_or(0, |r| r.consecutive_failures), + needs_attention: false, + next_retry_ms: 0, + outcome: None, + progress: None, + }; + let value = serde_json::to_vec(&record).map_err(|e| e.to_string())?; + let prior = match &old { + Some(old) => Compare::value( + key.as_str(), + CompareOp::Equal, + serde_json::to_vec(old).map_err(|e| e.to_string())?, + ), + None => Compare::version(key.as_str(), CompareOp::Equal, 0), + }; + if self + .client + .clone() + .txn( + Txn::new() + .when([ + prior, + Compare::version(self.slot_key(slot), CompareOp::Equal, 0), + Compare::version(active.as_str(), CompareOp::Equal, 0), + Compare::version(execution.as_str(), CompareOp::Equal, 0), + Compare::version(merge_claim.as_str(), CompareOp::Equal, 0), + Compare::version(target_lock.as_str(), CompareOp::Equal, 0), + ]) + .and_then([ + TxnOp::put(key.as_str(), value.clone(), None), + TxnOp::put(self.slot_key(slot), value, None), + TxnOp::put(active.as_str(), job.as_str(), None), + ]), + ) + .await + .map_err(|e| e.to_string())? + .succeeded() + { + let mut decision = Decision::new(target, "reserved"); + decision.job = Some(job); + return Ok(decision); + } + } + // Distinguish busy tables from an actually exhausted fleet budget. + let full = self.active().await?.len() >= self.max_jobs; + Ok(Decision::new( + target, + if full { + "capacity_exhausted" + } else { + "table_busy_or_admission_raced" + }, + )) + } + pub async fn complete(&self, record: &Record, success: bool, now: i64) -> Result<()> { + let mut terminal = record.clone(); + terminal.active = false; + terminal.finished_at_ms = Some(now); + terminal.consecutive_failures = if success { + 0 + } else { + record.consecutive_failures.saturating_add(1) + }; + terminal.needs_attention = terminal.consecutive_failures >= 3; + let delay_secs = if success { + 60 + } else { + (60u64.saturating_mul(1u64 << terminal.consecutive_failures.min(6))).min(3600) + }; + terminal.next_retry_ms = now.saturating_add(delay_secs as i64 * 1000); + terminal.outcome = Some( + if success { + "succeeded" + } else { + "failed; inspect Job status and executor logs" + } + .into(), + ); + let old = serde_json::to_vec(record).map_err(|e| e.to_string())?; + let new = serde_json::to_vec(&terminal).map_err(|e| e.to_string())?; + let active = active_key(&self.prefix, &record.target); + let changed = self + .client + .clone() + .txn( + Txn::new() + .when([ + Compare::value( + self.record_key(&record.target), + CompareOp::Equal, + old.clone(), + ), + Compare::value(self.slot_key(record.slot), CompareOp::Equal, old), + Compare::value(active.as_str(), CompareOp::Equal, record.job.as_bytes()), + ]) + .and_then([ + TxnOp::put(self.record_key(&record.target), new, None), + TxnOp::delete(self.slot_key(record.slot), None), + TxnOp::delete(active, None), + ]), + ) + .await + .map_err(|e| e.to_string())? + .succeeded(); + if changed { + tracing::info!(target = %record.target, job = %record.job, success, failures = terminal.consecutive_failures, "catch-up job terminal; capacity released"); + metrics::counter!("master_catchup_jobs_finished_total", "result" => if success { "ok" } else { "failed" }).increment(1); + } + Ok(()) + } +} diff --git a/crates/lance-context-master/src/catchup/tests.rs b/crates/lance-context-master/src/catchup/tests.rs new file mode 100644 index 0000000..1b30f6c --- /dev/null +++ b/crates/lance-context-master/src/catchup/tests.rs @@ -0,0 +1,488 @@ +use super::*; +use clap::Parser; +use lance_context_api::TaskKind; +use serde_json::json; + +fn config() -> MasterConfig { + let mut config = MasterConfig::parse_from(["master"]); + config.catchup.enabled = true; + config.catchup.shards = vec!["worker-0".into(), "worker-1".into()]; + config.merge_rollout.owned_targets = vec!["hot".into(), "other".into()]; + config +} +fn row(now: i64) -> StatRow { + StatRow { + name: "hot".into(), + uri: "/hot".into(), + row_count: 0, + fragment_count: 0, + last_updated: now, + pending_wal_generations: 1000, + last_compaction: -1, + total_compactions: 0, + scanned_at: now, + version: 1, + } +} +fn template() -> serde_json::Value { + json!({"containers":[{"name":"catchup","image":format!("example/native@sha256:{}", "a".repeat(64)),"command":["/usr/local/bin/lance-context-master"],"resources":{"requests":{"cpu":"2","memory":"8Gi"},"limits":{"cpu":"2","memory":"8Gi"}}}]}) +} +#[test] +fn admission_requires_fresh_owned_pressure_even_for_manual_requests() { + let mut c = config(); + let now = 10_000_000; + let mut r = row(now); + assert!(eligibility(&c, Some(&r), "hot", now).is_none()); + assert_eq!( + eligibility(&c, Some(&r), "legacy", now), + Some("requires_owned_target") + ); + r.scanned_at = now - 901_000; + assert_eq!(eligibility(&c, Some(&r), "hot", now), Some("stale_stats")); + r.scanned_at = now + 1; + assert_eq!(eligibility(&c, Some(&r), "hot", now), Some("stale_stats")); + r.scanned_at = now; + r.pending_wal_generations = 0; + assert_eq!( + eligibility(&c, Some(&r), "hot", now), + Some("below_threshold") + ); + c.catchup.enabled = false; + assert_eq!(eligibility(&c, Some(&r), "hot", now), Some("disabled")); +} +#[test] +fn job_is_native_and_bounded_and_overrides_unsafe_inherited_mode() { + let c = config(); + let r = Record { + target: "hot".into(), + job: "lc-catchup-test".into(), + job_uid: None, + slot: 0, + attempt: 1, + active: true, + termination_requested: false, + reason: "test".into(), + pending_at_admission: 1000, + created_at_ms: 0, + finished_at_ms: None, + consecutive_failures: 0, + needs_attention: false, + next_retry_ms: 0, + outcome: None, + progress: None, + }; + let mut pod = template(); + pod["containers"][0]["env"] = json!([{"name":"CATCHUP_ENABLED","value":"true"}]); + pod["activeDeadlineSeconds"] = json!(1); + let job = kubernetes::render_job(&c, &r, pod); + assert!(job["spec"].get("activeDeadlineSeconds").is_none()); + assert!(job["spec"]["template"]["spec"] + .get("activeDeadlineSeconds") + .is_none()); + assert_eq!(job["spec"]["parallelism"], 1); + assert_eq!(job["spec"]["backoffLimit"], 0); + assert_eq!(job["spec"]["template"]["spec"]["restartPolicy"], "Never"); + assert_eq!( + job["spec"]["template"]["spec"]["automountServiceAccountToken"], + false + ); + let env = job["spec"]["template"]["spec"]["containers"][0]["env"] + .as_array() + .unwrap(); + assert_eq!( + env.iter() + .filter(|e| e["name"] == "CATCHUP_ENABLED") + .count(), + 1 + ); + assert_eq!( + env.iter().find(|e| e["name"] == "CATCHUP_ENABLED").unwrap()["value"], + "false" + ); + let mut next = r.clone(); + next.attempt = 2; + let next_job = kubernetes::render_job(&c, &next, template()); + let env = next_job["spec"]["template"]["spec"]["containers"][0]["env"] + .as_array() + .unwrap(); + assert_eq!( + env.iter().find(|e| e["name"] == "CATCHUP_SHARDS").unwrap()["value"], + "worker-1,worker-0" + ); +} +async fn fixture() -> Option<(tempfile::TempDir, Arc)> { + let endpoints = std::env::var("ETCD_TEST_ENDPOINTS").expect("ETCD_TEST_ENDPOINTS is required"); + let dir = tempfile::tempdir().unwrap(); + let mut c = config(); + c.data_dir = dir.path().to_string_lossy().into(); + c.etcd.etcd_endpoints = endpoints.split(',').map(str::to_string).collect(); + c.etcd.etcd_prefix = format!("/catchup-test/{}", lance_context_core::generate_id()); + c.catchup.max_jobs = 1; + let file = dir.path().join("pod.json"); + std::fs::write(&file, template().to_string()).unwrap(); + c.catchup.pod_template = Some(file.to_string_lossy().into()); + Some((dir, MasterState::new(c).await.unwrap())) +} +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn replicas_share_one_slot_and_dedupe_duplicate_requests() { + let Some((_dir, state)) = fixture().await else { + return; + }; + let a = Inventory::new(&state); + let b = Inventory::new(&state); + let (one, two) = tokio::join!( + a.reserve("hot", "test", 1000, 100), + b.reserve("hot", "retry", 1000, 100) + ); + assert_eq!( + [one.unwrap(), two.unwrap()] + .iter() + .filter(|d| d.decision == "reserved") + .count(), + 1 + ); + assert_eq!( + a.reserve("other", "test", 1000, 100) + .await + .unwrap() + .decision, + "capacity_exhausted" + ); + assert_eq!(a.active().await.unwrap().len(), 1); + let record = a.get("hot").await.unwrap().unwrap(); + b.complete(&record, false, 1000).await.unwrap(); + let failed = a.get("hot").await.unwrap().unwrap(); + assert_eq!(failed.consecutive_failures, 1); + assert!(!failed.active); + assert_ne!( + a.reserve("hot", "retry", 1000, 1001) + .await + .unwrap() + .decision, + "reserved" + ); + assert_eq!( + a.reserve("other", "test", 1000, 1001) + .await + .unwrap() + .decision, + "reserved" + ); + // Replaying an old completion cannot release the new table's slot. + b.complete(&record, true, 2000).await.unwrap(); + assert_eq!(a.active().await.unwrap()[0].target, "other"); +} +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn dedicated_claim_excludes_normal_pollers_but_not_other_tables() { + let Some((_dir, state)) = fixture().await else { + return; + }; + let inventory = Inventory::new(&state); + let reserved = inventory.reserve("hot", "test", 1000, 100).await.unwrap(); + state + .task_store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + state + .task_store + .enqueue(TaskKind::MergeWal, "other", vec![]) + .await + .unwrap(); + let ordinary = state.task_store.claim_next().await.unwrap().unwrap(); + assert_eq!(ordinary.task.target, "other"); + assert!(state + .task_store + .claim_merge_target("hot", "wrong-job") + .await + .unwrap() + .is_none()); + let dedicated = state + .task_store + .claim_merge_target("hot", reserved.job.as_ref().unwrap()) + .await + .unwrap() + .unwrap(); + assert_eq!(dedicated.task.target, "hot"); + assert!(state.task_store.claim_next().await.unwrap().is_none()); + state + .task_store + .finish(dedicated, Ok("test".into())) + .await + .unwrap(); + state + .task_store + .finish(ordinary, Ok("test".into())) + .await + .unwrap(); +} +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn running_native_task_prevents_provisioning_empty_waiters() { + let Some((_dir, state)) = fixture().await else { + return; + }; + state + .task_store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + let claim = state.task_store.claim_next().await.unwrap().unwrap(); + let result = Inventory::new(&state) + .reserve("hot", "test", 1000, 100) + .await + .unwrap(); + assert_ne!(result.decision, "reserved"); + state + .task_store + .finish(claim, Ok("test".into())) + .await + .unwrap(); +} +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn policy_disagreement_cannot_expand_the_cluster_budget() { + let Some((_dir, state)) = fixture().await else { + return; + }; + let inventory = Inventory::new(&state); + inventory.ensure_policy(&state.config).await.unwrap(); + let mut other = state.config.clone(); + other.catchup.max_jobs += 1; + assert!(inventory.ensure_policy(&other).await.is_err()); +} + +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn native_executor_preserves_live_ingestion_and_drains_sealed_shards() { + use lance_context_core::{ + ColumnSpec, ColumnType, GenericStore, GenericStoreOptions, SchemaSpec, + }; + let (_dir, state) = fixture().await.unwrap(); + let target = "generic:hot"; + let uri = state.generic_uri("hot"); + let spec = SchemaSpec::new(vec![ + ( + "id".into(), + ColumnSpec::required(ColumnType::String { large: false }), + ), + ( + "text".into(), + ColumnSpec::new(ColumnType::String { large: true }), + ), + ]); + let mut writers = Vec::new(); + for shard in ["worker-0", "worker-1"] { + let writer = GenericStore::open( + &uri, + spec.clone(), + GenericStoreOptions { + shard_id: Some(shard.into()), + merge_after_generations: Some(0), + ..Default::default() + }, + ) + .await + .unwrap(); + for generation in 0..4 { + let row = json!({"id":format!("{shard}-{generation}"),"text":"x".repeat(512 * 1024)}) + .as_object() + .unwrap() + .clone(); + writer.add(&[row]).await.unwrap(); + writer.flush().await.unwrap(); + } + writers.push(writer); + } + assert_eq!(writers[0].pending_wal_generations().await.unwrap(), 8); + let mut cfg = state.config.clone(); + cfg.catchup.enabled = false; + cfg.merge_rollout.owned_targets.push(target.into()); + let decision = Inventory::new(&state) + .reserve(target, "integration", 8, 100) + .await + .unwrap(); + cfg.catchup.target = Some(target.into()); + cfg.catchup.job_name = decision.job; + let (executed, ()) = tokio::join!(execute(cfg, target), async { + for n in 0..8 { + for (i, writer) in writers.iter().enumerate() { + writer + .add( + &[json!({"id":format!("during-{i}-{n}"),"text":"continuous"}) + .as_object() + .unwrap() + .clone()], + ) + .await + .unwrap(); + writer.flush().await.unwrap(); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }); + executed.unwrap(); + // The old ingestion handles remain usable; the executor must not claim + // a writer epoch or seal somebody else's active memtable. + for (i, writer) in writers.iter().enumerate() { + writer + .add(&[json!({"id":format!("after-{i}"),"text":"live"}) + .as_object() + .unwrap() + .clone()]) + .await + .unwrap(); + writer.flush().await.unwrap(); + } + let reader = GenericStore::open_existing(&uri, GenericStoreOptions::default()) + .await + .unwrap(); + assert!(reader.pending_wal_generations().await.unwrap() <= 18); + let rows = reader.list(None, None).await.unwrap(); + assert_eq!(rows.len(), 26); + let ids: std::collections::HashSet<_> = + rows.iter().map(|r| r["id"].as_str().unwrap()).collect(); + assert_eq!(ids.len(), 26); + assert!(state + .task_store + .merge_coordinator() + .get(target) + .await + .unwrap() + .is_none()); +} + +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn repeated_failed_jobs_keep_backoff_across_controller_restarts() { + let (_dir, state) = fixture().await.unwrap(); + let mut now = 1000; + for attempt in 1..=4 { + let inventory = Inventory::new(&state); + assert_eq!( + inventory + .reserve("hot", "test", 1000, now) + .await + .unwrap() + .decision, + "reserved" + ); + let record = inventory.get("hot").await.unwrap().unwrap(); + assert_eq!(record.attempt, u64::from(attempt)); + inventory.complete(&record, false, now + 1).await.unwrap(); + let persisted = Inventory::new(&state).get("hot").await.unwrap().unwrap(); + assert_eq!(persisted.consecutive_failures, attempt); + assert_eq!(persisted.needs_attention, attempt >= 3); + assert!(persisted.next_retry_ms > now + 60_000); + now = persisted.next_retry_ms; + } +} + +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn catchup_runtime_ceiling_does_not_cancel_work_within_idle_window() { + let (_dir, state) = fixture().await.unwrap(); + let mut cfg = state.config.clone(); + cfg.catchup.enabled = false; + cfg.catchup.target = Some("hot".into()); + let inventory = Inventory::new(&state); + let admitted = inventory + .reserve("hot", "runtime test", 1000, 0) + .await + .unwrap(); + cfg.catchup.job_name = admitted.job.clone(); + cfg.maintenance.maintenance_timeout_secs = 1; + cfg.maintenance.maintenance_idle_timeout_secs = 10; + let dedicated = MasterState::new(cfg).await.unwrap(); + dedicated + .task_store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + let claim = dedicated + .task_store + .claim_merge_target("hot", admitted.job.as_ref().unwrap()) + .await + .unwrap() + .unwrap(); + let result = crate::maintenance_execution::run_as( + &dedicated, + &claim, + lance_context_merge::MaintenanceKind::Catchup, + async { + tokio::time::sleep(Duration::from_millis(2200)).await; + Ok("operation completed beyond old ceiling".into()) + }, + ) + .await; + assert!(result.is_ok(), "{result:?}"); + dedicated.task_store.finish(claim, result).await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS"] +async fn stale_progress_revoke_cannot_cancel_advancing_or_replaced_execution() { + let (_dir, state) = fixture().await.unwrap(); + let inventory = Inventory::new(&state); + let admitted = inventory + .reserve("hot", "stall test", 1000, 0) + .await + .unwrap(); + state + .task_store + .enqueue(TaskKind::MergeWal, "hot", vec![]) + .await + .unwrap(); + let claim = state + .task_store + .claim_merge_target("hot", admitted.job.as_ref().unwrap()) + .await + .unwrap() + .unwrap(); + let coordinator = state.task_store.merge_coordinator(); + let mut execution = lance_context_merge::Execution::new( + "hot", + "master:catchup", + admitted.job.as_ref().unwrap(), + 1, + ); + execution.maintenance = Some(lance_context_merge::MaintenanceKind::Catchup); + let proof = state.task_store.merge_claim(&claim); + assert!(coordinator.reserve(&proof, &execution).await.unwrap()); + let record = inventory.get("hot").await.unwrap().unwrap(); + let queued = progress::sample(&state, &record).await.unwrap(); + let running = coordinator.start(&execution).await.unwrap().unwrap(); + // Starting the same execution changes its phase before its first progress + // publication. The old queued sample must not cancel this fresh executor. + assert!(!progress::revoke(&state, &record, &queued).await.unwrap()); + assert!(coordinator.publish_progress(&running, 1).await.unwrap()); + let record = inventory.get("hot").await.unwrap().unwrap(); + let old = progress::sample(&state, &record).await.unwrap(); + assert!(coordinator.publish_progress(&running, 2).await.unwrap()); + assert!(!progress::revoke(&state, &record, &old).await.unwrap()); + let fresh = progress::sample(&state, &record).await.unwrap(); + assert!(progress::revoke(&state, &record, &fresh).await.unwrap()); + assert!(!coordinator.publish_progress(&running, 3).await.unwrap()); + assert!(coordinator + .authorize_commit(&running, "/hot", "base", 2) + .await + .is_err()); + let uncertain = coordinator.get("hot").await.unwrap().unwrap(); + assert_eq!(uncertain.phase, lance_context_merge::Phase::Uncertain); + assert!(coordinator.release(&proof, &uncertain).await.is_err()); + let frozen = coordinator + .freeze(&proof, &uncertain) + .await + .unwrap() + .unwrap(); + assert!(coordinator.finish_recovery(&proof, &frozen).await.unwrap()); + let recovered = coordinator.get("hot").await.unwrap().unwrap(); + assert!(coordinator.release(&proof, &recovered).await.unwrap()); + state + .task_store + .finish(claim, Err("injected stall".into())) + .await + .unwrap(); +} diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index a345f85..7ef3f8b 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -10,6 +10,8 @@ pub struct MasterConfig { pub merge_rollout: lance_context_merge::rollout::MergeRollout, #[command(flatten)] pub maintenance: MaintenanceConfig, + #[command(flatten)] + pub catchup: crate::catchup::CatchupConfig, /// Data directory / object-store prefix shared with the data-plane server. #[arg(long, env = "DATA_DIR", default_value = "./data")] pub data_dir: String, @@ -294,7 +296,7 @@ mod tests { /// Deadlines for local table mutations on explicitly owned targets. #[derive(Debug, Clone, clap::Args)] pub struct MaintenanceConfig { - /// Maximum local index/compact/repair execution time on owned targets. + /// Legacy wire field; owned work now uses the real-progress idle timeout. #[arg(long, env = "MAINTENANCE_TIMEOUT_SECS", default_value_t = 3600)] pub maintenance_timeout_secs: u64, /// Maximum time without a completed storage/processing step. diff --git a/crates/lance-context-master/src/lib.rs b/crates/lance-context-master/src/lib.rs index f1d23a0..3fda31c 100644 --- a/crates/lance-context-master/src/lib.rs +++ b/crates/lance-context-master/src/lib.rs @@ -1,5 +1,6 @@ //! Control-plane (master) library surface. +pub mod catchup; pub mod config; pub mod discovery; pub mod error; diff --git a/crates/lance-context-master/src/main.rs b/crates/lance-context-master/src/main.rs index 6fa3c1a..dc6d6c0 100644 --- a/crates/lance-context-master/src/main.rs +++ b/crates/lance-context-master/src/main.rs @@ -34,6 +34,14 @@ async fn main() { std::process::exit(1); } + if let Some(target) = config.catchup.target.clone() { + if let Err(error) = lance_context_master::catchup::execute(config, &target).await { + tracing::error!(%target, %error, "dedicated catch-up failed"); + std::process::exit(1); + } + return; + } + let ui_dir = config.ui_dir.clone(); let state = match MasterState::new(config).await { @@ -49,6 +57,7 @@ async fn main() { // Durable scheduler poller (+ optional coordinated auto-sweep). let _scheduler = scheduler::spawn_scheduler(&state); + let _catchup = lance_context_master::catchup::spawn(&state); let mut app = Router::new().nest("/api/v1", routes::api_router()); diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs index 4b159e3..59a31b6 100644 --- a/crates/lance-context-master/src/maintenance_execution.rs +++ b/crates/lance-context-master/src/maintenance_execution.rs @@ -20,6 +20,7 @@ pub(crate) fn retry_kind(endpoint: &str) -> Option { MaintenanceKind::Compact => Some(TaskKind::Compact), MaintenanceKind::IndexId => Some(TaskKind::IndexId), MaintenanceKind::Repair => Some(TaskKind::Repair), + MaintenanceKind::Catchup => Some(TaskKind::MergeWal), } } @@ -154,7 +155,22 @@ where return work.await; } let maintenance = kind(claim.task.kind).ok_or("invalid local maintenance kind")?; - let uri = state.rollout_uri(&claim.task.target); + run_as(state, claim, maintenance, work).await +} + +pub(crate) async fn run_as( + state: &Arc, + claim: &TaskClaim, + maintenance: MaintenanceKind, + work: F, +) -> Result +where + F: Future>, +{ + let uri = match claim.task.target.strip_prefix("generic:") { + Some(name) => state.generic_uri(name), + None => state.rollout_uri(&claim.task.target), + }; if !lance_context_core::merge_write_scope::supports_version_fencing(&uri) { return Err("invalid storage backend for maintenance fencing".into()); } @@ -175,7 +191,16 @@ where let mut reserved = Execution::new( &claim.task.target, maintenance.endpoint(), - &claim.task.id, + if maintenance == MaintenanceKind::Catchup { + state + .config + .catchup + .job_name + .as_deref() + .ok_or("missing catch-up Job identity")? + } else { + &claim.task.id + }, config.maintenance_timeout_secs, ); reserved.maintenance = Some(maintenance); @@ -193,14 +218,13 @@ where uri, })); let outcome = std::panic::AssertUnwindSafe(async { - tokio::time::timeout(Duration::from_secs(running.timeout_secs), async { + let watched = async { tokio::select! { result = scope.run(work) => result, error = watch(&coordinator, &running, &scope) => Err(error), } - }) - .await - .unwrap_or_else(|_| Err("maintenance execution deadline exceeded".into())) + }; + watched.await }) .catch_unwind() .await diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs index d4e3dee..dffe3e6 100644 --- a/crates/lance-context-master/src/merge_execution.rs +++ b/crates/lance-context-master/src/merge_execution.rs @@ -140,7 +140,6 @@ async fn run_legacy(state: &Arc, target: &str) -> Result, last_progress: tokio::time::Instant, @@ -151,7 +150,6 @@ impl Deadlines { fn new(execution: &Execution, now: tokio::time::Instant) -> Self { Self { queue: now + Duration::from_secs(execution.queue_timeout_secs) + RPC_TIMEOUT, - execution_timeout: Duration::from_secs(execution.timeout_secs), idle_timeout: Duration::from_secs(execution.idle_timeout_secs) + RPC_TIMEOUT, running: None, last_progress: now, @@ -168,10 +166,7 @@ impl Deadlines { fn expired(&self, now: tokio::time::Instant) -> bool { match self.running { None => now >= self.queue, - Some(started) => { - now.duration_since(started) >= self.execution_timeout + RPC_TIMEOUT - || now.duration_since(self.last_progress) >= self.idle_timeout - } + Some(_) => now.duration_since(self.last_progress) >= self.idle_timeout, } } } @@ -434,9 +429,6 @@ async fn reconcile_with_grace( loop { // Losing etcd responses must not turn this into an unbounded scheduler // wait. Ownership remains durable when the reconciliation slot exits. - if cancel_started.is_none() && deadlines.expired(tokio::time::Instant::now()) { - cancel_started = Some(tokio::time::Instant::now()); - } if cancel_started.is_some_and(|at| at.elapsed() >= grace) { let error = "merge ownership unresolved: executor did not acknowledge termination; fence retained pending storage version recovery"; coordinator @@ -448,6 +440,11 @@ async fn reconcile_with_grace( Ok(Some(current)) if current.id == initial.id => current, Ok(_) => return Err("merge execution ownership changed during reconciliation".into()), Err(error) => { + // Bound an unavailable coordinator independently. On successful + // reads, sample progress before evaluating the idle deadline. + if cancel_started.is_none() && deadlines.expired(tokio::time::Instant::now()) { + cancel_started = Some(tokio::time::Instant::now()); + } tracing::warn!(target = %initial.target, %error, "cannot confirm merge completion; retaining execution fence"); tokio::time::sleep(RETRY_DELAY).await; continue; @@ -572,9 +569,9 @@ mod tests { let stalled = started + Duration::from_secs(1131); deadlines.observe(10, stalled); assert!(deadlines.expired(stalled)); - // Even actual progress cannot extend the configured total ceiling. + // Real progress continues beyond the legacy total runtime ceiling. deadlines.observe(11, started + Duration::from_secs(1811)); - assert!(deadlines.expired(started + Duration::from_secs(1811))); + assert!(!deadlines.expired(started + Duration::from_secs(1811))); } #[test] diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index e57d006..6bef210 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -710,6 +710,10 @@ pub fn api_router() -> Router> { .route("/experiments/{name}/compact/status", get(compact_status)) .route("/experiments/{name}/rescan", post(rescan_experiment)) .route("/tasks", post(enqueue_task).get(list_tasks)) + .route( + "/catchup", + post(crate::catchup::trigger).get(crate::catchup::status), + ) .route("/scheduler/cooldowns", get(list_cooldowns)) .route("/scheduler/repairs", get(list_repairs)) .route("/registry/diff", get(registry_diff)) @@ -774,6 +778,7 @@ mod tests { fn test_config(dir: &TempDir) -> MasterConfig { MasterConfig { + catchup: Default::default(), maintenance: Default::default(), merge_rollout: Default::default(), data_dir: dir.path().to_string_lossy().to_string(), diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 6066492..0650184 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -942,6 +942,7 @@ mod tests { fn config(dir: &TempDir) -> MasterConfig { MasterConfig { + catchup: Default::default(), maintenance: Default::default(), merge_rollout: lance_context_merge::rollout::MergeRollout { owned_targets: ["exp", "generic:gs", "broken"] diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index f4a471a..f189587 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -124,6 +124,7 @@ impl MasterState { /// Open the registry, stats dataset, and configured durable task store. pub async fn new(config: MasterConfig) -> lance::Result> { config.merge_rollout.validate().map_err(lance::Error::io)?; + config.catchup.validate().map_err(lance::Error::io)?; if config.maintenance.maintenance_timeout_secs == 0 || config.maintenance.maintenance_idle_timeout_secs == 0 || config.maintenance.maintenance_drain_timeout_secs == 0 @@ -135,11 +136,16 @@ impl MasterState { let task_store = TaskStore::open(&config).await?; // Serialize first-time registry/stats creation and legacy backfill in // etcd mode. Followers wait briefly rather than racing Lance creates. - let init_guard = loop { - if let Some(guard) = task_store.try_coordination_lock("state-init").await? { - break guard; - } - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let dedicated = config.catchup.target.is_some(); + let init_guard = if dedicated { + None + } else { + Some(loop { + if let Some(guard) = task_store.try_coordination_lock("state-init").await? { + break guard; + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + }) }; let base_uri = config.data_dir.clone(); let rollout_session = build_rollout_session(config.rollout_cache_bytes); @@ -158,35 +164,41 @@ impl MasterState { &config.registry, ) .await?; - let backfilled = discovery::backfill_registry(&config.data_dir, &*registry).await?; - if backfilled > 0 { - tracing::info!( - experiments = backfilled, - "backfilled rollout registry from data directory" - ); - } - // With a mirror configured, seed it from the primary so a backend - // switch finds every store already there. Repeated on every - // maintenance round; this covers the very first start. - for (label, reg) in [ - ("rollout", ®istry), - ("generic", &generic_registry), - ("datagen", &datagen_registry), - ] { - if let Some(m) = reg.as_any().downcast_ref::() { - match backfill_registry(&*m.primary, &*m.mirror).await { - Ok(copied) if copied > 0 => { - tracing::info!(registry = label, copied, "seeded registry mirror") - } - Ok(_) => {} - Err(error) => { - tracing::warn!(registry = label, %error, "registry mirror seed failed") + // A dedicated one-table executor must not scan/backfill the whole + // registry or delay ordinary master startup behind its initialization. + if !dedicated { + let backfilled = discovery::backfill_registry(&config.data_dir, &*registry).await?; + if backfilled > 0 { + tracing::info!( + experiments = backfilled, + "backfilled rollout registry from data directory" + ); + } + // With a mirror configured, seed it from the primary so a backend + // switch finds every store already there. Repeated on every + // maintenance round; this covers the very first start. + for (label, reg) in [ + ("rollout", ®istry), + ("generic", &generic_registry), + ("datagen", &datagen_registry), + ] { + if let Some(m) = reg.as_any().downcast_ref::() { + match backfill_registry(&*m.primary, &*m.mirror).await { + Ok(copied) if copied > 0 => { + tracing::info!(registry = label, copied, "seeded registry mirror") + } + Ok(_) => {} + Err(error) => { + tracing::warn!(registry = label, %error, "registry mirror seed failed") + } } } } } let stats = StatsStore::open_or_create(&stats_uri, None).await?; - task_store.release_coordination_lock(init_guard).await?; + if let Some(guard) = init_guard { + task_store.release_coordination_lock(guard).await?; + } let state = Arc::new(Self { registry, generic_registry, @@ -281,6 +293,7 @@ mod tests { fn test_config(dir: &TempDir) -> MasterConfig { MasterConfig { + catchup: Default::default(), maintenance: Default::default(), merge_rollout: Default::default(), data_dir: dir.path().to_string_lossy().to_string(), diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index a9a513d..3fee22a 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -291,7 +291,7 @@ impl TaskStore { /// saturated one. pub async fn claim_next_of_kinds(&self, kinds: TaskKinds) -> lance::Result> { self.inner.recover_orphaned().await?; - let (claim, dependency_failed) = self.inner.claim_next(kinds).await?; + let (claim, dependency_failed) = self.inner.claim_next(kinds, None).await?; if dependency_failed { if let Err(error) = self.prune_terminal_history().await { tracing::warn!(error = %error, "failed to prune task history"); @@ -300,6 +300,23 @@ impl TaskStore { Ok(claim) } + /// Claim exactly this target, never consuming another table's queue entry. + pub(crate) async fn claim_merge_target( + &self, + target: &str, + job: &str, + ) -> lance::Result> { + self.inner.recover_orphaned().await?; + let Some(id) = self.get_active_id(TaskKind::MergeWal, target).await? else { + return Ok(None); + }; + Ok(self + .inner + .claim_next(TaskKinds::MERGE_WAL, Some((&id, job))) + .await? + .0) + } + /// Finish a claimed task and release its claim/target lock. pub async fn finish( &self, @@ -587,9 +604,19 @@ impl EtcdTaskStore { .map_err(|_| lance::Error::io("etcd returned an invalid queue count")) } - async fn claim_next(&self, kinds: TaskKinds) -> lance::Result<(Option, bool)> { - let prefix = self.queue_prefix(); - let range_end = prefix_range_end(prefix.as_bytes()); + async fn claim_next( + &self, + kinds: TaskKinds, + only_id: Option<(&str, &str)>, + ) -> lance::Result<(Option, bool)> { + let prefix = only_id.map_or_else(|| self.queue_prefix(), |(id, _)| self.queue_key(id)); + let range_end = if only_id.is_some() { + let mut end = prefix.as_bytes().to_vec(); + end.push(0); + end + } else { + prefix_range_end(prefix.as_bytes()) + }; let mut start_key = prefix.into_bytes(); let mut dependency_failed = false; @@ -667,6 +694,13 @@ impl EtcdTaskStore { Compare::version(claim_key.as_str(), CompareOp::Equal, 0), Compare::version(queue_key.as_str(), CompareOp::Greater, 0), ]; + let catchup_key = crate::catchup::store::active_key(&self.prefix, &task.target); + if requires_target_lock(task.kind) { + compares.push(match only_id { + Some((_, job)) => Compare::value(catchup_key, CompareOp::Equal, job), + None => Compare::version(catchup_key, CompareOp::Equal, 0), + }); + } if let Some(key) = &target_key { if let Some(execution) = &merge_execution { compares.push(Compare::value( @@ -1471,6 +1505,7 @@ mod tests { fn config(dir: &TempDir) -> MasterConfig { MasterConfig { + catchup: Default::default(), maintenance: Default::default(), merge_rollout: Default::default(), data_dir: dir.path().to_string_lossy().to_string(), diff --git a/crates/lance-context-merge/src/lib.rs b/crates/lance-context-merge/src/lib.rs index 0f423ea..f16a6a3 100644 --- a/crates/lance-context-merge/src/lib.rs +++ b/crates/lance-context-merge/src/lib.rs @@ -40,6 +40,7 @@ pub enum MaintenanceKind { Compact, IndexId, Repair, + Catchup, } impl MaintenanceKind { @@ -48,6 +49,7 @@ impl MaintenanceKind { Self::Compact => "master:compact", Self::IndexId => "master:index_id", Self::Repair => "master:repair", + Self::Catchup => "master:catchup", } } pub fn from_endpoint(endpoint: &str) -> Option { @@ -55,6 +57,7 @@ impl MaintenanceKind { "master:compact" => Some(Self::Compact), "master:index_id" => Some(Self::IndexId), "master:repair" => Some(Self::Repair), + "master:catchup" => Some(Self::Catchup), _ => None, } } @@ -338,13 +341,31 @@ impl Coordinator { pub async fn execute_scoped( work: F, timeout: std::time::Duration, + cancel: tokio::sync::watch::Receiver, +) -> Result +where + F: std::future::Future>, +{ + execute_until_cancelled( + async { + tokio::time::timeout(timeout, work) + .await + .unwrap_or_else(|_| Err("merge execution deadline exceeded".into())) + }, + cancel, + ) + .await +} + +/// The caller supplies a real-progress watchdog. Cancellation drops the storage +/// future before returning, without a total runtime ceiling on useful work. +pub async fn execute_until_cancelled( + work: F, mut cancel: tokio::sync::watch::Receiver, ) -> Result where F: std::future::Future>, { - // Keep the work in this inner scope: select! only drops branch borrows when - // the future was pinned outside it, which would publish completion too soon. tokio::select! { biased; _ = async { @@ -353,8 +374,7 @@ where if cancel.changed().await.is_err() { std::future::pending::<()>().await; } } } => Err("merge execution cancelled".into()), - result = tokio::time::timeout(timeout, work) => - result.unwrap_or_else(|_| Err("merge execution deadline exceeded".into())), + result = work => result, } } @@ -534,6 +554,7 @@ mod tests { MaintenanceKind::Compact, MaintenanceKind::IndexId, MaintenanceKind::Repair, + MaintenanceKind::Catchup, ] { let mut local = legacy.clone(); local.maintenance = Some(kind); diff --git a/crates/lance-context-merge/src/progress.rs b/crates/lance-context-merge/src/progress.rs index 54024f5..0cd8cb4 100644 --- a/crates/lance-context-merge/src/progress.rs +++ b/crates/lance-context-merge/src/progress.rs @@ -1,6 +1,6 @@ //! Execution progress is separate from commit ownership. A heartbeat with an //! unchanged sequence must never extend a no-progress deadline. -use crate::{encode, execution_key, Coordinator, Execution, Result}; +use crate::{encode, execution_key, Coordinator, Execution, Phase, Result}; use etcd_client::{Compare, CompareOp, TxnOp}; #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -47,6 +47,47 @@ impl Coordinator { .transpose() } + /// Close commit admission only if both ownership and the observed completed + /// work still match. A concurrent progress publication defeats this CAS. + /// This is not a storage fence: recovery must still cover admitted writes. + pub async fn revoke_stalled( + &self, + execution: &Execution, + sequence: Option, + ) -> Result { + if execution.phase != Phase::Running { + return Ok(false); + } + let mut next = execution.finished(Err( + "confirmed no-progress execution; storage recovery required".into(), + )); + next.phase = Phase::Uncertain; + let progress = match sequence { + Some(sequence) => Compare::value( + self.progress_key(execution), + CompareOp::Equal, + serde_json::to_vec(&ExecutionProgress { sequence }).unwrap(), + ), + None => Compare::version(self.progress_key(execution), CompareOp::Equal, 0), + }; + self.transact( + vec![ + Compare::value( + execution_key(&self.prefix, &execution.target), + CompareOp::Equal, + encode(execution), + ), + progress, + ], + vec![TxnOp::put( + execution_key(&self.prefix, &execution.target), + encode(&next), + None, + )], + ) + .await + } + pub(crate) fn remove_progress(&self, execution: &Execution) -> TxnOp { TxnOp::delete(self.progress_key(execution), None) } diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index 641cf3e..d9d15f3 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -7,8 +7,7 @@ pub struct ServerConfig { #[command(flatten)] pub merge_rollout: lance_context_merge::rollout::MergeRollout, - /// Maximum execution time AFTER slot acquisition. Already - /// started manifest writes must drain before ownership can be released. + /// Legacy wire field; owned merges now use the real-progress idle timeout. #[arg(long, env = "MERGE_EXECUTION_TIMEOUT_SECS", default_value_t = 3600)] pub merge_execution_timeout_secs: u64, diff --git a/crates/lance-context-server/src/merge_execution.rs b/crates/lance-context-server/src/merge_execution.rs index 3f87351..5423729 100644 --- a/crates/lance-context-server/src/merge_execution.rs +++ b/crates/lance-context-server/src/merge_execution.rs @@ -7,7 +7,7 @@ use crate::{ }; use axum::{extract::State, http::StatusCode, Json}; use futures::FutureExt; -use lance_context_merge::{execute_scoped, Coordinator, Execution, Phase}; +use lance_context_merge::{execute_scoped, execute_until_cancelled, Coordinator, Execution, Phase}; use std::{collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::{watch, Mutex, OnceCell}; @@ -281,14 +281,13 @@ async fn run( if !coordinator.publish_progress(&running, 0).await? { return Err("merge ownership revoked before execution".into()); } - execute_scoped( + execute_until_cancelled( async { tokio::select! { result = write_scope.run(work) => result, error = watch_progress(&coordinator, &running, &write_scope) => Err(error), } }, - Duration::from_secs(execution.timeout_secs), cancelled, ) .await @@ -385,7 +384,7 @@ async fn watch_ownership(coordinator: &Coordinator, execution: &Execution) -> St Ok(Ok(Some(current))) if current == *execution => {} Ok(Ok(_)) => return "merge ownership revoked during execution".into(), // An unavailable coordinator is not evidence of a completed fence. - // Queue, idle and total execution deadlines remain independent. + // Queue and real-progress idle deadlines remain independent. _ => {} } } @@ -587,6 +586,30 @@ mod tests { state.merge_executions.shutdown().await; } + #[tokio::test] + #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"] + async fn running_work_is_not_cancelled_by_legacy_total_timeout() { + let (state, coordinator, proof, _dir) = deadline_fixture().await; + let store = state.get_or_open_rollout_store("hot").await.unwrap(); + let held = store.write().await; + let mut execution = Execution::new("hot", "worker", &state.merge_executions.instance, 1); + execution.idle_timeout_secs = 10; + assert!(coordinator.reserve(&proof, &execution).await.unwrap()); + start(State(state.clone()), Json(execution.clone())) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(2200)).await; + assert_eq!( + coordinator.get("hot").await.unwrap().unwrap().phase, + Phase::Running + ); + drop(held); + let finished = terminal(&coordinator).await; + assert!(finished.error.is_none(), "{finished:?}"); + assert!(coordinator.release(&proof, &finished).await.unwrap()); + state.merge_executions.shutdown().await; + } + // No cancel HTTP request is sent in either case. Revocation must interrupt // both slot wait and a merge that cannot reach even its first checkpoint. async fn revoked_executor_exits(queue_blocked: bool) { diff --git a/deploy/catchup/pod-template.example.json b/deploy/catchup/pod-template.example.json new file mode 100644 index 0000000..e15a71b --- /dev/null +++ b/deploy/catchup/pod-template.example.json @@ -0,0 +1,15 @@ +{ + "containers": [ + { + "name": "executor", + "image": "YOUR_REGISTRY/lance-context@sha256:REPLACE_WITH_TESTED_IMAGE_DIGEST", + "command": ["/usr/local/bin/lance-context-master"], + "envFrom": [{"secretRef": {"name": "lance-storage-credentials"}}], + "resources": { + "requests": {"cpu": "2", "memory": "8Gi"}, + "limits": {"cpu": "2", "memory": "8Gi"} + }, + "securityContext": {"allowPrivilegeEscalation": false} + } + ] +} diff --git a/deploy/catchup/rbac.yaml b/deploy/catchup/rbac.yaml new file mode 100644 index 0000000..7f1e5e4 --- /dev/null +++ b/deploy/catchup/rbac.yaml @@ -0,0 +1,34 @@ +# Set metadata.namespace and the binding's subject namespace for your cluster. +# No worker/master Pod eviction, deletion, exec, or deployment mutation rights. +apiVersion: v1 +kind: ServiceAccount +metadata: + name: lance-context-master + namespace: default +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: lance-context-catchup + namespace: default +rules: + - apiGroups: ["batch"] + resources: ["jobs"] + verbs: ["get", "create", "patch"] + - apiGroups: [""] + resources: ["pods"] + verbs: ["list"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: lance-context-catchup + namespace: default +subjects: + - kind: ServiceAccount + name: lance-context-master + namespace: default +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: lance-context-catchup diff --git a/docs/master-catchup.md b/docs/master-catchup.md new file mode 100644 index 0000000..c742788 --- /dev/null +++ b/docs/master-catchup.md @@ -0,0 +1,193 @@ +# Native master catch-up jobs + +The master can provision bounded Kubernetes Jobs for tables whose scanned WAL +count exceeds a threshold. The same admission rules serve an automatic scan and +`POST /api/v1/catchup`, so an agent can request recovery without shell access, +custom Pod specs, arbitrary commands, or a bypass for ownership and resource +limits. The controller lives inside the master process, independently of its +ordinary scheduler pools. No developer workspace or external supervisor is needed +for this catch-up path. + +This feature is disabled by default. It requires owned merge targets. Deploy +compatible masters, including auxiliary maintenance masters, before activation: +the native executor adds a `catchup` maintenance discriminator to the existing +execution fence. Older readers reject that discriminator. This does not convert +legacy writers into fenced writers; existing per-table rollout/drain requirements +still apply. It does not enable owned targets or restart production workers. + +## Configure and activate + +1. Build the native master binary into a tested image and pin its digest in a + copy of `deploy/catchup/pod-template.example.json`. Mount this PodSpec JSON into + every master. Provide storage and etcd credentials through the reviewed + template's Secret references/mounts. Controller-generated environment values + override the target, data URI, etcd endpoints/prefix and execution budgets. + Image, command, credentials and resources never come from API callers. +2. Bind the namespace-scoped Role in `deploy/catchup/rbac.yaml` to the masters' + ServiceAccount. It permits creating/getting/patching Jobs and listing Pods; it grants + no Pod deletion, exec, or deployment modification. The executor Pod does not + receive the Kubernetes service account token. +3. Set `CATCHUP_POD_TEMPLATE` to the mounted JSON path, + `CATCHUP_NAMESPACE` to the deployment namespace, `CATCHUP_SHARDS` to the exact + stable worker identities used to derive WAL shards (for example + `worker-0,worker-1`), and `CATCHUP_ENABLED=true` on compatible masters. + Shard identities must include historical writers whose WAL should be drained. +4. Keep the same template, namespace, shard list, Job limit and execution budgets + on every master. The first controller persists this policy in etcd; mismatched + replicas fail closed. To change it, disable admission on every master, keep + reconciliation configured until all reservations finish, then remove only the + `/catchup-policy` key before starting the new shared configuration. + +Defaults: four Jobs cluster-wide, 256 pending generations, a 30-second scan +interval, and stats no older than 900 seconds. A Job has one container, one table, +serial shard visits, no Kubernetes retries, a 1800-second soft admission slice, a +30-second termination grace period, and a 24-hour terminal retention period. +The Job cannot preempt other Pods. CPU/memory requests must equal explicit +limits. The sample reserves 2 CPUs and 8 GiB per Job (8 CPUs/32 GiB total at the +default fleet cap); size these against available capacity before enabling. + +The native process uses a 128 MiB shared Lance cache, one merge at a time, +eight generations and 64 MiB per batch, and a 1 GiB merge-buffer budget. These +are buffering bounds, not a total RSS guarantee: one indivisible generation and +Lance working memory can exceed the buffer bound. Kubernetes limits provide the +final process bound. Validate real large generations before increasing concurrency. +A slice visits configured shards up to 16 times, stopping early when a +whole pass reclaims nothing or `CATCHUP_SLICE_SECS` ends. An already admitted +operation may finish beyond that slice while making progress. No Job or Pod +runtime deadline is installed at creation. Each +Job rotates the starting shard using a durable attempt number, including after +failures, so a slow first shard cannot always hide the rest. Further fresh +pressure can request another slice. + +## Agent/API contract + +These are internal admin endpoints with the same network/authentication boundary +as `/api/v1/tasks`. Do not expose them publicly; use the deployment's authenticated +admin ingress or an authenticated proxy for agents. This change adds no public +unauthenticated agent gateway. + +```http +POST /api/v1/catchup +Content-Type: application/json + +{"target":"hot_table","reason":"WAL pressure observed during reads","dry_run":true} +``` + +A dry run evaluates eligibility, active reservations, table locks and capacity. +Remove `dry_run` or set it to `false` to reserve a Job. Repeating the request +coalesces with the active Job and does not reset its failure budget. Generic +stores use scheduler target names such as `generic:events`. + +The response contains `target`, `decision`, and optional `job`. `reserved` means +admitted, not started or completed. Other decisions describe missing/stale stats, +a threshold not crossed, a non-owned target, a current execution needing recovery, +a table still busy, cooldown, capacity exhaustion, or an existing Job. Unknown +request fields (including attempts to supply a Pod spec or force flag) are rejected. + +```http +GET /api/v1/catchup?target=hot_table +``` + +Status exposes the current/latest Job, admission pending count and reason, active +reservation, start/finish timestamps, outcome, consecutive failures, next retry +and `needs_attention`, the last observed execution/sequence and observation times, +plus whether termination was requested. A missing target record returns JSON `null`. For failures, +inspect this status and the referenced Job logs; do not clear ownership keys or +blindly recreate Pods. + +An agent or skill should first read status, issue a dry run, request a Job only +when eligible, and poll status at a bounded interval. It must treat +`awaiting_execution_recovery` as recovery in progress, not permission to run an +unfenced writer. All resource and retry rules apply identically to agents and +automatic requests. Write/read pressure reports can use this endpoint; this PR +does not add synchronous metadata scans or HTTP calls to those data paths. + +## Failure and ownership behavior + +Admission atomically reserves a per-table key and one persistent fleet slot in +etcd. A native scheduler task already holding a target lock prevents provisioning; +there is no fleet of empty waiters. Ordinary merge claim transactions exclude +reserved targets, while the native executor must present the exact Job identity +to claim that target's queue entry. Other tables and master jobs continue normally. + +A lost master or Kubernetes create response does not release a slot: another +master reconciles the same deterministic Job name. Failed Jobs retain capacity +until every associated Pod is terminal. A failed/unknown Kubernetes lookup +retains the reservation and reports an error. A Kubernetes outage or a Pod stuck +terminating may therefore require cluster repair; claiming completion would allow +unbounded overlapping Pods. Controllers do not remove those reservations blindly. + +The native executor merges only sealed generations without claiming the live +writer's epoch. It uses the same durable target ownership, admission revocation, +progress timeout and storage version barrier as other owned maintenance. Losing +a task lease or terminating its Pod is not enough to authorize new storage writes; +unresolved execution fencing must still succeed. The native executor times out a +busy initial task claim after 30 seconds. No-progress work uses the existing +`MAINTENANCE_IDLE_TIMEOUT_SECS` (600 by default). The master independently +observes completed work, not heartbeats. Continuous no progress for that interval +plus a separate 60-second confirmation triggers recovery; the startup window +before an execution appears is `CATCHUP_STARTUP_TIMEOUT_SECS` (1800 by default). +A new execution receives a fresh observation window. Failed samples or a gap +longer than three controller intervals (at least 60 seconds) invalidate the old +window; an observation outage does not prove a stall. + +For a running execution, a final etcd transaction compares both ownership and the +observed progress sequence before revoking commit admission. A concurrent advance +wins over cancellation. Only then does the master patch the exact Job's deadline +to terminate a non-cooperative process, with UID/resource-version preconditions. +It persists the termination request for retry across master/API failures. This +is a reaction to confirmed lack of progress, not a runtime limit on healthy work. +The existing storage barrier must still fence previously admitted commits before +replacement writes; Job termination alone is insufficient. Confirmed Job UIDs +cannot be silently replaced or recreated after disappearance. + +Completed nonempty WAL batches and successful manifest commits count as progress; +conditional commit conflicts and unchanged publications do not. A pending count +may rise under ingestion despite useful merge work, so it is not the stall signal. +Some Lance-internal phases, including a single index build or long storage call, +do not yet expose intermediate checkpoints. Validate the idle threshold against +these phases on real large tables; completed-step telemetry cannot establish +whether an opaque call is still advancing internally. + +Failure counts survive master and Job replacement. Failed Jobs back off from +120 seconds to one hour, with `needs_attention` after three failures; the existing +merge failure ledger remains in force too. A successful Job waits at least a +minute and requires a newer stats observation before readmission, preventing an +old high count from creating repeated empty Jobs. There is no unlimited immediate +retry loop. Controller errors are visible in status, logs and +`master_catchup_*` counters/gauges. The controller examines a rotating window of +up to 128 eligible candidates so a busy group of large tables cannot indefinitely +hide the remainder. + +Disable new admission with `CATCHUP_ENABLED=false` while retaining the template +and policy settings until existing reservations finish. This leaves reconciliation +running. Removing the entire controller configuration before Jobs finish strands +reservations and is not the shutdown procedure. + +This is a catch-up controller, not a replacement for every external watchdog rule. +It does not evict ordinary workers or diagnose arbitrary OS-level deadlocks. Its +progress-based termination handles dedicated processes; ordinary owned-task recovery still +uses the existing worker identity/progress detection and cancellation protocol. + +## Validation before production activation + +The tests cover multi-master CAS admission, global capacity, wrong-Job claims, +ordinary-table progress, durable backoff, policy disagreement, an ambiguous Job +creation response, delayed Pod termination, and a native multi-shard merge with +live ingestion handles, progress beyond the old runtime ceiling, repeated unchanged +heartbeats, sampling gaps, stale-progress revocation races, and Job identity checks. Run the etcd-backed tests with `ETCD_TEST_ENDPOINTS` and +`--include-ignored`; the repository CI runs ignored etcd tests separately. + +In staging, use isolated storage and an etcd prefix. Verify real large-generation +RSS and throughput, continuous writes with result conservation, survival beyond +the slice while advancing, confirmed no-progress termination, master replacement after reservation/creation, storage failure during +a commit, and complete fencing before retry. Merely seeing a Pod become Ready +is not a throughput or recovery test. These changes do not themselves deploy or +activate the production controller. + +Owned worker merges and local owned maintenance also stop using a total execution +time ceiling in this change. Their existing idle watchdogs, ownership checks, +queue limits, cancellation and manifest-drain barriers remain. The legacy timeout +configuration/wire fields remain for compatibility; older master/worker binaries +may still enforce them during a mixed-version rollout. Deploy compatible versions +throughout before relying on the absence of a total execution ceiling. diff --git a/docs/merge-recovery.md b/docs/merge-recovery.md index d3c3867..7b2e63f 100644 --- a/docs/merge-recovery.md +++ b/docs/merge-recovery.md @@ -62,21 +62,23 @@ cancels/drains owned actors before closing resident stores. Admission and slot acquisition have a separate `MERGE_QUEUE_TIMEOUT_SECS` (default 600). The execution clock starts only after a merge slot is acquired. -`MERGE_EXECUTION_TIMEOUT_SECS` is a configurable total ceiling (default 3600), -not a hard-coded 600-second cap. `MERGE_IDLE_TIMEOUT_SECS` (default 600) limits -the interval without observed completed work. All three must be positive. - -The worker counts completed batch reads, merge phases, and shielded storage -operations. It publishes a changed sequence at most once a second, guarded by +`MERGE_IDLE_TIMEOUT_SECS` (default 600) limits the interval without observed +completed work. Progressing owned merges have no total runtime ceiling. +`MERGE_EXECUTION_TIMEOUT_SECS` remains a positive legacy wire field for rolling +compatibility; older masters/workers can still enforce it until upgraded. Queue +and idle limits must also be positive. + +The worker counts completed nonempty batch reads and successful manifest commits, +excluding conditional-write conflicts and failed storage operations. It publishes a changed sequence at most once a second, guarded by execution ownership. The master tracks the same sequence with bounded RPC -allowance. Repeated heartbeats do not extend the idle limit, and progressing -work cannot extend the total ceiling. A missing progress record means the +allowance. Repeated heartbeats do not extend the idle limit. Progressing work +may continue beyond the legacy total ceiling. A missing progress record means the executor has not reported acquiring its slot. Readiness and these intermediate checkpoints are NOT proof of WAL reclamation; only committed shard drains are. Progress observation is coarse inside opaque Lance index/build/storage calls. A single such operation exceeding the idle threshold can still be cancelled. -Measure its p99 on realistic blobs and tune the idle/total limits before enabling +Measure its p99 on realistic blobs and tune the idle limit before enabling a large target. These defaults are not production throughput measurements. Cancellation retains the slot until leaf commits drain or a barrier proves them fenced. It does not restart the pod. An etcd outage cannot indefinitely occupy a diff --git a/docs/owned-maintenance-recovery.md b/docs/owned-maintenance-recovery.md index 37d9960..0dd4288 100644 --- a/docs/owned-maintenance-recovery.md +++ b/docs/owned-maintenance-recovery.md @@ -8,9 +8,11 @@ compaction concurrency and all merge memory/byte budgets are unchanged. ## Cancellation and takeover -The default total deadline is 3600 seconds (`MAINTENANCE_TIMEOUT_SECS`), with a +Owned maintenance has no total runtime ceiling while making progress. It uses a 600-second no-progress deadline (`MAINTENANCE_IDLE_TIMEOUT_SECS`) and 30-second -commit drain grace (`MAINTENANCE_DRAIN_TIMEOUT_SECS`). All must be positive. +commit drain grace (`MAINTENANCE_DRAIN_TIMEOUT_SECS`). `MAINTENANCE_TIMEOUT_SECS` +remains a positive legacy wire field; older binaries may still enforce it during +a rolling upgrade. All timeout configuration fields must remain positive. Progress counts completed scoped processing steps or manifest commits, never heartbeats. Lance index building and individual compaction rewrites do not expose fine-grained scan progress here: a healthy long operation can reach the