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

[ggevay]

Persist reads batch parts, rollups, and the txns shard through Blob::get. We characterized a failure class where established pooled S3 connections die by TCP reset after a 5-15s hang, before any client timeout fires, and a single hung get on a hot shard starves a whole replica. This PR adds hedged blob gets: after a dyncfg delay (default 2s), a second request for the same key fires on a fully isolated sibling client (own pool, DNS, credentials, kept warm while enabled) and the first success wins.

Design doc (rendered) - motivation and evidence, the race and its error semantics, pool isolation per backend, the two amplification guards, correctness argument, observability, and rejected alternatives.

Highlights for review:

• HedgedBlob sits below Tasked/MetricsBlob in the blob stack, so loser cancellation is real and a hedged get is one blob_get at the winner's latency.
• open_hedge_sibling encodes per-backend sibling construction (Isolated for S3/Azure, SharedWithPrimary for file/mem/turmoil, best-effort degradation to Unavailable).
• Amplification is bounded by a concurrency cap (default 2, sized to the warm-socket count and the fetch semaphore's blindness to hedge bytes) and a token bucket (32-hedge burst, 1% long-run ratio).
• Enabling hedging inverts the old detection signals: mz_persist_s3_read_timeouts and the SDK poisoning log lines go quiet and hedges_won replaces them as the detector for this class.
• Flag state: initially off in production, on across CI with a 10ms delay (randomized 0s variant hedges every get); parallel-workload flips the flags at runtime; benchmark harnesses pin the production-enablement configuration (on, 2s, 0.01) so benchmark history accumulates ahead of rollout.

Tests added: a unit suite driving the race on tokio's paused clock (win/cancellation timing, error asymmetry incl. the grace window, budget exhaustion and refill, concurrency-slot release on dropped gets, warmer gating and cadence), a blob_impl_test conformance run with a hedge firing and winning on every get, and an env-gated variant against real S3 with two independent clients.

Rollout note: enablement is per environment via LaunchDarkly and requires the LD flags to be created first (the five dyncfgs are allowlisted in the LD-consistency test until then).

Fixes PER-57

Nightly:

• Original Nightly run: https://buildkite.com/materialize/nightly/builds/17879 (only unrelated or minor failures)
• Just Parallel Workload after fixing the config syntax issue: https://buildkite.com/materialize/nightly/builds/17893
• After addressing the first round of comments: https://buildkite.com/materialize/nightly/builds/18158

[petrosagg]

Overall shape looks good!

[petrosagg]

If the 10ms setting above is low enough to cause ~every request to hedge then setting this to one means CI will be doing double the network traffic, potentially messing up with other measurements and results. I'd keep this ratio to the production value of 0.01 in CI

[ggevay]

Done: the CI default is 0.01 now. Hedges still fire in every run (budget-capped at 1% of gets), and 1.0 stays in the variant list so randomized runs can pair a full budget with delay=0s.

[petrosagg]

Since the guard is responsible for encoding the restoration of the slot counter you can construct it as soon as we successfully get the slot and rely on its drop for cleanup.

Suggested change

	let took_token = self
	.micro_tokens
	.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |t| {
	t.checked_sub(HEDGE_COST_MICRO_TOKENS)
	})
	.is_ok();
	if !took_token {
	self.concurrent.fetch_sub(1, Ordering::SeqCst);
	return Err(HedgeRefused::Budget);
	}
	Ok(HedgeGuard(self))
	let guard = HedgeGuard(self);
	let token_res = self
	.micro_tokens
	.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |t| {
	t.checked_sub(HEDGE_COST_MICRO_TOKENS)
	});
	match token_res {
	Ok(_) => Ok(guard),
	Err(_) => Err(HedgeRefused::Budget)
	}

[ggevay]

Done. Kept the early return for the no-slot case, otherwise as suggested.

[petrosagg]

Relaxed ordering here too

