From e0e4fa5b3fc436cc89b359f3c618be93c4d79823 Mon Sep 17 00:00:00 2001
From: Sharang Gupta <sharang@sharanggupta.dev>
Date: Wed, 28 Oct 2026 23:22:46 +0000
Subject: [PATCH] Drop a PgSubscriber connection opened after close

Closing a subscriber while it was connecting, or while a reconnect was
scheduled, left the new connection open and listening.

Signed-off-by: Sharang Gupta <sharang@sharanggupta.dev>
Assisted-by: Anthropic Claude (claude-opus-5-5)
---
 .../impl/pubsub/PgSubscriberImpl.java         | 14 +++++++++++
 .../io/vertx/tests/pgclient/PubSubTest.java   | 25 +++++++++++++++++++
 2 files changed, 39 insertions(+)

diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/pubsub/PgSubscriberImpl.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/pubsub/PgSubscriberImpl.java
index f1dde7640..b45be0e79 100644
--- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/pubsub/PgSubscriberImpl.java
+++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/pubsub/PgSubscriberImpl.java
@@ -162,11 +162,25 @@ private void tryConnect(long delayMillis, Completable<Void> handler) {
   }
 
   private void doConnect(Completable<Void> completionHandler) {
+    synchronized (this) {
+      if (closed) {
+        // Closed while a reconnect was scheduled
+        connecting = false;
+        completionHandler.complete(null, new IllegalStateException("Subscriber closed"));
+        return;
+      }
+    }
     PgConnection.connect(vertx, options).onComplete(ar -> handleConnectResult(completionHandler, ar));
   }
 
   private synchronized void handleConnectResult(Completable<Void> completionHandler, AsyncResult<PgConnection> ar1) {
     connecting = false;
+    if (ar1.succeeded() && closed) {
+      // Closed while connecting: do not keep the connection
+      ar1.result().close();
+      completionHandler.complete(null, new IllegalStateException("Subscriber closed"));
+      return;
+    }
     if (ar1.succeeded()) {
       conn = ar1.result();
       conn.notificationHandler(PgSubscriberImpl.this::handleNotification);
diff --git a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PubSubTest.java b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PubSubTest.java
index af53a6bb5..bac0e7e21 100644
--- a/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PubSubTest.java
+++ b/vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PubSubTest.java
@@ -132,6 +132,31 @@ private void testConnect(TestContext ctx, String channel1Name, String channel2Na
     notifiedLatch.awaitSuccess(10000);
   }
 
+  @Test
+  public void testCloseWhileConnecting(TestContext ctx) {
+    ProxyServer proxy = ProxyServer.create(vertx, options.getPort(), options.getHost());
+    AtomicReference<ProxyServer.Connection> pending = new AtomicReference<>();
+    Async accepted = ctx.async();
+    // Hold the connection until the subscriber is closed
+    proxy.proxyHandler(conn -> {
+      pending.set(conn);
+      accepted.complete();
+    });
+    Async listening = ctx.async();
+    proxy.listen(8080, "localhost", ctx.asyncAssertSuccess(v -> listening.complete()));
+    listening.awaitSuccess(10000);
+    subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions(options).setPort(8080).setHost("localhost"));
+    subscriber.channel("the_channel").handler(notification -> ctx.fail("Notification after close: " + notification));
+    Async connected = ctx.async();
+    subscriber.connect().onComplete(ar -> connected.complete());
+    accepted.awaitSuccess(10000);
+    subscriber.close();
+    pending.get().connect();
+    connected.awaitSuccess(10000);
+    ctx.assertTrue(subscriber.closed());
+    ctx.assertNull(subscriber.actualConnection());
+  }
+
   @Test
   public void testSubscribe(TestContext ctx) {
     testSubscribe(ctx, "the_channel");