fix: defer native catch-up busy admission without failure backoff

.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

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<ExecuteOutcome> {
     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<T, F, Fut>(
     mut attempt: F,
     deadline: tokio::time::Instant,
-) -> Result<T>
+) -> Result<Option<T>>
 where
     F: FnMut() -> Fut,
     Fut: Future<Output = Result<Option<T>>>,
@@ -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::<Option<()>, _>("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);

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};

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() {

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;
     }

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