From 1d5170c723012f4530903c454a0098904e8251b8 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Sun, 2 Aug 2026 16:15:02 +0300 Subject: [PATCH 01/14] fix(connection): don't deadlock the connection loop during COPY IN MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `binary_copy::write_basic` hangs indefinitely when run on an otherwise idle connection. It passes only when a busy test suite happens to generate the concurrent activity that unsticks the loop, which is why the suite appeared healthy. The deadlock process_req's CopyIn arm blocked once the copy channel drained: Ok(None) => { nonblock_write(stream, write_buf)?; copy_in_msg = rcv.recv(); // blocks } That parks the only coroutine that reads the socket. Meanwhile copy_in() queues its initial buffer and waits on responses.next() for BindComplete — a message only that parked coroutine could decode. Two halves of one connection waiting on each other. The fix Copy data is now drained without blocking. The receiver is held as `pending_copy_in` state on the connection loop and polled each pass, so reads keep being serviced while a copy is in flight. Two consequences of no longer sitting inside the copy arm had to be made explicit, because the blocking recv had been providing them as side effects: - CopyInSink wakes the connection after queuing data. Copy data travels on its own channel rather than the request queue, so nothing else would tell the loop there is more to write and it would stay parked in wait_io. - process_req defers other queued requests while a copy is in flight. The connection is in COPY mode and the server accepts only CopyData/CopyDone/CopyFail; writing a queued request mid-copy injects it into the stream and the server rejects the transfer with "unexpected message type 0x43 during COPY from stdin". WaitIoWaker is not Clone, so Connection now holds it in an Arc to share with the sink. process_req no longer writes, so its unused stream parameter is dropped. Tests Four regression tests, each bounded by a 20s deadline so a deadlock fails legibly instead of hanging CI: - a COPY IN completes on an idle connection, with no concurrent activity — the case the existing test could not catch - the connection stays usable for ordinary queries afterwards - an empty copy completes, the boundary case where the channel is empty from the outset - abandoning a copy without finish() aborts it and leaves the connection answering Verified against PostgreSQL 16. binary_copy goes from hanging to 10/10 passing in 0.08s, and write_basic alone from a timeout to passing in 0.02s. Notably the suite now completes at all: on the base commit it aborts after 38 tests, and with this change all 81 run. The remaining failures are environmental — auth-mode tests needing pg_hba entries, and citext/hstore tests needing those extensions — and the citext/hstore ones are newly visible only because execution now reaches them. --- src/client.rs | 6 ++ src/connection.rs | 103 ++++++++++++++++++++------ src/copy_in.rs | 16 +++- tests/test/binary_copy.rs | 150 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 250 insertions(+), 25 deletions(-) diff --git a/src/client.rs b/src/client.rs index 301d7b4..a36f2f9 100644 --- a/src/client.rs +++ b/src/client.rs @@ -200,6 +200,12 @@ impl Client { } } + /// Waker for this client's connection coroutine, so a streaming request + /// can nudge the loop after queuing data outside the request queue. + pub(crate) fn connection_waker(&self) -> std::sync::Arc { + self.inner.sender.waker() + } + pub(crate) fn inner(&self) -> &Arc { &self.inner } diff --git a/src/connection.rs b/src/connection.rs index aa5487e..7274a87 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -71,7 +71,7 @@ pub struct Response { pub(crate) struct Connection { io_handle: JoinHandle<()>, req_queue: Arc>, - waker: WaitIoWaker, + waker: Arc, id: usize, } @@ -175,13 +175,59 @@ fn decode_messages( Ok(()) } +/// Move whatever copy-in data is currently available into the write buffer. +/// +/// Returns without waiting when the channel is empty, leaving the receiver in +/// `pending` so the caller retries on the next pass. Clearing `pending` signals +/// the copy is over — either finished (`Done`, which the receiver turns into +/// `CopyDone`/`Sync`) or abandoned, when the sender was dropped and the receiver +/// produces `CopyFail`/`Sync`. +#[inline] +fn drain_copy_in(pending: &mut Option, write_buf: &mut BytesMut) { + let Some(rcv) = pending.as_mut() else { + return; + }; + + loop { + match rcv.try_recv() { + Ok(Some(msg)) => { + reserve_buf(write_buf); + match msg { + FrontendMessage::Raw(buf) => write_buf.extend_from_slice(&buf), + FrontendMessage::CopyData(data) => data.write(write_buf), + } + } + // Nothing queued yet; the client is still writing. Yield so the loop + // can service reads, and pick this up again next pass. + Ok(None) => return, + // The copy has ended, in either direction. + Err(()) => { + *pending = None; + return; + } + } + } +} + #[inline] fn process_req( - stream: &mut impl Write, req_queue: &Queue, rsp_queue: &mut VecDeque, write_buf: &mut BytesMut, + pending_copy_in: &mut Option, ) -> io::Result<()> { + // While a copy is in flight the connection is in COPY mode: the server will + // only accept CopyData/CopyDone/CopyFail until it ends. Writing any other + // queued request now would inject it into the copy stream and the server + // would reject the whole thing with "unexpected message type during COPY". + // + // The previous blocking `rcv.recv()` gave this ordering for free, by never + // leaving the copy arm. Draining without blocking means the guard has to be + // explicit — requests simply wait their turn. + if pending_copy_in.is_some() { + return Ok(()); + } + while let Some(req) = req_queue.pop() { reserve_buf(write_buf); rsp_queue.push_back(Response { @@ -193,26 +239,16 @@ fn process_req( FrontendMessage::Raw(buf) => write_buf.extend_from_slice(&buf), FrontendMessage::CopyData(data) => data.write(write_buf), }, - RequestMessages::CopyIn(mut rcv) => { - let mut copy_in_msg = rcv.try_recv(); - loop { - match copy_in_msg { - Ok(Some(msg)) => { - match msg { - FrontendMessage::Raw(buf) => write_buf.extend_from_slice(&buf), - FrontendMessage::CopyData(data) => data.write(write_buf), - } - copy_in_msg = rcv.try_recv(); - } - Ok(None) => { - nonblock_write(stream, write_buf)?; - - // no data found we just write all the data and wait - copy_in_msg = rcv.recv(); - } - Err(_) => break, - } - } + RequestMessages::CopyIn(rcv) => { + // Take ownership and drain without blocking. Blocking here — as + // this used to, via `rcv.recv()` — parks the only coroutine that + // reads the socket, while the client waits for a response that + // only this coroutine can decode. That is a deadlock between the + // two halves of one connection, and it strikes whenever the copy + // channel drains before the client has queued more data, which + // on an otherwise idle connection is immediately. + *pending_copy_in = Some(rcv); + drain_copy_in(pending_copy_in, write_buf); } } } @@ -236,12 +272,21 @@ fn connection_loop( let mut read_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut write_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut rsp_queue = VecDeque::with_capacity(512); + let mut pending_copy_in: Option = None; let mut io_events = 1; // allow read loop { let inner_stream = stream.inner_mut(); - process_req(inner_stream, &req_queue, &mut rsp_queue, &mut write_buf).map_err(Error::io)?; + process_req( + &req_queue, + &mut rsp_queue, + &mut write_buf, + &mut pending_copy_in, + ) + .map_err(Error::io)?; + // An in-flight copy may have had more data queued since the last pass. + drain_copy_in(&mut pending_copy_in, &mut write_buf); nonblock_write(inner_stream, &mut write_buf).map_err(Error::io)?; let mut read_blocked = true; @@ -258,7 +303,7 @@ impl Connection { pub(crate) fn new(mut stream: TcpStream, parameters: HashMap) -> Connection { use std::os::fd::AsRawFd; let id = stream.as_raw_fd() as usize; - let waker = stream.waker(); + let waker = Arc::new(stream.waker()); let req_queue = Arc::new(Queue::new()); let req_queue_dup = req_queue.clone(); @@ -288,4 +333,14 @@ impl Connection { pub fn id(&self) -> usize { self.id } + + /// Waker for the connection coroutine. + /// + /// `send` wakes the loop when a request is queued. A copy-in streams its + /// data through a separate channel, so it must do the same, or the loop + /// stays parked in `wait_io` with data waiting to be written. + #[inline] + pub fn waker(&self) -> Arc { + self.waker.clone() + } } diff --git a/src/copy_in.rs b/src/copy_in.rs index 629a6b3..ddcface 100644 --- a/src/copy_in.rs +++ b/src/copy_in.rs @@ -3,6 +3,7 @@ use crate::codec::FrontendMessage; use crate::connection::RequestMessages; use crate::{query, Error, Statement}; use bytes::{Buf, BufMut, BytesMut}; +use may::io::WaitIoWaker; use may::sync::mpsc; use postgres_protocol::message::backend::Message; use postgres_protocol::message::frontend; @@ -84,6 +85,12 @@ impl CopyInReceiver { /// not, the copy will be aborted. pub struct CopyInSink { sender: mpsc::Sender, + /// Wakes the connection coroutine after data is queued. + /// + /// Copy data travels on its own channel rather than the request queue, so + /// nothing else would tell the loop there is more to write; it would stay + /// parked in `wait_io` until unrelated socket activity happened to wake it. + waker: std::sync::Arc, responses: Responses, buf: BytesMut, _p: PhantomData, @@ -113,7 +120,9 @@ where let data = CopyData::new(data).map_err(Error::encode)?; self.sender .send(CopyInMessage::Message(FrontendMessage::CopyData(data))) - .map_err(|_| Error::closed()) + .map_err(|_| Error::closed())?; + self.waker.wakeup(); + Ok(()) } /// send iterator of bufs @@ -139,11 +148,13 @@ where self.sender .send(CopyInMessage::Message(FrontendMessage::CopyData(data))) .map_err(|_| Error::closed())?; + self.waker.wakeup(); } self.sender .send(CopyInMessage::Done) .map_err(|_| Error::closed())?; + self.waker.wakeup(); match self.responses.next()? { Message::CommandComplete(body) => { @@ -172,11 +183,13 @@ where let (sender, receiver) = mpsc::channel(); let receiver = CopyInReceiver::new(receiver); + let waker = client.connection_waker(); let mut responses = client.send(RequestMessages::CopyIn(receiver))?; sender .send(CopyInMessage::Message(FrontendMessage::Raw(buf))) .map_err(|_| Error::closed())?; + waker.wakeup(); match responses.next()? { Message::BindComplete => {} @@ -190,6 +203,7 @@ where Ok(CopyInSink { sender, + waker, responses, buf: BytesMut::new(), _p: PhantomData, diff --git a/tests/test/binary_copy.rs b/tests/test/binary_copy.rs index 412ca99..9540d60 100644 --- a/tests/test/binary_copy.rs +++ b/tests/test/binary_copy.rs @@ -168,3 +168,153 @@ fn read_big_rows() { assert_eq!(row.get::<&[u8]>(1), &vec![i as u8; 128 * 1024][..]); } } + +/// Run `f` on a worker thread, failing rather than hanging if it does not finish. +/// +/// A deadlock in the connection loop would otherwise block the whole test binary +/// indefinitely, which in CI looks like an infrastructure problem rather than a +/// bug. Bounding it turns the hang into a legible failure. +fn with_deadline(what: &str, f: F) +where + F: FnOnce() + Send + 'static, +{ + use std::sync::mpsc; + use std::time::Duration; + + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + f(); + let _ = tx.send(()); + }); + + if rx.recv_timeout(Duration::from_secs(20)).is_err() { + panic!("{what} did not complete within 20s — the connection loop is deadlocked"); + } +} + +/// Regression: a COPY IN must complete on an otherwise idle connection. +/// +/// `process_req` used to block in `rcv.recv()` once the copy channel drained, +/// parking the only coroutine that reads the socket. The client was meanwhile +/// waiting for `BindComplete`, which that coroutine had to decode — a circular +/// wait between the two halves of one connection. +/// +/// The existing `write_basic` test did not catch this: it passes when a busy +/// test suite happens to generate the concurrent activity that unsticks the +/// loop. Isolation is the case that matters, so this test creates its own +/// connection and does nothing else. +#[test] +fn copy_in_completes_without_concurrent_activity() { + with_deadline("COPY IN on an idle connection", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE idle_copy (id INT, bar TEXT)") + .unwrap(); + + let sink = client + .copy_in("COPY idle_copy (id, bar) FROM STDIN BINARY") + .unwrap(); + let mut writer = BinaryCopyInWriter::new(sink, &[Type::INT4, Type::TEXT]); + writer.write(&[&1i32, &"alpha"]).unwrap(); + writer.write(&[&2i32, &"beta"]).unwrap(); + let rows_written = writer.finish().unwrap(); + + assert_eq!(rows_written, 2, "COPY should report the rows it inserted"); + + let rows = client + .query("SELECT id, bar FROM idle_copy ORDER BY id", &[]) + .unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].get::<_, i32>(0), 1); + assert_eq!(rows[0].get::<_, &str>(1), "alpha"); + }); +} + +/// The connection must stay usable for ordinary queries after a copy finishes. +/// +/// If the copy left the loop in a partial state, the next request would hang +/// rather than fail — the same deadlock one step later. +#[test] +fn connection_is_reusable_after_a_copy() { + with_deadline("query after COPY IN", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE reuse_copy (id INT)") + .unwrap(); + + let sink = client + .copy_in("COPY reuse_copy (id) FROM STDIN BINARY") + .unwrap(); + let mut writer = BinaryCopyInWriter::new(sink, &[Type::INT4]); + writer.write(&[&7i32]).unwrap(); + writer.finish().unwrap(); + + // Several round trips, to catch a loop that services one more request + // and then stalls. + for expected in 1..=3 { + let row = client.query_one("SELECT $1::INT", &[&expected]).unwrap(); + assert_eq!(row.get::<_, i32>(0), expected); + } + + let row = client + .query_one("SELECT count(*) FROM reuse_copy", &[]) + .unwrap(); + assert_eq!(row.get::<_, i64>(0), 1); + }); +} + +/// An empty copy must still complete rather than block waiting for data. +/// +/// This is the boundary case of the deadlock: the channel is empty from the +/// outset, so a loop that blocks when it finds nothing to write never returns. +#[test] +fn an_empty_copy_completes() { + with_deadline("empty COPY IN", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE empty_copy (id INT)") + .unwrap(); + + let sink = client + .copy_in("COPY empty_copy (id) FROM STDIN BINARY") + .unwrap(); + let writer = BinaryCopyInWriter::new(sink, &[Type::INT4]); + let rows_written = { + let mut w = writer; + w.finish().unwrap() + }; + + assert_eq!(rows_written, 0, "an empty copy inserts nothing"); + }); +} + +/// A copy that is abandoned without `finish()` must not wedge the connection. +/// +/// Dropping the sink closes the channel, which the receiver reports as an +/// error; the loop has to treat that as "copy over" and carry on rather than +/// waiting for data that will never arrive. +#[test] +fn abandoning_a_copy_does_not_wedge_the_connection() { + with_deadline("abandoned COPY IN", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE abandoned_copy (id INT)") + .unwrap(); + + { + let sink = client + .copy_in("COPY abandoned_copy (id) FROM STDIN BINARY") + .unwrap(); + let mut writer = BinaryCopyInWriter::new(sink, &[Type::INT4]); + writer.write(&[&1i32]).unwrap(); + // Deliberately no finish() — the sink is dropped here. + } + + // The copy is aborted, so the row must not be visible, and crucially + // the connection must still answer. + let row = client + .query_one("SELECT count(*) FROM abandoned_copy", &[]) + .unwrap(); + assert_eq!(row.get::<_, i64>(0), 0, "an aborted copy inserts nothing"); + }); +} From 78d8f21bd1e2d2ff71383265ea167d3376223f4e Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Sun, 2 Aug 2026 16:33:11 +0300 Subject: [PATCH 02/14] ci: add GitHub Actions workflow to run the test suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit There is currently no CI, so nothing checks a change before it lands. Workflow test — starts PostgreSQL, runs `cargo test --all` against 14 and 16 lint — `cargo fmt --check` and `cargo clippy --all-targets` Test fixtures The suite exercises each of PostgreSQL's password authentication methods and needs `citext` and `hstore`, none of which the stock image provides. Those are set up by ci/initdb/00-setup.sh, mounted into the container's docker-entrypoint-initdb.d hook: - md5_user, pass_user, scram_user, each with the matching verifier - a pg_hba.conf routing each role to its method pg_hba is written wholesale rather than appended, because it is first-match-wins and appending would leave the per-role lines behind the image's catch-all. Notes on the shape PostgreSQL is started with `docker run` rather than a `services:` block: the container needs fixture scripts from the repository, and service containers are created before checkout runs. The readiness check waits for a fixture role to exist, not just for pg_isready. The entrypoint restarts the server after running initdb scripts, so pg_isready alone can pass against the temporary instance and race the restart. Jobs carry `timeout-minutes`. The suite talks to the database over a coroutine-driven connection, where a protocol-level stall presents as a hang rather than a failure; without a limit a stalled run would occupy a runner for six hours instead of going red. Clippy is not `-D warnings`. That would fail on pre-existing lints unrelated to whatever change is under review. Warnings stay in the log and can be promoted once the tree is clean. Verified locally with nektos/act. With these fixtures all nine authentication tests and both extension tests pass, having previously been unrunnable. Expected result on this branch This will not go green yet: `copy_in_error`, `copy_in_large` and the `binary_copy` write tests hang on this base commit. That is the deadlock fixed in the COPY IN pull request, not a defect in this workflow — the job-level timeout is what turns it into a legible failure. CI should go green once that change lands. --- .github/workflows/ci.yml | 103 +++++++++++++++++++++++++++++++++++++++ ci/initdb/00-setup.sh | 38 +++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 .github/workflows/ci.yml create mode 100755 ci/initdb/00-setup.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..caea076 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,103 @@ +name: CI + +on: + push: + branches: [master] + pull_request: + +env: + CARGO_TERM_COLOR: always + +jobs: + test: + name: Test (PostgreSQL ${{ matrix.postgres }}) + runs-on: ubuntu-latest + # The suite talks to a database over a coroutine-driven connection, so a + # protocol-level stall shows up as a hang rather than a failure. A job-level + # limit turns that into a legible red build instead of a runner burning its + # full six hours. + timeout-minutes: 15 + + strategy: + fail-fast: false + matrix: + # PostgreSQL 14 is deliberately absent. The test binary segfaults + # against it — SIGSEGV after a handful of tests — and it does so on the + # unmodified base commit too, so this is a pre-existing runtime bug + # rather than anything to do with this workflow. A matrix leg that can + # never pass just teaches people to ignore red CI, so 14 stays out until + # that is fixed. Adding it back is a one-line change. + postgres: [16] + + steps: + - uses: actions/checkout@v4 + + # Started with `docker run` rather than a `services:` block because the + # container needs the fixture scripts from the repository, and service + # containers are created before the checkout step runs. + - name: Start PostgreSQL + run: | + # Removed first so a re-run on a persistent runner — or a local + # `act` invocation — does not collide with a container left behind + # by a previous attempt and then silently test against stale fixtures. + docker rm --force postgres 2>/dev/null || true + + docker run --detach --name postgres \ + --publish 127.0.0.1:5433:5432 \ + --env POSTGRES_USER=postgres \ + --env POSTGRES_HOST_AUTH_METHOD=trust \ + --volume "$PWD/ci/initdb:/docker-entrypoint-initdb.d:ro" \ + postgres:${{ matrix.postgres }} + + - name: Wait for PostgreSQL + run: | + for _ in $(seq 1 60); do + if docker exec postgres pg_isready --quiet --username postgres; then + # pg_isready reports the server is accepting connections, but the + # entrypoint restarts it once the initdb scripts have run. Confirm + # a fixture role exists so we do not race that restart. + if docker exec postgres psql --username postgres --tuples-only \ + --command "SELECT 1 FROM pg_roles WHERE rolname = 'scram_user'" \ + | grep --quiet 1; then + echo "PostgreSQL ready with fixtures applied" + exit 0 + fi + fi + sleep 1 + done + echo "PostgreSQL did not become ready" >&2 + docker logs postgres + exit 1 + + - uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + - name: Test + run: cargo test --all + + - name: PostgreSQL logs + if: failure() + run: docker logs postgres + + lint: + name: Lint + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - uses: Swatinem/rust-cache@v2 + + - name: Format + run: cargo fmt --all -- --check + + # Not `-D warnings`: that would make this job fail on pre-existing lints + # unrelated to whichever change is being reviewed. Warnings stay visible in + # the log, and can be promoted to errors once the tree is clean. + - name: Clippy + run: cargo clippy --all-targets diff --git a/ci/initdb/00-setup.sh b/ci/initdb/00-setup.sh new file mode 100755 index 0000000..69c1776 --- /dev/null +++ b/ci/initdb/00-setup.sh @@ -0,0 +1,38 @@ +#!/bin/bash +# Test fixtures for may_postgres' suite, applied by the postgres image's +# docker-entrypoint-initdb.d hook. +# +# The suite exercises each of PostgreSQL's password authentication methods, so +# it needs one role per method and a pg_hba.conf that routes them accordingly. +set -euo pipefail + +psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname postgres <<-'SQL' + -- Extensions used by the type round-trip tests. + CREATE EXTENSION IF NOT EXISTS citext; + CREATE EXTENSION IF NOT EXISTS hstore; + + -- One role per authentication method. `password_encryption` decides how the + -- verifier is stored, which is what makes md5 and scram distinguishable; + -- cleartext `password` auth works against either, so pass_user can reuse md5. + SET password_encryption TO 'md5'; + CREATE ROLE md5_user LOGIN PASSWORD 'password'; + CREATE ROLE pass_user LOGIN PASSWORD 'password'; + + SET password_encryption TO 'scram-sha-256'; + CREATE ROLE scram_user LOGIN PASSWORD 'password'; +SQL + +# Written wholesale rather than appended: pg_hba is first-match-wins, so the +# per-role lines must precede the catch-all that the image generates, and +# appending would leave them unreachable. +cat > "$PGDATA/pg_hba.conf" <<-'HBA' + # TYPE DATABASE USER ADDRESS METHOD + local all all trust + + host all md5_user all md5 + host all pass_user all password + host all scram_user all scram-sha-256 + + # Everything else, including the postgres superuser the tests connect as. + host all all all trust +HBA From f32d6d75c03e42489cf5e5d8705d2c3887ccd3a8 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Sun, 2 Aug 2026 16:33:11 +0300 Subject: [PATCH 03/14] ci: add GitHub Actions workflow to run the test suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit There is currently no CI, so nothing checks a change before it lands. Workflow test — starts PostgreSQL, runs `cargo test --all` against 14 and 16 lint — `cargo fmt --check` and `cargo clippy --all-targets` Test fixtures The suite exercises each of PostgreSQL's password authentication methods and needs `citext` and `hstore`, none of which the stock image provides. Those are set up by ci/initdb/00-setup.sh, mounted into the container's docker-entrypoint-initdb.d hook: - md5_user, pass_user, scram_user, each with the matching verifier - a pg_hba.conf routing each role to its method pg_hba is written wholesale rather than appended, because it is first-match-wins and appending would leave the per-role lines behind the image's catch-all. Notes on the shape PostgreSQL is started with `docker run` rather than a `services:` block: the container needs fixture scripts from the repository, and service containers are created before checkout runs. The readiness check waits for a fixture role to exist, not just for pg_isready. The entrypoint restarts the server after running initdb scripts, so pg_isready alone can pass against the temporary instance and race the restart. Jobs carry `timeout-minutes`. The suite talks to the database over a coroutine-driven connection, where a protocol-level stall presents as a hang rather than a failure; without a limit a stalled run would occupy a runner for six hours instead of going red. Clippy is not `-D warnings`. That would fail on pre-existing lints unrelated to whatever change is under review. Warnings stay in the log and can be promoted once the tree is clean. Verified locally with nektos/act. With these fixtures all nine authentication tests and both extension tests pass, having previously been unrunnable. Expected result on this branch This will not go green yet: `copy_in_error`, `copy_in_large` and the `binary_copy` write tests hang on this base commit. That is the deadlock fixed in the COPY IN pull request, not a defect in this workflow — the job-level timeout is what turns it into a legible failure. CI should go green once that change lands. --- .github/workflows/ci.yml | 103 +++++++++++++++++++++++++++++++++++++++ ci/initdb/00-setup.sh | 38 +++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 .github/workflows/ci.yml create mode 100755 ci/initdb/00-setup.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..caea076 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,103 @@ +name: CI + +on: + push: + branches: [master] + pull_request: + +env: + CARGO_TERM_COLOR: always + +jobs: + test: + name: Test (PostgreSQL ${{ matrix.postgres }}) + runs-on: ubuntu-latest + # The suite talks to a database over a coroutine-driven connection, so a + # protocol-level stall shows up as a hang rather than a failure. A job-level + # limit turns that into a legible red build instead of a runner burning its + # full six hours. + timeout-minutes: 15 + + strategy: + fail-fast: false + matrix: + # PostgreSQL 14 is deliberately absent. The test binary segfaults + # against it — SIGSEGV after a handful of tests — and it does so on the + # unmodified base commit too, so this is a pre-existing runtime bug + # rather than anything to do with this workflow. A matrix leg that can + # never pass just teaches people to ignore red CI, so 14 stays out until + # that is fixed. Adding it back is a one-line change. + postgres: [16] + + steps: + - uses: actions/checkout@v4 + + # Started with `docker run` rather than a `services:` block because the + # container needs the fixture scripts from the repository, and service + # containers are created before the checkout step runs. + - name: Start PostgreSQL + run: | + # Removed first so a re-run on a persistent runner — or a local + # `act` invocation — does not collide with a container left behind + # by a previous attempt and then silently test against stale fixtures. + docker rm --force postgres 2>/dev/null || true + + docker run --detach --name postgres \ + --publish 127.0.0.1:5433:5432 \ + --env POSTGRES_USER=postgres \ + --env POSTGRES_HOST_AUTH_METHOD=trust \ + --volume "$PWD/ci/initdb:/docker-entrypoint-initdb.d:ro" \ + postgres:${{ matrix.postgres }} + + - name: Wait for PostgreSQL + run: | + for _ in $(seq 1 60); do + if docker exec postgres pg_isready --quiet --username postgres; then + # pg_isready reports the server is accepting connections, but the + # entrypoint restarts it once the initdb scripts have run. Confirm + # a fixture role exists so we do not race that restart. + if docker exec postgres psql --username postgres --tuples-only \ + --command "SELECT 1 FROM pg_roles WHERE rolname = 'scram_user'" \ + | grep --quiet 1; then + echo "PostgreSQL ready with fixtures applied" + exit 0 + fi + fi + sleep 1 + done + echo "PostgreSQL did not become ready" >&2 + docker logs postgres + exit 1 + + - uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + - name: Test + run: cargo test --all + + - name: PostgreSQL logs + if: failure() + run: docker logs postgres + + lint: + name: Lint + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - uses: Swatinem/rust-cache@v2 + + - name: Format + run: cargo fmt --all -- --check + + # Not `-D warnings`: that would make this job fail on pre-existing lints + # unrelated to whichever change is being reviewed. Warnings stay visible in + # the log, and can be promoted to errors once the tree is clean. + - name: Clippy + run: cargo clippy --all-targets diff --git a/ci/initdb/00-setup.sh b/ci/initdb/00-setup.sh new file mode 100755 index 0000000..69c1776 --- /dev/null +++ b/ci/initdb/00-setup.sh @@ -0,0 +1,38 @@ +#!/bin/bash +# Test fixtures for may_postgres' suite, applied by the postgres image's +# docker-entrypoint-initdb.d hook. +# +# The suite exercises each of PostgreSQL's password authentication methods, so +# it needs one role per method and a pg_hba.conf that routes them accordingly. +set -euo pipefail + +psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname postgres <<-'SQL' + -- Extensions used by the type round-trip tests. + CREATE EXTENSION IF NOT EXISTS citext; + CREATE EXTENSION IF NOT EXISTS hstore; + + -- One role per authentication method. `password_encryption` decides how the + -- verifier is stored, which is what makes md5 and scram distinguishable; + -- cleartext `password` auth works against either, so pass_user can reuse md5. + SET password_encryption TO 'md5'; + CREATE ROLE md5_user LOGIN PASSWORD 'password'; + CREATE ROLE pass_user LOGIN PASSWORD 'password'; + + SET password_encryption TO 'scram-sha-256'; + CREATE ROLE scram_user LOGIN PASSWORD 'password'; +SQL + +# Written wholesale rather than appended: pg_hba is first-match-wins, so the +# per-role lines must precede the catch-all that the image generates, and +# appending would leave them unreachable. +cat > "$PGDATA/pg_hba.conf" <<-'HBA' + # TYPE DATABASE USER ADDRESS METHOD + local all all trust + + host all md5_user all md5 + host all pass_user all password + host all scram_user all scram-sha-256 + + # Everything else, including the postgres superuser the tests connect as. + host all all all trust +HBA From b1877b15f4d5335b9c7921940e3e103ee5e097ae Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Sun, 2 Aug 2026 16:15:02 +0300 Subject: [PATCH 04/14] fix(connection): don't deadlock the connection loop during COPY IN MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `binary_copy::write_basic` hangs indefinitely when run on an otherwise idle connection. It passes only when a busy test suite happens to generate the concurrent activity that unsticks the loop, which is why the suite appeared healthy. The deadlock process_req's CopyIn arm blocked once the copy channel drained: Ok(None) => { nonblock_write(stream, write_buf)?; copy_in_msg = rcv.recv(); // blocks } That parks the only coroutine that reads the socket. Meanwhile copy_in() queues its initial buffer and waits on responses.next() for BindComplete — a message only that parked coroutine could decode. Two halves of one connection waiting on each other. The fix Copy data is now drained without blocking. The receiver is held as `pending_copy_in` state on the connection loop and polled each pass, so reads keep being serviced while a copy is in flight. Two consequences of no longer sitting inside the copy arm had to be made explicit, because the blocking recv had been providing them as side effects: - CopyInSink wakes the connection after queuing data. Copy data travels on its own channel rather than the request queue, so nothing else would tell the loop there is more to write and it would stay parked in wait_io. - process_req defers other queued requests while a copy is in flight. The connection is in COPY mode and the server accepts only CopyData/CopyDone/CopyFail; writing a queued request mid-copy injects it into the stream and the server rejects the transfer with "unexpected message type 0x43 during COPY from stdin". WaitIoWaker is not Clone, so Connection now holds it in an Arc to share with the sink. process_req no longer writes, so its unused stream parameter is dropped. Tests Four regression tests, each bounded by a 20s deadline so a deadlock fails legibly instead of hanging CI: - a COPY IN completes on an idle connection, with no concurrent activity — the case the existing test could not catch - the connection stays usable for ordinary queries afterwards - an empty copy completes, the boundary case where the channel is empty from the outset - abandoning a copy without finish() aborts it and leaves the connection answering Verified against PostgreSQL 16. binary_copy goes from hanging to 10/10 passing in 0.08s, and write_basic alone from a timeout to passing in 0.02s. Notably the suite now completes at all: on the base commit it aborts after 38 tests, and with this change all 81 run. The remaining failures are environmental — auth-mode tests needing pg_hba entries, and citext/hstore tests needing those extensions — and the citext/hstore ones are newly visible only because execution now reaches them. --- src/client.rs | 6 ++ src/connection.rs | 103 ++++++++++++++++++++------ src/copy_in.rs | 16 +++- tests/test/binary_copy.rs | 150 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 250 insertions(+), 25 deletions(-) diff --git a/src/client.rs b/src/client.rs index 301d7b4..a36f2f9 100644 --- a/src/client.rs +++ b/src/client.rs @@ -200,6 +200,12 @@ impl Client { } } + /// Waker for this client's connection coroutine, so a streaming request + /// can nudge the loop after queuing data outside the request queue. + pub(crate) fn connection_waker(&self) -> std::sync::Arc { + self.inner.sender.waker() + } + pub(crate) fn inner(&self) -> &Arc { &self.inner } diff --git a/src/connection.rs b/src/connection.rs index aa5487e..7274a87 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -71,7 +71,7 @@ pub struct Response { pub(crate) struct Connection { io_handle: JoinHandle<()>, req_queue: Arc>, - waker: WaitIoWaker, + waker: Arc, id: usize, } @@ -175,13 +175,59 @@ fn decode_messages( Ok(()) } +/// Move whatever copy-in data is currently available into the write buffer. +/// +/// Returns without waiting when the channel is empty, leaving the receiver in +/// `pending` so the caller retries on the next pass. Clearing `pending` signals +/// the copy is over — either finished (`Done`, which the receiver turns into +/// `CopyDone`/`Sync`) or abandoned, when the sender was dropped and the receiver +/// produces `CopyFail`/`Sync`. +#[inline] +fn drain_copy_in(pending: &mut Option, write_buf: &mut BytesMut) { + let Some(rcv) = pending.as_mut() else { + return; + }; + + loop { + match rcv.try_recv() { + Ok(Some(msg)) => { + reserve_buf(write_buf); + match msg { + FrontendMessage::Raw(buf) => write_buf.extend_from_slice(&buf), + FrontendMessage::CopyData(data) => data.write(write_buf), + } + } + // Nothing queued yet; the client is still writing. Yield so the loop + // can service reads, and pick this up again next pass. + Ok(None) => return, + // The copy has ended, in either direction. + Err(()) => { + *pending = None; + return; + } + } + } +} + #[inline] fn process_req( - stream: &mut impl Write, req_queue: &Queue, rsp_queue: &mut VecDeque, write_buf: &mut BytesMut, + pending_copy_in: &mut Option, ) -> io::Result<()> { + // While a copy is in flight the connection is in COPY mode: the server will + // only accept CopyData/CopyDone/CopyFail until it ends. Writing any other + // queued request now would inject it into the copy stream and the server + // would reject the whole thing with "unexpected message type during COPY". + // + // The previous blocking `rcv.recv()` gave this ordering for free, by never + // leaving the copy arm. Draining without blocking means the guard has to be + // explicit — requests simply wait their turn. + if pending_copy_in.is_some() { + return Ok(()); + } + while let Some(req) = req_queue.pop() { reserve_buf(write_buf); rsp_queue.push_back(Response { @@ -193,26 +239,16 @@ fn process_req( FrontendMessage::Raw(buf) => write_buf.extend_from_slice(&buf), FrontendMessage::CopyData(data) => data.write(write_buf), }, - RequestMessages::CopyIn(mut rcv) => { - let mut copy_in_msg = rcv.try_recv(); - loop { - match copy_in_msg { - Ok(Some(msg)) => { - match msg { - FrontendMessage::Raw(buf) => write_buf.extend_from_slice(&buf), - FrontendMessage::CopyData(data) => data.write(write_buf), - } - copy_in_msg = rcv.try_recv(); - } - Ok(None) => { - nonblock_write(stream, write_buf)?; - - // no data found we just write all the data and wait - copy_in_msg = rcv.recv(); - } - Err(_) => break, - } - } + RequestMessages::CopyIn(rcv) => { + // Take ownership and drain without blocking. Blocking here — as + // this used to, via `rcv.recv()` — parks the only coroutine that + // reads the socket, while the client waits for a response that + // only this coroutine can decode. That is a deadlock between the + // two halves of one connection, and it strikes whenever the copy + // channel drains before the client has queued more data, which + // on an otherwise idle connection is immediately. + *pending_copy_in = Some(rcv); + drain_copy_in(pending_copy_in, write_buf); } } } @@ -236,12 +272,21 @@ fn connection_loop( let mut read_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut write_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut rsp_queue = VecDeque::with_capacity(512); + let mut pending_copy_in: Option = None; let mut io_events = 1; // allow read loop { let inner_stream = stream.inner_mut(); - process_req(inner_stream, &req_queue, &mut rsp_queue, &mut write_buf).map_err(Error::io)?; + process_req( + &req_queue, + &mut rsp_queue, + &mut write_buf, + &mut pending_copy_in, + ) + .map_err(Error::io)?; + // An in-flight copy may have had more data queued since the last pass. + drain_copy_in(&mut pending_copy_in, &mut write_buf); nonblock_write(inner_stream, &mut write_buf).map_err(Error::io)?; let mut read_blocked = true; @@ -258,7 +303,7 @@ impl Connection { pub(crate) fn new(mut stream: TcpStream, parameters: HashMap) -> Connection { use std::os::fd::AsRawFd; let id = stream.as_raw_fd() as usize; - let waker = stream.waker(); + let waker = Arc::new(stream.waker()); let req_queue = Arc::new(Queue::new()); let req_queue_dup = req_queue.clone(); @@ -288,4 +333,14 @@ impl Connection { pub fn id(&self) -> usize { self.id } + + /// Waker for the connection coroutine. + /// + /// `send` wakes the loop when a request is queued. A copy-in streams its + /// data through a separate channel, so it must do the same, or the loop + /// stays parked in `wait_io` with data waiting to be written. + #[inline] + pub fn waker(&self) -> Arc { + self.waker.clone() + } } diff --git a/src/copy_in.rs b/src/copy_in.rs index 629a6b3..ddcface 100644 --- a/src/copy_in.rs +++ b/src/copy_in.rs @@ -3,6 +3,7 @@ use crate::codec::FrontendMessage; use crate::connection::RequestMessages; use crate::{query, Error, Statement}; use bytes::{Buf, BufMut, BytesMut}; +use may::io::WaitIoWaker; use may::sync::mpsc; use postgres_protocol::message::backend::Message; use postgres_protocol::message::frontend; @@ -84,6 +85,12 @@ impl CopyInReceiver { /// not, the copy will be aborted. pub struct CopyInSink { sender: mpsc::Sender, + /// Wakes the connection coroutine after data is queued. + /// + /// Copy data travels on its own channel rather than the request queue, so + /// nothing else would tell the loop there is more to write; it would stay + /// parked in `wait_io` until unrelated socket activity happened to wake it. + waker: std::sync::Arc, responses: Responses, buf: BytesMut, _p: PhantomData, @@ -113,7 +120,9 @@ where let data = CopyData::new(data).map_err(Error::encode)?; self.sender .send(CopyInMessage::Message(FrontendMessage::CopyData(data))) - .map_err(|_| Error::closed()) + .map_err(|_| Error::closed())?; + self.waker.wakeup(); + Ok(()) } /// send iterator of bufs @@ -139,11 +148,13 @@ where self.sender .send(CopyInMessage::Message(FrontendMessage::CopyData(data))) .map_err(|_| Error::closed())?; + self.waker.wakeup(); } self.sender .send(CopyInMessage::Done) .map_err(|_| Error::closed())?; + self.waker.wakeup(); match self.responses.next()? { Message::CommandComplete(body) => { @@ -172,11 +183,13 @@ where let (sender, receiver) = mpsc::channel(); let receiver = CopyInReceiver::new(receiver); + let waker = client.connection_waker(); let mut responses = client.send(RequestMessages::CopyIn(receiver))?; sender .send(CopyInMessage::Message(FrontendMessage::Raw(buf))) .map_err(|_| Error::closed())?; + waker.wakeup(); match responses.next()? { Message::BindComplete => {} @@ -190,6 +203,7 @@ where Ok(CopyInSink { sender, + waker, responses, buf: BytesMut::new(), _p: PhantomData, diff --git a/tests/test/binary_copy.rs b/tests/test/binary_copy.rs index 412ca99..9540d60 100644 --- a/tests/test/binary_copy.rs +++ b/tests/test/binary_copy.rs @@ -168,3 +168,153 @@ fn read_big_rows() { assert_eq!(row.get::<&[u8]>(1), &vec![i as u8; 128 * 1024][..]); } } + +/// Run `f` on a worker thread, failing rather than hanging if it does not finish. +/// +/// A deadlock in the connection loop would otherwise block the whole test binary +/// indefinitely, which in CI looks like an infrastructure problem rather than a +/// bug. Bounding it turns the hang into a legible failure. +fn with_deadline(what: &str, f: F) +where + F: FnOnce() + Send + 'static, +{ + use std::sync::mpsc; + use std::time::Duration; + + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + f(); + let _ = tx.send(()); + }); + + if rx.recv_timeout(Duration::from_secs(20)).is_err() { + panic!("{what} did not complete within 20s — the connection loop is deadlocked"); + } +} + +/// Regression: a COPY IN must complete on an otherwise idle connection. +/// +/// `process_req` used to block in `rcv.recv()` once the copy channel drained, +/// parking the only coroutine that reads the socket. The client was meanwhile +/// waiting for `BindComplete`, which that coroutine had to decode — a circular +/// wait between the two halves of one connection. +/// +/// The existing `write_basic` test did not catch this: it passes when a busy +/// test suite happens to generate the concurrent activity that unsticks the +/// loop. Isolation is the case that matters, so this test creates its own +/// connection and does nothing else. +#[test] +fn copy_in_completes_without_concurrent_activity() { + with_deadline("COPY IN on an idle connection", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE idle_copy (id INT, bar TEXT)") + .unwrap(); + + let sink = client + .copy_in("COPY idle_copy (id, bar) FROM STDIN BINARY") + .unwrap(); + let mut writer = BinaryCopyInWriter::new(sink, &[Type::INT4, Type::TEXT]); + writer.write(&[&1i32, &"alpha"]).unwrap(); + writer.write(&[&2i32, &"beta"]).unwrap(); + let rows_written = writer.finish().unwrap(); + + assert_eq!(rows_written, 2, "COPY should report the rows it inserted"); + + let rows = client + .query("SELECT id, bar FROM idle_copy ORDER BY id", &[]) + .unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].get::<_, i32>(0), 1); + assert_eq!(rows[0].get::<_, &str>(1), "alpha"); + }); +} + +/// The connection must stay usable for ordinary queries after a copy finishes. +/// +/// If the copy left the loop in a partial state, the next request would hang +/// rather than fail — the same deadlock one step later. +#[test] +fn connection_is_reusable_after_a_copy() { + with_deadline("query after COPY IN", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE reuse_copy (id INT)") + .unwrap(); + + let sink = client + .copy_in("COPY reuse_copy (id) FROM STDIN BINARY") + .unwrap(); + let mut writer = BinaryCopyInWriter::new(sink, &[Type::INT4]); + writer.write(&[&7i32]).unwrap(); + writer.finish().unwrap(); + + // Several round trips, to catch a loop that services one more request + // and then stalls. + for expected in 1..=3 { + let row = client.query_one("SELECT $1::INT", &[&expected]).unwrap(); + assert_eq!(row.get::<_, i32>(0), expected); + } + + let row = client + .query_one("SELECT count(*) FROM reuse_copy", &[]) + .unwrap(); + assert_eq!(row.get::<_, i64>(0), 1); + }); +} + +/// An empty copy must still complete rather than block waiting for data. +/// +/// This is the boundary case of the deadlock: the channel is empty from the +/// outset, so a loop that blocks when it finds nothing to write never returns. +#[test] +fn an_empty_copy_completes() { + with_deadline("empty COPY IN", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE empty_copy (id INT)") + .unwrap(); + + let sink = client + .copy_in("COPY empty_copy (id) FROM STDIN BINARY") + .unwrap(); + let writer = BinaryCopyInWriter::new(sink, &[Type::INT4]); + let rows_written = { + let mut w = writer; + w.finish().unwrap() + }; + + assert_eq!(rows_written, 0, "an empty copy inserts nothing"); + }); +} + +/// A copy that is abandoned without `finish()` must not wedge the connection. +/// +/// Dropping the sink closes the channel, which the receiver reports as an +/// error; the loop has to treat that as "copy over" and carry on rather than +/// waiting for data that will never arrive. +#[test] +fn abandoning_a_copy_does_not_wedge_the_connection() { + with_deadline("abandoned COPY IN", || { + let client = crate::connect("user=postgres"); + client + .batch_execute("CREATE TEMPORARY TABLE abandoned_copy (id INT)") + .unwrap(); + + { + let sink = client + .copy_in("COPY abandoned_copy (id) FROM STDIN BINARY") + .unwrap(); + let mut writer = BinaryCopyInWriter::new(sink, &[Type::INT4]); + writer.write(&[&1i32]).unwrap(); + // Deliberately no finish() — the sink is dropped here. + } + + // The copy is aborted, so the row must not be visible, and crucially + // the connection must still answer. + let row = client + .query_one("SELECT count(*) FROM abandoned_copy", &[]) + .unwrap(); + assert_eq!(row.get::<_, i64>(0), 0, "an aborted copy inserts nothing"); + }); +} From f34edf41c74345cdc3bb80581dae2cc78e6bb77d Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Sun, 2 Aug 2026 15:55:11 +0300 Subject: [PATCH 05/14] feat(client): deliver LISTEN/NOTIFY notifications to the client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `NotificationResponse` was decoded and then dropped: BackendMessage::Async(Message::NotificationResponse(_body)) => {} so `LISTEN` could be issued but nothing could ever be observed. `Notification` and `AsyncMessage` remained in lib.rs as vestiges of the tokio-postgres origin, along with a commented-out `notifications` test, but nothing constructed them. The connection coroutine now decodes each notification and pushes it onto an unbounded queue that the client drains: let client = may_postgres::connect("host=localhost user=postgres")?; client.batch_execute("LISTEN orders")?; while let Some(n) = client.notifications().pop() { println!("{}: {}", n.channel(), n.payload()); } Design Uses `may::queue::mpsc::Queue`, the same primitive already carrying the request queue between client and connection. No futures, no streams, no reactor — draining fits in whatever loop the caller already runs, which suits the coroutine model. The upstream tokio-postgres approach (`connection.poll_message(cx)` yielding an `AsyncMessage` stream) is deliberately not revived; the commented-out test that used it is replaced with tests for the queue API. The queue is unbounded and only the caller drains it, so a client that subscribes and never reads will grow it without limit. Documented on the accessor. Tests Four covering delivery to a LISTENing client from a second session, ordering and payload preservation across several notifications, isolation so an unsubscribed channel is never queued, and that a client which never issues LISTEN receives nothing. Verified against PostgreSQL 16. Three tests fail in this environment (custom_array, pipelined_prepare, plain_password_wrong) — confirmed failing identically on the unmodified base commit, so unrelated. Example examples/listen_notify.rs shows the two-connection pattern and a long-running listener of the kind a cache invalidator or job runner would use. --- examples/listen_notify.rs | 81 +++++++++++++++++++++++ src/client.rs | 34 +++++++++- src/connection.rs | 32 ++++++++-- tests/test/main.rs | 131 ++++++++++++++++++++++++++++++-------- 4 files changed, 245 insertions(+), 33 deletions(-) create mode 100644 examples/listen_notify.rs diff --git a/examples/listen_notify.rs b/examples/listen_notify.rs new file mode 100644 index 0000000..566c5a7 --- /dev/null +++ b/examples/listen_notify.rs @@ -0,0 +1,81 @@ +//! `LISTEN` / `NOTIFY` — receiving asynchronous notifications. +//! +//! One connection subscribes to a channel and drains notifications; another +//! raises them. This is the shape behind change-data-capture and cache +//! invalidation: a trigger calls `pg_notify()` when a row changes, and a +//! listener reacts without polling the table. +//! +//! Run with a PostgreSQL on localhost: +//! +//! ```sh +//! cargo run --example listen_notify +//! ``` + +use may_postgres::{Client, Error}; +use std::time::Duration; + +fn main() -> Result<(), Error> { + let listener = may_postgres::connect("host=localhost user=postgres")?; + let notifier = may_postgres::connect("host=localhost user=postgres")?; + + // Only channels this connection has subscribed to are delivered. + listener.batch_execute("LISTEN orders")?; + println!("listening on 'orders'"); + + notifier.batch_execute( + "NOTIFY orders, 'order-1 created'; + NOTIFY orders, 'order-2 created';", + )?; + + // Notifications are decoded by the connection coroutine during I/O, so the + // listener has to do some work before they appear. Draining inside the same + // loop that does the work is the natural pattern — there is no separate + // reactor to run. + let mut seen = 0; + for _ in 0..50 { + listener.batch_execute("SELECT 1")?; + + while let Some(notification) = listener.notifications().pop() { + println!( + " [{}] {} -> {}", + notification.process_id(), + notification.channel(), + notification.payload(), + ); + seen += 1; + } + + if seen >= 2 { + break; + } + may::coroutine::sleep(Duration::from_millis(20)); + } + + println!("received {seen} notification(s)"); + Ok(()) +} + +/// A long-running listener, as a cache invalidator or job runner would use it. +/// +/// The queue is unbounded and only the caller drains it, so a process that +/// subscribes and then stops reading will grow it without limit. Draining every +/// iteration, as here, keeps it bounded. +#[allow(dead_code)] +fn run_forever(listener: &Client) -> Result<(), Error> { + listener.batch_execute("LISTEN row_changes")?; + + loop { + listener.batch_execute("SELECT 1")?; + + while let Some(notification) = listener.notifications().pop() { + handle(notification.channel(), notification.payload()); + } + + may::coroutine::sleep(Duration::from_millis(100)); + } +} + +#[allow(dead_code)] +fn handle(channel: &str, payload: &str) { + println!("{channel}: {payload}"); +} diff --git a/src/client.rs b/src/client.rs index a36f2f9..bf90732 100644 --- a/src/client.rs +++ b/src/client.rs @@ -7,8 +7,8 @@ use crate::query::RowStream; use crate::simple_query::SimpleQueryStream; use crate::types::{Oid, ToSql, Type}; use crate::{ - copy_in, copy_out, prepare, query, simple_query, CancelToken, CopyInSink, Error, Row, - SimpleQueryMessage, Statement, ToStatement, Transaction, TransactionBuilder, + copy_in, copy_out, prepare, query, simple_query, CancelToken, CopyInSink, Error, Notification, + Row, SimpleQueryMessage, Statement, ToStatement, Transaction, TransactionBuilder, }; use bytes::{Buf, BytesMut}; use fallible_iterator::FallibleIterator; @@ -174,6 +174,36 @@ impl Clone for Client { } impl Client { + /// Asynchronous notifications delivered by the server. + /// + /// A connection receives notifications after subscribing with `LISTEN`; + /// another session raises them with `NOTIFY` or `pg_notify()`. Before this + /// existed the connection decoded `NotificationResponse` and discarded it, + /// so `LISTEN` could be issued but nothing could ever be observed. + /// + /// The queue is unbounded and is only drained by the caller, so a client + /// that issues `LISTEN` and never reads will grow it without limit. Poll it + /// as part of the same loop that does the work: + /// + /// ```no_run + /// # fn main() -> Result<(), may_postgres::Error> { + /// let client = may_postgres::connect("host=localhost user=postgres")?; + /// client.batch_execute("LISTEN my_channel")?; + /// + /// loop { + /// // Any round trip lets the connection surface pending notifications. + /// client.batch_execute("SELECT 1")?; + /// while let Some(notification) = client.notifications().pop() { + /// println!("{}: {}", notification.channel(), notification.payload()); + /// } + /// } + /// # } + /// ``` + #[inline] + pub fn notifications(&self) -> &may::queue::mpsc::Queue { + self.inner.sender.notifications() + } + pub(crate) fn new(sender: Connection, process_id: i32, secret_key: i32) -> Client { let co_ch = { let (tx, rx) = spsc::channel(); diff --git a/src/connection.rs b/src/connection.rs index 7274a87..730be21 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -11,7 +11,7 @@ use postgres_protocol::message::frontend; use crate::codec::{BackendMessage, BackendMessages, FrontendMessage}; use crate::copy_in::CopyInReceiver; -use crate::Error; +use crate::{Error, Notification}; use std::collections::{HashMap, VecDeque}; use std::io::{self, Read, Write}; @@ -71,6 +71,9 @@ pub struct Response { pub(crate) struct Connection { io_handle: JoinHandle<()>, req_queue: Arc>, + /// Asynchronous notifications delivered by the server, filled by the + /// connection coroutine and drained by the client. + notifications: Arc>, waker: Arc, id: usize, } @@ -136,6 +139,7 @@ fn decode_messages( read_buf: &mut BytesMut, rsp_queue: &mut VecDeque, parameters: &mut HashMap, + notifications: &Queue, ) -> Result<(), Error> { use crate::codec::PostgresCodec; @@ -162,7 +166,13 @@ fn decode_messages( } } BackendMessage::Async(Message::NoticeResponse(_body)) => {} - BackendMessage::Async(Message::NotificationResponse(_body)) => {} + BackendMessage::Async(Message::NotificationResponse(body)) => { + notifications.push(Notification { + process_id: body.process_id(), + channel: body.channel().map_err(Error::parse)?.to_string(), + payload: body.message().map_err(Error::parse)?.to_string(), + }); + } BackendMessage::Async(Message::ParameterStatus(body)) => { parameters.insert( body.name().map_err(Error::parse)?.to_string(), @@ -268,6 +278,7 @@ fn connection_loop( stream: &mut TcpStream, req_queue: Arc>, mut params: HashMap, + notifications: Arc>, ) -> Result<(), Error> { let mut read_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut write_buf = BytesMut::with_capacity(IO_BUF_SIZE); @@ -292,7 +303,7 @@ fn connection_loop( let mut read_blocked = true; if io_events & 1 != 0 { read_blocked = nonblock_read(inner_stream, &mut read_buf).map_err(Error::io)?; - decode_messages(&mut read_buf, &mut rsp_queue, &mut params)?; + decode_messages(&mut read_buf, &mut rsp_queue, &mut params, ¬ifications)?; } io_events = if read_blocked { stream.wait_io() } else { 1 } @@ -307,8 +318,12 @@ impl Connection { let req_queue = Arc::new(Queue::new()); let req_queue_dup = req_queue.clone(); + let notifications = Arc::new(Queue::new()); + let notifications_dup = notifications.clone(); let io_handle = go!(move || { - if let Err(e) = connection_loop(&mut stream, req_queue_dup, parameters) { + if let Err(e) = + connection_loop(&mut stream, req_queue_dup, parameters, notifications_dup) + { log::error!("connection error = {:?}", e); terminate_connection(&mut stream); } @@ -317,6 +332,7 @@ impl Connection { Connection { io_handle, req_queue, + notifications, waker, id, } @@ -343,4 +359,12 @@ impl Connection { pub fn waker(&self) -> Arc { self.waker.clone() } + + /// Asynchronous notifications received from the server. + /// + /// Populated by `LISTEN`; see [`crate::Client::notifications`]. + #[inline] + pub fn notifications(&self) -> &Queue { + &self.notifications + } } diff --git a/tests/test/main.rs b/tests/test/main.rs index dd9d3d1..541c8a4 100644 --- a/tests/test/main.rs +++ b/tests/test/main.rs @@ -556,37 +556,114 @@ fn copy_out() { // ); // } -// #[test] -// fn notifications() { -// let (mut client, mut connection) = connect_raw("user=postgres").unwrap(); +#[test] +fn notifications_are_delivered_to_a_listening_client() { + let client = connect("user=postgres"); -// let (tx, rx) = mpsc::unbounded(); -// let stream = stream::poll_fn(move |cx| connection.poll_message(cx)).map_err(|e| panic!(e)); -// let connection = stream.forward(tx).map(|r| r.unwrap()); -// tokio::spawn(connection); + client.batch_execute("LISTEN test_notifications").unwrap(); + assert!( + client.notifications().pop().is_none(), + "nothing has been raised yet" + ); -// client -// .batch_execute( -// "LISTEN test_notifications; -// NOTIFY test_notifications, 'hello'; -// NOTIFY test_notifications, 'world';", -// ) -// .unwrap(); + // Raised from a second session, which is the shape that matters: one + // connection writes, another reacts. + let notifier = connect("user=postgres"); + notifier + .batch_execute("NOTIFY test_notifications, 'hello'") + .unwrap(); -// drop(client); + let notification = + wait_for_notification(&client).expect("a LISTENing client should receive the notification"); -// let notifications = rx -// .filter_map(|m| match m { -// AsyncMessage::Notification(n) => future::ready(Some(n)), -// _ => future::ready(None), -// }) -// .collect::>(); -// assert_eq!(notifications.len(), 2); -// assert_eq!(notifications[0].channel(), "test_notifications"); -// assert_eq!(notifications[0].payload(), "hello"); -// assert_eq!(notifications[1].channel(), "test_notifications"); -// assert_eq!(notifications[1].payload(), "world"); -// } + assert_eq!(notification.channel(), "test_notifications"); + assert_eq!(notification.payload(), "hello"); + assert_ne!( + notification.process_id(), + 0, + "the notifying backend pid should be reported" + ); +} + +#[test] +fn notifications_preserve_order_and_payloads() { + let client = connect("user=postgres"); + client.batch_execute("LISTEN test_ordering").unwrap(); + + let notifier = connect("user=postgres"); + notifier + .batch_execute( + "NOTIFY test_ordering, 'first'; + NOTIFY test_ordering, 'second'; + NOTIFY test_ordering, 'third';", + ) + .unwrap(); + + let mut payloads = Vec::new(); + for _ in 0..3 { + let n = wait_for_notification(&client).expect("all three should arrive"); + payloads.push(n.payload().to_string()); + } + + // A queue that reordered or coalesced messages would be worse than useless + // for change-data-capture, where each row event must be seen exactly once. + assert_eq!(payloads, vec!["first", "second", "third"]); +} + +#[test] +fn only_subscribed_channels_are_queued() { + let client = connect("user=postgres"); + client.batch_execute("LISTEN wanted_channel").unwrap(); + + let notifier = connect("user=postgres"); + notifier + .batch_execute( + "NOTIFY unwanted_channel, 'ignored'; + NOTIFY wanted_channel, 'kept';", + ) + .unwrap(); + + let notification = + wait_for_notification(&client).expect("the subscribed channel should arrive"); + assert_eq!(notification.channel(), "wanted_channel"); + assert_eq!(notification.payload(), "kept"); + assert!( + client.notifications().pop().is_none(), + "a channel we never subscribed to must not be queued" + ); +} + +#[test] +fn a_client_that_never_listens_receives_nothing() { + let client = connect("user=postgres"); + + let notifier = connect("user=postgres"); + notifier + .batch_execute("NOTIFY some_channel, 'nobody is listening'") + .unwrap(); + + client.batch_execute("SELECT 1").unwrap(); + assert!( + client.notifications().pop().is_none(), + "without LISTEN there is nothing to deliver" + ); +} + +/// Poll for a notification, giving the connection round trips to surface it. +/// +/// Notifications arrive asynchronously: the server sends them when it chooses, +/// and the connection coroutine decodes them during I/O. A test that read the +/// queue once, immediately, would race and pass or fail at random. +fn wait_for_notification(client: &Client) -> Option { + for _ in 0..50 { + if let Some(notification) = client.notifications().pop() { + return Some(notification); + } + client.batch_execute("SELECT 1").unwrap(); + may::coroutine::sleep(std::time::Duration::from_millis(20)); + } + None +} #[test] fn query_portal() { From 28ef2c29ff8c8714aefdca4e1c6a37aae4938777 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Sun, 2 Aug 2026 16:33:11 +0300 Subject: [PATCH 06/14] ci: add GitHub Actions workflow to run the test suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit There is currently no CI, so nothing checks a change before it lands. Workflow test — starts PostgreSQL, runs `cargo test --all` against 14 and 16 lint — `cargo fmt --check` and `cargo clippy --all-targets` Test fixtures The suite exercises each of PostgreSQL's password authentication methods and needs `citext` and `hstore`, none of which the stock image provides. Those are set up by ci/initdb/00-setup.sh, mounted into the container's docker-entrypoint-initdb.d hook: - md5_user, pass_user, scram_user, each with the matching verifier - a pg_hba.conf routing each role to its method pg_hba is written wholesale rather than appended, because it is first-match-wins and appending would leave the per-role lines behind the image's catch-all. Notes on the shape PostgreSQL is started with `docker run` rather than a `services:` block: the container needs fixture scripts from the repository, and service containers are created before checkout runs. The readiness check waits for a fixture role to exist, not just for pg_isready. The entrypoint restarts the server after running initdb scripts, so pg_isready alone can pass against the temporary instance and race the restart. Jobs carry `timeout-minutes`. The suite talks to the database over a coroutine-driven connection, where a protocol-level stall presents as a hang rather than a failure; without a limit a stalled run would occupy a runner for six hours instead of going red. Clippy is not `-D warnings`. That would fail on pre-existing lints unrelated to whatever change is under review. Warnings stay in the log and can be promoted once the tree is clean. Verified locally with nektos/act. With these fixtures all nine authentication tests and both extension tests pass, having previously been unrunnable. Expected result on this branch This will not go green yet: `copy_in_error`, `copy_in_large` and the `binary_copy` write tests hang on this base commit. That is the deadlock fixed in the COPY IN pull request, not a defect in this workflow — the job-level timeout is what turns it into a legible failure. CI should go green once that change lands. --- .github/workflows/ci.yml | 103 +++++++++++++++++++++++++++++++++++++++ ci/initdb/00-setup.sh | 38 +++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 .github/workflows/ci.yml create mode 100755 ci/initdb/00-setup.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..caea076 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,103 @@ +name: CI + +on: + push: + branches: [master] + pull_request: + +env: + CARGO_TERM_COLOR: always + +jobs: + test: + name: Test (PostgreSQL ${{ matrix.postgres }}) + runs-on: ubuntu-latest + # The suite talks to a database over a coroutine-driven connection, so a + # protocol-level stall shows up as a hang rather than a failure. A job-level + # limit turns that into a legible red build instead of a runner burning its + # full six hours. + timeout-minutes: 15 + + strategy: + fail-fast: false + matrix: + # PostgreSQL 14 is deliberately absent. The test binary segfaults + # against it — SIGSEGV after a handful of tests — and it does so on the + # unmodified base commit too, so this is a pre-existing runtime bug + # rather than anything to do with this workflow. A matrix leg that can + # never pass just teaches people to ignore red CI, so 14 stays out until + # that is fixed. Adding it back is a one-line change. + postgres: [16] + + steps: + - uses: actions/checkout@v4 + + # Started with `docker run` rather than a `services:` block because the + # container needs the fixture scripts from the repository, and service + # containers are created before the checkout step runs. + - name: Start PostgreSQL + run: | + # Removed first so a re-run on a persistent runner — or a local + # `act` invocation — does not collide with a container left behind + # by a previous attempt and then silently test against stale fixtures. + docker rm --force postgres 2>/dev/null || true + + docker run --detach --name postgres \ + --publish 127.0.0.1:5433:5432 \ + --env POSTGRES_USER=postgres \ + --env POSTGRES_HOST_AUTH_METHOD=trust \ + --volume "$PWD/ci/initdb:/docker-entrypoint-initdb.d:ro" \ + postgres:${{ matrix.postgres }} + + - name: Wait for PostgreSQL + run: | + for _ in $(seq 1 60); do + if docker exec postgres pg_isready --quiet --username postgres; then + # pg_isready reports the server is accepting connections, but the + # entrypoint restarts it once the initdb scripts have run. Confirm + # a fixture role exists so we do not race that restart. + if docker exec postgres psql --username postgres --tuples-only \ + --command "SELECT 1 FROM pg_roles WHERE rolname = 'scram_user'" \ + | grep --quiet 1; then + echo "PostgreSQL ready with fixtures applied" + exit 0 + fi + fi + sleep 1 + done + echo "PostgreSQL did not become ready" >&2 + docker logs postgres + exit 1 + + - uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + - name: Test + run: cargo test --all + + - name: PostgreSQL logs + if: failure() + run: docker logs postgres + + lint: + name: Lint + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - uses: Swatinem/rust-cache@v2 + + - name: Format + run: cargo fmt --all -- --check + + # Not `-D warnings`: that would make this job fail on pre-existing lints + # unrelated to whichever change is being reviewed. Warnings stay visible in + # the log, and can be promoted to errors once the tree is clean. + - name: Clippy + run: cargo clippy --all-targets diff --git a/ci/initdb/00-setup.sh b/ci/initdb/00-setup.sh new file mode 100755 index 0000000..69c1776 --- /dev/null +++ b/ci/initdb/00-setup.sh @@ -0,0 +1,38 @@ +#!/bin/bash +# Test fixtures for may_postgres' suite, applied by the postgres image's +# docker-entrypoint-initdb.d hook. +# +# The suite exercises each of PostgreSQL's password authentication methods, so +# it needs one role per method and a pg_hba.conf that routes them accordingly. +set -euo pipefail + +psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname postgres <<-'SQL' + -- Extensions used by the type round-trip tests. + CREATE EXTENSION IF NOT EXISTS citext; + CREATE EXTENSION IF NOT EXISTS hstore; + + -- One role per authentication method. `password_encryption` decides how the + -- verifier is stored, which is what makes md5 and scram distinguishable; + -- cleartext `password` auth works against either, so pass_user can reuse md5. + SET password_encryption TO 'md5'; + CREATE ROLE md5_user LOGIN PASSWORD 'password'; + CREATE ROLE pass_user LOGIN PASSWORD 'password'; + + SET password_encryption TO 'scram-sha-256'; + CREATE ROLE scram_user LOGIN PASSWORD 'password'; +SQL + +# Written wholesale rather than appended: pg_hba is first-match-wins, so the +# per-role lines must precede the catch-all that the image generates, and +# appending would leave them unreachable. +cat > "$PGDATA/pg_hba.conf" <<-'HBA' + # TYPE DATABASE USER ADDRESS METHOD + local all all trust + + host all md5_user all md5 + host all pass_user all password + host all scram_user all scram-sha-256 + + # Everything else, including the postgres superuser the tests connect as. + host all all all trust +HBA From 82ba69bfe0afb5c16eed4131c7f592a51abc5848 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 13:01:38 +0300 Subject: [PATCH 07/14] fix(connection): graceful shutdown on drop - close the socket, and make the response sender owned rather than a transmuted borrow Dropping a Client leaves its server backend open until process exit: drop cancels the io coroutine, cancel() does not run the coroutine destructors, and the TcpStream is owned by the coroutine closure, so the fd is never closed. Ten connect/query/drop cycles leave ten backends in pg_stat_activity and ten sockets in /proc/self/fd, one for one, and a process that opens connections over time eventually meets the servers too-many-clients limit. The obvious repair - have the io loop exit gracefully so it closes its own socket - trips over a second, latent bug. Requests carry the response sender as RefOrValue with a transmuted static borrow of the CoChannel field inside Client, under the stated safety condition that the client waits until all responses come back. An instantly-cancelled coroutine upholds that condition by force; a graceful one runs at least one more pass and pushes responses through a pointer into freed Client memory. Under load that shows up as SIGSEGV and SIGABRT inside the spsc send. So the fix has two halves that only work together: 1. The sender is now Arc-owned, cloned into each Request, and the RefOrValue transmute is deleted. Queued requests keep the channel alive however long the io coroutine outlives the Client, making shutdown ordering safe by construction. 2. Connection::drop sets a shared AtomicBool and wakes the loop with the same waker send() uses, then DETACHES. The loop polls the flag at the top of each pass, and on every exit path sends the protocol Terminate and shuts down its own socket on its own stack. Drop deliberately does not join: a Client can be dropped inside a coroutine that is itself being cancelled, and parking there aborts the process. The waker holds an Arc into the reactor slot, so a wake after the loop has already exited is a no-op. The unsafe cancel() is gone. Verified, each against PostgreSQL 15: - examples/leak_probe.rs: backends AND open socket fds stay at baseline across 10 connect/query/drop cycles; previously both climbed one per cycle and never returned. - Soak: 500 sequential connect/query/drop against a server with max_connections set to 100 - impossible without live reclamation - both counters pinned at baseline throughout. - The added regression test asserts pg_stat_activity returns to baseline, tagged via application_name so it is immune to parallel tests. - A downstream ORM integration suite of 139 tests that opens a connection per test: previously failed at exactly the connection limit; now 138 passed / 0 failed across 12 consecutive runs with no crashes. - This crate-s own suite: the same tests pass and fail before and after (the failures are auth roles and custom types the local server does not provide). --- examples/leak_probe.rs | 75 ++++++++++++++++++++++++++++++++++++++++++ src/client.rs | 30 ++++++++++------- src/connection.rs | 69 +++++++++++++++++++++++--------------- tests/test/main.rs | 33 +++++++++++++++++++ 4 files changed, 170 insertions(+), 37 deletions(-) create mode 100644 examples/leak_probe.rs diff --git a/examples/leak_probe.rs b/examples/leak_probe.rs new file mode 100644 index 0000000..717f60f --- /dev/null +++ b/examples/leak_probe.rs @@ -0,0 +1,75 @@ +//! Diagnostic: does dropping a Client actually release its resources? +//! +//! Two independent measurements per iteration: +//! - server side: backends in pg_stat_activity for our database +//! - client side: open socket fds in /proc/self/fd +//! +//! If BOTH climb one-per-iteration, the process itself is holding the socket +//! after drop - a genuine client-side leak. If fds return to baseline but +//! backends linger, the client is fine and the server is just slow to reap. + +fn socket_fds() -> usize { + std::fs::read_dir("/proc/self/fd") + .unwrap() + .filter_map(|e| e.ok()) + .filter(|e| { + std::fs::read_link(e.path()) + .map(|t| t.to_string_lossy().starts_with("socket:")) + .unwrap_or(false) + }) + .count() +} + +fn main() { + let url = std::env::var("TEST_URL").expect("set TEST_URL"); + + let observer = may_postgres::connect(&url).expect("observer connect"); + let backends = |o: &may_postgres::Client| -> i64 { + o.query_one( + "SELECT count(*) FROM pg_stat_activity WHERE datname = current_database()", + &[], + ) + .unwrap() + .get(0) + }; + + let base_be = backends(&observer); + let base_fd = socket_fds(); + println!("BASELINE backends={base_be} socket_fds={base_fd}"); + + let iters: u32 = std::env::var("ITERS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(10); + let verbose = iters <= 20; + for i in 1..=iters { + { + let client = may_postgres::connect(&url).expect("connect"); + let rows = client.query("SELECT 1", &[]).expect("query"); + assert_eq!(rows.len(), 1); + drop(client); + } + std::thread::sleep(std::time::Duration::from_millis(if verbose { + 300 + } else { + 5 + })); + if !verbose && i % 100 != 0 { + continue; + } + println!( + "ITER {i} backends={} socket_fds={}", + backends(&observer), + socket_fds() + ); + } + + // Give the server every chance to reap on its own. + println!("WAITING 15s..."); + std::thread::sleep(std::time::Duration::from_secs(15)); + println!( + "FINAL backends={} socket_fds={} (baseline {base_be}/{base_fd})", + backends(&observer), + socket_fds() + ); +} diff --git a/src/client.rs b/src/client.rs index 301d7b4..69c60a9 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,7 +1,7 @@ use crate::cancel_query; use crate::codec::BackendMessages; use crate::config::Host; -use crate::connection::{Connection, RefOrValue, Request, RequestMessages}; +use crate::connection::{Connection, Request, RequestMessages}; use crate::copy_out::CopyOutStream; use crate::query::RowStream; use crate::simple_query::SimpleQueryStream; @@ -64,16 +64,16 @@ pub struct InnerClient { struct CoChannel { tag: Cell, rx: Rc>, - tx: spsc::Sender, + tx: Arc>, } impl CoChannel { - fn sender(&self) -> RefOrValue<'static, spsc::Sender> { - // Safety: - // 1. there is only one sender - // 2. we will wait until all response come back - let tx: &'static spsc::Sender = unsafe { std::mem::transmute(&self.tx) }; - RefOrValue::Ref(tx) + fn sender(&self) -> Arc> { + // Shared ownership, not a transmuted borrow. The old 'static borrow + // relied on the client outliving every queued request, which drop + // breaks - the io coroutine would then push responses through a + // dangling pointer into freed Client memory. + self.tx.clone() } fn tag(&self) -> usize { @@ -91,7 +91,7 @@ impl InnerClient { /// ignore the result pub fn raw_send(&self, messages: RequestMessages) -> Result<(), Error> { let (sender, _rx) = spsc::channel(); - let request = Request::new(0, messages, RefOrValue::Value(sender)); + let request = Request::new(0, messages, Arc::new(sender)); self.sender.send(request); Ok(()) } @@ -160,7 +160,11 @@ impl Clone for Client { let (tx, rx) = spsc::channel(); let rx = Rc::new(rx); let tag = Cell::new(0); - CoChannel { tag, rx, tx } + CoChannel { + tag, + rx, + tx: Arc::new(tx), + } }; Client { inner: self.inner.clone(), @@ -179,7 +183,11 @@ impl Client { let (tx, rx) = spsc::channel(); let rx = Rc::new(rx); let tag = Cell::new(0); - CoChannel { tag, rx, tx } + CoChannel { + tag, + rx, + tx: Arc::new(tx), + } }; Client { inner: Arc::new(InnerClient { diff --git a/src/connection.rs b/src/connection.rs index aa5487e..f48ebec 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -15,27 +15,11 @@ use crate::Error; use std::collections::{HashMap, VecDeque}; use std::io::{self, Read, Write}; -use std::ops::Deref; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; const IO_BUF_SIZE: usize = 4096 * 2; -pub enum RefOrValue<'a, T> { - Ref(&'a T), - Value(T), -} - -impl Deref for RefOrValue<'_, T> { - type Target = T; - #[inline] - fn deref(&self) -> &Self::Target { - match self { - RefOrValue::Ref(r) => r, - RefOrValue::Value(ref v) => v, - } - } -} - pub enum RequestMessages { Single(FrontendMessage), CopyIn(CopyInReceiver), @@ -44,7 +28,11 @@ pub enum RequestMessages { pub struct Request { tag: usize, messages: RequestMessages, - sender: RefOrValue<'static, spsc::Sender>, + // OWNED (shared) sender, never a transmuted borrow: requests can sit in + // the queues after the originating Client is dropped, so a borrow here + // would dangle into freed Client memory the moment drop stopped being + // an instant kill. + sender: Arc>, } impl Request { @@ -52,7 +40,7 @@ impl Request { pub fn new( tag: usize, messages: RequestMessages, - sender: RefOrValue<'static, spsc::Sender>, + sender: Arc>, ) -> Request { Request { tag, @@ -64,21 +52,36 @@ impl Request { pub struct Response { tag: usize, - tx: RefOrValue<'static, spsc::Sender>, + tx: Arc>, } /// A connection to a PostgreSQL database. pub(crate) struct Connection { - io_handle: JoinHandle<()>, + /// Held only to keep the io coroutine owned; dropping it DETACHES. The + /// coroutine is never cancelled (cancel skips destructors, leaking the + /// socket) and never joined (drop can run inside a coroutine that is + /// itself being cancelled, where parking to join aborts the process). + _io_handle: JoinHandle<()>, req_queue: Arc>, waker: WaitIoWaker, + /// Cooperative shutdown: `drop` sets this and wakes the coroutine; the + /// connection loop polls it and exits on its own stack, closing its own + /// socket. Reaching in from outside (cancelling, or shutting the fd down + /// from `drop`) corrupts the coroutine mid-flight. + shutdown: Arc, id: usize, } impl Drop for Connection { fn drop(&mut self) { - let rx = self.io_handle.coroutine(); - unsafe { rx.cancel() }; + // Signal the loop and wake it out of wait_io(); it exits on its own + // stack, sends the protocol Terminate and closes its own socket. This + // must NOT block: a Client can be dropped inside a coroutine that is + // itself being cancelled, and parking there (to join) aborts. The + // waker holds an Arc to the reactor slot, so waking after the loop + // has already exited is a harmless no-op. + self.shutdown.store(true, Ordering::Release); + self.waker.wakeup(); } } @@ -232,6 +235,7 @@ fn connection_loop( stream: &mut TcpStream, req_queue: Arc>, mut params: HashMap, + shutdown: &AtomicBool, ) -> Result<(), Error> { let mut read_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut write_buf = BytesMut::with_capacity(IO_BUF_SIZE); @@ -239,6 +243,12 @@ fn connection_loop( let mut io_events = 1; // allow read loop { + // Cooperative shutdown, set by Connection::drop. Exiting HERE - + // rather than being cancelled from outside - is what guarantees the + // stream is dropped on this stack with destructors run. + if shutdown.load(Ordering::Acquire) { + return Ok(()); + } let inner_stream = stream.inner_mut(); process_req(inner_stream, &req_queue, &mut rsp_queue, &mut write_buf).map_err(Error::io)?; @@ -262,17 +272,24 @@ impl Connection { let req_queue = Arc::new(Queue::new()); let req_queue_dup = req_queue.clone(); + let shutdown = Arc::new(AtomicBool::new(false)); + let shutdown_dup = shutdown.clone(); let io_handle = go!(move || { - if let Err(e) = connection_loop(&mut stream, req_queue_dup, parameters) { + if let Err(e) = connection_loop(&mut stream, req_queue_dup, parameters, &shutdown_dup) { log::error!("connection error = {:?}", e); - terminate_connection(&mut stream); } + // On BOTH exit paths: send the protocol Terminate and close the + // socket, on this coroutine-s own stack. This is what releases the + // server backend the moment the client goes away, instead of at + // process exit. + terminate_connection(&mut stream); }); Connection { - io_handle, + _io_handle: io_handle, req_queue, waker, + shutdown, id, } } diff --git a/tests/test/main.rs b/tests/test/main.rs index dd9d3d1..b61e38f 100644 --- a/tests/test/main.rs +++ b/tests/test/main.rs @@ -710,3 +710,36 @@ fn query_opt() { .unwrap(); client.query_one("SELECT * FROM foo", &[]).err().unwrap(); } + +#[test] +fn dropping_a_client_closes_the_server_backend() { + // The probe connections carry their own application_name so this count is + // immune to every other test opening and dropping connections in parallel. + const TAG: &str = "drop_leak_probe"; + let observer = connect("user=postgres dbname=postgres"); + let backends = || -> i64 { + observer + .query_one( + "SELECT count(*) FROM pg_stat_activity WHERE application_name = $1", + &[&TAG], + ) + .unwrap() + .get(0) + }; + + for _ in 0..5 { + let client = connect(&format!( + "user=postgres dbname=postgres application_name={TAG}" + )); + client.query("SELECT 1", &[]).unwrap(); + drop(client); + // The server reaps the backend once the socket closes; give it a moment. + std::thread::sleep(std::time::Duration::from_millis(250)); + } + + let after = backends(); + assert_eq!( + after, 0, + "{after} dropped client(s) left their backend open" + ); +} From 3b4fdb0a2c245f7c8c55e8a8d34aaea7ecf0d654 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Sun, 2 Aug 2026 16:33:11 +0300 Subject: [PATCH 08/14] ci: add GitHub Actions workflow to run the test suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit There is currently no CI, so nothing checks a change before it lands. Workflow test — starts PostgreSQL, runs `cargo test --all` against 14 and 16 lint — `cargo fmt --check` and `cargo clippy --all-targets` Test fixtures The suite exercises each of PostgreSQL's password authentication methods and needs `citext` and `hstore`, none of which the stock image provides. Those are set up by ci/initdb/00-setup.sh, mounted into the container's docker-entrypoint-initdb.d hook: - md5_user, pass_user, scram_user, each with the matching verifier - a pg_hba.conf routing each role to its method pg_hba is written wholesale rather than appended, because it is first-match-wins and appending would leave the per-role lines behind the image's catch-all. Notes on the shape PostgreSQL is started with `docker run` rather than a `services:` block: the container needs fixture scripts from the repository, and service containers are created before checkout runs. The readiness check waits for a fixture role to exist, not just for pg_isready. The entrypoint restarts the server after running initdb scripts, so pg_isready alone can pass against the temporary instance and race the restart. Jobs carry `timeout-minutes`. The suite talks to the database over a coroutine-driven connection, where a protocol-level stall presents as a hang rather than a failure; without a limit a stalled run would occupy a runner for six hours instead of going red. Clippy is not `-D warnings`. That would fail on pre-existing lints unrelated to whatever change is under review. Warnings stay in the log and can be promoted once the tree is clean. Verified locally with nektos/act. With these fixtures all nine authentication tests and both extension tests pass, having previously been unrunnable. Expected result on this branch This will not go green yet: `copy_in_error`, `copy_in_large` and the `binary_copy` write tests hang on this base commit. That is the deadlock fixed in the COPY IN pull request, not a defect in this workflow — the job-level timeout is what turns it into a legible failure. CI should go green once that change lands. --- .github/workflows/ci.yml | 103 +++++++++++++++++++++++++++++++++++++++ ci/initdb/00-setup.sh | 38 +++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 .github/workflows/ci.yml create mode 100755 ci/initdb/00-setup.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..caea076 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,103 @@ +name: CI + +on: + push: + branches: [master] + pull_request: + +env: + CARGO_TERM_COLOR: always + +jobs: + test: + name: Test (PostgreSQL ${{ matrix.postgres }}) + runs-on: ubuntu-latest + # The suite talks to a database over a coroutine-driven connection, so a + # protocol-level stall shows up as a hang rather than a failure. A job-level + # limit turns that into a legible red build instead of a runner burning its + # full six hours. + timeout-minutes: 15 + + strategy: + fail-fast: false + matrix: + # PostgreSQL 14 is deliberately absent. The test binary segfaults + # against it — SIGSEGV after a handful of tests — and it does so on the + # unmodified base commit too, so this is a pre-existing runtime bug + # rather than anything to do with this workflow. A matrix leg that can + # never pass just teaches people to ignore red CI, so 14 stays out until + # that is fixed. Adding it back is a one-line change. + postgres: [16] + + steps: + - uses: actions/checkout@v4 + + # Started with `docker run` rather than a `services:` block because the + # container needs the fixture scripts from the repository, and service + # containers are created before the checkout step runs. + - name: Start PostgreSQL + run: | + # Removed first so a re-run on a persistent runner — or a local + # `act` invocation — does not collide with a container left behind + # by a previous attempt and then silently test against stale fixtures. + docker rm --force postgres 2>/dev/null || true + + docker run --detach --name postgres \ + --publish 127.0.0.1:5433:5432 \ + --env POSTGRES_USER=postgres \ + --env POSTGRES_HOST_AUTH_METHOD=trust \ + --volume "$PWD/ci/initdb:/docker-entrypoint-initdb.d:ro" \ + postgres:${{ matrix.postgres }} + + - name: Wait for PostgreSQL + run: | + for _ in $(seq 1 60); do + if docker exec postgres pg_isready --quiet --username postgres; then + # pg_isready reports the server is accepting connections, but the + # entrypoint restarts it once the initdb scripts have run. Confirm + # a fixture role exists so we do not race that restart. + if docker exec postgres psql --username postgres --tuples-only \ + --command "SELECT 1 FROM pg_roles WHERE rolname = 'scram_user'" \ + | grep --quiet 1; then + echo "PostgreSQL ready with fixtures applied" + exit 0 + fi + fi + sleep 1 + done + echo "PostgreSQL did not become ready" >&2 + docker logs postgres + exit 1 + + - uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + - name: Test + run: cargo test --all + + - name: PostgreSQL logs + if: failure() + run: docker logs postgres + + lint: + name: Lint + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - uses: Swatinem/rust-cache@v2 + + - name: Format + run: cargo fmt --all -- --check + + # Not `-D warnings`: that would make this job fail on pre-existing lints + # unrelated to whichever change is being reviewed. Warnings stay visible in + # the log, and can be promoted to errors once the tree is clean. + - name: Clippy + run: cargo clippy --all-targets diff --git a/ci/initdb/00-setup.sh b/ci/initdb/00-setup.sh new file mode 100755 index 0000000..69c1776 --- /dev/null +++ b/ci/initdb/00-setup.sh @@ -0,0 +1,38 @@ +#!/bin/bash +# Test fixtures for may_postgres' suite, applied by the postgres image's +# docker-entrypoint-initdb.d hook. +# +# The suite exercises each of PostgreSQL's password authentication methods, so +# it needs one role per method and a pg_hba.conf that routes them accordingly. +set -euo pipefail + +psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname postgres <<-'SQL' + -- Extensions used by the type round-trip tests. + CREATE EXTENSION IF NOT EXISTS citext; + CREATE EXTENSION IF NOT EXISTS hstore; + + -- One role per authentication method. `password_encryption` decides how the + -- verifier is stored, which is what makes md5 and scram distinguishable; + -- cleartext `password` auth works against either, so pass_user can reuse md5. + SET password_encryption TO 'md5'; + CREATE ROLE md5_user LOGIN PASSWORD 'password'; + CREATE ROLE pass_user LOGIN PASSWORD 'password'; + + SET password_encryption TO 'scram-sha-256'; + CREATE ROLE scram_user LOGIN PASSWORD 'password'; +SQL + +# Written wholesale rather than appended: pg_hba is first-match-wins, so the +# per-role lines must precede the catch-all that the image generates, and +# appending would leave them unreachable. +cat > "$PGDATA/pg_hba.conf" <<-'HBA' + # TYPE DATABASE USER ADDRESS METHOD + local all all trust + + host all md5_user all md5 + host all pass_user all password + host all scram_user all scram-sha-256 + + # Everything else, including the postgres superuser the tests connect as. + host all all all trust +HBA From c15d6f202782887e67f99de89e10bc5a06827e93 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 27 Oct 2026 20:14:04 +0200 Subject: [PATCH 09/14] client: implement is_closed Client::_is_closed was a stub (`unimplemented!()`, with the intended body commented out), so there was no way to tell that a connection had died. The connection coroutine now sets a flag when its loop exits (an I/O or protocol error, or the server ending the session), and Client::is_closed() returns it, as in tokio-postgres. Test: a connection terminated with pg_terminate_backend reports closed. --- src/client.rs | 5 ++--- src/connection.rs | 14 ++++++++++++++ tests/test/main.rs | 21 +++++++++++++++++++++ 3 files changed, 37 insertions(+), 3 deletions(-) diff --git a/src/client.rs b/src/client.rs index 60a46ea..2f15a1f 100644 --- a/src/client.rs +++ b/src/client.rs @@ -504,9 +504,8 @@ impl Client { /// Determines if the connection to the server has already closed. /// /// In that case, all future queries will fail. - pub fn _is_closed(&self) -> bool { - // self.inner.sender.is_closed() - unimplemented!() + pub fn is_closed(&self) -> bool { + self.inner.sender.is_closed() } #[inline] diff --git a/src/connection.rs b/src/connection.rs index ad3dce2..ad53f33 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -16,6 +16,7 @@ use crate::{Error, Notification}; use std::collections::{HashMap, VecDeque}; use std::io::{self, Read, Write}; use std::ops::Deref; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; const IO_BUF_SIZE: usize = 4096 * 2; @@ -74,6 +75,8 @@ pub(crate) struct Connection { /// Asynchronous notifications delivered by the server, filled by the /// connection coroutine and drained by the client. notifications: Arc>, + /// Set by the connection coroutine when its loop exits. + closed: Arc, waker: WaitIoWaker, id: usize, } @@ -275,6 +278,8 @@ impl Connection { let req_queue_dup = req_queue.clone(); let notifications = Arc::new(Queue::new()); let notifications_dup = notifications.clone(); + let closed = Arc::new(AtomicBool::new(false)); + let closed_dup = closed.clone(); let io_handle = go!(move || { if let Err(e) = connection_loop(&mut stream, req_queue_dup, parameters, notifications_dup) @@ -282,12 +287,14 @@ impl Connection { log::error!("connection error = {:?}", e); terminate_connection(&mut stream); } + closed_dup.store(true, Ordering::Release); }); Connection { io_handle, req_queue, notifications, + closed, waker, id, } @@ -312,4 +319,11 @@ impl Connection { pub fn notifications(&self) -> &Queue { &self.notifications } + + /// Whether the connection coroutine has exited: the connection failed or + /// the server closed it, so no request can complete any more. + #[inline] + pub fn is_closed(&self) -> bool { + self.closed.load(Ordering::Acquire) + } } diff --git a/tests/test/main.rs b/tests/test/main.rs index 541c8a4..c6b36bb 100644 --- a/tests/test/main.rs +++ b/tests/test/main.rs @@ -787,3 +787,24 @@ fn query_opt() { .unwrap(); client.query_one("SELECT * FROM foo", &[]).err().unwrap(); } + +#[test] +fn is_closed_reports_a_terminated_connection() { + let client = connect("user=postgres"); + assert!(!client.is_closed()); + let pid: i32 = client + .query_one("SELECT pg_backend_pid()", &[]) + .unwrap() + .get(0); + + let admin = connect("user=postgres"); + admin + .query_one("SELECT pg_terminate_backend($1)", &[&pid]) + .unwrap(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !client.is_closed() && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + assert!(client.is_closed()); +} From 86456fac0e9785db84e1968ced66529644be572c Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 27 Oct 2026 20:14:04 +0200 Subject: [PATCH 10/14] query: expose the affected-row count on RowStream A statement streamed with `query_raw` ends with a CommandComplete message carrying its row count, but RowStream dropped it, so a caller that pipelines writes through `query_raw` could not learn how many rows each changed without a separate `execute`. RowStream::rows_affected() returns the count once the stream has ended: Some(n) from the CommandComplete tag (Some(0) for an empty query), None before the end, for a suspended portal, or for a tag without a count. Test: an UPDATE without and with RETURNING, and two pipelined streams that each report their own count. --- src/query.rs | 32 ++++++++++++++++++++++++++++--- tests/test/main.rs | 48 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 3 deletions(-) diff --git a/src/query.rs b/src/query.rs index e84b5c6..cebfe8c 100644 --- a/src/query.rs +++ b/src/query.rs @@ -17,6 +17,7 @@ pub fn query( Ok(RowStream { statement, responses, + rows_affected: None, }) } @@ -32,6 +33,7 @@ pub fn query_portal(client: &Client, portal: &Portal, max_rows: i32) -> Result, +} + +impl RowStream { + /// The number of rows the statement affected, from its `CommandComplete` + /// tag (`INSERT 0 3`, `UPDATE 5`, `SELECT 2`, ...). + /// + /// `None` until the stream has been read to its end, and for a portal + /// that was suspended before completing. This lets a statement sent with + /// `query_raw` (for example to pipeline it with others) report its count + /// the way `execute` does, without a second round trip. + pub fn rows_affected(&self) -> Option { + self.rows_affected + } } impl Iterator for RowStream { @@ -135,9 +151,19 @@ impl Iterator for RowStream { Message::DataRow(body) => { return Some(Ok(o_try!(Row::new(self.statement.clone(), body)))) } - Message::EmptyQueryResponse - | Message::CommandComplete(_) - | Message::PortalSuspended => return None, + Message::CommandComplete(body) => { + self.rows_affected = body + .tag() + .ok() + .and_then(|tag| tag.rsplit(' ').next()) + .and_then(|n| n.parse().ok()); + return None; + } + Message::EmptyQueryResponse => { + self.rows_affected = Some(0); + return None; + } + Message::PortalSuspended => return None, Message::ErrorResponse(body) => return Some(Err(Error::db(body))), Message::BindComplete => {} _ => return Some(Err(Error::unexpected_message())), diff --git a/tests/test/main.rs b/tests/test/main.rs index dd9d3d1..3ba0cee 100644 --- a/tests/test/main.rs +++ b/tests/test/main.rs @@ -627,6 +627,54 @@ fn query_portal() { assert_eq!(r3.len(), 0); } +#[test] +fn row_stream_reports_rows_affected() { + let client = connect("user=postgres"); + client + .batch_execute( + "CREATE TEMPORARY TABLE rows_affected_test (id INT PRIMARY KEY, n INT); + INSERT INTO rows_affected_test VALUES (1, 0), (2, 0), (3, 0);", + ) + .unwrap(); + + // A write without RETURNING: no rows, but a count, as `execute` reports. + let stmt = client + .prepare("UPDATE rows_affected_test SET n = n + 1 WHERE id <= $1") + .unwrap(); + let mut stream = client.query_raw(&stmt, &[&2i32]).unwrap(); + assert_eq!( + stream.rows_affected(), + None, + "unknown until the stream ends" + ); + assert!(stream.next().is_none()); + assert_eq!(stream.rows_affected(), Some(2)); + + // With RETURNING: the rows and the count. + let stmt = client + .prepare("UPDATE rows_affected_test SET n = n + 1 RETURNING id") + .unwrap(); + let mut stream = client.query_raw(&stmt, &[]).unwrap(); + let rows = stream.by_ref().collect::, _>>().unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!(stream.rows_affected(), Some(3)); + + // Pipelined: both statements are sent before either is read, and each + // stream reports its own count. + let insert = client + .prepare("INSERT INTO rows_affected_test VALUES ($1, 0)") + .unwrap(); + let delete = client + .prepare("DELETE FROM rows_affected_test WHERE id <= $1") + .unwrap(); + let mut a = client.query_raw(&insert, &[&10i32]).unwrap(); + let mut b = client.query_raw(&delete, &[&3i32]).unwrap(); + assert!(a.next().is_none()); + assert!(b.next().is_none()); + assert_eq!(a.rows_affected(), Some(1)); + assert_eq!(b.rows_affected(), Some(3)); +} + #[test] fn check_send() { fn is_send(_: &T) {} From 3a3a19a785ce3a9ffd8f61141cdd308d5dfbda06 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 27 Oct 2026 20:14:04 +0200 Subject: [PATCH 11/14] client: deliver LISTEN/NOTIFY notifications The connection decoded NotificationResponse and dropped it, so a client could issue LISTEN but never observe a notification. Notification and AsyncMessage were left in lib.rs from the tokio-postgres port, with a commented-out test, but nothing constructed them. The connection coroutine now pushes each notification onto an unbounded queue (may's mpsc Queue, as used for requests), which Client::notifications() exposes for the caller to drain: client.batch_execute("LISTEN orders")?; while let Some(n) = client.notifications().pop() { println!("{}: {}", n.channel(), n.payload()); } The commented-out test, written against tokio-postgres's poll_message stream, is replaced by tests for delivery from another session, order and payloads, channel isolation, and nothing delivered without LISTEN. --- src/client.rs | 30 ++++++++++- src/connection.rs | 32 +++++++++-- tests/test/main.rs | 131 +++++++++++++++++++++++++++++++++++---------- 3 files changed, 160 insertions(+), 33 deletions(-) diff --git a/src/client.rs b/src/client.rs index 301d7b4..60a46ea 100644 --- a/src/client.rs +++ b/src/client.rs @@ -7,8 +7,8 @@ use crate::query::RowStream; use crate::simple_query::SimpleQueryStream; use crate::types::{Oid, ToSql, Type}; use crate::{ - copy_in, copy_out, prepare, query, simple_query, CancelToken, CopyInSink, Error, Row, - SimpleQueryMessage, Statement, ToStatement, Transaction, TransactionBuilder, + copy_in, copy_out, prepare, query, simple_query, CancelToken, CopyInSink, Error, Notification, + Row, SimpleQueryMessage, Statement, ToStatement, Transaction, TransactionBuilder, }; use bytes::{Buf, BytesMut}; use fallible_iterator::FallibleIterator; @@ -174,6 +174,32 @@ impl Clone for Client { } impl Client { + /// Asynchronous notifications delivered by the server. + /// + /// A connection receives notifications after subscribing with `LISTEN`; + /// another session raises them with `NOTIFY` or `pg_notify()`. The + /// connection coroutine queues each one as it arrives, and the caller + /// drains the queue. + /// + /// The queue is unbounded and is only drained by the caller, so a client + /// that issues `LISTEN` and never reads will grow it without limit. + /// + /// ```no_run + /// # fn main() -> Result<(), may_postgres::Error> { + /// let client = may_postgres::connect("host=localhost user=postgres")?; + /// client.batch_execute("LISTEN my_channel")?; + /// + /// while let Some(notification) = client.notifications().pop() { + /// println!("{}: {}", notification.channel(), notification.payload()); + /// } + /// # Ok(()) + /// # } + /// ``` + #[inline] + pub fn notifications(&self) -> &may::queue::mpsc::Queue { + self.inner.sender.notifications() + } + pub(crate) fn new(sender: Connection, process_id: i32, secret_key: i32) -> Client { let co_ch = { let (tx, rx) = spsc::channel(); diff --git a/src/connection.rs b/src/connection.rs index aa5487e..ad3dce2 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -11,7 +11,7 @@ use postgres_protocol::message::frontend; use crate::codec::{BackendMessage, BackendMessages, FrontendMessage}; use crate::copy_in::CopyInReceiver; -use crate::Error; +use crate::{Error, Notification}; use std::collections::{HashMap, VecDeque}; use std::io::{self, Read, Write}; @@ -71,6 +71,9 @@ pub struct Response { pub(crate) struct Connection { io_handle: JoinHandle<()>, req_queue: Arc>, + /// Asynchronous notifications delivered by the server, filled by the + /// connection coroutine and drained by the client. + notifications: Arc>, waker: WaitIoWaker, id: usize, } @@ -136,6 +139,7 @@ fn decode_messages( read_buf: &mut BytesMut, rsp_queue: &mut VecDeque, parameters: &mut HashMap, + notifications: &Queue, ) -> Result<(), Error> { use crate::codec::PostgresCodec; @@ -162,7 +166,13 @@ fn decode_messages( } } BackendMessage::Async(Message::NoticeResponse(_body)) => {} - BackendMessage::Async(Message::NotificationResponse(_body)) => {} + BackendMessage::Async(Message::NotificationResponse(body)) => { + notifications.push(Notification { + process_id: body.process_id(), + channel: body.channel().map_err(Error::parse)?.to_string(), + payload: body.message().map_err(Error::parse)?.to_string(), + }); + } BackendMessage::Async(Message::ParameterStatus(body)) => { parameters.insert( body.name().map_err(Error::parse)?.to_string(), @@ -232,6 +242,7 @@ fn connection_loop( stream: &mut TcpStream, req_queue: Arc>, mut params: HashMap, + notifications: Arc>, ) -> Result<(), Error> { let mut read_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut write_buf = BytesMut::with_capacity(IO_BUF_SIZE); @@ -247,7 +258,7 @@ fn connection_loop( let mut read_blocked = true; if io_events & 1 != 0 { read_blocked = nonblock_read(inner_stream, &mut read_buf).map_err(Error::io)?; - decode_messages(&mut read_buf, &mut rsp_queue, &mut params)?; + decode_messages(&mut read_buf, &mut rsp_queue, &mut params, ¬ifications)?; } io_events = if read_blocked { stream.wait_io() } else { 1 } @@ -262,8 +273,12 @@ impl Connection { let req_queue = Arc::new(Queue::new()); let req_queue_dup = req_queue.clone(); + let notifications = Arc::new(Queue::new()); + let notifications_dup = notifications.clone(); let io_handle = go!(move || { - if let Err(e) = connection_loop(&mut stream, req_queue_dup, parameters) { + if let Err(e) = + connection_loop(&mut stream, req_queue_dup, parameters, notifications_dup) + { log::error!("connection error = {:?}", e); terminate_connection(&mut stream); } @@ -272,6 +287,7 @@ impl Connection { Connection { io_handle, req_queue, + notifications, waker, id, } @@ -288,4 +304,12 @@ impl Connection { pub fn id(&self) -> usize { self.id } + + /// Asynchronous notifications received from the server. + /// + /// Populated by `LISTEN`; see [`crate::Client::notifications`]. + #[inline] + pub fn notifications(&self) -> &Queue { + &self.notifications + } } diff --git a/tests/test/main.rs b/tests/test/main.rs index dd9d3d1..541c8a4 100644 --- a/tests/test/main.rs +++ b/tests/test/main.rs @@ -556,37 +556,114 @@ fn copy_out() { // ); // } -// #[test] -// fn notifications() { -// let (mut client, mut connection) = connect_raw("user=postgres").unwrap(); +#[test] +fn notifications_are_delivered_to_a_listening_client() { + let client = connect("user=postgres"); -// let (tx, rx) = mpsc::unbounded(); -// let stream = stream::poll_fn(move |cx| connection.poll_message(cx)).map_err(|e| panic!(e)); -// let connection = stream.forward(tx).map(|r| r.unwrap()); -// tokio::spawn(connection); + client.batch_execute("LISTEN test_notifications").unwrap(); + assert!( + client.notifications().pop().is_none(), + "nothing has been raised yet" + ); -// client -// .batch_execute( -// "LISTEN test_notifications; -// NOTIFY test_notifications, 'hello'; -// NOTIFY test_notifications, 'world';", -// ) -// .unwrap(); + // Raised from a second session, which is the shape that matters: one + // connection writes, another reacts. + let notifier = connect("user=postgres"); + notifier + .batch_execute("NOTIFY test_notifications, 'hello'") + .unwrap(); -// drop(client); + let notification = + wait_for_notification(&client).expect("a LISTENing client should receive the notification"); -// let notifications = rx -// .filter_map(|m| match m { -// AsyncMessage::Notification(n) => future::ready(Some(n)), -// _ => future::ready(None), -// }) -// .collect::>(); -// assert_eq!(notifications.len(), 2); -// assert_eq!(notifications[0].channel(), "test_notifications"); -// assert_eq!(notifications[0].payload(), "hello"); -// assert_eq!(notifications[1].channel(), "test_notifications"); -// assert_eq!(notifications[1].payload(), "world"); -// } + assert_eq!(notification.channel(), "test_notifications"); + assert_eq!(notification.payload(), "hello"); + assert_ne!( + notification.process_id(), + 0, + "the notifying backend pid should be reported" + ); +} + +#[test] +fn notifications_preserve_order_and_payloads() { + let client = connect("user=postgres"); + client.batch_execute("LISTEN test_ordering").unwrap(); + + let notifier = connect("user=postgres"); + notifier + .batch_execute( + "NOTIFY test_ordering, 'first'; + NOTIFY test_ordering, 'second'; + NOTIFY test_ordering, 'third';", + ) + .unwrap(); + + let mut payloads = Vec::new(); + for _ in 0..3 { + let n = wait_for_notification(&client).expect("all three should arrive"); + payloads.push(n.payload().to_string()); + } + + // A queue that reordered or coalesced messages would be worse than useless + // for change-data-capture, where each row event must be seen exactly once. + assert_eq!(payloads, vec!["first", "second", "third"]); +} + +#[test] +fn only_subscribed_channels_are_queued() { + let client = connect("user=postgres"); + client.batch_execute("LISTEN wanted_channel").unwrap(); + + let notifier = connect("user=postgres"); + notifier + .batch_execute( + "NOTIFY unwanted_channel, 'ignored'; + NOTIFY wanted_channel, 'kept';", + ) + .unwrap(); + + let notification = + wait_for_notification(&client).expect("the subscribed channel should arrive"); + assert_eq!(notification.channel(), "wanted_channel"); + assert_eq!(notification.payload(), "kept"); + assert!( + client.notifications().pop().is_none(), + "a channel we never subscribed to must not be queued" + ); +} + +#[test] +fn a_client_that_never_listens_receives_nothing() { + let client = connect("user=postgres"); + + let notifier = connect("user=postgres"); + notifier + .batch_execute("NOTIFY some_channel, 'nobody is listening'") + .unwrap(); + + client.batch_execute("SELECT 1").unwrap(); + assert!( + client.notifications().pop().is_none(), + "without LISTEN there is nothing to deliver" + ); +} + +/// Poll for a notification, giving the connection round trips to surface it. +/// +/// Notifications arrive asynchronously: the server sends them when it chooses, +/// and the connection coroutine decodes them during I/O. A test that read the +/// queue once, immediately, would race and pass or fail at random. +fn wait_for_notification(client: &Client) -> Option { + for _ in 0..50 { + if let Some(notification) = client.notifications().pop() { + return Some(notification); + } + client.batch_execute("SELECT 1").unwrap(); + may::coroutine::sleep(std::time::Duration::from_millis(20)); + } + None +} #[test] fn query_portal() { From 4a2653a280a2337c000583428952272c8c158aed Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 27 Oct 2026 20:14:04 +0200 Subject: [PATCH 12/14] client: implement is_closed Client::_is_closed was a stub (`unimplemented!()`, with the intended body commented out), so there was no way to tell that a connection had died. The connection coroutine now sets a flag when its loop exits (an I/O or protocol error, or the server ending the session), and Client::is_closed() returns it, as in tokio-postgres. Test: a connection terminated with pg_terminate_backend reports closed. --- src/client.rs | 5 ++--- src/connection.rs | 14 ++++++++++++++ tests/test/main.rs | 21 +++++++++++++++++++++ 3 files changed, 37 insertions(+), 3 deletions(-) diff --git a/src/client.rs b/src/client.rs index 301d7b4..fe22cb9 100644 --- a/src/client.rs +++ b/src/client.rs @@ -478,9 +478,8 @@ impl Client { /// Determines if the connection to the server has already closed. /// /// In that case, all future queries will fail. - pub fn _is_closed(&self) -> bool { - // self.inner.sender.is_closed() - unimplemented!() + pub fn is_closed(&self) -> bool { + self.inner.sender.is_closed() } #[inline] diff --git a/src/connection.rs b/src/connection.rs index aa5487e..e4592c2 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -16,6 +16,7 @@ use crate::Error; use std::collections::{HashMap, VecDeque}; use std::io::{self, Read, Write}; use std::ops::Deref; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; const IO_BUF_SIZE: usize = 4096 * 2; @@ -71,6 +72,8 @@ pub struct Response { pub(crate) struct Connection { io_handle: JoinHandle<()>, req_queue: Arc>, + /// Set by the connection coroutine when its loop exits. + closed: Arc, waker: WaitIoWaker, id: usize, } @@ -262,16 +265,20 @@ impl Connection { let req_queue = Arc::new(Queue::new()); let req_queue_dup = req_queue.clone(); + let closed = Arc::new(AtomicBool::new(false)); + let closed_dup = closed.clone(); let io_handle = go!(move || { if let Err(e) = connection_loop(&mut stream, req_queue_dup, parameters) { log::error!("connection error = {:?}", e); terminate_connection(&mut stream); } + closed_dup.store(true, Ordering::Release); }); Connection { io_handle, req_queue, + closed, waker, id, } @@ -288,4 +295,11 @@ impl Connection { pub fn id(&self) -> usize { self.id } + + /// Whether the connection coroutine has exited: the connection failed or + /// the server closed it, so no request can complete any more. + #[inline] + pub fn is_closed(&self) -> bool { + self.closed.load(Ordering::Acquire) + } } diff --git a/tests/test/main.rs b/tests/test/main.rs index dd9d3d1..682cef0 100644 --- a/tests/test/main.rs +++ b/tests/test/main.rs @@ -710,3 +710,24 @@ fn query_opt() { .unwrap(); client.query_one("SELECT * FROM foo", &[]).err().unwrap(); } + +#[test] +fn is_closed_reports_a_terminated_connection() { + let client = connect("user=postgres"); + assert!(!client.is_closed()); + let pid: i32 = client + .query_one("SELECT pg_backend_pid()", &[]) + .unwrap() + .get(0); + + let admin = connect("user=postgres"); + admin + .query_one("SELECT pg_terminate_backend($1)", &[&pid]) + .unwrap(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !client.is_closed() && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + assert!(client.is_closed()); +} From e62970cb1e50d0eedce192a2391858650e8e53b7 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 27 Oct 2026 20:14:04 +0200 Subject: [PATCH 13/14] client: add next_notification to wait for a notification Client::notifications() is a queue the caller has to poll. Add Client::next_notification(timeout), which parks until a notification is delivered, the timeout passes, or the connection closes. The notification queue is paired with a semaphore posted on every push, and once more when the connection coroutine exits. A waiter returns queued notifications first, wakes on delivery, returns None at its timeout, and returns promptly once the connection is closed. Waits are never shorter than 1 ms, because may (up to 0.3.51) treats a park timeout under 1 ms as no timeout. Tests: a parked waiter wakes on arrival; queued notifications come back in order, then an empty wait lasts about its timeout; a terminated connection releases the waiter; sub-millisecond timeouts return. --- src/client.rs | 57 ++++++++++++++++++++++- src/connection.rs | 49 +++++++++++++++++--- tests/test/main.rs | 112 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 210 insertions(+), 8 deletions(-) diff --git a/src/client.rs b/src/client.rs index 2f15a1f..398cc61 100644 --- a/src/client.rs +++ b/src/client.rs @@ -182,7 +182,9 @@ impl Client { /// drains the queue. /// /// The queue is unbounded and is only drained by the caller, so a client - /// that issues `LISTEN` and never reads will grow it without limit. + /// that issues `LISTEN` and never reads will grow it without limit. To wait + /// for the next notification rather than poll, use + /// [`Client::next_notification`]. /// /// ```no_run /// # fn main() -> Result<(), may_postgres::Error> { @@ -197,7 +199,58 @@ impl Client { /// ``` #[inline] pub fn notifications(&self) -> &may::queue::mpsc::Queue { - self.inner.sender.notifications() + self.inner.sender.notifications().queue() + } + + /// Wait up to `timeout` for the next notification. + /// + /// Returns as soon as one is queued (or immediately if one already is), + /// parking the coroutine (or thread) in the meantime, so a `LISTEN` + /// consumer needs no polling interval. Returns `None` when `timeout` + /// passes with nothing delivered, or when the connection has closed and + /// nothing is left in the queue. + /// + /// ```no_run + /// # fn main() -> Result<(), may_postgres::Error> { + /// # use std::time::Duration; + /// let client = may_postgres::connect("host=localhost user=postgres")?; + /// client.batch_execute("LISTEN my_channel")?; + /// + /// while !client.is_closed() { + /// if let Some(n) = client.next_notification(Duration::from_secs(1)) { + /// println!("{}: {}", n.channel(), n.payload()); + /// } + /// } + /// # Ok(()) + /// # } + /// ``` + pub fn next_notification(&self, timeout: std::time::Duration) -> Option { + let notifications = self.inner.sender.notifications(); + let deadline = std::time::Instant::now() + timeout; + loop { + if let Some(n) = notifications.queue().pop() { + return Some(n); + } + if self.is_closed() { + // Pass the wake-up on, so every waiter on this connection + // returns rather than one of them. + notifications.ready().post(); + return notifications.queue().pop(); + } + let left = deadline.saturating_duration_since(std::time::Instant::now()); + if left.is_zero() { + return notifications.queue().pop(); + } + // may keeps park timeouts in whole milliseconds and, up to 0.3.51, + // truncates a shorter one to "no timeout", so a wait under 1 ms + // would never return: never ask for less than 1 ms. + let left = left.max(std::time::Duration::from_millis(1)); + if !notifications.ready().wait_timeout(left) { + return notifications.queue().pop(); + } + // A permit can outnumber queued notifications when the queue was + // also drained with `notifications()`; loop and re-check. + } } pub(crate) fn new(sender: Connection, process_id: i32, secret_key: i32) -> Client { diff --git a/src/connection.rs b/src/connection.rs index ad53f33..2a055b2 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -5,7 +5,7 @@ use may::go; use may::io::{WaitIo, WaitIoWaker}; use may::net::TcpStream; use may::queue::mpsc::Queue; -use may::sync::spsc; +use may::sync::{spsc, Semphore}; use postgres_protocol::message::backend::Message; use postgres_protocol::message::frontend; @@ -68,13 +68,48 @@ pub struct Response { tx: RefOrValue<'static, spsc::Sender>, } +/// Asynchronous notifications delivered by the server: a queue the connection +/// coroutine fills and the client drains, and a count of pushes a consumer can +/// park on instead of polling the queue. +pub(crate) struct Notifications { + queue: Queue, + ready: Semphore, +} + +impl Notifications { + fn new() -> Self { + Notifications { + queue: Queue::new(), + ready: Semphore::new(0), + } + } + + fn push(&self, notification: Notification) { + self.queue.push(notification); + self.ready.post(); + } + + /// Wake a consumer without a notification (the connection closed). + fn wake(&self) { + self.ready.post(); + } + + pub(crate) fn queue(&self) -> &Queue { + &self.queue + } + + pub(crate) fn ready(&self) -> &Semphore { + &self.ready + } +} + /// A connection to a PostgreSQL database. pub(crate) struct Connection { io_handle: JoinHandle<()>, req_queue: Arc>, /// Asynchronous notifications delivered by the server, filled by the /// connection coroutine and drained by the client. - notifications: Arc>, + notifications: Arc, /// Set by the connection coroutine when its loop exits. closed: Arc, waker: WaitIoWaker, @@ -142,7 +177,7 @@ fn decode_messages( read_buf: &mut BytesMut, rsp_queue: &mut VecDeque, parameters: &mut HashMap, - notifications: &Queue, + notifications: &Notifications, ) -> Result<(), Error> { use crate::codec::PostgresCodec; @@ -245,7 +280,7 @@ fn connection_loop( stream: &mut TcpStream, req_queue: Arc>, mut params: HashMap, - notifications: Arc>, + notifications: Arc, ) -> Result<(), Error> { let mut read_buf = BytesMut::with_capacity(IO_BUF_SIZE); let mut write_buf = BytesMut::with_capacity(IO_BUF_SIZE); @@ -276,11 +311,12 @@ impl Connection { let req_queue = Arc::new(Queue::new()); let req_queue_dup = req_queue.clone(); - let notifications = Arc::new(Queue::new()); + let notifications = Arc::new(Notifications::new()); let notifications_dup = notifications.clone(); let closed = Arc::new(AtomicBool::new(false)); let closed_dup = closed.clone(); let io_handle = go!(move || { + let waiters = notifications_dup.clone(); if let Err(e) = connection_loop(&mut stream, req_queue_dup, parameters, notifications_dup) { @@ -288,6 +324,7 @@ impl Connection { terminate_connection(&mut stream); } closed_dup.store(true, Ordering::Release); + waiters.wake(); }); Connection { @@ -316,7 +353,7 @@ impl Connection { /// /// Populated by `LISTEN`; see [`crate::Client::notifications`]. #[inline] - pub fn notifications(&self) -> &Queue { + pub(crate) fn notifications(&self) -> &Notifications { &self.notifications } diff --git a/tests/test/main.rs b/tests/test/main.rs index c6b36bb..3c8fd45 100644 --- a/tests/test/main.rs +++ b/tests/test/main.rs @@ -808,3 +808,115 @@ fn is_closed_reports_a_terminated_connection() { } assert!(client.is_closed()); } + +#[test] +fn next_notification_wakes_a_waiting_client_without_polling() { + let client = connect("user=postgres"); + client.batch_execute("LISTEN test_wakeup").unwrap(); + + // Raise the notification from another coroutine after the waiter is parked; + // the waiter issues no request in the meantime, so only the connection's + // own read can deliver it. + let notify = may::go!(|| { + may::coroutine::sleep(std::time::Duration::from_millis(200)); + let notifier = connect("user=postgres"); + notifier + .batch_execute("NOTIFY test_wakeup, 'woken'") + .unwrap(); + }); + + let started = std::time::Instant::now(); + let n = client + .next_notification(std::time::Duration::from_secs(10)) + .expect("the notification should wake the waiter"); + let waited = started.elapsed(); + notify.join().unwrap(); + + assert_eq!(n.payload(), "woken"); + assert!( + waited >= std::time::Duration::from_millis(150), + "returned before the notification was raised: {waited:?}" + ); + assert!( + waited < std::time::Duration::from_secs(5), + "the waiter must wake on arrival, not at its timeout: {waited:?}" + ); +} + +#[test] +fn next_notification_returns_queued_notifications_first_and_times_out_empty() { + let client = connect("user=postgres"); + client.batch_execute("LISTEN test_queued").unwrap(); + let notifier = connect("user=postgres"); + notifier + .batch_execute("NOTIFY test_queued, 'a'; NOTIFY test_queued, 'b';") + .unwrap(); + + let a = client + .next_notification(std::time::Duration::from_secs(5)) + .expect("first"); + let b = client + .next_notification(std::time::Duration::from_secs(5)) + .expect("second"); + assert_eq!((a.payload(), b.payload()), ("a", "b")); + + let started = std::time::Instant::now(); + assert!(client + .next_notification(std::time::Duration::from_millis(100)) + .is_none()); + let waited = started.elapsed(); + assert!( + waited >= std::time::Duration::from_millis(90) + && waited < std::time::Duration::from_secs(2), + "an empty wait lasts about its timeout: {waited:?}" + ); +} + +#[test] +fn next_notification_returns_when_the_connection_closes() { + let client = connect("user=postgres"); + client.batch_execute("LISTEN test_closed").unwrap(); + let pid: i32 = client + .query_one("SELECT pg_backend_pid()", &[]) + .unwrap() + .get(0); + + let killer = may::go!(move || { + may::coroutine::sleep(std::time::Duration::from_millis(200)); + let admin = connect("user=postgres"); + admin + .query_one("SELECT pg_terminate_backend($1)", &[&pid]) + .unwrap(); + }); + + let started = std::time::Instant::now(); + assert!(client + .next_notification(std::time::Duration::from_secs(10)) + .is_none()); + killer.join().unwrap(); + assert!( + started.elapsed() < std::time::Duration::from_secs(5), + "a closed connection must release the waiter, not hold it to its timeout" + ); + assert!(client.is_closed()); +} + +#[test] +fn next_notification_returns_for_sub_millisecond_timeouts() { + // In a coroutine, so the wait parks on may's timer. + let (tx, rx) = std::sync::mpsc::channel(); + let _waiter = may::go!(move || { + let client = connect("user=postgres"); + client.batch_execute("LISTEN test_short_wait").unwrap(); + for us in [1u64, 100, 500, 999, 1500] { + assert!(client + .next_notification(std::time::Duration::from_micros(us)) + .is_none()); + } + tx.send(()).unwrap(); + }); + assert!( + rx.recv_timeout(std::time::Duration::from_secs(10)).is_ok(), + "a sub-millisecond wait did not return" + ); +} From 64de36f18cbc12d123559b8e8f4707342ce12acd Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Wed, 28 Oct 2026 12:39:01 +0200 Subject: [PATCH 14/14] docs: CONTRIBUTING: the integration branch is integration/microscaler-fork Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Y3TSRmEZAYU6RBUXhkn7WF --- CONTRIBUTING.md | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 448e728..6a8e54e 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1,17 +1,20 @@ # Contributing to this fork -This repository is a fork. Its integration branch is `master`: it carries every +This repository is a fork. Its integration branch is `integration/microscaler-fork`: it carries every change made in this fork, and it is the only branch that projects depending on the fork should consume (a dependent can follow just one branch). -- **Changes land here first.** Every fix or feature goes to `master` through a pull +- **Changes land here first.** Every fix or feature goes to `integration/microscaler-fork` through a pull request merged with a merge commit. Branches that dependents pin are never rebased or force-pushed. -- **Upstream proposals are cut separately.** Once a change is on `master`, the +- **Upstream proposals are cut separately.** Once a change is on `integration/microscaler-fork`, the pull request for the upstream project is cut fresh from upstream's default branch as a single-purpose change (one concern, one commit, following - upstream's conventions). It is not merged from `master`. + upstream's conventions). It is not merged from `integration/microscaler-fork`. - **Upstream is merged in regularly.** Upstream's default branch is merged into - `master` so the fork stays a superset of upstream. + `integration/microscaler-fork` so the fork stays a superset of upstream. - **Before an upstream pull request merges or closes,** check that its content is - on `master` (by content, not commit hashes: upstream branches are re-cut). + on `integration/microscaler-fork` (by content, not commit hashes: upstream branches are re-cut). + +`master` is kept as it was when the integration branch was created, because existing lockfiles pin its +commits; new work goes to `integration/microscaler-fork`.