request: treat an undetermined apply outcome as terminal at commit points

The re-vendored kvproto exposes errorpb.UndeterminedResult: the server telling
us the raft apply outcome is UNKNOWN. Today it falls through to the generic
region-error arm and is retried like any transient failure. For most requests
that is right, but not for a commit point, where "failed" must never be claimed
for a transaction that may already be durable.

Follow client-go's shape: its transport never retries UndeterminedResult
("should not retry ... processed by the caller", region_request.go) and each
ACTION decides. So the plan layer keeps RETRY as the default — replaying an
idempotent request resolves the uncertainty, and on backoff exhaustion the
error escapes UNCHANGED so callers can still classify it — while plans for
which a replay is unsafe opt in to terminal-on-first-sight via
retry_multi_region_terminal_on_undetermined:

- raw CAS: a replay would compare against its OWN effect and report
  succeed = false for a write that happened.
- the primary commit: client-go returns ErrResultUndetermined here (commit.go)
  rather than retrying.
- async-commit / 1PC prewrites: the prewrite IS the commit point. Stricter
  than client-go, in the safe direction.

Shard aggregation is fixed alongside: collect::<Result<_>>() returns the first
error BY SHARD INDEX, so a lower-index shard's determinate error could mask a
higher-index shard's UndeterminedResult and tell the caller the request
definitely failed. An undetermined error from any shard now wins.

Commit paths classify the surfaced error as UndeterminedError. Two places are
deliberately conservative and say so in comments: a 1PC prewrite that sharding
downgraded to 2PC, and an undetermined error arriving from the commit plan's
resolve_lock layer rather than the commit dispatch. Both only over-report
uncertainty, which makes the caller verify instead of replay.

Split out of the kvproto re-vendor at pingyu's suggestion (#550).

Tests: 4 new — terminal for opted-in plans under a generous backoff (proving
terminal, not retry-exhausted), preserved unchanged on retry exhaustion,
recognizer positive/negative, and the shard-masking hazard. 71 lib tests green.

Signed-off-by: Eduard R. <eduard@ralphovi.net>

Author: eduralph

src/raw/client.rs

@@ -697,7 +697,10 @@ impl<PdC: PdClient> Client<PdC> {
             self.cf.clone(),
         );
         let plan = crate::request::PlanBuilder::new(self.rpc.clone(), self.keyspace, req)
-            .retry_multi_region(self.backoff.clone())
+            // CAS is not replay-safe once the outcome is unknown: if the first attempt
+            // applied, a retry would compare against the swapped-in value and report
+            // `succeed = false` for a write that HAPPENED. Terminal on first sight.
+            .retry_multi_region_terminal_on_undetermined(self.backoff.clone())
             .merge(CollectSingle)
             .post_process_default()
             .plan();

src/request/plan.rs

@@ -132,6 +132,14 @@ where
         .collect())
 }
 
+/// Did the server say the request's outcome is UNKNOWN — e.g. a raft timeout where the
+/// apply result was never observed (`errorpb.UndeterminedResult`)? A commit receiving
+/// this must be reported as undetermined: reporting plain failure would invite the
+/// caller to retry effects that may already be durable.
+pub(crate) fn is_undetermined_region_error(e: &Error) -> bool {
+    matches!(e, Error::RegionError(re) if re.undetermined_result.is_some())
+}
+
 pub struct RetryableMultiRegion<P: Plan, PdC: PdClient> {
     pub(super) inner: P,
     pub pd_client: Arc<PdC>,
@@ -141,6 +149,19 @@ pub struct RetryableMultiRegion<P: Plan, PdC: PdClient> {
     /// If true, return Ok and preserve all regions' results, even if some of them are Err.
     /// Otherwise, return the first Err if there is any.
     pub preserve_region_results: bool,
+
+    /// Terminal treatment of `errorpb.UndeterminedResult` (an unknown raft apply
+    /// outcome). client-go's transport never retries it and each ACTION decides
+    /// (region_request.go: "should not retry ... processed by the caller"); its one
+    /// terminal action is the primary non-async commit (commit.go: returns
+    /// ErrResultUndetermined). Plans opt in when a replay could produce a WRONG
+    /// RESULT (raw CAS replaying against its own effect) or when the request is a
+    /// commit point and a later, different retry error must not overwrite the
+    /// uncertainty (primary commit; async/1PC prewrite — stricter than client-go,
+    /// in the safe direction). Everything else retries: re-applying an idempotent
+    /// request resolves the uncertainty, and on backoff exhaustion the error
+    /// escapes UNCHANGED for the caller to classify.
+    pub terminal_on_undetermined: bool,
 }
 
 impl<P: Plan + Shardable, PdC: PdClient> RetryableMultiRegion<P, PdC>
@@ -155,6 +176,7 @@ where
         backoff: Backoff,
         permits: Arc<Semaphore>,
         preserve_region_results: bool,
+        terminal_on_undetermined: bool,
     ) -> Result<<Self as Plan>::Result> {
         let shards = current_plan.shards(&pd_client).collect::<Vec<_>>().await;
         let shards_len = shards.len();
@@ -182,6 +204,7 @@ where
                         backoff,
                         permits,
                         preserve_region_results,
+                        terminal_on_undetermined,
                     )
                     .await,
                 )
