fix(flow): preserve stale fence recovery for wrapped errors

Signed-off-by: discord9 <discord9@163.com>

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;

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, .. } => {

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<String, String>,
-    ) -> 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(

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<i64> {
         self.last_exec_time_millis
     }

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<Arc<Schema>, 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,

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,

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<String>) {}
+    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 {

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::{