From 979beaa7d93bc8231b82dde7627695e66d269bab Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Tue, 6 Oct 2026 02:58:56 +0000 Subject: [PATCH] fix: defer native catch-up when admission is busy before claim --- .github/workflows/native-supervisor-test.yml | 2 +- .../src/catchup/executor.rs | 46 ++++- .../lance-context-master/src/catchup/mod.rs | 2 +- .../lance-context-master/src/catchup/tests.rs | 96 ++++++++++ crates/lance-context-master/src/main.rs | 25 ++- deploy/catchup/native-publisher.md | 23 +++ deploy/catchup/native_supervisor.py | 80 +++++++- deploy/catchup/test_native_supervisor.py | 172 ++++++++++++++++++ 8 files changed, 429 insertions(+), 17 deletions(-) diff --git a/.github/workflows/native-supervisor-test.yml b/.github/workflows/native-supervisor-test.yml index ba0a08d5..6fa49d1e 100644 --- a/.github/workflows/native-supervisor-test.yml +++ b/.github/workflows/native-supervisor-test.yml @@ -18,7 +18,7 @@ jobs: - uses: actions/checkout@v4 - name: Install disposable etcd fixture dependency run: sudo apt-get update && sudo apt-get install -y etcd-server - - name: Verify supervisor identity, CAS, backoff and join behavior + - name: Verify supervisor identity, busy deferral, CAS, backoff and join behavior env: ETCD_BIN: /usr/bin/etcd run: python3 -m unittest discover -s deploy/catchup -p 'test_native_supervisor.py' -v diff --git a/crates/lance-context-master/src/catchup/executor.rs b/crates/lance-context-master/src/catchup/executor.rs index d6810703..7cde32c6 100644 --- a/crates/lance-context-master/src/catchup/executor.rs +++ b/crates/lance-context-master/src/catchup/executor.rs @@ -8,9 +8,17 @@ use lance_context_core::{ use lance_context_merge::MaintenanceKind; use std::{future::Future, sync::Arc, time::Duration}; +/// A busy admission never acquired a task or entered recovery/payload work. +/// Transport errors, including an unknown claim outcome, remain errors. +#[derive(Debug, PartialEq, Eq)] +pub enum ExecuteOutcome { + Completed, + AdmissionBusy, +} + /// One dedicated process, one table, ordered commits, 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<()> { +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) { @@ -31,7 +39,7 @@ pub async fn execute(mut config: MasterConfig, target: &str) -> Result<()> { .await .map_err(|e| e.to_string())?; let admission_started = tokio::time::Instant::now(); - let claim = claim_before_deadline( + let Some(claim) = claim_before_deadline( || async { state .task_store @@ -41,7 +49,10 @@ pub async fn execute(mut config: MasterConfig, target: &str) -> Result<()> { }, admission_started + Duration::from_secs(30), ) - .await?; + .await? + else { + return Ok(ExecuteOutcome::AdmissionBusy); + }; tracing::info!(%target, admission_seconds = admission_started.elapsed().as_secs_f64(), "catch-up task claimed"); let result = async { @@ -76,13 +87,13 @@ pub async fn execute(mut config: MasterConfig, target: &str) -> Result<()> { if let Some(error) = error { return Err(error); } - Ok(()) + Ok(ExecuteOutcome::Completed) } async fn claim_before_deadline( mut attempt: F, deadline: tokio::time::Instant, -) -> Result +) -> Result> where F: FnMut() -> Fut, Fut: Future>>, @@ -92,10 +103,10 @@ where // mutating claim RPC. Individual etcd RPCs have their own timeout; // an accepted claim must be delivered even at the retry deadline. if tokio::time::Instant::now() >= deadline { - return Err("catch-up admission remained busy for 30 seconds".into()); + return Ok(None); } if let Some(claim) = attempt().await? { - return Ok(claim); + return Ok(Some(claim)); } tokio::time::sleep_until( deadline.min(tokio::time::Instant::now() + Duration::from_secs(1)), @@ -317,7 +328,7 @@ mod tests { ) .await .unwrap(); - assert_eq!(result, "accepted claim"); + assert_eq!(result, Some("accepted claim")); } #[tokio::test] @@ -332,10 +343,27 @@ mod tests { deadline, ) .await; - assert!(result.unwrap_err().contains("admission remained busy")); + assert_eq!(result.unwrap(), None); assert_eq!(calls.load(Ordering::Relaxed), 1); } + #[tokio::test] + async fn admission_unknown_claim_outcome_is_not_busy_even_after_deadline() { + let deadline = tokio::time::Instant::now() + Duration::from_millis(50); + let result = claim_before_deadline( + || async { + tokio::time::sleep_until(deadline + Duration::from_millis(10)).await; + Err::, _>("claim response lost after accepted transaction".into()) + }, + deadline, + ) + .await; + assert_eq!( + result.unwrap_err(), + "claim response lost after accepted transaction" + ); + } + #[tokio::test] async fn pipeline_overlaps_reads_with_one_ordered_commit_and_bounds_lookahead() { let prepared = AtomicUsize::new(0); diff --git a/crates/lance-context-master/src/catchup/mod.rs b/crates/lance-context-master/src/catchup/mod.rs index 71d0672a..c92ddd73 100644 --- a/crates/lance-context-master/src/catchup/mod.rs +++ b/crates/lance-context-master/src/catchup/mod.rs @@ -11,7 +11,7 @@ use axum::{ extract::{Query, State}, Json, }; -pub use executor::execute; +pub use executor::{execute, ExecuteOutcome}; use kubernetes::Kubernetes; use serde::{Deserialize, Serialize}; use std::{sync::Arc, time::Duration}; diff --git a/crates/lance-context-master/src/catchup/tests.rs b/crates/lance-context-master/src/catchup/tests.rs index 14106521..8a917687 100644 --- a/crates/lance-context-master/src/catchup/tests.rs +++ b/crates/lance-context-master/src/catchup/tests.rs @@ -541,6 +541,102 @@ async fn stale_progress_revoke_cannot_cancel_advancing_or_replaced_execution() { /// Full native executor comparison; setup and verification are outside timing. /// Opt in explicitly so CI's ignored etcd suite does not run a benchmark. +#[tokio::test] +#[ignore = "requires ETCD_TEST_ENDPOINTS; exercises the real 30-second claim boundary"] +async fn busy_executor_preserves_live_claim_then_merges_after_release() { + 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 schema = SchemaSpec::new(vec![ + ( + "id".into(), + ColumnSpec::required(ColumnType::String { large: false }), + ), + ( + "text".into(), + ColumnSpec::new(ColumnType::String { large: true }), + ), + ]); + let writer = GenericStore::open( + &uri, + schema, + GenericStoreOptions { + shard_id: Some("worker-0".into()), + merge_after_generations: Some(0), + ..Default::default() + }, + ) + .await + .unwrap(); + for generation in 0..4 { + writer + .add(&[ + json!({"id": format!("row-{generation}"), "text": format!("payload-{generation}")}) + .as_object() + .unwrap() + .clone(), + ]) + .await + .unwrap(); + writer.flush().await.unwrap(); + } + assert_eq!(writer.pending_wal_generations().await.unwrap(), 4); + let mut cfg = state.config.clone(); + cfg.catchup.enabled = false; + cfg.catchup.shards = vec!["worker-0".into()]; + cfg.merge_rollout.owned_targets.push(target.into()); + cfg.catchup.target = Some(target.into()); + cfg.catchup.job_name = Inventory::new(&state) + .reserve(target, "busy-test", 4, 100) + .await + .unwrap() + .job; + let task = state + .task_store + .enqueue(TaskKind::MergeWal, target, vec![]) + .await + .unwrap(); + let held = state + .task_store + .claim_merge_target(target, cfg.catchup.job_name.as_ref().unwrap()) + .await + .unwrap() + .unwrap(); + let started = std::time::Instant::now(); + assert_eq!( + execute(cfg.clone(), target).await.unwrap(), + ExecuteOutcome::AdmissionBusy + ); + assert!(started.elapsed() >= std::time::Duration::from_secs(30)); + assert_eq!( + state.task_store.get(&task.id).await.unwrap().unwrap().state, + lance_context_api::TaskState::Running + ); + assert_eq!(writer.pending_wal_generations().await.unwrap(), 4); + state + .task_store + .finish(held, Ok("release fixture claim".into())) + .await + .unwrap(); + assert_eq!( + execute(cfg, target).await.unwrap(), + ExecuteOutcome::Completed + ); + let reader = GenericStore::open_existing(&uri, GenericStoreOptions::default()) + .await + .unwrap(); + assert_eq!(reader.pending_wal_generations().await.unwrap(), 0); + let rows = reader.list(None, None).await.unwrap(); + assert_eq!(rows.len(), 4); + for row in rows { + let index = row["id"].as_str().unwrap().strip_prefix("row-").unwrap(); + assert_eq!(row["text"], format!("payload-{index}")); + } +} + #[tokio::test] #[ignore = "manual benchmark: CATCHUP_BENCH=1 ETCD_TEST_ENDPOINTS required"] async fn benchmark_native_catchup_pipeline() { diff --git a/crates/lance-context-master/src/main.rs b/crates/lance-context-master/src/main.rs index 26f1b397..b0cb777e 100644 --- a/crates/lance-context-master/src/main.rs +++ b/crates/lance-context-master/src/main.rs @@ -35,9 +35,28 @@ async fn main() { } 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); + use lance_context_master::catchup::ExecuteOutcome; + match lance_context_master::catchup::execute(config, &target).await { + Ok(ExecuteOutcome::Completed) => {} + Ok(ExecuteOutcome::AdmissionBusy) => { + // Machine-readable, attempt-bound contract with the supervisor. + // Emitted only before a task claim; never infer this from error text. + println!( + "{}", + serde_json::json!({ + "event": "catchup_admission_deferred", + "version": 1, + "reason": "busy_before_claim", + "target": target, + "attempt_id": std::env::var("CATCHUP_ATTEMPT_ID").ok(), + }) + ); + std::process::exit(75); + } + Err(error) => { + tracing::error!(%target, %error, "dedicated catch-up failed"); + std::process::exit(1); + } } return; } diff --git a/deploy/catchup/native-publisher.md b/deploy/catchup/native-publisher.md index 5c13a0c0..bf405ca6 100644 --- a/deploy/catchup/native-publisher.md +++ b/deploy/catchup/native-publisher.md @@ -61,6 +61,29 @@ responsible for claims, cancellation, memory reservations, storage barriers and watermarks. This script has no general automatic recreation loop. Migration to the native Job controller remains separate work. +## Busy before task claim + +When every completed claim attempt returns no claim for 30 seconds, the native +executor returns the typed `AdmissionBusy` outcome. It has not entered execution +recovery or payload work. The CLI exits 75 and emits one JSON record with event +`catchup_admission_deferred`, version 1, reason `busy_before_claim`, target, and +`CATCHUP_ATTEMPT_ID`. A claim accepted across the deadline is still delivered; +an RPC error or lost claim response remains a failure, even after the deadline. + +The supervisor supplies a fresh attempt ID to each child, joins it, and requires +both exit 75 and that exact record with no observed claim or payload work. Only +then does it record `state=deferred` and retry after 2–5 seconds of jitter rather +than charging another 120-second-or-longer failure penalty. It preserves previous +failure counts, attention flags, and earned deadlines, rereads the native failure +ledger after joining, and never changes that ledger. The next admission always +checks the latest durable failure deadline again. A failed terminal CAS retains +the pessimistic admitted attempt; it does not authorize a short retry. + +This requires both the updated native binary and supervisor. An old binary's +busy error text, exit 75 alone, a mismatched/duplicate marker, partial work, +transport errors and unknown outcomes retain normal failure handling. The +separate native Kubernetes Job controller does not consume this wrapper protocol. + ## Tests ```sh diff --git a/deploy/catchup/native_supervisor.py b/deploy/catchup/native_supervisor.py index 6dda1477..44ab3037 100644 --- a/deploy/catchup/native_supervisor.py +++ b/deploy/catchup/native_supervisor.py @@ -10,6 +10,7 @@ import json import os import pathlib +import random import re import signal import subprocess @@ -34,6 +35,30 @@ def authorized(operator, job, pod_name, pod_uid): ) +def confirmed_busy(code, markers, attempt_id, target, claimed, reclaimed, errors): + """Only the new binary's joined, attempt-bound pre-claim outcome can defer. + + An old binary's error text, an arbitrary exit75, or any claim/payload evidence + retains the ordinary failure path. The caller must have joined the child. + """ + return ( + code == 75 + and not claimed + and reclaimed == 0 + and errors == 0 + and len(markers) == 1 + and type(markers[0].get("version")) is int + and markers[0] + == { + "event": "catchup_admission_deferred", + "version": 1, + "reason": "busy_before_claim", + "target": target, + "attempt_id": attempt_id, + } + ) + + def main(): UID = os.environ.get("POD_UID", "").strip() if not UID: @@ -177,18 +202,36 @@ def halt(*args): started = time.monotonic() reclaimed = 0 pass_errors = 0 + claimed = False + markers = [] proc = subprocess.Popen( [NATIVE_BINARY], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, + env={**os.environ, "CATCHUP_ATTEMPT_ID": attempt["id"]}, ) log("native_pass_started", pid=proc.pid) for line in proc.stdout: print(line.rstrip(), flush=True) clean = re.sub(r"\x1b\[[0-9;]*m", "", line) + if ( + "catch-up task claimed" in clean + or "dedicated catch-up committed" in clean + ): + claimed = True + try: + event = json.loads(clean) + except ValueError: + event = None + if ( + isinstance(event, dict) + and event.get("event") == "catchup_admission_deferred" + ): + markers.append(event) if "parallel rollout append completed" in clean: + claimed = True n = re.search(r"\breclaimed=(\d+)", clean) if n: reclaimed += int(n.group(1)) @@ -206,7 +249,30 @@ def halt(*args): reclaimed=reclaimed, seconds=seconds, ) - if code == 0: + busy = confirmed_busy( + code, markers, attempt["id"], T, claimed, reclaimed, pass_errors + ) + if busy: + # Read again after joining: do not shorten existing earned debt, + # including debt recorded concurrently by another executor. + failure = json.loads( + read("merge-failures/" + H + "/" + b"master:catchup".hex()) or "{}" + ) + terminal.update( + state="deferred", + deferred_admission={"class": "busy_before_claim"}, + consecutive_failures=previous.get("consecutive_failures", 0), + needs_attention=bool( + previous.get("needs_attention") + or failure.get("needs_attention") + ), + next_retry_ms=max( + previous.get("next_retry_ms", 0), + failure.get("next_retry_ms", 0), + int((time.time() + random.uniform(2, 5)) * 1000), + ), + ) + elif code == 0: terminal.update( consecutive_failures=0, needs_attention=False, @@ -220,7 +286,8 @@ def halt(*args): ) * 1000 ) - if cas_attempt(admitted, terminal, operator) is None: + recorded = cas_attempt(admitted, terminal, operator) is not None + if not recorded: log("attempt_record_changed_ownership_untouched") log( "native_pass_finished", @@ -229,9 +296,16 @@ def halt(*args): seconds=seconds, generations_per_second=reclaimed / seconds, failed_shards=pass_errors, + outcome=terminal["state"], + attempt_recorded=recorded, ) # Let ordinary flushes proceed; zero-work passes must not hammer etcd/catalogs. - stop.wait(2 if code == 0 and reclaimed > 0 else 30) + if busy and recorded: + stop.wait( + max(0, min(60, terminal["next_retry_ms"] / 1000 - time.time())) + ) + else: + stop.wait(2 if code == 0 and reclaimed > 0 else 30) except Exception as e: # noqa: BLE001 - preserve native child join on any supervisor error log("supervisor_error_ownership_untouched", error=repr(e)) if proc is not None: diff --git a/deploy/catchup/test_native_supervisor.py b/deploy/catchup/test_native_supervisor.py index 2b0afb4f..30c2bcb9 100644 --- a/deploy/catchup/test_native_supervisor.py +++ b/deploy/catchup/test_native_supervisor.py @@ -276,6 +276,136 @@ def test_matching_uid_preserves_failure_and_attempt_backoff(self): self.assertEqual(self.read("failure"), failure) self.assertFalse((self.folder / "started").exists()) + def configure_outcome(self, *, code=75, marker=True, extra="", foreign=False): + # The child cannot report its terminal outcome until explicitly released. + text = self.stub.read_text() + text = text[: text.index('print("parallel rollout')] + text += "import json,sys\n" + if extra: + text += "print(" + repr(extra) + ",flush=True)\n" + if marker: + text += ( + 'print(json.dumps({"event":"catchup_admission_deferred","version":1,' + '"reason":"busy_before_claim","target":"table","attempt_id":' + + ( + '"foreign-attempt"' + if foreign + else 'os.environ["CATCHUP_ATTEMPT_ID"]' + ) + + "}),flush=True)\n" + ) + text += "sys.exit(" + str(code) + ")\n" + self.stub.write_text(text) + + def release_terminal(self): + self.wait_for(lambda: (self.folder / "started").exists()) + (self.folder / "release").touch() + self.wait_for(lambda: json.loads(self.read("attempt"))["state"] != "admitted") + return json.loads(self.read("attempt")) + + def test_confirmed_busy_defers_without_earning_failure_and_retries(self): + self.configure_outcome() + previous = { + "job": "persistent-job", + "consecutive_failures": 2, + "needs_attention": True, + "next_retry_ms": 1, + "state": "failed", + } + self.put("attempt", json.dumps(previous)) + failure = json.dumps( + {"next_retry_ms": 1, "last_error": "earned-storage-failure"} + ) + self.put("failure", failure) + self.start() + terminal = self.release_terminal() + self.assertEqual(terminal["state"], "deferred") + self.assertEqual(terminal["consecutive_failures"], 2) + self.assertTrue(terminal["needs_attention"]) + self.assertGreaterEqual( + terminal["next_retry_ms"], terminal["finished_at"] * 1000 + 1900 + ) + self.assertLessEqual(terminal["next_retry_ms"], time.time() * 1000 + 5100) + self.assertEqual(self.read("failure"), failure) + self.wait_for( + lambda: len((self.folder / "started").read_text().splitlines()) >= 2, + seconds=7, + ) + self.assertEqual(self.read("owner"), "persistent-job") + + def test_unproven_or_partial_outcomes_keep_failure_backoff(self): + # Each case runs a fresh child+real-etcd prefix in a separate test fixture. + # Pure marker variants are additionally covered by the classifier tests. + self.configure_outcome( + extra="parallel rollout append completed reclaimed=41 failed=0" + ) + self.start() + terminal = self.release_terminal() + self.assertEqual(terminal["state"], "failed") + self.assertEqual(terminal["reclaimed"], 41) + self.assertGreater(terminal["next_retry_ms"], time.time() * 1000 + 110000) + + def test_exit75_without_marker_is_failure(self): + self.configure_outcome(marker=False) + self.start() + self.assertEqual(self.release_terminal()["state"], "failed") + + def test_foreign_attempt_marker_is_failure(self): + self.configure_outcome(foreign=True) + self.start() + self.assertEqual(self.release_terminal()["state"], "failed") + + def test_claimed_zero_work_is_not_preclaim_busy(self): + self.configure_outcome(extra="catch-up task claimed target=table") + self.start() + self.assertEqual(self.release_terminal()["state"], "failed") + + def test_old_error_text_does_not_shorten_failure_backoff(self): + self.configure_outcome( + code=1, + marker=False, + extra="dedicated catch-up failed target=table error=catch-up admission remained busy for 30 seconds", + ) + self.start() + self.assertEqual(self.release_terminal()["state"], "failed") + + def test_concurrent_native_failure_deadline_is_preserved_after_busy(self): + self.configure_outcome() + self.start() + self.wait_for(lambda: (self.folder / "started").exists()) + deadline = int((time.time() + 600) * 1000) + failure = json.dumps( + { + "next_retry_ms": deadline, + "needs_attention": True, + "last_error": "unknown storage outcome", + } + ) + self.put("failure", failure) + terminal = self.release_terminal() + self.assertEqual(terminal["state"], "deferred") + self.assertEqual(terminal["next_retry_ms"], deadline) + self.assertTrue(terminal["needs_attention"]) + self.assertEqual(self.read("failure"), failure) + time.sleep(0.2) + self.assertEqual(len((self.folder / "started").read_text().splitlines()), 1) + + def test_terminal_operator_race_leaves_admitted_debt_intact(self): + self.configure_outcome() + self.start() + self.wait_for(lambda: (self.folder / "started").exists()) + admitted = self.read("attempt") + self.operator["native_pod_uid"] = "replacement" + self.put("operator", json.dumps(self.operator)) + (self.folder / "release").touch() + self.wait_for( + lambda: ( + "attempt_record_changed" in (self.folder / "supervisor.log").read_text() + ) + ) + self.assertEqual(self.read("attempt"), admitted) + self.assertEqual(self.read("owner"), "persistent-job") + def test_operator_change_after_uid_check_is_rejected_by_full_cas(self): fixture = self raced = threading.Event() @@ -311,5 +441,47 @@ def log_message(self, *_args): self.assertEqual(self.read("owner"), "persistent-job") +class BusyContractTests(unittest.TestCase): + def test_requires_exact_terminal_contract(self): + from native_supervisor import confirmed_busy + + event = { + "event": "catchup_admission_deferred", + "version": 1, + "reason": "busy_before_claim", + "target": "table", + "attempt_id": "id", + } + baseline = [75, [event], "id", "table", False, 0, 0] + self.assertTrue(confirmed_busy(*baseline)) + variants = [ + (0, 1), + (0, 0), + (0, -9), + (1, []), + (1, [event, event]), + (2, "other"), + (3, "other"), + (4, True), + (5, 1), + (6, 1), + ] + for index, value in variants: + with self.subTest(index=index, value=value): + args = list(baseline) + args[index] = value + self.assertFalse(confirmed_busy(*args)) + for field, value in [ + ("version", True), + ("version", 2), + ("reason", "unknown"), + ("attempt_id", None), + ("extra", "unexpected"), + ]: + with self.subTest(field=field, value=value): + bad = {**event, field: value} + self.assertFalse(confirmed_busy(75, [bad], "id", "table", False, 0, 0)) + + if __name__ == "__main__": unittest.main()