@@ -200,12 +223,30 @@ where
                 })
                 .collect())
         } else {
-            Ok(results
-                .into_iter()
-                .collect::<Result<Vec<_>>>()?
-                .into_iter()
-                .flatten()
-                .collect())
+            // A terminal undetermined outcome must never be masked by another shard's
+            // determinate error. `collect::<Result<_>>()` would return the first error
+            // by shard index, so if a lower-index shard failed with (say) a gRPC error
+            // while a higher-index shard reported `UndeterminedResult`, the caller would
+            // be told the request definitely failed — and might replay a transaction
+            // that may already be durable. So an undetermined error
+            // from ANY shard wins over a determinate one.
+            let mut oks = Vec::with_capacity(results.len());
+            let mut first_err: Option<Error> = None;
+            let mut undetermined: Option<Error> = None;
+            for r in results {
+                match r {
+                    Ok(v) => oks.push(v),
+                    Err(e) if undetermined.is_none() && is_undetermined_region_error(&e) => {
+                        undetermined = Some(e)
+                    }
+                    Err(e) if first_err.is_none() => first_err = Some(e),
+                    Err(_) => {}
+                }
+            }
+            if let Some(e) = undetermined.or(first_err) {
+                return Err(e);
+            }
+            Ok(oks.into_iter().flatten().collect())
         }
     }
 
@@ -217,6 +258,7 @@ where
         mut backoff: Backoff,
         permits: Arc<Semaphore>,
         preserve_region_results: bool,
+        terminal_on_undetermined: bool,
     ) -> Result<<Self as Plan>::Result> {
         let region_ver_id = region.ver_id();
         let store_id = region.get_store_id().ok();
@@ -243,6 +285,7 @@ where
                     backoff,
                     permits,
                     preserve_region_results,
+                    terminal_on_undetermined,
                     err,
                 )
                 .await;
@@ -266,6 +309,7 @@ where
                     backoff,
                     permits,
                     preserve_region_results,
+                    terminal_on_undetermined,
                     e,
                 )
                 .await;
@@ -284,6 +328,13 @@ where
                 "single_shard_handler:execute: region error: {:?}, region: {:?}",
                 e, region_ver_id
             );
+            // See `terminal_on_undetermined`: for CAS and commit points, an unknown
+            // apply outcome must surface on FIRST sight — a replay could contradict
+            // its own effect, and a later different error must not overwrite the
+            // uncertainty. The caller classifies (commit maps it to UndeterminedError).
+            if terminal_on_undetermined && e.undetermined_result.is_some() {
+                return Err(Error::RegionError(Box::new(e)));
+            }
             match backoff.next_delay_duration() {
                 Some(duration) => {
                     let region_error_resolved =
@@ -298,6 +349,7 @@ where
                         backoff,
                         permits,
                         preserve_region_results,
+                        terminal_on_undetermined,
                     )
                     .await
                 }
@@ -323,6 +375,7 @@ where
         mut backoff: Backoff,
         permits: Arc<Semaphore>,
         preserve_region_results: bool,
