From 9d36b4c2b76f980e68307c35e2d190821a0f3a99 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 13:18:55 -0600 Subject: [PATCH 1/4] feat(o11y): replay real inputs through cost-selected production plans --- .../examples/compile_workload_artifact.rs | 40 +++ .../examples/offline_planner_replay.rs | 4 +- .../src/physical/colored_dag/emitter.rs | 3 +- control_plane/src/physical/compiler.rs | 4 - .../src/physical/post_asap/cost_model.rs | 5 +- control_plane/src/query_plan.rs | 5 +- control_plane/tests/offline_evidence.rs | 6 +- data_plane/src/drivers/query/fallback/mod.rs | 25 +- data_plane/tests/o11y_remote_write_wire.rs | 29 ++ tools/o11y-execution/README.md | 64 +++++ tools/o11y-execution/replay.py | 259 ++++++++++++++++++ tools/o11y-execution/test_replay.py | 45 +++ 12 files changed, 474 insertions(+), 15 deletions(-) create mode 100644 control_plane/examples/compile_workload_artifact.rs create mode 100644 data_plane/tests/o11y_remote_write_wire.rs create mode 100644 tools/o11y-execution/README.md create mode 100644 tools/o11y-execution/replay.py create mode 100644 tools/o11y-execution/test_replay.py diff --git a/control_plane/examples/compile_workload_artifact.rs b/control_plane/examples/compile_workload_artifact.rs new file mode 100644 index 00000000..0cd9115d --- /dev/null +++ b/control_plane/examples/compile_workload_artifact.rs @@ -0,0 +1,40 @@ +//! Control-plane entry point: cost-select a workload and emit its atomic install request. +use control_plane::physical::compiler::BackendLocalPlanningSnapshot; +use serde_json::json; + +fn main() -> Result<(), Box> { + let path = std::env::args() + .nth(1) + .ok_or("usage: compile_workload_artifact SNAPSHOT.json")?; + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?; + if snapshot.snapshot_version != 2 { + return Err( + "execution evaluation requires version 2 complete workload cost evidence".into(), + ); + } + let start = std::time::Instant::now(); + let plan = snapshot.compile()?; + let elapsed = start.elapsed().as_nanos(); + let comparison = plan + .cost_comparison + .ok_or("missing complete-plan comparison")?; + println!( + "{}", + serde_json::to_string_pretty(&json!({ + "schema_version": 1, + "planning_elapsed_ns": elapsed, + "envelope": plan.envelope, + "cost_comparison": comparison, + "lifecycle_estimates": plan.lifecycle_estimates, + "install_request": { + "precompute_plan": plan.precompute_plan, + "transmission_plan": plan.transmission_plan, + "backend_plan": plan.backend_plan.encode_to_vec(), + "query_plan": plan.query_plan, + "storage_routing": null, + "adaptation_evidence": [] + } + }))? + ); + Ok(()) +} diff --git a/control_plane/examples/offline_planner_replay.rs b/control_plane/examples/offline_planner_replay.rs index a28a598a..d7941ad3 100644 --- a/control_plane/examples/offline_planner_replay.rs +++ b/control_plane/examples/offline_planner_replay.rs @@ -38,8 +38,8 @@ fn inspect( Some(json!({ "record_id": row.id, "provenance": row.provenance, - "update_cpu_ns": row.metrics.update_cpu_ns, - "retained_bytes": row.metrics.retained_bytes, + "update_cpu_ns": row.metrics.resources.cpu.update_cpu_ns, + "retained_bytes": row.metrics.resources.retained_memory_bytes, "offline_error_observation": row.error, "error_applies_to_current_query": false, })), diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index d64ef0b6..787d43b5 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -116,8 +116,7 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> { SummaryExpr::BinaryOp { .. } | SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } - | SummaryExpr::SummaryDelete { .. } - | SummaryExpr::BinaryOp { .. } => NodeKind::Other, + | SummaryExpr::SummaryDelete { .. } => NodeKind::Other, }, PhysicalExpr::Committed(PostAsapPlan::LetBinding { name, .. }) => { NodeKind::LetBinding { name } diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 9a0557db..cbb2b40d 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -2111,10 +2111,6 @@ fn summary_agg_metric(node: &SummaryNode) -> Option { walk(right, metrics); } SummaryExpr::SummaryDelete { summary_input, .. } => walk(summary_input, metrics), - SummaryExpr::BinaryOp { lhs, rhs, .. } => { - walk(lhs, metrics); - walk(rhs, metrics); - } } } let mut metrics = BTreeSet::new(); diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index a9569a69..9cdf0624 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -200,7 +200,10 @@ impl ControlPlaneCostModel { .map(|algorithm| { let params = self.size_params(algorithm.clone(), intent, eps, delta); let row = provider.lookup(algorithm, ¶ms).ok()?; - Some((algorithm.clone(), row.metrics.update_cpu_ns.as_ref()?.value)) + Some(( + algorithm.clone(), + row.metrics.resources.cpu.update_cpu_ns.as_ref()?.value, + )) }) .collect(); let Some(mut costs) = costs else { diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 4d46306f..29329aaf 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -392,7 +392,10 @@ where } if !exact_value_executable(node) => QueryPlanNode::ExactFallback { reason: "only temporal observation counts are supported".into(), }, - SummaryExpr::BinaryOp { .. } | SummaryExpr::KeepPreAsap(_) => { + SummaryExpr::BinaryOp { .. } => QueryPlanNode::ExactFallback { + reason: "summary binary operation is not executable by the warm tier".into(), + }, + SummaryExpr::KeepPreAsap(_) => { QueryPlanNode::ExactFallback { reason: "post-ASAP node requires exact execution".into(), } diff --git a/control_plane/tests/offline_evidence.rs b/control_plane/tests/offline_evidence.rs index 96b5a82d..1aa0aca0 100644 --- a/control_plane/tests/offline_evidence.rs +++ b/control_plane/tests/offline_evidence.rs @@ -38,9 +38,9 @@ fn frequency_comparison() -> ( artifact.records[1].error.as_mut().unwrap().mean = Some(0.001); for row in &mut artifact.records { row.error.as_mut().unwrap().query = query.clone(); - row.metrics.build_cpu_ns = + row.metrics.resources.cpu.build_cpu_ns = serde_json::from_value(json!({"value":1.0,"samples":3,"stddev":0.0})).unwrap(); - row.metrics.read_cpu_ns = row.metrics.build_cpu_ns.clone(); + row.metrics.resources.cpu.read_cpu_ns = row.metrics.resources.cpu.build_cpu_ns.clone(); } let bindings: Vec<_> = artifact .records @@ -174,7 +174,7 @@ fn offline_frequency_filters_backend_layout_before_comparison() { width: 1500, depth: 5, }; - unsupported.metrics.update_cpu_ns.as_mut().unwrap().value = 0.001; + unsupported.metrics.resources.cpu.update_cpu_ns.as_mut().unwrap().value = 0.001; let mut query_binding = evidence.query_bindings[1].clone(); query_binding.record_id = unsupported.id.clone(); evidence.query_bindings.push(query_binding); diff --git a/data_plane/src/drivers/query/fallback/mod.rs b/data_plane/src/drivers/query/fallback/mod.rs index 5a15b454..2f4c50d5 100644 --- a/data_plane/src/drivers/query/fallback/mod.rs +++ b/data_plane/src/drivers/query/fallback/mod.rs @@ -19,7 +19,7 @@ pub enum FallbackResponse { impl IntoResponse for FallbackResponse { fn into_response(self) -> Response { - match self { + let mut response = match self { FallbackResponse::Json(value) => Json(value).into_response(), FallbackResponse::Text(text) => { // Return plain text with appropriate content type @@ -32,7 +32,28 @@ impl IntoResponse for FallbackResponse { ) .into_response() } - } + }; + response.headers_mut().insert( + "x-asap-execution", + axum::http::HeaderValue::from_static("exact_fallback"), + ); + response + } +} + +#[cfg(test)] +mod execution_attribution_tests { + use super::*; + + // An HTTP fallback must be distinguishable from warm execution, even if + // its upstream response carries misleading or absent source annotations. + #[test] + fn forwarded_response_has_backend_owned_execution_marker() { + let response = FallbackResponse::Json(serde_json::json!({ + "status": "success", "infos": ["data_source: asap_query"] + })) + .into_response(); + assert_eq!(response.headers()["x-asap-execution"], "exact_fallback"); } } diff --git a/data_plane/tests/o11y_remote_write_wire.rs b/data_plane/tests/o11y_remote_write_wire.rs new file mode 100644 index 00000000..88a146b5 --- /dev/null +++ b/data_plane/tests/o11y_remote_write_wire.rs @@ -0,0 +1,29 @@ +//! Validate the replay adapter's wire format with the production decoding libraries. +use data_plane::drivers::ingest::prometheus_remote_write::WriteRequest; +use prost::Message; + +// The Python adapter encodes x{job="a"} 2 at 1.234 OpenMetrics seconds. +#[test] +fn replay_wire_preserves_labels_value_and_millisecond_timestamp() { + let hex = + "29a00a270a0d0a085f5f6e616d655f5f1201780a080a036a6f62120161120c09000000000000004010d209"; + let bytes: Vec<_> = (0..hex.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap()) + .collect(); + let protobuf = snap::raw::Decoder::new().decompress_vec(&bytes).unwrap(); + let write = WriteRequest::decode(protobuf.as_slice()).unwrap(); + assert_eq!(write.timeseries.len(), 1); + let series = &write.timeseries[0]; + assert_eq!( + series + .labels + .iter() + .map(|l| (l.name.as_str(), l.value.as_str())) + .collect::>(), + vec![("__name__", "x"), ("job", "a")] + ); + assert_eq!(series.samples.len(), 1); + assert_eq!(series.samples[0].value, 2.0); + assert_eq!(series.samples[0].timestamp, 1234); +} diff --git a/tools/o11y-execution/README.md b/tools/o11y-execution/README.md new file mode 100644 index 00000000..6d9f9226 --- /dev/null +++ b/tools/o11y-execution/README.md @@ -0,0 +1,64 @@ +# Real-workload execution replay + +Developer acceptance tooling, stacked on #524 and the deployment/cost-selection +foundation through #505. This is an executable harness, not a published benefit +result. It invokes the **control-plane compiler**, which calls the pinned Planner, +then boots the production data plane with that compiler's atomic install request. +No family override, candidate index, or benchmark-selected winner is accepted. + +## Inputs and prerequisites + +- Original timestamped o11ybench OpenMetrics data. The strict numeric subset + preserves names, labels and finite values; seconds become Remote Write + milliseconds without rounding. Unsupported lines fail before any writes. +- A JSON corpus with `upstream_revision` and `queries`, preserving every upstream + occurrence: `{id, query, eval_timestamp_ms}` (additional provenance is retained). + Export the pinned upstream task queries, not the historical 27-query fixture. + Provide generator revision, command and seed separately; a file hash cannot + establish that provenance. +- A canonical **version 2** planning snapshot registering exactly the corpus's + unique queries and containing valid complete-workload provider cost evidence. + Collect evidence using `workload_cost_manifest`; keep calibration inputs and + evaluation inputs distinct. Do not reuse the demo snapshot's declared costs. + The harness does not invent registrations, cost quotes or unsupported bindings. +- A dedicated, empty Prometheus instance with its Remote Write receiver enabled, + configured for the input's timestamp range. This process is also the fallback + service; do not point the runner at a production/shared instance. + +The harness does not provision Prometheus or calibrate the cost provider. These +are required run inputs, not completed end-to-end acceptance evidence. + +## Run + +```sh +cargo build --locked -p control_plane --example compile_workload_artifact +cargo build --locked -p data_plane --bin data_plane +python3 tools/o11y-execution/replay.py \ + --metrics /path/metrics.txt --queries /path/queries.json \ + --snapshot /path/o11y-costed-snapshot.json \ + --compiler target/debug/examples/compile_workload_artifact \ + --data-plane target/debug/data_plane \ + --exact-url http://127.0.0.1:9090 --output /path/new-run +python3 -m unittest discover -s tools/o11y-execution -v +``` + +Use an unused backend port (`--port`, default 18089). The output directory must +not exist. The runner validates all samples, compiles the supplied workload, +records candidates/selected plan/costs, starts its own backend, sends identical +Remote Write batches to both services, and queries every corpus occurrence at its +original evaluation time. It stops only the backend process it started. Partial +ingest is journaled without automatic retries: restart with fresh services and a +new output directory after investigating a failure. + +`planning.json` contains the envelope, cost comparison, lifecycle estimates and +install request. `installed.json` records runtime status. `queries.json` preserves +every response, occurrence ID, timing and `warm`, `exact_fallback` or `failed` +classification. Forwarded responses carry a backend-owned `x-asap-execution` +header; an unmarked success is not counted as warm or fallback. `ingestion.json` +records accepted batches; acceptance does not prove worker completion. + +The settle interval is recorded, not a completion barrier. The first traversal is +called `first_pass`, not “cold cache”; later traversals are `repeat`. All failures +and fallback responses remain in the denominator. A completion file means the +replay finished, **not** that accuracy or benefit acceptance passed. The next +stacked PR adds matched exact queries and measurement/reporting. diff --git a/tools/o11y-execution/replay.py b/tools/o11y-execution/replay.py new file mode 100644 index 00000000..942ab506 --- /dev/null +++ b/tools/o11y-execution/replay.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +"""Replay supplied o11ybench inputs through the production backend. No winner overrides.""" +import argparse +from decimal import Decimal +import hashlib +import json +import math +from pathlib import Path +import re +import socket +import struct +import subprocess +import time +import urllib.error +import urllib.parse +import urllib.request + + +def classify(response, headers=None): + if response.get("status") != "success": + return "failed" + if (headers or {}).get("x-asap-execution") == "exact_fallback": + return "exact_fallback" + sources = {x for x in response.get("infos", []) if isinstance(x, str) and x.startswith("data_source:")} + if sources == {"data_source: asap_query"}: + return "warm" + if sources == {"data_source: exact_fallback"}: + return "exact_fallback" + return "failed" + + +def validate_workload(snapshot, corpus): + rows = corpus["queries"] + if not corpus.get("upstream_revision") or not rows: + raise ValueError("a versioned, nonempty upstream query corpus is required") + ids = [row["id"] for row in rows] + if len(set(ids)) != len(ids): + raise ValueError("query occurrence IDs must be unique") + for row in rows: + if not isinstance(row["eval_timestamp_ms"], int) or row["eval_timestamp_ms"] < 0: + raise ValueError("each query needs its original nonnegative evaluation timestamp") + workload = snapshot["query_workload"] + if workload.get("query_batch"): + raise ValueError("this runner currently accepts repeating workload registrations only") + registered = {row["query"] for row in workload["repeating_queries"]} + if registered != {row["query"] for row in rows}: + raise ValueError("snapshot registrations must match the complete unique corpus exactly") + return rows + + +_SAMPLE = re.compile(r'([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{(.*)\})?\s+(\S+)\s+(\d+(?:\.\d+)?)') +_LABEL = re.compile(r'([a-zA-Z_][a-zA-Z0-9_]*)="((?:[^"\\]|\\[\\"n])*)"') + + +def parse_samples(lines): + """Strict OpenMetrics subset: seconds converted losslessly to Remote Write milliseconds.""" + rows, seen, latest = [], {}, -1 + for number, line in enumerate(lines, 1): + line = line.strip() + if not line or line.startswith("#"): + continue + match = _SAMPLE.fullmatch(line) + if not match: + raise ValueError(f"unsupported sample at line {number}") + metric, raw_labels, value, timestamp = match.groups() + labels = {"__name__": metric} + rest = raw_labels or "" + while rest: + label = _LABEL.match(rest) + if not label or label[1] in labels: + raise ValueError(f"invalid or duplicate label at line {number}") + labels[label[1]] = re.sub(r'\\([\\"n])', lambda m: '\n' if m[1] == 'n' else m[1], label[2]) + rest = rest[label.end():] + if rest: + if not rest.startswith(",") or len(rest) == 1: + raise ValueError(f"invalid label separator at line {number}") + rest = rest[1:] + millis = Decimal(timestamp) * 1000 + if millis != millis.to_integral_value(): + raise ValueError(f"submillisecond timestamp at line {number}") + value, timestamp = float(value), int(millis) + key = tuple(sorted(labels.items())) + if not math.isfinite(value) or timestamp > 2**63 - 1 or timestamp < latest or timestamp <= seen.get(key, -1): + raise ValueError(f"nonfinite, duplicate, or out-of-order sample at line {number}") + latest, seen[key] = timestamp, timestamp + rows.append((labels, value, timestamp)) + if not rows: + raise ValueError("empty dataset") + return rows + + +def varint(value): + result = bytearray() + while value > 127: + result.append((value & 127) | 128) + value >>= 7 + result.append(value) + return bytes(result) + + +def field(number, payload): + return varint(number * 8 + 2) + varint(len(payload)) + payload + + +def encode_write(rows): + """Remote Write v1 protobuf in an uncompressed-literal raw Snappy block.""" + series = {} + for labels, value, timestamp in rows: + series.setdefault(tuple(sorted(labels.items())), []).append((value, timestamp)) + wire = bytearray() + for labels, samples in series.items(): + ts = b"".join(field(1, field(1, k.encode()) + field(2, v.encode())) for k, v in labels) + ts += b"".join(field(2, b"\x09" + struct.pack(" Date: Tue, 8 Sep 2026 14:11:11 -0600 Subject: [PATCH 2/4] style: fix formatting gate for o11y execution stack --- control_plane/src/query_plan.rs | 8 +++----- control_plane/tests/offline_evidence.rs | 9 ++++++++- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 29329aaf..306b24e4 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -395,11 +395,9 @@ where SummaryExpr::BinaryOp { .. } => QueryPlanNode::ExactFallback { reason: "summary binary operation is not executable by the warm tier".into(), }, - SummaryExpr::KeepPreAsap(_) => { - QueryPlanNode::ExactFallback { - reason: "post-ASAP node requires exact execution".into(), - } - } + SummaryExpr::KeepPreAsap(_) => QueryPlanNode::ExactFallback { + reason: "post-ASAP node requires exact execution".into(), + }, SummaryExpr::SummaryAgg { family, reduction, diff --git a/control_plane/tests/offline_evidence.rs b/control_plane/tests/offline_evidence.rs index 1aa0aca0..beb7037b 100644 --- a/control_plane/tests/offline_evidence.rs +++ b/control_plane/tests/offline_evidence.rs @@ -174,7 +174,14 @@ fn offline_frequency_filters_backend_layout_before_comparison() { width: 1500, depth: 5, }; - unsupported.metrics.resources.cpu.update_cpu_ns.as_mut().unwrap().value = 0.001; + unsupported + .metrics + .resources + .cpu + .update_cpu_ns + .as_mut() + .unwrap() + .value = 0.001; let mut query_binding = evidence.query_bindings[1].clone(); query_binding.record_id = unsupported.id.clone(); evidence.query_bindings.push(query_binding); From 18674b1ef524f39b05d77961d1e18c0e19452e13 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 14:31:27 -0600 Subject: [PATCH 3/4] fix: acknowledge finite replay only after worker output completion --- .../drivers/ingest/prometheus_remote_write.rs | 37 ++++++ data_plane/src/drivers/query/servers/http.rs | 26 ++++ .../src/precompute_engine/output_sink.rs | 35 +++-- .../src/precompute_engine/series_router.rs | 22 +++- data_plane/src/precompute_engine/worker.rs | 120 ++++++++++++++++++ tools/o11y-execution/README.md | 16 +++ tools/o11y-execution/replay.py | 9 +- 7 files changed, 249 insertions(+), 16 deletions(-) diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index b90cd013..1b8832fc 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -100,6 +100,7 @@ struct ReceiverInner { #[derive(Default)] struct DedupState { + input_closed: bool, values: HashMap<(u64, u64, String, i64), DedupValue>, expiry: VecDeque<(Instant, u64, u64, String, i64)>, } @@ -112,6 +113,8 @@ enum DedupValue { #[derive(Debug, thiserror::Error)] pub enum RemoteWriteError { + #[error("finite input has been closed by the drain barrier")] + InputClosed, #[error("compressed request exceeds {0} bytes")] CompressedTooLarge(usize), #[error("decompressed request exceeds {0} bytes")] @@ -171,6 +174,15 @@ impl PrometheusRemoteWriteReceiver { self.inner.stats.clone() } + /// Permanently seal this finite source before queuing worker barriers. + pub async fn drain(&self) -> Result<(), String> { + { + let mut state = self.inner.dedup.lock().map_err(|e| e.to_string())?; + state.input_closed = true; + } + self.inner.ingest.router.drain().await + } + /// Decode, validate, deduplicate, and enqueue one whole v1 request. /// All validation and all queue reservations complete before any message /// becomes visible to a worker. @@ -235,6 +247,9 @@ impl PrometheusRemoteWriteReceiver { .dedup .lock() .expect("remote write dedup poisoned"); + if dedup.input_closed { + return Err(RemoteWriteError::InputClosed); + } dedup.evict_before(now.checked_sub(config.dedup_horizon).unwrap_or(now)); // Validate conflicts both against committed history and inside this @@ -664,6 +679,28 @@ mod tests { }) } + // Closing finite input prevents writes racing behind the completion barrier. + #[tokio::test] + async fn finite_input_drain_seals_receiver_and_propagates_worker_failure() { + let (receiver, mut worker) = configured_receiver(); + receiver.accept(&one_sample(1.0)).unwrap(); + let handle = receiver.clone(); + let drain = tokio::spawn(async move { handle.drain().await }); + assert!(matches!( + worker.recv().await.unwrap(), + WorkerMessage::GroupSamples { .. } + )); + let WorkerMessage::Drain(reply) = worker.recv().await.unwrap() else { + panic!("expected barrier") + }; + assert!(matches!( + receiver.accept(&one_sample(1.0)), + Err(RemoteWriteError::InputClosed) + )); + reply.send(Err("sink write failed".into())).unwrap(); + assert_eq!(drain.await.unwrap().unwrap_err(), "sink write failed"); + } + #[test] fn rejects_writes_without_an_active_physical_plan() { let (sender, _worker) = mpsc::channel(1); diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 717cb081..088cc0dd 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -470,6 +470,7 @@ impl HttpServer { .layer(DefaultBodyLimit::max(request_body_limit)), ) .route("/api/v1/store/metrics", get(handle_store_metrics)) + .route("/api/v1/precompute/drain", post(handle_precompute_drain)) .route( "/api/v1/streaming-config", get(handle_get_streaming_config).post(handle_post_streaming_config), @@ -5577,6 +5578,8 @@ async fn handle_prometheus_remote_write( match receiver.accept(&body) { Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::InputClosed) => + (StatusCode::CONFLICT, "finite input is closed").into_response(), Err( crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::CompressedTooLarge( _, @@ -5636,6 +5639,29 @@ async fn handle_health(State(state): State) -> axum::response::Respons (StatusCode::OK, "ok").into_response() } +/// Explicit end-of-input; this seals Remote Write for the lifetime of the process. +async fn handle_precompute_drain(State(state): State) -> Response { + let Some(receiver) = state.remote_write.as_ref() else { + return (StatusCode::NOT_FOUND, "Remote Write is disabled").into_response(); + }; + match receiver.drain().await { + Ok(()) => ( + StatusCode::OK, + axum::Json(serde_json::json!({ + "status": "success", "input_closed": true, "complete": true + })), + ) + .into_response(), + Err(error) => ( + StatusCode::INTERNAL_SERVER_ERROR, + axum::Json(serde_json::json!({ + "status": "error", "input_closed": true, "complete": false, "error": error + })), + ) + .into_response(), + } +} + /// Return list of metrics currently in the store. async fn handle_store_metrics(State(state): State) -> axum::response::Response { use axum::http::StatusCode; diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index b139f741..0bd2908c 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -100,11 +100,9 @@ impl SketchStoreSink { } } - /// Best-effort write to `SketchStore` for one PrecomputedOutput. - /// Logs and skips on missing agg_config or other transient - /// inconsistencies — a SketchStore miss is recoverable in - /// practice because the control plane will re-emit the agg_config - /// on its next reconcile pass. + /// Write one PrecomputedOutput; false reports a rejected output to the caller. + /// Missing configuration or incompatible state must surface as failure: + /// a finite-input completion barrier cannot acknowledge dropped outputs. /// /// PR-6 follow-up: resolves the source `AggregationConfig` via /// `PolicyRegistry::get(output.policy_fp)`. The legacy @@ -160,8 +158,18 @@ impl OutputSink for SketchStoreSink { return Ok(()); } let _span = debug_span!("sketch_index_insert", batch_size = outputs.len()).entered(); + let mut failed = 0; for (output, accumulator) in &outputs { - self.append_to_index(output, accumulator.as_ref()); + if !self.append_to_index(output, accumulator.as_ref()) { + failed += 1; + } + } + if failed > 0 { + return Err(format!( + "SketchStore rejected {failed} of {} completed outputs", + outputs.len() + ) + .into()); } Ok(()) } @@ -375,9 +383,9 @@ mod tests { } #[test] - fn sketch_index_sink_skips_unknown_agg_id_gracefully() { + fn sketch_index_sink_reports_unknown_policy_as_failure() { // Streaming config does NOT contain agg_id=99 — the sink - // skips it (warn log) rather than panicking. + // reports a recoverable error rather than acknowledging a lost write. let streaming = StreamingConfig::new(HashMap::new()); let hot_reload = HotReloadStreamingConfig::new(streaming.clone()); let sketch_index = Arc::new(SketchStore::new()); @@ -389,12 +397,13 @@ mod tests { let output = PrecomputedOutput::new(1000, 2000, None, asap_types::PolicyFingerprint(99)); let acc: Box = Box::new(SumAccumulator::with_sum(1.0)); - sink.emit_batch(vec![(output, acc)]).expect("emit ok"); + sink.emit_batch(vec![(output, acc)]) + .expect_err("unpersisted output must not be acknowledged"); assert_eq!(sketch_index.instance_count(), 0); } /// CQ-6 — a registry-miss (policy_fp not in the running streaming - /// config) is a silent write-skip; with an `IngestObservability` + /// config) reports a failed write; with an `IngestObservability` /// handle wired in, the `dropped_policy_miss` counter must tick. #[test] fn sink_increments_policy_miss_counter_on_registry_miss() { @@ -412,7 +421,8 @@ mod tests { // policy_fp=42 is absent from the empty registry → registry miss. let output = PrecomputedOutput::new(1000, 2000, None, asap_types::PolicyFingerprint(42)); let acc: Box = Box::new(SumAccumulator::with_sum(1.0)); - sink.emit_batch(vec![(output, acc)]).expect("emit ok"); + sink.emit_batch(vec![(output, acc)]) + .expect_err("unpersisted output must not be acknowledged"); assert_eq!( sketch_index.instance_count(), @@ -430,7 +440,8 @@ mod tests { // miss — it must not bump the counter. let unset = PrecomputedOutput::new(1000, 2000, None, asap_types::PolicyFingerprint::UNSET); let acc2: Box = Box::new(SumAccumulator::with_sum(1.0)); - sink.emit_batch(vec![(unset, acc2)]).expect("emit ok"); + sink.emit_batch(vec![(unset, acc2)]) + .expect_err("unpersisted output must not be acknowledged"); assert_eq!( obs.dropped_policy_miss .load(std::sync::atomic::Ordering::Relaxed), diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs index f66b2073..475620b8 100644 --- a/data_plane/src/precompute_engine/series_router.rs +++ b/data_plane/src/precompute_engine/series_router.rs @@ -85,6 +85,8 @@ pub enum WorkerMessage { }, /// Signal the worker to flush/check idle windows. Flush, + /// Finite-input barrier: acknowledge only after queued input and trailing panes reach the sink. + Drain(tokio::sync::oneshot::Sender>), /// Graceful shutdown. Shutdown, } @@ -126,6 +128,7 @@ impl fmt::Debug for WorkerMessage { .field("accumulator_type", &accumulator.type_name()) .finish(), Self::Flush => f.write_str("Flush"), + Self::Drain(_) => f.write_str("Drain"), Self::Shutdown => f.write_str("Shutdown"), } } @@ -205,7 +208,7 @@ impl SeriesRouter { WorkerMessage::GroupSamples { sid, .. } | WorkerMessage::AccumulatorInput { sid, .. } => self.worker_for_sid(*sid), WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key), - WorkerMessage::Flush | WorkerMessage::Shutdown => 0, + WorkerMessage::Flush | WorkerMessage::Drain(_) | WorkerMessage::Shutdown => 0, }; let permit = self.senders[worker_idx] .clone() @@ -244,6 +247,23 @@ impl SeriesRouter { Ok(()) } + /// Caller must stop new input before invoking this finite-source barrier. + pub async fn drain(&self) -> Result<(), String> { + let mut replies = Vec::new(); + for sender in &self.senders { + let (tx, rx) = tokio::sync::oneshot::channel(); + sender + .send(WorkerMessage::Drain(tx)) + .await + .map_err(|e| e.to_string())?; + replies.push(rx); + } + for reply in replies { + reply.await.map_err(|e| e.to_string())??; + } + Ok(()) + } + /// Determine which worker handles a given sid bucket. /// /// Hashes the sid alone — the legacy `(agg_id, group_key)` tuple folded diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 34ce3c56..d6b53aaa 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -202,6 +202,7 @@ impl Worker { pub async fn run(mut self) { info!("Worker {} started", self.id); + let mut processing_error: Option = None; while let Some(msg) = self.receiver.recv().await { match msg { WorkerMessage::GroupSamples { @@ -223,6 +224,7 @@ impl Worker { .entered(); if let Err(e) = self.process_group_samples(sid, policy_fp, &group_key, samples) { + processing_error = Some(e.to_string()); warn!( "Worker {} error processing sid={} (policy_fp={}, group={}): {}", self.id, sid, policy_fp, group_key, e @@ -246,6 +248,7 @@ impl Worker { ) .entered(); if let Err(e) = self.process_samples_raw(&series_key, samples) { + processing_error = Some(e.to_string()); warn!("Worker {} raw error for {}: {}", self.id, series_key, e); } debug!( @@ -278,6 +281,7 @@ impl Worker { timestamp_ms, accumulator, ) { + processing_error = Some(e.to_string()); warn!( "Worker {} accumulator input error for sid={} (policy_fp={}, group={}): {}", self.id, sid, policy_fp, group_key, e @@ -290,6 +294,7 @@ impl Worker { } WorkerMessage::Flush => { if let Err(e) = self.flush_all() { + processing_error = Some(e.to_string()); warn!("Worker {} flush error: {}", self.id, e); } // Evict orphaned GroupStates whose agg_id has been @@ -298,6 +303,13 @@ impl Worker { // closed their windows); empty ones are freed. self.evict_orphaned_groups(); } + WorkerMessage::Drain(reply) => { + let result = self.force_close_all().map_err(|e| e.to_string()); + if let Err(error) = &result { + processing_error = Some(error.clone()); + } + let _ = reply.send(processing_error.clone().map_or(Ok(()), Err)); + } WorkerMessage::Shutdown => { info!("Worker {} shutting down", self.id); if let Err(e) = self.flush_all() { @@ -3259,6 +3271,114 @@ aggregations: // sketch (`sketch_panes`) paths. // ----------------------------------------------------------------------- + // Acknowledgement proves FIFO input processing and trailing-window publication. + #[tokio::test] + async fn finite_input_drain_publishes_before_acknowledging() { + let config = make_agg_config( + 1, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + 10, + 0, + vec![], + ); + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker( + HashMap::from([(1, config)]), + sink.clone(), + false, + 0, + LateDataPolicy::Drop, + ); + let (tx, rx) = tokio::sync::mpsc::channel(8); + worker.receiver = rx; + let task = tokio::spawn(worker.run()); + tx.send(WorkerMessage::GroupSamples { + sid: 1, + policy_fp: PolicyFingerprint(1), + group_key: "".into(), + samples: group_samples("cpu", vec![(1000, 2.0), (2000, 3.0)]), + ingest_received_at: std::time::Instant::now(), + }) + .await + .unwrap(); + let (ack, completed) = tokio::sync::oneshot::channel(); + tx.send(WorkerMessage::Drain(ack)).await.unwrap(); + completed.await.unwrap().unwrap(); + let outputs = sink.drain(); + assert_eq!(outputs.len(), 1); + assert_eq!( + outputs[0] + .1 + .as_any() + .downcast_ref::() + .unwrap() + .sum, + 5.0 + ); + let (ack, completed) = tokio::sync::oneshot::channel(); + tx.send(WorkerMessage::Drain(ack)).await.unwrap(); + completed.await.unwrap().unwrap(); + assert_eq!(sink.len(), 0); + tx.send(WorkerMessage::Shutdown).await.unwrap(); + task.await.unwrap(); + } + + // A sink failure remains visible on repeated barriers after panes were consumed. + #[tokio::test] + async fn finite_input_drain_does_not_hide_sink_failure_on_retry() { + struct FailedSink; + impl OutputSink for FailedSink { + fn emit_batch( + &self, + _: Vec<(PrecomputedOutput, Box)>, + ) -> Result<(), Box> { + Err("deliberate sink failure".into()) + } + } + let config = make_agg_config( + 1, + "cpu", + AggregationType::SingleSubpopulation, + "Sum", + 10, + 0, + vec![], + ); + let mut worker = make_worker( + HashMap::from([(1, config)]), + Arc::new(CapturingOutputSink::new()), + false, + 0, + LateDataPolicy::Drop, + ); + worker.output_sink = Arc::new(FailedSink); + let (tx, rx) = tokio::sync::mpsc::channel(8); + worker.receiver = rx; + let task = tokio::spawn(worker.run()); + tx.send(WorkerMessage::GroupSamples { + sid: 1, + policy_fp: PolicyFingerprint(1), + group_key: "".into(), + samples: group_samples("cpu", vec![(1000, 2.0)]), + ingest_received_at: std::time::Instant::now(), + }) + .await + .unwrap(); + for _ in 0..2 { + let (ack, completed) = tokio::sync::oneshot::channel(); + tx.send(WorkerMessage::Drain(ack)).await.unwrap(); + assert!(completed + .await + .unwrap() + .unwrap_err() + .contains("deliberate sink failure")); + } + tx.send(WorkerMessage::Shutdown).await.unwrap(); + task.await.unwrap(); + } + #[test] fn shutdown_force_close_emits_trailing_sample_window() { // 10s tumbling window; make_worker uses grace=0, isolating the diff --git a/tools/o11y-execution/README.md b/tools/o11y-execution/README.md index 6d9f9226..deade70f 100644 --- a/tools/o11y-execution/README.md +++ b/tools/o11y-execution/README.md @@ -62,3 +62,19 @@ called `first_pass`, not “cold cache”; later traversals are `repeat`. All fa and fallback responses remain in the denominator. A completion file means the replay finished, **not** that accuracy or benefit acceptance passed. The next stacked PR adds matched exact queries and measurement/reporting. + + +## Finite-input completion + +After all Remote Write batches are accepted, the runner calls +`POST /api/v1/precompute/drain` and saves `drain.json`. The receiver seals input +before worker barriers are queued. Each worker publishes its trailing panes, +then acknowledges completion; prior processing or sink failures remain failures +on repeated drains. Queries begin only after a successful completion response. +Subsequent Remote Write requests return HTTP 409 for that process. Start a new +backend process for another input generation. + +This endpoint is for a finite replay, not a live ingestion watermark. Closing a +trailing pane does not by itself prove its coverage matches every query window; +unsupported or incomplete readouts must still follow exact fallback. This change +adds no local raw-query storage or hybrid operator execution. diff --git a/tools/o11y-execution/replay.py b/tools/o11y-execution/replay.py index 942ab506..a9b58f14 100644 --- a/tools/o11y-execution/replay.py +++ b/tools/o11y-execution/replay.py @@ -188,7 +188,7 @@ def main(): parser.add_argument("--data-plane", type=Path, required=True) parser.add_argument("--exact-url", required=True, help="dedicated empty Prometheus with Remote Write receiver enabled") parser.add_argument("--port", type=int, default=18089) - parser.add_argument("--settle-seconds", type=float, default=2) + parser.add_argument("--settle-seconds", type=float, default=0, help="deprecated; completion uses explicit finite-input drain") parser.add_argument("--repetitions", type=int, default=2) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() @@ -205,7 +205,7 @@ def main(): for p in [args.metrics, args.queries, args.snapshot, args.compiler, args.data_plane]}, "samples": len(samples), "timestamp_min_ms": samples[0][2], "timestamp_max_ms": samples[-1][2], "query_occurrences": len(queries), "configuration": {k: str(v) for k, v in vars(args).items()}, - "limitations": ["generator provenance must be supplied separately", "first pass is not a guaranteed cold cache", "ingest acceptance is not materialization completion"]} + "limitations": ["generator provenance must be supplied separately", "first pass is not a guaranteed cold cache", "finite-input drain closes trailing panes and permanently seals Remote Write; no live-ingestion claim"]} write_json(args.output / "run.json", provenance) with (args.output / "planning.stderr").open("w") as log: compiled = subprocess.run([str(args.compiler.resolve()), str(args.snapshot.resolve())], check=True, @@ -240,7 +240,10 @@ def main(): and p["phase"] == "active" for p in installed["response"].get("plans", [])): raise RuntimeError("runtime has not activated the selected plan generation") ingest(samples, [args.exact_url, backend], args.output) - time.sleep(args.settle_seconds) + drained = request(backend + "/api/v1/precompute/drain", b"") + write_json(args.output / "drain.json", drained) + if drained["http_status"] != 200 or drained["response"].get("complete") is not True: + raise RuntimeError("finite-input materialization drain failed; see drain.json") results = replay(queries, backend, args.output, args.repetitions) write_json(args.output / "store.json", request(backend + "/api/v1/store/metrics")) write_json(args.output / "completion.json", {"complete": True, From 43f9e2111d2984b5d829eb801c9a64143903f4c5 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 14:42:54 -0600 Subject: [PATCH 4/4] fix: keep merged TopK dev dependency on the Planner revision --- data_plane/Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 4528b962..1677d4e5 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -129,7 +129,7 @@ crc32fast = "1.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21"