fix: bound each ListOffsets partition read by a deadline

[solace-aross]

Summary

ListOffsets made one Storage::list_offsets call per partition in the
request, with no time limit. Engines answer partitions one after another, so
a slow storage read for one partition delayed every partition after it, and
the whole request could outlast the client's own read timeout. The client
then reconnects and re-sends the request while the abandoned one keeps
reading.

Change

• Each partition is now read with its own Storage::list_offsets call, up
  to 4 concurrently, all bounded by one deadline 5 seconds after the request
  arrives.
• A partition still unread at the deadline answers REQUEST_TIMED_OUT,
  which the Java consumer/admin client, librdkafka, and franz-go all retry.
  5 seconds sits under the shortest client read deadline for ListOffsets
  (franz-go, 10s); the new deadline module has the full table.
• A storage error still fails the whole request.
• On PostgreSQL, a read dropped at the deadline asks the server to cancel its statement and keeps its
  connection and its permit until the statement has ended, whether or not the cancel request reached the
  server, then returns the connection to the pool. The connection is closed and taken out of the pool only
  when it fails or when the statement has not ended within the connection's statement_timeout plus 5
  seconds (35 seconds by default; an operator's own statement_timeout in the URL moves it), logged at
  warn with the host the cancel went to. The engine keeps at most 8 ListOffsets partition reads in flight
  across all requests, half its pool of 16, counting an abandoned read until its statement has ended, so the
  broker never holds more than its 16 server connections. Each cancelled statement logs ERROR: canceling statement due to user request on the server. The other engines have only the per-request limit of 4.
• nisshi_storage_read_deadline_exceeded has a stage attribute: queued is a read that was waiting for
  one of the 8 permits, for a pooled connection, or for the libSQL semaphore or request channel; reading is
  a read whose storage call was slow; not_started is a read the deadline passed before it began. A request
  with timed-out partitions logs one warning with the count per stage and a sample of the partitions.

The Storage trait is unchanged.

Test plan

• New unit tests in nisshi-storage/src/service/deadline.rs and
  service/list_offsets.rs cover: answers returned in request order,
  a slow partition timing out alone without blocking others, the
  deadline already passed before a read starts, a storage error
  failing the whole request, and the counter under each stage.
• New tests in nisshi-storage-sql/src/pg.rs cover the Postgres read limit without a server, the cancel and
  connection return against a real server (pg_sleep(30), dropped, then pg_stat_activity polled until the
  backend is idle and the connection is back in the pool), three tests against a real server with the cancel
  pointed at a closed port (statement outlives its cancel, statement still running at the bound, connection
  terminated by the server), the statement_timeout options parser, and the bound derived from the URL.
• A read waiting for the libSQL semaphore or request channel counts as queued.
• many_partitions_with_latency: 64 partitions with injected latency
  on every engine (in-memory, libSQL, SlateDB, Postgres).
• Existing list_offsets integration tests pass across all four
  backends.
• cargo clippy --workspace --all-features --all-targets -- -D warnings
  clean.
• cargo fmt --all --check clean.

🤖 Generated with Claude Code

https://claude.ai/code/session_01D8pQUJmdicavN4zsumUdaD

[sgamelin]

Thanks for this. Putting a time limit on ListOffsets is the right call, and the unit tests are tight: paused time, the not-started-after-the-deadline guard and request order are all pinned down.

I checked the client side, and REQUEST_TIMED_OUT is a safe code to send. The Java consumer retries only the failed partitions (OffsetFetcherUtils.java#L170-L173), the admin client retries it as a RetriableException (ListOffsetsHandler.java#L177), and librdkafka retries it (rdkafka_request.c#L1045). The timed-out row (offset and timestamp -1) also matches what Kafka 4.0 sends from its remote list-offsets purgatory.

I'm requesting changes for two things. Both are about what happens to a read after the deadline drops it.

1. Merge after #836, or include its fix. In libSQL mode=mpsc, one task runs all storage. That task exits the first time it can't deliver a response (service.rs#L1011-L1013): tx.send fails because the caller dropped its receiver, and the ? ends the loop. After that, every storage call fails with UnableToSend until the broker restarts. Before this PR, that took a client disconnecting mid-request. Now any ListOffsets read slower than 5s does it. #836 fixes exactly this, so landing it first is enough.
2. On PostgreSQL, a dropped read keeps its statement running on a connection that the pool treats as idle. Details inline.

The inline comments also cover the cost of one engine call per partition, logging, and a note on the Kafka difference.

[solace-aross]

On the first summary point: this PR now depends on #836 merging first, since #836 is what keeps the mpsc supervisor alive when a caller drops its receiver. Rebased onto main and updated the CHANGELOG entry in 8174f88.

[solace-aross]

Rebased on main. The Postgres list_offsets body now lives in list_offsets_on with main's no-row default for Timestamp lookups, so timestamp_no_match still holds. #836's supervisor fix is on main, so the merge-order note no longer applies.

[sgamelin]

Thanks for turning this round so quickly. What I asked for is in: #836 is on main, the warning and the counter test look good (inverting the is_none() check now fails counts_each_abandoned_read), and the 64-partition test shows where the per-request limit sits. I'm happy to leave the trait change and a configurable deadline for later.

I'm still requesting changes, for two problems with the new limits rather than the old ones. Details inline.

1. The shared limit of 8 applies to every backend, and the wait for it counts against the deadline. I suggested a limit shared by all ListOffsets requests but didn't say where it should live; it belongs with the Postgres pool. As a static in nisshi-storage, it lets a few partitions that always read slower than 5s hold every permit through their retries. Then every ListOffsets on the broker answers REQUEST_TIMED_OUT, on S3 and memory too.
2. Each abandoned read now opens two new server connections, and its backend no longer counts against the pool. This load arrives when Postgres is already slow.

The third inline comment suggests a test that sends the cancel to a real server.

[solace-aross]

Rebased on main and ready for another look. The mechanism you reviewed on daac0ff is unchanged. Three things differ since then:

• The rustdoc on cancel_and_return now says what happens when a cancel fails or outlasts the grace period: Object::take drops the client, but tokio-postgres only sends Terminate once the pending statement has answered, so that connection stays open on the server until the statement ends (up to statement_timeout), beside the pool's replacement.
• abandoned_statement_is_cancelled_and_its_connection_returned now asserts the cancel task still holds its LIST_OFFSETS_READS permit right after the drop.
• many_partitions_with_latency has a SlateDB leg, so it runs on all four engines.

The real-server cancel test and the Postgres leg only run in CI's Postgres matrix.

[sgamelin]

Thanks, this round closes most of it. Both earlier points are in. The limit lives in Postgres now, so memory, S3 and SlateDB keep only the per-request 4. The stage attribute separates queue waits from slow reads. A successful cancel now returns the connection to the pool. I checked that abandoned_statement_is_cancelled_and_its_connection_returned really runs against the server in the postgres:17 job, so it isn't silently skipped. It asserts size == 2 and available == 1, so it would catch a take on that path. Using ClosedByPeer to wait for the server's EOF, as libpq does, is a good catch.

I'm still requesting changes for one thing, and part of it is my fault: the short timeout I suggested last round leaves the gap open when a cancel misses its statement. Details inline. The other inline comments are optional.

[solace-aross]

Filed the two follow-ups from the design review:

• Give the metadata reader's DbReaderOptions an explicit retry budget that fits inside this PR's 5s deadline. Today it is unbounded in practice, so a dropped read keeps retrying against the object store after nisshi has already answered the client.
• Once the Fetch half lands, a persistently slow partition can starve every partition after it in the same request, since Fetch reads partitions sequentially (unlike ListOffsets here). Accepted for this first cut, tracked separately.