persist: hedge slow blob gets to absorb dead-connection stalls

doc/user/data/metrics.yml

@@ -1659,6 +1659,50 @@ metrics:
   - honeycomb
   source: src/persist-client/src/internal/metrics.rs
   visibility: internal
+- name: mz_persist_blob_hedge_armed
+  help: 1 if this process opened a hedge sibling and can hedge when enabled
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedge_errors
+  help: hedge requests (the hedge leg only, not the primary) that completed with an error
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedge_rtt_latency
+  help: roundtrip-time of the most recent successful warm-path liveness gets on the hedge sibling
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedge_warm_errors
+  help: warm-path liveness gets on the hedge sibling that failed or timed out
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedge_won_seconds_bucket
+  help: end-to-end latency of blob gets won by the hedge request
+  labels:
+  - le
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedge_won_seconds_count
+  help: end-to-end latency of blob gets won by the hedge request
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedge_won_seconds_sum
+  help: end-to-end latency of blob gets won by the hedge request
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedges_fired
+  help: blob gets that fired a hedge request
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedges_skipped
+  help: hedge requests not fired for a get that exceeded the hedge delay, by reason
+  labels:
+  - reason
+  source: src/persist/src/metrics.rs
+  visibility: internal
+- name: mz_persist_blob_hedges_won
+  help: blob gets where the hedge request won the race
+  source: src/persist/src/metrics.rs
+  visibility: internal
 - name: mz_persist_cmd_cas_mismatch_count
   help: count of command retries from CaS mismatch
   labels:

misc/python/materialize/mzcompose/__init__.py

@@ -49,7 +49,15 @@
 # a new feature causes benchmarks to become flaky, consider that this can also
 # impact customers' experience and try to find a solution other than disabling
 # the feature here!
-ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS = {}
+ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS = {
+    # Benchmarks measure the intended production configuration. For hedged
+    # blob gets that is the planned enablement state (on, at production
+    # tuning), not the CI-wide coverage tuning below, whose short delay
+    # would add duplicate fetches to any measured get slower than it.
+    "persist_blob_hedged_get_enabled": "true",
+    "persist_blob_hedged_get_delay": "2s",
+    "persist_blob_hedged_get_budget_ratio": "0.01",
+}
 
 
 def sanitizer_enabled() -> bool:
