diff --git a/doc/developer/design/20260806_hedged_blob_gets.md b/doc/developer/design/20260806_hedged_blob_gets.md new file mode 100644 index 0000000000000..c77736842e0d4 --- /dev/null +++ b/doc/developer/design/20260806_hedged_blob_gets.md @@ -0,0 +1,517 @@ +# Hedged blob gets + +Persist reads batch parts and rollups, including those of the txn-wal txns +shard, through `Blob::get` calls against an object store. A single slow get on a +hot path can stall a whole replica, because everything downstream waits on +that fetch. This document describes a hedging mechanism for those gets: if +a get has not completed within a short delay, persist fires a second +request for the same key on an isolated connection and takes whichever +succeeds first. + +## Goals + +- Cap the freshness impact of the dead-connection failure class (described + below) at roughly the hedge delay plus one normal get, whenever a healthy + connection is reachable. When it is not (both requests stall), behavior + falls back to today's, with `retry_external` (persist's retry wrapper) + as the backstop. +- Change nothing about persist's semantics, error surface, or write paths. + +## Non-Goals + +- Hedging writes, deletes, or lists. Writes and deletes have side effects. + Lists are idempotent like gets but not latency-critical: they serve + usage collection and admin paths, not the dataflow read path that a + stalled get freezes. +- Consensus. Stalls of the CockroachDB-backed consensus layer are a + different failure class and are untouched by this design. +- Improving tail latency in the healthy case. The delay is chosen so that + hedges are rare, and the budget (see Bounding amplification) bounds the + exceptions. + +## The Problem + +Established connections to S3 occasionally die in ways that surface only +after several seconds. The observed pattern is a pooled connection that +stops responding, then delivers a TCP reset 5 to 15 seconds later. The AWS +SDK then marks the connection as unusable ("Connection encountered an issue +and should not be re-used") and the request is retried on a fresh +connection, either by the SDK within the same call or by `retry_external`, +persist's retry-with-backoff wrapper around every external call. That +fresh-connection retry succeeded promptly in every event we examined. The +cost is the hang before the reset, which nothing bounds tightly today. + +Persist's client-side timeouts do not catch the hang. Persist configures +the SDK with `persist_blob_read_timeout` (10s, but it only limits time to +the first response byte), `persist_blob_connect_timeout` (7s, but it only +applies to connection establishment), and a 90s per-attempt timeout that is +far too long to help at freshness timescales. In the events we +characterized, the TCP reset arrived before any of these fired, so those +hung requests never incremented the corresponding timeout counters +(`mz_persist_s3_read_timeouts` and friends). A different, smaller set of +events fleet-wide does trip those counters at a low background rate. + +We characterized three manifestations of the class: + +- A background rate of roughly two connection events per hour per pod. + Almost all are invisible, absorbed by the retry off any hot path. +- A single hung get on the part fetch path of a high-fan-in shard (one + whose data feeds many dataflows, such as the txns shard), which starved + an entire replica for about 10 seconds. Such visible starvations + occurred roughly once per day in the busy environment where we + characterized the class, each a multi-second freshness stall for a + workload with single-digit-second freshness expectations. +- A regionally synchronized burst, in which the table shards of at least 17 + environments stalled 7 to 15 seconds simultaneously (each environment's + single txns shard fanning the stall out to all of its tables) while S3 + itself served normally. + +The evidence says a hedge on a separate connection would have won almost +all of these races. During one event, 58 of 67 storage collections on the +same pod (each a table or source backed by its own shard) stayed at their +normal latency, and the stalled nine were the collections behind the +affected fetch path, so the other established connections were fine at +that instant. +A peer replica of the affected one read the same shard with normal latency +at the same moment. Fleet-wide latency histograms showed no elevated gets +during the regional burst. And every observed recovery was itself a +successful fresh-connection retry, which is exactly a hedge that fired +late. One caveat feeds the design below: during the correlated bursts the +fleet's connect-timeout counters also stepped, so brand-new connections +were sometimes slow to establish mid-event, and a useful hedge therefore +needs an already-established connection waiting. + +Hedging slow requests is what the object-store vendors recommend. AWS's +S3 performance guidance advises aggressively retrying slow operations on +a new connection with a fresh DNS lookup, and suggests a 2 second +threshold for small gets. Google Cloud Storage documentation recommends +hedged requests for latency-sensitive applications, and Microsoft ships +a first-class hedging strategy in its official resilience stack. + +## The Design + +### The wrapper + +`HedgedBlob` (`src/persist/src/hedge.rs`) is a decorator implementing the +`Blob` trait. It holds two `Arc` handles on the same durable +store: the primary, and a second handle we call the sibling, on which hedge +requests run. Only `get` is hedged. Every other trait method forwards to +the primary untouched. + +Because the wrapper works against the `Blob` trait object, one +implementation covers every backend: S3, GCS through the S3 interop mode, +and Azure. + +### The race + +A hedged get runs up to two requests, called legs below. The primary leg +starts immediately and races a timer set to the configured delay. If the +primary completes first (the full blob, not just its first byte: `Blob::get` +returns a complete result), its result is returned verbatim, success or +error. +A fast error therefore still takes today's path into `retry_external`'s +retries. Hedging targets hangs, not failures. + +If the delay elapses, the hedge leg fires on the sibling, subject to the +two admission guards described under Bounding amplification below. If the +guards refuse, the get simply continues waiting on the primary, exactly +today's behavior. + +Once both legs are running, the first success wins and the losing future is +dropped, which cancels its request. An error on one leg does not end the +race, and the two error cases are asymmetric. If the hedge leg errors, the +get keeps waiting on the primary alone, so a fast-failing hedge cannot turn +a get that was about to succeed into an error. If the primary errors after +the hedge fired, the get waits a delay-sized grace window for the hedge and +then returns the primary's error, so a slow hedge cannot hold the get past +the point where handing the error to `retry_external` is the better move. +Either way, when the get does fail, the caller sees the primary's error +object, unchanged. This matters because `ExternalError::is_timeout` matches +on the error string, so even attaching the hedge's error as context could +change how a caller classifies the failure. + +The race operates entirely within one `retry_external` attempt, leaving +the existing retry machinery untouched as the backstop. + +### Where the wrapper sits + +`PersistClientCache::open_blob` (in `mz_persist_client::cache`) composes +the production blob stack. With `HedgedBlob` in place, from the outside in: + +- `BlobMemCache`, an in-memory cache of recently fetched blobs, +- `Tasked`, which runs each call in a spawned tokio task so that + Timely-polled callers cannot stall blob calls by polling them lazily, +- `MetricsBlob`, which records per-operation metrics, +- `HedgedBlob`, +- the backend (`S3Blob` or `AzureBlob`). + +Each neighbor constrains that position ("below" meaning closer to the +backend). `HedgedBlob` must be below `BlobMemCache` so that cache hits +never hedge. It must be below `MetricsBlob` so that a hedged get is +recorded as one `blob_get` operation at the winner's latency, rather than +as two operations, one of which would carry the loser's hang into the very +latency histogram used to detect this failure class. + +It must also be below `Tasked`, so that the race runs inside the spawned +task rather than around it. That buys two properties. First, cancellation +is real: dropping the losing leg aborts its request in flight, whereas a +`Tasked` boundary between the race and the backend would merely detach a +spawned task that keeps running, holding its socket and buffers. Second, +the hedge timer is driven by tokio. Outside the spawned task it would be +polled by Timely operators, which can poll futures arbitrarily late, and a +timer that fires late is a hedge that does not fire. + +### Pool isolation + +The sibling shares no connection state with the primary, so a hedge +cannot be handed a connection dying in the same event that stalled the +primary. The full case for isolation, including what a shared pool +forecloses, is under Alternatives. Three facts make the isolation work. + +First, isolation cannot come from cloning: the AWS SDK client embedded in +`S3BlobConfig` is reference-counted, so cloning the config shares the HTTP +connection pool. The sibling is instead built from a second, from-scratch +config. Every `mz_aws_util::defaults()` call installs a fresh hyper client +with its own pool, and hyper resolves DNS per connection establishment, +which also satisfies the fresh-DNS guidance. Azure behaves the same way, +since `AzureBlobConfig::new` builds its own client per call. (One shared +piece remains: the SDK keeps process-global retry bookkeeping per service +and region. It never gates first attempts, only the SDK's own internal +retries.) + +Second, a from-scratch config also carries its own credential provider +chain, so the sibling authenticates independently of the primary. That is +a small robustness gain and a new failure mode to watch (see +Observability). + +Third, per-backend knowledge lives in one place, `open_hedge_sibling` in +`mz_persist::cfg`. For S3 and Azure it opens the independent second +handle. For file, mem, and turmoil (persist's deterministic +network-simulation backend for tests) it returns the primary instance +itself, because a second open of those would observe an independent store: +a hedged get against a different store can return `Ok(None)`, a legitimate +success, for data that exists, winning races and reporting live batch +parts as missing. Sharing the instance eliminates the hazard by +construction while still exercising the race path in tests. + +Opening the sibling is best-effort. On failure, persist logs a warning and +runs without hedging for the process lifetime, visible in metrics as +`hedges_skipped{reason="unavailable"}` and as `hedge_armed` staying 0. +Persist startup must not regress for this feature. + +### Keeping the sibling warm + +A cold hedge can pay a connection handshake of up to the 7 second connect +timeout, and the burst evidence above shows fresh connects are sometimes +slow exactly during the correlated events. So the sibling's pool must hold +already-established connections. A background task issues concurrent +liveness gets on the sibling (fetching a reserved key that never exists, so +each ping is a cheap not-found response) immediately at startup and then +every 20 seconds, well inside hyper's eviction of connections idle for 90 +seconds. (NOTE: that 90 is coincidentally equal to, and unrelated to, the +SDK's 90 second attempt timeout.) The number of concurrent pings follows +the hedge concurrency cap, because HTTP/1.1 allows one in-flight request +per connection: N concurrent pings force N warm sockets, so hedges the cap +admits are normally served warm. (A raised cap grows the warm pool at the +next cycle.) + +Each warm cycle is bounded by a timeout: an unbounded hung ping would block +warming past the idle eviction, going cold exactly during the correlated +events warming exists for, and the timeout also drops the hung request, +which closes its dying socket. A successful cycle records its round-trip +time in a gauge. A failed or timed-out cycle increments a warm-error +counter and leaves the gauge alone, so a fast-failing sibling (for +example, one whose credentials rotted) cannot masquerade as a fast healthy +one. + +Because the killed ping closes its socket, the warm interval bounds how +long a dead sibling socket lingers in the pool: about two intervals (one +until a ping lands on it, one more until the cycle timeout kills it). +Until that purge a hedge can check out the dead socket, hang until the +read timeout while contributing nothing, and hold its concurrency slot +the whole time, so an event's first hedges can pin both default slots +and leave later gets unhedged with +`hedges_skipped{reason="concurrency"}` climbing. At 20 seconds, hedging +therefore survives a correlated event that also hits the sibling's +sockets only if the sibling escaped it. + +A single-digit interval would cut that exposure to a few seconds and let +the pool heal mid-event, and its restarted handshakes probe better than +one patient handshake: each resets TCP's exponential SYN backoff (about +one probe per second) and re-resolves DNS, which can retarget a rotated +S3 front-end address once the record's short TTL lapses. Nothing waits +on a warming handshake, so aborting a viable-but-slow one costs nothing. +The interval nevertheless stays at 20 seconds on cost: there the warmer +is a modest fraction of the fleet's organic blob gets, while at a few +seconds it would rival or exceed the fleet's entire organic blob-get +request volume. The evidence that would justify that spend is correlated +events still visible in freshness after enablement, hedges erroring on +dead sibling sockets, concurrency skips clustering at event times, and +the warm-error counter spiking alongside. The interval is a dyncfg, so +lowering it needs no release. + +The warmer runs only while hedging is enabled, re-checking the flag at its +normal cadence while idle. While hedging is disabled, the sibling is +therefore fully idle: no requests, no credential refreshes (SDK providers +refresh lazily, on use), and its startup sockets idle out. The trade-off +is a short cold window after a runtime enablement: until the warmer's +first cycle, up to one warm interval plus a handshake later, a hedge can +land on an empty pool and pay a cold connect (bounded by the connect +timeout). Hedges in that window are merely no better than no hedge, never +worse, since the primary keeps racing regardless. Setting the warm +interval to zero stops warmer traffic while keeping hedging on. + +### Bounding amplification + +Two admission guards bound how much extra load hedging can create, each +protecting a different resource. Every hedge must pass both. + +A concurrency cap (default 2) bounds memory. Batch parts can be up to the +128 MiB blob target size, and the hedge leg's buffers are invisible to the +fetch-path memory accounting (the fetch semaphore sizes its memory budget +before the blob layer runs, and on cc replicas that memory budget is tied to +the process memory limit), so the cap is what keeps unaccounted transient +buffers to about two parts. The warmer holds one warm socket per admitted +hedge (see above), so the cap also sizes the warm pool. + +A token bucket, called the budget, bounds request rate and egress. The +bucket holds up to 32 tokens, a hedge costs one token, and every completed +get adds `budget_ratio` tokens (default 0.01), so under sustained slowness +hedging settles at one percent of gets. Without it, a store-wide brownout +that pushes every get past the delay would deterministically double the +request rate onto an already-degraded dependency. The same guard keeps +large gets that legitimately exceed the delay (a 128 MiB part on a +bandwidth-constrained pod) from settling into permanent double egress. +The bucket starts full, so a low-traffic process can still hedge the +rare event that motivates the feature, and 32 is an order of magnitude +above the handful of rescues one event needs per process. The blind spot +mirrors the protection: where more than one percent of gets are +legitimately slow, the drained bucket also refuses the occasional +genuine dead-connection hang, visible as `hedges_skipped{reason="budget"}`. + +### Configuration + +Five dyncfgs, all readable per call so LaunchDarkly changes apply live: + +| Name | Default | Purpose | +| --- | --- | --- | +| `persist_blob_hedged_get_enabled` | `false` | Master switch for hedging. | +| `persist_blob_hedged_get_delay` | `2s` | Time in flight before the hedge fires. | +| `persist_blob_hedged_get_max_concurrent` | `2` | Memory bound. | +| `persist_blob_hedged_get_budget_ratio` | `0.01` | Rate bound. | +| `persist_blob_hedged_get_warm_interval` | `20s` | Warm interval. `0` disables the warmer without disabling hedging. | + +The bucket capacity (32) is hard-coded: it is the bucket's shape rather +than an operational lever, and fewer knobs means less to review and fewer +LaunchDarkly entries. Note that `budget_ratio = 0` is not a kill switch, +because the bucket starts full. `enabled` is the kill switch for hedging. + +Retuning the delay trades two effects. Lowering it moves the rescue +floor (a hedge win arrives no +earlier than the delay plus one round trip) but grows the false-fire +population: the race triggers on full-blob completion, not first byte, +so the gets that legitimately outlast the delay are large parts on +constrained bandwidth, and each false fire both duplicates that part's +egress and spends a budget token, so a lower delay can drain the bucket +with healthy traffic and leave a real event refused. Raising it has a +wide plateau in the other direction: a rescue stays useful as long as +the delay plus one normal get fits the freshness target, which holds up +to several seconds. Any retune should start by re-measuring the +would-be fire rate at the candidate delay (the rate in Rollout is +workload-dependent, dominated by part sizes and pod bandwidth). + +What the kill switch does not cover: `enabled = false` leaves the +sibling idle (see Keeping the sibling warm), but its client and +credential chain are constructed at process start regardless, which is +what keeps enablement restart-free. The disabled standing costs are one +extra SDK client's memory per process, one extra credential resolution, +and a doubled blob-open (including the backend's own health-check get) +at startup. Removing the sibling machinery is a rollback, not a flag +flip. + +### Testing + +Unit tests drive the race deterministically on tokio's paused test clock: + +- the hedge winning, with the get resolving at exactly the hedge delay + while the primary is still pending (cancellation), +- the primary winning after the hedge fired, +- a fast-failing hedge not failing a get whose primary later succeeds, +- the primary failing after the hedge fired, with the hedge winning inside + the grace window in one test and the grace window expiring in another, +- both legs failing, returning the primary's error verbatim, +- budget exhaustion and refill, the concurrency cap, and release of the + concurrency slot when a hedged get is dropped mid-race, +- the warmer's cadence, and its absence for a same-instance sibling. + +The existing `Blob` conformance suite runs against `HedgedBlob` with the +delay at zero and the primary's gets artificially slowed, so a hedge fires +and wins on every get through the full set/get/delete/list matrix over one +shared store (the test asserts hedges actually fired and won). An +additional test runs the same suite against real S3 with two genuinely +independent clients. Like the existing external-storage S3 tests it is +ignored by default and runs on demand against the external test bucket, so +true pool isolation has manual but not continuous coverage. + +In CI the feature is on everywhere (repo convention for new feature +flags, wired through the mzcompose system-parameter defaults) at a 10ms +delay rather than the production 2s, so hedges genuinely fire in every +run. CI's system-parameters randomization mode can additionally set the +delay to 0s, making every blob get hedge under real workloads, and the +parallel-workload suite flips `enabled` and the delay mid-workload. +Benchmarks are the exception: the harnesses pin the planned production +configuration (hedging on, production delay and budget) through the +shared benchmarking parameter overrides, so benchmark history +accumulates against what production will run. + +## Correctness + +Racing two gets is sound because blobs are write-once and modify-never, and +the store is linearizable (`BlobMemCache` already relies on both). Both +legs start after the caller invokes `get`, so any result either leg returns +is a correct answer for some point during the call, including `Ok(None)` +for a key that does not exist. The one way this argument breaks is if the +two handles observe different stores, which is exactly what +`open_hedge_sibling`'s per-backend contract prevents. + +Write-once is a statement about blob contents, not key existence: a +concurrent delete can legitimately change the answer between the two legs, +and the hedge leg can observe the store up to the delay later than the +primary would have. What makes that irrelevant in practice is the same +thing that protects sequential gets today: persist only deletes blobs that +no live reader can reference, enforced by seqno leases. (This is also why +a runtime cross-check that both legs agree was rejected, see Alternatives.) + +The error surface is unchanged: callers see a success or the primary's +error exactly as they would have without hedging, so the error text, the +timeout heuristic, and the determinate versus indeterminate +classification are all preserved. + +## Observability + +New metrics, all prefixed `mz_persist_blob_` (elided in prose below): + +| Metric | Type | Meaning | +| --- | --- | --- | +| `hedges_fired` | counter | Gets that fired a hedge request. | +| `hedges_won` | counter | Gets the hedge request won. | +| `hedge_won_seconds` | histogram | End-to-end latency of gets the hedge won. | +| `hedges_skipped` | counter, by `reason` | Hedges refused: `budget`, `concurrency`, or `unavailable`. | +| `hedge_errors` | counter | Hedge legs (not primaries) that completed with an error. | +| `hedge_warm_errors` | counter | Warm cycles that failed or timed out. | +| `hedge_armed` | gauge | 1 if the process opened a sibling and can hedge. | +| `hedge_rtt_latency` | gauge | Round-trip time of the last successful warm cycle. | + +Enabling hedging makes the old detection signals go quiet. The hung +primary is cancelled at the delay, so the background increments of +`mz_persist_s3_read_timeouts`, the SDK's connection-poisoning log lines, +and spikes in the persist-observed blob round-trip gauge +(`mz_persist_external_rtt_latency`, which measures through the hedged +stack) largely stop occurring for this class. `hedges_won` replaces them +as the detector. + +Two further notes. On accounting: the pre-existing S3 request counters +(`mz_persist_s3_operations`, for example its `get_part` label) count both +legs of a hedged get plus the warmer's pings, so they can exceed the logical +`blob_get` operation count. On health: `hedge_warm_errors` and +`hedge_errors` are the signals that the sibling's independent credential +chain has rotted, which would otherwise silently turn the feature into a +no-op. Both only move while hedging is enabled, so expect any rot to +surface within the first warm cycles after enablement. `hedge_armed` only certifies +that the sibling opened at process start. + +## Alternatives + +### Hedging on the primary's pool + +Hedging on the primary's own connection pool would need no sibling +machinery, and in the dominant event shape (one connection dies while +its pool-mates stay healthy, which is why the same-pool retry succeeds +promptly today) it would usually work. Isolation is for the residual +minority of correlated, path-scoped events, where connection kills +clustered within a second on one pod and fresh connects stalled +fleet-wide, so a shared pool is most likely to hand the hedge a sick +connection exactly when the hedge matters most. A shared pool also +forecloses levers that matter regardless of fate sharing: it pins the +hedge to the pool's cached DNS answer and front-end address, lets a +checkout queue behind the very jam the hung get is part of, and cannot +be kept warm without distorting the primary's pool. + +### Just lowering the client timeouts + +Lowering the client timeouts instead of hedging was the 2023 answer to +this class (connect and read were cut from 30 and 60 seconds to today's +7 and 10), and the residual hang is what that lever left behind. +Tightening further runs into the structural limits The Problem lists, +and a mid-body hang stays bounded only by the 90 second attempt timeout, +which cannot come down without killing legitimately long large-part +fetches. A timeout is also sequential and forces a trade: it converts a +possibly-about-to-succeed request into an error, pays backoff and +restart, and retries on the same pool with no fresh connection or DNS +resolution guaranteed, so its threshold must stay conservative. A hedge +is concurrent, so firing early costs a bounded duplicate rather than a +lost success, which is why hedging affords a 2 second trigger where a 2 +second timeout could not. + +### Per-chunk hedging + +Hedging per multipart chunk inside `S3Blob::get` is a possible refinement: +`S3Blob::get` already fetches large blobs in 8 MiB parts, so a per-chunk +hedge would duplicate one chunk instead of the whole blob and could trigger +on missing first bytes rather than wall clock. It is S3-only, duplicates +the race logic per backend, and is a natural follow-up once the generic +version has proven itself, not the first version. + +### A pool-less sibling + +Disabling pooling on the sibling client (max idle connections of zero) +would guarantee isolation and fresh DNS with no warmer machinery, but it +puts a full TCP and TLS handshake on the critical path of every hedge, +which is up to 7 seconds during exactly the correlated events the feature +targets. + +### A both-legs debug mode + +A debug mode that awaits both legs and asserts they agree was rejected as +unsound, not merely unnecessary: as Correctness explains, two gets of the +same key are only required to agree for keys protected by a live lease, so +the assertion would misfire on legitimate delete races. + +### A hedging-aware Blob trait + +Extending the `Blob` trait with a hedging-aware method was rejected because +it pushes a transport concern into a five-method correctness-critical trait +that most implementations have no answer for. + +### Hedging in maelstrom + +A configuration for maelstrom was considered and deferred. Maelstrom's +blob is neither S3 nor Azure, so its sibling would be the same instance, +and `UnreliableBlob` injects errors but not delays, so hedges would +essentially never fire. The +configuration becomes worthwhile together with delay injection in +`UnreliableBlob`. + +## Rollout + +The feature ships dark: the code path is present everywhere, but hedging +is off by default in production (and on in CI, see Testing) until enabled +at runtime. Enablement happens per environment through LaunchDarkly, which +requires the LD flags (at minimum `persist_blob_hedged_get_enabled` and +the delay) to be created first: until then the dyncfgs exist only with +their code defaults. On enablement, +expect the old detection signals to fade (see Observability), +`hedges_fired` to run at a low background rate (measured on a busy +reference environment: gets over the 2s delay ran at roughly 4 per day +on the busiest replica and a few per hour environment-wide, so +single-digit fires per day per process is the expected order, with +hydration bursts of large parts as the exception the budget caps), and +`hedges_won` to step where the old signals would have fired. +Success is the dead-connection class becoming sub-breach: each +instance should cap near the hedge delay plus one normal get, and the +roughly-daily visible freshness stalls attributed to the class in a busy +environment should disappear from per-minute freshness data. The known +residual is the case where both legs stall (a correlated event defeating +the warm pool, or a drained budget), which falls back to today's behavior +and stays visible as `hedge_won_seconds` outliers and +`hedges_skipped` increments. diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index fee34ab44368d..d269dca4d9ca7 100644 --- a/doc/user/data/metrics.yml +++ b/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: diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 322c6f05cf0fa..82bfd75d0224b 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/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", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 58de840777d63..b2b90a9f2bfa8 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/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", diff --git a/src/persist-client/src/cache.rs b/src/persist-client/src/cache.rs index 2f7707cf59091..7604dc01f23f0 100644 --- a/src/persist-client/src/cache.rs +++ b/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( diff --git a/src/persist-client/src/internal/metrics.rs b/src/persist-client/src/internal/metrics.rs index c60fe7b818eac..e21908f9e5223 100644 --- a/src/persist-client/src/internal/metrics.rs +++ b/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, diff --git a/src/persist/src/azure.rs b/src/persist/src/azure.rs index f48658c789a6b..c0deb6b49303f 100644 --- a/src/persist/src/azure.rs +++ b/src/persist/src/azure.rs @@ -310,6 +310,10 @@ fn token_credential() -> Arc { } /// 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, diff --git a/src/persist/src/cfg.rs b/src/persist/src/cfg.rs index cd598d314be1e..67b1764d33220 100644 --- a/src/persist/src/cfg.rs +++ b/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, + 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]. diff --git a/src/persist/src/hedge.rs b/src/persist/src/hedge.rs new file mode 100644 index 0000000000000..86daf82a1cc2c --- /dev/null +++ b/src/persist/src/hedge.rs @@ -0,0 +1,879 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! A [Blob] decorator that hedges slow `get` requests. +//! +//! Established connections to the blob store occasionally die in ways that +//! surface only after multiple seconds (a TCP reset after a hang, or a black +//! hole), well before any client timeout fires. A `get` riding such a +//! connection stalls everything downstream of it, while other connections on +//! the same process serve the same store normally. The mitigation, endorsed by +//! the major object stores for idempotent reads, is a hedged request: if the +//! first `get` has not completed within a short delay, race a second one on a +//! connection the first cannot have poisoned, and take whichever succeeds +//! first. +//! +//! Only `get` is hedged. All other [Blob] methods are forwarded to the +//! primary handle untouched: writes, deletes, and restores have side +//! effects, and lists are not latency-critical enough to justify racing a +//! streaming interface. Extending hedging to any of them is forbidden. +//! +//! The hedge handle must not share a connection pool (or DNS state) with the +//! primary, otherwise the hedge can be handed a connection dying in the same +//! event that stalled the primary, exactly when a hedge matters most. See +//! [crate::cfg::open_hedge_sibling] for how that isolation is constructed +//! per backend. +//! +//! Hedging operates within a single `retry_external` attempt, before any +//! failure surfaces. The retrying in `retry_external`, which is what +//! recovers this failure class when hedging is off (at the cost of the full +//! hang), stays untouched as the backstop. The governing principle for +//! every race below: the primary's outcome is authoritative, and the hedge +//! is opportunistic, invisible unless it wins. Nothing here assumes callers +//! retry: every branch of the race degrades to the outcome of the un-hedged +//! get, delayed by at most one hedge delay, so a caller that treats a get +//! error as fatal sees the same error it would have seen without hedging, +//! at most that one delay later. +//! +//! NOTE: enabling hedging largely suppresses the old fingerprints of the +//! dead-connection class (client timeout counters, the SDK's +//! connection-poisoning log lines), because the hung request is cancelled +//! before they trigger. The `hedges_won` counter is the replacement signal. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use async_trait::async_trait; +use bytes::Bytes; +use futures_util::future::{Either, select}; +use mz_dyncfg::{Config, ConfigSet}; +use mz_ore::bytes::SegmentedBytes; +use mz_ore::cast::CastLossy; +use mz_ore::task::AbortOnDropHandle; +use tracing::{debug, warn}; + +use crate::location::{BLOB_GET_LIVENESS_KEY, Blob, BlobMetadata, ExternalError}; +use crate::metrics::BlobHedgeMetrics; + +pub(crate) const BLOB_HEDGED_GET_ENABLED: Config = Config::new( + "persist_blob_hedged_get_enabled", + false, + "Whether to hedge slow blob gets with a second request on a separate \ + connection pool (Materialize).", +); + +pub(crate) const BLOB_HEDGED_GET_DELAY: Config = Config::new( + "persist_blob_hedged_get_delay", + Duration::from_secs(2), + "How long a blob get may be in flight before a hedge request is fired \ + (Materialize).", +); + +pub(crate) const BLOB_HEDGED_GET_MAX_CONCURRENT: Config = Config::new( + "persist_blob_hedged_get_max_concurrent", + // Bounds the extra in-flight bytes (which the fetch memory semaphore + // cannot see) to about two batch parts. The warmer holds this many + // sockets open, so every admitted hedge can be served warm. + 2, + "Maximum concurrent hedge requests per blob handle, bounding the memory \ + held by raced gets and the number of warm sockets (Materialize).", +); + +pub(crate) const BLOB_HEDGED_GET_BUDGET_RATIO: Config = Config::new( + "persist_blob_hedged_get_budget_ratio", + 0.01, + "Long-run bound on hedge requests as a fraction of blob gets \ + (Materialize).", +); + +// NOTE: the warmer only runs while hedging is enabled, so `enabled` stops +// its traffic too. Setting this knob to 0 additionally stops the warmer +// while keeping hedging on, which is why it must be changeable at runtime. +pub(crate) const BLOB_HEDGED_GET_WARM_INTERVAL: Config = Config::new( + "persist_blob_hedged_get_warm_interval", + Duration::from_secs(20), + "How often to issue liveness gets that keep the hedge connection pool \ + warm, 0 disables warming without disabling hedging (Materialize).", +); + +/// The cost of one hedge in bucket tokens. Micro-token granularity keeps +/// small `budget_ratio` values (down to 1e-6) from rounding to "never +/// refill". +const HEDGE_COST_MICRO_TOKENS: u64 = 1_000_000; + +/// Token-bucket capacity: 32 hedges. +/// +/// The bucket's shape, not an operational lever: the tuning lever is +/// `persist_blob_hedged_get_budget_ratio` and the kill switch is +/// `persist_blob_hedged_get_enabled`. NOTE: because the bucket starts full, +/// `budget_ratio = 0` still permits ~32 banked hedges before draining. It is +/// not an instant stop, `enabled` is. +const BUDGET_BURST_MICRO_TOKENS: u64 = 32 * HEDGE_COST_MICRO_TOKENS; + +/// Why a hedge was not fired for a get that exceeded the delay. +enum HedgeRefused { + Concurrency, + Budget, +} + +/// Bounds hedge amplification with two independent guards: the concurrency +/// cap bounds memory held by raced gets, the token bucket bounds long-run +/// request-rate/egress amplification (e.g. a store-wide brownout making +/// every get slow, or large gets that legitimately exceed the delay, must +/// not settle into hedging every request). +#[derive(Debug)] +struct HedgeBudget { + concurrent: AtomicUsize, + micro_tokens: AtomicU64, +} + +impl HedgeBudget { + fn new() -> Self { + HedgeBudget { + concurrent: AtomicUsize::new(0), + micro_tokens: AtomicU64::new(BUDGET_BURST_MICRO_TOKENS), + } + } + + /// Attempts to acquire both guards. The returned guard releases the + /// concurrency slot on drop. Spent tokens come back only via + /// [HedgeBudget::replenish]. + fn try_acquire(&self, max_concurrent: usize) -> Result, HedgeRefused> { + let got_slot = self + .concurrent + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + (c < max_concurrent).then_some(c + 1) + }) + .is_ok(); + if !got_slot { + return Err(HedgeRefused::Concurrency); + } + // Constructed before the token take so its drop releases the slot + // on the budget-refusal path. + let guard = HedgeGuard(self); + let took_token = self + .micro_tokens + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |t| { + t.checked_sub(HEDGE_COST_MICRO_TOKENS) + }) + .is_ok(); + if took_token { + Ok(guard) + } else { + Err(HedgeRefused::Budget) + } + } + + /// Adds `ratio` tokens, called once per completed get (hedged or not), + /// so under sustained slowness hedging settles at `ratio` of traffic. + fn replenish(&self, ratio: f64) { + let add = u64::cast_lossy(ratio.clamp(0.0, 1.0) * f64::cast_lossy(HEDGE_COST_MICRO_TOKENS)); + if add == 0 { + return; + } + let _ = self + .micro_tokens + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |t| { + Some((t + add).min(BUDGET_BURST_MICRO_TOKENS)) + }); + } +} + +struct HedgeGuard<'a>(&'a HedgeBudget); + +impl Drop for HedgeGuard<'_> { + fn drop(&mut self) { + self.0.concurrent.fetch_sub(1, Ordering::Relaxed); + } +} + +/// The sibling handle a [HedgedBlob] runs hedge requests on, produced by +/// [crate::cfg::open_hedge_sibling]. +#[derive(Debug)] +pub enum HedgeSibling { + /// A handle onto the same durable store with fully separate connection + /// state, kept warm by the wrapper. + Isolated(Arc), + /// The backend has no connection state to isolate (or a second open + /// would observe a different store): hedge on the primary instance + /// itself, with nothing to warm. + SharedWithPrimary, + /// Opening the sibling failed: hedging is unavailable for this process + /// lifetime. + Unavailable, +} + +/// A [Blob] decorator that hedges slow `get` requests, per the module docs. +#[derive(Debug)] +pub struct HedgedBlob { + primary: Arc, + /// The handle hedge requests run on. `None` = hedging unavailable, + /// visible as `hedges_skipped{reason="unavailable"}`. + hedge: Option>, + cfg: Arc, + metrics: BlobHedgeMetrics, + budget: HedgeBudget, + _warmer: Option>, +} + +/// Keeps the sibling's connection pool warm with periodic concurrent +/// liveness gets, while hedging is enabled. A cold hedge can stall up to +/// the connect timeout, which during correlated connection events is +/// exactly when it must not. While hedging is disabled the warmer idles +/// and the sibling sees no traffic at all, so a freshly enabled flag can +/// find a cold pool for up to one warm interval plus a handshake. Hedges +/// in that window are merely no better than no hedge, never worse. +fn spawn_warmer( + hedge: Arc, + cfg: Arc, + metrics: BlobHedgeMetrics, +) -> AbortOnDropHandle<()> { + mz_ore::task::spawn(|| "persist::blob_hedge_warmer", async move { + loop { + let interval = BLOB_HEDGED_GET_WARM_INTERVAL.get(&cfg); + if !BLOB_HEDGED_GET_ENABLED.get(&cfg) || interval == Duration::ZERO { + // Nothing to keep warm. Re-check at the configured cadence + // (or its default while warming is set to 0), so a dyncfg + // flip takes effect without a restart. + let recheck = if interval == Duration::ZERO { + *BLOB_HEDGED_GET_WARM_INTERVAL.default() + } else { + interval + }; + tokio::time::sleep(recheck).await; + continue; + } + // Ping first, sleep second, so the pool is warm from process + // start. As many concurrent pings as hedges can run at once + // (HTTP/1.1 allows one in-flight request per connection, so N + // concurrent pings force N warm sockets). + let start = Instant::now(); + let sockets = BLOB_HEDGED_GET_MAX_CONCURRENT.get(&cfg); + let pings = (0..sockets).map(|_| hedge.get(BLOB_GET_LIVENESS_KEY)); + let pings = futures_util::future::join_all(pings); + // Bound the cycle: an unbounded hung ping would block warming + // past hyper's pool idle eviction, going cold exactly during the + // correlated events warming exists for. The timeout also drops + // the hung request, which closes its dying socket. + match tokio::time::timeout(interval, pings).await { + Ok(results) if results.iter().all(|r| r.is_ok()) => { + metrics.rtt_latency.set(start.elapsed().as_secs_f64()); + } + Ok(_) | Err(_) => { + // A failing or hung warm path means hedges cannot be + // trusted to be fast. Surface it, and do not update the + // gauge: a fast failure must not report as a fast + // healthy path. + metrics.warm_errors.inc(); + } + } + tokio::time::sleep(interval).await; + } + }) + .abort_on_drop() +} + +impl HedgedBlob { + /// Returns a new [HedgedBlob]. + /// + /// Must be called from within a tokio runtime: it spawns the sibling + /// warming task. + pub fn new( + primary: Arc, + sibling: HedgeSibling, + cfg: Arc, + metrics: BlobHedgeMetrics, + ) -> HedgedBlob { + let (hedge, warmer) = match sibling { + HedgeSibling::Isolated(h) => { + let warmer = spawn_warmer(Arc::clone(&h), Arc::clone(&cfg), metrics.clone()); + (Some(h), Some(warmer)) + } + HedgeSibling::SharedWithPrimary => (Some(Arc::clone(&primary)), None), + HedgeSibling::Unavailable => (None, None), + }; + metrics.armed.set(i64::from(hedge.is_some())); + HedgedBlob { + primary, + hedge, + cfg, + metrics, + budget: HedgeBudget::new(), + _warmer: warmer, + } + } + + /// Returns the sibling handle and a budget guard, or `None` (having + /// already recorded why) if this get must not hedge. + fn admit(&self) -> Option<(&Arc, HedgeGuard<'_>)> { + let Some(hedge_blob) = &self.hedge else { + self.metrics.skipped_unavailable.inc(); + return None; + }; + match self + .budget + .try_acquire(BLOB_HEDGED_GET_MAX_CONCURRENT.get(&self.cfg)) + { + Ok(guard) => Some((hedge_blob, guard)), + Err(HedgeRefused::Concurrency) => { + self.metrics.skipped_concurrency.inc(); + None + } + Err(HedgeRefused::Budget) => { + self.metrics.skipped_budget.inc(); + None + } + } + } + + fn record_win(&self, key: &str, start: Instant) { + self.metrics.won.inc(); + self.metrics + .won_seconds + .observe(start.elapsed().as_secs_f64()); + debug!(%key, elapsed = ?start.elapsed(), "blob get won by hedge request"); + } + + async fn get_hedged(&self, key: &str) -> Result, ExternalError> { + let start = Instant::now(); + let delay = BLOB_HEDGED_GET_DELAY.get(&self.cfg); + let mut primary = std::pin::pin!(self.primary.get(key)); + // NOTE: `Timeout` polls the wrapped future before checking the + // deadline, so a primary that is ready exactly at the delay + // boundary wins here without firing a hedge. A deadline-first + // combinator would not be incorrect, just wasteful: it would fire + // a redundant hedge (spending budget and skewing metrics) whenever + // the primary completes right at the boundary. + if let Ok(res) = tokio::time::timeout(delay, primary.as_mut()).await { + // A fast error is returned verbatim: hedging targets hangs, + // not failures. + return res; + } + let Some((hedge_blob, guard)) = self.admit() else { + return primary.await; + }; + self.metrics.fired.inc(); + let mut hedge = std::pin::pin!(hedge_blob.get(key)); + // The losing future is dropped, which cancels the request in flight + // (there is no task boundary between here and the backend). An error + // on one leg does not end the race: the slow leg is expected to be a + // hung request, and a fast-failing hedge must not convert a get that + // was about to succeed into an error. + // NOTE: `select` polls its first argument first, so a primary that + // is ready simultaneously with the hedge is never miscredited as a + // hedge win, which matters because hedges_won is the detection + // signal that replaces the suppressed timeout counters (see the + // module doc). tokio::select! does NOT have this property unless + // marked `biased`. + match select(primary.as_mut(), hedge.as_mut()).await { + Either::Left((Ok(res), _hedge)) => Ok(res), + Either::Right((Ok(res), _primary)) => { + self.record_win(key, start); + Ok(res) + } + Either::Left((Err(primary_err), hedge)) => { + // The primary failed after the hedge fired. Give the hedge + // a bounded grace window before returning the error: the + // window is only there to let an already-healthy hedge win, + // which takes about one round trip; reusing the hedge delay + // as its length caps the added latency of every branch at + // one delay (see the module doc). Beyond that, returning the + // primary's error is the better move: it is the un-hedged + // outcome, and the callers that wrap gets in retry_external + // recover this failure class promptly on a fresh + // connection, while an unbounded wait would gamble that + // recovery on the hedge leg's health, holding the get and + // its hedge slot for up to the blob client's per-attempt + // timeout when both legs are unhealthy. The guard stays + // held across the window on purpose: the hedge is still in + // flight, so the slot still bounds real work (contrast the + // hedge-error branch below). + match tokio::time::timeout(delay, hedge).await { + Ok(Ok(res)) => { + self.record_win(key, start); + Ok(res) + } + Ok(Err(hedge_err)) => { + self.metrics.errors.inc(); + warn!(%key, %hedge_err, "hedged blob get: both requests failed"); + // Do not attach the hedge error as context (the + // warning above records it): the error surface + // must not depend on whether a hedge fired (see + // the module doc), and hedge text mentioning + // timeouts could make a string-matching consumer + // like ExternalError::is_timeout misclassify a + // non-timeout error. + Err(primary_err) + } + Err(_elapsed) => { + warn!(%key, "hedged blob get: primary failed, hedge still pending"); + Err(primary_err) + } + } + } + Either::Right((Err(hedge_err), primary)) => { + self.metrics.errors.inc(); + warn!(%key, %hedge_err, "hedge request failed, awaiting primary"); + // The hedge leg is gone, so the concurrency slot no longer + // bounds any in-flight memory. Release it rather than + // pinning it for the primary's remaining hang, which could + // starve other gets of their hedges during exactly the + // events hedging exists for. + drop(guard); + primary.await + } + } + } +} + +#[async_trait] +impl Blob for HedgedBlob { + async fn get(&self, key: &str) -> Result, ExternalError> { + if !BLOB_HEDGED_GET_ENABLED.get(&self.cfg) { + return self.primary.get(key).await; + } + let res = self.get_hedged(key).await; + self.budget + .replenish(BLOB_HEDGED_GET_BUDGET_RATIO.get(&self.cfg)); + res + } + + async fn list_keys_and_metadata( + &self, + key_prefix: &str, + f: &mut (dyn FnMut(BlobMetadata) + Send + Sync), + ) -> Result<(), ExternalError> { + self.primary.list_keys_and_metadata(key_prefix, f).await + } + + async fn set(&self, key: &str, value: Bytes) -> Result<(), ExternalError> { + self.primary.set(key, value).await + } + + async fn delete(&self, key: &str) -> Result, ExternalError> { + self.primary.delete(key).await + } + + async fn restore(&self, key: &str) -> Result<(), ExternalError> { + self.primary.restore(key).await + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use anyhow::anyhow; + use mz_dyncfg::ConfigUpdates; + use mz_ore::metrics::MetricsRegistry; + + use crate::location::tests::blob_impl_test; + use crate::mem::MemMultiRegistry; + + use super::*; + + /// A test [Blob] whose `get` sleeps a fixed delay and then returns a + /// fixed outcome, counting calls. + #[derive(Debug)] + struct TestBlob { + delay: Duration, + outcome: Result, &'static str>, + gets: AtomicUsize, + } + + impl TestBlob { + fn new( + delay: Duration, + outcome: Result, &'static str>, + ) -> Arc { + Arc::new(TestBlob { + delay, + outcome, + gets: AtomicUsize::new(0), + }) + } + } + + #[async_trait] + impl Blob for TestBlob { + async fn get(&self, _key: &str) -> Result, ExternalError> { + self.gets.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(self.delay).await; + match self.outcome { + Ok(x) => Ok(x.map(|x| SegmentedBytes::from(Bytes::from(x)))), + Err(msg) => Err(ExternalError::from(anyhow!(msg))), + } + } + + async fn list_keys_and_metadata( + &self, + _key_prefix: &str, + _f: &mut (dyn FnMut(BlobMetadata) + Send + Sync), + ) -> Result<(), ExternalError> { + unreachable!("test blob only supports get") + } + + async fn set(&self, _key: &str, _value: Bytes) -> Result<(), ExternalError> { + unreachable!("test blob only supports get") + } + + async fn delete(&self, _key: &str) -> Result, ExternalError> { + unreachable!("test blob only supports get") + } + + async fn restore(&self, _key: &str) -> Result<(), ExternalError> { + unreachable!("test blob only supports get") + } + } + + fn test_cfg(customize: impl FnOnce(&mut ConfigUpdates)) -> Arc { + let cfg = crate::cfg::all_dyn_configs(ConfigSet::default()); + let mut updates = ConfigUpdates::default(); + updates.add(&BLOB_HEDGED_GET_ENABLED, true); + customize(&mut updates); + updates.apply(&cfg); + Arc::new(cfg) + } + + fn metrics() -> BlobHedgeMetrics { + BlobHedgeMetrics::new(&MetricsRegistry::new()) + } + + fn hedged(primary: &Arc, hedge: &Arc, cfg: Arc) -> HedgedBlob { + let primary: Arc = Arc::::clone(primary); + let hedge: Arc = Arc::::clone(hedge); + HedgedBlob::new(primary, HedgeSibling::Isolated(hedge), cfg, metrics()) + } + + const SECS: fn(u64) -> Duration = Duration::from_secs; + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn fast_primary_no_hedge() { + let primary = TestBlob::new(SECS(0), Ok(Some("x"))); + let hedge = TestBlob::new(SECS(0), Ok(Some("x"))); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + assert!(blob.get("k").await.unwrap().is_some()); + assert_eq!(hedge.gets.load(Ordering::SeqCst), 0); + assert_eq!(blob.metrics.fired.get(), 0); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn hedge_wins_and_cancels_primary() { + let primary = TestBlob::new(SECS(3600), Ok(Some("slow"))); + let hedge = TestBlob::new(SECS(0), Ok(Some("fast"))); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + let start = tokio::time::Instant::now(); + let res = blob.get("k").await.unwrap().expect("some"); + // The hedge's value won, and it won at exactly the hedge delay, not + // at the primary's 3600s: the primary was cancelled while pending. + assert_eq!(res.into_contiguous(), b"fast".to_vec()); + assert_eq!(start.elapsed(), SECS(2)); + assert_eq!(blob.metrics.fired.get(), 1); + assert_eq!(blob.metrics.won.get(), 1); + assert_eq!(blob.metrics.won_seconds.get_sample_count(), 1); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn primary_wins_after_hedge_fired() { + let primary = TestBlob::new(SECS(3), Ok(Some("primary"))); + let hedge = TestBlob::new(SECS(3600), Ok(Some("hedge"))); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + let res = blob.get("k").await.unwrap().expect("some"); + assert_eq!(res.into_contiguous(), b"primary".to_vec()); + assert_eq!(blob.metrics.fired.get(), 1); + assert_eq!(blob.metrics.won.get(), 0); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn hedge_error_does_not_fail_get() { + let primary = TestBlob::new(SECS(5), Ok(Some("primary"))); + let hedge = TestBlob::new(SECS(0), Err("hedge boom")); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + // First success wins, not first completion: the hedge fails fast at + // the 2s mark but the primary's later success is returned. + let res = blob.get("k").await.unwrap().expect("some"); + assert_eq!(res.into_contiguous(), b"primary".to_vec()); + assert_eq!(blob.metrics.errors.get(), 1); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn primary_error_then_hedge_success() { + let primary = TestBlob::new(SECS(3), Err("primary boom")); + let hedge = TestBlob::new(SECS(2), Ok(Some("hedge"))); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + let res = blob.get("k").await.unwrap().expect("some"); + assert_eq!(res.into_contiguous(), b"hedge".to_vec()); + assert_eq!(blob.metrics.won.get(), 1); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn primary_error_then_hedge_error() { + // Both legs fail with the primary failing first: the hedge's error + // within the grace window is counted, the primary's error returned. + let primary = TestBlob::new(SECS(3), Err("primary boom")); + let hedge = TestBlob::new(Duration::from_millis(1500), Err("hedge boom")); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + let err = blob.get("k").await.unwrap_err(); + assert!(err.to_string().contains("primary boom"), "{}", err); + assert!(!err.to_string().contains("hedge boom"), "{}", err); + assert_eq!(blob.metrics.errors.get(), 1); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn primary_error_hedge_timeout() { + // The primary fails after the hedge fired, and the hedge is slow: + // the get returns the primary's error after a bounded extra wait + // instead of holding on the hedge indefinitely. + let primary = TestBlob::new(SECS(3), Err("primary boom")); + let hedge = TestBlob::new(SECS(3600), Ok(Some("hedge"))); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + let start = tokio::time::Instant::now(); + let err = blob.get("k").await.unwrap_err(); + assert!(err.to_string().contains("primary boom"), "{}", err); + // Primary error at 3s plus the delay-sized grace window. + assert_eq!(start.elapsed(), SECS(5)); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn dropped_get_releases_concurrency_slot() { + // Dropping a hedged get mid-race must release the concurrency slot, + // else abandoned gets would permanently disable hedging. + let primary = TestBlob::new(SECS(3600), Ok(Some("slow"))); + let hedge = TestBlob::new(SECS(3600), Ok(Some("slow"))); + let cfg = test_cfg(|u| u.add(&BLOB_HEDGED_GET_MAX_CONCURRENT, 1)); + let blob = hedged(&primary, &hedge, cfg); + for expected_fired in [1, 2] { + let res = tokio::time::timeout(SECS(10), blob.get("k")).await; + assert!(res.is_err(), "get should still be pending at timeout"); + assert_eq!(blob.metrics.fired.get(), expected_fired); + } + assert_eq!(blob.metrics.skipped_concurrency.get(), 0); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn hedge_error_then_primary_error() { + // Both legs fail with the hedge failing first: the get falls back to + // awaiting the primary and returns the primary's error. + let primary = TestBlob::new(SECS(3), Err("primary boom")); + let hedge = TestBlob::new(SECS(0), Err("hedge boom")); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + let err = blob.get("k").await.unwrap_err(); + assert!(err.to_string().contains("primary boom"), "{}", err); + assert!(!err.to_string().contains("hedge boom"), "{}", err); + assert!(!err.is_timeout()); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn fast_primary_error_passthrough() { + let primary = TestBlob::new(SECS(0), Err("fast fail")); + let hedge = TestBlob::new(SECS(0), Ok(Some("hedge"))); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + assert!(blob.get("k").await.is_err()); + assert_eq!(hedge.gets.load(Ordering::SeqCst), 0); + assert_eq!(blob.metrics.fired.get(), 0); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn ok_none_wins() { + let primary = TestBlob::new(SECS(3600), Ok(None)); + let hedge = TestBlob::new(SECS(0), Ok(None)); + let blob = hedged(&primary, &hedge, test_cfg(|_| {})); + assert!(blob.get("k").await.unwrap().is_none()); + assert_eq!(blob.metrics.won.get(), 1); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn disabled_passthrough() { + let primary = TestBlob::new(SECS(0), Ok(Some("x"))); + let hedge = TestBlob::new(SECS(0), Ok(Some("x"))); + let cfg = test_cfg(|u| u.add(&BLOB_HEDGED_GET_ENABLED, false)); + let blob = hedged(&primary, &hedge, cfg); + assert!(blob.get("k").await.unwrap().is_some()); + assert_eq!(hedge.gets.load(Ordering::SeqCst), 0); + assert_eq!(blob.metrics.fired.get(), 0); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn unavailable_sibling() { + let primary = TestBlob::new(SECS(3), Ok(Some("x"))); + let primary_blob: Arc = Arc::::clone(&primary); + let blob = HedgedBlob::new( + primary_blob, + HedgeSibling::Unavailable, + test_cfg(|_| {}), + metrics(), + ); + assert_eq!(blob.metrics.armed.get(), 0); + assert!(blob.get("k").await.unwrap().is_some()); + assert_eq!(blob.metrics.skipped_unavailable.get(), 1); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn budget_exhausts_and_refills() { + let primary = TestBlob::new(SECS(10), Ok(Some("slow"))); + let hedge = TestBlob::new(SECS(0), Ok(Some("fast"))); + // No refill, so the bucket only ever drains. + let cfg = test_cfg(|u| u.add(&BLOB_HEDGED_GET_BUDGET_RATIO, 0.0)); + let blob = hedged(&primary, &hedge, Arc::clone(&cfg)); + for _ in 0..32 { + assert!(blob.get("k").await.unwrap().is_some()); + } + assert_eq!(blob.metrics.fired.get(), 32); + assert!(blob.get("k").await.unwrap().is_some()); + assert_eq!(blob.metrics.fired.get(), 32); + assert_eq!(blob.metrics.skipped_budget.get(), 1); + // Turn refill up to one token per completed get. The next get still + // finds an empty bucket (refill lands at completion), the one after + // hedges again. + let mut updates = ConfigUpdates::default(); + updates.add(&BLOB_HEDGED_GET_BUDGET_RATIO, 1.0); + updates.apply(&cfg); + assert!(blob.get("k").await.unwrap().is_some()); + assert_eq!(blob.metrics.skipped_budget.get(), 2); + assert!(blob.get("k").await.unwrap().is_some()); + assert_eq!(blob.metrics.fired.get(), 33); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn concurrency_cap() { + let primary = TestBlob::new(SECS(10), Ok(Some("slow"))); + let hedge = TestBlob::new(SECS(5), Ok(Some("fast"))); + let cfg = test_cfg(|u| u.add(&BLOB_HEDGED_GET_MAX_CONCURRENT, 1)); + let blob = hedged(&primary, &hedge, cfg); + let (a, b) = tokio::join!(blob.get("k1"), blob.get("k2")); + assert!(a.is_ok() && b.is_ok()); + assert_eq!(blob.metrics.fired.get(), 1); + assert_eq!(blob.metrics.skipped_concurrency.get(), 1); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn warmer_pings_isolated_sibling() { + let primary = TestBlob::new(SECS(0), Ok(None)); + let hedge = TestBlob::new(SECS(0), Ok(None)); + let cfg = test_cfg(|_| {}); + let sockets = BLOB_HEDGED_GET_MAX_CONCURRENT.get(&cfg); + let blob = hedged(&primary, &hedge, cfg); + // The warmer pings immediately at start, then every 20s, holding as + // many sockets as hedges can run at once. + tokio::time::sleep(SECS(1)).await; + tokio::task::yield_now().await; + assert_eq!(hedge.gets.load(Ordering::SeqCst), sockets); + tokio::time::sleep(SECS(20)).await; + tokio::task::yield_now().await; + assert_eq!(hedge.gets.load(Ordering::SeqCst), 2 * sockets); + drop(blob); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn warmer_gated_on_enabled() { + let primary = TestBlob::new(SECS(0), Ok(None)); + let hedge = TestBlob::new(SECS(0), Ok(None)); + let cfg = test_cfg(|u| u.add(&BLOB_HEDGED_GET_ENABLED, false)); + let blob = hedged(&primary, &hedge, Arc::clone(&cfg)); + // Disabled: the warmer idles, the sibling sees no traffic. + tokio::time::sleep(SECS(120)).await; + tokio::task::yield_now().await; + assert_eq!(hedge.gets.load(Ordering::SeqCst), 0); + // Enabling at runtime starts warming within one warm interval. + let mut updates = ConfigUpdates::default(); + updates.add(&BLOB_HEDGED_GET_ENABLED, true); + updates.apply(&cfg); + tokio::time::sleep(*BLOB_HEDGED_GET_WARM_INTERVAL.default() + SECS(1)).await; + tokio::task::yield_now().await; + assert!(hedge.gets.load(Ordering::SeqCst) > 0); + drop(blob); + } + + #[mz_ore::test(tokio::test(start_paused = true))] + async fn shared_sibling_gets_no_warmer() { + let primary = TestBlob::new(SECS(0), Ok(None)); + let primary_blob: Arc = Arc::::clone(&primary); + let blob = HedgedBlob::new( + primary_blob, + HedgeSibling::SharedWithPrimary, + test_cfg(|_| {}), + metrics(), + ); + assert!(blob._warmer.is_none()); + assert_eq!(blob.metrics.armed.get(), 1); + tokio::time::sleep(SECS(60)).await; + assert_eq!(primary.gets.load(Ordering::SeqCst), 0); + } + + /// A test [Blob] that delays gets so the hedge (delay 0) fires and wins + /// on every get in the conformance run below. Non-get methods pass + /// through undelayed. + #[derive(Debug)] + struct SlowGetBlob(Arc); + + #[async_trait] + impl Blob for SlowGetBlob { + async fn get(&self, key: &str) -> Result, ExternalError> { + tokio::time::sleep(Duration::from_millis(2)).await; + self.0.get(key).await + } + + async fn list_keys_and_metadata( + &self, + key_prefix: &str, + f: &mut (dyn FnMut(BlobMetadata) + Send + Sync), + ) -> Result<(), ExternalError> { + self.0.list_keys_and_metadata(key_prefix, f).await + } + + async fn set(&self, key: &str, value: Bytes) -> Result<(), ExternalError> { + self.0.set(key, value).await + } + + async fn delete(&self, key: &str) -> Result, ExternalError> { + self.0.delete(key).await + } + + async fn restore(&self, key: &str) -> Result<(), ExternalError> { + self.0.restore(key).await + } + } + + /// Runs the full [Blob] conformance suite with a hedge racing on every + /// single get: the primary's gets are artificially delayed while the + /// hedge reads the same underlying store undelayed, so the hedge fires + /// and wins throughout. + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] // unsupported operation: returning ready events from epoll_wait is not yet implemented + async fn hedged_blob_conformance() { + let registry = Arc::new(tokio::sync::Mutex::new(MemMultiRegistry::new(false))); + let cfg = test_cfg(|u| { + u.add(&BLOB_HEDGED_GET_DELAY, Duration::ZERO); + u.add(&BLOB_HEDGED_GET_BUDGET_RATIO, 1.0); + }); + let metrics = metrics(); + let metrics_check = metrics.clone(); + blob_impl_test(move |path| { + let path = path.to_owned(); + let registry = Arc::clone(®istry); + let cfg = Arc::clone(&cfg); + let metrics = metrics.clone(); + async move { + let store: Arc = Arc::new(registry.lock().await.blob(&path)); + let primary: Arc = Arc::new(SlowGetBlob(Arc::clone(&store))); + Ok(HedgedBlob::new( + primary, + HedgeSibling::Isolated(store), + cfg, + metrics, + )) + } + }) + .await + .expect("conformance"); + assert!(metrics_check.fired.get() > 0, "no hedge ever fired"); + assert!(metrics_check.won.get() > 0, "no hedge ever won"); + } +} diff --git a/src/persist/src/lib.rs b/src/persist/src/lib.rs index 0d2177becb091..7ae5740ceea9d 100644 --- a/src/persist/src/lib.rs +++ b/src/persist/src/lib.rs @@ -24,6 +24,7 @@ pub mod file; #[cfg(feature = "foundationdb")] pub mod foundationdb; pub mod generated; +pub mod hedge; pub mod indexed; pub mod intercept; pub mod location; diff --git a/src/persist/src/metrics.rs b/src/persist/src/metrics.rs index d09fe7c60062f..85daad38678de 100644 --- a/src/persist/src/metrics.rs +++ b/src/persist/src/metrics.rs @@ -13,7 +13,69 @@ use std::time::Instant; use mz_ore::metric; use mz_ore::metrics::{Counter, IntCounter, MetricsRegistry}; -use prometheus::IntCounterVec; +use mz_ore::stats::histogram_seconds_buckets; +use prometheus::{Gauge, Histogram, IntCounterVec, IntGauge}; + +/// Metrics for [crate::hedge::HedgedBlob]. +#[derive(Debug, Clone)] +pub struct BlobHedgeMetrics { + pub(crate) fired: IntCounter, + pub(crate) won: IntCounter, + pub(crate) won_seconds: Histogram, + pub(crate) skipped_budget: IntCounter, + pub(crate) skipped_concurrency: IntCounter, + pub(crate) skipped_unavailable: IntCounter, + pub(crate) errors: IntCounter, + pub(crate) warm_errors: IntCounter, + pub(crate) armed: IntGauge, + pub(crate) rtt_latency: Gauge, +} + +impl BlobHedgeMetrics { + /// Returns a new [BlobHedgeMetrics] instance connected to the given + /// registry. + pub fn new(registry: &MetricsRegistry) -> Self { + let skipped: IntCounterVec = registry.register(metric!( + name: "mz_persist_blob_hedges_skipped", + help: "hedge requests not fired for a get that exceeded the hedge delay, by reason", + var_labels: ["reason"], + )); + BlobHedgeMetrics { + fired: registry.register(metric!( + name: "mz_persist_blob_hedges_fired", + help: "blob gets that fired a hedge request", + )), + won: registry.register(metric!( + name: "mz_persist_blob_hedges_won", + help: "blob gets where the hedge request won the race", + )), + won_seconds: registry.register(metric!( + name: "mz_persist_blob_hedge_won_seconds", + help: "end-to-end latency of blob gets won by the hedge request", + buckets: histogram_seconds_buckets(0.000_500, 32.0), + )), + skipped_budget: skipped.with_label_values(&["budget"]), + skipped_concurrency: skipped.with_label_values(&["concurrency"]), + skipped_unavailable: skipped.with_label_values(&["unavailable"]), + errors: registry.register(metric!( + name: "mz_persist_blob_hedge_errors", + help: "hedge requests (the hedge leg only, not the primary) that completed with an error", + )), + warm_errors: registry.register(metric!( + name: "mz_persist_blob_hedge_warm_errors", + help: "warm-path liveness gets on the hedge sibling that failed or timed out", + )), + armed: registry.register(metric!( + name: "mz_persist_blob_hedge_armed", + help: "1 if this process opened a hedge sibling and can hedge when enabled", + )), + rtt_latency: registry.register(metric!( + name: "mz_persist_blob_hedge_rtt_latency", + help: "roundtrip-time of the most recent successful warm-path liveness gets on the hedge sibling", + )), + } + } +} /// Metrics specific to S3Blob's internal workings. #[derive(Debug, Clone)] diff --git a/src/persist/src/s3.rs b/src/persist/src/s3.rs index 9da27ad5739af..bf32439330ab8 100644 --- a/src/persist/src/s3.rs +++ b/src/persist/src/s3.rs @@ -44,6 +44,10 @@ use crate::location::{Blob, BlobMetadata, Determinate, ExternalError}; use crate::metrics::S3BlobMetrics; /// Configuration for opening an [S3Blob]. +/// +/// NOTE: cloning shares the underlying `S3Client` and therefore its HTTP +/// connection pool. Connection-pool isolation (as hedged gets require, see +/// [crate::hedge]) needs a fresh [S3BlobConfig::new]. #[derive(Clone, Debug)] pub struct S3BlobConfig { metrics: S3BlobMetrics, @@ -1129,6 +1133,76 @@ mod tests { Ok(()) } + /// Runs the conformance suite through [crate::hedge::HedgedBlob] with two + /// genuinely independent S3 clients (separate connection pools) pointed + /// at the same bucket/prefix, a hedge racing on every get. Ignored by + /// default like `s3_blob` above. When run against the external test + /// bucket, it is the one exercise of the real pool-isolation path. + #[mz_ore::test(tokio::test(flavor = "multi_thread"))] + #[cfg_attr(coverage, ignore)] // https://github.com/MaterializeInc/database-issues/issues/5586 + #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `TLS_method` on OS `linux` + #[ignore] // TODO: Reenable against minio so it can run locally + async fn s3_blob_hedged() -> Result<(), ExternalError> { + use crate::hedge::{ + BLOB_HEDGED_GET_BUDGET_RATIO, BLOB_HEDGED_GET_DELAY, BLOB_HEDGED_GET_ENABLED, + HedgeSibling, HedgedBlob, + }; + use crate::metrics::BlobHedgeMetrics; + use mz_dyncfg::{ConfigSet, ConfigUpdates}; + + let config = match S3BlobConfig::new_for_test().await? { + Some(client) => client, + None => return Ok(()), + }; + // A second client with its own connection pool. Its generated prefix + // is discarded below: both sides must point at the same store. + let sibling = match S3BlobConfig::new_for_test().await? { + Some(client) => client, + None => return Ok(()), + }; + + let cfg = crate::cfg::all_dyn_configs(ConfigSet::default()); + let mut updates = ConfigUpdates::default(); + updates.add(&BLOB_HEDGED_GET_ENABLED, true); + updates.add(&BLOB_HEDGED_GET_DELAY, Duration::ZERO); + updates.add(&BLOB_HEDGED_GET_BUDGET_RATIO, 1.0); + updates.apply(&cfg); + let cfg = Arc::new(cfg); + + blob_impl_test(move |path| { + let path = path.to_owned(); + let config = config.clone(); + let sibling = sibling.clone(); + let cfg = Arc::clone(&cfg); + async move { + let prefix = format!("{}/s3_blob_hedged_test/{}", config.prefix, path); + let primary_config = S3BlobConfig { + metrics: config.metrics.clone(), + client: config.client.clone(), + bucket: config.bucket.clone(), + prefix: prefix.clone(), + }; + let hedge_config = S3BlobConfig { + metrics: sibling.metrics.clone(), + client: sibling.client.clone(), + bucket: config.bucket.clone(), + prefix, + }; + let primary: Arc = Arc::new(S3Blob::open(primary_config).await?); + let hedge: Arc = Arc::new(S3Blob::open(hedge_config).await?); + Ok(HedgedBlob::new( + primary, + HedgeSibling::Isolated(hedge), + cfg, + BlobHedgeMetrics::new(&MetricsRegistry::new()), + )) + } + }) + .await?; + + Ok(()) + } + #[mz_ore::test] fn should_multipart() { let config = MultipartConfig::default(); diff --git a/test/cluster-spec-sheet/mzcompose.py b/test/cluster-spec-sheet/mzcompose.py index 88595a3744345..0cdcbdab910c2 100644 --- a/test/cluster-spec-sheet/mzcompose.py +++ b/test/cluster-spec-sheet/mzcompose.py @@ -35,6 +35,7 @@ from materialize import MZ_ROOT, buildkite from materialize.mz_env_util import print_environment_id from materialize.mz_version import MzVersion +from materialize.mzcompose import ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS from materialize.mzcompose.composition import ( Composition, WorkflowArgumentParser, @@ -90,7 +91,7 @@ def staging_credentials() -> tuple[str, str]: return staging_username_for_account(index), app_password -MATERIALIZED_ADDITIONAL_SYSTEM_PARAMETER_DEFAULTS = { +MATERIALIZED_ADDITIONAL_SYSTEM_PARAMETER_DEFAULTS = ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS | { "memory_limiter_interval": "0s", "max_credit_consumption_rate": "1024", # Headroom over the 30k default cap from `--envd-objects-scalability-sizes`, diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index bd8d9cb6baac4..43acdfdba2e9c 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/test/launchdarkly-flag-consistency/mzcompose.py @@ -327,6 +327,11 @@ persist_blob_cache_scale_factor_bytes persist_blob_cache_scale_with_threads persist_blob_connect_timeout + persist_blob_hedged_get_budget_ratio + persist_blob_hedged_get_delay + persist_blob_hedged_get_enabled + persist_blob_hedged_get_max_concurrent + persist_blob_hedged_get_warm_interval persist_blob_operation_attempt_timeout persist_blob_operation_timeout persist_blob_read_timeout