fix(fetch): bound each partition's read so one slow partition can't starve the rest

[solace-aross]

What does this change do, and why?

Fetch reads its partitions in order, and nothing bounded a single storage read. If one partition's read never finished, the request waited until the client gave up (30s for Java), and every partition after it in the same request got nothing, on every retry. This bounds each partition's Fetch reads by a deadline of its own, in nisshi-storage/src/service/fetch.rs.

• Read deadline per request: max_wait + 5s, under the Fetch read timeouts of Java (30s), librdkafka (60s + fetch.wait.max.ms) and franz-go (10s + max_wait).
• Share per partition: each partition may spend half of the time left, and the last partition all of it. A fast partition passes its unused time on.
• Storage budget: dynostore, lite and postgres stop assembling batches once their budget is spent, checking it only between records or object reads. A partition's budget is max(end of max_wait, now + max_wait / 2), capped at cap - margin, where cap is the partition's share and margin is a quarter of the share, at least 50ms, so a read that spends its whole budget returns what it assembled before the broker abandons it.
  • A slow-but-healthy partition keeps its max_wait budget while max_wait is under 3s. From 3s the first partition gets three quarters of its share: 3.75s at franz-go's 5s default.
  • A partition read after a stall gets half of max_wait of its own, as far as its share allows, and no budget once the share is under 50ms.
  • The budget starts once the partition's offsets are read.
  • If every partition spends its full budget, the request can run out to the read deadline rather than stopping at max_wait. That trades latency in that case for no partition going unread, still under every client's own timeout.
