[zzylol]

Base: #824 (stored outputs built in a QueryRun). Ports #814 to the plan-executing engine.

Why

A pane closes only when a later sample of a routed metric passes its end (plus the allowed lateness). When a series stops sending, its last pane stays open, and every planned query reading that window is forwarded. Remote Write gives the producer no way to say "everything up to T is written". #814 added that statement to the old backend; this PR adds it to the engine that executes the Planner's plan.

What

• Endpoint: POST /api/v1/precompute/watermark?time_ms=T. It answers {"barrier_ms": B}, the barrier in force: T capped at the wall clock, or a higher earlier barrier. A missing or non-integer time_ms gets 400 bad_data with the reason, and a stopped engine gets 500. Input stays open.
• Ordering: the barrier is queued behind every write already acknowledged, and the response is sent once it has run.
• Effect: every pane with end <= T is closed, keeping the left-open (end − width, end] convention. The last pane of a series that stopped sending is built, and a planned query reading it is answered by the plan.
• Cap: T is capped at the wall clock, like the sample watermark.
• Repeats: a repeated or lower T does nothing.
• Overflow: closing reuses the expired-pane skip and checked arithmetic, so T = i64::MAX terminates and builds only the retained panes (at most seven for Example 4b).
• Later samples: a sample that arrives later for a pane the barrier closed (one ending at or before T) counts as late. A later sample in a pane still open is accepted.
• Several producers: send the barrier only after all of them have written through T; the README says so. The backend keeps no producer roster.
• Design doc: for the "idle shards" open decision, this is the producer-side answer. A wall-clock idle close stays open.
• Later in the stack: with sharded ingestion workers, the barrier advances every worker.

Key code interfaces added / modified

// data_plane/src/precompute.rs
impl Precompute {
    /// `advance_watermark_at` the wall clock.
    pub fn advance_watermark(&mut self, barrier_ms: i64) -> i64;
    /// Closes every pane with end <= min(barrier_ms, now_ms); a repeated or
    /// lower barrier does nothing. Returns the barrier in force.
    pub fn advance_watermark_at(&mut self, barrier_ms: i64, now_ms: i64) -> i64;
}
// data_plane/src/engine.rs
impl EngineHandle {
    /// Queues the barrier behind acknowledged writes and waits until it has run.
    pub async fn advance_watermark(&self, barrier_ms: i64) -> Result<i64, IngestError>;
}

Before this PR

Example 4b without raw data, with one series that stops at 69:45 and 15 s allowed lateness: the pane (60, 70 min] stays open, and quantile_over_time(0.99, latency_ms[60m]) at 70 min is forwarded.

After this PR

After POST /api/v1/precompute/watermark?time_ms=4200000, the same query at 70 min is answered by the plan, and later writes are still accepted.

Verification

• data_plane/tests/answers.rs: barrier_serves_the_window_of_a_stopped_series. It fails before this PR, because the endpoint did not exist. Before the barrier the query is forwarded; after it, the plan answers. Input is still accepted. barrier_ms is in the response. A missing, non-numeric or fractional time_ms gets 400 bad_data with the reason.
• data_plane/tests/precompute.rs:
  • barrier_closes_the_last_pane_of_a_stopped_series
  • barrier_is_monotonic_and_capped_at_the_wall_clock
  • barrier_at_i64_max_terminates
  • input_stays_open_after_a_barrier
• cargo test --workspace --locked passes.