Recycle HTTP client lease when an unsent request is reset or cancelled

vertx-core/src/main/java/io/vertx/core/http/impl/http2/DefaultHttp2Stream.java

@@ -182,8 +182,12 @@ public void init(int id, boolean writable) {
   }
 
   public void onClose(GoAway goAway) {
+    close(goAway, true);
+  }
+
+  private void close(GoAway goAway, boolean observe) {
     if (!trailersSent || !trailersReceived) {
-      StreamObserver observer = observer();
+      StreamObserver observer = observe ? observer() : null;
       if (observer != null) {
         observer.observeReset();
       }
@@ -501,6 +505,8 @@ private void writeReset0(long code, Promise<Void> promise) {
         } else {
           // Reset happening before stream allocation
           handleReset(code);
+          // Nothing was sent, so the observer did not begin the request
+          close(null, false);
           promise.complete();
         }
       }

vertx-core/src/main/java/io/vertx/core/http/impl/http3/Http3ClientStream.java

@@ -25,6 +25,7 @@
 import io.vertx.core.http.impl.headers.HttpResponseHeaders;
 import io.vertx.core.http.impl.observability.ClientStreamObserver;
 import io.vertx.core.internal.ContextInternal;
+import io.vertx.core.internal.PromiseInternal;
 import io.vertx.core.internal.net.QuicStreamInternal;
 
 import java.util.Map;
@@ -47,6 +48,32 @@ public Http3ClientStream(Http3ClientConnection connection, QuicStreamInternal st
     this.observer = observer;
   }
 
+  @Override
+  public Future<Void> writeReset(long code) {
+    boolean unsent = !isHeadersSent();
+    Future<Void> fut = super.writeReset(code);
+    return unsent ? fut.compose(v -> closeUnsent((int) code)) : fut;
+  }
+
+  @Override
+  public Future<Boolean> cancel() {
+    boolean unsent = !isHeadersSent();
+    Future<Boolean> fut = super.cancel();
+    return unsent ? fut.compose(res -> closeUnsent(Http3ErrorCode.H3_REQUEST_CANCELLED.code()).map(res)) : fut;
+  }
+
+  /**
+   * Close a stream reset before the request headers were sent, so the stream closes and its connection lease
+   * is recycled: abort the inbound side as well, and close the channel since Netty does not close it after a reset.
+   */
+  private Future<Void> closeUnsent(int code) {
+    return stream.abort(code).compose(v -> {
+      PromiseInternal<Void> promise = context.promise();
+      stream.channelHandlerContext().channel().close().addListener(promise);
+      return promise.future();
+    });
+  }
+
   @Override
   protected HttpHeaders headOf(Http3Headers headers) {
     return new HttpResponseHeaders(headers);

vertx-core/src/main/java/io/vertx/core/http/impl/http3/Http3Stream.java

@@ -360,6 +360,10 @@ public Future<Void> writeReset(long code) {
     return stream.reset((int)code);
   }
 
+  final boolean isHeadersSent() {
+    return headersSent;
+  }
+
   public final Object metric() {
     return observer != null ? observer.metric() : null;
   }

vertx-core/src/test/java/io/vertx/tests/http/HttpTest.java

@@ -5788,6 +5788,39 @@ public void testResetClientRequestResponseInProgress(Checkpoint checkpoint) thro
     }));
   }
 
+  @Test
+  public void testResetClientRequestNotYetSentRecyclesLease() throws Exception {
+    testClientRequestNotYetSentRecyclesLease(req -> req.reset());
+  }
+
+  @Test
+  public void testCancelClientRequestNotYetSentRecyclesLease() throws Exception {
+    testClientRequestNotYetSentRecyclesLease(HttpClientRequest::cancel);
+  }
+
+  private void testClientRequestNotYetSentRecyclesLease(Function<HttpClientRequest, Future<?>> abort) throws Exception {
+    // Allow a single request at a time, the second request can only proceed when the lease is recycled
+    HttpServerConfig serverConfig = config.forServer().config();
+    if (serverConfig.getHttp2Config() != null) {
+      serverConfig.getHttp2Config().getInitialSettings().setMaxConcurrentStreams(1);
+    }
+    serverConfig.getQuicConfig().getTransportConfig().setInitialMaxStreamsBidi(1);
+    server = vertx.httpServerBuilder()
+      .with(serverConfig)
+      .with(config.forServer().sslOptions())
+      .build();
+    server.requestHandler(req -> req.response().end());
+    startServer(testAddress);
+    PoolOptions poolOptions = new PoolOptions().setHttp1MaxSize(1).setHttp2MaxSize(1).setHttp3MaxSize(1);
+    client = config.forClient().create(vertx, poolOptions);
+    HttpClientRequest request = client.request(requestOptions).await();
+    HttpConnection connection = request.connection();
+    abort.apply(request).await();
+    HttpClientRequest next = client.request(requestOptions).await(10, TimeUnit.SECONDS);
+    assertSame(connection, next.connection());
+    next.send().compose(HttpClientResponse::end).await(10, TimeUnit.SECONDS);
+  }
+
   @Test
   public void testResetPartialClientRequest(Checkpoint checkpoint) throws Exception {
     server.requestHandler(req -> {