Integration/microscaler fork

.github/workflows/ci.yml

@@ -2,7 +2,7 @@ name: CI
 
 on:
   push:
-    branches: [master]
+    branches: [master, integration/microscaler-fork]
   pull_request:
 
 env:
@@ -74,6 +74,9 @@ jobs:
       - uses: Swatinem/rust-cache@v2
 
       - name: Test
+        env:
+          # severed_connection proxies to this port (defaults to 5432).
+          TEST_PG_PORT: "5433"
         run: cargo test --all
 
       - name: PostgreSQL logs

.gitignore

@@ -4,3 +4,6 @@ Cargo.lock
 .idea
 *.iml
 .vscode
+
+# NFS silly-rename leftovers
+.nfs*

.pre-commit-config.yaml

@@ -0,0 +1,18 @@
+# Guards against pushing what would fail a Rust build or review: install once per checkout with
+#   pre-commit install
+# Run on everything: pre-commit run --all-files
+repos:
+  - repo: local
+    hooks:
+      - id: cargo-fmt
+        name: cargo fmt --check
+        entry: cargo fmt --all -- --check
+        language: system
+        files: \.rs$
+        pass_filenames: false
+      - id: cargo-clippy
+        name: cargo clippy -D warnings
+        entry: cargo clippy --all-targets -- -D warnings
+        language: system
+        files: (\.rs|Cargo\.toml|Cargo\.lock)$
+        pass_filenames: false

CONTRIBUTING.md

@@ -0,0 +1,33 @@
+# Contributing to this fork
+
+This repository is a fork with two long-lived branches:
+
+- **`master` tracks upstream.** It mirrors upstream's default branch; nothing of ours is committed to it.
+- **`integration/microscaler-fork` carries every change made in this fork** on top of `master`. It is the
+  default branch and the only branch that projects depending on the fork should consume (a dependent can
+  follow just one branch).
+
+## Making a change
+
+- Every fix or feature goes to `integration/microscaler-fork` first, through a pull request merged with a
+  merge commit.
+- The pull request for the upstream project is cut separately, fresh from upstream's default branch, as a
+  single-purpose change (one concern, one commit, following upstream's conventions). It is not merged
+  from the integration branch.
+- Before an upstream pull request merges or closes, check that its content is on the integration branch
+  (by content, not commit hashes: upstream branches are re-cut).
+
+## After an upstream update
+
+1. Update `master` to upstream's default branch.
+2. **Before rebasing, tag the current integration head and push the tag**, e.g.
+   `git tag integration-2026-10-28 origin/integration/microscaler-fork && git push origin integration-2026-10-28`.
+   Lockfiles pin integration commits; the tag keeps those commits reachable after the rebase, so older
+   releases still rebuild.
+3. Rebase `integration/microscaler-fork` onto `master`, dropping changes upstream now contains, run the
+   tests, and force-push the integration branch (the only branch that is ever force-pushed).
+4. Re-lock every consumer to the new integration head and check each has a single copy of this crate in
+   its dependency graph.
+
+Never reset or rewrite `master` while consumers still pin commits that exist only on it; move them to the
+integration branch (or tag those commits) first.

Cargo.toml

@@ -48,6 +48,9 @@ fallible-iterator = "0.2"
 postgres-protocol = "0.6"
 
 may = { version = "0.3", default-features = false }
+# TLS (sslmode, sslrootcert, sslcert/sslkey). ring: no C toolchain beyond cc, no cmake.
+rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] }
+rustls-pemfile = "2"
 
 [dev-dependencies]
 env_logger = "0.11"
@@ -70,3 +73,7 @@ lto = true
 
 # [patch.crates-io]
 # may = { path = "../may" }
+
+[lints.rust]
+# Test code that upstream disabled on purpose (no unix-socket support in this port; a geo-types version never added).
+unexpected_cfgs = { level = "warn", check-cfg = ["cfg(unix_nosupport)", "cfg(feature, values(\"with-geo-0_10\"))"] }

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()
+    );
+}

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;
@@ -22,10 +22,12 @@ use std::rc::Rc;
 use std::sync::Arc;
 use std::time::Duration;
 
