From a944de85f97d840996a42548eccd6c7c657475fd Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Sat, 18 Apr 2026 15:58:05 -0400 Subject: [PATCH] feat(sketch-db): create_checked retention validation (Method B) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends `BackfillRegistry::create_checked` with the Method-B check from the design discussion: if the requested backfill `start_ms` is older than `persistence_delete_older_than_ms` (the SimpleMapStore retention horizon), reject up-front with `CreateError::OutOfRetention` instead of silently letting the backfill produce windows the retention sweep will immediately evict. ## What's landed New `CreateError::OutOfRetention { agg_id, requested_start_ms, earliest_retained_ms }` variant alongside existing `UnknownAgg` and `Overlap`. `Display` includes a specific actionable message: "extend persistence_delete_older_than before creating this job". `create_checked` signature gains a trailing `data_retention_ms: Option` parameter: - `Some(N)`: reject if `now - start_ms > N`. - `None`: skip the check (testing / retention-disabled deployments). Existing tests that called `create_checked` updated to pass `None`; behavior unchanged there. No other production callsites exist yet — the HTTP endpoint in `handle_post_backfill_job` calls the unchecked `create()` today, so this PR doesn't break any production flow. A follow-up PR will wire the HTTP endpoint through `create_checked` with the deployment's retention value. ## Test plan - [x] 3 new tests: * `create_checked_rejects_start_older_than_data_retention` — start=0 + retention=1h → `OutOfRetention` with correct fields. * `create_checked_none_retention_skips_check` — `None` accepts a start that would otherwise fail. * `create_checked_accepts_start_within_retention` — start=now-30m, retention=1h → accepted (guard against off-by-one at boundary). - [x] 3 existing tests (Overlap, boundary-at-created_at, UnknownAgg) pass with new signature. - [x] 713 lib tests total (up from 710). - [x] clippy `--workspace --all-targets --tests -- -D warnings` clean. - [x] `cargo fmt -- --check` clean. ## Next PR 4: `SchemaEvictionService` tokio task consuming the primitives from PRs 2 + 3 — polls schema registry, cancels in-flight backfills for Expired schemas, drops their agg_id, removes from registry. `--enable-schema-eviction` + `--schema-eviction-dry-run` flags. Default retention 24h. Startup warn if `persistence_delete_older_than < retirement_retention`. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/stores/sketch_db/backfill.rs | 66 +++++++++++-- .../stores/sketch_db/backfill_processor.rs | 96 +++++++++++++++++++ 2 files changed, 152 insertions(+), 10 deletions(-) diff --git a/asap-query-engine/src/stores/sketch_db/backfill.rs b/asap-query-engine/src/stores/sketch_db/backfill.rs index fe8f8dc0..a41675aa 100644 --- a/asap-query-engine/src/stores/sketch_db/backfill.rs +++ 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,17 +522,23 @@ 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, @@ -515,6 +546,7 @@ impl BackfillRegistry { time_range: (u64, u64), source: BackfillSource, windows_total: u64, + data_retention_ms: Option, ) -> Result { 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)) } diff --git a/asap-query-engine/src/stores/sketch_db/backfill_processor.rs b/asap-query-engine/src/stores/sketch_db/backfill_processor.rs index e9019c62..574c21bd 100644 --- a/asap-query-engine/src/stores/sketch_db/backfill_processor.rs +++ b/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()); + } }