From bd564839b70db928422da9245f437e85243e1d0f Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Tue, 22 Sep 2026 19:19:53 +0800 Subject: [PATCH] feat(flow): freeze explicit recovery windows and retention bounds Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/flow/src/batching_mode/state.rs | 139 ++++++++++++++++++++++-- src/flow/src/batching_mode/task.rs | 97 +++++++++++------ src/flow/src/batching_mode/task/test.rs | 58 +++++++++- 3 files changed, 245 insertions(+), 49 deletions(-) diff --git a/src/flow/src/batching_mode/state.rs b/src/flow/src/batching_mode/state.rs index b9710a25a0d..55307dbcdec 100644 --- a/src/flow/src/batching_mode/state.rs +++ b/src/flow/src/batching_mode/state.rs @@ -208,6 +208,25 @@ impl TaskState { self.pending_fenced_repair.as_ref() } + /// Start repairing explicit bounded windows under a frozen high `H`. + /// + /// The caller supplies the complete repair scope. Live dirty windows are + /// left unchanged so signals received after `H` remain separate. + pub fn start_fenced_repair_windows( + &mut self, + high: BTreeMap, + windows: Vec<(Timestamp, Timestamp)>, + ) { + let mut pending_windows = self.dirty_time_windows.clone(); + pending_windows.clean(); + pending_windows.add_windows(windows); + self.pending_fenced_repair = Some(FencedRepair { + high, + pending_windows, + }); + self.checkpoint_mode = CheckpointMode::FullSnapshot; + } + /// Finish the fenced repair and promote the frozen high watermark to the /// checkpoint map. Incremental-disabled flows stay in FullSnapshot mode. pub fn finish_fenced_repair(&mut self) -> Option> { @@ -259,22 +278,18 @@ impl TaskState { task_ctx: Option<&BatchingTask>, ) -> Result, Error> { if let Some(repair) = self.pending_fenced_repair.as_mut() { - let expr = repair.pending_windows.gen_filter_exprs( + // Fenced windows are an explicit frozen repair scope. They must not + // be pruned by the moving live-data expiration boundary, and an + // empty repair must remain active until its high watermark is + // explicitly finished. + return repair.pending_windows.gen_filter_exprs( col_name, - expire_lower_bound, + None, window_size, window_cnt, flow_id, task_ctx, - )?; - if expr.is_some() || !repair.pending_windows.is_empty() { - return Ok(expr); - } - - // All pending repair windows may have expired during merge. Clear - // the empty repair so this call can fall back to live dirty windows - // instead of routing future executions to an empty queue forever. - self.pending_fenced_repair = None; + ); } self.dirty_time_windows.gen_filter_exprs( @@ -1393,6 +1408,108 @@ mod test { assert_eq!(state.dirty_time_windows.len(), 2); } + #[test] + fn test_explicit_fenced_repair_windows_keep_live_windows_separate() { + let mut state = state_with_past_update(Duration::from_secs(1)); + state + .dirty_time_windows + .add_window(Timestamp::new_second(0), Some(Timestamp::new_second(1_000))); + let high = BTreeMap::from([(1, 10)]); + state.start_fenced_repair_windows( + high.clone(), + vec![(Timestamp::new_second(10), Timestamp::new_second(15))], + ); + + assert_eq!(state.checkpoint_mode(), CheckpointMode::FullSnapshot); + assert_eq!(state.pending_fenced_repair().unwrap().high(), &high); + assert_eq!( + state + .pending_fenced_repair() + .unwrap() + .pending_windows() + .len(), + 1 + ); + assert_eq!(state.dirty_time_windows.len(), 1); + + let filter = state + .gen_scoped_filter_exprs( + "ts", + Some(Timestamp::new_second(100)), + chrono::Duration::seconds(5), + 1, + 1, + None, + ) + .unwrap() + .unwrap(); + assert_eq!( + filter.time_ranges, + vec![(Timestamp::new_second(10), Timestamp::new_second(15))] + ); + state + .dirty_time_windows + .add_window(Timestamp::new_second(20), Some(Timestamp::new_second(25))); + state.restore_scoped_windows(&filter); + + assert_eq!( + state + .pending_fenced_repair() + .unwrap() + .pending_windows() + .len(), + 1 + ); + assert_eq!(state.dirty_time_windows.len(), 2); + } + + #[test] + fn test_explicit_fenced_repair_keeps_empty_high_and_old_windows() { + let mut state = state_with_past_update(Duration::from_secs(1)); + state + .dirty_time_windows + .add_window(Timestamp::new_second(0), Some(Timestamp::new_second(1_000))); + let high = BTreeMap::from([(1, 10)]); + state.start_fenced_repair_windows(high.clone(), Vec::new()); + + assert!( + state + .gen_scoped_filter_exprs( + "ts", + Some(Timestamp::new_second(100)), + chrono::Duration::seconds(5), + 1, + 1, + None, + ) + .unwrap() + .is_none() + ); + assert_eq!(state.pending_fenced_repair().unwrap().high(), &high); + assert_eq!(state.dirty_time_windows.len(), 1); + assert_eq!(state.finish_fenced_repair(), Some(high)); + + state.start_fenced_repair_windows( + BTreeMap::from([(1, 11)]), + vec![(Timestamp::new_second(-20), Timestamp::new_second(-15))], + ); + let filter = state + .gen_scoped_filter_exprs( + "ts", + Some(Timestamp::new_second(100)), + chrono::Duration::seconds(5), + 1, + 1, + None, + ) + .unwrap() + .unwrap(); + assert_eq!( + filter.time_ranges, + vec![(Timestamp::new_second(-20), Timestamp::new_second(-15))] + ); + } + #[test] fn test_disable_incremental_persists_full_snapshot_mode() { let query_ctx = QueryContext::arc(); diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index 8715e7aab0a..4fe4d0dde17 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -539,16 +539,75 @@ impl BatchingTask { Arc::new(query_ctx.fork()) } + /// Returns the retention lower bound aligned to this task's time window. + pub fn recovery_retention_lower_bound(&self) -> Result, Error> { + let Some(expire_after) = self.config.expire_after else { + return Ok(None); + }; + let expire_after = u64::try_from(expire_after).map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Flow {} has negative expire_after {expire_after}", + self.config.flow_id + ), + } + .build() + })?; + let now = Timestamp::new_second( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|err| { + UnexpectedSnafu { + reason: format!("Failed to read recovery wall clock: {err}"), + } + .build() + })? + .as_secs() as i64, + ); + let lower = now + .sub_duration(Duration::from_secs(expire_after)) + .map_err(BoxedError::new) + .context(ExternalSnafu)?; + self.config + .time_window_expr + .as_ref() + .context(UnexpectedSnafu { + reason: "Recovery expiry requires a time-window expression".to_string(), + })? + .eval(lower)? + .0 + .context(UnexpectedSnafu { + reason: "Recovery expiry time-window expression returned no lower bound" + .to_string(), + }) + .map(Some) + } + /// Discovers the source time windows touched by the exact sequence range `(C, H]`. /// - /// Terminal proof must cover ALL regions in `C`; subset proofs and pruning are rejected. - /// The caller owns the execution guard. This method is read-only: it executes a - /// timestamp projection of the aggregate input and leaves task state untouched. + /// This compatibility wrapper captures the retention bound before discovery. pub async fn capture_recovery_windows( &self, engine: &QueryEngineRef, frontend_client: &FrontendClient, lower: &BTreeMap, + ) -> Result<(BTreeMap, Vec<(Timestamp, Timestamp)>), Error> { + let retention_lower = self.recovery_retention_lower_bound()?; + self.capture_recovery_windows_since(engine, frontend_client, lower, retention_lower) + .await + } + + /// Discovers windows under `lower` using a caller-frozen retention bound. + /// + /// The caller owns the execution guard and freezes this bound with its recovery scope. + /// This read-only method leaves task state untouched. Terminal proof must cover every + /// region in `lower`; subset or pruned proofs are rejected. + pub async fn capture_recovery_windows_since( + &self, + engine: &QueryEngineRef, + frontend_client: &FrontendClient, + lower: &BTreeMap, + retention_lower: Option, ) -> Result<(BTreeMap, Vec<(Timestamp, Timestamp)>), Error> { if lower.is_empty() { return UnexpectedSnafu { @@ -600,37 +659,9 @@ impl BatchingTask { self.config.flow_id ), })?; - let input = if let Some(expire_after) = self.config.expire_after { - let expire_after = u64::try_from(expire_after).map_err(|_| { - UnexpectedSnafu { - reason: format!( - "Flow {} has negative expire_after {expire_after}", - self.config.flow_id - ), - } - .build() - })?; - let now = Timestamp::new_second( - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_err(|err| { - UnexpectedSnafu { - reason: format!("Failed to read recovery wall clock: {err}"), - } - .build() - })? - .as_secs() as i64, - ); - let lower = now - .sub_duration(Duration::from_secs(expire_after)) - .map_err(BoxedError::new) - .context(ExternalSnafu)?; - let lower = time_window_expr.eval(lower)?.0.context(UnexpectedSnafu { - reason: "Recovery expiry time-window expression returned no lower bound" - .to_string(), - })?; + let input = if let Some(retention_lower) = retention_lower { let mut add_filter = AddFilterRewriter::new( - col(&time_window_expr.column_name).gt_eq(lit(to_df_literal(lower)?)), + col(&time_window_expr.column_name).gt_eq(lit(to_df_literal(retention_lower)?)), ); input .rewrite(&mut add_filter) diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index f40b108154a..ea4d759effe 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -1749,6 +1749,47 @@ async fn test_capture_recovery_windows_decodes_raw_timestamp_projection_with_whe assert_recovery_state_unchanged(&task, &before_checkpoints, &before_dirty); } +#[tokio::test] +async fn test_capture_recovery_windows_since_uses_supplied_retention_lower() { + let TestTaskParts { + task, + query_engine, + .. + } = new_time_window_test_task_with_query( + "SELECT max(number) AS output_value, date_bin(INTERVAL '5 second', ts) AS output_window FROM numbers_with_ts WHERE number = 42 GROUP BY output_window", + ) + .await; + configure_source_capability(&query_engine, "mito", true).await; + let lower = BTreeMap::from([(1, 10)]); + let retention_lower = Timestamp::new_second(-10); + let handler = Arc::new(RecoveryCaptureHandler { + output: std::sync::Mutex::new(Some(recovery_stream_output( + vec![], + Some(vec![(1, Some(10))]), + false, + ))), + lower: serde_json::to_string(&lower).unwrap(), + expire_after: Some(1), + expiry_lower: std::sync::Mutex::new(None), + }); + let frontend = FrontendClient::from_grpc_handler( + Arc::downgrade( + &(handler.clone() + as Arc), + ), + QueryOptions::default(), + ); + + let (high, windows) = task + .capture_recovery_windows_since(&query_engine, &frontend, &lower, Some(retention_lower)) + .await + .unwrap(); + + assert_eq!(high, lower); + assert!(windows.is_empty()); + assert_eq!(*handler.expiry_lower.lock().unwrap(), Some(retention_lower)); +} + #[tokio::test] async fn test_capture_recovery_windows_terminal_proof_cases_leave_state_unchanged() { struct Case { @@ -2915,7 +2956,7 @@ async fn test_full_snapshot_scoped_plan_marks_checkpoint_advance_safe_only_after } #[tokio::test] -async fn test_expired_empty_fenced_repair_generates_scoped_base_repair_plan() { +async fn test_expired_fenced_repair_uses_frozen_scope() { let TestTaskParts { mut task, query_engine, @@ -2953,11 +2994,18 @@ async fn test_expired_empty_fenced_repair_generates_scoped_base_repair_plan() { ) .await .unwrap() - .expect("expired empty repair should fall back to live dirty"); + .unwrap(); - assert!(matches!(plan.coverage, QueryCoverage::ScopedBaseRepair)); - assert!(plan.coverage.snapshot_seqs().is_empty()); - assert!(task.state.read().unwrap().pending_fenced_repair().is_none()); + // A fenced repair is a frozen explicit scope: expiry cannot discard its + // old window or make this query consume the live post-fence signal. + assert!(matches!( + plan.coverage, + QueryCoverage::FencedRepairChunk { .. } + )); + let state = task.state.read().unwrap(); + assert!(state.pending_fenced_repair().is_some()); + assert!(state.fenced_repair_pending_is_empty()); + assert_eq!(state.dirty_time_windows.len(), 1); } #[tokio::test]