Suggested change

	.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |t| {
	.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |t| {

[ggevay]

Done for all the budget atomics (acquire, replenish, guard drop).

[petrosagg]

Suggested change

	self.0.concurrent.fetch_sub(1, Ordering::SeqCst);
	self.0.concurrent.fetch_sub(1, Ordering::Relaxed);

[ggevay]

Done (part of the sweep above).

[petrosagg]

There is a dedicated method for this https://docs.rs/tokio/latest/tokio/time/fn.timeout.html

[ggevay]

Done. Timeout polls the wrapped future before checking the deadline, so a primary ready exactly at the boundary still wins without firing a hedge; the NOTE here covers this now.

[petrosagg]

Why isn't this branch symmetric to the one below? i.e I would expect this to be just hedge.await plus the metrics

[ggevay]

The asymmetry is deliberate. The principle: the primary's outcome is authoritative, and the hedge is opportunistic, invisible unless it wins. Two consequences here:

• Bounded rather than plain hedge.await: the window is only there to let an already-healthy hedge win, which takes about one round trip. Beyond that, returning the error puts us on the known-good recovery path (callers wrap gets in retry_external, whose retry on a fresh connection recovers this failure class promptly). An unbounded wait gambles that on the hedge leg's health, up to its 90s attempt timeout in a correlated both-legs event, and pins one of the (default 2) hedge slots for that long.
• The primary's error rather than the hedge's outcome: it is exactly what the caller would have seen without hedging. Surfacing the hedge's error would make the error surface depend on whether a hedge happened to fire (and the hedge's failure is already logged).

The mirror branch has no such shortcut: when the hedge errors, we hold no outcome for the caller's request, so awaiting the primary (bounded by its own timeouts) is the only transparent option.

I've expanded the code comments to spell this out; primary_error_hedge_timeout pins the timing.

[ggevay]

Thank you for the review! I've addressed the comments, ready for the next review round.

Edit: I've also added one more commit, which tweaks the design doc.

[def-]

QA LLM Review

1. MEDIUM -- The concurrency guard does not bound hedge memory

src/persist/src/hedge.rs:148

The guard charges every hedge one slot regardless of response size, so it does not provide the advertised memory bound. A slow oversized batch part, hollow run, or rollup can therefore allocate a second full response outside the existing fetch accounting and OOM a replica when hedging is enabled.

Details

The 128 MiB persist_blob_target_size used to justify the default of two slots is explicitly best-effort and does not apply to an individual update (src/persist-client/src/batch.rs:415). The columnar representation permits key/value buffers up to i32::MAX bytes (src/persist/src/indexed/columnar.rs:46). Rollups and hollow runs are also read through this wrapper without the batch fetch semaphore. For batch reads, the caller acquires permits once for the logical blob's encoded size before calling blob.get (src/persist-client/src/fetch.rs:228), but the hedge can concurrently build another complete SegmentedBytes that those permits do not cover. The full initial 32-token bucket admits this immediately, with up to two oversized duplicates in flight. A large incompressible row or large state blob that takes longer than the two-second delay can therefore add hundreds of MiB or more beyond the memory budget. This needs byte-aware admission, a hard hedge-size limit, or chunk-level hedging. Counting calls cannot establish the claimed memory or egress bound.

[ggevay]

The mechanism is the documented trade rather than a missed one: the design doc's "Bounding amplification" section states that the hedge leg's buffers are invisible to the fetch-path accounting and that the concurrency cap is what bounds the unaccounted transient buffers, to about two parts. Two refinements here are fair, though: that bound inherits the best-effort nature of the 128 MiB target (a single oversized update raises the ceiling with it), and rollup and hollow-run gets sit outside the fetch semaphore for the primary leg too, so hedging doubles an already-unaccounted read there, still capped at two in flight.

Byte-aware admission is not implementable at this layer: Blob::get(key) carries no size metadata, so the wrapper learns the size only when the bytes arrive, and plumbing a size hint through would extend the correctness-critical Blob trait (rejected in Alternatives). The remedy that genuinely bounds bytes is per-chunk hedging, which is the recorded follow-up. Sustained duplication of large parts is separately throttled by the budget (one percent of gets).

[ggevay]

Thank you for the review!