From bc51ad3af1e00246ddb79e65f2e965bf29ac901d Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 18 May 2026 10:05:21 -0600 Subject: [PATCH] fix(control_plane): retry transient errors on startup POST to backend (404/connect-refused) In the multinode harness (asap arm from ASAPCollector PR #394), the controller and backend start concurrently. The backend logs `HTTP server listening on port 9091` before its `/api/v1/streaming-config` route is actually registered, and the controller's PR #290 startup `Replanner::replan_all()` tick fires before the backend's HTTP server has finished accepting connections. Observed symptom: all seven startup `POST /api/v1/streaming-config` calls returned 404; convergence relied entirely on the `OpampServer::on_connect` re-fire when the first agent connected. This patch adds retry-with-backoff for transient failures in `emit::backend_push::post_typed_backend_for_role` (the single helper PR #290 unified both call sites through). Transient: 404, 5xx, connection refused / DNS / connect timeout, reqwest `IsTimeout` / `IsConnect`. Permanent (no retry): 4xx other than 404. Policy: 5 attempts, exponential 100ms -> 300ms -> 900ms -> 2.7s (capped) with full jitter, total span ~5-8s. Lock-discipline unchanged: the `backend_routing_cache` lock is dropped before any HTTP call, so retries never stall sibling replan cycles. `BackendClient`'s existing public methods are untouched (new typed variants `_typed` returning `BackendPostError` sit alongside); `OpampServer::on_connect` remains as the secondary convergence mechanism. Unit tests: 4 new tests in `emit::backend_push::tests` cover recovery-after-404, no-retry-on-permanent, exhaustion-reports- attempts, no-retry-on-first-success; 4 new tests in `backend_client::tests` cover the typed classifier (404/500 transient, 400 permanent, connection-refused transient). All 748 `cargo test -p control_plane --lib` pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/backend_client.rs | 221 +++++++++++++++++++++ control_plane/src/emit/backend_push.rs | 255 ++++++++++++++++++++++++- 2 files changed, 471 insertions(+), 5 deletions(-) diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index a71bb530..aca5484c 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -19,6 +19,77 @@ use anyhow::{Context, Result}; use reqwest::Client; use tracing::{debug, warn}; +/// Classification of an HTTP push failure used by the retry layer in +/// [`crate::emit::backend_push`]. Transient errors are safe to retry +/// (the controller raced ahead of the backend's route bind, the host +/// hasn't finished accepting TCP yet, a transient 5xx during backend +/// startup, etc.); permanent errors indicate the call itself is wrong +/// (bad payload, 4xx other than 404) and retrying just amplifies the +/// log noise without improving the outcome. +/// +/// This type sits next to [`BackendClient`] because the classification +/// is a property of how the backend responded, not of how the caller +/// retries — keeping the classifier here lets every endpoint method +/// share the same transient/permanent definition. +#[derive(Debug)] +pub enum BackendPostError { + /// Worth another attempt after backoff: connection refused, + /// connect timeout, DNS resolution failure, 404 (route not yet + /// registered), or any 5xx. + Transient(anyhow::Error), + /// Will fail again the same way: 4xx other than 404 (bad payload, + /// auth, etc.). The retry layer surfaces these immediately. + Permanent(anyhow::Error), +} + +impl BackendPostError { + pub fn is_transient(&self) -> bool { + matches!(self, BackendPostError::Transient(_)) + } + + pub fn into_inner(self) -> anyhow::Error { + match self { + BackendPostError::Transient(e) | BackendPostError::Permanent(e) => e, + } + } +} + +impl std::fmt::Display for BackendPostError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BackendPostError::Transient(e) => write!(f, "transient: {e}"), + BackendPostError::Permanent(e) => write!(f, "permanent: {e}"), + } + } +} + +impl std::error::Error for BackendPostError {} + +/// Inspect a `reqwest::Error` and decide whether the failure is worth +/// retrying. Connection-level failures (no TCP socket / DNS / connect +/// timeout) and request-side timeouts are transient — the backend is +/// likely still coming up. +fn classify_reqwest_error(err: reqwest::Error) -> BackendPostError { + if err.is_connect() || err.is_timeout() || err.is_request() { + BackendPostError::Transient(anyhow::Error::new(err)) + } else { + // body decode, redirect loops, etc. — these will not resolve + // themselves on retry. + BackendPostError::Permanent(anyhow::Error::new(err)) + } +} + +/// Map a non-2xx HTTP status to a [`BackendPostError`]. 404 and 5xx +/// are transient; every other 4xx is permanent. +fn classify_http_status(status: reqwest::StatusCode, body: String, what: &str) -> BackendPostError { + let err = anyhow::anyhow!("backend returned {} for {}: {}", status, what, body); + if status == reqwest::StatusCode::NOT_FOUND || status.is_server_error() { + BackendPostError::Transient(err) + } else { + BackendPostError::Permanent(err) + } +} + /// Minimal HTTP client for ASAPQuery-backend's streaming-config endpoint. /// Built once at control-plane startup from the /// `CONTROL_PLANE_BACKEND_ENDPOINT` environment variable and shared @@ -132,6 +203,79 @@ impl BackendClient { } } + /// Typed sibling of [`Self::post_streaming_config_json`] for the + /// retry layer. Returns the same `Ok(())` on 2xx, but on failure + /// classifies the underlying cause as [`BackendPostError::Transient`] + /// or [`BackendPostError::Permanent`] so the caller can decide + /// whether another attempt is worthwhile. + /// + /// Public-interface preservation: the existing + /// [`Self::post_streaming_config_json`] remains untouched — callers + /// that don't need retry semantics keep their `anyhow::Result` + /// shape. The retry layer in `emit::backend_push` uses this typed + /// variant. + pub async fn post_streaming_config_json_typed( + &self, + json: String, + ) -> std::result::Result<(), BackendPostError> { + debug!( + endpoint = %self.endpoint, + json_bytes = json.len(), + "posting streaming-config JSON to ASAPQuery-backend (typed)" + ); + let resp = self + .http + .post(&self.endpoint) + .header("content-type", "application/json") + .body(json) + .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, + "streaming-config JSON POST", + )) + } + } + + /// Typed sibling of [`Self::post_storage_routing_json`] for the + /// retry layer. Identical contract to + /// [`Self::post_streaming_config_json_typed`]. + pub async fn post_storage_routing_json_typed( + &self, + json: String, + ) -> std::result::Result<(), BackendPostError> { + let url = derive_storage_routing_url(&self.endpoint); + debug!( + endpoint = %url, + json_bytes = json.len(), + "posting storage-routing JSON to ASAPQuery-backend (typed)" + ); + let resp = self + .http + .post(&url) + .header("content-type", "application/json") + .body(json) + .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, "storage-routing JSON POST")) + } + } + /// Phase α (MVP) sibling of [`Self::post_streaming_config_json`]: /// POSTs the control-plane-emitted `BackendStorageRouting` JSON /// document to the backend's `POST /api/v1/storage_routing` @@ -397,6 +541,83 @@ mod tests { assert_eq!(received[0], json); } + /// Typed-variant classification: 404 from the backend is + /// transient (route not yet registered during startup race). + #[tokio::test] + async fn typed_streaming_config_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_streaming_config_json_typed("{}".to_string()) + .await + .expect_err("404 should surface as Err"); + assert!(err.is_transient(), "404 must classify as transient: {err}"); + } + + /// Typed-variant classification: 500 is transient. + #[tokio::test] + async fn typed_streaming_config_500_is_transient() { + let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); + let url = + start_mock_backend(sink.clone(), axum::http::StatusCode::INTERNAL_SERVER_ERROR).await; + + let client = BackendClient::new(url); + let err = client + .post_streaming_config_json_typed("{}".to_string()) + .await + .expect_err("500 should surface as Err"); + assert!(err.is_transient(), "500 must classify as transient: {err}"); + } + + /// Typed-variant classification: 400 (other 4xx) is permanent — + /// retrying won't fix a malformed payload. + #[tokio::test] + async fn typed_streaming_config_400_is_permanent() { + let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); + let url = start_mock_backend(sink.clone(), axum::http::StatusCode::BAD_REQUEST).await; + + let client = BackendClient::new(url); + let err = client + .post_streaming_config_json_typed("{}".to_string()) + .await + .expect_err("400 should surface as Err"); + assert!( + !err.is_transient(), + "400 must classify as permanent: {err}" + ); + } + + /// Typed-variant classification: connection refused (no listener + /// on the target port) is transient — the backend may still be + /// binding. + #[tokio::test] + async fn typed_connection_refused_is_transient() { + // Port 1 on loopback is reserved and refuses connections. + let client = BackendClient::new("http://127.0.0.1:1/api/v1/streaming-config"); + let err = client + .post_streaming_config_json_typed("{}".to_string()) + .await + .expect_err("connection refused should surface as Err"); + assert!( + err.is_transient(), + "connection-refused must classify as transient: {err}" + ); + } + + /// Typed-variant happy path: 2xx returns Ok. + #[tokio::test] + async fn typed_streaming_config_success_is_ok() { + let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); + let url = start_mock_backend(sink.clone(), axum::http::StatusCode::OK).await; + let client = BackendClient::new(url); + client + .post_streaming_config_json_typed("{}".to_string()) + .await + .expect("2xx must be Ok"); + } + #[tokio::test] async fn storage_routing_post_non_2xx_is_error() { let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index e70387bf..340cb84f 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -39,16 +39,120 @@ //! plan. use std::collections::{BTreeMap, HashMap}; +use std::future::Future; use std::sync::Arc; +use std::time::{Duration, Instant}; use tokio::sync::Mutex; use tracing::{info, warn}; -use crate::backend_client::BackendClient; +use crate::backend_client::{BackendClient, BackendPostError}; use crate::emit::{emit_backend_storage_routing, emit_backend_streaming_config_json}; use crate::physical::colored_dag::emitter::BackendStageConfig; use crate::workload::AggRole; +/// Retry policy for transient POST failures. Tuned to bridge the +/// startup race window in the multinode harness (asap arm from +/// ASAPCollector PR #394), where the controller may issue its first +/// `POST /api/v1/streaming-config` before the backend's HTTP server +/// has finished registering its routes: +/// +/// * backend log: `HTTP server listening on port 9091` at T+0 +/// * route bind for `/api/v1/streaming-config` lands T+~hundreds-of-ms later +/// * controller's startup `Replanner::replan_all()` POSTs at T+~few-seconds +/// +/// Five attempts spanning ~5-8 s cover both the route-bind delay and +/// any TCP-accept race when the backend's compose container is still +/// initialising. Each delay is exponential (3x) with full jitter to +/// avoid synchronised retries from a fleet of controllers. +const RETRY_MAX_ATTEMPTS: u32 = 5; +const RETRY_BASE_DELAY: Duration = Duration::from_millis(100); +const RETRY_DELAY_CAP: Duration = Duration::from_millis(2700); + +/// 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 +/// already needed for the backoff itself. Returns a value in `0..=cap_ms`. +fn jitter_ms(start: Instant, cap_ms: u64) -> u64 { + if cap_ms == 0 { + return 0; + } + // Nanos since program start, folded into the jitter range. Good + // enough to break up synchronous retry storms; not a CSPRNG. + let nanos = start.elapsed().as_nanos() as u64; + nanos % (cap_ms + 1) +} + +/// Compute the delay before attempt `n` (1-indexed). Returns a value +/// `<= RETRY_DELAY_CAP` so the total span is bounded. +fn backoff_delay(attempt: u32, start: Instant) -> Duration { + // Exponential base: 100ms, 300ms, 900ms, 2.7s, 2.7s (capped). + let exp = 3u64.saturating_pow(attempt.saturating_sub(1)); + let base_ms = RETRY_BASE_DELAY + .as_millis() + .saturating_mul(exp as u128) as u64; + let base_ms = base_ms.min(RETRY_DELAY_CAP.as_millis() as u64); + // Full jitter: pick a value in [0, base_ms]. + let with_jitter = jitter_ms(start, base_ms); + Duration::from_millis(with_jitter) +} + +/// Retry the given POST closure on [`BackendPostError::Transient`] +/// outcomes with exponential backoff + jitter, capped at +/// `RETRY_MAX_ATTEMPTS` attempts. Permanent failures short-circuit on +/// the first attempt. Returns the final attempt count and outcome — +/// the caller is expected to log appropriately and never propagate +/// (preserve the outer fire-and-forget contract). +/// +/// Type parameters allow the closure to capture per-attempt context +/// (clones of the JSON body, the endpoint label) without forcing the +/// caller to box the future. +async fn retry_transient( + label: &str, + mut op: F, +) -> (u32, std::result::Result<(), BackendPostError>) +where + F: FnMut() -> Fut, + Fut: Future>, +{ + let start = Instant::now(); + let mut last_err: Option = None; + + for attempt in 1..=RETRY_MAX_ATTEMPTS { + match op().await { + Ok(()) => return (attempt, Ok(())), + Err(BackendPostError::Permanent(e)) => { + // 4xx other than 404 — won't get better with retry. + return (attempt, Err(BackendPostError::Permanent(e))); + } + Err(BackendPostError::Transient(e)) => { + last_err = Some(BackendPostError::Transient(e)); + if attempt < RETRY_MAX_ATTEMPTS { + let delay = backoff_delay(attempt, start); + warn!( + op = %label, + attempt, + max_attempts = RETRY_MAX_ATTEMPTS, + retry_in_ms = delay.as_millis() as u64, + error = %last_err.as_ref().unwrap(), + "transient backend POST failure; will retry after backoff" + ); + tokio::time::sleep(delay).await; + } + } + } + } + + ( + RETRY_MAX_ATTEMPTS, + Err(last_err.unwrap_or_else(|| { + BackendPostError::Transient(anyhow::anyhow!( + "retry loop exhausted without recording a final error" + )) + })), + ) +} + /// Per-`(metric, role)` `BackendStageConfig` cache type alias. The /// cache is owned by the controller's `AppState` and shared with the /// `Replanner` via `Arc>` so both call sites read/write the @@ -126,17 +230,30 @@ pub async fn post_typed_backend_for_role( ); if let Some(client) = backend_client { let body = json_doc.to_string(); - match client.post_streaming_config_json(body).await { + // Retry-with-backoff for transient errors (404 + // route-not-bound, 5xx, connection refused, connect + // timeout) so the controller's startup `replan_all()` + // tick can outwait the backend's HTTP-server bind + + // route-registration window. See PR for the multinode + // race we're patching here. + let (attempts, outcome) = retry_transient("streaming-config", || { + let body = body.clone(); + async move { client.post_streaming_config_json_typed(body).await } + }) + .await; + match outcome { Ok(()) => info!( stage = "backend", endpoint = %client.endpoint(), + attempts, "[USE_TYPED_STAGE_SPLIT] typed backend JSON push succeeded" ), Err(e) => warn!( stage = "backend", endpoint = %client.endpoint(), + attempts, error = %e, - "[USE_TYPED_STAGE_SPLIT] typed backend JSON push failed; \ + "[USE_TYPED_STAGE_SPLIT] typed backend JSON push failed after retries; \ next replan cycle will retry" ), } @@ -191,17 +308,28 @@ pub async fn post_typed_backend_for_role( ); if let Some(client) = backend_client { let body = routing_doc.to_string(); - match client.post_storage_routing_json(body).await { + // Same retry policy as the streaming-config POST + // above — the storage-routing endpoint lives on the + // same backend HTTP server and binds at the same time, + // so it shares the same startup-race window. + let (attempts, outcome) = retry_transient("storage-routing", || { + let body = body.clone(); + async move { client.post_storage_routing_json_typed(body).await } + }) + .await; + match outcome { Ok(()) => info!( stage = "backend", metric = %metric, + attempts, "[USE_TYPED_STAGE_SPLIT] storage-routing JSON push succeeded" ), Err(e) => warn!( stage = "backend", metric = %metric, + attempts, error = %e, - "[USE_TYPED_STAGE_SPLIT] storage-routing JSON push failed; \ + "[USE_TYPED_STAGE_SPLIT] storage-routing JSON push failed after retries; \ next replan cycle will retry" ), } @@ -220,6 +348,123 @@ pub async fn post_typed_backend_for_role( #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::{AtomicU32, Ordering}; + + /// Retry-loop happy path with transient recovery: closure returns + /// `Transient(404)` on the first call and `Ok` on the second. The + /// helper must (a) reach attempt 2, (b) report final outcome Ok. + /// This is the regression test for the controller-startup vs + /// backend-route-bind race the parent PR addresses. + #[tokio::test(start_paused = true)] + async fn retry_transient_recovers_after_first_404() { + let counter = AtomicU32::new(0); + let (attempts, outcome) = retry_transient("test", || { + let n = counter.fetch_add(1, Ordering::SeqCst) + 1; + async move { + if n == 1 { + Err(BackendPostError::Transient(anyhow::anyhow!( + "backend returned 404 for streaming-config JSON POST: " + ))) + } else { + Ok(()) + } + } + }) + .await; + + assert!( + attempts > 1, + "expected retry to happen at least once, got {attempts} attempt(s)" + ); + assert_eq!(attempts, 2, "should succeed on the 2nd attempt"); + assert!(outcome.is_ok(), "expected Ok after recovery, got {outcome:?}"); + } + + /// Permanent failures (4xx other than 404) MUST short-circuit on + /// the first attempt — retrying a bad payload just floods the + /// logs without ever succeeding. + #[tokio::test(start_paused = true)] + async fn retry_transient_does_not_retry_permanent_errors() { + let counter = AtomicU32::new(0); + let (attempts, outcome) = retry_transient("test", || { + counter.fetch_add(1, Ordering::SeqCst); + async move { + Err(BackendPostError::Permanent(anyhow::anyhow!( + "backend returned 400 for streaming-config JSON POST: bad payload" + ))) + } + }) + .await; + + assert_eq!(attempts, 1, "permanent error must not retry: got {attempts} attempts"); + assert!(outcome.is_err(), "permanent error should surface as Err"); + assert!(!outcome.unwrap_err().is_transient(), "outcome must remain permanent"); + assert_eq!(counter.load(Ordering::SeqCst), 1, "closure called exactly once"); + } + + /// Exhausting all retries returns the final Transient error with + /// `attempts == RETRY_MAX_ATTEMPTS` so the caller's WARN log can + /// report how hard we tried. + #[tokio::test(start_paused = true)] + async fn retry_transient_exhausts_and_reports_attempts() { + let counter = AtomicU32::new(0); + let (attempts, outcome) = retry_transient("test", || { + counter.fetch_add(1, Ordering::SeqCst); + async move { + Err(BackendPostError::Transient(anyhow::anyhow!( + "connection refused" + ))) + } + }) + .await; + + assert_eq!( + attempts, RETRY_MAX_ATTEMPTS, + "all attempts should have fired" + ); + assert!(outcome.is_err(), "exhausted retries should surface Err"); + assert!( + outcome.unwrap_err().is_transient(), + "final error must still be transient" + ); + assert_eq!(counter.load(Ordering::SeqCst), RETRY_MAX_ATTEMPTS); + } + + /// Happy path on the first attempt: zero retries, Ok outcome, + /// attempts == 1. This protects the smoke-test invariant that the + /// fire-and-forget happy path is unchanged when the backend is up + /// before the controller's first POST. + #[tokio::test(start_paused = true)] + async fn retry_transient_no_retry_on_first_success() { + let counter = AtomicU32::new(0); + let (attempts, outcome) = retry_transient("test", || { + counter.fetch_add(1, Ordering::SeqCst); + async move { Ok(()) } + }) + .await; + + assert_eq!(attempts, 1, "first-attempt success must not retry"); + assert!(outcome.is_ok()); + assert_eq!(counter.load(Ordering::SeqCst), 1); + } + + /// Backoff schedule sanity-check: delays grow exponentially up to + /// the cap. Doesn't assert exact ms (jitter makes that flaky); just + /// asserts each delay is `<= RETRY_DELAY_CAP` and at least one + /// later attempt has a larger nominal base than the first. + #[test] + fn backoff_delay_respects_cap() { + let start = Instant::now(); + for attempt in 1..=RETRY_MAX_ATTEMPTS { + let d = backoff_delay(attempt, start); + assert!( + d <= RETRY_DELAY_CAP, + "attempt {attempt} delay {d:?} exceeds cap {:?}", + RETRY_DELAY_CAP + ); + } + } + fn make_be(metric: &str, agg_id: &str) -> BackendStageConfig { use crate::physical::colored_dag::emitter::{