[zzylol]

Base: main (#819 merged). Next: #821.

Retention and stored outputs still come from the backend's own frontier and "largest offset plus two panes". The next PR (M5, stacked on this one) switches the data plane to InstallablePlan.stored_outputs and its retention_ms.

Why

Remote Write samples were accepted and dropped, so no stored summary was ever built.

What

A precompute engine on the engine thread builds stored panes from Remote Write:

• Frontier: Sds::ingestion_frontier finds the outputs to store from each node's timing: ingestion-time nodes read by a query-time node or as a query's answer. At the pinned ASAPPlanner revision the executor exports frontier_from_timing only under test-support, so the data plane no longer calls it; a test checks the two agree on every example, with and without raw data.
• Selection: samples are buffered per definition and pane. The executor's RawSampleSelector, built from the definition's Scan and TimeRange, turns a pane's samples into the build's input, so the plan's label matchers apply.
• Building: a pane (end − width, end] (left-open, like PromQL ranges and the executor's windows) is built once the watermark passes end. The watermark is the newest sample of a metric some definition reads, capped at the wall clock, minus the allowed lateness, so one series stamped ahead of the others cannot close panes early. The build runs the definition's builder, compiled from the DAG with asap_executor::physical_planner::compile, under ingestion scope, and stores the summary batch. Panes without samples are stored empty.
• Lateness and startup: a sample arriving after its pane was built is dropped and counted. The first pane after startup is skipped because it may be partial; its samples are counted separately.
• Bad timestamps: only samples of a metric some definition reads move the watermark. A sample more than --max-future-ms (default 5 min) ahead of the wall clock is dropped and counted. When the watermark jumps, panes that would already be evicted are skipped and counted, not built. Pane arithmetic is checked, so timestamps near i64::MIN/MAX neither overflow nor loop.
• Stale markers: ASAPPlanner is pinned to main ecea897f, whose RawSampleSelector drops Prometheus stale markers: a range window keeps the samples around one, and an instant selector whose latest sample is one has no row. The data plane passes markers through, and the HTTP-layer filter is removed. The same pin bump brings in ASAPPlanner test: select, install and execute every admitted query-ensemble candidate #775: the executor runs exact per-series range reductions (quantile/sum/avg/min/max/count_over_time), and Stage 3 rejects per-entity aggregate candidates the executor cannot run.
• Eviction: panes older than the largest offset plus two widths are evicted.
• Back-pressure: the engine queue is bounded by queued samples (--max-queued-samples, default 2,000,000). /api/v1/write returns 503 when admitting a request would exceed the bound, so senders retry. A request into an empty queue is always admitted, so a bound smaller than one request cannot refuse it forever. A request without samples is not queued. If the engine thread has stopped, it returns 500 and /-/ready returns 503.
• Counters: /api/v1/status/asap returns JSON with engine_running, samples_queued and the precompute counters. Each samples_* counter counts a sample once: accepted, dropped_late, skipped_first_pane, dropped_future, unrouted. The panes_* counters count built, failed and skipped_expired panes.
• data_plane --allowed-lateness-ms (default 15 s, unsigned). A zero-width pane is refused when the engine is built.
• Store: SummaryStore keeps BTreeMap<DefinitionId, BTreeMap<i64, S>>. get borrows the definition id instead of building a key, and evict_through uses split_off on one definition's panes instead of scanning every pane.
• Forwarding (from main): DataPlane::new gives forwarded queries the default 120 s timeout; with_forward_timeout changes it, and data_plane --forward-timeout-seconds sets it. Header forwarding and the Remote Write body limit are unchanged.

How

The Planner IR holds Rcs, so one engine thread parses the plan once and owns it, the precompute engine and the store. HTTP handlers send it batches over a channel; a shared counter of queued samples bounds the queue, and the engine publishes its counters after each batch.

Known cost: window_batch takes Samples with owned label maps, so each sample's labels are cloned once per pane build. An ASAPPlanner follow-up could take borrowed or interned labels; the series interning later in this stack does not remove this copy.

Key code interfaces added / modified

// data_plane/src/precompute.rs
pub struct RawSample {
    /// The series' full label set, including `__name__`.
    pub labels: Arc<BTreeMap<String, String>>,
    /// Sample time.
    pub timestamp_ms: i64,
    /// Sample value.
    pub value: f64,
}

pub struct PrecomputeStats {
    /// Samples buffered for at least one pane.
    pub samples_accepted: u64,
    /// Samples of a pane already built or skipped, other than the first pane.
    pub samples_dropped_late: u64,
    /// Samples of the first pane after startup, which is never built.
    pub samples_skipped_first_pane: u64,
    /// Samples more than `max_future_ms` ahead of the wall clock.
    pub samples_dropped_future: u64,
    /// Samples of a metric no stored summary reads.
    pub samples_unrouted: u64,
    /// Panes built and stored.
    pub panes_built: u64,
    /// Panes whose build failed; they are skipped, not retried.
    pub panes_failed: u64,
    /// Panes skipped because they would be evicted as soon as built.
    pub panes_skipped_expired: u64,
}

pub struct PrecomputeOptions {
    /// A pane closes when the newest routed sample is this far past its end.
    pub allowed_lateness_ms: u64,
    /// Samples further ahead of the wall clock than this are dropped.
    pub max_future_ms: u64,
}

pub struct Precompute {
    /// Built panes.
    pub store: SummaryStore<Batch>,
    /// What happened to incoming samples.
    pub stats: PrecomputeStats,
    /* private: the DAG, one compiled build per definition, open panes */
}
impl Precompute {
    /// Compiles one build per definition of `sds`, a description of `dag`.
    pub fn new(dag: PhysicalASAPDAG, sds: &Sds, options: PrecomputeOptions) -> Result<Self, PrecomputeError>;
    /// Newest sample time minus the allowed lateness.
    pub fn watermark_ms(&self) -> Option<i64>;
    /// `ingest_at` the wall-clock time.
    pub fn ingest(&mut self, samples: &[RawSample]);
    /// Buffers `samples` (dropping those beyond `now_ms + max_future_ms`),
    /// then builds every pane the watermark has passed.
    pub fn ingest_at(&mut self, samples: &[RawSample], now_ms: i64);
}
/// End of the pane of width `pane_ms` containing `timestamp_ms`, `None` on
/// overflow; panes are `(end - pane_ms, end]`.
pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> Option<i64>;

// data_plane/src/engine.rs
pub struct EngineOptions {
    pub precompute: PrecomputeOptions,
    /// Queued samples beyond which Remote Write gets 503.
    pub max_queued_samples: u64,
}
pub enum IngestError {
    /// Too many samples queued; retry later (503).
    Busy,
    /// The engine thread has exited (500).
    Stopped,
}
/// The HTTP side of the engine thread.
pub struct EngineHandle { /* private */ }
impl EngineHandle {
    /// Starts the engine, which parses the plan once; returns its queries.
    pub fn spawn(plan_json: String, options: EngineOptions) -> anyhow::Result<(Self, Vec<String>)>;
    /// Queues samples for precompute.
    pub fn try_ingest(&self, samples: Vec<RawSample>) -> Result<(), IngestError>;
    pub fn is_running(&self) -> bool;
    pub fn queued_samples(&self) -> u64;
    /// The counters as of the last batch ingested.
    pub fn stats(&self) -> PrecomputeStats;
    /// Stops the engine after the commands queued before.
    pub fn stop(&self);
}

// data_plane/src/sds.rs
impl Sds {
    /// Every ingestion-time node read at query time (by a query-time node,
    /// as a root, or by a scalar root), in node order.
    pub fn ingestion_frontier(dag: &PhysicalASAPDAG) -> Result<Vec<u64>, SdsError>;
}
pub enum SdsError {
    // ...
    /// An ingestion-time node reads a query-time node.
    QueryTimeFeedsIngestion(u64),
}

// data_plane/src/lib.rs
impl DataPlane {
    /// Installs the plan and starts the engine with default options and
    /// `allowed_lateness_ms`; forwarded queries time out after 120 s.
    pub fn new(plan_json: String, prometheus_url: String, allowed_lateness_ms: u64) -> anyhow::Result<Self>;
    /// The same with every engine option.
    pub fn with_options(plan_json: String, prometheus_url: String, options: EngineOptions) -> anyhow::Result<Self>;
    /// Bounds each forwarded query by `timeout`.
    pub fn with_forward_timeout(self, timeout: Duration) -> Self;
}

Before this PR

Remote Write samples were decoded and dropped.

After this PR

Example 4b without raw data, one series sampled every 15 s from 1 to 75 min: the pane ending at 10 min is skipped as possibly partial, and panes ending at 20..70 min are built.

Usage

// The plan and SDS of #819's usage (Example 4b, no raw data).
// Build panes with zero allowed lateness.
let options = PrecomputeOptions { allowed_lateness_ms: 0, ..PrecomputeOptions::default() };
let mut precompute = Precompute::new(dag.clone(), &sds, options)?;
// One series, a sample every 15 s from 1 min to 75 min, values cycling 0..240.
let labels = Arc::new(BTreeMap::from([
    ("__name__".to_string(), "latency_ms".to_string()),
    ("service".to_string(), "a".to_string()),
]));
let samples: Vec<RawSample> = (60_000..4_500_000).step_by(15_000)
    .map(|t| RawSample { labels: labels.clone(), timestamp_ms: t, value: ((t / 15_000) % 240) as f64 })
    .collect();
precompute.ingest(&samples);
// samples_accepted: 259, samples_skipped_first_pane: 37, panes_built: 6, every other counter 0; 6 panes stored

Verification

• data_plane/tests/precompute.rs (14 tests): the watermark and skipped first pane, ignored metrics and dropped late samples, empty panes, label matching, left-open boundaries of pane_end, and the engine putting a sample at exactly 20 min into the pane (10, 20] and one at 20 min + 1 ms into (20, 30]. Added with the review fixes:
  • unrouted_future_sample_does_not_close_panes is the review's repro: an other_metric sample 30 days ahead, then a real sample. It failed on the previous head (4,320 panes built instead of 2) and passes now.
  • drops_samples_beyond_the_future_bound
  • watermark_jump_builds_only_retained_panes: a 30-day jump builds the 7 retained panes and skips the rest.
  • extreme_timestamps_do_not_overflow
  • rejects_zero_width_panes
  • evicts_panes_past_retention
  • accepts_out_of_order_samples_within_lateness
  • stale_markers_are_not_samples
  • series_ahead_of_the_clock_does_not_close_panes_early: the re-review's repro (wall clock base+9m, one series at +13m). It failed on ac6cec1b, where the on-time sample at base+9m was dropped as late, and passes now.
  • far_future_unrouted_sample_counts_as_unrouted (failed on ac6cec1b)
• data_plane/src/engine.rs unit tests: admission (always into an empty queue, Busy otherwise past the bound) and running_guard_clears_on_panic.
• data_plane/tests/sds.rs: ingestion_frontier_matches_the_executor (examples 1, 3b, 4b, with and without raw data) and the store test against the new layout.
• data_plane/tests/server.rs: main's body-limit, header-forwarding and timeout tests run against the engine-backed data plane. New: admits_a_large_request_into_an_empty_queue, does_not_queue_empty_requests, stopped_engine_fails_writes_and_readiness (500, /-/ready 503) and reports_sample_counters (/api/v1/status/asap).
• Ran the MVP CI steps on this tree: sync-script unit tests, node scripts/ci/test_planner_merge.js, cargo fmt --check, cargo check, cargo clippy --all-targets -D warnings and cargo test --workspace, all --locked. All passed.
• Usage output above is from running it as a scratch test on this PR's head.