From 6ddb2d2ec4df2b3255de11a74b6d1b9dd3073d55 Mon Sep 17 00:00:00 2001 From: discord9 Date: Tue, 8 Sep 2026 13:19:15 +0800 Subject: [PATCH] fix(flow): preserve stale fence recovery for wrapped errors Signed-off-by: discord9 --- src/flow/src/batching_mode.rs | 8 - src/flow/src/batching_mode/checkpoint.rs | 4 +- src/flow/src/batching_mode/engine.rs | 8 - src/flow/src/batching_mode/state.rs | 4 - src/flow/src/batching_mode/task.rs | 9 -- src/flow/src/batching_mode/task/ckpt.rs | 50 ++++++- src/flow/src/batching_mode/task/test.rs | 182 +++++++++++++++++++++++ src/flow/src/lib.rs | 2 +- 8 files changed, 231 insertions(+), 36 deletions(-) diff --git a/src/flow/src/batching_mode.rs b/src/flow/src/batching_mode.rs index bd884f8fa0..920f847083 100644 --- a/src/flow/src/batching_mode.rs +++ b/src/flow/src/batching_mode.rs @@ -20,14 +20,6 @@ use common_grpc::channel_manager::ClientTlsOption; use serde::{Deserialize, Serialize}; use session::ReadPreference; -/// Runtime-only mode used for incremental source scans. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub enum IncrementalMode { - #[default] - MemtableOnly, - SequenceRange, -} - pub(crate) mod batching_execution; pub(crate) mod checkpoint; pub(crate) mod engine; diff --git a/src/flow/src/batching_mode/checkpoint.rs b/src/flow/src/batching_mode/checkpoint.rs index b7a0d9f9ea..8c32c717fa 100644 --- a/src/flow/src/batching_mode/checkpoint.rs +++ b/src/flow/src/batching_mode/checkpoint.rs @@ -115,8 +115,8 @@ impl FlowCheckpointDecision { } Self::AdvancedIncremental { .. } => checkpoint_mode_label(CheckpointMode::Incremental), Self::ContinuedFencedRepair { .. } => { - // Fenced repair and completion of a requested full repair are - // FullSnapshot sub-states, not third top-level modes. + // Fenced repair is a FullSnapshot sub-state, not a third + // top-level mode. checkpoint_mode_label(CheckpointMode::FullSnapshot) } Self::FallbackToFullSnapshot { previous_mode, .. } => { diff --git a/src/flow/src/batching_mode/engine.rs b/src/flow/src/batching_mode/engine.rs index e681a56c9e..a89f8ab63d 100644 --- a/src/flow/src/batching_mode/engine.rs +++ b/src/flow/src/batching_mode/engine.rs @@ -519,14 +519,6 @@ impl BatchingEngine { .is_some_and(|value| value.eq_ignore_ascii_case("true")) } - fn table_options_enable_merge_mode_last_non_null( - extra_options: &HashMap, - ) -> bool { - extra_options - .get(store_api::mito_engine_options::MERGE_MODE_KEY) - .is_some_and(|value| value.eq_ignore_ascii_case("last_non_null")) - } - /// SQL flows without a usable time-window expression can only run as an /// explicit full-query flow, so require `EVAL INTERVAL` at creation time. fn ensure_sql_flow_has_twe_or_eval_interval( diff --git a/src/flow/src/batching_mode/state.rs b/src/flow/src/batching_mode/state.rs index 2d339a4041..e515a449f8 100644 --- a/src/flow/src/batching_mode/state.rs +++ b/src/flow/src/batching_mode/state.rs @@ -112,10 +112,6 @@ impl TaskState { } } - pub(crate) fn last_query_duration(&self) -> Duration { - self.last_query_duration - } - pub fn last_execution_time_millis(&self) -> Option { self.last_exec_time_millis } diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index 7f1e2fb633..4f5e858b1d 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -473,15 +473,6 @@ impl BatchingTask { .await } - pub(crate) async fn validate_sink_table_schema_with_table( - &self, - engine: &QueryEngineRef, - table: TableRef, - ) -> Result, Error> { - self.validate_sink_table_schema_with_table_and_values(engine, table, &BTreeMap::new()) - .await - } - async fn validate_sink_table_schema_with_table_and_values( &self, engine: &QueryEngineRef, diff --git a/src/flow/src/batching_mode/task/ckpt.rs b/src/flow/src/batching_mode/task/ckpt.rs index c6b8503c32..960e541c73 100644 --- a/src/flow/src/batching_mode/task/ckpt.rs +++ b/src/flow/src/batching_mode/task/ckpt.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::error::Error as StdError; use std::time::Duration; use client::OutputWithMetrics; @@ -30,6 +31,42 @@ use crate::metrics::{ }; use crate::{Error, FlowId}; +/// Liveness guard: when a fenced repair query fails with a wrapped error whose +/// text indicates a stale snapshot fence (even when `StatusCode::RequestOutdated` +/// was lost through client layers), classify it as `SnapshotFenceExpired` to +/// break the retry loop and force a rebind of the fence high `H`. +/// +/// Long-term the structured `StatusCode` / retry hint path should be preserved +/// end-to-end; this text fallback is a narrow safety measure. +fn matches_stale_snapshot_fence_text(err: &Error) -> bool { + let markers = [ + "STALE_SNAPSHOT_FENCE", + "REBIND_SNAPSHOT_FENCE", + "snapshot upper bound stale", + ]; + // Check the top-level error Display and Debug. + let debug_str = format!("{:?}", err); + let display_str = err.to_string(); + for marker in &markers { + if debug_str.contains(marker) || display_str.contains(marker) { + return true; + } + } + // Walk the error source chain. + let mut source = err.source(); + while let Some(s) = source { + let debug_str = format!("{:?}", s); + let display_str = s.to_string(); + for marker in &markers { + if debug_str.contains(marker) || display_str.contains(marker) { + return true; + } + } + source = s.source(); + } + false +} + impl BatchingTask { /// Classify execution errors into checkpoint fallback reasons. A stale /// snapshot fence is special only for fenced repair chunks. @@ -43,6 +80,14 @@ impl BatchingTask { } else { FlowQueryFallbackReason::StaleCursor } + } else if matches!(coverage, QueryCoverage::FencedRepairChunk { .. }) + && matches_stale_snapshot_fence_text(err) + { + // Narrow text-based fallback for wrapped errors where the + // structured StatusCode::RequestOutdated was lost through + // frontend/client layers. Without this fenced repair will + // retry the same stale `given_seq` every refresh tick. + FlowQueryFallbackReason::SnapshotFenceExpired } else if matches!(coverage, QueryCoverage::IncrementalDelta) { FlowQueryFallbackReason::IncrementalQueryFailure } else { @@ -87,10 +132,7 @@ impl BatchingTask { } /// Apply checkpoint transitions for a successfully executed query using its - /// terminal watermark proof and declared coverage, while retaining whether this attempt - /// started with a persisted full-repair request. That fact is deliberately - /// captured by the caller before execution; a repair request raised later - /// must not turn an ordinary full snapshot into a completed repair. + /// terminal watermark proof and declared coverage. pub(super) fn apply_query_result_to_state( state: &mut TaskState, res: &OutputWithMetrics, diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index 5ac695a530..1f78e23fd3 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -526,6 +526,50 @@ fn flow_error_with_status(status_code: StatusCode) -> Error { .unwrap_err() } +/// Test-only error that carries a non-RequestOutdated status code but +/// displays a stale-snapshot-fence marker string, simulating the real-world +/// scenario where the structured status code is lost through frontend/client +/// wrapping layers. +#[derive(Debug)] +struct StaleFenceTextError { + code: StatusCode, + message: String, +} + +impl std::fmt::Display for StaleFenceTextError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.message) + } +} + +impl std::error::Error for StaleFenceTextError {} + +impl common_error::ext::ErrorExt for StaleFenceTextError { + fn status_code(&self) -> StatusCode { + self.code + } + fn as_any(&self) -> &dyn std::any::Any { + self + } +} + +impl common_error::ext::StackError for StaleFenceTextError { + fn debug_fmt(&self, _: usize, _: &mut Vec) {} + fn next(&self) -> Option<&dyn common_error::ext::StackError> { + None + } +} + +fn flow_error_with_code_and_text(code: StatusCode, text: &str) -> Error { + let inner = StaleFenceTextError { + code, + message: text.to_string(), + }; + Err::<(), _>(BoxedError::new(inner)) + .context(crate::error::ExternalSnafu) + .unwrap_err() +} + fn dirty_range(start: i64, end: i64) -> DirtyTimeWindows { let mut dirty = DirtyTimeWindows::default(); dirty.add_window( @@ -1567,6 +1611,73 @@ fn test_query_failure_reason_distinguishes_fenced_repair_stale_fence() { ); } +#[test] +fn test_query_failure_reason_text_fallback_stale_snapshot_fence() { + let high = BTreeMap::new(); + let fenced = QueryCoverage::FencedRepairChunk { high: high.clone() }; + + // STALE_SNAPSHOT_FENCE marker with a non-RequestOutdated status code + let err = flow_error_with_code_and_text( + StatusCode::Internal, + "gRPC error: STALE_SNAPSHOT_FENCE: snapshot upper bound stale, region: 1024/0", + ); + assert_eq!( + BatchingTask::query_failure_reason(&err, &fenced), + FlowQueryFallbackReason::SnapshotFenceExpired + ); + + // REBIND_SNAPSHOT_FENCE marker + let err = flow_error_with_code_and_text( + StatusCode::Internal, + "STALE_SNAPSHOT_FENCE ... retry_hint: REBIND_SNAPSHOT_FENCE", + ); + assert_eq!( + BatchingTask::query_failure_reason(&err, &fenced), + FlowQueryFallbackReason::SnapshotFenceExpired + ); + + // snapshot upper bound stale marker (the natural-language fragment) + let err = flow_error_with_code_and_text( + StatusCode::Internal, + "query failed: snapshot upper bound stale, consider rebinding", + ); + assert_eq!( + BatchingTask::query_failure_reason(&err, &fenced), + FlowQueryFallbackReason::SnapshotFenceExpired + ); + + // Fenced coverage with a generic wrapped error (no stale-fence marker) → + // still QueryFailure + let generic_err = + flow_error_with_code_and_text(StatusCode::Internal, "some transient network error"); + assert_eq!( + BatchingTask::query_failure_reason(&generic_err, &fenced), + FlowQueryFallbackReason::QueryFailure + ); + + // Non-fenced incremental coverage with stale-fence marker text must NOT + // classify as SnapshotFenceExpired; it should remain IncrementalQueryFailure. + let err = flow_error_with_code_and_text( + StatusCode::Internal, + "STALE_SNAPSHOT_FENCE blob in unexpected context", + ); + assert_eq!( + BatchingTask::query_failure_reason(&err, &QueryCoverage::IncrementalDelta), + FlowQueryFallbackReason::IncrementalQueryFailure + ); + + // Existing RequestOutdated behavior is unchanged. + let outdated_err = flow_error_with_status(StatusCode::RequestOutdated); + assert_eq!( + BatchingTask::query_failure_reason(&outdated_err, &fenced), + FlowQueryFallbackReason::SnapshotFenceExpired + ); + assert_eq!( + BatchingTask::query_failure_reason(&outdated_err, &QueryCoverage::IncrementalDelta), + FlowQueryFallbackReason::StaleCursor + ); +} + #[test] fn test_fenced_repair_coverage_produces_snapshot_seq_map_for_distributed_metadata_path() { // Covers the metadata boundary between QueryCoverage and the @@ -1714,6 +1825,77 @@ fn test_fenced_repair_transient_non_stale_failure_retries_same_high() { ); } +#[tokio::test] +async fn test_text_fallback_stale_fence_produces_scoped_base_repair() { + let TestTaskParts { + task, + query_engine, + .. + } = new_time_window_test_task_with_query( + "SELECT number, date_bin(INTERVAL '5 second', ts) AS time_window FROM numbers_with_ts GROUP BY time_window, number", + ) + .await; + let high = BTreeMap::from([(1_u64, 10_u64), (2_u64, 20_u64)]); + let filter = { + let mut state = task.state.write().unwrap(); + state + .dirty_time_windows + .add_window(Timestamp::new_second(10), Some(Timestamp::new_second(15))); + state + .dirty_time_windows + .add_window(Timestamp::new_second(100), Some(Timestamp::new_second(105))); + state.start_fenced_repair(high.clone()).unwrap(); + next_fenced_repair_filter(&mut state, 1) + }; + + // Construct a wrapped error that hits the text fallback (non-RequestOutdated + // status code with STALE_SNAPSHOT_FENCE marker text). + let err = flow_error_with_code_and_text( + StatusCode::Internal, + "STALE_SNAPSHOT_FENCE: snapshot upper bound stale, retry_hint: REBIND_SNAPSHOT_FENCE", + ); + let coverage = QueryCoverage::FencedRepairChunk { high }; + let reason = BatchingTask::query_failure_reason(&err, &coverage); + assert_eq!(reason, FlowQueryFallbackReason::SnapshotFenceExpired); + + { + let mut state = task.state.write().unwrap(); + let decision = BatchingTask::apply_query_failure_to_state( + &mut state, + std::time::Duration::from_millis(1), + &coverage, + reason, + ); + assert_eq!( + decision, + Some(FlowCheckpointDecision::FallbackToFullSnapshot { + previous_mode: CheckpointMode::FullSnapshot, + reason: FlowQueryFallbackReason::SnapshotFenceExpired, + }) + ); + assert!(state.pending_fenced_repair().is_none()); + + // Simulate the outer execution failure restore for the in-flight chunk. + state.restore_scoped_windows(&filter); + } + + let plan = task + .gen_query_with_time_window( + query_engine, + &aggregate_time_window_sink_schema(), + &[], + false, + Some(1), + ) + .await + .unwrap() + .expect("text-fallback stale fence should restore dirty windows for a fresh scoped repair"); + assert!( + matches!(plan.coverage, QueryCoverage::ScopedBaseRepair), + "next plan after text-fallback stale fence should be ScopedBaseRepair" + ); +} + #[test] fn test_checkpoint_decision_labels_are_stable() { let advance = FlowCheckpointDecision::AdvancedIncremental { diff --git a/src/flow/src/lib.rs b/src/flow/src/lib.rs index 4e5ee26ce5..5bfaf8114f 100644 --- a/src/flow/src/lib.rs +++ b/src/flow/src/lib.rs @@ -43,6 +43,7 @@ mod test_utils; pub use adapter::flownode_impl::FlowDualEngineRef; pub use adapter::{FlowConfig, FlowStreamingEngineRef, StreamingEngine}; +pub use batching_mode::BatchingModeOptions; pub use batching_mode::batching_execution::{BatchingExecution, BatchingExecutionFactory}; pub use batching_mode::frontend_client::{ FrontendClient, GrpcQueryHandlerWithBoxedError, PeerDesc, @@ -52,7 +53,6 @@ pub use batching_mode::task::{ }; pub use batching_mode::time_window::{TimeWindowExpr, find_time_window_expr}; pub use batching_mode::utils::sql_to_df_plan; -pub use batching_mode::{BatchingModeOptions, IncrementalMode}; pub(crate) use engine::{CreateFlowArgs, FlowId, TableName}; pub use error::{Error, Result}; pub use server::{