Fix PgEncoder to fully close the connection on idle timeout (#1760) (#1761)

See #1755

The `PgEncoder` close override sends the Terminate message but was not propagating the close, so the channel never fully closed when `readIdleTimeout` fired on an unresponsive connection.

Propagate `ctx.close(promise)` when `closeSent` is false; otherwise wire the promise to `ch.closeFuture()` to let the normal `shutdownOutput` cycle complete without cutting off in-flight server responses.

Add `testReadIdleTimeoutClosesConnection` using a Vert.x `NetServer` with `ALLOW_HALF_CLOSURE` to simulate an unresponsive server that doesn't send FIN back.

Fix `testMaximumParametersExceeded` to use `asyncAssertFailure` so the VertxUnit runner waits for the query result.
Without this, the runner tears down the `Vertx` instance immediately after the test method returns, closing the PG connection before the server response arrives, which changes the failure reason from a parameter count error to a connection lost error.

Assisted-By: Claude Opus 4.6


(cherry picked from commit faf9310)

Signed-off-by: Thomas Segismont <tsegismont@gmail.com>

Author: tsegismont

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgEncoder.java

@@ -23,13 +23,15 @@
 import io.netty.channel.ChannelOutboundHandlerAdapter;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.socket.SocketChannel;
-import io.vertx.sqlclient.Tuple;
+import io.netty.util.concurrent.PromiseNotifier;
 import io.vertx.pgclient.impl.util.Util;
+import io.vertx.sqlclient.Tuple;
 import io.vertx.sqlclient.impl.HexSequence;
 import io.vertx.sqlclient.internal.RowDescriptorBase;
 import io.vertx.sqlclient.spi.protocol.CloseConnectionCommand;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Map;
 
 import static io.vertx.pgclient.impl.util.Util.writeCString;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -98,6 +100,9 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exce
       CloseConnectionCommand cmd = CloseConnectionCommand.INSTANCE;
       PgCommandMessage<?, ?> codec = PgCommandMessage.wrap(cmd);
       codec.encode(this);
+      ctx.close(promise);
+    } else {
+      ctx.channel().closeFuture().addListener(new PromiseNotifier<>(promise));
     }
   }
 

vertx-pg-client/src/test/java/io/vertx/tests/pgclient/CloseConnectionTest.java

@@ -1,19 +1,26 @@
 package io.vertx.tests.pgclient;
 
 import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.ChannelOption;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.internal.net.NetSocketInternal;
 import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.pgclient.PgBuilder;
+import io.vertx.pgclient.PgConnectOptions;
 import io.vertx.pgclient.PgConnection;
+import io.vertx.pgclient.SslMode;
 import io.vertx.sqlclient.Pool;
 import io.vertx.sqlclient.PoolOptions;
 import io.vertx.tests.sqlclient.ProxyServer;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+
 public class CloseConnectionTest extends PgTestBase {
 
   protected Vertx vertx;
@@ -74,6 +81,51 @@ private void testCloseConnection(TestContext ctx, Runnable test) {
     }));
   }
 
+  @Test
+  public void testReadIdleTimeoutClosesConnection(TestContext ctx) {
+    Async async = ctx.async();
+    vertx.createNetServer().connectHandler(socket -> {
+      // Enable half-close so the server keeps the connection open after receiving FIN,
+      // otherwise Vert.x auto-closes and sends FIN back, masking the bug
+      ((NetSocketInternal) socket).channelHandlerContext().channel().config()
+        .setOption(ChannelOption.ALLOW_HALF_CLOSURE, true);
+      socket.handler(data -> {
+        // Received startup message, send PG handshake response then stay silent
+        Buffer response = Buffer.buffer()
+          .appendByte((byte) 'R').appendInt(8).appendInt(0)
+          .appendByte((byte) 'S').appendInt(25)
+          .appendBytes("client_encoding\0UTF8\0".getBytes(StandardCharsets.UTF_8))
+          .appendByte((byte) 'K').appendInt(12).appendInt(1).appendInt(0)
+          .appendByte((byte) 'Z').appendInt(5).appendByte((byte) 'I');
+        socket.write(response);
+        socket.handler(null);
+      });
+    }).listen(0, "localhost").onComplete(ctx.asyncAssertSuccess(server -> {
+      PgConnectOptions connectOptions = new PgConnectOptions()
+        .setHost("localhost")
+        .setPort(server.actualPort())
+        .setUser("test")
+        .setPassword("test")
+        .setDatabase("test")
+        .setSslMode(SslMode.DISABLE);
+      Pool pool = PgBuilder.pool()
+        .connectingTo(connectOptions)
+        .with(new PoolOptions().setMaxSize(1))
+        .with(new io.vertx.core.net.NetClientOptions()
+          .setReadIdleTimeout(200)
+          .setIdleTimeoutUnit(TimeUnit.MILLISECONDS))
+        .using(vertx)
+        .build();
+      pool.getConnection().onComplete(ctx.asyncAssertSuccess(conn ->
+        conn.closeHandler(v -> {
+          pool.close();
+          server.close();
+          async.complete();
+        })
+      ));
+    }));
+  }
+
   @Test
   public void testTransactionInProgressShouldFail(TestContext ctx) {
     ProxyServer proxy = ProxyServer.create(vertx, options.getPort(), options.getHost());

vertx-pg-client/src/test/java/io/vertx/tests/pgclient/tck/PgPreparedQueryTestBase.java

@@ -52,10 +52,9 @@ public void testMaximumParametersExceeded(TestContext ctx) {
     connect(ctx.asyncAssertSuccess(conn -> {
       conn.preparedQuery(query)
         .execute(tuple)
-        .onComplete(ar -> {
-          ctx.assertTrue(ar.failed());
-          ctx.assertTrue(ar.cause().getMessage().contains("exceeds the maximum of 65535"));
-        });
+        .onComplete(ctx.asyncAssertFailure(t -> {
+          ctx.assertTrue(t.getMessage().contains("exceeds the maximum of 65535"));
+        }));
     }));
   }
 }