+use crate::connection::ResponseResult;
+
 pub struct Responses {
     tag: usize,
     cur: BackendMessages,
-    rx: Rc<spsc::Receiver<BackendMessages>>,
+    rx: Rc<spsc::Receiver<ResponseResult>>,
 }
 
 impl Responses {
@@ -36,12 +38,16 @@ impl Responses {
                 Some((_, Message::ErrorResponse(body))) => return Err(Error::db(body)),
                 Some((_, message)) => return Ok(message),
                 None => match self.rx.recv() {
-                    Ok(messages) => {
+                    Ok(Ok(messages)) => {
                         if messages.tag != self.tag {
                             continue;
                         }
                         self.cur = messages
                     }
+                    // The connection I/O loop exited and failed this pending
+                    // request. The error is untagged — the connection is dead,
+                    // every outstanding request on it fails the same way.
+                    Ok(Err(e)) => return Err(e),
                     Err(_) => return Err(Error::closed()),
                 },
             }
@@ -63,17 +69,17 @@ pub struct InnerClient {
 
 struct CoChannel {
     tag: Cell<usize>,
-    rx: Rc<spsc::Receiver<BackendMessages>>,
-    tx: spsc::Sender<BackendMessages>,
+    rx: Rc<spsc::Receiver<ResponseResult>>,
+    tx: Arc<spsc::Sender<ResponseResult>>,
 }
 
 impl CoChannel {
-    fn sender(&self) -> RefOrValue<'static, spsc::Sender<BackendMessages>> {
-        // Safety:
-        // 1. there is only one sender
-        // 2. we will wait until all response come back
-        let tx: &'static spsc::Sender<BackendMessages> = unsafe { std::mem::transmute(&self.tx) };
-        RefOrValue::Ref(tx)
+    fn sender(&self) -> Arc<spsc::Sender<ResponseResult>> {
+        // 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 {
@@ -82,7 +88,7 @@ impl CoChannel {
         tag
     }
 
-    fn receiver(&self) -> Rc<spsc::Receiver<BackendMessages>> {
+    fn receiver(&self) -> Rc<spsc::Receiver<ResponseResult>> {
         self.rx.clone()
     }
 }
@@ -91,7 +97,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 +166,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(),
@@ -182,34 +192,76 @@ impl Client {
     /// 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:
+    /// that issues `LISTEN` and never reads will grow it without limit. The
+    /// connection fills it as notifications arrive, without any request from
+    /// the client; to wait for the next one rather than poll, use
+    /// [`Client::next_notification`].
+    #[inline]
+    pub fn notifications(&self) -> &may::queue::mpsc::Queue<Notification> {
+        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")?;
     ///
-    /// 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());
+    /// while !client.is_closed() {
+    ///     if let Some(n) = client.next_notification(Duration::from_secs(1)) {
+    ///         println!("{}: {}", n.channel(), n.payload());
     ///     }
     /// }
+    /// # Ok(())
     /// # }
     /// ```
-    #[inline]
-    pub fn notifications(&self) -> &may::queue::mpsc::Queue<Notification> {
-        self.inner.sender.notifications()
+    pub fn next_notification(&self, timeout: std::time::Duration) -> Option<Notification> {
+        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 releases up to
+            // 0.3.51 truncate anything shorter to "no timeout" (the wait would
+            // never return): never ask it 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().pop()`; loop and re-check.
+        }
     }
 
     pub(crate) fn new(sender: Connection, process_id: i32, secret_key: i32) -> Client {
         let co_ch = {
             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 {
@@ -513,10 +565,17 @@ impl Client {
 
     /// Determines if the connection to the server has already closed.
     ///
-    /// In that case, all future queries will fail.
+    /// In that case, all future queries will fail — fast, with a
+    /// connection-closed error, rather than blocking. Pool implementations can
+    /// use this to retire a dead client before dispatching work to it.
+    pub fn is_closed(&self) -> bool {
+        self.inner.sender.is_dead()
+    }
+
+    #[doc(hidden)]
+    #[deprecated(note = "use `is_closed`")]
     pub fn _is_closed(&self) -> bool {
-        // self.inner.sender.is_closed()
-        unimplemented!()
+        self.is_closed()
     }
 
     #[inline]