Skip to content

feat(control-plane): require complete plan publication #466

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zzylol merged 2 commits into from
Sep 3, 2026
Merged

feat(control-plane): require complete plan publication #466

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aea6aee
feat(control-plane): require complete plan publication
zzylol Sep 2, 2026
dfae514
test(control-plane): mock complete plan publication
zzylol Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 41 additions & 46 deletions control_plane/src/emit/backend_push.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,34 +184,29 @@ where
/// same cumulative state.
pub type BackendRoutingCache = Mutex<HashMap<(String, AggRole), BackendStageConfig>>;

/// Combined outcome of one cumulative push cycle (P2-3).
/// Combined outcome of one cumulative publication cycle.
///
/// The streaming-config and storage-routing documents are TWO independent
/// HTTP POSTs to the backend. Before P2-3 they were retried separately and
/// the function returned `()`, so a cycle where one POST succeeded and the
/// other exhausted its retries left the backend running a streaming-config
/// that disagreed with its storage-routing table until the next replan
/// re-pushed both. This enum surfaces the coupled result so callers (and
/// tests) can observe a partial-failure desync rather than silently
/// proceeding.
/// Streaming config, storage routing and BackendPlan are independent HTTP
/// POSTs. The outcome records all three so callers never treat a generation
/// with a missing authoritative plan as successfully published. It surfaces
/// partial failure so the next replan can republish the complete generation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PushOutcome {
/// No backend client configured — nothing was POSTed. Cache was still
/// updated.
Skipped,
/// A document failed to even serialise; nothing was POSTed.
EmitFailed,
/// Both documents were accepted by the backend (each on a 2xx). The
/// backend's streaming-config and storage-routing are consistent.
BothApplied,
/// At least one document failed to land after retries. The two
/// documents may now disagree on the backend; the next replan cycle
/// re-POSTs both cumulatively (idempotent swap) to restore
/// All three documents were accepted by the backend.
AllApplied,
/// At least one document failed to land. The documents may now disagree
/// on the backend; the next replan cycle re-POSTs them to restore
/// consistency. The carried flags say which succeeded so logs / tests
/// can tell which side is stale.
Desynced {
streaming_ok: bool,
routing_ok: bool,
plan_ok: bool,
},
}

Expand Down Expand Up @@ -294,26 +289,27 @@ async fn push_documents_coupled(
(streaming_ok, routing_ok, RETRY_MAX_ATTEMPTS)
}

/// Best-effort, single-attempt push of the encoded `BackendPlan` — no
/// Required 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>) {
/// WARN on failure and return it to the coupled publication outcome.
async fn push_backend_plan_required(client: &Arc<BackendClient>, bytes: Vec<u8>) -> bool {
match client.post_backend_plan_typed(bytes).await {
Ok(()) => {
debug!(stage = "backend", endpoint = %client.endpoint(), "BackendPlan push succeeded");
true
}
Err(e) => {
warn!(
stage = "backend",
endpoint = %client.endpoint(),
error = %e,
"BackendPlan push failed; next replan cycle will retry"
"BackendPlan push failed; publication generation is incomplete"
);
false
}
}
}
Expand Down Expand Up @@ -475,21 +471,19 @@ 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.
// three describe one consistent generation of planning state. Until
// BackendPlan fully replaces the compatibility documents, publication
// succeeds only when all three are accepted.
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()),
Ok(plan) => plan.encode_to_vec(),
Err(e) => {
warn!(error = %e, "backend_plan::from_stage_config failed; skipping BackendPlan push (legacy push unaffected)");
None
warn!(error = %e, "backend_plan::from_stage_config failed; refusing partial publication");
return PushOutcome::EmitFailed;
}
};

Expand Down Expand Up @@ -553,35 +547,32 @@ 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;
}
let plan_ok = push_backend_plan_required(client, plan_bytes).await;