+        terminal_on_undetermined: bool,
         e: Error,
     ) -> Result<<Self as Plan>::Result> {
         debug!("handle_other_error: {:?}", e);
@@ -341,6 +394,7 @@ where
                     backoff,
                     permits,
                     preserve_region_results,
+                    terminal_on_undetermined,
                 )
                 .await
             }
@@ -392,6 +446,15 @@ pub(crate) async fn handle_region_error<PdC: PdClient>(
     } else if e.stale_command.is_some() || e.region_not_found.is_some() {
         pd_client.invalidate_region_cache(ver_id).await;
         Ok(false)
+    } else if e.undetermined_result.is_some() {
+        // The apply outcome is UNKNOWN (a raft timeout, errorpb.UndeterminedResult).
+        // Default: retry — matching client-go's ACTION layers (ordinary prewrites and
+        // secondary commits back off and re-send; re-applying an idempotent request
+        // resolves the uncertainty). Routing is not suspect, so nothing is
+        // invalidated. On backoff exhaustion the error escapes UNCHANGED, and commit
+        // paths classify it via `is_undetermined_region_error`. Plans for which a
+        // replay is unsafe never reach this arm — see `terminal_on_undetermined`.
+        Ok(false)
     } else if e.server_is_busy.is_some()
         || e.raft_entry_too_large.is_some()
         || e.max_timestamp_not_synced.is_some()
@@ -457,6 +520,7 @@ impl<P: Plan, PdC: PdClient> Clone for RetryableMultiRegion<P, PdC> {
             pd_client: self.pd_client.clone(),
             backoff: self.backoff.clone(),
             preserve_region_results: self.preserve_region_results,
+            terminal_on_undetermined: self.terminal_on_undetermined,
         }
     }
 }
@@ -479,6 +543,7 @@ where
             self.backoff.clone(),
             concurrency_permits.clone(),
             self.preserve_region_results,
+            self.terminal_on_undetermined,
         )
         .await
     }
@@ -1042,6 +1107,7 @@ mod test {
             pd_client: Arc::new(MockPdClient::default()),
             backoff: Backoff::no_backoff(),
             preserve_region_results: false,
+            terminal_on_undetermined: false,
         };
         assert!(plan.execute().await.is_err())
     }
@@ -1062,4 +1128,172 @@ mod test {
 
         assert_eq!(results, vec![0, 1, 2]);
     }
