diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index c3d8228a..d2bd0e75 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -324,6 +324,41 @@ impl BackendClient { )) } } + + /// POST an encoded `BackendPlan` (protobuf bytes) to the backend's + /// `POST /api/v1/backend-plan` endpoint (see + /// `control_plane/docs/design-backend-plan-wire-format.md`), sent + /// alongside the streaming-config/storage-routing push, not in place + /// of it (see `emit::backend_push`'s call site). Same + /// transient/permanent classification as the other typed POST + /// methods. + pub async fn post_backend_plan_typed( + &self, + bytes: Vec, + ) -> std::result::Result<(), BackendPostError> { + let url = derive_backend_plan_url(&self.endpoint); + debug!( + endpoint = %url, + plan_bytes = bytes.len(), + "posting BackendPlan to ASAPQuery-backend (typed)" + ); + let resp = self + .http + .post(&url) + .header("content-type", "application/x-protobuf") + .body(bytes) + .send() + .await + .map_err(classify_reqwest_error)?; + + let status = resp.status(); + if status.is_success() { + Ok(()) + } else { + let body = resp.text().await.unwrap_or_default(); + Err(classify_http_status(status, body, "BackendPlan POST")) + } + } } /// Map a streaming-config endpoint URL to the sibling storage-routing @@ -344,6 +379,21 @@ fn derive_storage_routing_url(endpoint: &str) -> String { endpoint.to_string() } +/// Map a streaming-config endpoint URL to the sibling `backend-plan` +/// endpoint, same rewrite convention as [`derive_storage_routing_url`]. +fn derive_backend_plan_url(endpoint: &str) -> String { + const STREAMING_PATH_DASH: &str = "/api/v1/streaming-config"; + const STREAMING_PATH_UNDERSCORE: &str = "/api/v1/streaming_config"; + const PLAN_PATH: &str = "/api/v1/backend-plan"; + if let Some(stripped) = endpoint.strip_suffix(STREAMING_PATH_DASH) { + return format!("{stripped}{PLAN_PATH}"); + } + if let Some(stripped) = endpoint.strip_suffix(STREAMING_PATH_UNDERSCORE) { + return format!("{stripped}{PLAN_PATH}"); + } + endpoint.to_string() +} + /// Fire-and-forget convenience helper used by the replanner. Logs /// errors at WARN and never propagates them — the replanner should /// never fail an entire replan because the backend was temporarily @@ -497,6 +547,73 @@ mod tests { assert_eq!(derive_storage_routing_url("http://x/foo"), "http://x/foo"); } + #[test] + fn backend_plan_url_rewrites_streaming_path() { + assert_eq!( + derive_backend_plan_url("http://backend:8088/api/v1/streaming-config"), + "http://backend:8088/api/v1/backend-plan" + ); + assert_eq!( + derive_backend_plan_url("http://backend:8088/api/v1/streaming_config"), + "http://backend:8088/api/v1/backend-plan" + ); + } + + #[test] + fn backend_plan_url_preserves_unknown_paths_for_tests() { + assert_eq!( + derive_backend_plan_url("http://127.0.0.1:1/api/v1/backend-plan"), + "http://127.0.0.1:1/api/v1/backend-plan" + ); + assert_eq!(derive_backend_plan_url("http://x/foo"), "http://x/foo"); + } + + #[tokio::test] + async fn backend_plan_post_round_trips_bytes_via_url_rewrite() { + let hits: StdArc>>> = StdArc::new(Mutex::new(Vec::new())); + let hits_for_route = hits.clone(); + let app = Router::new() + .route( + "/api/v1/backend-plan", + post(move |body: axum::body::Bytes| { + let hits = hits_for_route.clone(); + async move { + hits.lock().unwrap().push(body.to_vec()); + axum::http::StatusCode::OK + } + }), + ); + 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(); + }); + tokio::time::sleep(Duration::from_millis(50)).await; + + let client = BackendClient::new(format!("http://{addr}/api/v1/streaming-config")); + let bytes = vec![1u8, 2, 3, 4]; + client + .post_backend_plan_typed(bytes.clone()) + .await + .expect("backend-plan post ok"); + + let received = hits.lock().unwrap(); + assert_eq!(received.len(), 1); + assert_eq!(received[0], bytes); + } + + #[tokio::test] + async fn backend_plan_post_404_is_transient() { + let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); + let url = start_mock_backend(sink.clone(), axum::http::StatusCode::NOT_FOUND).await; + let client = BackendClient::new(url); + let err = client + .post_backend_plan_typed(vec![1, 2, 3]) + .await + .expect_err("404 should surface as Err"); + assert!(err.is_transient(), "404 must classify as transient: {err}"); + } + /// Phase α: full happy path. A mock backend hosts the storage /// routing endpoint; the client POSTs the control-plane-emitted JSON /// and the body round-trips verbatim. Mirrors `json_post_round_trips_body`. diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index 7fcba9e1..9e8f7a2a 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -44,11 +44,12 @@ use std::collections::{BTreeMap, HashMap}; // import is gated to keep the non-test build warning-free. #[cfg(test)] use std::future::Future; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::Mutex; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; use crate::backend_client::{BackendClient, BackendPostError}; use crate::emit::{emit_backend_storage_routing, emit_backend_streaming_config_json}; @@ -73,6 +74,19 @@ const RETRY_MAX_ATTEMPTS: u32 = 5; const RETRY_BASE_DELAY: Duration = Duration::from_millis(100); const RETRY_DELAY_CAP: Duration = Duration::from_millis(2700); +/// Monotonic counter for `BackendPlan.plan_id` — observability only, not +/// identity (see `BackendPlan`'s own doc). One process-wide sequence is +/// enough; there's no existing streaming-config version counter to +/// reuse for parity. +static PLAN_ID_COUNTER: AtomicU64 = AtomicU64::new(1); + +fn now_unix_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + /// Cheap process-wide jitter source. We don't have `rand` in the /// control plane's dependency set and don't want to add it for one /// call site — `Instant::elapsed` reads the monotonic clock which is @@ -280,6 +294,30 @@ async fn push_documents_coupled( (streaming_ok, routing_ok, RETRY_MAX_ATTEMPTS) } +/// Best-effort, single-attempt push of the encoded `BackendPlan` — no +/// in-function retry loop, unlike [`push_documents_coupled`]. A dropped +/// push just leaves `data_plane`'s serving-time lookup falling back to +/// `SketchStore` reconstruction until the next replan cycle re-pushes, +/// so the next cycle is itself the retry backstop — same contract +/// [`push_or_log`] already establishes for the legacy YAML path. Logs at +/// WARN on failure; never affects [`PushOutcome`], which real callers +/// key legacy-path behavior on. +async fn push_backend_plan_best_effort(client: &Arc, bytes: Vec) { + match client.post_backend_plan_typed(bytes).await { + Ok(()) => { + debug!(stage = "backend", endpoint = %client.endpoint(), "BackendPlan push succeeded"); + } + Err(e) => { + warn!( + stage = "backend", + endpoint = %client.endpoint(), + error = %e, + "BackendPlan push failed; next replan cycle will retry" + ); + } + } +} + /// Update the cumulative cache with `be` for `(metric, role)` and /// POST the cumulative streaming-config + storage-routing JSON /// documents to the backend. @@ -435,6 +473,26 @@ async fn push_cumulative_entries( } }; + // BackendPlan (design-backend-plan-wire-format.md): built from the + // SAME `cumulative_be` snapshot as the legacy documents above, so all + // three describe one consistent generation of planning state. This + // is a dual-push, alongside (not instead of) the legacy + // streaming-config / storage-routing documents — a failure here must + // never affect `PushOutcome`, which existing callers key real + // behavior on. + let plan_bytes = match crate::backend_plan::from_stage_config( + &cumulative_be, + monitors, + PLAN_ID_COUNTER.fetch_add(1, Ordering::Relaxed), + now_unix_ms(), + ) { + Ok(plan) => Some(plan.encode_to_vec()), + Err(e) => { + warn!(error = %e, "backend_plan::from_stage_config failed; skipping BackendPlan push (legacy push unaffected)"); + None + } + }; + // Storage-routing: the routing classifier (`build_routing_entry` in // `emit/stage_config.rs`) reads `cfg.aggregations` to derive shape // routing, so we MUST merge every role's aggregations for one metric @@ -495,6 +553,13 @@ async fn push_cumulative_entries( let (streaming_ok, routing_ok, attempts) = push_documents_coupled(client, streaming_body, routing_body).await; + // Best-effort BackendPlan push — same backoff schedule as the legacy + // documents, but its own outcome never feeds into `PushOutcome` (see + // this function's doc above `plan_bytes`). + if let Some(bytes) = plan_bytes { + push_backend_plan_best_effort(client, bytes).await; + } + if streaming_ok && routing_ok { info!( stage = "backend", @@ -770,6 +835,7 @@ mod tests { struct DualMock { streaming_hits: StdArc, routing_hits: StdArc, + plan_hits: StdArc, streaming_status: axum::http::StatusCode, routing_status: axum::http::StatusCode, } @@ -781,6 +847,7 @@ mod tests { let mock = DualMock { streaming_hits: StdArc::new(StdAtomicU32::new(0)), routing_hits: StdArc::new(StdAtomicU32::new(0)), + plan_hits: StdArc::new(StdAtomicU32::new(0)), streaming_status, routing_status, }; @@ -803,6 +870,15 @@ mod tests { }, ), ) + .route( + "/api/v1/backend-plan", + post( + |State(m): State, _body: axum::body::Bytes| async move { + m.plan_hits.fetch_add(1, StdOrdering::SeqCst); + axum::http::StatusCode::OK + }, + ), + ) .with_state(mock.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -836,6 +912,70 @@ mod tests { assert_eq!(mock.routing_hits.load(StdOrdering::SeqCst), 1); } + /// The dual-push also fires a best-effort `POST /api/v1/backend-plan`, + /// alongside — not instead of — the legacy documents. + #[tokio::test] + async fn coupled_push_also_fires_backend_plan_push() { + let (url, mock) = + start_dual_mock(axum::http::StatusCode::OK, axum::http::StatusCode::OK).await; + let client = StdArc::new(BackendClient::new(url)); + let cache = Mutex::new(HashMap::new()); + let outcome = post_typed_backend_for_role( + Some(&client), + &cache, + "latency", + AggRole::Quantile, + make_be("latency", "q"), + &[], + ) + .await; + assert_eq!(outcome, PushOutcome::BothApplied); + assert_eq!(mock.plan_hits.load(StdOrdering::SeqCst), 1); + } + + /// A BackendPlan push failure (backend doesn't implement the + /// endpoint yet, or returns an error) must NOT affect `PushOutcome` + /// — nothing depends on the plan push succeeding in this phase. + #[tokio::test] + async fn backend_plan_push_failure_does_not_affect_push_outcome() { + // A mock that only serves the legacy endpoints (no + // `/api/v1/backend-plan` route) — the plan push 404s. + let app = Router::new() + .route( + "/api/v1/streaming-config", + post(|_body: axum::body::Bytes| async { axum::http::StatusCode::OK }), + ) + .route( + "/api/v1/storage_routing", + post(|_body: axum::body::Bytes| async { axum::http::StatusCode::OK }), + ); + 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(); + }); + tokio::time::sleep(Duration::from_millis(50)).await; + + let client = StdArc::new(BackendClient::new(format!( + "http://{addr}/api/v1/streaming-config" + ))); + let cache = Mutex::new(HashMap::new()); + let outcome = post_typed_backend_for_role( + Some(&client), + &cache, + "latency", + AggRole::Quantile, + make_be("latency", "q"), + &[], + ) + .await; + assert_eq!( + outcome, + PushOutcome::BothApplied, + "legacy documents must still report success even though the plan push 404s" + ); + } + /// P2-3: streaming-config succeeds (200) but storage-routing always /// returns a PERMANENT 400. The coupled push surfaces /// `Desynced { streaming_ok: true, routing_ok: false }` rather than a diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 0a6fbb6c..b7d2b284 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -166,6 +166,12 @@ pub struct HttpServer { /// Hot-reloadable `StreamingConfig` source. `None` when hot-reload /// is not wired up by the caller (unit tests, legacy binaries). hot_reload_config: Option, + /// Hot-reloadable `BackendPlan` handle for `GET/POST + /// /api/v1/backend-plan` (see `control_plane/docs/design-backend-plan-wire-format.md`). + /// `None` when not wired up (unit tests, legacy binaries) — the + /// endpoints return `503`. Shared with `ASAPQueryEngine` so its + /// serving-time lookup sees the same installed plan. + hot_reload_backend_plan: Option, /// Per-metric storage-backend routing table consulted by the HTTP /// instant-query handler at request time. When `Some(..)` and the /// query parses, the handler extracts the metric name from the @@ -228,6 +234,8 @@ struct AppState { adapter: Arc, fallback: Option>, hot_reload_config: Option, + /// See [`HttpServer::hot_reload_backend_plan`]. + hot_reload_backend_plan: Option, /// See [`HttpServer::backend_storage_routing`]. backend_storage_routing: Option, /// Backfill registry (sketch DB §10). See `HttpServer::backfill`. @@ -257,6 +265,7 @@ impl HttpServer { query_router, sketch_index, hot_reload_config: None, + hot_reload_backend_plan: None, backend_storage_routing: None, backfill: None, data_retention_ms: None, @@ -305,6 +314,20 @@ impl HttpServer { self } + /// Attach a `HotReloadBackendPlan` handle so the + /// `GET/POST /api/v1/backend-plan` endpoints can install and read + /// the control plane's typed `BackendPlan` push. Additive alongside + /// [`Self::with_hot_reload_config`] — without this handle the + /// endpoints return `503 Service Unavailable`, same contract as the + /// legacy streaming-config handle. + pub fn with_hot_reload_backend_plan( + mut self, + handle: crate::storage_engines::types::HotReloadBackendPlan, + ) -> Self { + self.hot_reload_backend_plan = Some(handle); + self + } + /// Attach a per-metric storage-backend routing table. The table is /// wrapped in a hot-reload handle internally so the /// `POST /api/v1/storage_routing` endpoint (Phase α) can swap it @@ -415,6 +438,7 @@ impl HttpServer { adapter: adapter.clone(), fallback: self.config.adapter_config.fallback.clone(), hot_reload_config: self.hot_reload_config.clone(), + hot_reload_backend_plan: self.hot_reload_backend_plan.clone(), backend_storage_routing: self.backend_storage_routing.clone(), backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, @@ -452,6 +476,13 @@ impl HttpServer { "/api/v1/streaming-config", get(handle_get_streaming_config).post(handle_post_streaming_config), ) + // BackendPlan wire format (design-backend-plan-wire-format.md): + // sibling of streaming-config above, read by ASAPQueryEngine's + // serving-time lookup. POST body is raw protobuf bytes. + .route( + "/api/v1/backend-plan", + get(handle_get_backend_plan).post(handle_post_backend_plan), + ) // Phase α (MVP): control-plane-pushed `BackendStorageRouting` // table. POST replaces the current table atomically; GET // returns a JSON snapshot for operator diagnostics. @@ -507,6 +538,7 @@ impl HttpServer { adapter: adapter.clone(), fallback: self.config.adapter_config.fallback.clone(), hot_reload_config: self.hot_reload_config.clone(), + hot_reload_backend_plan: self.hot_reload_backend_plan.clone(), backend_storage_routing: self.backend_storage_routing.clone(), backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, @@ -526,6 +558,13 @@ impl HttpServer { "/api/v1/streaming-config", get(handle_get_streaming_config).post(handle_post_streaming_config), ) + // BackendPlan wire format (design-backend-plan-wire-format.md): + // sibling of streaming-config above, read by ASAPQueryEngine's + // serving-time lookup. POST body is raw protobuf bytes. + .route( + "/api/v1/backend-plan", + get(handle_get_backend_plan).post(handle_post_backend_plan), + ) // Phase α (MVP): control-plane-pushed `BackendStorageRouting` // table. POST replaces the current table atomically; GET // returns a JSON snapshot for operator diagnostics. @@ -2209,6 +2248,34 @@ mod tests { .expect("Failed to start test server") } + async fn setup_test_server_with_backend_plan( + hot_reload: Option, + ) -> u16 { + let adapter_config = AdapterConfig::prometheus_promql( + "http://127.0.0.1:9999".to_string(), + false, + ); + let config = HttpServerConfig { + port: 0, + handle_http_requests: true, + adapter_config, + }; + let streaming_config = Arc::new(StreamingConfig::default()); + let query_engine = Arc::new(ASAPQueryEngine::new(streaming_config.clone(), 15000)); + let mut server = HttpServer::new( + config, + query_engine, + Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()), + ); + if let Some(handle) = hot_reload { + server = server.with_hot_reload_backend_plan(handle); + } + server + .start_test_server() + .await + .expect("Failed to start test server") + } + #[tokio::test] async fn test_get_endpoint_plus_symbol_decoding() { // Enable debug logging for this test @@ -2421,6 +2488,105 @@ aggregations: assert_eq!(body["status"], "error"); } + // ── BackendPlan hot-reload (design-backend-plan-wire-format.md) ───── + + /// POST an encoded `BackendPlan` and verify the active state via GET + /// reflects the swap, and that the underlying hot-reload handle + /// (cloned into the server at setup) sees it too — mirroring + /// `test_streaming_config_hot_reload_round_trip`. + #[tokio::test] + async fn test_backend_plan_hot_reload_round_trip() { + use control_plane::backend_plan::BackendPlan; + + let hot_reload = + crate::storage_engines::types::HotReloadBackendPlan::new(BackendPlan::default()); + let server_port = setup_test_server_with_backend_plan(Some(hot_reload.clone())).await; + let client = Client::new(); + + let initial = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan")) + .send() + .await + .expect("GET failed"); + assert!(initial.status().is_success()); + let initial_body: serde_json::Value = initial.json().await.unwrap(); + assert_eq!(initial_body["materialization_count"], 0); + + let new_plan = BackendPlan { + plan_id: 7, + generated_at_unix_ms: 123, + ..Default::default() + }; + let bytes = new_plan.encode_to_vec(); + + let post_resp = client + .post(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan")) + .header("content-type", "application/x-protobuf") + .body(bytes) + .send() + .await + .expect("POST failed"); + let post_status = post_resp.status(); + let post_body: serde_json::Value = post_resp.json().await.unwrap(); + assert!( + post_status.is_success(), + "POST returned {post_status}: {post_body}" + ); + assert_eq!(post_body["status"], "success"); + assert_eq!(post_body["plan_id"], 7); + + let after = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan")) + .send() + .await + .expect("GET after swap failed"); + let after_body: serde_json::Value = after.json().await.unwrap(); + assert_eq!(after_body["plan_id"], 7); + assert_eq!(after_body["generated_at_unix_ms"], 123); + + assert_eq!(hot_reload.snapshot().plan_id, 7); + } + + #[tokio::test] + async fn test_backend_plan_hot_reload_missing_handle_503() { + let server_port = setup_test_server_with_backend_plan(None).await; + let client = Client::new(); + + let get_resp = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan")) + .send() + .await + .unwrap(); + assert_eq!(get_resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE); + + let post_resp = client + .post(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan")) + .body("anything") + .send() + .await + .unwrap(); + assert_eq!(post_resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE); + } + + #[tokio::test] + async fn test_backend_plan_hot_reload_rejects_bad_bytes() { + use control_plane::backend_plan::BackendPlan; + let hot_reload = + crate::storage_engines::types::HotReloadBackendPlan::new(BackendPlan::default()); + let server_port = setup_test_server_with_backend_plan(Some(hot_reload)).await; + let client = Client::new(); + + let resp = client + .post(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan")) + .body(vec![0xFFu8, 0xFF, 0xFF]) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["status"], "error"); + } + /// Set up a test server wired with a hot-reload handle and a /// shared `SketchStore` (the sid catalog the new sid-level /// reconcile reads + writes). Returns `(port, sketch_index)` so @@ -5380,6 +5546,75 @@ async fn handle_post_streaming_config( (StatusCode::OK, axum::Json(body)).into_response() } +// ── BackendPlan hot-reload (design-backend-plan-wire-format.md) ───────── +// +// `GET /api/v1/backend-plan` — return the currently installed plan as +// JSON (debug / verification). +// `POST /api/v1/backend-plan` — accept a protobuf body, decode, and +// atomically swap via ArcSwap. +// +// Sits alongside `/api/v1/streaming-config`, not in place of it — the +// swap here does NOT touch the sid catalog / SketchStore reconciliation; +// that lifecycle management stays on the streaming-config path. + +async fn handle_get_backend_plan(State(state): State) -> axum::response::Response { + use axum::http::StatusCode; + use axum::response::IntoResponse; + + let Some(handle) = state.hot_reload_backend_plan else { + let body = serde_json::json!({ + "status": "error", + "error": "hot-reload backend-plan handle not attached; backend was built without HttpServer::with_hot_reload_backend_plan"}); + return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response(); + }; + let snap = handle.snapshot(); + let body = serde_json::json!({ + "status": "success", + "plan_id": snap.plan_id, + "generated_at_unix_ms": snap.generated_at_unix_ms, + "materialization_count": snap.materializations.len(), + "routing_count": snap.routing.len(), + "monitor_count": snap.monitors.len()}); + (StatusCode::OK, axum::Json(body)).into_response() +} + +async fn handle_post_backend_plan( + State(state): State, + body: axum::body::Bytes, +) -> axum::response::Response { + use axum::http::StatusCode; + use axum::response::IntoResponse; + + let Some(handle) = state.hot_reload_backend_plan else { + let body = serde_json::json!({ + "status": "error", + "error": "hot-reload backend-plan handle not attached; backend was built without HttpServer::with_hot_reload_backend_plan"}); + return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response(); + }; + + let new_plan = match control_plane::backend_plan::BackendPlan::decode(&body) { + Ok(p) => p, + Err(e) => { + let body = serde_json::json!({ + "status": "error", + "error": format!("BackendPlan decode error: {e}")}); + return (StatusCode::BAD_REQUEST, axum::Json(body)).into_response(); + } + }; + + let materialization_count = new_plan.materializations.len(); + let routing_count = new_plan.routing.len(); + let plan_id = new_plan.plan_id; + handle.swap(new_plan); + + let body = serde_json::json!({ + "status": "success", + "plan_id": plan_id, + "materialization_count": materialization_count, + "routing_count": routing_count}); + (StatusCode::OK, axum::Json(body)).into_response() +} + // ── Phase α: BackendStorageRouting hot-reload endpoints ──────────── /// `GET /api/v1/storage_routing` — return a JSON snapshot of the diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 9be3dc2b..375dd277 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -712,8 +712,18 @@ async fn main() -> Result<()> { // `SchemaRegistry`. `POST /api/v1/streaming-config` drives // lifecycle transitions at the sid level via the shared // `SketchStore` (already passed in below). + // BackendPlan wire format (design-backend-plan-wire-format.md): + // install an empty hot-reload handle so `GET/POST + // /api/v1/backend-plan` don't 503 before the control plane's first + // push lands — same "install empty, let the first push fill it in" + // pattern as `bootstrap_routing` below. + let hot_reload_backend_plan = data_plane::storage_engines::types::HotReloadBackendPlan::new( + control_plane::backend_plan::BackendPlan::default(), + ); + let mut server = HttpServer::new(http_config, engine, sketch_index.clone()) .with_hot_reload_config(hot_reload_config.clone()) + .with_hot_reload_backend_plan(hot_reload_backend_plan) .with_probe_cache(probe_cache.clone()); // Per-metric storage-backend routing table (issue #46 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 b81c832d..66b56372 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -81,6 +81,97 @@ use arc_swap::ArcSwap; use crate::storage_engines::types::StreamingConfig; +/// Hot-reloadable `BackendPlan` state — same `ArcSwap` shape as +/// [`HotReloadStreamingConfig`], applied to +/// `control_plane::backend_plan::BackendPlan` (see +/// `control_plane/docs/design-backend-plan-wire-format.md`). Lives +/// alongside [`HotReloadStreamingConfig`], not in place of it: +/// `POST /api/v1/backend-plan` installs the latest plan here for +/// `ASAPQueryEngine`'s serving-time lookup to read, while +/// `POST /api/v1/streaming-config` still drives sid-catalog lifecycle +/// (registration/retirement) on its own path. +#[derive(Clone)] +pub struct HotReloadBackendPlan { + inner: Arc>, +} + +impl HotReloadBackendPlan { + pub fn new(initial: control_plane::backend_plan::BackendPlan) -> Self { + Self { + inner: Arc::new(ArcSwap::new(Arc::new(initial))), + } + } + + pub fn from_arc(initial: Arc) -> Self { + Self { + inner: Arc::new(ArcSwap::new(initial)), + } + } + + pub fn snapshot(&self) -> Arc { + self.inner.load_full() + } + + pub fn swap( + &self, + new: control_plane::backend_plan::BackendPlan, + ) -> Arc { + self.inner.swap(Arc::new(new)) + } +} + +impl Default for HotReloadBackendPlan { + fn default() -> Self { + Self::new(control_plane::backend_plan::BackendPlan::default()) + } +} + +impl std::fmt::Debug for HotReloadBackendPlan { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let snap = self.snapshot(); + f.debug_struct("HotReloadBackendPlan") + .field("plan_id", &snap.plan_id) + .field("materializations", &snap.materializations.len()) + .field("routing", &snap.routing.len()) + .finish() + } +} + +#[cfg(test)] +mod hot_reload_backend_plan_tests { + use super::*; + use control_plane::backend_plan::BackendPlan; + + fn plan(plan_id: u64) -> BackendPlan { + BackendPlan { + plan_id, + ..Default::default() + } + } + + #[test] + fn snapshot_reflects_initial_plan() { + let hr = HotReloadBackendPlan::new(plan(1)); + assert_eq!(hr.snapshot().plan_id, 1); + } + + #[test] + fn swap_replaces_plan_atomically() { + let hr = HotReloadBackendPlan::new(plan(1)); + let old = hr.swap(plan(2)); + assert_eq!(old.plan_id, 1, "swap returns the pre-swap snapshot"); + assert_eq!(hr.snapshot().plan_id, 2); + } + + #[test] + fn clones_share_underlying_swap() { + let hr = HotReloadBackendPlan::new(plan(1)); + let hr_clone = hr.clone(); + hr.swap(plan(2)); + assert_eq!(hr_clone.snapshot().plan_id, 2); + } +} + /// Thin wrapper around `ArcSwap` with ergonomic /// snapshot + swap helpers. Cloneable; clones share the same /// underlying `ArcSwap` so all holders see the same swaps.