@@ -206,6 +214,20 @@ def get_variable_system_parameters(
         VariableSystemParameter(
             "persist_source_fetch_concurrency", "1", ["1", "2", "8", "16"]
         ),
+        VariableSystemParameter(
+            "persist_blob_hedged_get_enabled", "true", ["true", "false"]
+        ),
+        # 10ms (vs the 2s production default) makes hedges actually fire in
+        # every CI run; 0s makes every blob get hedge under randomized seeds.
+        VariableSystemParameter(
+            "persist_blob_hedged_get_delay", "10ms", ["0s", "10ms", "2s"]
+        ),
+        # The production ratio: with the 10ms delay above, a full refill
+        # would hedge nearly every get and double CI blob traffic. The 1.0
+        # variant lets randomized runs pair a full budget with delay=0s.
+        VariableSystemParameter(
+            "persist_blob_hedged_get_budget_ratio", "0.01", ["1.0", "0.01"]
+        ),
         # -----
         # Others (ordered by name),
         VariableSystemParameter(
@@ -660,6 +682,8 @@ def get_default_system_parameters(
     "persist_blob_operation_attempt_timeout",
     "persist_blob_connect_timeout",
     "persist_blob_read_timeout",
+    "persist_blob_hedged_get_max_concurrent",
+    "persist_blob_hedged_get_warm_interval",
     "persist_stats_collection_enabled",
     "persist_stats_filter_enabled",
     "persist_stats_budget_bytes",

misc/python/materialize/parallel_workload/action.py

@@ -2877,6 +2877,12 @@ def __init__(
             "8",
             "16",
         ]
+        self.flags_with_values["persist_blob_hedged_get_enabled"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["persist_blob_hedged_get_delay"] = [
+            "'0s'",
+            "'10ms'",
+            "'2s'",
+        ]
         self.flags_with_values["enable_variadic_left_join_lowering"] = (
             BOOLEAN_FLAG_VALUES
         )
@@ -3113,6 +3119,9 @@ def __init__(
             # takes effect after a restart. Flipping it here would be a no-op
             # for the running process.
             "enable_adapter_frontend_occ_read_then_write",
+            "persist_blob_hedged_get_budget_ratio",
+            "persist_blob_hedged_get_max_concurrent",
+            "persist_blob_hedged_get_warm_interval",
             "enable_compute_half_join2",
             "enable_mz_join_core",
             "enable_compute_correction_v2",

src/persist-client/src/cache.rs

@@ -24,7 +24,8 @@ use mz_ore::instrument;
 use mz_ore::metrics::MetricsRegistry;
 use mz_ore::task::{AbortOnDropHandle, JoinHandle};
 use mz_ore::url::SensitiveUrl;
-use mz_persist::cfg::{BlobConfig, ConsensusConfig};
+use mz_persist::cfg::{BlobConfig, ConsensusConfig, open_hedge_sibling};
+use mz_persist::hedge::HedgedBlob;
 use mz_persist::location::{
     BLOB_GET_LIVENESS_KEY, Blob, CONSENSUS_HEAD_LIVENESS_KEY, Consensus, ExternalError, Tasked,
     VersionedData,
@@ -249,6 +250,29 @@ impl PersistClientCache {
                     blob.clone().open()
                 })
                 .await;
+                // Hedged gets need a second handle on an isolated connection
+                // pool. Built unconditionally (best-effort): the wrapper
+                // reads its enable flag dynamically per call.
+                //
+                // NOTE: HedgedBlob must stay below Tasked in this stack. Its
+                // race relies on dropping the losing future to cancel the
+                // request in flight, and on hedged gets running to
+                // completion once started (Tasked detaches). A task boundary
+                // between HedgedBlob and the backend would break the former,
+                // and an aborting layer above would slowly leak budget
+                // tokens via the latter.
+                let sibling = open_hedge_sibling(
+                    x.key(),
+                    Box::new(self.cfg.clone()),
+                    self.metrics.s3_blob.clone(),
+                )
+                .await;
+                let blob = Arc::new(HedgedBlob::new(
+                    blob,
+                    sibling,
+                    Arc::clone(&self.cfg.configs),
+                    self.metrics.blob_hedge.clone(),
+                ));
                 let blob = Arc::new(MetricsBlob::new(blob, Arc::clone(&self.metrics)));
                 let blob = Arc::new(Tasked(blob));
                 let task = blob_rtt_latency_task(

src/persist-client/src/internal/metrics.rs

@@ -32,7 +32,7 @@ use mz_ore::stats::histogram_seconds_buckets;
 use mz_persist::location::{
     Blob, BlobMetadata, CaSResult, Consensus, ExternalError, ResultStream, SeqNo, VersionedData,
 };
-use mz_persist::metrics::{ColumnarMetrics, S3BlobMetrics};
+use mz_persist::metrics::{BlobHedgeMetrics, ColumnarMetrics, S3BlobMetrics};
 use mz_persist::retry::RetryStream;
 use mz_persist_types::Codec64;
 use mz_postgres_client::metrics::PostgresClientMetrics;
@@ -110,6 +110,8 @@ pub struct Metrics {
 
     /// Metrics for S3-backed blob implementation
     pub s3_blob: S3BlobMetrics,
+    /// Metrics for hedged blob gets
+    pub blob_hedge: BlobHedgeMetrics,
     /// Metrics for Postgres-backed consensus implementation
     pub postgres_consensus: PostgresClientMetrics,
 
@@ -168,6 +170,7 @@ impl Metrics {
             semaphore: SemaphoreMetrics::new(cfg.clone(), registry.clone()),
             sink: SinkMetrics::new(registry),
             s3_blob,
+            blob_hedge: BlobHedgeMetrics::new(registry),
             postgres_consensus: PostgresClientMetrics::new(registry, "mz_persist"),
             _vecs: vecs,
             _uptime: uptime,

src/persist/src/azure.rs

@@ -310,6 +310,10 @@ fn token_credential() -> Arc<dyn TokenCredential> {
 }
 
 /// Configuration for opening an [AzureBlob].
+///
+/// NOTE: cloning shares the underlying client and therefore its HTTP
+/// connection pool. Connection-pool isolation (as hedged gets require, see
+/// [crate::hedge]) needs a fresh [AzureBlobConfig::new].
 #[derive(Clone, Debug)]
 pub struct AzureBlobConfig {
     metrics: S3BlobMetrics,

src/persist/src/cfg.rs

@@ -25,6 +25,7 @@ use crate::azure::{AzureBlob, AzureBlobConfig};
 use crate::file::{FileBlob, FileBlobConfig};
 #[cfg(feature = "foundationdb")]
 use crate::foundationdb::{FdbConsensus, FdbConsensusConfig};
+use crate::hedge::HedgeSibling;
 use crate::location::{Blob, Consensus, Determinate, ExternalError};
 use crate::mem::{MemBlob, MemBlobConfig, MemConsensus};
 use crate::metrics::S3BlobMetrics;
@@ -33,7 +34,68 @@ use crate::s3::{S3Blob, S3BlobConfig};
 
 /// Adds the full set of all mz_persist `Config`s.
 pub fn all_dyn_configs(configs: ConfigSet) -> ConfigSet {
-    configs.add(&crate::postgres::PG_CONSENSUS_READ_COMMITTED)
+    configs
+        .add(&crate::postgres::PG_CONSENSUS_READ_COMMITTED)
+        .add(&crate::hedge::BLOB_HEDGED_GET_ENABLED)
+        .add(&crate::hedge::BLOB_HEDGED_GET_DELAY)
+        .add(&crate::hedge::BLOB_HEDGED_GET_MAX_CONCURRENT)
+        .add(&crate::hedge::BLOB_HEDGED_GET_BUDGET_RATIO)
+        .add(&crate::hedge::BLOB_HEDGED_GET_WARM_INTERVAL)
+}
+
+/// Opens the sibling handle that [crate::hedge::HedgedBlob] runs hedge
+/// requests on for `url`.
+///
+/// Contract:
+/// - An [HedgeSibling::Isolated] handle observes exactly the same durable
+///   store as a handle opened from the same `url`, but is built from a
+///   scratch client: it shares no HTTP connection pool, DNS state, or
+///   credential chain, so a hedge request on it can never be assigned a
+///   connection the primary's pool has already half-killed.
+/// - Backends where a second open would observe an independent store (mem,
+///   turmoil's simulated store), or that have no connection state to isolate
+///   (file), return [HedgeSibling::SharedWithPrimary] instead.
+/// - Callers must use the handle only for idempotent reads.
+///
+/// Errors opening the sibling degrade to [HedgeSibling::Unavailable] with a
+/// warning rather than failing: persist must come up even if hedging cannot.
+/// A process that hits this keeps hedging unavailable until restart, visible
+/// as `mz_persist_blob_hedges_skipped{reason="unavailable"}` and
+/// `mz_persist_blob_hedge_armed` staying 0.
+pub async fn open_hedge_sibling(
+    url: &SensitiveUrl,
+    knobs: Box<dyn BlobKnobs>,
+    metrics: S3BlobMetrics,
+) -> HedgeSibling {
+    let config = match BlobConfig::try_from(url, knobs, metrics).await {
+        Ok(config) => config,
+        Err(err) => {
+            warn!(
+                "hedged blob gets unavailable, sibling config failed: {}",
+                err
+            );
+            return HedgeSibling::Unavailable;
+        }
+    };
+    match config {
+        // A second S3/Azure config builds its own SDK client and therefore
+        // its own connection pool, with DNS resolved per connect.
+        config @ (BlobConfig::S3(_) | BlobConfig::Azure(_)) => match config.open().await {
+            Ok(blob) => HedgeSibling::Isolated(blob),
+            Err(err) => {
+                warn!("hedged blob gets unavailable, sibling open failed: {}", err);
+                HedgeSibling::Unavailable
+            }
+        },
+        // File has no connection pool to isolate, so a second instance would
+        // buy nothing. A second open of Mem (or of turmoil's simulated
+        // store) would be actively wrong: it creates an INDEPENDENT store,
+        // and a hedged get against a different store can return `Ok(None)`
+        // for data that exists.
+        BlobConfig::File(_) | BlobConfig::Mem(_) => HedgeSibling::SharedWithPrimary,
+        #[cfg(feature = "turmoil")]
+        BlobConfig::Turmoil(_) => HedgeSibling::SharedWithPrimary,
+    }
 }
 
 /// Config for an implementation of [Blob].