Repository navigation
LISTEN wake-up (next_notification) and RowStream::rows_affected - #1
Merged
Merged
Conversation
`execute` reports how many rows a statement affected, but a statement sent with `query_raw` returns a `RowStream`, which dropped the `CommandComplete` tag. A caller that pipelines several statements with `query_raw` (send them all, then read the streams in order) therefore had no way to learn a write's row count without a separate round trip. `RowStream::rows_affected()` returns the count parsed from the tag (`UPDATE 5`, `INSERT 0 3`, `SELECT 2`) once the stream has been read to its end, `Some(0)` for an empty query, and `None` before the end or for a suspended portal. tokio-postgres has the same method. Test: counts for an UPDATE without RETURNING, with RETURNING, and for two pipelined statements read in order. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y3TSRmEZAYU6RBUXhkn7WF
Notifications are queued by the connection coroutine as they arrive, but the queue had no wake-up, so a consumer had to poll it: either burning CPU on a short interval or adding up to the interval in latency. The queue now carries a semaphore that the connection posts once per notification, and once more when the connection closes. `Client::next_notification(timeout)` returns a queued notification at once, otherwise parks the coroutine (or thread) until one arrives, the timeout passes, or the connection closes. `Client::notifications()` is unchanged; mixing direct pops with waits only costs a spurious wake-up. Tests: a parked waiter wakes on a notification raised by another session with no request of its own; queued notifications come first and an empty wait lasts its timeout; terminating the backend releases a waiter well before its timeout. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y3TSRmEZAYU6RBUXhkn7WF
The test postgres is published on 127.0.0.1:5433, but tests/severed_connection.rs reads TEST_PG_PORT and defaults to 5432, so both of its tests failed on every CI run (master included). Set TEST_PG_PORT=5433 for the test step. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y3TSRmEZAYU6RBUXhkn7WF
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Two independent additions, one commit each.
1.
RowStream::rows_affected()— parses theCommandCompletetag of the statement the stream ran and exposes the affected-row count (Noneuntil the stream is drained or for tags without a count;Some(0)for an empty query). Lets callers pipeline write statements throughquery_rawand still learn how many rows changed, without a separateexecuteround trip.2.
Client::next_notification(timeout)— LISTEN consumers can now block on a notification instead of polling the queue. The connection's notification queue is paired with a semaphore that is posted on every push and once more when the I/O loop exits, so a waiter wakes immediately on delivery, returns queued notifications first, times out withNone, and returns promptly (None) when the connection closes.Client::notifications()keeps its existing queue-based behaviour.Tests (in
tests/test/main.rs):row_stream_reports_rows_affected— UPDATE with and without RETURNING, plus two pipelined streams.next_notification_wakes_a_waiting_client_without_pollingnext_notification_returns_queued_notifications_first_and_times_out_emptynext_notification_returns_when_the_connection_closes(usespg_terminate_backend).Local:
cargo test --allagainst postgres:16 on 127.0.0.1:5433 (TEST_PG_PORT=5433forsevered_connection) all green; pre-commit (fmt --check, clippy -D warnings) passes.🤖 Generated with Claude Code
https://claude.ai/code/session_01Y3TSRmEZAYU6RBUXhkn7WF