• Permit wait: SemaphoreProxy::fetch (libSQL's default mode, one permit per broker) charges the wait for its permit to the budget it passes on.
• Partition whose records stall: answered with ErrorCode::None, the batches it had read, and the offsets read before them, with the high watermark raised to cover the batches (and the last stable offset too, under ReadCommitted), so the client fetches it again. The bytes those batches took stay counted against max_bytes.
• Partition whose offsets stall: the offsets are read first (fix: answer OFFSET_OUT_OF_RANGE for a fetch offset outside the partition #826), under the same share. If they are not read in time the partition answers None with no records and -1 for the high watermark, last stable offset and log start offset. Java ignores negative offsets; librdkafka and franz-go report -1 watermarks for that partition until the next full answer. Kafka answers an expired remote read (DelayedRemoteFetch) with None and the offsets it holds; the -1 case is nisshi's own.
• Long poll: the partition count is taken again on every round. Time is measured with Instant instead of SystemTime.
• No late reads: a read is not started once its deadline has passed, because timeout_at polls its future once before checking the clock.
• Diagnosis: each abandoned read adds 1 to nisshi_storage_read_deadline_exceeded with operation (fetch, offset_stage) and stage (reading, not_started), the names and values fix: bound each ListOffsets partition read by a deadline #827 uses. A read abandoned in storage logs fetch read deadline exceeded at warn with topic, partition and offset; a read never started logs fetch read not started at debug.

What clients see: the stalled partition is retried on every round, so the partitions behind it are served once per round, not once per client timeout. librdkafka moves the stalled partition to the end of its next request; Java does not, so for Java it stays first and costs up to half the read deadline each round.

Not covered by the deadline:

• Postgres: dropping a read does not cancel a statement already sent, and the connection goes back to the pool with it still running, so the next checkout waits for it, up to the 30s statement_timeout. The budget margin means pg's own budget check usually ends the read first. fix: bound each ListOffsets partition read by a deadline #827 adds a guard that takes a dropped read's connection out of the pool and cancels its statement; pg's fetch can use it once fix: bound each ListOffsets partition read by a deadline #827 lands.
• A synchronous libSQL statement (only permit and pool waits are bounded), and libSQL's mpsc mode, where a fetch already in the request channel still runs with its whole budget after Fetch abandons it. Passing an Instant deadline through Storage::fetch would close this and the permit-wait accounting together.
• SlateDB, which ignores the budget.
• storage.metadata, which is still unbounded. Partitions are still read one after another.

before and Missed here match deadline::within and deadline::Missed from #827, and the counter has the same name, attributes and values. Whichever PR merges second drops its copy and calls the other's.

How was this tested?

The unit tests in fetch.rs use a storage double whose reads are scripted per partition: a read answers at once, never answers, or, like dynostore and lite, reads nothing with a zero budget and returns once its budget is spent plus an overshoot. They cover a stalled partition not starving the rest, stalled partitions ending at the read deadline, a partition that stalls after one batch keeping it (under both isolation levels, asserting the raised last stable offset under ReadCommitted), long-poll rounds sharing the deadline again, both budget margin rules, a zero budget under the least margin, the budget starting after the offsets, offsets stalling before and after the records (both isolation levels), two topics sharing one deadline, an unknown topic counting as read, no polling after the deadline, and SemaphoreProxy::fetch charging its permit wait.

Each rule was mutated and the tests re-run:

Mutation	Failing tests
no per-partition timeout	11
no halving	9
RequestTimedOut instead of None for stalled offsets	1
storage budget from the client's max_wait	8
partition count taken once	1
partial batches dropped on stall	1
no guard against polling after the deadline	1
budget margin a hundredth of the share instead of a quarter	1
no 50ms least margin	2
offsets read again after a stall	1
first offsets read outside the deadline	1
budget computed before the offsets are read	1
permit wait not charged to the budget	1
last stable offset not raised under ReadCommitted	2

Locally: just fmt, just clippy, cargo nextest run -p nisshi-storage --all-features (53 passed), and the nisshi-broker fetch, fetch_wire, produce_fetch, storage_fetch, storage_list_offsets and list_offsets tests on in-memory, libSQL, Postgres and SlateDB (81 passed).

Checklist

• Commits are signed off (git commit -s) — see CONTRIBUTING.md
• just fmt, just clippy, and just test pass locally (targeted tests as above; CI runs the full suite)
• Tests added or updated for the behavior changed
• Docs updated if user-facing behavior changed

🤖 Generated with Claude Code

[sgamelin]

Bounding each partition's read is the right fix for this, and the description makes the trade-offs easy to follow. I checked two of its Kafka claims against 3.9.1, and both hold:

• DelayedRemoteFetch.onComplete answers an expired remote read with NONE, no records, and the offsets the broker already holds (DelayedRemoteFetch.scala#L98-L125).
• The Java consumer does not add fetch.max.wait.ms to its Fetch request timeout.

I'm requesting changes for one regression, in the first inline comment. When a partition's storage budget equals its share of the read deadline, a read that uses its whole budget is always abandoned, and everything it read is lost. With a max_wait of 5s or more (franz-go's default is 5s), this covers the first partition of every Fetch with more than one partition.

The other inline comments:

• Postgres: an abandoned read puts a busy connection back into the pool. libSQL and SlateDB are only partly covered by the bound.
• Tests: the offset_stage deadline and the left count across topics have no test.
• Diagnosis: a warn on each partition is the only signal of a stall.
• A non-blocking question on the halving share.
• The CHANGELOG entry.

[solace-aross]

Rebased on main after #834. fetch_topic's else branch keeps the metadata-error warning from #834 and the left decrement from here. No other change. This PR and #826 both change fetch.rs, so whichever lands second needs a rebase.

[sgamelin]

The margin fixes the regression I requested changes for. I checked it by mutation: with Deadlines::storage capped at cap again, a_read_that_spends_its_budget_keeps_what_it_read fails, and with the margin all 12 fetch tests pass. Approving. The points below don't block the merge.

Earlier comments

• Budget reaches the cap: addressed and pinned by a test.
• Postgres connection back in the pool: partly addressed. The doc on before is now correct. #827 wraps only ListOffsets in its guard (pg.rs#L2908 on #827), so pg's fetch stays uncovered after both PRs merge, and no issue tracks it. Could you open one before this merges? The description's "Postgres" bullet also still says "the abandoned request held its connection just as long", which contradicts the header above it.
• Tests for offset_stage and left: addressed by the three new tests.
• Diagnosis: addressed by the counter and the debug level for reads that never started. One follow-up is inline.
• Halving: your reasoning about wide requests is convincing. Leaving it.
• CHANGELOG: addressed. One wording point is inline.

New

• libSQL permit wait: SemaphoreProxy::fetch passes the budget on unchanged after it waits for its permit (proxy.rs#L180-L197). libSQL's default Semaphore mode has one permit per broker. A read that gets the permit just before its cap then runs synchronously for its whole budget, up to three quarters of its share past the cap. This isn't a regression, since main has no cap at all. Passing max_wait.saturating_sub(start.elapsed()) there would charge the wait to the budget, and it's a one-line change.
• Description: a few parts predate the margin and are now stale:
  • the "Storage budget" formula;
  • "keeps today's timing, bounded by max_wait", which no longer holds for the first partition from a max_wait of 3s;
  • the mutation table;
  • the header's reference to 06cb089, which is not on the branch.
• Merge order with #826: #826 reads offset_stage before the fetch, outside any deadline, and its partition_data takes an OffsetStage rather than an Option. Whichever PR merges second has to bring that first read under before, and choose what a partition answers when it stalls there.

[solace-aross]

Rebased on main after #826 and ready for another look. The three inline threads are answered and resolved. The rest of your approval notes:

• libSQL permit wait: SemaphoreProxy::fetch now passes max_wait.saturating_sub(started_at.elapsed()), so the permit wait counts against the budget. fetch_charges_the_permit_wait_to_the_budget holds the permit for 200ms of a 500ms budget and checks storage gets 300ms.
• Merge order with fix: answer OFFSET_OUT_OF_RANGE for a fetch offset outside the partition #826: the first offsets read now sits under the partition's share. If it misses, the partition answers with no records and -1 offsets, and its records are not read (a_stalled_offset_stage_before_the_read_answers_unknown_offsets). A read that stalls after the offsets answers the offsets it has, raised to cover the records, without reading them again (a_partition_that_stalls_keeps_what_it_read).
• Postgres fetch and the stale description parts: tickets have been raised for Postgres fetch using fix: bound each ListOffsets partition read by a deadline #827's connection guard, SlateDB ignoring its budget, and libSQL statements not being interruptible. The description is rewritten to match.

Whichever of #827 and #852 merges second folds its deadline helpers onto the other's. That is about 70 lines out of fetch.rs, and the counter name and label values already match.

[sgamelin]

The rebase onto #826 works well. The first offset stage read now runs under the partition's share, the budget starts once it's read, and a partition whose records stall answers with the offsets it read first rather than -1. I checked the new tests by mutation: reading the first offset stage outside before fails a_stalled_offset_stage_before_the_read_answers_unknown_offsets, and all 22 fetch and proxy tests pass at 0125c1c. The -1 answer for a partition whose offsets weren't read in time is safe for Java: at 3.9.1 the consumer updates the high watermark, log start and last stable offset only when each is non-negative (FetchCollector.java#L292-L311). Approving again. The points below don't block the merge.

Earlier comments

• Margin doc and tests: addressed. The two short-share tests pin the 50ms least margin and the zero budget.
• offset in the warn, and the labels: addressed, using #827's operation and stage.
• CHANGELOG wording: addressed.
• Permit wait in SemaphoreProxy::fetch: addressed, with a test.
• Stale description: addressed.
• Merge order with #826: settled by this rebase.
• An issue for pg fetch returning a busy connection to the pool: partly addressed. The description says these gaps are tracked, but I couldn't find GitHub issues for pg fetch, the synchronous libSQL statement, or SlateDB ignoring its budget. Could you open them here and link them from "Not covered", so a reader can follow them from the PR?

New

Two inline comments: a missing test for the ReadCommitted arm of covering, and libSQL's mpsc mode.