backend-plan: wire push + data_plane consume endpoint (Phase 2+3)

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<u8>,
+    ) -> 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<Mutex<Vec<Vec<u8>>>> = 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`.

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<BackendClient>, bytes: Vec<u8>) {
+    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<StdAtomicU32>,
         routing_hits: StdArc<StdAtomicU32>,
+        plan_hits: StdArc<StdAtomicU32>,
         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<DualMock>, _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