fix(precompute): preserve deadline corrections

data_plane/src/main.rs

@@ -512,11 +512,9 @@ async fn main() -> Result<()> {
             channel_buffer_size: args.precompute_channel_buffer_size,
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
-            late_data_policy: LateDataPolicy::Drop,
+            late_data_policy: LateDataPolicy::ForwardToStore,
             wall_clock_idle_grace_period_ms: 5_000,
-            // Enabled only after deadline-triggered late corrections are
-            // guaranteed to append rather than drop.
-            wall_clock_max_open_grace_period_ms: 0,
+            wall_clock_max_open_grace_period_ms: 5_000,
             schema_persist_path: args.schema_persist_path.clone(),
         };
         // M2.3.6 — sketch-only sink. Precompute writes now go to

data_plane/src/precompute_engine/config.rs

@@ -40,9 +40,10 @@ pub struct PrecomputeEngineConfig {
     pub wall_clock_idle_grace_period_ms: i64,
     /// Additional grace for the absolute wall-clock deadline. A pane closes
     /// after `window_size + max_open_grace` from its first input even if it is
-    /// still active. Non-positive disables the deadline. It stays disabled by
-    /// default until late corrections are guaranteed not to be dropped.
-    #[serde(default)]
+    /// still active. Non-positive disables the deadline. The worker applies an
+    /// absolute deadline only with `ForwardToStore`, ensuring later inputs are
+    /// emitted as mergeable corrections. Default: 5000 ms.
+    #[serde(default = "default_wall_clock_max_open_grace_period_ms")]
     pub wall_clock_max_open_grace_period_ms: i64,
     /// Optional path where the `SchemaRegistry` persists per-`agg_id`
     /// lifecycle state across restarts (sketch DB Phase 2c). When
@@ -65,9 +66,9 @@ impl Default for PrecomputeEngineConfig {
             channel_buffer_size: 10_000,
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
-            late_data_policy: LateDataPolicy::Drop,
+            late_data_policy: LateDataPolicy::ForwardToStore,
             wall_clock_idle_grace_period_ms: default_wall_clock_idle_grace_period_ms(),
-            wall_clock_max_open_grace_period_ms: 0,
+            wall_clock_max_open_grace_period_ms: default_wall_clock_max_open_grace_period_ms(),
             schema_persist_path: None,
         }
     }
@@ -77,6 +78,10 @@ fn default_wall_clock_idle_grace_period_ms() -> i64 {
     5_000
 }
 
+fn default_wall_clock_max_open_grace_period_ms() -> i64 {
+    5_000
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -91,9 +96,9 @@ mod tests {
         assert_eq!(config.channel_buffer_size, 10_000);
         assert!(!config.pass_raw_samples);
         assert_eq!(config.raw_mode_aggregation_id, 0);
-        assert_eq!(config.late_data_policy, LateDataPolicy::Drop);
+        assert_eq!(config.late_data_policy, LateDataPolicy::ForwardToStore);
         assert_eq!(config.wall_clock_idle_grace_period_ms, 5_000);
-        assert_eq!(config.wall_clock_max_open_grace_period_ms, 0);
+        assert_eq!(config.wall_clock_max_open_grace_period_ms, 5_000);
     }
 
     #[test]
@@ -113,6 +118,6 @@ wall_clock_grace_period_ms: 7000
         )
         .expect("legacy config should deserialize");
         assert_eq!(config.wall_clock_idle_grace_period_ms, 7_000);
-        assert_eq!(config.wall_clock_max_open_grace_period_ms, 0);
+        assert_eq!(config.wall_clock_max_open_grace_period_ms, 5_000);
     }
 }

data_plane/src/precompute_engine/metrics.rs

@@ -0,0 +1,37 @@
+use lazy_static::lazy_static;
+use prometheus::{register_counter_vec, CounterVec};
+
+lazy_static! {
+    static ref LATE_INPUTS_TOTAL: CounterVec = register_counter_vec!(
+        "asap_precompute_late_inputs_total",
+        "Precompute inputs handled after their event-time window was late or closed",
+        &["action", "input_kind"]
+    )
+    .expect("late-input counter registration must succeed");
+}
+
+pub fn record_late_input(action: &'static str, input_kind: &'static str) {
+    LATE_INPUTS_TOTAL
+        .with_label_values(&[action, input_kind])
+        .inc();
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn late_input_actions_are_exported() {
+        record_late_input("drop", "raw_sample");
+        let family = prometheus::gather()
+            .into_iter()
+            .find(|family| family.get_name() == "asap_precompute_late_inputs_total")
+            .expect("late-input counter must be registered");
+        assert!(family.get_metric().iter().any(|metric| {
+            metric
+                .get_label()
+                .iter()
+                .any(|label| label.get_name() == "action" && label.get_value() == "drop")
+        }));
+    }
+}

data_plane/src/precompute_engine/mod.rs

@@ -2,6 +2,7 @@ pub mod accumulator_factory;
 pub mod config;
 mod engine;
 pub mod ingest_handler;
+mod metrics;
 pub mod operators;
 pub mod output_sink;
 pub mod series_buffer;

data_plane/src/precompute_engine/precompute_engine_design_doc.md

