es_entity/operation/batch/
transient.rs

1//! Classifying a probe failure as transient.
2
3/// The Postgres sqlstate of a failure that says nothing about the statement
4/// that hit it, or `None`.
5///
6/// Walks the whole [`source`](std::error::Error::source) chain, so a
7/// [`sqlx::Error`] wrapped several layers deep in a caller's own error type is
8/// still recognised.
9///
10/// - `40P01` — deadlock detected. This transaction was chosen as the victim;
11///   another one made progress.
12/// - `40001` — serialization failure.
13///
14/// Both are properties of the contention, not of the items being probed, so a
15/// bisect re-probes the same range unsplit.
16pub fn retryable_conflict_code(err: &(dyn std::error::Error + 'static)) -> Option<&'static str> {
17    let mut source = Some(err);
18    while let Some(err) = source {
19        if let Some(db) = err
20            .downcast_ref::<sqlx::Error>()
21            .and_then(|err| err.as_database_error())
22        {
23            match db.code().as_deref() {
24                Some("40P01") => return Some("40P01"),
25                Some("40001") => return Some("40001"),
26                _ => {}
27            }
28        }
29        source = err.source();
30    }
31    None
32}
33
34/// [`retryable_conflict_code`] as a predicate.
35pub fn is_retryable_conflict(err: &(dyn std::error::Error + 'static)) -> bool {
36    retryable_conflict_code(err).is_some()
37}
38
39/// Which probe failures are transient, and how many re-probes they may buy.
40///
41/// The default classification is [`is_retryable_conflict`]. Override it to add
42/// error types of your own that describe contention — an optimistic-concurrency
43/// conflict, say — so the search re-probes their ranges too.
44#[derive(Debug, Clone, Copy)]
45pub struct TransientPolicy<P> {
46    /// Returns `true` when a probe failure carries no information about the
47    /// range's contents.
48    pub is_transient: P,
49    /// How many transient re-probes the whole search may take before it is
50    /// abandoned.
51    pub max_retries: usize,
52}
53
54impl<P> TransientPolicy<P> {
55    /// A policy with the default retry allowance.
56    pub fn new(is_transient: P) -> Self {
57        Self {
58            is_transient,
59            max_retries: super::DEFAULT_MAX_TRANSIENT_RETRIES,
60        }
61    }
62
63    /// Overrides the retry allowance.
64    #[must_use]
65    pub fn with_max_retries(self, max_retries: usize) -> Self {
66        Self {
67            max_retries,
68            ..self
69        }
70    }
71}
72
73/// The default classifier, as a plain function so it can be named in a
74/// [`TransientPolicy`] without boxing.
75pub(super) fn sqlstate_is_transient<E: std::error::Error + 'static>(error: &E) -> bool {
76    is_retryable_conflict(error)
77}
es_entity/operation/batch/transient.rs

es_entity/operation/batch/
transient.rs
