Fix a bug where closing a websocket session could send a RST_STREAM frame

core/src/main/java/com/linecorp/armeria/internal/common/Http2ObjectEncoder.java

@@ -121,7 +121,7 @@ protected final boolean isStreamPresentAndWritable(int streamId) {
     }
 
     @Override
-    public final void close(Throwable unused) {
+    public void close(Throwable unused) {
         closed = true;
         keepAliveHandler().destroy();
     }

core/src/main/java/com/linecorp/armeria/internal/common/websocket/WebSocketUtil.java

@@ -45,6 +45,7 @@ public final class WebSocketUtil {
     public static final long DEFAULT_REQUEST_RESPONSE_TIMEOUT_MILLIS = 0;
     public static final long DEFAULT_MAX_REQUEST_RESPONSE_LENGTH = 0;
     public static final long DEFAULT_REQUEST_AUTO_ABORT_DELAY_MILLIS = 5000;
+    public static final long DEFAULT_CLOSE_HTTP2_STREAM_DELAY_MILLIS = 10_000;
 
     public static boolean isHttp1WebSocketUpgradeRequest(RequestHeaders headers) {
         requireNonNull(headers, "headers");

core/src/main/java/com/linecorp/armeria/server/Http2RequestDecoder.java

@@ -278,6 +278,10 @@ public void onStreamClosed(Http2Stream stream) {
             // Ignored if the stream has already been closed.
             req.close(ClosedStreamException.get());
         }
+        final ServerHttp2ObjectEncoder encoder = this.encoder;
+        if (encoder != null) {
+            encoder.notifyStreamClosed(stream.id());
+        }
     }
 
     @Override

core/src/main/java/com/linecorp/armeria/server/Http2StreamLifecycleHandler.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2025 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.server;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.util.SafeCloseable;
+import com.linecorp.armeria.internal.common.AbstractHttp2ConnectionHandler;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.Http2ConnectionEncoder;
+import io.netty.handler.codec.http2.Http2Error;
+import io.netty.handler.codec.http2.Http2Stream;
+import io.netty.util.collection.IntObjectHashMap;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.ScheduledFuture;
+
+/**
+ * Synchronizes the lifecycle of an HTTP2 stream with its corresponding user-facing
+ * HTTP constructs. All methods in this class are expected to be invoked from
+ * the {@link Channel}'s {@link EventExecutor}.
+ */
+final class Http2StreamLifecycleHandler implements SafeCloseable {
+
+    private final Http2ConnectionEncoder encoder;
+    private final ChannelHandlerContext ctx;
+
+    private final Map<Integer, ScheduledFuture<?>> streamResetFutures = new IntObjectHashMap<>();
+
+    Http2StreamLifecycleHandler(ChannelHandlerContext ctx,
+                                AbstractHttp2ConnectionHandler handler) {
+        encoder = handler.encoder();
+        this.ctx = ctx;
+    }
+
+    /**
+     * Invoked when a {@link Http2Stream}'s corresponding {@link HttpRequest} and {@link HttpResponse}
+     * are closed.
+     */
+    void maybeResetStream(int streamId, Http2Error http2Error, long delayMillis) {
+        if (!canResetStream(streamId)) {
+            return;
+        }
+        if (delayMillis == 0) {
+            maybeResetStream0(streamId, http2Error);
+        } else if (delayMillis > 0) {
+            final ScheduledFuture<?> scheduled = ctx.executor().schedule(() -> {
+                maybeResetStream0(streamId, http2Error);
+            }, delayMillis, TimeUnit.MILLISECONDS);
+            streamResetFutures.put(streamId, scheduled);
+        }
+    }
+
+    private void maybeResetStream0(int streamId, Http2Error http2Error) {
+        if (!canResetStream(streamId)) {
+            return;
+        }
+        encoder.writeRstStream(ctx, streamId, http2Error.code(), ctx.voidPromise());
+        ctx.flush();
+    }
+
+    private boolean canResetStream(int streamId) {
+        if (!ctx.channel().isActive()) {
+            return false;
+        }
+        final Http2Stream stream = encoder.connection().stream(streamId);
+        if (stream == null) {
+            return false;
+        }
+        return stream.state().remoteSideOpen();
+    }
+
+    /**
+     * Invoked every time a stream is closed, which allows clean up of pre-scheduled
+     * stream close futures.
+     */
+    void notifyStreamClosed(int streamId) {
+        final ScheduledFuture<?> future = streamResetFutures.remove(streamId);
+        if (future != null) {
+            future.cancel(true);
+        }
+    }
+
+    @Override
+    public void close() {
+        if (!ctx.executor().inEventLoop()) {
+            ctx.executor().execute(this::close);
+            return;
+        }
+        for (ScheduledFuture<?> future : streamResetFutures.values()) {
+            future.cancel(true);
+        }
+        streamResetFutures.clear();
+    }
+}

core/src/main/java/com/linecorp/armeria/server/HttpServerHandler.java

@@ -774,13 +774,15 @@ private final class RequestAndResponseCompleteHandler {
         private final ChannelHandlerContext ctx;
         private final DecodedHttpRequest req;
         private final boolean isTransientService;
+        private final long closeHttp2StreamDelayMillis;
 
         RequestAndResponseCompleteHandler(EventLoop eventLoop, ChannelHandlerContext ctx,
                                           ServiceRequestContext reqCtx, DecodedHttpRequest req,
                                           boolean isTransientService) {
             this.ctx = ctx;
             this.req = req;
             this.isTransientService = isTransientService;
+            closeHttp2StreamDelayMillis = reqCtx.config().service().options().closeHttp2StreamDelayMillis();
 
             assert responseEncoder != null;
 
@@ -863,7 +865,7 @@ private void handleRequestOrResponseComplete() {
 
                 if (!isNeedsDisconnection() && responseEncoder instanceof ServerHttp2ObjectEncoder) {
                     ((ServerHttp2ObjectEncoder) responseEncoder)
-                            .maybeResetStream(req.streamId(), Http2Error.CANCEL);
+                            .maybeResetStream(req.streamId(), Http2Error.CANCEL, closeHttp2StreamDelayMillis);
                 }
 
                 final boolean needsDisconnection = ctx.channel().isActive() &&

core/src/main/java/com/linecorp/armeria/server/ServerHttp2ObjectEncoder.java

@@ -39,11 +39,14 @@
 
 final class ServerHttp2ObjectEncoder extends Http2ObjectEncoder implements ServerHttpObjectEncoder {
 
+    final Http2StreamLifecycleHandler streamLifecycleHandler;
+
     ServerHttp2ObjectEncoder(ChannelHandlerContext connectionHandlerCtx,
                              AbstractHttp2ConnectionHandler connectionHandler) {
         super(connectionHandlerCtx, connectionHandler);
         assert keepAliveHandler() instanceof Http2ServerKeepAliveHandler ||
                keepAliveHandler() instanceof NoopKeepAliveHandler;
+        streamLifecycleHandler = new  Http2StreamLifecycleHandler(connectionHandlerCtx, connectionHandler);
     }
 
     @Override
@@ -138,14 +141,17 @@ public ChannelFuture writeErrorResponse(int id, int streamId,
         return future;
     }
 
-    public void maybeResetStream(int streamId, Http2Error http2Error) {
-        final Http2Stream stream = encoder().connection().stream(streamId);
-        if (stream == null) {
-            return;
-        }
-        if (stream.state().remoteSideOpen()) {
-            encoder().writeRstStream(ctx(), streamId, http2Error.code(), ctx().voidPromise());
-            ctx().flush();
-        }
+    void maybeResetStream(int streamId, Http2Error http2Error, long delayMillis) {
+        streamLifecycleHandler.maybeResetStream(streamId, http2Error, delayMillis);
+    }
+
+    void notifyStreamClosed(int streamId) {
+        streamLifecycleHandler.notifyStreamClosed(streamId);
+    }
+
+    @Override
+    public void close(Throwable unused) {
+        super.close(unused);
+        streamLifecycleHandler.close();
     }
 }

core/src/main/java/com/linecorp/armeria/server/ServiceOptions.java

@@ -49,11 +49,14 @@ public static ServiceOptionsBuilder builder() {
     private final long requestTimeoutMillis;
     private final long maxRequestLength;
     private final long requestAutoAbortDelayMillis;
+    private final long closeHttp2StreamDelayMillis;
 
-    ServiceOptions(long requestTimeoutMillis, long maxRequestLength, long requestAutoAbortDelayMillis) {
+    ServiceOptions(long requestTimeoutMillis, long maxRequestLength, long requestAutoAbortDelayMillis,
+                   long closeHttp2StreamDelayMillis) {
         this.requestTimeoutMillis = requestTimeoutMillis;
         this.maxRequestLength = maxRequestLength;
         this.requestAutoAbortDelayMillis = requestAutoAbortDelayMillis;
+        this.closeHttp2StreamDelayMillis = closeHttp2StreamDelayMillis;
     }
 
     /**
@@ -78,6 +81,17 @@ public long requestAutoAbortDelayMillis() {
         return requestAutoAbortDelayMillis;
     }
 
+    /**
+     * Sets the amount of time to wait after an {@link HttpRequest} and {@link HttpResponse}
+     * is complete before closing the underlying HTTP2 stream. This value will default to {@code 0},
+     * which closes the stream immediately. If negative, the delay is disabled.
+     * This may be useful for protocols which have a separate lifecycle from the underlying
+     * HTTP2 stream such as WebSockets.
+     */
+    public long closeHttp2StreamDelayMillis() {
+        return closeHttp2StreamDelayMillis;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -91,12 +105,14 @@ public boolean equals(Object o) {
 
         return requestTimeoutMillis == that.requestTimeoutMillis &&
                maxRequestLength == that.maxRequestLength &&
-               requestAutoAbortDelayMillis == that.requestAutoAbortDelayMillis;
+               requestAutoAbortDelayMillis == that.requestAutoAbortDelayMillis &&
+               closeHttp2StreamDelayMillis == that.closeHttp2StreamDelayMillis;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis);
+        return Objects.hash(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis,
+                            closeHttp2StreamDelayMillis);
     }
 
     @Override
@@ -105,6 +121,7 @@ public String toString() {
                           .add("requestTimeoutMillis", requestTimeoutMillis)
                           .add("maxRequestLength", maxRequestLength)
                           .add("requestAutoAbortDelayMillis", requestAutoAbortDelayMillis)
+                          .add("closeHttp2StreamDelayMillis", closeHttp2StreamDelayMillis)
                           .toString();
     }
 }

core/src/main/java/com/linecorp/armeria/server/ServiceOptionsBuilder.java

@@ -30,11 +30,12 @@ public final class ServiceOptionsBuilder {
     private long requestTimeoutMillis = -1;
     private long maxRequestLength = -1;
     private long requestAutoAbortDelayMillis = -1;
+    private long closeHttp2StreamDelayMillis;
 
     ServiceOptionsBuilder() {}
 
     /**
-     * Returns the server-side timeout of a request in milliseconds.
+     * Sets the server-side timeout of a request in milliseconds.
      */
     public ServiceOptionsBuilder requestTimeoutMillis(long requestTimeoutMillis) {
         checkArgument(requestTimeoutMillis >= 0, "requestTimeoutMillis: %s (expected: >= 0)",
@@ -44,7 +45,7 @@ public ServiceOptionsBuilder requestTimeoutMillis(long requestTimeoutMillis) {
     }
 
     /**
-     * Returns the server-side maximum length of a request.
+     * Sets the server-side maximum length of a request.
      */
     public ServiceOptionsBuilder maxRequestLength(long maxRequestLength) {
         checkArgument(maxRequestLength >= 0, "maxRequestLength: %s (expected: >= 0)", maxRequestLength);
@@ -63,10 +64,23 @@ public ServiceOptionsBuilder requestAutoAbortDelayMillis(long requestAutoAbortDe
         return this;
     }
 
+    /**
+     * Sets the amount of time to wait after an {@link HttpRequest} and {@link HttpResponse}
+     * is complete before closing the underlying HTTP2 stream. This value will default to {@code 0},
+     * which closes the stream immediately. If negative, the delay is disabled.
+     * This may be useful for protocols which have a separate lifecycle from the underlying
+     * HTTP2 stream such as WebSockets.
+     */
+    public ServiceOptionsBuilder closeHttp2StreamDelayMillis(long closeHttp2StreamDelayMillis) {
+        this.closeHttp2StreamDelayMillis = closeHttp2StreamDelayMillis;
+        return this;
+    }
+
     /**
      * Returns a newly created {@link ServiceOptions} based on the properties of this builder.
      */
     public ServiceOptions build() {
-        return new ServiceOptions(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis);
+        return new ServiceOptions(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis,
+                                  closeHttp2StreamDelayMillis);
     }
 }

core/src/main/java/com/linecorp/armeria/server/websocket/WebSocketServiceBuilder.java

@@ -51,6 +51,8 @@
  *       {@value WebSocketUtil#DEFAULT_MAX_REQUEST_RESPONSE_LENGTH}.</li>
  *   <li>{@link ServiceConfig#requestAutoAbortDelayMillis()} is
  *       {@value WebSocketUtil#DEFAULT_REQUEST_AUTO_ABORT_DELAY_MILLIS}.</li>
+ *   <li>{@link ServiceOptions#closeHttp2StreamDelayMillis()} is
+ *       {@value WebSocketUtil#DEFAULT_CLOSE_HTTP2_STREAM_DELAY_MILLIS}.</li>
  * </ul>
  */
 @UnstableApi
@@ -67,6 +69,7 @@ public final class WebSocketServiceBuilder {
             .requestTimeoutMillis(WebSocketUtil.DEFAULT_REQUEST_RESPONSE_TIMEOUT_MILLIS)
             .maxRequestLength(WebSocketUtil.DEFAULT_MAX_REQUEST_RESPONSE_LENGTH)
             .requestAutoAbortDelayMillis(WebSocketUtil.DEFAULT_REQUEST_AUTO_ABORT_DELAY_MILLIS)
+            .closeHttp2StreamDelayMillis(WebSocketUtil.DEFAULT_CLOSE_HTTP2_STREAM_DELAY_MILLIS)
             .build();
 
     private final WebSocketServiceHandler handler;