diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 2742ebf..ed00937 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -297,6 +297,7 @@ async fn run_merge_wal(state: &Arc, target: &str) -> Result, target: &str) -> Result { ok_workers += 1; } - Err(e) => last_err = Some(e.to_string()), + Err(e) => { + failed_workers += 1; + last_err = Some(e.to_string()); + } } } metrics::counter!("master_merge_wal_generations_reclaimed_total").increment(reclaimed as u64); - if ok_workers == 0 { - return Err(last_err.unwrap_or_else(|| "all workers failed".to_string())); + // A worker that owns no shard for this store answers 404 or reclaims 0, + // and that is fine -- but it must not launder a real failure elsewhere + // into success. If any worker errored and the task as a whole made no + // progress, it failed: this is what a store with a broken base table + // looks like (the one worker holding its data 500s, the rest have + // nothing), and it is what the failure cooldown needs to see. Partial + // progress with some errors is still success; the next sweep retries + // the stragglers. + if ok_workers == 0 || (failed_workers > 0 && reclaimed == 0) { + return Err(format!( + "{}/{total_workers} workers failed and nothing was merged: {}", + failed_workers, + last_err.unwrap_or_else(|| "all workers failed".to_string()) + )); } Ok(format!( "merged {reclaimed} generations across {ok_workers}/{total_workers} workers" @@ -940,6 +956,49 @@ mod tests { worker.abort(); } + /// A worker that reclaims nothing plus a worker that errors is a failed + /// task, not "merged 0 across 1/2 workers": that shape is exactly a store + /// whose base table is broken (the shard owner 500s, the others have no + /// data), and the failure cooldown can only act on it if it is reported + /// as a failure. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn merge_wal_with_an_erroring_worker_and_no_progress_fails() { + use axum::{http::StatusCode, routing::post, Json, Router}; + + async fn spawn(ok: bool) -> String { + let app = Router::new().route( + "/api/v1/internal/merge-wal/{name}", + post(move || async move { + if ok { + (StatusCode::OK, Json(serde_json::json!({ "reclaimed": 0 }))) + } else { + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": "Not found: data/frag.lance" })), + ) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") + } + + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.worker_endpoints = vec![spawn(true).await, spawn(false).await]; + let state = MasterState::new(cfg).await.unwrap(); + let worker = spawn_scheduler(&state); + + let rec = enqueue(&state, TaskKind::MergeWal, "broken").await.unwrap(); + let status = await_terminal(&state, &rec.id).await; + assert_eq!(status.state, TaskState::Failed, "got {status:?}"); + assert!(status.error.unwrap().contains("nothing was merged")); + worker.abort(); + } + /// Workers are called one at a time. Every worker's merge commits to the /// same base table and Lance's commit-conflict retry gives up after 30s, so /// concurrent fan-out is N writers racing one commit point -- the failure