+
+    #[derive(Clone)]
+    struct UndeterminedPlan;
+
+    #[async_trait]
+    impl Plan for UndeterminedPlan {
+        type Result = BatchGetResponse;
+
+        async fn execute(&self) -> Result<Self::Result> {
+            // The server says the apply outcome is UNKNOWN.
+            Ok(BatchGetResponse {
+                region_error: Some(errorpb::Error {
+                    undetermined_result: Some(errorpb::UndeterminedResult::default()),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            })
+        }
+    }
+
+    impl Shardable for UndeterminedPlan {
+        type Shard = ();
+
+        fn shards(
+            &self,
+            _: &Arc<impl crate::pd::PdClient>,
+        ) -> BoxStream<'static, crate::Result<(Self::Shard, RegionWithLeader)>> {
+            Box::pin(stream::iter(vec![Ok(((), MockPdClient::region1()))])).boxed()
+        }
+
+        fn apply_shard(&mut self, _: Self::Shard) {}
+
+        fn apply_store(&mut self, _: &crate::store::RegionStore) -> Result<()> {
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn an_undetermined_result_is_terminal_on_first_sight_for_optedin_plans() {
+        // CAS and commit points: a replay could contradict its own effect, and a
+        // later different error must not overwrite the uncertainty. A GENEROUS
+        // backoff proves the error is terminal, not merely retry-exhausted.
+        let plan = RetryableMultiRegion {
+            inner: UndeterminedPlan,
+            pd_client: Arc::new(MockPdClient::default()),
+            backoff: Backoff::no_jitter_backoff(1, 2, 10),
+            preserve_region_results: false,
+            terminal_on_undetermined: true,
+        };
+        let err = plan.execute().await.unwrap_err();
+        assert!(
+            is_undetermined_region_error(&err),
+            "want the undetermined region error surfaced unchanged, got {err:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn an_undetermined_result_stays_recognizable_when_retries_exhaust() {
+        // Everything else retries (client-go's action-layer behavior for ordinary
+        // prewrites and secondary commits); when the backoff exhausts, the error
+        // must escape UNCHANGED so commit paths can still classify it as undetermined.
+        let plan = RetryableMultiRegion {
+            inner: UndeterminedPlan,
+            pd_client: Arc::new(MockPdClient::default()),
+            backoff: Backoff::no_backoff(),
+            preserve_region_results: false,
+            terminal_on_undetermined: false,
+        };
+        let err = plan.execute().await.unwrap_err();
+        assert!(
+            is_undetermined_region_error(&err),
+            "want the undetermined region error preserved, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn undetermined_region_errors_are_recognized() {
+        // errorpb.UndeterminedResult: the apply outcome is UNKNOWN. The commit path
+        // must map this to UndeterminedError — a plain failure would invite the caller
+        // to retry effects that may already be durable.
+        let undetermined = Error::RegionError(Box::new(errorpb::Error {
+            undetermined_result: Some(errorpb::UndeterminedResult::default()),
+            ..Default::default()
+        }));
+        assert!(is_undetermined_region_error(&undetermined));
+
+        let busy = Error::RegionError(Box::new(errorpb::Error {
+            server_is_busy: Some(errorpb::ServerIsBusy::default()),
+            ..Default::default()
+        }));
+        assert!(!is_undetermined_region_error(&busy));
+        assert!(!is_undetermined_region_error(&Error::StringError(
+            "x".to_owned()
+        )));
+    }
+
+    /// A two-shard plan: the LOWER-index shard fails determinately, the HIGHER-index
+    /// shard reports `UndeterminedResult`. Reproduces the masking hazard the shard
+    /// aggregation guards against.
+    #[derive(Clone)]
+    struct MaskingPlan {
+        idx: usize,
+    }
+
+    #[async_trait]
+    impl Plan for MaskingPlan {
+        type Result = BatchGetResponse;
+
+        async fn execute(&self) -> Result<Self::Result> {
+            if self.idx == 0 {
+                // Determinate, lower index — the error that MUST NOT win.
+                Err(Error::Unimplemented)
+            } else {
+                // Undetermined, higher index.
+                Ok(BatchGetResponse {
+                    region_error: Some(errorpb::Error {
+                        undetermined_result: Some(errorpb::UndeterminedResult::default()),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                })
+            }
+        }
+    }
+
+    impl Shardable for MaskingPlan {
+        type Shard = usize;
+
+        fn shards(
+            &self,
+            _: &Arc<impl crate::pd::PdClient>,
+        ) -> BoxStream<'static, crate::Result<(Self::Shard, RegionWithLeader)>> {
+            Box::pin(stream::iter(vec![
+                Ok((0usize, MockPdClient::region1())),
+                Ok((1usize, MockPdClient::region2())),
+            ]))
+            .boxed()
+        }
+
+        fn apply_shard(&mut self, shard: Self::Shard) {
+            self.idx = shard;
+        }
+
+        fn apply_store(&mut self, _: &crate::store::RegionStore) -> Result<()> {
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn an_undetermined_shard_is_not_masked_by_another_shards_error() {
+        // THE MASKING HAZARD. Shard 0 fails determinately (lower index); shard 1 is
+        // undetermined. First-error-by-index would report shard 0's determinate
+        // failure and hide the fact that shard 1 may have applied — claiming clean
+        // failure for a write that may be durable.
+        // The aggregation must surface the undetermined error instead.
+        let plan = RetryableMultiRegion {
+            inner: MaskingPlan { idx: 0 },
+            pd_client: Arc::new(MockPdClient::default()),
+            backoff: Backoff::no_backoff(),
+            preserve_region_results: false,
+            terminal_on_undetermined: true,
+        };
+        let err = plan.execute().await.unwrap_err();
+        assert!(
+            is_undetermined_region_error(&err),
+            "the undetermined shard must win over the determinate one, got {err:?}"
+        );
+    }
 }