feat(precompute): build stored summary panes from Remote Write samples

[zzylol]

Stack 3/5. Base: #819. Next: #821.

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:

• Selection: samples are buffered per definition and pane. The plan's own Scan selects them, so its label matchers apply.
• Building: a pane (end − width, end] (left-open, like PromQL ranges and the executor's windows) is built once the watermark (newest sample − allowed lateness) passes end. 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.
• Eviction: panes older than the largest offset plus two widths are evicted.
• /api/v1/write returns 503 when the engine queue is full, so senders retry.
• data_plane --allowed-lateness-ms (default 15 s).

How

The Planner IR holds Rcs, so one engine thread owns the plan, the precompute engine and the store; HTTP handlers send it commands over a bounded channel.

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 accepted: u64,
    /// Samples for a pane already built or skipped.
    pub dropped: u64,
    /// Panes built and stored.
    pub panes_built: u64,
    /// Panes whose build failed; they are skipped, not retried.
    pub panes_failed: 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, allowed_lateness_ms: i64) -> Result<Self, PrecomputeError>;
    /// Newest sample time minus the allowed lateness.
    pub fn watermark_ms(&self) -> Option<i64>;
    /// Buffers `samples`, then builds every pane the watermark has passed.
    pub fn ingest(&mut self, samples: &[RawSample]);
}
/// End of the pane of width `pane_ms` containing `timestamp_ms`; panes are `(end - pane_ms, end]`.
pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> i64;

// data_plane/src/engine.rs
/// The HTTP side of the engine thread.
pub struct EngineHandle { /* private */ }
impl EngineHandle {
    /// Starts the engine for the serialized plan; fails if it cannot be executed.
    pub fn spawn(plan_json: String, allowed_lateness_ms: i64) -> anyhow::Result<Self>;
    /// Queues samples for precompute, or `Busy` if the engine is behind.
    pub fn try_ingest(&self, samples: Vec<RawSample>) -> Result<(), Busy>;
    /// Precompute counters once every command queued before has run.
    pub async fn stats(&self) -> PrecomputeStats;
}

DataPlane::new(plan_json, prometheus_url, allowed_lateness_ms) now starts the engine.

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 mut precompute = Precompute::new(dag.clone(), &sds, 0)?;
// 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);
// PrecomputeStats { accepted: 259, dropped: 37, panes_built: 6, panes_failed: 0 }, 6 panes stored

Verification

• data_plane/tests/precompute.rs (5 tests): the watermark and skipped first pane, ignored metrics and dropped late samples, empty panes, label matching, left-open boundaries.
• cargo test -p control_plane -p data_plane, fmt, clippy on this PR's tree.
• Usage output above is from running it on the feat(query): answer planned instant queries by running the plan's roots #821 tree, whose precompute API is this PR's.