From e041109130a0e0a9c47e3ce28d5b5d08188d9e20 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 18:51:46 -0600 Subject: [PATCH] Roll back failed staging and verify successful cutover under concurrent queries --- control_plane/src/backend_client.rs | 27 +++++ control_plane/src/main.rs | 5 +- data_plane/src/drivers/query/servers/http.rs | 26 +++++ .../types/hot_reload_config.rs | 40 +++++++ data_plane/tests/backend_process_e2e.rs | 103 +++++++++++++++++- 5 files changed, 199 insertions(+), 2 deletions(-) diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 230f7095..de50a861 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -396,6 +396,33 @@ impl BackendClient { } } + pub async fn discard_staged_physical_plan( + &self, + plan_id: u64, + plan_version: u64, + ) -> std::result::Result<(), BackendPostError> { + let response = self + .http + .post(format!( + "{}/discard", + derive_physical_plan_url(&self.endpoint) + )) + .json(&serde_json::json!({"plan_id": plan_id, "plan_version": plan_version})) + .send() + .await + .map_err(classify_reqwest_error)?; + let status = response.status(); + if status.is_success() { + Ok(()) + } else { + Err(classify_http_status( + status, + response.text().await.unwrap_or_default(), + "PhysicalPlan discard POST", + )) + } + } + pub async fn activate_physical_plan( &self, plan_id: u64, diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 263bfed3..73838cc7 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -675,9 +675,12 @@ async fn handle_compile_and_publish_physical_plan( .publish_collector_plans(&bundle.collector_plans, apply_timeout) .await { + let cleanup = backend + .discard_staged_physical_plan(bundle.envelope.plan_id, bundle.envelope.plan_version) + .await; return ( StatusCode::BAD_GATEWAY, - format!("collector physical-plan publication failed: {error}"), + format!("collector physical-plan publication failed: {error}; staged backend cleanup: {cleanup:?}"), ) .into_response(); } diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 5bdd2a4b..717cb081 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -482,6 +482,10 @@ impl HttpServer { get(handle_get_backend_plan).post(handle_post_backend_plan), ) .route("/api/v1/physical-plan", post(handle_post_physical_plan)) + .route( + "/api/v1/physical-plan/discard", + post(handle_discard_physical_plan), + ) .route( "/api/v1/physical-plan/activate", post(handle_activate_physical_plan), @@ -588,6 +592,10 @@ impl HttpServer { get(handle_get_backend_plan).post(handle_post_backend_plan), ) .route("/api/v1/physical-plan", post(handle_post_physical_plan)) + .route( + "/api/v1/physical-plan/discard", + post(handle_discard_physical_plan), + ) .route( "/api/v1/physical-plan/activate", post(handle_activate_physical_plan), @@ -6076,6 +6084,24 @@ async fn handle_activate_physical_plan( .into_response() } +async fn handle_discard_physical_plan( + State(state): State, + axum::Json(request): axum::Json, +) -> axum::response::Response { + use axum::response::IntoResponse; + let Some(lifecycle) = state.physical_plan_lifecycle.as_ref() else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "physical-plan lifecycle is not attached", + ) + .into_response(); + }; + match lifecycle.discard_staged(request.plan_id, request.plan_version) { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(error) => (StatusCode::CONFLICT, error.to_string()).into_response(), + } +} + async fn handle_physical_plan_status(State(state): State) -> axum::response::Response { use axum::response::IntoResponse; let Some(lifecycle) = state.physical_plan_lifecycle.as_ref() else { diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index 959a123b..5d75f2b6 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -309,6 +309,27 @@ impl PhysicalPlanLifecycle { Ok(()) } + /// Roll back a failed publication without touching active readers or state. + pub fn discard_staged( + &self, + plan_id: u64, + plan_version: u64, + ) -> Result<(), PhysicalPlanLifecycleError> { + let key = (plan_id, plan_version); + let mut state = self + .state + .lock() + .expect("physical-plan lifecycle lock poisoned"); + if state.staged.remove(&key).is_none() { + return Err(PhysicalPlanLifecycleError::NotStaged { + plan_id, + plan_version, + }); + } + state.statuses.remove(&key); + Ok(()) + } + pub fn activate( &self, plan_id: u64, @@ -981,6 +1002,25 @@ mod tests { assert_eq!(lifecycle.statuses()[0].phase, PhysicalPlanPhase::Retired); } + // Failed publication releases only its staging slot; active readers remain valid. + #[test] + fn discard_staged_allows_retry_and_never_discards_active() { + let active = HotReloadActivePhysicalPlan::new(physical_plan(7, 1, 100, None)); + let held_reader = active.snapshot(); + let lifecycle = PhysicalPlanLifecycle::new(active.clone()); + lifecycle + .stage(physical_plan(7, 2, 200, None), 150) + .unwrap(); + lifecycle.discard_staged(7, 2).unwrap(); + lifecycle + .stage(physical_plan(7, 2, 300, None), 250) + .unwrap(); + lifecycle.activate(7, 2, 300).unwrap(); + assert!(lifecycle.discard_staged(7, 2).is_err()); + assert_eq!(active.snapshot().backend_plan.plan_version, 2); + assert_eq!(held_reader.backend_plan.plan_version, 1); + } + #[test] fn materialization_readiness_is_generation_scoped_and_monotonic() { let active = HotReloadActivePhysicalPlan::new(physical_plan(7, 1, 100, None)); diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index beca3d17..2495614b 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -626,7 +626,7 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { failed_body.contains("injected Collector staging failure"), "{failed_body}" ); - let (rejected_plan, _collector_socket) = collector.await.unwrap(); + let (rejected_plan, collector_socket) = collector.await.unwrap(); let rejected_frame = client .post(format!("http://{otlp_http}/v1/metrics")) .header("content-type", "application/x-protobuf") @@ -671,6 +671,107 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { .await .unwrap(); assert_eq!(first_scalar(&still_warm), Some(value)); + // Retry the staged successor while queries are in flight. Each + // request must retain a complete active snapshot through cutover. + request["activation_unix_ms"] = (std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 + + 250) + .into(); + let reader_client = client.clone(); + let reader_base = data_base.clone(); + let readers = tokio::spawn(async move { + for _ in 0..40 { + let response: serde_json::Value = reader_client + .get(format!("{reader_base}/api/v1/query")) + .query(&[ + ("query", query.to_string()), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!( + first_scalar(&response), + Some(value), + "torn serving snapshot: {response}" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + }); + let collector = tokio::spawn(respond_next_collector_plan(collector_socket, true)); + let activated = client + .post(format!( + "{control_base}/api/v1/physical-plan/compile-and-publish" + )) + .json(&request) + .send() + .await + .unwrap(); + let status = activated.status(); + let body = activated.text().await.unwrap(); + assert!( + status.is_success(), + "successful successor rejected: {status}: {body}" + ); + let activated: serde_json::Value = serde_json::from_str(&body).unwrap(); + assert_eq!(activated["plan_version"], 2); + let (successor, _collector_socket) = collector.await.unwrap(); + readers.await.unwrap(); + let old_frame = client + .post(format!("http://{otlp_http}/v1/metrics")) + .header("content-type", "application/x-protobuf") + .body(ddsketch_export( + "whole_process_e2e_latency_ms", + sample_ns, + &[9999.0], + planned_alpha, + &collector_plan, + 2, + )) + .send() + .await + .unwrap(); + assert_eq!( + old_frame.status().as_u16(), + 422, + "retired generation accepted a write" + ); + let new_frame = client + .post(format!("http://{otlp_http}/v1/metrics")) + .header("content-type", "application/x-protobuf") + .body(ddsketch_export( + "whole_process_e2e_latency_ms", + sample_ns + 1_000_000_000, + &[200.0; 101], + planned_alpha, + &successor, + 1, + )) + .send() + .await + .unwrap(); + assert!(new_frame.status().is_success()); + let new_result: serde_json::Value = client + .get(format!("{data_base}/api/v1/query")) + .query(&[ + ("query", query.to_string()), + ("time", ((window_end_ms + 1000) as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!( + (first_scalar(&new_result).unwrap() - 200.0).abs() <= 200.0 * planned_alpha * 1.05, + "{new_result}" + ); return; } last_response = response;