From fb8d5687e5599c34def898c5e4a42aa2311e494e Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Mon, 5 Oct 2026 08:25:55 +0000 Subject: [PATCH] fix: keep dedicated merge admission local to its target --- .../src/catchup/executor.rs | 81 ++- crates/lance-context-master/src/task_store.rs | 494 ++++++++++++++++-- docs/merge-task-admission.md | 56 ++ 3 files changed, 573 insertions(+), 58 deletions(-) create mode 100644 docs/merge-task-admission.md diff --git a/crates/lance-context-master/src/catchup/executor.rs b/crates/lance-context-master/src/catchup/executor.rs index a8373041..d6810703 100644 --- a/crates/lance-context-master/src/catchup/executor.rs +++ b/crates/lance-context-master/src/catchup/executor.rs @@ -21,28 +21,29 @@ pub async fn execute(mut config: MasterConfig, target: &str) -> Result<()> { .job_name .clone() .ok_or("missing catch-up Job identity")?; + let startup = tokio::time::Instant::now(); let state = MasterState::new(config).await.map_err(|e| e.to_string())?; + tracing::info!(%target, startup_seconds = startup.elapsed().as_secs_f64(), "catch-up state ready"); // 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 + let admission_started = tokio::time::Instant::now(); + let claim = claim_before_deadline( + || async { + 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")??; + .map_err(|e| e.to_string()) + }, + admission_started + Duration::from_secs(30), + ) + .await?; + tracing::info!(%target, admission_seconds = admission_started.elapsed().as_secs_f64(), + "catch-up task claimed"); 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. @@ -78,6 +79,31 @@ pub async fn execute(mut config: MasterConfig, target: &str) -> Result<()> { Ok(()) } +async fn claim_before_deadline( + mut attempt: F, + deadline: tokio::time::Instant, +) -> Result +where + F: FnMut() -> Fut, + Fut: Future>>, +{ + loop { + // Bound retries between attempts, not by dropping an in-flight + // 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()); + } + if let Some(claim) = attempt().await? { + return Ok(claim); + } + tokio::time::sleep_until( + deadline.min(tokio::time::Instant::now() + Duration::from_secs(1)), + ) + .await; + } +} + async fn merge_passes(state: &Arc, target: &str) -> Result { if state.config.append.enabled(target) { return crate::rollout_append::run(state, target).await; @@ -279,6 +305,37 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::{Notify, Semaphore}; + #[tokio::test] + async fn admission_delivers_claim_accepted_across_retry_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; + Ok(Some("accepted claim")) + }, + deadline, + ) + .await + .unwrap(); + assert_eq!(result, "accepted claim"); + } + + #[tokio::test] + async fn admission_stops_retrying_unclaimed_work_at_deadline() { + let calls = AtomicUsize::new(0); + let deadline = tokio::time::Instant::now() + Duration::from_millis(50); + let result = claim_before_deadline( + || async { + calls.fetch_add(1, Ordering::Relaxed); + Ok(None::<()>) + }, + deadline, + ) + .await; + assert!(result.unwrap_err().contains("admission remained busy")); + assert_eq!(calls.load(Ordering::Relaxed), 1); + } + #[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/task_store.rs b/crates/lance-context-master/src/task_store.rs index f7dbaf47..ddedd35a 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -12,7 +12,7 @@ use std::sync::Arc; use std::time::Duration; use chrono::Utc; -use etcd_client::{Client, Compare, CompareOp, GetOptions, PutOptions, Txn, TxnOp}; +use etcd_client::{Client, Compare, CompareOp, GetOptions, PutOptions, Txn, TxnOp, TxnOpResponse}; use lance_context_api::{RepairRecord, TaskCooldown, TaskKind, TaskRecord, TaskState}; use lance_context_core::generate_id; use tokio::sync::oneshot; @@ -71,6 +71,7 @@ pub struct TaskStore { history_limit: usize, history_ttl_secs: u64, cooldown: CooldownPolicy, + global_maintenance: bool, } /// When to stop re-enqueueing a target that keeps failing, and for how long. @@ -239,9 +240,15 @@ impl TaskStore { base: Duration::from_secs(config.task_cooldown_base_secs), max: Duration::from_secs(config.task_cooldown_max_secs), }, + global_maintenance: config.catchup.target.is_none(), }; - store.recover_orphaned().await?; - store.prune_terminal_history().await?; + // Short-lived exact-target executors must not sweep the fleet's running + // tasks or terminal history on every invocation. Ordinary masters own + // that maintenance; dedicated admission reconciles its own task below. + if store.global_maintenance { + store.recover_orphaned().await?; + store.prune_terminal_history().await?; + } Ok(store) } @@ -306,10 +313,14 @@ impl TaskStore { 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); }; + if let Some(task) = self.inner.get(&id).await? { + if task.state == TaskState::Running { + self.inner.recover_tasks(&[task]).await?; + } + } Ok(self .inner .claim_next(TaskKinds::MERGE_WAL, Some((&id, job))) @@ -342,7 +353,10 @@ impl TaskStore { tracing::warn!(kind = ?kind, target = %target, %error, "cooldown bookkeeping failed"); } } - self.prune_terminal_history().await.map(|_| ()) + if self.global_maintenance { + self.prune_terminal_history().await?; + } + Ok(()) } /// Whether the sweeps should skip this target for now because it has @@ -663,11 +677,37 @@ impl EtcdTaskStore { } } - let merge_execution = - lance_context_merge::Coordinator::new(self.client.clone(), self.prefix.clone()) - .get(&task.target) - .await - .map_err(lance::Error::io)?; + let execution_key = lance_context_merge::execution_key(&self.prefix, &task.target); + let snapshot = self + .read_values(&[ + execution_key.clone(), + self.target_lock_key(&task.target), + crate::catchup::store::active_key(&self.prefix, &task.target), + execution_key.replace("/merge-executions/", "/merge-claims/"), + self.claim_key(&task.id), + ]) + .await?; + let merge_execution: Option = snapshot[0] + .as_deref() + .map(serde_json::from_slice) + .transpose() + .map_err(|e| lance::Error::io(format!("invalid merge execution: {e}")))?; + // Avoid lease grant/revoke for a clearly blocked candidate. + // The final transaction still checks every ownership predicate. + let expected_owner = merge_execution + .as_ref() + .map(lance_context_merge::execution_owner); + let expected_job = only_id.map(|(_, job)| job.as_bytes()); + if snapshot[4].is_some() + || (requires_target_lock(task.kind) + && (snapshot[2].as_deref() != expected_job + || snapshot[3].is_some() + || snapshot[1].as_deref() + != expected_owner.as_deref().map(str::as_bytes))) + { + metrics::counter!("master_task_admission_blocked_total").increment(1); + continue; + } // Only MergeWal reconciles a worker execution. A local fenced // mutation may be recovered by any table writer after the // exclusive reconciler lease has expired. @@ -680,6 +720,15 @@ impl EtcdTaskStore { } let token = generate_id(); let lease_id = self.grant_lease().await?; + // Establish renewal before publishing ownership, so a slow + // keepalive handshake cannot strand an already accepted claim. + let keepalive = match self.start_keepalive(lease_id).await { + Ok(keepalive) => keepalive, + Err(error) => { + let _ = self.revoke_lease(lease_id).await; + return Err(error); + } + }; let claim_key = self.claim_key(&task.id); let target_key = requires_target_lock(task.kind).then(|| self.target_lock_key(&task.target)); @@ -749,13 +798,21 @@ impl EtcdTaskStore { } } let mut client = self.client.clone(); - let claimed = client + let response = client .txn(Txn::new().when(compares).and_then(operations)) + .await; + let claimed = match self + .resolve_claim_response(response, &claim_key, &token) .await - .map_err(etcd_error("claim task"))? - .succeeded(); + { + Ok(claimed) => claimed, + Err(error) => { + drop(keepalive); + let _ = self.revoke_lease(lease_id).await; + return Err(error); + } + }; if claimed { - let keepalive = self.start_keepalive(lease_id).await?; return Ok(( Some(TaskClaim { task, @@ -770,6 +827,7 @@ impl EtcdTaskStore { dependency_failed, )); } + drop(keepalive); self.revoke_lease(lease_id).await?; } @@ -857,47 +915,148 @@ impl EtcdTaskStore { Ok(()) } + async fn resolve_claim_response( + &self, + response: Result, + claim_key: &str, + token: &str, + ) -> lance::Result { + match response { + Ok(response) => Ok(response.succeeded()), + Err(error) => { + // A transport deadline can arrive after the atomic claim was + // accepted. Adopt only our unique token, never another owner. + if self.get_text(claim_key).await?.as_deref() == Some(token) { + metrics::counter!("master_task_claim_response_recovered_total").increment(1); + Ok(true) + } else { + Err(etcd_error("claim task")(error)) + } + } + } + } + async fn recover_orphaned(&self) -> lance::Result { - let mut client = self.client.clone(); - let response = client - .get(self.running_prefix(), Some(GetOptions::new().with_prefix())) - .await - .map_err(etcd_error("list running tasks"))?; - let running = response - .kvs() - .iter() - .map(|kv| decode_task(kv.value(), &String::from_utf8_lossy(kv.key()))) - .collect::>>()?; + // Bound each read and batch live-claim checks: a healthy running task + // needs no conditional recovery transaction. Keep the original CAS + // authority for missing claims, since preflight reads may become stale. + let mut start = self.running_prefix().into_bytes(); + let end = prefix_range_end(&start); let mut recovered = 0; - for mut task in running { - let running_value = encode_task(&task)?; - let claim_key = self.claim_key(&task.id); - let running_key = self.running_key(&task.id); - requeue(&mut task); - let txn = Txn::new() - .when([ - Compare::value(self.task_key(&task.id), CompareOp::Equal, running_value), - Compare::version(claim_key.as_str(), CompareOp::Equal, 0), - Compare::version(running_key.as_str(), CompareOp::Greater, 0), - ]) - .and_then([ - TxnOp::put(self.task_key(&task.id), encode_task(&task)?, None), - TxnOp::put(self.queue_key(&task.id), encode_task(&task)?, None), - TxnOp::delete(running_key, None), - ]); - let mut client = self.client.clone(); - if client - .txn(txn) + loop { + let page = self + .client + .clone() + .get( + start, + Some(GetOptions::new().with_range(end.clone()).with_limit(64)), + ) .await - .map_err(etcd_error("recover orphaned task"))? - .succeeded() - { + .map_err(etcd_error("list running tasks"))?; + let tasks = page + .kvs() + .iter() + .map(|kv| decode_task(kv.value(), &String::from_utf8_lossy(kv.key()))) + .collect::>>()?; + recovered += self.recover_tasks(&tasks).await?; + if !page.more() { + return Ok(recovered); + } + start = page + .kvs() + .last() + .ok_or_else(|| lance::Error::io("empty recovery page"))? + .key() + .to_vec(); + start.push(0); + } + } + + /// Exact keys in one read-only transaction; never a fleet prefix scan. + async fn read_values(&self, keys: &[String]) -> lance::Result>>> { + let response = self + .client + .clone() + .txn( + Txn::new().and_then( + keys.iter() + .map(|key| TxnOp::get(key.as_str(), None)) + .collect::>(), + ), + ) + .await + .map_err(etcd_error("read admission snapshot"))?; + let responses = response.op_responses(); + if responses.len() != keys.len() { + return Err(lance::Error::io("incomplete admission snapshot")); + } + responses + .into_iter() + .map(|op| match op { + TxnOpResponse::Get(value) => Ok(value.kvs().first().map(|kv| kv.value().to_vec())), + _ => Err(lance::Error::io("invalid admission snapshot response")), + }) + .collect() + } + + async fn recover_tasks(&self, tasks: &[TaskRecord]) -> lance::Result { + if tasks.is_empty() { + return Ok(0); + } + let claims = self + .read_values( + &tasks + .iter() + .map(|t| self.claim_key(&t.id)) + .collect::>(), + ) + .await?; + let mut recovered = 0; + for (task, claim) in tasks.iter().zip(claims) { + if claim.is_some() { + continue; + } + if self.recover_task(task).await? { recovered += 1; } } + metrics::counter!("master_task_recovery_inspected_total").increment(tasks.len() as u64); + metrics::counter!("master_task_recovery_requeued_total").increment(recovered as u64); Ok(recovered) } + async fn recover_task(&self, task: &TaskRecord) -> lance::Result { + let mut queued = task.clone(); + requeue(&mut queued); + let value = encode_task(&queued)?; + let txn = Txn::new() + .when([ + Compare::value( + self.task_key(&task.id), + CompareOp::Equal, + encode_task(task)?, + ), + Compare::version(self.claim_key(&task.id), CompareOp::Equal, 0), + Compare::value( + self.running_key(&task.id), + CompareOp::Equal, + encode_task(task)?, + ), + ]) + .and_then([ + TxnOp::put(self.task_key(&task.id), value.clone(), None), + TxnOp::put(self.queue_key(&task.id), value, None), + TxnOp::delete(self.running_key(&task.id), None), + ]); + Ok(self + .client + .clone() + .txn(txn) + .await + .map_err(etcd_error("recover orphaned task"))? + .succeeded()) + } + async fn try_coordination_lock(&self, name: &str) -> lance::Result> { let lease_id = self.grant_lease().await?; let token = generate_id(); @@ -1461,6 +1620,249 @@ mod tests { use super::*; use tempfile::TempDir; + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn benchmark_target_admission_with_eighty_healthy_tasks() { + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + cfg.etcd.etcd_prefix = format!("/admission-benchmark/{}", generate_id()); + cfg.catchup.target = Some("hot".into()); + let store = TaskStore::open(&cfg).await.unwrap(); + let lease = store.inner.grant_lease().await.unwrap(); + let keepalive = store.inner.start_keepalive(lease).await.unwrap(); + let mut client = store.inner.client.clone(); + for i in 0..80 { + let mut task = new_task(TaskKind::MergeWal, &format!("healthy-{i}"), Vec::new()); + task.state = TaskState::Running; + let value = encode_task(&task).unwrap(); + client + .txn(Txn::new().and_then([ + TxnOp::put(store.inner.task_key(&task.id), value.clone(), None), + TxnOp::put(store.inner.running_key(&task.id), value, None), + TxnOp::put( + store.inner.claim_key(&task.id), + "live", + Some(PutOptions::new().with_lease(lease)), + ), + ])) + .await + .unwrap(); + } + // Reproduce the old sweep's sequential conditional transactions. All + // predicates fail because these fixture tasks have healthy claims. + let baseline = std::time::Instant::now(); + let running = client + .get( + store.inner.running_prefix(), + Some(GetOptions::new().with_prefix()), + ) + .await + .unwrap(); + for kv in running.kvs() { + let task = decode_task(kv.value(), "fixture").unwrap(); + assert!(!store.inner.recover_task(&task).await.unwrap()); + } + let old_sweep_seconds = baseline.elapsed().as_secs_f64(); + let batch = std::time::Instant::now(); + assert_eq!(store.inner.recover_orphaned().await.unwrap(), 0); + let batched_sweep_seconds = batch.elapsed().as_secs_f64(); + client + .put( + crate::catchup::store::active_key(&store.inner.prefix, "hot"), + "job", + None, + ) + .await + .unwrap(); + store + .enqueue(TaskKind::MergeWal, "hot", Vec::new()) + .await + .unwrap(); + let start = std::time::Instant::now(); + let claim = store + .claim_merge_target("hot", "job") + .await + .unwrap() + .unwrap(); + let target_claim_seconds = start.elapsed().as_secs_f64(); + println!( + "{}", + serde_json::json!({"healthy_running":80, "old_sweep_seconds":old_sweep_seconds, + "batched_sweep_seconds":batched_sweep_seconds, "target_claim_seconds":target_claim_seconds}) + ); + store.finish(claim, Ok("done".into())).await.unwrap(); + drop(keepalive); + store.inner.revoke_lease(lease).await.unwrap(); + client + .delete( + cfg.etcd.etcd_prefix, + Some(etcd_client::DeleteOptions::new().with_prefix()), + ) + .await + .unwrap(); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn dedicated_admission_ignores_unrelated_history_and_recovers_own_orphan() { + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + cfg.etcd.etcd_prefix = format!("/admission-test/{}", generate_id()); + cfg.catchup.target = Some("hot".into()); + let backend = EtcdTaskStore::connect(&cfg).await.unwrap(); + let mut client = backend.client.clone(); + // These intentionally undecodable records expose any accidental fleet + // scan during dedicated open, claim, recovery, or finish. + let unrelated_running = backend.running_key("unrelated"); + let unrelated_history = backend.task_key("unrelated"); + for key in [&unrelated_running, &unrelated_history] { + client + .put(key.as_str(), "not a task record", None) + .await + .unwrap(); + } + client + .put( + crate::catchup::store::active_key(&backend.prefix, "hot"), + "job", + None, + ) + .await + .unwrap(); + let store = TaskStore::open(&cfg).await.unwrap(); + let task = store + .enqueue(TaskKind::MergeWal, "hot", Vec::new()) + .await + .unwrap(); + assert!(store + .claim_merge_target("hot", "wrong-job") + .await + .unwrap() + .is_none()); + let claim = store + .claim_merge_target("hot", "job") + .await + .unwrap() + .unwrap(); + assert_eq!(claim.task.id, task.id); + assert!( + store + .claim_merge_target("hot", "job") + .await + .unwrap() + .is_none(), + "a live claim must remain protected" + ); + store.abandon_claim_for_test(claim).await.unwrap(); + let recovered = store + .claim_merge_target("hot", "job") + .await + .unwrap() + .unwrap(); + assert_eq!(recovered.task.id, task.id); + store.finish(recovered, Ok("merged".into())).await.unwrap(); + assert_eq!( + store.get(&task.id).await.unwrap().unwrap().state, + TaskState::Done + ); + for key in [&unrelated_running, &unrelated_history] { + assert_eq!( + backend.get_text(key).await.unwrap().as_deref(), + Some("not a task record") + ); + } + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn lost_claim_response_adopts_only_its_own_token() { + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + cfg.etcd.etcd_prefix = format!("/admission-test/{}", generate_id()); + let store = TaskStore::open(&cfg).await.unwrap(); + store + .enqueue(TaskKind::MergeWal, "hot", Vec::new()) + .await + .unwrap(); + let claim = store.claim_next().await.unwrap().unwrap(); + let lost = || { + Err(etcd_client::Error::Internal( + "simulated lost response".into(), + )) + }; + assert!(store + .inner + .resolve_claim_response(lost(), &claim.backend.claim_key, &claim.backend.token) + .await + .unwrap()); + assert!(store + .inner + .resolve_claim_response(lost(), &claim.backend.claim_key, "another-owner") + .await + .is_err()); + store.abandon_claim_for_test(claim).await.unwrap(); + assert!(store + .inner + .resolve_claim_response(lost(), &store.inner.claim_key("absent"), "another-owner") + .await + .is_err()); + } + + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn batched_recovery_preserves_live_claims_and_crosses_pages() { + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS") + .unwrap() + .split(',') + .map(str::to_owned) + .collect(); + cfg.etcd.etcd_prefix = format!("/admission-test/{}", generate_id()); + let store = TaskStore::open(&cfg).await.unwrap(); + store + .enqueue(TaskKind::MergeWal, "live", Vec::new()) + .await + .unwrap(); + let live = store.claim_next().await.unwrap().unwrap(); + let mut client = store.inner.client.clone(); + for i in 0..70 { + let mut task = new_task(TaskKind::MergeWal, &format!("orphan-{i}"), Vec::new()); + task.state = TaskState::Running; + let value = encode_task(&task).unwrap(); + client + .txn(Txn::new().and_then([ + TxnOp::put(store.inner.task_key(&task.id), value.clone(), None), + TxnOp::put(store.inner.running_key(&task.id), value, None), + ])) + .await + .unwrap(); + } + assert_eq!(store.inner.recover_orphaned().await.unwrap(), 70); + assert_eq!( + store.get(&live.task.id).await.unwrap().unwrap().state, + TaskState::Running + ); + assert_eq!(store.inner.recover_orphaned().await.unwrap(), 0); + assert_eq!(store.queue_depth().await.unwrap(), 70); + store.finish(live, Ok("done".into())).await.unwrap(); + } + #[tokio::test] #[ignore = "requires ETCD_TEST_ENDPOINTS"] async fn compact_noop_lease_expiry_allows_bounded_recheck() { diff --git a/docs/merge-task-admission.md b/docs/merge-task-admission.md new file mode 100644 index 00000000..f2d9ef45 --- /dev/null +++ b/docs/merge-task-admission.md @@ -0,0 +1,56 @@ +# Bounded work for merge task admission + +Dedicated catch-up processes previously scanned all running tasks during both +startup and exact-target admission. Each scan sent a conditional recovery +transaction for every running task, even when its claim was healthy. They also +read global task history on startup and completion. A busy fleet could therefore +make a free target miss its admission deadline and enter external retry backoff. + +Dedicated processes now skip global recovery/history maintenance and reconcile +only the task referenced by their target's merge dedupe key. Ordinary masters +retain global maintenance. Their running-task scan uses 64-record pages and one +batched claim read per page; only tasks without observed claims reach the existing +recovery CAS. The CAS still checks the canonical task, absent claim and exact +running record. A stale observation cannot requeue a newly claimed task. + +Candidate admission reads the execution, target owner, native guard, merge claim +and task claim together before allocating a lease. Clearly blocked candidates +are skipped without a lease grant/revoke pair. This preflight is an optimization: +all final atomic ownership comparisons remain. Dependency handling, queue +pagination and kind filtering remain in their original order. + +Lease renewal starts before the claim transaction. A transport error after that +transaction triggers a read of the unique claim token; an accepted claim is +adopted only when the token matches this attempt. Otherwise the attempt stops +renewing and requests lease revocation. If cleanup cannot reach etcd, lease +expiry and ordinary/target-local recovery still apply. No storage fence or +persistent execution ownership is cleared by this recovery. + +The dedicated 30-second admission limit bounds retry admission, not an already +in-flight claim attempt. Individual etcd RPCs still have ten-second timeouts. +A successful attempt may finish after the retry deadline and must be delivered +to its executor, rather than dropped by an outer timeout. This does not raise +merge execution or no-progress deadlines. + +## Validation and observation + +Regression tests cover unrelated malformed records, exact native Job ownership, +live-claim exclusion, recovery of the target's own expired claim, recovery across +64-record page boundaries, token verification after a simulated lost response, +and delivery of a successful attempt across its retry deadline. Existing etcd +tests continue to exercise HA exclusion, dependencies and worker execution locks. + +The ignored `benchmark_target_admission_with_eighty_healthy_tasks` test creates an +isolated etcd prefix and compares the old sequential recovery transactions with +batched healthy-claim checking. It also times a complete exact-target claim. +It asserts correctness, not a timing ratio, and removes its fixture prefix. +Run it with `ETCD_TEST_ENDPOINTS` pointing to local or staging etcd. + +`catch-up state ready` and `catch-up task claimed` logs expose startup and +admission seconds separately. Bounded-cardinality counters record inspected and +requeued recovery tasks, preflight-blocked candidates, and accepted claims whose +transaction response was lost. Production acceptance requires shorter admission +times and actual subsequent WAL publication; Pod readiness alone is insufficient. + +This change does not modify merge slots, staging concurrency, append commits, +retry budgets, native ownership guards, or master/worker deployment settings.