@@ -146,9 +146,9 @@ pub struct PrecomputeEngineConfig {
     pub channel_buffer_size: usize,      // default: 10,000
     pub pass_raw_samples: bool,          // default: false
     pub raw_mode_aggregation_id: u64,    // default: 0
-    pub late_data_policy: LateDataPolicy, // default: Drop
+    pub late_data_policy: LateDataPolicy, // default: ForwardToStore
     pub wall_clock_idle_grace_period_ms: i64, // default: 5,000
-    pub wall_clock_max_open_grace_period_ms: i64, // default: 0 (disabled)
+    pub wall_clock_max_open_grace_period_ms: i64, // default: 5,000
 }
 
 pub enum LateDataPolicy {
@@ -163,6 +163,12 @@ without a touch. When enabled, the absolute deadline fires after
 wall-clock decisions advance only the closure watermark; they never modify the
 maximum observed event timestamp.
 
+The absolute deadline is applied only with `ForwardToStore`: input arriving
+after a deadline close is appended as a mergeable correction for the same
+logical window. With explicit `Drop`, the absolute deadline is disabled so a
+timer cannot introduce silent loss. Both actions increment
+`asap_precompute_late_inputs_total{action,input_kind}`.
+
 ### 3.3 SeriesRouter (`series_router.rs`)
 
 Deterministic hash-based routing using XXHash64:
@@ -1032,19 +1038,12 @@ automatically combined with original window data at query time.
 
 ## 7. Late Data Handling
 
-Two checks determine whether a sample is "late":
-
-1. **Watermark check** (sample-level): `ts < watermark - allowed_lateness_ms` →
-   sample is dropped entirely before reaching any aggregation logic.
-
-2. **Window closure check** (window-level): the sample passes the watermark check
-   but targets a window that is already closed
-   (`window not in active_windows && watermark >= window_end`).
+Two checks classify an input as late: its timestamp is behind the event-time
+watermark, or its target window has already closed. `LateDataPolicy` applies to
+both cases and to both raw samples and prebuilt sketches:
 
-For case 2, the `LateDataPolicy` controls behavior:
-
-- **Drop**: log at debug level and skip. No ghost accumulator is created
-  (fixing the original bug where `or_insert_with` would create orphaned entries).
+- **Drop**: increment `asap_precompute_late_inputs_total{action="drop",...}`,
+  log at debug level, and skip. No ghost accumulator is created.
 
 - **ForwardToStore**: create a fresh `AccumulatorUpdater`, feed the single
   late sample, wrap as `PrecomputedOutput`, and push into the same `emit_batch`
@@ -1129,7 +1128,7 @@ store with the Kafka consumer path.
   | `test_tumbling_window_correctness` | Samples at t=1s/5s/9s; window [0,10s) closes on t=10s; `sum=6` |
   | `test_sliding_window_pane_sharing` | Sample at t=15s in 30s/10s window -> 2 emits for [0,30s) and [10s,40s), both `sum=42` via shared pane snapshot/take |
   | `test_groupby_separate_emits_per_series` | Two series (`host=A`, `host=B`) on same worker -> 2 independent `MultipleSumAccumulator` emits (no ingest-time cross-series merge) |
-  | `test_late_data_drop` | Sample behind `watermark - allowed_lateness_ms` with `Drop` policy -> 0 emits |
+  | `test_late_data_drop` | Sample behind the event watermark with `Drop` policy -> 0 emits and records the action |
   | `test_late_data_forward_to_store` | Late sample for evicted pane with `ForwardToStore` -> 1 emit as mini-accumulator with correct window bounds and sum |
 
 - **Unit tests -- other modules**: `window_manager.rs` (tumbling/sliding arithmetic, pane enumeration, closure detection), `series_buffer.rs` (ordering, watermark), `accumulator_factory.rs` (updater creation and reset), `series_router.rs` (consistent hash routing), `config.rs` (defaults).
@@ -1149,18 +1148,12 @@ The engine is currently in-memory and single-process with no persistence of in-f
 
 | # | Case | When it occurs | Mitigation status |
 |---|---|---|---|
-| 1 | **Explicit late drop** | `LateDataPolicy::Drop` + `ts < watermark - allowed_lateness_ms` | Intended; use `ForwardToStore` to avoid |
-| 2 | **Intra-batch lateness** | Within a single `process_samples` call, `current_wm` is set to the batch's max timestamp before pane routing; with `allowed_lateness_ms=0` every sample below the batch max is dropped | Set `allowed_lateness_ms` >= max timestamp spread within a producer batch |
-| 3 | **Evicted pane + Drop** | Sample passes watermark check but its pane was already evicted (window closed); `Drop` policy discards it | Use `ForwardToStore` |
-| 4 | **No matching config** | `matching_agg_configs` returns empty -- metric name in the series key does not match any config's `metric` or `spatial_filter`; worker silently returns `Ok(())` | No warning is logged. TODO: emit a metric or log at warn level for unmatched series |
-| 5 | **Open panes on shutdown** | `flush_all` only emits windows already closed by the watermark; panes that are still open at shutdown are discarded | TODO (see below) |
-| 6 | **Worker panic** | Tokio task dies; all series owned by that worker lose their pane state; subsequent sends log a warning and drop | TODO (see below) |
-
-### TODO: open-pane flush on shutdown
-
-`flush_all` currently only closes windows whose `end <= watermark`. On graceful shutdown it should optionally force-close all open panes by advancing each series watermark to `i64::MAX` (or to `current_wm + window_size_ms`) before the final flush. This would emit partial windows with whatever samples have accumulated, allowing downstream consumers to decide whether to use them.
+| 1 | **Explicit late drop** | `LateDataPolicy::Drop` handles a late or already-closed-window input | Intended and counted; use `ForwardToStore` to retain corrections |
+| 2 | **No matching config** | Metric name does not match a configured policy | Warn and expose the existing policy-miss diagnostics |
+| 3 | **Worker panic** | Tokio task dies with in-memory panes | TODO: worker supervision and durable pane recovery |
 
-This behaviour should be opt-in (a `force_flush_on_shutdown: bool` config flag) because partial windows can be misleading for consumers that expect complete windows.
+On graceful shutdown, `force_close_all` emits all remaining raw and sketch
+panes. These are partial windows when the event-time boundary was not reached.
 
 ### TODO: warn on unmatched series