feat(sketch-db): create_checked retention validation (Method B)

asap-query-engine/src/stores/sketch_db/backfill.rs

@@ -294,6 +294,21 @@ pub enum CreateError {
         requested_end_ms: u64,
         created_at_ms: u64,
     },
+    /// The requested `start_ms` is older than the `SimpleMapStore`
+    /// data-retention horizon — any windows the backfill writes
+    /// at that range would immediately be evicted by the
+    /// retention sweep. Method B from the design discussion: fail
+    /// fast at job creation rather than silently letting the
+    /// backfill produce windows that get wiped.
+    ///
+    /// `earliest_retained_ms` is `now - persistence_delete_older_than_ms`,
+    /// i.e. the smallest timestamp that would still survive
+    /// retention at creation time.
+    OutOfRetention {
+        agg_id: u64,
+        requested_start_ms: u64,
+        earliest_retained_ms: u64,
+    },
 }
 
 impl std::fmt::Display for CreateError {
@@ -311,6 +326,16 @@ impl std::fmt::Display for CreateError {
                 "backfill end_ms {requested_end_ms} > agg {agg_id} created_at_ms {created_at_ms}; \
                  live ingest already owns [{created_at_ms}, ∞), refuse to race"
             ),
+            Self::OutOfRetention {
+                agg_id,
+                requested_start_ms,
+                earliest_retained_ms,
+            } => write!(
+                f,
+                "backfill start_ms {requested_start_ms} for agg {agg_id} is older than the store's \
+                 earliest_retained_ms {earliest_retained_ms}; any written windows would be evicted \
+                 by retention — extend persistence_delete_older_than before creating this job"
+            ),
         }
     }
 }
@@ -497,24 +522,31 @@ impl BackfillRegistry {
         job_id
     }
 
-    /// Create a job with the §10.5 time-disjoint invariant enforced:
-    /// `time_range.1` must be `<= agg_id`'s `created_at_ms` in the
-    /// schema registry, so backfill writes never race live writes
-    /// on the same `(agg_id, window)` pair. See the module doc for
-    /// why disjoint-by-construction beats locking.
+    /// Create a job with all §10.5 invariants enforced:
+    ///
+    /// * **Known agg**: `schemas.get(agg_id)` must return `Some`.
+    /// * **Time-disjoint**: `time_range.1 <= schema.created_at_ms`
+    ///   so backfill writes don't race live writes on the same
+    ///   `(agg_id, window)` pair.
+    /// * **Within data retention** (if `data_retention_ms` is
+    ///   provided): `time_range.0 >= now - data_retention_ms`.
+    ///   Method B from the design discussion — fail fast instead
+    ///   of letting the backfill produce windows that the
+    ///   SimpleMapStore retention sweep would immediately evict.
+    ///   Pass `None` to skip the check (tests, or deployments
+    ///   where retention is disabled).
     ///
-    /// Returns `Err(CreateError::Overlap { created_at_ms })` if the
-    /// requested `end_ms` is strictly after the agg's creation
-    /// time, and `Err(CreateError::UnknownAgg)` if the agg_id isn't
-    /// in the schema registry at all (a backfill can't target an
-    /// aggregation the backend doesn't know about).
+    /// Errors map to distinct [`CreateError`] variants so the
+    /// controller-facing HTTP endpoint can return specific 404 /
+    /// 409 / 400 statuses.
     pub fn create_checked(
         &self,
         schemas: &super::SchemaRegistry,
         agg_id: u64,
         time_range: (u64, u64),
         source: BackfillSource,
         windows_total: u64,
+        data_retention_ms: Option<u64>,
     ) -> Result<u64, CreateError> {
         let schema = match schemas.get(agg_id) {
             Some(s) => s,
@@ -530,6 +562,20 @@ impl BackfillRegistry {
                 created_at_ms: schema.created_at_ms,
             });
         }
+        // Data-retention check (Method B): if the store would
+        // immediately evict the windows this job would write,
+        // reject up-front with a clear message rather than
+        // silently wasting CPU + I/O.
+        if let Some(retention_ms) = data_retention_ms {
+            let earliest_retained_ms = now_ms().saturating_sub(retention_ms);
+            if time_range.0 < earliest_retained_ms {
+                return Err(CreateError::OutOfRetention {
+                    agg_id,
+                    requested_start_ms: time_range.0,
+                    earliest_retained_ms,
+                });
+            }
+        }
         Ok(self.create(agg_id, time_range, source, windows_total))
     }
 