if streaming_ok && routing_ok {
if streaming_ok && routing_ok && plan_ok {
info!(
stage = "backend",
endpoint = %client.endpoint(),
attempts,
"[USE_TYPED_STAGE_SPLIT] coupled backend JSON push succeeded (both documents applied)"
);
PushOutcome::BothApplied
PushOutcome::AllApplied
} else {
warn!(
stage = "backend",
endpoint = %client.endpoint(),
attempts,
streaming_ok,
routing_ok,
plan_ok,
"[USE_TYPED_STAGE_SPLIT] coupled backend JSON push DESYNCED after retries \
(one document landed, the other did not); next replan cycle re-POSTs both \
cumulatively to restore consistency"
);
PushOutcome::Desynced {
streaming_ok,
routing_ok,
plan_ok,
}
}
}
Expand Down Expand Up @@ -907,7 +898,7 @@ mod tests {
&[],
)
.await;
assert_eq!(outcome, PushOutcome::BothApplied);
assert_eq!(outcome, PushOutcome::AllApplied);
assert_eq!(mock.streaming_hits.load(StdOrdering::SeqCst), 1);
assert_eq!(mock.routing_hits.load(StdOrdering::SeqCst), 1);
}
Expand All @@ -929,15 +920,14 @@ mod tests {
&[],
)
.await;
assert_eq!(outcome, PushOutcome::BothApplied);
assert_eq!(outcome, PushOutcome::AllApplied);
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.
/// A BackendPlan push failure makes the publication generation
/// explicitly incomplete even if both compatibility documents landed.
#[tokio::test]
async fn backend_plan_push_failure_does_not_affect_push_outcome() {
async fn backend_plan_push_failure_is_reported_as_desync() {
// A mock that only serves the legacy endpoints (no
// `/api/v1/backend-plan` route) — the plan push 404s.
let app = Router::new()
Expand Down Expand Up @@ -971,8 +961,12 @@ mod tests {
.await;
assert_eq!(
outcome,
PushOutcome::BothApplied,
"legacy documents must still report success even though the plan push 404s"
PushOutcome::Desynced {
streaming_ok: true,
routing_ok: true,
plan_ok: false,
},
"publication must not report success when the authoritative plan is missing"
);
}

Expand Down Expand Up @@ -1004,7 +998,8 @@ mod tests {
outcome,
PushOutcome::Desynced {
streaming_ok: true,
routing_ok: false
routing_ok: false,
plan_ok: true,
},
"one-sided failure must surface as Desynced, not silent success"
);
Expand Down Expand Up @@ -1048,7 +1043,7 @@ mod tests {

// The periodic re-POST reads the SAME cache and re-pushes everything.
let outcome = repost_cumulative_backend_config(Some(&client2), &cache, &[]).await;
assert_eq!(outcome, PushOutcome::BothApplied);
assert_eq!(outcome, PushOutcome::AllApplied);
assert_eq!(
mock2.streaming_hits.load(StdOrdering::SeqCst),
1,
Expand Down
18 changes: 12 additions & 6 deletions control_plane/src/replan.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -824,7 +824,7 @@ impl Replanner {
ticker.tick().await;
let outcome = self.repost_cumulative_backend_config().await;
match outcome {
PushOutcome::BothApplied => info!(
PushOutcome::AllApplied => info!(
"periodic backend re-POST applied cumulative streaming-config + storage-routing"
),
PushOutcome::Skipped => { /* no backend / empty cache — nothing logged each tick */
Expand All @@ -835,9 +835,11 @@ impl Replanner {
PushOutcome::Desynced {
streaming_ok,
routing_ok,
plan_ok,
} => warn!(
streaming_ok,
routing_ok,
plan_ok,
"periodic backend re-POST desynced after retries; will retry next tick"
),
}
Expand Down Expand Up @@ -1209,9 +1211,9 @@ mod tests {
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc as StdArc;

/// Start a mock backend serving both the streaming-config and
/// storage-routing endpoints, returning the streaming-config URL and a
/// shared hit-counter for the streaming endpoint.
/// Start a mock backend serving the complete publication contract,
/// returning the streaming-config URL and a shared hit-counter for that
/// endpoint.
async fn start_repost_mock() -> (String, StdArc<AtomicU32>) {
use axum::extract::State;
use axum::routing::post;
Expand All @@ -1231,6 +1233,10 @@ mod tests {
"/api/v1/storage_routing",
post(|_b: axum::body::Bytes| async move { axum::http::StatusCode::OK }),
)
.route(
"/api/v1/backend-plan",
post(|_b: axum::body::Bytes| async move { axum::http::StatusCode::OK }),
)
.with_state(StdArc::clone(&hits));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
Expand Down Expand Up @@ -1321,7 +1327,7 @@ mod tests {
// the full cumulative config WITHOUT any replan. The (now-restarted)
// backend receives the streaming-config again.
let outcome = r.repost_cumulative_backend_config().await;
assert_eq!(outcome, PushOutcome::BothApplied);
assert_eq!(outcome, PushOutcome::AllApplied);
assert_eq!(
hits.load(Ordering::SeqCst),
1,
Expand All @@ -1331,7 +1337,7 @@ mod tests {
// Idempotent: a second tick re-POSTs again (the data plane no-ops on
// a matching config; the controller still re-sends each cycle).
let outcome2 = r.repost_cumulative_backend_config().await;
assert_eq!(outcome2, PushOutcome::BothApplied);
assert_eq!(outcome2, PushOutcome::AllApplied);
assert_eq!(hits.load(Ordering::SeqCst), 2);
}

Expand Down
5 changes: 4 additions & 1 deletion data_plane/src/query_engines/asap_query_engine/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,10 @@ impl ASAPQueryEngine {
/// handle is wired up. `None` otherwise — callers fall back to the
/// `SketchStore`-reconstruction path.
fn backend_plan_snapshot(&self) -> Option<Arc<control_plane::backend_plan::BackendPlan>> {
self.hot_reload_backend_plan.as_ref().map(|h| h.snapshot())
self.hot_reload_backend_plan
.as_ref()
.map(|h| h.snapshot())
.filter(|plan| plan.plan_id != 0)
}

/// Phase-5 hybrid-stitch builder — attach an archive engine the
Expand Down
Loading