Respond with 408 status when the server didn't receive the request fu…

…lly.

Motivation:
Currently, the server always responds with a 503 status if a `RequestTimeoutException` is raised.
However, according to the RFC 9110, the correct response in this case should be a 408 status code.
https://httpwg.org/specs/rfc9110.html#status.408
```
The 408 (Request Timeout) status code indicates that the server did not receive a complete request message within the time that it was prepared to wait.
```

Modification:
- Introduced `DecodedHttpRequest.isNormallyClosed()` to check if the request was received fully.
- Updated the server to send a 408 response when a request times out and the service didn't receive the request fully.

Result:
- The server now returns a 408 status if a service didn't receive the request fully and the request times out.
- Issue line#5579 has been closed.

core/src/main/java/com/linecorp/armeria/common/stream/DeferredStreamMessage.java

@@ -482,10 +482,8 @@ public CompletableFuture<List<T>> collect(EventExecutor executor, SubscriptionOp
  }
 
  private static SubscriptionImpl noopSubscription() {
- final DefaultStreamMessage<?> streamMessage = new DefaultStreamMessage<>();
- streamMessage.close();
- return new SubscriptionImpl(streamMessage, NoopSubscriber.get(), ImmediateEventExecutor.INSTANCE,
- EMPTY_OPTIONS);
+ return new SubscriptionImpl(NoopCancellableStreamMessage.INSTANCE, NoopSubscriber.get(),
+ ImmediateEventExecutor.INSTANCE, EMPTY_OPTIONS);
  }
 
  private final class ForwardingSubscriber implements Subscriber<T> {

core/src/main/java/com/linecorp/armeria/common/stream/NoopCancellableStreamMessage.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2024 LINE Corporation
+ *
+ * LINE 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.common.stream;
+
+final class NoopCancellableStreamMessage extends CancellableStreamMessage<Object> {
+
+ static final NoopCancellableStreamMessage INSTANCE = new NoopCancellableStreamMessage();
+
+ @Override
+ SubscriptionImpl subscribe(SubscriptionImpl subscription) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ void request(long n) {}
+
+ @Override
+ void cancel() {}
+
+ @Override
+ public boolean isOpen() {
+ return false;
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public long demand() {
+ return 0;
+ }
+
+ @Override
+ public void abort() {}
+
+ @Override
+ public void abort(Throwable cause) {}
+
+ private NoopCancellableStreamMessage() {}
+}

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

@@ -56,6 +56,8 @@ final class AggregatingDecodedHttpRequest extends AggregatingStreamMessage<HttpO
  @Nullable
  private Throwable abortResponseCause;
 
+ private boolean isNormallyClosed;
+
  private final CompletableFuture<Void> aggregationFuture = new CompletableFuture<>();
 
  AggregatingDecodedHttpRequest(EventLoop eventLoop, int id, int streamId, RequestHeaders headers,
@@ -194,6 +196,7 @@ public void abortResponse(Throwable cause, boolean cancel) {
 
  @Override
  public void close() {
+ isNormallyClosed = true;
  super.close();
  aggregationFuture.complete(null);
  }
@@ -204,6 +207,11 @@ public void close(Throwable cause) {
  aggregationFuture.complete(null);
  }
 
+ @Override
+ public boolean isNormallyClosed() {
+ return isNormallyClosed;
+ }
+
  @Override
  public void abort() {
  super.abort();

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

@@ -88,6 +88,8 @@ static DecodedHttpRequest of(boolean endOfStream, EventLoop eventLoop, int id, i
 
  void close(Throwable cause);
 
+ boolean isNormallyClosed();
+
  /**
  * Sets the specified {@link HttpResponse} which responds to this request. This is always called
  * by the {@link HttpServerHandler} after the handler gets the {@link HttpResponse} from an

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

@@ -25,13 +25,16 @@
 import com.linecorp.armeria.common.AggregatedHttpResponse;
 import com.linecorp.armeria.common.ContentTooLargeException;
 import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpMethod;
 import com.linecorp.armeria.common.HttpResponse;
 import com.linecorp.armeria.common.HttpStatus;
 import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.common.Request;
 import com.linecorp.armeria.common.RequestHeaders;
 import com.linecorp.armeria.common.ResponseHeaders;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.util.Exceptions;
+import com.linecorp.armeria.internal.common.RequestContextExtension;
 import com.linecorp.armeria.internal.common.util.TemporaryThreadLocals;
 import com.linecorp.armeria.server.annotation.AnnotatedService;
 
@@ -86,8 +89,24 @@ public HttpResponse onServiceException(ServiceRequestContext ctx, Throwable caus
  }
 
  if (cause instanceof RequestTimeoutException) {
- return internalRenderStatus(ctx, ctx.request().headers(),
- HttpStatus.SERVICE_UNAVAILABLE, cause);
+ final HttpStatus status;
+ if (ctx.method() == HttpMethod.GET) {
+ // Consider GET requests as having received the request fully.
+ status = HttpStatus.SERVICE_UNAVAILABLE;
+ } else {
+ final RequestContextExtension ctxExtension = ctx.as(RequestContextExtension.class);
+ assert ctxExtension != null;
+ final Request request = ctxExtension.originalRequest();
+
+ if (request instanceof DecodedHttpRequest &&
+ ((DecodedHttpRequest) request).isNormallyClosed()) {
+ status = HttpStatus.SERVICE_UNAVAILABLE;
+ } else {
+ // The server didn't receive the request fully yet.
+ status = HttpStatus.REQUEST_TIMEOUT;
+ }
+ }
+ return internalRenderStatus(ctx, ctx.request().headers(), status, cause);
  }
 
  return internalRenderStatus(ctx, ctx.request().headers(),

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

@@ -192,6 +192,11 @@ public void close() {}
  @Override
  public void close(Throwable cause) {}
 
+ @Override
+ public boolean isNormallyClosed() {
+ return true;
+ }
+
  @Override
  public void setResponse(HttpResponse response) {
  // TODO(ikhoon): Dedup

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

@@ -54,6 +54,8 @@ final class StreamingDecodedHttpRequest extends DefaultHttpRequest implements De
  @Nullable
  private Throwable abortResponseCause;
 
+ private boolean isNormallyClosed;
+
  StreamingDecodedHttpRequest(EventLoop eventLoop, int id, int streamId, RequestHeaders headers,
  boolean keepAlive, InboundTrafficController inboundTrafficController,
  long maxRequestLength, RoutingContext routingCtx, ExchangeType exchangeType,
@@ -165,6 +167,17 @@ protected void onRemoval(HttpObject obj) {
  }
  }
 
+ @Override
+ public void close() {
+ isNormallyClosed = true;
+ super.close();
+ }
+
+ @Override
+ public boolean isNormallyClosed() {
+ return isNormallyClosed;
+ }
+
  @Override
  public void setResponse(HttpResponse response) {
  if (abortResponseCause != null) {

core/src/test/java/com/linecorp/armeria/server/AggregatingRequestTimeoutTest.java

@@ -91,6 +91,6 @@ void asynchronouslyHandleFailedRequest(SessionProtocol protocol) {
  // The service waits for the full request body and then responds.
  assertThat(WebClient.of(protocol, server.endpoint(protocol))
  .execute(request).aggregate().join().status())
- .isSameAs(HttpStatus.SERVICE_UNAVAILABLE);
+ .isSameAs(HttpStatus.REQUEST_TIMEOUT);
  }
 }

core/src/test/java/com/linecorp/armeria/server/RequestTimeoutTest.java

@@ -103,9 +103,18 @@ public ExchangeType exchangeType(RoutingContext routingContext) {
  request.write(HttpData.ofUtf8("foo"));
  }
  request.close();
- assertThat(WebClient.of(protocol, server.endpoint(protocol))
- .execute(request).aggregate().join().status())
- .isSameAs(HttpStatus.SERVICE_UNAVAILABLE);
+
+ if (!exchangeType.isRequestStreaming() || (!clientSendsOneBody && !useServiceEventLoop)) {
+ // The server received the request fully in these cases.
+ assertThat(WebClient.of(protocol, server.endpoint(protocol))
+ .execute(request).aggregate().join().status())
+ .isSameAs(HttpStatus.SERVICE_UNAVAILABLE);
+ } else {
+ // The server did or did not receive the request fully depending on the thread scheduling timing.
+ assertThat(WebClient.of(protocol, server.endpoint(protocol))
+ .execute(request).aggregate().join().status())
+ .isIn(HttpStatus.REQUEST_TIMEOUT, HttpStatus.SERVICE_UNAVAILABLE);
+ }
  }
 
  private static class TimeoutByTimerArgumentsProvider implements ArgumentsProvider {
@@ -147,12 +156,16 @@ public ExchangeType exchangeType(RoutingContext routingContext) {
  });
  });
  final HttpRequestWriter request = HttpRequest.streaming(RequestHeaders.of(HttpMethod.POST, "/"));
+ final HttpStatus expectedStatus;
  if (clientCloseRequest) {
  request.close();
+ expectedStatus = HttpStatus.SERVICE_UNAVAILABLE;
+ } else {
+ expectedStatus = HttpStatus.REQUEST_TIMEOUT;
  }
  assertThat(WebClient.builder(protocol, server.endpoint(protocol))
  .build()
  .execute(request).aggregate().join().status())
- .isSameAs(HttpStatus.SERVICE_UNAVAILABLE);
+ .isSameAs(expectedStatus);
  }
 }