asap-query-engine/src/stores/sketch_db/backfill_processor.rs

@@ -574,6 +574,7 @@ mod tests {
                 (0, created + 1),
                 BackfillSource::Prometheus { url: "x".into() },
                 1,
+                None,
             )
             .expect_err("overlap should be rejected");
         match err {
@@ -599,6 +600,7 @@ mod tests {
                 (0, created),
                 BackfillSource::Prometheus { url: "x".into() },
                 1,
+                None,
             )
             .expect("boundary-touching range should be accepted");
         assert!(registry.get(job_id).is_some());
@@ -618,8 +620,102 @@ mod tests {
                 (0, 100),
                 BackfillSource::Prometheus { url: "x".into() },
                 1,
+                None,
             )
             .expect_err("unknown agg should be rejected");
         assert!(matches!(err, CreateError::UnknownAgg { agg_id: 999 }));
     }
+
+    /// Retention guard: requesting a start_ms older than the store's
+    /// data-retention horizon is rejected. Method B from the design
+    /// discussion — fail fast at creation rather than let the backfill
+    /// write windows the retention sweep will immediately delete.
+    #[test]
+    fn create_checked_rejects_start_older_than_data_retention() {
+        use super::super::CreateError;
+        let cfg = sum_config(1, "m", vec![]);
+        let streaming = streaming_config_with(cfg);
+        let schemas = Arc::new(SchemaRegistry::from_streaming_config(&streaming));
+        let registry = BackfillRegistry::new();
+        // Schema's created_at_ms is now_ms(), so data retention of
+        // 1 hour with `start_ms = 0` means we're requesting data
+        // from the epoch — way outside retention.
+        let err = registry
+            .create_checked(
+                &schemas,
+                1,
+                (0, 1_000),
+                BackfillSource::Prometheus { url: "x".into() },
+                1,
+                Some(3_600_000), // 1 hour
+            )
+            .expect_err("out-of-retention should be rejected");
+        match err {
+            CreateError::OutOfRetention {
+                agg_id,
+                requested_start_ms,
+                earliest_retained_ms,
+            } => {
+                assert_eq!(agg_id, 1);
+                assert_eq!(requested_start_ms, 0);
+                assert!(earliest_retained_ms > 0);
+            }
+            other => panic!("expected OutOfRetention, got {other:?}"),
+        }
+    }
+
+    /// `data_retention_ms = None` skips the retention check entirely —
+    /// lets tests and retention-disabled deployments bypass.
+    #[test]
+    fn create_checked_none_retention_skips_check() {
+        let cfg = sum_config(1, "m", vec![]);
+        let streaming = streaming_config_with(cfg);
+        let schemas = Arc::new(SchemaRegistry::from_streaming_config(&streaming));
+        let created = schemas.get(1).unwrap().created_at_ms;
+        let registry = BackfillRegistry::new();
+        // start_ms = 0 would normally fail any realistic retention
+        // window, but None skips the check.
+        let job_id = registry
+            .create_checked(
+                &schemas,
+                1,
+                (0, created),
+                BackfillSource::Prometheus { url: "x".into() },
+                1,
+                None,
+            )
+            .expect("None retention → accepted");
+        assert!(registry.get(job_id).is_some());
+    }
+
+    /// Within-retention start is accepted even when a retention is
+    /// configured. Guards against off-by-one at the boundary.
+    #[test]
+    fn create_checked_accepts_start_within_retention() {
+        let cfg = sum_config(1, "m", vec![]);
+        let streaming = streaming_config_with(cfg);
+        let schemas = Arc::new(SchemaRegistry::from_streaming_config(&streaming));
+        let registry = BackfillRegistry::new();
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+        let retention = 3_600_000_u64; // 1h
+                                       // start_ms = now - 30 min: well within 1h retention.
+        let start = now.saturating_sub(1_800_000);
+        let created = schemas.get(1).unwrap().created_at_ms;
+        // Clip end to the schema boundary so time-disjoint passes.
+        let end = created.min(now);
+        let job_id = registry
+            .create_checked(
+                &schemas,
+                1,
+                (start, end),
+                BackfillSource::Prometheus { url: "x".into() },
+                1,
+                Some(retention),
+            )
+            .expect("within-retention start should be accepted");
+        assert!(registry.get(job_id).is_some());
+    }
 }