fix: detect lost merge owners and stop revoked work promptly

crates/lance-context-master/src/merge_execution.rs

@@ -7,6 +7,14 @@ const RPC_TIMEOUT: Duration = Duration::from_secs(10);
 const POLL_DELAY: Duration = Duration::from_millis(250);
 const RETRY_DELAY: Duration = Duration::from_secs(2);
 const ATTEMPTS: usize = 3;
+const OWNER_PROBE_INTERVAL: Duration = Duration::from_secs(10);
+const OWNER_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+const OWNER_PROBE_FAILURES: u8 = 3;
+
+#[derive(serde::Deserialize)]
+struct ExecutorIdentity {
+    instance: String,
+}
 
 #[derive(serde::Deserialize)]
 struct Capabilities {
@@ -194,6 +202,12 @@ pub(crate) async fn recover_execution(
     old: Execution,
 ) -> Result<(), String> {
     let result = async {
+        // Completion can race a failed identity probe or the end of cancel
+        // grace. A terminal executor needs release, not a new storage barrier.
+        if matches!(old.phase, Phase::Finished | Phase::Recovered) {
+            return coordinator.release(proof, &old).await?.then_some(())
+                .ok_or_else(|| "claim lost before releasing terminal merge execution".to_string());
+        }
         let frozen = coordinator.freeze(proof, &old).await?
             .ok_or_else(|| "merge execution changed while freezing commit admission".to_string())?;
         let watermarks = coordinator.watermarks(&frozen).await?;
@@ -415,6 +429,8 @@ async fn reconcile_with_grace(
     let mut deadlines = Deadlines::new(&initial, tokio::time::Instant::now());
     let mut cancel_started = cancel_immediately.then(tokio::time::Instant::now);
     let mut next_cancel = tokio::time::Instant::now();
+    let mut next_probe = tokio::time::Instant::now() + OWNER_PROBE_INTERVAL;
+    let mut failed_probes = 0;
     loop {
         // Losing etcd responses must not turn this into an unbounded scheduler
         // wait. Ownership remains durable when the reconciliation slot exits.
@@ -454,12 +470,64 @@ async fn reconcile_with_grace(
         }
         if cancel_started.is_none() {
             if let Ok(Some(progress)) = coordinator.progress(&current).await {
+                if deadlines
+                    .sequence
+                    .is_none_or(|previous| progress.sequence > previous)
+                {
+                    // Completed work outweighs an unavailable HTTP endpoint.
+                    failed_probes = 0;
+                }
                 deadlines.observe(progress.sequence, tokio::time::Instant::now());
             }
             if deadlines.expired(tokio::time::Instant::now()) {
                 cancel_started = Some(tokio::time::Instant::now());
             }
         }
+        if tokio::time::Instant::now() >= next_probe {
+            let probe = async {
+                http.get(format!(
+                    "{}/api/v1/internal/merge-executor",
+                    current.endpoint.trim_end_matches('/')
+                ))
+                .timeout(OWNER_PROBE_TIMEOUT)
+                .send()
+                .await?
+                .error_for_status()?
+                .json::<ExecutorIdentity>()
+                .await
+            }
+            .await;
+            next_probe = tokio::time::Instant::now() + OWNER_PROBE_INTERVAL;
+            let reason = match probe {
+                Ok(identity) if identity.instance == current.instance => {
+                    failed_probes = 0;
+                    None
+                }
+                Ok(_) => Some("merge executor incarnation changed"),
+                Err(_) => {
+                    failed_probes += 1;
+                    (failed_probes >= OWNER_PROBE_FAILURES)
+                        .then_some("merge executor unreachable without progress")
+                }
+            };
+            if let Some(reason) = reason {
+                if current.phase == Phase::Reserved {
+                    // No execution has started. CAS cancellation either wins
+                    // admission or the next observation sees the running owner.
+                    let _ = coordinator.cancel_reserved(&current).await;
+                    continue;
+                }
+                // This is a reason to revoke admission and run the storage
+                // barrier, never permission to release ownership directly.
+                let error = format!("merge ownership unresolved: {reason}; recovery required");
+                coordinator
+                    .record_failure(proof, &current.target, &current.endpoint, &error)
+                    .await?;
+                tracing::warn!(target = %current.target, endpoint = %current.endpoint,
+                    id = %current.id, %reason, "recovering lost merge executor before execution deadline");
+                return Err(error);
+            }
+        }
         if cancel_started.is_some() && tokio::time::Instant::now() >= next_cancel {
             next_cancel = tokio::time::Instant::now() + RETRY_DELAY;
             // CAS reserved work to terminal, or ask its owner to cancel the
@@ -699,6 +767,18 @@ mod tests {
             .await
             .is_err());
 
+        // A probe may decide to recover just as the executor publishes its
+        // terminal result. Release that result without trying to freeze it.
+        let completed = Execution::new("table", &lost, "late-completion", 1);
+        assert!(coordinator.reserve(&proof, &completed).await.unwrap());
+        let completed = coordinator.start(&completed).await.unwrap().unwrap();
+        assert!(coordinator.finish(&completed, Ok(0)).await.unwrap());
+        let completed = coordinator.get("table").await.unwrap().unwrap();
+        recover_execution(&state, &coordinator, &proof, completed)
+            .await
+            .unwrap();
+        assert!(coordinator.get("table").await.unwrap().is_none());
+
         // A mismatched worker namespace must remain fenced, never recover by
         // committing a barrier to an unrelated same-named table.
         let wrong = Execution::new("table", &lost, "misconfigured-worker", 1);
@@ -731,6 +811,106 @@ mod tests {
             .unwrap();
     }
 
+    async fn verify_lost_executor(restarted: bool) {
+        let (coordinator, proof) = fixture().await;
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let endpoint = format!("http://{}", listener.local_addr().unwrap());
+        let app = if restarted {
+            Router::new().route(
+                "/api/v1/internal/merge-executor",
+                get(|| async { Json(serde_json::json!({"instance": "new-process"})) }),
+            )
+        } else {
+            Router::new()
+        };
+        let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        let execution = Execution::new("table", &endpoint, "old-process", 3600);
+        assert!(coordinator.reserve(&proof, &execution).await.unwrap());
+        let running = coordinator.start(&execution).await.unwrap().unwrap();
+        assert!(coordinator.publish_progress(&running, 0).await.unwrap());
+        let error = tokio::time::timeout(
+            Duration::from_secs(if restarted { 15 } else { 40 }),
+            reconcile(
+                &reqwest::Client::new(),
+                &coordinator,
+                &proof,
+                running.clone(),
+                false,
+            ),
+        )
+        .await
+        .expect("lost executor must be detected before the 600-second idle timeout")
+        .unwrap_err();
+        assert!(
+            error.contains(if restarted {
+                "incarnation changed"
+            } else {
+                "unreachable"
+            }),
+            "{error}"
+        );
+        assert_eq!(coordinator.get("table").await.unwrap(), Some(running));
+        assert!(!coordinator
+            .reserve(&proof, &Execution::new("table", "other", "new", 3600))
+            .await
+            .unwrap());
+        server.abort();
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn restarted_executor_is_detected_before_idle_deadline() {
+        verify_lost_executor(true).await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn unreachable_executor_is_detected_before_idle_deadline() {
+        verify_lost_executor(false).await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn real_progress_prevents_takeover_when_http_probes_fail() {
+        let (coordinator, proof) = fixture().await;
+        let execution = Execution::new("table", "http://127.0.0.1:1", "worker", 3600);
+        assert!(coordinator.reserve(&proof, &execution).await.unwrap());
+        let running = coordinator.start(&execution).await.unwrap().unwrap();
+        assert!(coordinator.publish_progress(&running, 0).await.unwrap());
+        let publisher = coordinator.clone();
+        let working = running.clone();
+        let work = tokio::spawn(async move {
+            for step in 1..=35 {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                assert!(publisher.publish_progress(&working, step).await.unwrap());
+            }
+            assert!(publisher.finish(&working, Ok(7)).await.unwrap());
+        });
+        assert_eq!(
+            tokio::time::timeout(
+                Duration::from_secs(45),
+                reconcile(
+                    &reqwest::Client::new(),
+                    &coordinator,
+                    &proof,
+                    running,
+                    false
+                )
+            )
+            .await
+            .unwrap()
+            .unwrap(),
+            7
+        );
+        work.await.unwrap();
+        assert!(coordinator.get("table").await.unwrap().is_none());
+        assert!(coordinator
+            .failure("table", &execution.endpoint)
+            .await
+            .unwrap()
+            .is_none());
+    }
+
     #[tokio::test]
     #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
     async fn missing_executor_returns_attention_without_unlocking_surviving_write() {

crates/lance-context-server/src/merge_execution.rs

@@ -267,7 +267,9 @@ async fn run(
             dataset_uri: uri.clone(),
         }),
     );
-    let outcome = std::panic::AssertUnwindSafe(async {
+    // Erase the nested merge future so adding the independent watchdog does
+    // not exceed the compiler's async layout recursion limit.
+    let execute: futures::future::BoxFuture<'_, Result<usize, String>> = Box::pin(async {
         // No payload work starts before admission to the process-wide slot.
         slot = execute_scoped(
             async { Ok(state.acquire_merge_slot().await) },
@@ -290,6 +292,12 @@ async fn run(
             cancelled,
         )
         .await
+    });
+    let outcome = std::panic::AssertUnwindSafe(async {
+        tokio::select! {
+            result = execute => result,
+            error = watch_ownership(&coordinator, &running) => Err(error),
+        }
     })
     .catch_unwind()
     .await
@@ -365,6 +373,24 @@ async fn run(
     Ok(())
 }
 
+// Check independently of progress and slot admission. A revoked executor may
+// be stuck before its first checkpoint, and the HTTP cancel may never arrive.
+// This only cancels preparation; the caller still drains or fences admitted
+// storage writes before releasing the merge slot and publishing a terminal state.
+async fn watch_ownership(coordinator: &Coordinator, execution: &Execution) -> String {
+    loop {
+        tokio::time::sleep(Duration::from_secs(2)).await;
+        match tokio::time::timeout(Duration::from_secs(2), coordinator.get(&execution.target)).await
+        {
+            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.
+            _ => {}
+        }
+    }
+}
+
 async fn watch_progress(
     coordinator: &Coordinator,
     execution: &Execution,
@@ -561,6 +587,82 @@ mod tests {
         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) {
+        let (state, coordinator, proof, _dir) = deadline_fixture().await;
+        let store = state.get_or_open_rollout_store("hot").await.unwrap();
+        let held_store = store.write().await;
+        let held_slot = if queue_blocked {
+            state.acquire_merge_slot().await
+        } else {
+            None
+        };
+        let execution = Execution::new("hot", "worker", &state.merge_executions.instance, 3600);
+        assert!(coordinator.reserve(&proof, &execution).await.unwrap());
+        start(State(state.clone()), Json(execution.clone()))
+            .await
+            .unwrap();
+        let running = tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                let current = coordinator.get("hot").await.unwrap().unwrap();
+                if current.phase == Phase::Running
+                    && (queue_blocked || coordinator.progress(&current).await.unwrap().is_some())
+                {
+                    break current;
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .unwrap();
+        let frozen = coordinator.freeze(&proof, &running).await.unwrap().unwrap();
+        tokio::time::timeout(Duration::from_secs(6), async {
+            while !state.merge_executions.running.lock().await.is_empty() {
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .expect("revoked executor must stop without waiting for idle/queue timeout");
+        assert_eq!(
+            state.merge_slots.as_ref().unwrap().available_permits(),
+            usize::from(!queue_blocked)
+        );
+        // Local task exit alone does not unlock the table or acknowledge the
+        // storage barrier. An attempted replacement is still rejected.
+        assert_eq!(coordinator.get("hot").await.unwrap(), Some(frozen.clone()));
+        let replacement = Execution::new("hot", "worker", &state.merge_executions.instance, 3600);
+        assert!(!coordinator.reserve(&proof, &replacement).await.unwrap());
+        let watermarks = coordinator.watermarks(&frozen).await.unwrap();
+        assert!(watermarks.versions.is_empty());
+        // There were no admitted writes, so the storage barrier is empty.
+        assert!(coordinator.finish_recovery(&proof, &frozen).await.unwrap());
+        let recovered = coordinator.get("hot").await.unwrap().unwrap();
+        assert!(coordinator.release(&proof, &recovered).await.unwrap());
+        drop(held_slot);
+        drop(held_store);
+        assert!(coordinator.reserve(&proof, &replacement).await.unwrap());
+        start(State(state.clone()), Json(replacement))
+            .await
+            .unwrap();
+        let finished = terminal(&coordinator).await;
+        assert!(finished.error.is_none(), "{finished:?}");
+        assert!(coordinator.release(&proof, &finished).await.unwrap());
+        state.merge_executions.shutdown().await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn revoked_executor_stops_while_waiting_for_a_slot() {
+        revoked_executor_exits(true).await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn revoked_executor_stops_without_any_progress() {
+        revoked_executor_exits(false).await;
+    }
+
     #[tokio::test]
     #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
     async fn disconnected_http_call_is_cancelled_before_ownership_handoff() {

docs/owned-maintenance-recovery.md

@@ -59,3 +59,28 @@ this change cannot retroactively fence an old unguarded write. It does not enabl
 owned targets, alter production pods, or complete registry migration. Recovery
 continues on draining targets while an owned execution remains. Validate large
 table timings and fault recovery before selecting tighter deadlines.
+
+## Prompt worker loss detection
+
+For owned WAL merges, the master probes the executor identity every 10 seconds
+with a two-second request timeout. A changed process incarnation triggers storage
+recovery on the next probe. Three consecutive failed probes without observed
+merge progress also trigger recovery (normally about 30–36 seconds). A successful
+probe or a newly completed merge step clears the failed-probe count. These are
+recovery triggers, not evidence that a remote storage commit has stopped: the
+existing version barrier must still succeed before a replacement can write.
+
+Workers independently check execution ownership every two seconds, including
+while waiting for a merge slot and before the first processing checkpoint. A
+revoked execution cancels its work even when its cancel HTTP request was lost.
+The merge slot remains held until admitted storage writes drain or the durable
+storage barrier permits their cancellation. This terminates the individual merge
+execution, not the worker process or other tables' work.
+
+A live worker with no completed processing steps still uses
+`MERGE_IDLE_TIMEOUT_SECS` (default 600). Identity probes are not merge progress
+and cannot extend this deadline. Reduce that setting only after measuring the
+longest individual generation read or base-table write for the target workload;
+there is no sub-operation progress for a single long storage call. These changes
+do not activate ownership fencing for legacy tables, launch replacement pods, or
+provide an OS-level watchdog for a completely wedged worker process.