feat: Drain in-flight requests on stop

HttpServer.stop(delay) can wait out the full delay with nothing in
flight: when a client closes an idle keep-alive connection just before
stop, the JDK never counts its finished latch down (JDK 25 ServerImpl).

OpenApiServer now counts in-flight requests with a filter on every
context it creates. stop(delay) closes the listener at once through a
background HttpServer.stop(delay), waits on the counter for at most
the delay, then calls HttpServer.stop(0). Later and concurrent stop
calls wait for the first. inFlightRequests() exposes the count.

Author: thced

CLAUDE.md

@@ -27,7 +27,7 @@ Request flow when `OpenApiServer` boots (`src/main/java/com/retailsvc/http/OpenA
 
 1. `HttpServer` is created on a port with a virtual-thread-per-task executor.
 2. One `HttpContext` is registered per spec binding at `spec.basePath()` (the first `servers[].url` path from the OpenAPI doc). Unless a binding owns `/`, a catch-all `/` context serves extra routes via `ExtrasRouter` and 404s everything else; `ExceptionFilter` wraps that context only.
-3. On a binding context, two filters run in order, then the handler:
+3. Every context the server creates, the binding contexts and the extras one, starts with an `InFlightCounter` filter that counts the exchange for `OpenApiServer.inFlightRequests()` and `stop(int)`. On a binding context, two more filters run in order, then the handler:
     - `RequestPreparationFilter` — reads the request body through `RequestBodyReader` (which decodes a registered `Content-Encoding` — gzip is built in — under a size cap), resolves the route, runs OpenAPI parameter + body validation via `DefaultValidator`, and binds the resulting `Request` into the `DispatchHandler.CURRENT` scoped value. It renders its own failures through the `ExceptionHandler` rather than relying on `ExceptionFilter`.
     - `SecurityFilter` — enforces the spec's `securitySchemes` / `security`, re-binding the `Request` with resolved principals. It writes its 401/403 responses straight to the exchange.
     - `DispatchHandler` — looks up the `RequestHandler` registered for the resolved `operationId` in the user-supplied map and invokes it, applying interceptors and response decorators. Handler coverage is verified at boot, so the lookup never returns `null`.

README.md

@@ -574,6 +574,22 @@ try (var server = OpenApiServer.builder()
 `stop(int)` and `shutdownTimeoutSeconds(int)` reject negative values with
 `IllegalArgumentException`.
 
+Stopping closes the listening socket at once, so no new connection is accepted. Requests on
+keep-alive connections that are already open are served until the JDK sees no exchange in flight,
+which can come before the after-response runnables of the last request have finished; the JDK then
+closes every connection, keep-alive ones included. The wait ends as soon as `inFlightRequests()` reads `0`, or when the delay elapses; every connection
+is then closed, which cuts off what is still running. A request counts as in flight from the first
+filter until its after-response runnables (`Request.afterResponse`, hooks) have run, so `stop` waits
+for those too. A request the JDK rejects before it dispatches it, such as a malformed one, is not
+counted.
+
+Only the first `stop` call stops the server. A later or concurrent call waits until that stop is
+complete, then returns (or throws `IllegalStateException` if that stop failed); its own delay is
+ignored. A thread interrupted while it waits stops waiting and keeps its interrupt flag.
+
+`inFlightRequests()` is the server's own count, readable at any time, for example to log it or to
+wait for `0` before stopping.
+
 ## Interceptors and response decorators
 
 ### Response decorators

src/main/java/com/retailsvc/http/OpenApiServer.java

@@ -9,6 +9,7 @@
 import com.retailsvc.http.internal.ExceptionFilter;
 import com.retailsvc.http.internal.ExtrasRouter;
 import com.retailsvc.http.internal.FormTypeMapper;
+import com.retailsvc.http.internal.InFlightCounter;
 import com.retailsvc.http.internal.PemSslContext;
 import com.retailsvc.http.internal.RequestBodyReader;
 import com.retailsvc.http.internal.RequestPreparationFilter;
@@ -40,6 +41,10 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import javax.net.ssl.SSLContext;
 import org.slf4j.Logger;
@@ -57,9 +62,13 @@ public class OpenApiServer implements AutoCloseable {
   private static final int DEFAULT_HTTPS_PORT = 8443;
   private static final String JSON = "application/json";
   private static final String GSON_CLASS = "com.google.gson.Gson";
+  private static final long STOP_JOIN_SECONDS = 5;
 
   private final HttpServer httpServer;
   private final int shutdownTimeoutSeconds;
+  private final InFlightCounter counter;
+  private final AtomicBoolean stopping = new AtomicBoolean();
+  private final CompletableFuture<Void> stopped = new CompletableFuture<>();
 
   /** Internal grouping of handler-related configuration to keep the constructor signature small. */
   record HandlerConfig(
@@ -98,8 +107,10 @@ record HandlerConfig(
     this.httpServer = createHttpServer(socketAddress, sslContext);
     httpServer.setExecutor(newThreadPerTaskExecutor(ofVirtual().name("http-", 0).factory()));
 
-    boolean anyBindingAtRoot = wireBindings(httpServer, bindings, bodyMappers, handlerConfig);
-    wireExtras(httpServer, anyBindingAtRoot, handlerConfig);
+    this.counter = new InFlightCounter();
+    boolean anyBindingAtRoot =
+        wireBindings(httpServer, bindings, bodyMappers, handlerConfig, counter);
+    wireExtras(httpServer, anyBindingAtRoot, handlerConfig, counter);
 
     httpServer.start();
     this.shutdownTimeoutSeconds = shutdownTimeoutSeconds;
@@ -120,12 +131,13 @@ private static boolean wireBindings(
       HttpServer httpServer,
       List<SpecBinding> bindings,
       Map<String, TypeMapper> bodyMappers,
-      HandlerConfig handlerConfig) {
+      HandlerConfig handlerConfig,
+      InFlightCounter counter) {
     boolean anyBindingAtRoot = false;
     for (SpecBinding binding : bindings) {
       String basePath = Optional.ofNullable(binding.spec().basePath()).orElse("/");
       anyBindingAtRoot |= "/".equals(basePath);
-      wireBinding(httpServer, basePath, binding, bodyMappers, handlerConfig);
+      wireBinding(httpServer, basePath, binding, bodyMappers, handlerConfig, counter);
     }
     return anyBindingAtRoot;
   }
@@ -135,11 +147,13 @@ private static void wireBinding(
       String basePath,
       SpecBinding binding,
       Map<String, TypeMapper> bodyMappers,
-      HandlerConfig handlerConfig) {
+      HandlerConfig handlerConfig,
+      InFlightCounter counter) {
     Map<String, Operation> operationsById =
         binding.spec().operations().stream()
             .collect(Collectors.toUnmodifiableMap(Operation::operationId, op -> op));
     HttpContext ctx = httpServer.createContext(basePath);
+    ctx.getFilters().add(counter.filter());
     ctx.getFilters()
         .add(
             new RequestPreparationFilter(
@@ -170,7 +184,10 @@ private static void wireBinding(
   }
 
   private static void wireExtras(
-      HttpServer httpServer, boolean anyBindingAtRoot, HandlerConfig handlerConfig) {
+      HttpServer httpServer,
+      boolean anyBindingAtRoot,
+      HandlerConfig handlerConfig,
+      InFlightCounter counter) {
     Map<String, RequestHandler> extras = handlerConfig.extras();
     if (anyBindingAtRoot) {
       if (!extras.isEmpty()) {
@@ -186,6 +203,7 @@ private static void wireExtras(
             handlerConfig.bodyReader(),
             handlerConfig.streamingLimit());
     HttpContext extrasCtx = httpServer.createContext("/", extrasRouter);
+    extrasCtx.getFilters().add(counter.filter());
     extrasCtx
         .getFilters()
         .add(new ExceptionFilter(handlerConfig.exceptionHandler(), handlerConfig.renderer()));
@@ -213,18 +231,90 @@ public InetAddress bindAddress() {
     return httpServer.getAddress().getAddress();
   }
 
+  /**
+   * The number of requests being handled now, from the first filter to the end of the
+   * after-response hooks. A request the JDK rejects before it dispatches it, such as a malformed
+   * one, is not counted. Reads {@code 0} once the server is idle, so a caller can wait for that
+   * before {@link #stop(int)}.
+   */
+  public int inFlightRequests() {
+    return counter.inFlight();
+  }
+
   /**
    * Stops the server, waiting up to {@code delaySeconds} for active exchanges to finish before
    * closing them. {@code 0} stops immediately.
    *
+   * <p>The listening socket closes at once, so no new connection is accepted. Requests on
+   * keep-alive connections that are already open are served until the JDK sees no exchange in
+   * flight, which can come before the after-response runnables of the last request have finished;
+   * the JDK then closes every connection, keep-alive ones included. The wait ends when {@link
+   * #inFlightRequests()} reads {@code 0} or the delay elapses, whichever comes first; the server
+   * then closes every connection, which cuts off whatever is still running. The wait is on this
+   * server's own count, not on the JDK's, which can wait out the whole delay with nothing in
+   * flight.
+   *
+   * <p>Only the first call stops the server. A later call, also a concurrent one, waits until that
+   * stop is complete and then returns, or throws {@link IllegalStateException} if that stop failed;
+   * its own {@code delaySeconds} is ignored. A thread that is interrupted while it waits stops
+   * waiting and keeps its interrupt flag.
+   *
    * @param delaySeconds maximum seconds to wait for in-flight exchanges; must be non-negative
    */
   public void stop(int delaySeconds) {
     if (delaySeconds < 0) {
       throw new IllegalArgumentException("delaySeconds must be non-negative, got " + delaySeconds);
     }
-    if (httpServer != null) {
-      httpServer.stop(delaySeconds);
+    if (stopping.compareAndSet(false, true)) {
+      try {
+        drainAndStop(delaySeconds);
+        stopped.complete(null);
+      } catch (RuntimeException | Error e) {
+        stopped.completeExceptionally(e);
+        throw e;
+      }
+    } else {
+      awaitStopped();
+    }
+  }
+
+  /**
+   * {@code HttpServer.stop(delay)} closes the listening socket at once and then waits on a latch
+   * that, in the JDK this was written against, can stay unreleased with nothing in flight. It runs
+   * on its own thread for the first, and {@code stop(0)} ends it: {@code ServerImpl.stop} is safe
+   * to call again while another call waits. Each call only sets {@code terminating}, posts another
+   * {@code StopRequested} event, closes the already closed socket and waits on the latch; its
+   * {@code finally} counts the latch down for every waiter and clears connections and timers, which
+   * tolerates repeats.
+   */
+  private void drainAndStop(int delaySeconds) {
+    Thread background = ofVirtual().name("http-stop").start(() -> httpServer.stop(delaySeconds));
+    counter.awaitIdle(delaySeconds, TimeUnit.SECONDS);
+    // The JDK swallows an interrupt in stop(0), so take the flag now and set it again at the end.
+    boolean interrupted = Thread.interrupted();
+    httpServer.stop(0);
+    try {
+      background.join(Duration.ofSeconds(STOP_JOIN_SECONDS));
+    } catch (InterruptedException _) {
+      interrupted = true;
+    }
+    if (background.isAlive()) {
+      LOG.warn(
+          "The JDK stop thread is still running {} s after the server was stopped",
+          STOP_JOIN_SECONDS);
+    }
+    if (interrupted) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  private void awaitStopped() {
+    try {
+      stopped.get();
+    } catch (InterruptedException _) {
+      Thread.currentThread().interrupt();
+    } catch (ExecutionException e) {
+      throw new IllegalStateException("The first stop failed", e.getCause());
     }
   }
 

src/main/java/com/retailsvc/http/internal/InFlightCounter.java

@@ -0,0 +1,102 @@
+package com.retailsvc.http.internal;
+
+import com.sun.net.httpserver.Filter;
+import com.sun.net.httpserver.HttpExchange;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Counts the exchanges the server is handling. One counter is shared by every {@code HttpContext}
+ * the server creates, through {@link #filter()}, installed ahead of every other filter so the
+ * counted window covers the whole exchange: the handler, the rendering of the response and the
+ * after-response hooks.
+ *
+ * <p>The server waits on this count instead of on {@code HttpServer.stop(delay)}, because the JDK
+ * can leave that call waiting for the whole delay with nothing in flight: a client that closes an
+ * idle keep-alive connection just as {@code stop} is called leaves the JDK's internal latch
+ * unreleased. Thread-safe.
+ */
+public final class InFlightCounter {
+
+  private final Lock lock = new ReentrantLock();
+  private final Condition idle = lock.newCondition();
+  private final Filter filter = new CountingFilter();
+  private int inFlight;
+
+  /** The filter that counts the exchanges of the context it is added to. Add it first. */
+  public Filter filter() {
+    return filter;
+  }
+
+  /** The number of exchanges being handled now. */
+  public int inFlight() {
+    lock.lock();
+    try {
+      return inFlight;
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  /**
+   * Waits until no exchange is in flight, for at most {@code timeout} of {@code unit}. If the
+   * thread is interrupted the wait ends and the interrupt flag is set again.
+   *
+   * @return whether the count is zero when the wait ends
+   */
+  public boolean awaitIdle(long timeout, TimeUnit unit) {
+    long remaining = unit.toNanos(timeout);
+    lock.lock();
+    try {
+      while (inFlight > 0 && remaining > 0) {
+        remaining = idle.awaitNanos(remaining);
+      }
+    } catch (InterruptedException _) {
+      Thread.currentThread().interrupt();
+    } finally {
+      lock.unlock();
+    }
+    return inFlight() == 0;
+  }
+
+  private void enter() {
+    lock.lock();
+    try {
+      inFlight++;
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  private void exit() {
+    lock.lock();
+    try {
+      if (--inFlight == 0) {
+        idle.signalAll();
+      }
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  private final class CountingFilter extends Filter {
+
+    @Override
+    public void doFilter(HttpExchange exchange, Chain chain) throws IOException {
+      enter();
+      try {
+        chain.doFilter(exchange);
+      } finally {
+        exit();
+      }
+    }
+
+    @Override
+    public String description() {
+      return "In-flight counter";
+    }
+  }
+}