fix(master): read merge progress at one etcd revision with a typed response

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<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    execution: Option<lance_context_merge::Execution>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    progress: Option<lance_context_merge::progress::ExecutionProgress>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    work_report: Option<serde_json::Value>,
+    /// etcd revision the execution/progress/work triple was read at.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    revision: Option<i64>,
+    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<Arc<MasterState>>,
     Query(query): Query<MergeProgressQuery>,
-) -> Result<Json<serde_json::Value>, MasterError> {
+) -> Result<Json<MergeProgressReport>, 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 {

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

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<ExecutionProgress>,
+    pub work_report: Option<serde_json::Value>,
+    /// 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<Option<ProgressSnapshot>> {
+        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<ExecutionProgress> = values
+                .next()
+                .flatten()
+                .map(|v| serde_json::from_slice(&v).map_err(|e| e.to_string()))
+                .transpose()?;
+            let work_report: Option<serde_json::Value> = 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)
     }

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