From 100f87ce1cdb4e2432886da310feb3d5e937b685 Mon Sep 17 00:00:00 2001 From: Beinan Date: Thu, 8 Oct 2026 19:08:43 +0000 Subject: [PATCH] fix(master): read merge progress at one etcd revision with a typed response `/merge-progress` did four independent etcd reads (execution, progress, work report, execution again) and, when they disagreed, returned `progress_changed_retry` or `execution_changed_retry`, pushing read consistency onto the caller. The response was hand-assembled JSON with no schema. Add `Coordinator::progress_snapshot`: one read for the execution id, then one transaction guarded on the execution bytes that fetches progress and work report together, so all three come from a single revision (bounded to three attempts; a target flipping executions faster than two etcd reads is not a status-endpoint concern). A work report whose sequence trails the current progress sequence is dropped rather than rendered beside a newer heartbeat. The handler returns `MergeProgressReport`, a serde struct with a fixed field set and a `status` enum of four values; the two retry statuses are gone. Absent fields are omitted so existing consumers that only read `status`/`task_id` are unaffected. `revision` is new. Validation: merge-crate test extended for the snapshot, stale-report dropping and the absent-target case; routes test asserts on the typed struct and its wire form. Both pass against real etcd. Clippy -D warnings, fmt and typos clean. Refs #333. Co-Authored-By: Claude Fable 5 --- crates/lance-context-master/src/routes.rs | 123 +++++++++++++-------- crates/lance-context-merge/src/lib.rs | 29 +++++ crates/lance-context-merge/src/progress.rs | 80 +++++++++++++- docs/merge-work-progress.md | 14 ++- 4 files changed, 197 insertions(+), 49 deletions(-) diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index bb3f7a2..9af5fdb 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -9,7 +9,7 @@ use axum::http::{header, HeaderValue, StatusCode}; use axum::response::Response; use axum::routing::{get, post}; use axum::{Json, Router}; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use lance_context_api::{ CompactJobStatus, EnqueueTaskRequest, ExperimentDetail, ExperimentListResponse, @@ -711,12 +711,51 @@ struct MergeProgressQuery { target: String, } +/// What `/merge-progress` can say about a target. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +enum MergeProgressStatus { + /// Execution, progress sequence and a matching work report were read + /// at one etcd revision. + Reported, + /// An execution exists but has published no work report for its + /// current progress sequence (older executor binary, or not yet). + ProgressDetailsUnavailable, + /// A MergeWal task is live but no execution record exists: a legacy + /// worker RPC with no telemetry. Not evidence of health. + ProgressUnknown, + /// No live task and no execution record. + NoActiveRecord, +} + +/// Typed response for `GET /api/v1/merge-progress?target=`. Field names are +/// a contract; add fields, never rename them. +#[derive(Debug, Serialize)] +struct MergeProgressReport { + target: String, + status: MergeProgressStatus, + #[serde(skip_serializing_if = "Option::is_none")] + task_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + execution: Option, + #[serde(skip_serializing_if = "Option::is_none")] + progress: Option, + #[serde(skip_serializing_if = "Option::is_none")] + work_report: Option, + /// etcd revision the execution/progress/work triple was read at. + #[serde(skip_serializing_if = "Option::is_none")] + revision: Option, + sampled_at_ms: u64, +} + /// Exact target reads only. A live task/lease without execution telemetry is /// explicitly unknown, not evidence that a legacy merge is making progress. +/// Execution, progress and work report come from one etcd revision, so the +/// caller never has to retry on a torn read. async fn merge_progress( State(state): State>, Query(query): Query, -) -> Result, MasterError> { +) -> Result, MasterError> { if query.target.is_empty() || query.target.len() > 512 { return Err(MasterError::InvalidRequest("invalid merge target".into())); } @@ -726,48 +765,41 @@ async fn merge_progress( .get_active_id(TaskKind::MergeWal, &query.target) .await .map_err(MasterError::from_lance)?; - let Some(execution) = coordinator - .get(&query.target) + let sampled_at_ms = lance_context_merge::failure::now_ms(); + let Some(snapshot) = coordinator + .progress_snapshot(&query.target) .await .map_err(MasterError::Internal)? else { - return Ok(Json( - serde_json::json!({"target": query.target, "task_id": task_id, "status": if task_id.is_some() { "progress_unknown" } else { "no_active_record" }}), - )); + return Ok(Json(MergeProgressReport { + target: query.target, + status: if task_id.is_some() { + MergeProgressStatus::ProgressUnknown + } else { + MergeProgressStatus::NoActiveRecord + }, + task_id, + execution: None, + progress: None, + work_report: None, + revision: None, + sampled_at_ms, + })); }; - let progress = coordinator - .progress(&execution) - .await - .map_err(MasterError::Internal)?; - let work = coordinator - .work_progress(&execution) - .await - .map_err(MasterError::Internal)?; - if work - .as_ref() - .is_some_and(|report| report["sequence"].as_u64() != progress.as_ref().map(|p| p.sequence)) - { - return Ok(Json( - serde_json::json!({"target":query.target,"status":"progress_changed_retry"}), - )); - } - if coordinator - .get(&query.target) - .await - .map_err(MasterError::Internal)? - .as_ref() - != Some(&execution) - { - return Ok(Json( - serde_json::json!({"target":query.target,"status":"execution_changed_retry"}), - )); - } - Ok(Json(serde_json::json!({ - "target": query.target, "task_id": task_id, "execution": execution, - "progress": progress, "work_report": work, - "sampled_at_ms": lance_context_merge::failure::now_ms(), - "status": if work.is_some() { "reported" } else { "progress_details_unavailable" }, - }))) + Ok(Json(MergeProgressReport { + target: query.target, + status: if snapshot.work_report.is_some() { + MergeProgressStatus::Reported + } else { + MergeProgressStatus::ProgressDetailsUnavailable + }, + task_id, + execution: Some(snapshot.execution), + progress: snapshot.progress, + work_report: snapshot.work_report, + revision: Some(snapshot.revision), + sampled_at_ms, + })) } async fn drain_executor( @@ -937,9 +969,12 @@ mod tests { ) .await .unwrap(); - assert_eq!(report["status"], "progress_unknown"); - assert_eq!(report["task_id"], claim.task.id); - assert!(report.get("work_report").is_none()); + assert_eq!(report.status, MergeProgressStatus::ProgressUnknown); + assert_eq!(report.task_id.as_deref(), Some(claim.task.id.as_str())); + assert!(report.work_report.is_none()); + let wire = serde_json::to_value(&report).unwrap(); + assert_eq!(wire["status"], "progress_unknown"); + assert!(wire.get("execution").is_none(), "absent fields are omitted"); state .task_store .finish(claim, Ok("done".into())) @@ -953,7 +988,7 @@ mod tests { ) .await .unwrap(); - assert_eq!(report["status"], "no_active_record"); + assert_eq!(report.status, MergeProgressStatus::NoActiveRecord); } fn test_record(id: &str, with_blob: bool) -> RolloutRecord { diff --git a/crates/lance-context-merge/src/lib.rs b/crates/lance-context-merge/src/lib.rs index ab2cf4b..35e782e 100644 --- a/crates/lance-context-merge/src/lib.rs +++ b/crates/lance-context-merge/src/lib.rs @@ -731,6 +731,35 @@ mod tests { .sequence, 4 ); + // One-revision snapshot: execution, sequence and report agree. + let snapshot = coordinator + .progress_snapshot("table") + .await + .unwrap() + .unwrap(); + assert_eq!(snapshot.execution, running); + assert_eq!(snapshot.progress.as_ref().map(|p| p.sequence), Some(4)); + assert_eq!(snapshot.work_report, Some(report.clone())); + assert!(snapshot.revision > 0); + // A heartbeat that advances the sequence without a new report must + // not render the old report as current. + assert!(coordinator.publish_progress(&running, 5).await.unwrap()); + let snapshot = coordinator + .progress_snapshot("table") + .await + .unwrap() + .unwrap(); + assert_eq!(snapshot.progress.as_ref().map(|p| p.sequence), Some(5)); + assert!(snapshot.work_report.is_none(), "stale report dropped"); + assert!(coordinator + .publish_work_progress(&running, 4, report.clone()) + .await + .unwrap()); + assert!(coordinator + .progress_snapshot("absent") + .await + .unwrap() + .is_none()); // A stale observation cannot revoke completed new work. Details have // their own key, so the established exact-byte sequence CAS still works. assert!(!coordinator.revoke_stalled(&running, Some(3)).await.unwrap()); diff --git a/crates/lance-context-merge/src/progress.rs b/crates/lance-context-merge/src/progress.rs index 5cbdcf6..253fc1d 100644 --- a/crates/lance-context-merge/src/progress.rs +++ b/crates/lance-context-merge/src/progress.rs @@ -1,13 +1,26 @@ //! 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, Phase, Result}; -use etcd_client::{Compare, CompareOp, TxnOp}; +use etcd_client::{Compare, CompareOp, Txn, TxnOp, TxnOpResponse}; #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct ExecutionProgress { pub sequence: u64, } +/// Execution, progress sequence and work report read together at one etcd +/// revision. `work_report` is only present when it was published for the +/// same sequence as `progress`; a report from an older sequence is dropped +/// rather than shown beside a newer heartbeat. +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct ProgressSnapshot { + pub execution: Execution, + pub progress: Option, + pub work_report: Option, + /// etcd revision the snapshot was taken at. + pub revision: i64, +} + impl Coordinator { fn work_key(&self, execution: &Execution) -> String { format!( @@ -62,6 +75,71 @@ impl Coordinator { .transpose() } + /// Read execution, progress and work report for `target` atomically. + /// + /// Two round trips: the first learns the execution id (the progress and + /// work keys are per execution, not per target), the second reads all + /// three keys in one transaction so they come from a single revision. + /// If the execution changed between the two reads the transaction's + /// compare fails and we loop; a target that flips executions faster than + /// two etcd reads is not something a status endpoint needs to render. + pub async fn progress_snapshot(&self, target: &str) -> Result> { + let execution_key = execution_key(&self.prefix, target); + for _ in 0..3 { + let Some(execution) = self.get(target).await? else { + return Ok(None); + }; + let encoded = encode(&execution); + let response = self + .client + .clone() + .txn( + Txn::new() + .when(vec![Compare::value( + execution_key.clone(), + CompareOp::Equal, + encoded, + )]) + .and_then(vec![ + TxnOp::get(self.progress_key(&execution), None), + TxnOp::get(self.work_key(&execution), None), + ]), + ) + .await + .map_err(|e| e.to_string())?; + if !response.succeeded() { + continue; + } + let revision = response.header().map_or(0, |h| h.revision()); + let mut values = response.op_responses().into_iter().map(|op| match op { + TxnOpResponse::Get(get) => get.kvs().first().map(|kv| kv.value().to_vec()), + _ => None, + }); + let progress: Option = values + .next() + .flatten() + .map(|v| serde_json::from_slice(&v).map_err(|e| e.to_string())) + .transpose()?; + let work_report: Option = values + .next() + .flatten() + .map(|v| serde_json::from_slice(&v).map_err(|e| e.to_string())) + .transpose()?; + let work_report = work_report.filter(|report| { + report["sequence"].as_u64() == progress.as_ref().map(|p| p.sequence) + }); + return Ok(Some(ProgressSnapshot { + execution, + progress, + work_report, + revision, + })); + } + Err(format!( + "merge execution for {target} changed on every read; retry later" + )) + } + pub(crate) fn remove_work_progress(&self, execution: &Execution) -> TxnOp { TxnOp::delete(self.work_key(execution), None) } diff --git a/docs/merge-work-progress.md b/docs/merge-work-progress.md index de1bfb8..2e51ecf 100644 --- a/docs/merge-work-progress.md +++ b/docs/merge-work-progress.md @@ -27,10 +27,16 @@ for progress detection. Detailed reports and the established sequence are published in one execution-qualified transaction. The sequence keeps its original wire format, -so mixed-version no-progress compare-and-swap remains valid. The API rejects a -sample if the execution changed or independently read sequence/report versions -disagree. Consumers must compare samples of the same execution and check the -report timestamp; `reported` is not a health verdict. Zero append counters on a +so mixed-version no-progress compare-and-swap remains valid. The endpoint reads +execution, sequence and report in one etcd transaction guarded on the execution +bytes, so every response is a single-revision sample; the `revision` field +carries it. A report published for an older sequence than the current one is +omitted rather than shown beside a newer heartbeat. The response is a fixed +schema (`target`, `status`, `task_id`, `execution`, `progress`, `work_report`, +`revision`, `sampled_at_ms`; absent fields are omitted) with `status` one of +`reported`, `progress_details_unavailable`, `progress_unknown`, +`no_active_record`. Consumers must compare samples of the same execution and +check the report timestamp; `reported` is not a health verdict. Zero append counters on a different maintenance kind are not evidence that its work is stalled. The report is live execution state and is removed on normal release. A final