Pipelining depth grows unbounded when prepared statement is retried with explicit parameter types (#1646) (#1668)

* Increment inflight counter on prepare-with-parametes-types query



* test

* Fix prepared statement reprepare inflight accounting



* Fix prepared statement reprepare inflight accounting



* Fix prepared statement reprepare inflight accounting



* Update vertx-sql-client-codec/src/main/java/io/vertx/sqlclient/codec/SocketConnectionBase.java

---------

Signed-off-by: doxlik <doxlikx@gmail.com>
Co-authored-by: doxlik <168666511+doxlik@users.noreply.github.com>

Author: tsegismont

vertx-pg-client/src/test/java/io/vertx/pgclient/PreparedStatementReprepareTest.java

@@ -0,0 +1,71 @@
+package io.vertx.pgclient;
+
+import io.vertx.core.Vertx;
+import io.vertx.ext.unit.Async;
+import io.vertx.ext.unit.TestContext;
+import io.vertx.pgclient.impl.PgSocketConnection;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowSet;
+import io.vertx.sqlclient.Tuple;
+import io.vertx.sqlclient.impl.SqlConnectionInternal;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class PreparedStatementReprepareTest extends PgTestBase {
+
+  private Vertx vertx;
+
+  @Before
+  public void setup() throws Exception {
+    super.setup();
+    vertx = Vertx.vertx();
+  }
+
+  @After
+  public void tearDown(TestContext ctx) {
+    vertx.close().onComplete(ctx.asyncAssertSuccess());
+  }
+
+  @Test
+  public void testReprepareDoesNotMakeInflightNegativeWithEnabledCache(TestContext ctx) {
+    testReprepareDoesNotMakeInflightNegative(ctx, true);
+  }
+
+  @Test
+  public void testReprepareDoesNotMakeInflightNegativeWithDisabledCache(TestContext ctx) {
+    testReprepareDoesNotMakeInflightNegative(ctx, false);
+  }
+
+  private void testReprepareDoesNotMakeInflightNegative(TestContext ctx, boolean cachePreparedStatements) {
+    Async async = ctx.async();
+
+    PgConnectOptions options = new PgConnectOptions(this.options)
+      .setCachePreparedStatements(cachePreparedStatements);
+
+    PgConnection.connect(vertx, options).onComplete(ctx.asyncAssertSuccess(conn -> {
+      PgSocketConnection socket = (PgSocketConnection) ((SqlConnectionInternal) conn).unwrap();
+
+      conn
+        .preparedQuery("SELECT CONCAT('HELLO ', $1)")
+        .execute(Tuple.of("WORLD"))
+        .map(rows -> {
+          RowSet<Row> result = rows;
+
+          ctx.assertEquals(1, result.size());
+          ctx.assertEquals("HELLO WORLD", result.iterator().next().getString(0));
+
+          ctx.assertEquals(
+            0,
+            socket.inflight(),
+            "Inflight count should be zero after reprepare query completion " +
+              "(cachePreparedStatements=" + cachePreparedStatements + ")"
+          );
+
+          return rows;
+        })
+        .eventually(() -> conn.close())
+        .onComplete(ctx.asyncAssertSuccess(v -> async.complete()));
+    }));
+  }
+}

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SocketConnectionBase.java

@@ -105,6 +105,11 @@ public ClientMetrics metrics() {
     return metrics;
   }
 
+  // Visible for testing
+  public int inflight() {
+    return inflight;
+  }
+
   @Override
   public TracingPolicy tracingPolicy() {
     return connectOptions().getTracingPolicy();
@@ -305,6 +310,8 @@ private PrepareStatementCommand prepareCommand(ExtendedQueryCommand<?> queryCmd,
         Throwable cause = ar.cause();
         if (queryCmd.autoCommit() && isIndeterminatePreparedStatementError(cause) && !sendParameterTypes) {
           ChannelHandlerContext ctx = socket.channelHandlerContext();
+          // We need to increment inflight because a new prepare command will be submitted
+          inflight++;
           // We cannot cache this prepared statement because it might be executed with another type
           ctx.write(prepareCommand(queryCmd, false, true), ctx.voidPromise());
           ctx.flush();