From 7f5c6ddb25bb16a9f284dd17f2fed10a5e0517fd Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Tue, 25 Aug 2026 12:56:18 +0300 Subject: [PATCH 1/9] Optimized KoraRequestProcessingHttpHandler --- .../UndertowCamundaRestHttpHandler.java | 4 +- .../UndertowHttpServerFactoryModule.java | 4 +- .../KoraRequestProcessingHttpHandler.java | 48 +++++++++------ ...readPerConnectionDispatchHttpHandler.java} | 4 +- .../request/UndertowRequestHttpBody.java | 5 +- .../undertow/UndertowHttpServerTest.java | 4 +- .../request/UndertowRequestHttpBodyTest.java | 61 +++++++++++++++++++ 7 files changed, 102 insertions(+), 28 deletions(-) rename http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/{KoraVirtualThreadDispatchHttpHandler.java => KoraVirtualThreadPerConnectionDispatchHttpHandler.java} (89%) create mode 100644 http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java diff --git a/experimental/camunda-rest-undertow/src/main/java/io/koraframework/camunda/rest/undertow/UndertowCamundaRestHttpHandler.java b/experimental/camunda-rest-undertow/src/main/java/io/koraframework/camunda/rest/undertow/UndertowCamundaRestHttpHandler.java index 6d762104b5..31d2370986 100644 --- a/experimental/camunda-rest-undertow/src/main/java/io/koraframework/camunda/rest/undertow/UndertowCamundaRestHttpHandler.java +++ b/experimental/camunda-rest-undertow/src/main/java/io/koraframework/camunda/rest/undertow/UndertowCamundaRestHttpHandler.java @@ -11,7 +11,7 @@ import io.koraframework.common.util.TimeUtils; import io.koraframework.http.server.undertow.handler.KoraRequestProcessingHttpHandler; import io.koraframework.http.server.undertow.UndertowContext; -import io.koraframework.http.server.undertow.handler.KoraVirtualThreadDispatchHttpHandler; +import io.koraframework.http.server.undertow.handler.KoraVirtualThreadPerConnectionDispatchHttpHandler; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; import io.undertow.server.HttpHandler; import io.undertow.server.handlers.PathHandler; @@ -147,7 +147,7 @@ public void init() throws Exception { }); root.addPrefixPath("/", new OpenApiHttpHandler(camundaRestConfig)); - this.realhttpHandler = new KoraVirtualThreadDispatchHttpHandler("camunda-rest", root); + this.realhttpHandler = new KoraVirtualThreadPerConnectionDispatchHttpHandler("camunda-rest", root); logger.info("Camunda Rest Handler (Undertow) configured in {}", TimeUtils.tookForLogging(started)); } diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java index 28b32e316d..497cb16df3 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java @@ -10,7 +10,7 @@ import io.koraframework.http.server.common.router.HttpServerRouter; import io.koraframework.http.server.common.telemetry.HttpServerTelemetryFactory; import io.koraframework.http.server.undertow.handler.KoraRequestProcessingHttpHandler; -import io.koraframework.http.server.undertow.handler.KoraVirtualThreadDispatchHttpHandler; +import io.koraframework.http.server.undertow.handler.KoraVirtualThreadPerConnectionDispatchHttpHandler; import io.undertow.Undertow; import io.undertow.server.HttpHandler; import org.jspecify.annotations.Nullable; @@ -41,7 +41,7 @@ public HttpHandler handler(@Tag(Tag.Factory.class) HttpServerConfig httpServerCo HttpServerTelemetryFactory telemetryFactory) { var telemetry = telemetryFactory.get(this.name, httpServerConfig.port(), httpServerConfig.telemetry()); var handler = (HttpHandler) new KoraRequestProcessingHttpHandler(httpServerConfig, httpServerRouter, telemetry); - handler = new KoraVirtualThreadDispatchHttpHandler(this.name, handler); + handler = new KoraVirtualThreadPerConnectionDispatchHttpHandler(this.name, handler); return handler; } } diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java index ad47134ac2..1591497c9d 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java @@ -10,6 +10,7 @@ import io.koraframework.http.server.common.router.HttpServerRouter; import io.koraframework.http.server.common.telemetry.HttpServerObservation; import io.koraframework.http.server.common.telemetry.HttpServerTelemetry; +import io.koraframework.http.server.common.telemetry.impl.NoopHttpServerTelemetry; import io.koraframework.http.server.undertow.UndertowContext; import io.koraframework.http.server.undertow.request.UndertowUnroutedHttpRequest; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; @@ -43,6 +44,7 @@ public final class KoraRequestProcessingHttpHandler implements HttpHandler { private final HttpServerConfig httpServerConfig; private final HttpServerTelemetry telemetry; private final HttpServerRouter httpServerRouter; + private final boolean telemetryEnabled; public KoraRequestProcessingHttpHandler(HttpServerConfig httpServerConfig, HttpServerRouter httpServerRouter, @@ -50,11 +52,14 @@ public KoraRequestProcessingHttpHandler(HttpServerConfig httpServerConfig, this.httpServerConfig = httpServerConfig; this.telemetry = telemetry; this.httpServerRouter = httpServerRouter; + this.telemetryEnabled = !(telemetry instanceof NoopHttpServerTelemetry); } @Override public void handleRequest(HttpServerExchange exchange) { - var rootCtx = W3CTraceContextPropagator.getInstance().extract(Context.root(), exchange.getRequestHeaders(), HttpServerExchangeMapGetter.INSTANCE); + var rootCtx = this.telemetryEnabled + ? W3CTraceContextPropagator.getInstance().extract(Context.root(), exchange.getRequestHeaders(), HttpServerExchangeMapGetter.INSTANCE) + : Context.root(); ScopedValue .where(UndertowContext.VALUE, new UndertowContext(exchange)) .where(io.koraframework.logging.common.MDC.VALUE, new io.koraframework.logging.common.MDC()) @@ -62,20 +67,21 @@ public void handleRequest(HttpServerExchange exchange) { .run(() -> { MDC.clear(); try { - exchange.startBlocking(); var request = new UndertowUnroutedHttpRequest(exchange); var invocation = this.httpServerRouter.route(request); var observation = this.telemetry.observe(invocation.routedRequest()); var ctx = rootCtx.with(observation.span()); - W3CTraceContextPropagator.getInstance().inject( - ctx, - exchange.getResponseHeaders(), - HttpServerExchangeMapGetter.INSTANCE - ); - exchange.addExchangeCompleteListener((e, nextListener) -> { - observation.end(); - nextListener.proceed(); - }); + if (telemetryEnabled) { + W3CTraceContextPropagator.getInstance().inject( + ctx, + exchange.getResponseHeaders(), + HttpServerExchangeMapGetter.INSTANCE + ); + exchange.addExchangeCompleteListener((e, nextListener) -> { + observation.end(); + nextListener.proceed(); + }); + } ScopedValue .where(OpentelemetryContext.VALUE, ctx) .where(Observation.VALUE, observation) @@ -122,23 +128,27 @@ private void sendResponse(HttpServerObservation observation, HttpServerExchange this.setHeaders(exchange.getResponseHeaders(), headers, null); return; } + var contentType = body.contentType(); + this.setHeaders(exchange.getResponseHeaders(), headers, contentType); + if (contentType != null) { + exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, contentType); + } try (body) { - var contentType = body.contentType(); - this.setHeaders(exchange.getResponseHeaders(), headers, contentType); - if (contentType != null) { - exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, contentType); - } - var contentLength = body.contentLength(); - if (contentLength >= 0) { - exchange.setResponseContentLength(contentLength); + if (!exchange.isBlocking()) { + exchange.startBlocking(); } try (var os = exchange.getOutputStream()) { try { var full = body.getFullContentIfAvailable(); if (full != null) { + exchange.setResponseContentLength(full.remaining()); this.writeBuffer(exchange, os, full); return; } + var contentLength = body.contentLength(); + if (contentLength >= 0) { + exchange.setResponseContentLength(contentLength); + } body.write(os); } catch (Throwable t) { if (!exchange.isResponseStarted()) { diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraVirtualThreadDispatchHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraVirtualThreadPerConnectionDispatchHttpHandler.java similarity index 89% rename from http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraVirtualThreadDispatchHttpHandler.java rename to http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraVirtualThreadPerConnectionDispatchHttpHandler.java index 08c88f1107..4a82a22879 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraVirtualThreadDispatchHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraVirtualThreadPerConnectionDispatchHttpHandler.java @@ -7,14 +7,14 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -public final class KoraVirtualThreadDispatchHttpHandler implements HttpHandler { +public final class KoraVirtualThreadPerConnectionDispatchHttpHandler implements HttpHandler { private final AttachmentKey executorServiceAttachmentKey = AttachmentKey.create(ExecutorService.class); private final String name; private final HttpHandler delegate; - public KoraVirtualThreadDispatchHttpHandler(String name, HttpHandler delegate) { + public KoraVirtualThreadPerConnectionDispatchHttpHandler(String name, HttpHandler delegate) { this.name = name; this.delegate = delegate; } diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBody.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBody.java index a8d281dbeb..fc112a5c6a 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBody.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBody.java @@ -32,12 +32,15 @@ public String contentType() { @Override @NonNull public InputStream asInputStream() { + if (!this.exchange.isBlocking()) { + this.exchange.startBlocking(); + } return this.exchange.getInputStream(); } @Override public void close() throws IOException { - this.exchange.getInputStream().close(); + this.asInputStream().close(); } @Override diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java index 90765bb64b..a75483815c 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java @@ -7,7 +7,7 @@ import io.koraframework.http.server.common.router.HttpServerRouter; import io.koraframework.http.server.common.telemetry.HttpServerTelemetry; import io.koraframework.http.server.undertow.handler.KoraRequestProcessingHttpHandler; -import io.koraframework.http.server.undertow.handler.KoraVirtualThreadDispatchHttpHandler; +import io.koraframework.http.server.undertow.handler.KoraVirtualThreadPerConnectionDispatchHttpHandler; class UndertowHttpServerTest extends HttpServerTestKit { @@ -15,7 +15,7 @@ class UndertowHttpServerTest extends HttpServerTestKit { protected HttpServer httpServer(ValueOf config, HttpServerRouter httpServerRouter, HttpServerTelemetry telemetry) { return new UndertowHttpServer( "test", - valueOf(new KoraVirtualThreadDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(config.get(), httpServerRouter, telemetry))), + valueOf(new KoraVirtualThreadPerConnectionDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(config.get(), httpServerRouter, telemetry))), null, config, null diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java new file mode 100644 index 0000000000..ace4e7d9d7 --- /dev/null +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java @@ -0,0 +1,61 @@ +package io.koraframework.http.server.undertow.request; + +import io.undertow.server.HttpServerExchange; +import org.junit.jupiter.api.Test; + +import java.io.InputStream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class UndertowRequestHttpBodyTest { + + @Test + void startsBlockingModeWhenInputStreamIsRequested() { + var exchange = mock(HttpServerExchange.class); + var inputStream = mock(InputStream.class); + when(exchange.getInputStream()).thenReturn(inputStream); + + var body = new UndertowRequestHttpBody(exchange); + + assertThat(body.asInputStream()).isSameAs(inputStream); + verify(exchange).startBlocking(); + } + + @Test + void doesNotStartBlockingModeUntilBodyIsUsed() { + var exchange = mock(HttpServerExchange.class); + + new UndertowRequestHttpBody(exchange); + + verify(exchange, never()).startBlocking(); + verify(exchange, never()).getInputStream(); + } + + @Test + void closesOpenedInputStream() throws Exception { + var exchange = mock(HttpServerExchange.class); + var inputStream = mock(InputStream.class); + when(exchange.isBlocking()).thenReturn(true); + when(exchange.getInputStream()).thenReturn(inputStream); + + new UndertowRequestHttpBody(exchange).close(); + + verify(inputStream).close(); + } + + @Test + void startsBlockingModeWhenUnopenedBodyIsClosed() throws Exception { + var exchange = mock(HttpServerExchange.class); + var inputStream = mock(InputStream.class); + when(exchange.getInputStream()).thenReturn(inputStream); + + new UndertowRequestHttpBody(exchange).close(); + + verify(exchange).startBlocking(); + verify(inputStream).close(); + } +} From 4914f47d46b934eacdd9e85ecbd11c7a27c7d147 Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Tue, 25 Aug 2026 14:08:18 +0300 Subject: [PATCH 2/9] Optimized KoraRequestProcessingHttpHandler --- .../http/common/body/JsonHttpBodyOutput.java | 8 +- .../KoraRequestProcessingHttpHandler.java | 283 ++++++++++++------ .../undertow/UndertowHttpServerTest.java | 56 ++++ 3 files changed, 259 insertions(+), 88 deletions(-) diff --git a/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java b/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java index f9f2633574..5d9565db54 100644 --- a/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java +++ b/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java @@ -1,13 +1,14 @@ package io.koraframework.http.common.body; -import org.jspecify.annotations.Nullable; import io.koraframework.json.common.JsonModule; import io.koraframework.json.common.JsonWriter; +import org.jspecify.annotations.Nullable; import tools.jackson.core.JsonEncoding; import tools.jackson.core.ObjectWriteContext; import java.io.IOException; import java.io.OutputStream; +import java.nio.ByteBuffer; public final class JsonHttpBodyOutput implements HttpBodyOutput { private final JsonWriter writer; @@ -29,6 +30,11 @@ public String contentType() { return "application/json"; } + @Override + public ByteBuffer getFullContentIfAvailable() { + return ByteBuffer.wrap(this.writer.toByteArray(value)); + } + @Override public void write(OutputStream os) throws IOException { try (var gen = JsonModule.JSON_FACTORY.createGenerator(ObjectWriteContext.empty(), os, JsonEncoding.UTF8)) { diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java index 1591497c9d..ff33770e1e 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java @@ -11,6 +11,7 @@ import io.koraframework.http.server.common.telemetry.HttpServerObservation; import io.koraframework.http.server.common.telemetry.HttpServerTelemetry; import io.koraframework.http.server.common.telemetry.impl.NoopHttpServerTelemetry; +import io.koraframework.http.server.common.telemetry.impl.NoopHttpServerObservation; import io.koraframework.http.server.undertow.UndertowContext; import io.koraframework.http.server.undertow.request.UndertowUnroutedHttpRequest; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; @@ -18,18 +19,23 @@ import io.opentelemetry.context.propagation.TextMapGetter; import io.opentelemetry.context.propagation.TextMapSetter; import io.undertow.UndertowMessages; +import io.undertow.io.AsyncSenderImpl; import io.undertow.io.BufferWritableOutputStream; +import io.undertow.io.IoCallback; +import io.undertow.io.Sender; import io.undertow.server.HttpHandler; import io.undertow.server.HttpServerExchange; import io.undertow.util.HeaderMap; import io.undertow.util.HeaderValues; import io.undertow.util.Headers; import io.undertow.util.HttpString; +import io.undertow.util.Methods; import org.jspecify.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; import java.nio.ByteBuffer; @@ -57,137 +63,118 @@ public KoraRequestProcessingHttpHandler(HttpServerConfig httpServerConfig, @Override public void handleRequest(HttpServerExchange exchange) { + ProcessedResponse processedResponse; + try { + processedResponse = this.processRequest(exchange); + } catch (Throwable e) { + logger.warn("HTTP request processing failed", e); + processedResponse = errorResponse(NoopHttpServerObservation.INSTANCE, Context.root(), null, e); + } + var response = processedResponse; + exchange.dispatch(exchange.getIoThread(), () -> this.sendResponse(exchange, response)); + } + + private ProcessedResponse processRequest(HttpServerExchange exchange) { var rootCtx = this.telemetryEnabled ? W3CTraceContextPropagator.getInstance().extract(Context.root(), exchange.getRequestHeaders(), HttpServerExchangeMapGetter.INSTANCE) : Context.root(); - ScopedValue + return ScopedValue .where(UndertowContext.VALUE, new UndertowContext(exchange)) .where(io.koraframework.logging.common.MDC.VALUE, new io.koraframework.logging.common.MDC()) .where(OpentelemetryContext.VALUE, rootCtx) - .run(() -> { + .call(() -> { MDC.clear(); try { var request = new UndertowUnroutedHttpRequest(exchange); var invocation = this.httpServerRouter.route(request); var observation = this.telemetry.observe(invocation.routedRequest()); var ctx = rootCtx.with(observation.span()); - if (telemetryEnabled) { - W3CTraceContextPropagator.getInstance().inject( - ctx, - exchange.getResponseHeaders(), - HttpServerExchangeMapGetter.INSTANCE - ); - exchange.addExchangeCompleteListener((e, nextListener) -> { - observation.end(); - nextListener.proceed(); - }); - } - ScopedValue + return ScopedValue .where(OpentelemetryContext.VALUE, ctx) .where(Observation.VALUE, observation) - .run(() -> { + .call(() -> { HttpServerResponse response; + if (httpServerConfig.headerServerNameEnabled()) { + exchange.getResponseHeaders().put(Headers.SERVER, "Kora"); + } try { var httpServerRequest = observation.observeRequest(invocation.routedRequest()); response = invocation.proceed(httpServerRequest); } catch (Throwable e) { observation.observeError(e); if (e instanceof HttpServerResponse rs) { - this.sendResponse(observation, exchange, rs); + response = rs; } else { - this.sendResponse(observation, exchange, HttpServerResponse.of(500, HttpBody.plaintext(Objects.requireNonNullElse(e.getMessage(), "Unknown error")))); + return errorResponse(observation, ctx, null, e); } - return; } - this.sendResponse(observation, exchange, response); + return this.materializeResponse(observation, ctx, response, exchange.getRequestMethod().equals(Methods.HEAD)); }); - } catch (Throwable exception) { - exchange.setStatusCode(500); - try { - exchange.getResponseSender().send(StandardCharsets.UTF_8.encode(Objects.requireNonNullElse(exception.getMessage(), "Unknown error"))); - exchange.getConnection().close(); - } catch (Exception e) { - exception.addSuppressed(e); - } - logger.warn("Error dropped", exception); } finally { - exchange.endExchange(); + MDC.clear(); } }); } - private void sendResponse(HttpServerObservation observation, HttpServerExchange exchange, HttpServerResponse httpResponse) { - httpResponse = observation.observeResponse(httpResponse); - var headers = httpResponse.headers(); - exchange.setStatusCode(httpResponse.code()); - if (httpServerConfig.headerServerNameEnabled()) { - exchange.getResponseHeaders().put(Headers.SERVER, "Kora"); - } - var body = httpResponse.body(); + private ProcessedResponse materializeResponse(HttpServerExchange exchange, HttpServerObservation observation, Context context, HttpServerResponse response, boolean headRequest) { + response = observation.observeResponse(response); + var body = response.body(); if (body == null) { - this.setHeaders(exchange.getResponseHeaders(), headers, null); - return; + return new ProcessedResponse(observation, context, response.code(), response.headers(), null, null, -1, null, null); } - var contentType = body.contentType(); - this.setHeaders(exchange.getResponseHeaders(), headers, contentType); - if (contentType != null) { - exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, contentType); + var declaredLength = body.contentLength(); + if (headRequest) { + return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), ByteBuffer.allocate(0), declaredLength, body, null); } - try (body) { - if (!exchange.isBlocking()) { - exchange.startBlocking(); - } - try (var os = exchange.getOutputStream()) { - try { - var full = body.getFullContentIfAvailable(); - if (full != null) { - exchange.setResponseContentLength(full.remaining()); - this.writeBuffer(exchange, os, full); - return; - } - var contentLength = body.contentLength(); - if (contentLength >= 0) { - exchange.setResponseContentLength(contentLength); - } - body.write(os); - } catch (Throwable t) { - if (!exchange.isResponseStarted()) { - observation.observeResponse(HttpServerResponse.of(500, HttpBody.plaintext(Objects.requireNonNullElse(t.getMessage(), "Unknown error")))); - exchange.setStatusCode(500); - exchange.getResponseHeaders().remove(Headers.CONTENT_LENGTH); - exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "text/plain"); - exchange.getResponseSender().send(Objects.requireNonNullElse(t.getMessage(), "Unknown error")); - } else { - observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); - } - throw t; - } + ByteArrayOutputStream output = null; + try { + var contentType = body.contentType(); + var content = body.getFullContentIfAvailable(); + if (content == null) { + output = new ByteArrayOutputStream(declaredLength > 0 && declaredLength <= Integer.MAX_VALUE + ? (int) declaredLength + : 256); + body.write(output); + content = ByteBuffer.wrap(output.toByteArray()); + } else { + content = content.slice(); } + var contentLength = declaredLength >= 0 ? declaredLength : content.remaining(); + return new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null); } catch (Throwable e) { observation.observeError(e); + if (output != null && output.size() > 0) { + var content = ByteBuffer.wrap(output.toByteArray()); + var contentLength = declaredLength >= 0 ? declaredLength : content.remaining() + 1L; + return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), content, contentLength, body, e); + } + return errorResponse(observation, context, body, e); } } - private void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { - for (var header : headers) { - var key = header.getKey(); - if (key.equals("server")) { - continue; - } - if (key.equals("content-type") && contentType != null) { - continue; - } - if (key.equals("content-length")) { - continue; + private void writeBuffer(HttpServerExchange exchange, OutputStream outputStream, ByteBuffer buffer) throws IOException { + if (outputStream instanceof BufferWritableOutputStream bufferWritableOutputStream) { + //fast path, if the stream can take a buffer directly just write to it + bufferWritableOutputStream.write(buffer); + return; + } + if (buffer.hasArray()) { + outputStream.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining()); + return; + } + try (var pooled = exchange.getConnection().getByteBufferPool().getArrayBackedPool().allocate()) { + if (pooled == null) { + throw UndertowMessages.MESSAGES.failedToAllocateResource(); } - if (key.equals("transfer-encoding")) { - continue; + while (buffer.hasRemaining()) { + var toRead = Math.min(buffer.remaining(), pooled.getBuffer().remaining()); + buffer.get(pooled.getBuffer().array(), pooled.getBuffer().arrayOffset(), toRead); + outputStream.write(pooled.getBuffer().array(), pooled.getBuffer().arrayOffset(), toRead); } - responseHeaders.addAll(HttpString.tryFromString(key), header.getValue()); } } - private void writeBuffer(HttpServerExchange exchange, OutputStream outputStream, ByteBuffer buffer) throws IOException { + private void writeBufferNew(HttpServerExchange exchange, OutputStream outputStream, ByteBuffer buffer) throws IOException { if (outputStream instanceof BufferWritableOutputStream bufferWritableOutputStream) { //fast path, if the stream can take a buffer directly just write to it bufferWritableOutputStream.write(buffer); @@ -209,6 +196,128 @@ private void writeBuffer(HttpServerExchange exchange, OutputStream outputStream, } } + private static ProcessedResponse errorResponse(HttpServerObservation observation, Context context, @Nullable HttpBody bodyToClose, Throwable error) { + var message = Objects.requireNonNullElse(error.getMessage(), "Unknown error"); + var content = message.getBytes(StandardCharsets.UTF_8); + var response = observation.observeResponse(HttpServerResponse.of(500, HttpBody.plaintext(message))); + return new ProcessedResponse( + observation, + context, + response.code(), + response.headers(), + "text/plain;charset=utf-8", + ByteBuffer.wrap(content), + content.length, + bodyToClose, + null + ); + } + + private void sendResponse(HttpServerExchange exchange, ProcessedResponse response) { + var observation = response.observation(); + try { + if (this.telemetryEnabled) { + W3CTraceContextPropagator.getInstance().inject( + response.context(), + exchange.getResponseHeaders(), + HttpServerExchangeMapGetter.INSTANCE + ); + exchange.addExchangeCompleteListener((e, nextListener) -> { + observation.end(); + nextListener.proceed(); + }); + } + exchange.setStatusCode(response.code()); + exchange.getResponseHeaders().put(Headers.SERVER, "Kora"); + this.setHeaders(exchange.getResponseHeaders(), response.headers(), response.contentType()); + if (response.contentType() != null) { + exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, response.contentType()); + } + var content = response.content(); + if (content == null) { + closeBody(observation, response.body()); + exchange.endExchange(); + return; + } + exchange.setResponseContentLength(response.contentLength()); + new AsyncSenderImpl(exchange).send(content, new IoCallback() { + @Override + public void onComplete(HttpServerExchange exchange, Sender sender) { + closeBody(observation, response.body()); + if (response.sendFailure() != null) { + observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); + try { + exchange.getConnection().close(); + } catch (IOException closeException) { + response.sendFailure().addSuppressed(closeException); + } + return; + } + IoCallback.END_EXCHANGE.onComplete(exchange, sender); + } + + @Override + public void onException(HttpServerExchange exchange, Sender sender, IOException exception) { + observation.observeError(exception); + observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); + closeBody(observation, response.body()); + IoCallback.END_EXCHANGE.onException(exchange, sender, exception); + } + }); + } catch (Throwable e) { + observation.observeError(e); + observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); + closeBody(observation, response.body()); + exchange.endExchange(); + try { + exchange.getConnection().close(); + } catch (IOException closeException) { + e.addSuppressed(closeException); + } + logger.warn("HTTP response send failed", e); + } + } + + private static void closeBody(HttpServerObservation observation, @Nullable HttpBody body) { + if (body == null) { + return; + } + try { + body.close(); + } catch (IOException e) { + observation.observeError(e); + } + } + + private record ProcessedResponse(HttpServerObservation observation, + Context context, + int code, + HttpHeaders headers, + @Nullable String contentType, + @Nullable ByteBuffer content, + long contentLength, + @Nullable HttpBody body, + @Nullable Throwable sendFailure) {} + + private void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { + for (var header : headers) { + var key = header.getKey(); + if (key.equals("server")) { + continue; + } + if (key.equals("content-type") && contentType != null) { + continue; + } + if (key.equals("content-length")) { + continue; + } + if (key.equals("transfer-encoding")) { + continue; + } + responseHeaders.addAll(HttpString.tryFromString(key), header.getValue()); + } + } + public static class HttpServerExchangeMapGetter implements TextMapGetter, TextMapSetter { public static final HttpServerExchangeMapGetter INSTANCE = new HttpServerExchangeMapGetter(); diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java index a75483815c..55b64a3e58 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java @@ -4,13 +4,69 @@ import io.koraframework.http.server.common.HttpServer; import io.koraframework.http.server.common.HttpServerConfig; import io.koraframework.http.server.common.HttpServerTestKit; +import io.koraframework.http.common.body.HttpBodyOutput; +import io.koraframework.http.server.common.request.HttpServerRequestHandlerImpl; +import io.koraframework.http.server.common.response.HttpServerResponse; import io.koraframework.http.server.common.router.HttpServerRouter; import io.koraframework.http.server.common.telemetry.HttpServerTelemetry; import io.koraframework.http.server.undertow.handler.KoraRequestProcessingHttpHandler; import io.koraframework.http.server.undertow.handler.KoraVirtualThreadPerConnectionDispatchHttpHandler; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; class UndertowHttpServerTest extends HttpServerTestKit { + @Test + void responseBodyIsMaterializedOnVirtualThreadAndClosedOnIoThread() throws Exception { + var writeThread = new AtomicReference(); + var closeThread = new AtomicReference(); + var closed = new CountDownLatch(1); + var body = new HttpBodyOutput() { + @Override + public long contentLength() { + return -1; + } + + @Override + public String contentType() { + return "application/json"; + } + + @Override + public void write(OutputStream os) throws IOException { + writeThread.set(Thread.currentThread()); + os.write("{\"message\":\"ok\"}".getBytes(StandardCharsets.UTF_8)); + } + + @Override + public void close() { + closeThread.set(Thread.currentThread()); + closed.countDown(); + } + }; + startServer(HttpServerRequestHandlerImpl.get("/thread-ownership", _ -> HttpServerResponse.of(200, body))); + + try (var response = client.newCall(request("/thread-ownership").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().string()).isEqualTo("{\"message\":\"ok\"}"); + } + + assertThat(closed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(writeThread.get()).isNotNull(); + assertThat(writeThread.get().isVirtual()).isTrue(); + assertThat(closeThread.get()).isNotNull(); + assertThat(closeThread.get().isVirtual()).isFalse(); + assertThat(closeThread.get().getName()).contains("XNIO").contains("I/O"); + } + @Override protected HttpServer httpServer(ValueOf config, HttpServerRouter httpServerRouter, HttpServerTelemetry telemetry) { return new UndertowHttpServer( From d2df03b8803ce95367242f0a36ac7e63e709af12 Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Wed, 26 Aug 2026 02:55:21 +0300 Subject: [PATCH 3/9] perf(undertow): optimize response body writes Buffer small bodies on request VTs and stream large bodies through a bounded XNIO pipe. Keep send, close, and exchange completion on I/O threads to avoid buffer-pool contention and unbounded memory use. --- .../mapper/HttpClientRequestMapperModule.java | 9 +- .../http/common/body/JsonHttpBodyOutput.java | 6 - .../KoraRequestProcessingHttpHandler.java | 394 ++++++++++++++---- .../undertow/UndertowHttpServerTest.java | 48 +++ 4 files changed, 362 insertions(+), 95 deletions(-) diff --git a/http/http-client-common/src/main/java/io/koraframework/http/client/common/request/mapper/HttpClientRequestMapperModule.java b/http/http-client-common/src/main/java/io/koraframework/http/client/common/request/mapper/HttpClientRequestMapperModule.java index aed08c454c..ab6aac734f 100644 --- a/http/http-client-common/src/main/java/io/koraframework/http/client/common/request/mapper/HttpClientRequestMapperModule.java +++ b/http/http-client-common/src/main/java/io/koraframework/http/client/common/request/mapper/HttpClientRequestMapperModule.java @@ -1,10 +1,11 @@ package io.koraframework.http.client.common.request.mapper; import io.koraframework.common.annotation.DefaultComponent; -import io.koraframework.common.annotation.Tag; import io.koraframework.http.client.common.request.HttpClientRequestMapper; import io.koraframework.http.common.body.HttpBody; import io.koraframework.http.common.body.HttpBodyOutput; +import io.koraframework.http.common.form.FormMultipart; +import io.koraframework.http.common.form.FormUrlEncoded; import io.koraframework.json.common.JsonWriter; import io.koraframework.json.common.annotation.Json; @@ -33,18 +34,18 @@ default HttpClientRequestMapper httpClientRequestStringMapper() { } @DefaultComponent - default FormUrlEncodedClientRequestMapper httpClientRequestFormUrlEncodedMapper() { + default HttpClientRequestMapper httpClientRequestFormUrlEncodedMapper() { return new FormUrlEncodedClientRequestMapper(); } @DefaultComponent - default FormMultipartClientRequestMapper httpClientRequestFormMultipartMapper() { + default HttpClientRequestMapper httpClientRequestFormMultipartMapper() { return new FormMultipartClientRequestMapper(); } @Json @DefaultComponent - default JsonHttpClientRequestMapper httpClientRequestJsonMapper(JsonWriter writer) { + default HttpClientRequestMapper httpClientRequestJsonMapper(JsonWriter writer) { return new JsonHttpClientRequestMapper<>(writer); } } diff --git a/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java b/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java index 5d9565db54..d248fb9b3d 100644 --- a/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java +++ b/http/http-common/src/main/java/io/koraframework/http/common/body/JsonHttpBodyOutput.java @@ -8,7 +8,6 @@ import java.io.IOException; import java.io.OutputStream; -import java.nio.ByteBuffer; public final class JsonHttpBodyOutput implements HttpBodyOutput { private final JsonWriter writer; @@ -30,11 +29,6 @@ public String contentType() { return "application/json"; } - @Override - public ByteBuffer getFullContentIfAvailable() { - return ByteBuffer.wrap(this.writer.toByteArray(value)); - } - @Override public void write(OutputStream os) throws IOException { try (var gen = JsonModule.JSON_FACTORY.createGenerator(ObjectWriteContext.empty(), os, JsonEncoding.UTF8)) { diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java index ff33770e1e..a15cb68261 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java @@ -4,6 +4,7 @@ import io.koraframework.common.telemetry.OpentelemetryContext; import io.koraframework.http.common.HttpResultCode; import io.koraframework.http.common.body.HttpBody; +import io.koraframework.http.common.body.HttpBodyOutput; import io.koraframework.http.common.header.HttpHeaders; import io.koraframework.http.server.common.HttpServerConfig; import io.koraframework.http.server.common.response.HttpServerResponse; @@ -18,9 +19,7 @@ import io.opentelemetry.context.Context; import io.opentelemetry.context.propagation.TextMapGetter; import io.opentelemetry.context.propagation.TextMapSetter; -import io.undertow.UndertowMessages; import io.undertow.io.AsyncSenderImpl; -import io.undertow.io.BufferWritableOutputStream; import io.undertow.io.IoCallback; import io.undertow.io.Sender; import io.undertow.server.HttpHandler; @@ -30,22 +29,27 @@ import io.undertow.util.Headers; import io.undertow.util.HttpString; import io.undertow.util.Methods; +import io.undertow.util.SameThreadExecutor; import org.jspecify.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; -import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.Iterator; import java.util.Objects; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; public final class KoraRequestProcessingHttpHandler implements HttpHandler { private static final Logger logger = LoggerFactory.getLogger(KoraRequestProcessingHttpHandler.class); + public static final ByteBuffer BODY_EMPTY = ByteBuffer.allocate(0); private final HttpServerConfig httpServerConfig; private final HttpServerTelemetry telemetry; @@ -71,7 +75,9 @@ public void handleRequest(HttpServerExchange exchange) { processedResponse = errorResponse(NoopHttpServerObservation.INSTANCE, Context.root(), null, e); } var response = processedResponse; - exchange.dispatch(exchange.getIoThread(), () -> this.sendResponse(exchange, response)); + if (!response.streamingStarted()) { + exchange.dispatch(exchange.getIoThread(), () -> this.sendResponse(exchange, response)); + } } private ProcessedResponse processRequest(HttpServerExchange exchange) { @@ -108,7 +114,7 @@ private ProcessedResponse processRequest(HttpServerExchange exchange) { return errorResponse(observation, ctx, null, e); } } - return this.materializeResponse(observation, ctx, response, exchange.getRequestMethod().equals(Methods.HEAD)); + return this.prepareResponse(exchange, observation, ctx, response, exchange.getRequestMethod().equals(Methods.HEAD)); }); } finally { MDC.clear(); @@ -116,87 +122,55 @@ private ProcessedResponse processRequest(HttpServerExchange exchange) { }); } - private ProcessedResponse materializeResponse(HttpServerExchange exchange, HttpServerObservation observation, Context context, HttpServerResponse response, boolean headRequest) { + private ProcessedResponse prepareResponse(HttpServerExchange exchange, HttpServerObservation observation, Context context, HttpServerResponse response, boolean headRequest) { response = observation.observeResponse(response); var body = response.body(); if (body == null) { - return new ProcessedResponse(observation, context, response.code(), response.headers(), null, null, -1, null, null); + return new ProcessedResponse(observation, context, response.code(), response.headers(), null, null, -1, null, null, false); } + var declaredLength = body.contentLength(); if (headRequest) { - return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), ByteBuffer.allocate(0), declaredLength, body, null); + return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), BODY_EMPTY, declaredLength, body, null, false); } - ByteArrayOutputStream output = null; + + AdaptiveBodyOutputStream output = null; try { var contentType = body.contentType(); var content = body.getFullContentIfAvailable(); - if (content == null) { - output = new ByteArrayOutputStream(declaredLength > 0 && declaredLength <= Integer.MAX_VALUE - ? (int) declaredLength - : 256); - body.write(output); - content = ByteBuffer.wrap(output.toByteArray()); - } else { - content = content.slice(); + if (content != null) { + var contentLength = declaredLength >= 0 ? declaredLength : content.remaining(); + return new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); } + + var streamingResponse = new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, null, declaredLength, body, null, true); + output = new AdaptiveBodyOutputStream(this, exchange, streamingResponse); + body.write(output); + if (output.isStreaming()) { + output.completeStreaming(); + return streamingResponse; + } + + content = output.fullContent(); var contentLength = declaredLength >= 0 ? declaredLength : content.remaining(); - return new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null); + return new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); } catch (Throwable e) { observation.observeError(e); + if (output != null && output.isStreaming()) { + output.failStreaming(e); + return output.streamingResponse(); + } + if (output != null && output.size() > 0) { - var content = ByteBuffer.wrap(output.toByteArray()); + var content = output.fullContent(); var contentLength = declaredLength >= 0 ? declaredLength : content.remaining() + 1L; - return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), content, contentLength, body, e); + return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), content, contentLength, body, e, false); } return errorResponse(observation, context, body, e); } } - private void writeBuffer(HttpServerExchange exchange, OutputStream outputStream, ByteBuffer buffer) throws IOException { - if (outputStream instanceof BufferWritableOutputStream bufferWritableOutputStream) { - //fast path, if the stream can take a buffer directly just write to it - bufferWritableOutputStream.write(buffer); - return; - } - if (buffer.hasArray()) { - outputStream.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining()); - return; - } - try (var pooled = exchange.getConnection().getByteBufferPool().getArrayBackedPool().allocate()) { - if (pooled == null) { - throw UndertowMessages.MESSAGES.failedToAllocateResource(); - } - while (buffer.hasRemaining()) { - var toRead = Math.min(buffer.remaining(), pooled.getBuffer().remaining()); - buffer.get(pooled.getBuffer().array(), pooled.getBuffer().arrayOffset(), toRead); - outputStream.write(pooled.getBuffer().array(), pooled.getBuffer().arrayOffset(), toRead); - } - } - } - - private void writeBufferNew(HttpServerExchange exchange, OutputStream outputStream, ByteBuffer buffer) throws IOException { - if (outputStream instanceof BufferWritableOutputStream bufferWritableOutputStream) { - //fast path, if the stream can take a buffer directly just write to it - bufferWritableOutputStream.write(buffer); - return; - } - if (buffer.hasArray()) { - outputStream.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining()); - return; - } - try (var pooled = exchange.getConnection().getByteBufferPool().getArrayBackedPool().allocate()) { - if (pooled == null) { - throw UndertowMessages.MESSAGES.failedToAllocateResource(); - } - while (buffer.hasRemaining()) { - var toRead = Math.min(buffer.remaining(), pooled.getBuffer().remaining()); - buffer.get(pooled.getBuffer().array(), pooled.getBuffer().arrayOffset(), toRead); - outputStream.write(pooled.getBuffer().array(), pooled.getBuffer().arrayOffset(), toRead); - } - } - } - - private static ProcessedResponse errorResponse(HttpServerObservation observation, Context context, @Nullable HttpBody bodyToClose, Throwable error) { + private static ProcessedResponse errorResponse(HttpServerObservation observation, Context context, @Nullable HttpBodyOutput bodyToClose, Throwable error) { var message = Objects.requireNonNullElse(error.getMessage(), "Unknown error"); var content = message.getBytes(StandardCharsets.UTF_8); var response = observation.observeResponse(HttpServerResponse.of(500, HttpBody.plaintext(message))); @@ -209,30 +183,15 @@ private static ProcessedResponse errorResponse(HttpServerObservation observation ByteBuffer.wrap(content), content.length, bodyToClose, - null + null, + false ); } private void sendResponse(HttpServerExchange exchange, ProcessedResponse response) { var observation = response.observation(); try { - if (this.telemetryEnabled) { - W3CTraceContextPropagator.getInstance().inject( - response.context(), - exchange.getResponseHeaders(), - HttpServerExchangeMapGetter.INSTANCE - ); - exchange.addExchangeCompleteListener((e, nextListener) -> { - observation.end(); - nextListener.proceed(); - }); - } - exchange.setStatusCode(response.code()); - exchange.getResponseHeaders().put(Headers.SERVER, "Kora"); - this.setHeaders(exchange.getResponseHeaders(), response.headers(), response.contentType()); - if (response.contentType() != null) { - exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, response.contentType()); - } + this.prepareExchange(exchange, response); var content = response.content(); if (content == null) { closeBody(observation, response.body()); @@ -278,6 +237,40 @@ public void onException(HttpServerExchange exchange, Sender sender, IOException } } + private void sendStreamingResponse(HttpServerExchange exchange, ProcessedResponse response, AsyncBodyPipe pipe) { + try { + this.prepareExchange(exchange, response); + if (response.contentLength() >= 0) { + exchange.setResponseContentLength(response.contentLength()); + } + pipe.attach(new AsyncSenderImpl(exchange)); + } catch (Throwable e) { + pipe.abort(e); + exchange.endExchange(); + logger.warn("HTTP streaming response setup failed", e); + } + } + + private void prepareExchange(HttpServerExchange exchange, ProcessedResponse response) { + if (this.telemetryEnabled) { + W3CTraceContextPropagator.getInstance().inject( + response.context(), + exchange.getResponseHeaders(), + HttpServerExchangeMapGetter.INSTANCE + ); + exchange.addExchangeCompleteListener((e, nextListener) -> { + response.observation().end(); + nextListener.proceed(); + }); + } + exchange.setStatusCode(response.code()); + exchange.getResponseHeaders().put(Headers.SERVER, "Kora"); + this.setHeaders(exchange.getResponseHeaders(), response.headers(), response.contentType()); + if (response.contentType() != null) { + exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, response.contentType()); + } + } + private static void closeBody(HttpServerObservation observation, @Nullable HttpBody body) { if (body == null) { return; @@ -296,8 +289,239 @@ private record ProcessedResponse(HttpServerObservation observation, @Nullable String contentType, @Nullable ByteBuffer content, long contentLength, - @Nullable HttpBody body, - @Nullable Throwable sendFailure) {} + @Nullable HttpBodyOutput body, + @Nullable Throwable sendFailure, + boolean streamingStarted) {} + + private static final class AdaptiveBodyOutputStream extends OutputStream { + + private static final int SMALL_BODY_THRESHOLD = 64 * 1024; + private static final int STREAM_CHUNK_SIZE = 16 * 1024; + + private final KoraRequestProcessingHttpHandler handler; + private final HttpServerExchange exchange; + private final ProcessedResponse streamingResponse; + private byte[] buffer = new byte[256]; + private int size; + private @Nullable AsyncBodyPipe pipe; + + private AdaptiveBodyOutputStream(KoraRequestProcessingHttpHandler handler, + HttpServerExchange exchange, + ProcessedResponse streamingResponse) { + this.handler = handler; + this.exchange = exchange; + this.streamingResponse = streamingResponse; + } + + @Override + public void write(int value) throws IOException { + this.write(new byte[]{(byte) value}, 0, 1); + } + + @Override + public void write(byte[] bytes, int offset, int length) throws IOException { + Objects.checkFromIndexSize(offset, length, bytes.length); + if (length == 0) { + return; + } + if (this.pipe == null && this.size + length <= SMALL_BODY_THRESHOLD) { + this.ensureCapacity(this.size + length); + System.arraycopy(bytes, offset, this.buffer, this.size, length); + this.size += length; + return; + } + if (this.pipe == null) { + this.startStreaming(); + } + while (length > 0) { + var chunkLength = Math.min(length, STREAM_CHUNK_SIZE); + var chunk = Arrays.copyOfRange(bytes, offset, offset + chunkLength); + this.pipe.offer(ByteBuffer.wrap(chunk)); + offset += chunkLength; + length -= chunkLength; + } + } + + private void ensureCapacity(int required) { + if (required <= this.buffer.length) { + return; + } + var capacity = Math.min(SMALL_BODY_THRESHOLD, Math.max(required, this.buffer.length << 1)); + this.buffer = Arrays.copyOf(this.buffer, capacity); + } + + private void startStreaming() throws IOException { + this.pipe = new AsyncBodyPipe(this.handler, this.exchange, this.streamingResponse); + this.pipe.start(); + if (this.size > 0) { + this.pipe.offer(ByteBuffer.wrap(this.buffer, 0, this.size)); + this.buffer = new byte[0]; + this.size = 0; + } + } + + private boolean isStreaming() { + return this.pipe != null; + } + + private int size() { + return this.size; + } + + private ByteBuffer fullContent() { + return ByteBuffer.wrap(this.buffer, 0, this.size); + } + + private ProcessedResponse streamingResponse() { + return this.streamingResponse; + } + + private void completeStreaming() { + Objects.requireNonNull(this.pipe).complete(); + } + + private void failStreaming(Throwable failure) { + Objects.requireNonNull(this.pipe).fail(failure); + } + } + + private static final class AsyncBodyPipe { + + private static final int MAX_PENDING_CHUNKS = 4; + + private final KoraRequestProcessingHttpHandler handler; + private final HttpServerExchange exchange; + private final ProcessedResponse response; + private final ConcurrentLinkedQueue chunks = new ConcurrentLinkedQueue<>(); + private final Semaphore freeSlots = new Semaphore(MAX_PENDING_CHUNKS); + private final AtomicBoolean drainScheduled = new AtomicBoolean(); + private final AtomicBoolean terminal = new AtomicBoolean(); + private final IoCallback callback = new IoCallback() { + @Override + public void onComplete(HttpServerExchange exchange, Sender sender) { + freeSlots.release(); + sending = false; + drain(); + } + + @Override + public void onException(HttpServerExchange exchange, Sender sender, IOException exception) { + freeSlots.release(); + finishException(exception); + } + }; + + private volatile @Nullable Sender sender; + private volatile boolean producerComplete; + private volatile @Nullable Throwable producerFailure; + private boolean sending; + + private AsyncBodyPipe(KoraRequestProcessingHttpHandler handler, + HttpServerExchange exchange, + ProcessedResponse response) { + this.handler = handler; + this.exchange = exchange; + this.response = response; + } + + private void start() { + // Keep Connectors.executeRootHandler() from auto-ending the exchange when the VT producer returns. + this.exchange.dispatch(SameThreadExecutor.INSTANCE, () -> {}); + // dispatch(...) itself is deferred until the handler returns, so the consumer must be started directly. + this.exchange.getIoThread().execute(() -> this.handler.sendStreamingResponse(this.exchange, this.response, this)); + } + + private void attach(Sender sender) { + this.sender = sender; + this.scheduleDrain(); + } + + private void offer(ByteBuffer chunk) throws IOException { + try { + this.freeSlots.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for HTTP response backpressure", e); + } + if (this.terminal.get()) { + this.freeSlots.release(); + throw new IOException("HTTP response stream is already closed"); + } + this.chunks.add(chunk); + this.scheduleDrain(); + } + + private void complete() { + this.producerComplete = true; + this.scheduleDrain(); + } + + private void fail(Throwable failure) { + this.producerFailure = failure; + this.producerComplete = true; + this.scheduleDrain(); + } + + private void abort(Throwable failure) { + this.producerFailure = failure; + this.producerComplete = true; + this.finishException(failure); + } + + private void scheduleDrain() { + if (this.sender != null && this.drainScheduled.compareAndSet(false, true)) { + this.exchange.getIoThread().execute(this::drain); + } + } + + private void drain() { + this.drainScheduled.set(false); + if (this.terminal.get() || this.sending) { + return; + } + var chunk = this.chunks.poll(); + if (chunk != null) { + this.sending = true; + Objects.requireNonNull(this.sender).send(chunk, this.callback); + return; + } + if (this.producerComplete) { + var failure = this.producerFailure; + if (failure == null) { + this.finishComplete(); + } else { + this.finishException(failure); + } + } + } + + private void finishComplete() { + if (!this.terminal.compareAndSet(false, true)) { + return; + } + closeBody(this.response.observation(), this.response.body()); + IoCallback.END_EXCHANGE.onComplete(this.exchange, Objects.requireNonNull(this.sender)); + } + + private void finishException(Throwable failure) { + if (!this.terminal.compareAndSet(false, true)) { + return; + } + var abandoned = 0; + while (this.chunks.poll() != null) { + abandoned++; + } + this.freeSlots.release(abandoned + MAX_PENDING_CHUNKS); + this.response.observation().observeError(failure); + this.response.observation().observeResultCode(HttpResultCode.CONNECTION_ERROR); + closeBody(this.response.observation(), this.response.body()); + try { + this.exchange.getConnection().close(); + } catch (IOException closeException) { + failure.addSuppressed(closeException); + } + } + } private void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { for (var header : headers) { diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java index 55b64a3e58..3dd7f94d39 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java @@ -16,6 +16,7 @@ import java.io.IOException; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -67,6 +68,53 @@ public void close() { assertThat(closeThread.get().getName()).contains("XNIO").contains("I/O"); } + @Test + void largeResponseBodyIsProducedOnVirtualThreadAndStreamedByIoThread() throws Exception { + var expected = new byte[256 * 1024]; + Arrays.fill(expected, (byte) 'a'); + var writeThread = new AtomicReference(); + var closeThread = new AtomicReference(); + var closed = new CountDownLatch(1); + var body = new HttpBodyOutput() { + @Override + public long contentLength() { + return -1; + } + + @Override + public String contentType() { + return "application/octet-stream"; + } + + @Override + public void write(OutputStream os) throws IOException { + writeThread.set(Thread.currentThread()); + for (var offset = 0; offset < expected.length; offset += 4096) { + os.write(expected, offset, Math.min(4096, expected.length - offset)); + } + } + + @Override + public void close() { + closeThread.set(Thread.currentThread()); + closed.countDown(); + } + }; + startServer(HttpServerRequestHandlerImpl.get("/large-thread-ownership", _ -> HttpServerResponse.of(200, body))); + + try (var response = client.newCall(request("/large-thread-ownership").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().bytes()).isEqualTo(expected); + } + + assertThat(closed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(writeThread.get()).isNotNull(); + assertThat(writeThread.get().isVirtual()).isTrue(); + assertThat(closeThread.get()).isNotNull(); + assertThat(closeThread.get().isVirtual()).isFalse(); + assertThat(closeThread.get().getName()).contains("XNIO").contains("I/O"); + } + @Override protected HttpServer httpServer(ValueOf config, HttpServerRouter httpServerRouter, HttpServerTelemetry telemetry) { return new UndertowHttpServer( From 6a48a7ab5a25d78ccf7ab5d53c97a5308c02107b Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Wed, 26 Aug 2026 16:28:59 +0300 Subject: [PATCH 4/9] fix(undertow): process requests off-call, cut per-request allocation Streaming responses larger than the socket buffer aborted mid-body. AsyncBodyPipe started the I/O-thread consumer while the producer was still inside handleRequest, leaving the exchange IN_CALL | DISPATCHED. Undertow forbids resuming writes in that state, so AsyncSenderImpl threw UT000146 on the first partial write. Small bodies never hit it because their writes complete synchronously. handleRequest now only claims the dispatch slot and defers the request to the dispatch task, which Connectors.executeRootHandler runs after setInCall(false)/unDispatch(). Async writes are legal for the whole response, and AsyncBodyPipe no longer needs its keep-alive dispatch hack. Also fixes a double observeError: once the pipe reported its terminal failure, the producer's next offer() threw and prepareResponse counted that echo as a second error. - Cache lowercase header name -> interned HttpString. Kora lowercases every header name while Undertow's Headers cache is keyed by the canonical spelling, so tryFromString() missed on every header of every response and allocated an HttpString plus its backing array. Also covers traceparent/tracestate on the injection path. - Reuse exchange.getResponseSender() instead of allocating a second AsyncSenderImpl per response (3.7% of baseline allocation). - Let ProcessedResponse be its own Runnable, IoCallback and ExchangeCompletionListener, removing three allocations per request. - Size the adaptive body buffer from the declared content length, keeping 256 bytes when it is unknown. A 1 KiB default made AdaptiveBodyOutputStream the largest byte[] source in the process and inflated total allocation by 18% on /db. - Hoist the W3C propagator and response HeaderMap, switch on reserved header names, drop the redundant leading MDC.clear(). Wire the new tracing contextPropagation flag through the handler so W3C extract/inject can be disabled independently of telemetry. Fix a BufferUnderflowException in DefaultFullHttpBody.write() for non-heap-backed buffers with under 1024 bytes remaining. --- .../http/common/body/DefaultFullHttpBody.java | 2 +- .../telemetry/HttpServerTelemetryConfig.java | 4 + .../server/undertow/UndertowHttpServer.java | 1 + .../UndertowHttpServerFactoryModule.java | 6 +- .../KoraRequestProcessingHttpHandler.java | 359 +++++++++++++----- .../undertow/UndertowHttpServerTest.java | 3 +- 6 files changed, 284 insertions(+), 91 deletions(-) diff --git a/http/http-common/src/main/java/io/koraframework/http/common/body/DefaultFullHttpBody.java b/http/http-common/src/main/java/io/koraframework/http/common/body/DefaultFullHttpBody.java index 62c322d4ad..c8a147ec01 100644 --- a/http/http-common/src/main/java/io/koraframework/http/common/body/DefaultFullHttpBody.java +++ b/http/http-common/src/main/java/io/koraframework/http/common/body/DefaultFullHttpBody.java @@ -43,7 +43,7 @@ public void write(OutputStream os) throws IOException { data = data.slice(); while (data.hasRemaining()) { var len = Math.min(data.remaining(), buf.length); - data.get(buf); + data.get(buf, 0, len); os.write(buf, 0, len); } } diff --git a/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java b/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java index 253b186d87..ac8ca372be 100644 --- a/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java +++ b/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java @@ -53,6 +53,10 @@ interface HttpServerMetricsConfig extends TelemetryConfig.MetricsConfig { } @ConfigMapper interface HttpServerTracingConfig extends TelemetryConfig.TracingConfig { + default boolean contextPropagation() { + return true; + } + default boolean tracePathFull() { return true; } diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java index deda0bee4f..1f288bf16b 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java @@ -40,6 +40,7 @@ public class UndertowHttpServer implements HttpServer, ReadinessProbe { private volatile Undertow undertow; public UndertowHttpServer(String name, + ValueOf undertowConfig, ValueOf httpHandler, XnioWorker xnioWorker, ValueOf config, diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java index 497cb16df3..e81c085ae0 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java @@ -28,15 +28,17 @@ public UndertowHttpServerFactoryModule(String name, String configPath) { @Root @Tag(Tag.Factory.class) public UndertowHttpServer server(XnioWorker worker, + ValueOf undertowConfig, @Tag(Tag.Factory.class) ValueOf httpHandler, @Tag(Tag.Factory.class) ValueOf config, @Tag(Tag.Factory.class) @Nullable Configurer configurer) { - return new UndertowHttpServer(this.name, httpHandler, worker, config, configurer); + return new UndertowHttpServer(this.name, undertowConfig, httpHandler, worker, config, configurer); } @DefaultComponent @Tag(Tag.Factory.class) - public HttpHandler handler(@Tag(Tag.Factory.class) HttpServerConfig httpServerConfig, + public HttpHandler handler(ValueOf undertowConfig, + @Tag(Tag.Factory.class) HttpServerConfig httpServerConfig, @Tag(Tag.Factory.class) HttpServerRouter httpServerRouter, HttpServerTelemetryFactory telemetryFactory) { var telemetry = telemetryFactory.get(this.name, httpServerConfig.port(), httpServerConfig.telemetry()); diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java index a15cb68261..d47c30df88 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java @@ -1,5 +1,6 @@ package io.koraframework.http.server.undertow.handler; +import io.koraframework.application.graph.ValueOf; import io.koraframework.common.telemetry.Observation; import io.koraframework.common.telemetry.OpentelemetryContext; import io.koraframework.http.common.HttpResultCode; @@ -13,15 +14,16 @@ import io.koraframework.http.server.common.telemetry.HttpServerTelemetry; import io.koraframework.http.server.common.telemetry.impl.NoopHttpServerTelemetry; import io.koraframework.http.server.common.telemetry.impl.NoopHttpServerObservation; +import io.koraframework.http.server.undertow.UndertowConfig; import io.koraframework.http.server.undertow.UndertowContext; import io.koraframework.http.server.undertow.request.UndertowUnroutedHttpRequest; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; import io.opentelemetry.context.Context; import io.opentelemetry.context.propagation.TextMapGetter; import io.opentelemetry.context.propagation.TextMapSetter; -import io.undertow.io.AsyncSenderImpl; import io.undertow.io.IoCallback; import io.undertow.io.Sender; +import io.undertow.server.ExchangeCompletionListener; import io.undertow.server.HttpHandler; import io.undertow.server.HttpServerExchange; import io.undertow.util.HeaderMap; @@ -38,9 +40,10 @@ import java.io.IOException; import java.io.OutputStream; import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.HashMap; import java.util.Iterator; +import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Semaphore; @@ -49,47 +52,172 @@ public final class KoraRequestProcessingHttpHandler implements HttpHandler { private static final Logger logger = LoggerFactory.getLogger(KoraRequestProcessingHttpHandler.class); - public static final ByteBuffer BODY_EMPTY = ByteBuffer.allocate(0); + private static final ByteBuffer BODY_EMPTY = ByteBuffer.allocate(0); + + private static final W3CTraceContextPropagator PROPAGATOR = W3CTraceContextPropagator.getInstance(); + + private static final HttpString TRACE_PARENT = new HttpString("traceparent"); + private static final HttpString TRACE_STATE = new HttpString("tracestate"); + + /** + * Kora normalizes every header name to lower case, while Undertow's {@link Headers} cache is keyed by the + * canonical spelling ({@code Content-Type}). Looking a lower case name up via {@link HttpString#tryFromString} + * therefore always misses that cache and allocates a fresh {@link HttpString} plus its backing byte array for + * every header of every response. This table restores the cache hit: {@link HttpString} hashes and compares + * case-insensitively, so the canonical instances behave exactly like freshly built lower case ones. + */ + private static final Map HEADER_NAMES = lowercaseHeaderNames(); private final HttpServerConfig httpServerConfig; private final HttpServerTelemetry telemetry; private final HttpServerRouter httpServerRouter; private final boolean telemetryEnabled; + private final boolean contextPropagationEnabled; - public KoraRequestProcessingHttpHandler(HttpServerConfig httpServerConfig, - HttpServerRouter httpServerRouter, - HttpServerTelemetry telemetry) { - this.httpServerConfig = httpServerConfig; + public KoraRequestProcessingHttpHandler(ValueOf undertowConfig, + ValueOf config, + HttpServerTelemetry telemetry, + HttpServerRouter httpServerRouter) { this.telemetry = telemetry; this.httpServerRouter = httpServerRouter; this.telemetryEnabled = !(telemetry instanceof NoopHttpServerTelemetry); + this.contextPropagationEnabled = this.telemetryEnabled && config.get().telemetry().tracing().contextPropagation(); + this.httpServerConfig = config.get(); + } + + private static Map lowercaseHeaderNames() { + var names = new HashMap(256); + names.put("accept", Headers.ACCEPT); + names.put("accept-charset", Headers.ACCEPT_CHARSET); + names.put("accept-encoding", Headers.ACCEPT_ENCODING); + names.put("accept-language", Headers.ACCEPT_LANGUAGE); + names.put("accept-ranges", Headers.ACCEPT_RANGES); + names.put("age", Headers.AGE); + names.put("allow", Headers.ALLOW); + names.put("authentication-info", Headers.AUTHENTICATION_INFO); + names.put("authorization", Headers.AUTHORIZATION); + names.put("cache-control", Headers.CACHE_CONTROL); + names.put("connection", Headers.CONNECTION); + names.put("content-disposition", Headers.CONTENT_DISPOSITION); + names.put("content-encoding", Headers.CONTENT_ENCODING); + names.put("content-language", Headers.CONTENT_LANGUAGE); + names.put("content-length", Headers.CONTENT_LENGTH); + names.put("content-location", Headers.CONTENT_LOCATION); + names.put("content-md5", Headers.CONTENT_MD5); + names.put("content-range", Headers.CONTENT_RANGE); + names.put("content-security-policy", Headers.CONTENT_SECURITY_POLICY); + names.put("content-transfer-encoding", Headers.CONTENT_TRANSFER_ENCODING); + names.put("content-transfer-encoding", Headers.CONTENT_TRANSFER_ENCODING); + names.put("content-type", Headers.CONTENT_TYPE); + names.put("cookie", Headers.COOKIE); + names.put("cookie2", Headers.COOKIE2); + names.put("date", Headers.DATE); + names.put("etag", Headers.ETAG); + names.put("expect", Headers.EXPECT); + names.put("expires", Headers.EXPIRES); + names.put("forwarded", Headers.FORWARDED); + names.put("from", Headers.FROM); + names.put("host", Headers.HOST); + names.put("if-match", Headers.IF_MATCH); + names.put("if-modified-since", Headers.IF_MODIFIED_SINCE); + names.put("if-none-match", Headers.IF_NONE_MATCH); + names.put("if-range", Headers.IF_RANGE); + names.put("if-unmodified-since", Headers.IF_UNMODIFIED_SINCE); + names.put("last-modified", Headers.LAST_MODIFIED); + names.put("location", Headers.LOCATION); + names.put("max-forwards", Headers.MAX_FORWARDS); + names.put("origin", Headers.ORIGIN); + names.put("pragma", Headers.PRAGMA); + names.put("proxy-authenticate", Headers.PROXY_AUTHENTICATE); + names.put("proxy-authorization", Headers.PROXY_AUTHORIZATION); + names.put("range", Headers.RANGE); + names.put("referer", Headers.REFERER); + names.put("referrer-policy", Headers.REFERRER_POLICY); + names.put("refresh", Headers.REFRESH); + names.put("retry-after", Headers.RETRY_AFTER); + names.put("sec-websocket-accept", Headers.SEC_WEB_SOCKET_ACCEPT); + names.put("sec-websocket-extensions", Headers.SEC_WEB_SOCKET_EXTENSIONS); + names.put("sec-websocket-key", Headers.SEC_WEB_SOCKET_KEY); + names.put("sec-websocket-key1", Headers.SEC_WEB_SOCKET_KEY1); + names.put("sec-websocket-key2", Headers.SEC_WEB_SOCKET_KEY2); + names.put("sec-websocket-location", Headers.SEC_WEB_SOCKET_LOCATION); + names.put("sec-websocket-origin", Headers.SEC_WEB_SOCKET_ORIGIN); + names.put("sec-websocket-protocol", Headers.SEC_WEB_SOCKET_PROTOCOL); + names.put("sec-websocket-version", Headers.SEC_WEB_SOCKET_VERSION); + names.put("secure_protocol", Headers.SECURE_PROTOCOL); + names.put("server", Headers.SERVER); + names.put("servlet-engine", Headers.SERVLET_ENGINE); + names.put("set-cookie", Headers.SET_COOKIE); + names.put("set-cookie2", Headers.SET_COOKIE2); + names.put("ssl_cipher", Headers.SSL_CIPHER); + names.put("ssl_cipher_usekeysize", Headers.SSL_CIPHER_USEKEYSIZE); + names.put("ssl_client_cert", Headers.SSL_CLIENT_CERT); + names.put("ssl_session_id", Headers.SSL_SESSION_ID); + names.put("status", Headers.STATUS); + names.put("strict-transport-security", Headers.STRICT_TRANSPORT_SECURITY); + names.put("te", Headers.TE); + names.put("trailer", Headers.TRAILER); + names.put("transfer-encoding", Headers.TRANSFER_ENCODING); + names.put("upgrade", Headers.UPGRADE); + names.put("user-agent", Headers.USER_AGENT); + names.put("vary", Headers.VARY); + names.put("via", Headers.VIA); + names.put("warning", Headers.WARNING); + names.put("www-authenticate", Headers.WWW_AUTHENTICATE); + names.put("x-content-length", Headers.X_CONTENT_LENGTH); + names.put("x-content-type-options", Headers.X_CONTENT_TYPE_OPTIONS); + names.put("x-disable-push", Headers.X_DISABLE_PUSH); + names.put("x-forwarded-for", Headers.X_FORWARDED_FOR); + names.put("x-forwarded-host", Headers.X_FORWARDED_HOST); + names.put("x-forwarded-port", Headers.X_FORWARDED_PORT); + names.put("x-forwarded-proto", Headers.X_FORWARDED_PROTO); + names.put("x-forwarded-server", Headers.X_FORWARDED_SERVER); + names.put("x-frame-options", Headers.X_FRAME_OPTIONS); + names.put("x-xss-protection", Headers.X_XSS_PROTECTION); + names.put("traceparent", TRACE_PARENT); + names.put("tracestate", TRACE_STATE); + return Map.copyOf(names); + } + + /** + * @return interned {@link HttpString} for an already lower cased header name, or null if the name can not be + * encoded as a header name at all. + */ + @Nullable + private static HttpString headerName(String lowercaseName) { + var cached = HEADER_NAMES.get(lowercaseName); + return cached != null + ? cached + : HttpString.tryFromString(lowercaseName); } @Override public void handleRequest(HttpServerExchange exchange) { + exchange.dispatch(SameThreadExecutor.INSTANCE, () -> this.process(exchange)); + } + + private void process(HttpServerExchange exchange) { ProcessedResponse processedResponse; try { processedResponse = this.processRequest(exchange); } catch (Throwable e) { logger.warn("HTTP request processing failed", e); - processedResponse = errorResponse(NoopHttpServerObservation.INSTANCE, Context.root(), null, e); + processedResponse = this.errorResponse(exchange, NoopHttpServerObservation.INSTANCE, Context.root(), null, e); } - var response = processedResponse; - if (!response.streamingStarted()) { - exchange.dispatch(exchange.getIoThread(), () -> this.sendResponse(exchange, response)); + if (!processedResponse.streamingStarted()) { + exchange.getIoThread().execute(processedResponse); } } private ProcessedResponse processRequest(HttpServerExchange exchange) { - var rootCtx = this.telemetryEnabled - ? W3CTraceContextPropagator.getInstance().extract(Context.root(), exchange.getRequestHeaders(), HttpServerExchangeMapGetter.INSTANCE) + var rootCtx = this.contextPropagationEnabled + ? PROPAGATOR.extract(Context.root(), exchange.getRequestHeaders(), HttpServerExchangeMapGetter.INSTANCE) : Context.root(); return ScopedValue .where(UndertowContext.VALUE, new UndertowContext(exchange)) .where(io.koraframework.logging.common.MDC.VALUE, new io.koraframework.logging.common.MDC()) .where(OpentelemetryContext.VALUE, rootCtx) .call(() -> { - MDC.clear(); try { var request = new UndertowUnroutedHttpRequest(exchange); var invocation = this.httpServerRouter.route(request); @@ -111,7 +239,7 @@ private ProcessedResponse processRequest(HttpServerExchange exchange) { if (e instanceof HttpServerResponse rs) { response = rs; } else { - return errorResponse(observation, ctx, null, e); + return this.errorResponse(exchange, observation, ctx, null, e); } } return this.prepareResponse(exchange, observation, ctx, response, exchange.getRequestMethod().equals(Methods.HEAD)); @@ -126,12 +254,12 @@ private ProcessedResponse prepareResponse(HttpServerExchange exchange, HttpServe response = observation.observeResponse(response); var body = response.body(); if (body == null) { - return new ProcessedResponse(observation, context, response.code(), response.headers(), null, null, -1, null, null, false); + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), null, null, -1, null, null, false); } var declaredLength = body.contentLength(); if (headRequest) { - return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), BODY_EMPTY, declaredLength, body, null, false); + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), body.contentType(), BODY_EMPTY, declaredLength, body, null, false); } AdaptiveBodyOutputStream output = null; @@ -140,11 +268,13 @@ private ProcessedResponse prepareResponse(HttpServerExchange exchange, HttpServe var content = body.getFullContentIfAvailable(); if (content != null) { var contentLength = declaredLength >= 0 ? declaredLength : content.remaining(); - return new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); } - var streamingResponse = new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, null, declaredLength, body, null, true); - output = new AdaptiveBodyOutputStream(this, exchange, streamingResponse); + var streamingResponse = new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), contentType, null, declaredLength, body, null, true); + output = new AdaptiveBodyOutputStream(this, exchange, streamingResponse, declaredLength); + // HttpBodyOutput.write() must flush whatever it wraps around this stream before returning: the buffer is + // read back as soon as write() returns and close() is never called on the writer's behalf. body.write(output); if (output.isStreaming()) { output.completeStreaming(); @@ -153,35 +283,42 @@ private ProcessedResponse prepareResponse(HttpServerExchange exchange, HttpServe content = output.fullContent(); var contentLength = declaredLength >= 0 ? declaredLength : content.remaining(); - return new ProcessedResponse(observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); } catch (Throwable e) { - observation.observeError(e); if (output != null && output.isStreaming()) { + // AsyncBodyPipe observes its own terminal failure. Once it has terminated, the next write() only + // fails because the pipe is already closed, so reporting it here would count the error twice. output.failStreaming(e); return output.streamingResponse(); } + observation.observeError(e); if (output != null && output.size() > 0) { var content = output.fullContent(); + // deliberately one byte longer than what is actually sent: the client has to see the response as + // truncated, the connection is then closed by ProcessedResponse.onComplete() var contentLength = declaredLength >= 0 ? declaredLength : content.remaining() + 1L; - return new ProcessedResponse(observation, context, response.code(), response.headers(), body.contentType(), content, contentLength, body, e, false); + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), body.contentType(), content, contentLength, body, e, false); } - return errorResponse(observation, context, body, e); + return this.errorResponse(exchange, observation, context, body, e); } } - private static ProcessedResponse errorResponse(HttpServerObservation observation, Context context, @Nullable HttpBodyOutput bodyToClose, Throwable error) { + private ProcessedResponse errorResponse(HttpServerExchange exchange, HttpServerObservation observation, Context context, @Nullable HttpBodyOutput bodyToClose, Throwable error) { var message = Objects.requireNonNullElse(error.getMessage(), "Unknown error"); - var content = message.getBytes(StandardCharsets.UTF_8); - var response = observation.observeResponse(HttpServerResponse.of(500, HttpBody.plaintext(message))); + var errorBody = HttpBody.plaintext(message); + var response = observation.observeResponse(HttpServerResponse.of(500, errorBody)); + var content = errorBody.getFullContentIfAvailable(); return new ProcessedResponse( + this, + exchange, observation, context, response.code(), response.headers(), - "text/plain;charset=utf-8", - ByteBuffer.wrap(content), - content.length, + errorBody.contentType(), + content, + content.remaining(), bodyToClose, null, false @@ -199,30 +336,8 @@ private void sendResponse(HttpServerExchange exchange, ProcessedResponse respons return; } exchange.setResponseContentLength(response.contentLength()); - new AsyncSenderImpl(exchange).send(content, new IoCallback() { - @Override - public void onComplete(HttpServerExchange exchange, Sender sender) { - closeBody(observation, response.body()); - if (response.sendFailure() != null) { - observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); - try { - exchange.getConnection().close(); - } catch (IOException closeException) { - response.sendFailure().addSuppressed(closeException); - } - return; - } - IoCallback.END_EXCHANGE.onComplete(exchange, sender); - } - - @Override - public void onException(HttpServerExchange exchange, Sender sender, IOException exception) { - observation.observeError(exception); - observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); - closeBody(observation, response.body()); - IoCallback.END_EXCHANGE.onException(exchange, sender, exception); - } - }); + // getResponseSender() reuses the sender cached on the exchange, ProcessedResponse is its own IoCallback + exchange.getResponseSender().send(content, response); } catch (Throwable e) { observation.observeError(e); observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); @@ -243,7 +358,7 @@ private void sendStreamingResponse(HttpServerExchange exchange, ProcessedRespons if (response.contentLength() >= 0) { exchange.setResponseContentLength(response.contentLength()); } - pipe.attach(new AsyncSenderImpl(exchange)); + pipe.attach(exchange.getResponseSender()); } catch (Throwable e) { pipe.abort(e); exchange.endExchange(); @@ -252,22 +367,19 @@ private void sendStreamingResponse(HttpServerExchange exchange, ProcessedRespons } private void prepareExchange(HttpServerExchange exchange, ProcessedResponse response) { + var responseHeaders = exchange.getResponseHeaders(); + if (this.contextPropagationEnabled) { + PROPAGATOR.inject(response.context(), responseHeaders, HttpServerExchangeMapGetter.INSTANCE); + } if (this.telemetryEnabled) { - W3CTraceContextPropagator.getInstance().inject( - response.context(), - exchange.getResponseHeaders(), - HttpServerExchangeMapGetter.INSTANCE - ); - exchange.addExchangeCompleteListener((e, nextListener) -> { - response.observation().end(); - nextListener.proceed(); - }); + exchange.addExchangeCompleteListener(response); } exchange.setStatusCode(response.code()); - exchange.getResponseHeaders().put(Headers.SERVER, "Kora"); - this.setHeaders(exchange.getResponseHeaders(), response.headers(), response.contentType()); - if (response.contentType() != null) { - exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, response.contentType()); + responseHeaders.put(Headers.SERVER, "Kora"); + var contentType = response.contentType(); + setHeaders(responseHeaders, response.headers(), contentType); + if (contentType != null) { + responseHeaders.put(Headers.CONTENT_TYPE, contentType); } } @@ -282,7 +394,14 @@ private static void closeBody(HttpServerObservation observation, @Nullable HttpB } } - private record ProcessedResponse(HttpServerObservation observation, + /** + * Carries everything needed to write one response back to the client. It implements {@link Runnable}, + * {@link IoCallback} and {@link ExchangeCompletionListener} itself so that the send path does not allocate a + * dispatch lambda, an I/O callback and a completion listener on top of it for every single request. + */ + private record ProcessedResponse(KoraRequestProcessingHttpHandler handler, + HttpServerExchange exchange, + HttpServerObservation observation, Context context, int code, HttpHeaders headers, @@ -291,30 +410,90 @@ private record ProcessedResponse(HttpServerObservation observation, long contentLength, @Nullable HttpBodyOutput body, @Nullable Throwable sendFailure, - boolean streamingStarted) {} + boolean streamingStarted) implements Runnable, IoCallback, ExchangeCompletionListener { + + @Override + public void run() { + this.handler.sendResponse(this.exchange, this); + } + + @Override + public void exchangeEvent(HttpServerExchange exchange, NextListener nextListener) { + this.observation.end(); + nextListener.proceed(); + } + + @Override + public void onComplete(HttpServerExchange exchange, Sender sender) { + closeBody(this.observation, this.body); + var failure = this.sendFailure; + if (failure != null) { + this.observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); + try { + exchange.getConnection().close(); + } catch (IOException closeException) { + failure.addSuppressed(closeException); + } + return; + } + IoCallback.END_EXCHANGE.onComplete(exchange, sender); + } + + @Override + public void onException(HttpServerExchange exchange, Sender sender, IOException exception) { + this.observation.observeError(exception); + this.observation.observeResultCode(HttpResultCode.CONNECTION_ERROR); + closeBody(this.observation, this.body); + IoCallback.END_EXCHANGE.onException(exchange, sender, exception); + } + } private static final class AdaptiveBodyOutputStream extends OutputStream { private static final int SMALL_BODY_THRESHOLD = 64 * 1024; private static final int STREAM_CHUNK_SIZE = 16 * 1024; + private static final int MIN_BUFFER_SIZE = 64; + private static final int DEFAULT_BUFFER_SIZE = 128; + private static final byte[] EMPTY_BUFFER = new byte[0]; private final KoraRequestProcessingHttpHandler handler; private final HttpServerExchange exchange; private final ProcessedResponse streamingResponse; - private byte[] buffer = new byte[256]; + private byte[] buffer; private int size; private @Nullable AsyncBodyPipe pipe; private AdaptiveBodyOutputStream(KoraRequestProcessingHttpHandler handler, HttpServerExchange exchange, - ProcessedResponse streamingResponse) { + ProcessedResponse streamingResponse, + long declaredLength) { this.handler = handler; this.exchange = exchange; this.streamingResponse = streamingResponse; + this.buffer = new byte[initialCapacity(declaredLength)]; + } + + /** + * Most bodies declare their length up front, so the buffer can be sized once instead of doubling up from a + * fixed 256 bytes (an 8 KiB body used to cost 6 allocations and 5 array copies). + */ + private static int initialCapacity(long declaredLength) { + if (declaredLength < 0) { + return DEFAULT_BUFFER_SIZE; + } + if (declaredLength > SMALL_BODY_THRESHOLD) { + return MIN_BUFFER_SIZE; + } + return (int) Math.max(MIN_BUFFER_SIZE, declaredLength); } @Override public void write(int value) throws IOException { + if (this.pipe == null && this.size < SMALL_BODY_THRESHOLD) { + this.ensureCapacity(this.size + 1); + this.buffer[this.size++] = (byte) value; + return; + } this.write(new byte[]{(byte) value}, 0, 1); } @@ -355,7 +534,7 @@ private void startStreaming() throws IOException { this.pipe.start(); if (this.size > 0) { this.pipe.offer(ByteBuffer.wrap(this.buffer, 0, this.size)); - this.buffer = new byte[0]; + this.buffer = EMPTY_BUFFER; this.size = 0; } } @@ -425,9 +604,8 @@ private AsyncBodyPipe(KoraRequestProcessingHttpHandler handler, } private void start() { - // Keep Connectors.executeRootHandler() from auto-ending the exchange when the VT producer returns. - this.exchange.dispatch(SameThreadExecutor.INSTANCE, () -> {}); - // dispatch(...) itself is deferred until the handler returns, so the consumer must be started directly. + // handleRequest() already took the dispatch slot, so executeRootHandler() will not auto-end the exchange + // and the producer is off-call by the time the consumer touches the sender. this.exchange.getIoThread().execute(() -> this.handler.sendStreamingResponse(this.exchange, this.response, this)); } @@ -523,25 +701,29 @@ private void finishException(Throwable failure) { } } - private void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { + private static void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { for (var header : headers) { var key = header.getKey(); - if (key.equals("server")) { - continue; - } - if (key.equals("content-type") && contentType != null) { + if (isReservedHeader(key, contentType)) { continue; } - if (key.equals("content-length")) { + var name = headerName(key); + if (name == null) { + logger.warn("HTTP response header with unsupported name was skipped: {}", key); continue; } - if (key.equals("transfer-encoding")) { - continue; - } - responseHeaders.addAll(HttpString.tryFromString(key), header.getValue()); + responseHeaders.addAll(name, header.getValue()); } } + private static boolean isReservedHeader(String key, @Nullable String contentType) { + return switch (key) { + case "server", "content-length", "transfer-encoding" -> true; + case "content-type" -> contentType != null; + default -> false; + }; + } + public static class HttpServerExchangeMapGetter implements TextMapGetter, TextMapSetter { public static final HttpServerExchangeMapGetter INSTANCE = new HttpServerExchangeMapGetter(); @@ -569,7 +751,10 @@ public String get(HeaderMap headers, String key) { @Override public void set(HeaderMap headers, String key, String value) { - headers.add(HttpString.tryFromString(key), value); + var name = headerName(key); + if (name != null) { + headers.add(name, value); + } } } } diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java index 3dd7f94d39..463dc92fca 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java @@ -119,7 +119,8 @@ public void close() { protected HttpServer httpServer(ValueOf config, HttpServerRouter httpServerRouter, HttpServerTelemetry telemetry) { return new UndertowHttpServer( "test", - valueOf(new KoraVirtualThreadPerConnectionDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(config.get(), httpServerRouter, telemetry))), + valueOf(new UndertowConfig() {}), + valueOf(new KoraVirtualThreadPerConnectionDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(valueOf(new UndertowConfig() {}), (ValueOf) config, telemetry, httpServerRouter))), null, config, null From 8ff5be156f1c914dcc89ab05fed4a5ca031832dc Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Wed, 26 Aug 2026 16:32:26 +0300 Subject: [PATCH 5/9] Cleanup --- .../request/UndertowRequestHttpBodyTest.java | 61 ------------------- 1 file changed, 61 deletions(-) delete mode 100644 http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java deleted file mode 100644 index ace4e7d9d7..0000000000 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/request/UndertowRequestHttpBodyTest.java +++ /dev/null @@ -1,61 +0,0 @@ -package io.koraframework.http.server.undertow.request; - -import io.undertow.server.HttpServerExchange; -import org.junit.jupiter.api.Test; - -import java.io.InputStream; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -class UndertowRequestHttpBodyTest { - - @Test - void startsBlockingModeWhenInputStreamIsRequested() { - var exchange = mock(HttpServerExchange.class); - var inputStream = mock(InputStream.class); - when(exchange.getInputStream()).thenReturn(inputStream); - - var body = new UndertowRequestHttpBody(exchange); - - assertThat(body.asInputStream()).isSameAs(inputStream); - verify(exchange).startBlocking(); - } - - @Test - void doesNotStartBlockingModeUntilBodyIsUsed() { - var exchange = mock(HttpServerExchange.class); - - new UndertowRequestHttpBody(exchange); - - verify(exchange, never()).startBlocking(); - verify(exchange, never()).getInputStream(); - } - - @Test - void closesOpenedInputStream() throws Exception { - var exchange = mock(HttpServerExchange.class); - var inputStream = mock(InputStream.class); - when(exchange.isBlocking()).thenReturn(true); - when(exchange.getInputStream()).thenReturn(inputStream); - - new UndertowRequestHttpBody(exchange).close(); - - verify(inputStream).close(); - } - - @Test - void startsBlockingModeWhenUnopenedBodyIsClosed() throws Exception { - var exchange = mock(HttpServerExchange.class); - var inputStream = mock(InputStream.class); - when(exchange.getInputStream()).thenReturn(inputStream); - - new UndertowRequestHttpBody(exchange).close(); - - verify(exchange).startBlocking(); - verify(inputStream).close(); - } -} From 69a61e43ca831c407fe2117f226ad63f2c8a0794 Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Wed, 23 Sep 2026 15:14:07 +0300 Subject: [PATCH 6/9] Merged --- .../undertow/UndertowHttpServerFactoryModule.java | 2 +- .../handler/KoraRequestProcessingHttpHandler.java | 10 +++++----- .../server/undertow/UndertowHttpServerRefreshTest.java | 2 +- .../http/server/undertow/UndertowHttpServerTest.java | 2 +- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java index e81c085ae0..3dc7ccc8e0 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java @@ -42,7 +42,7 @@ public HttpHandler handler(ValueOf undertowConfig, @Tag(Tag.Factory.class) HttpServerRouter httpServerRouter, HttpServerTelemetryFactory telemetryFactory) { var telemetry = telemetryFactory.get(this.name, httpServerConfig.port(), httpServerConfig.telemetry()); - var handler = (HttpHandler) new KoraRequestProcessingHttpHandler(httpServerConfig, httpServerRouter, telemetry); + var handler = (HttpHandler) new KoraRequestProcessingHttpHandler(undertowConfig, httpServerConfig, httpServerRouter, telemetry); handler = new KoraVirtualThreadPerConnectionDispatchHttpHandler(this.name, handler); return handler; } diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java index d47c30df88..e3968992ae 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java @@ -75,14 +75,14 @@ public final class KoraRequestProcessingHttpHandler implements HttpHandler { private final boolean contextPropagationEnabled; public KoraRequestProcessingHttpHandler(ValueOf undertowConfig, - ValueOf config, - HttpServerTelemetry telemetry, - HttpServerRouter httpServerRouter) { + HttpServerConfig config, + HttpServerRouter httpServerRouter, + HttpServerTelemetry telemetry) { this.telemetry = telemetry; this.httpServerRouter = httpServerRouter; this.telemetryEnabled = !(telemetry instanceof NoopHttpServerTelemetry); - this.contextPropagationEnabled = this.telemetryEnabled && config.get().telemetry().tracing().contextPropagation(); - this.httpServerConfig = config.get(); + this.contextPropagationEnabled = this.telemetryEnabled && config.telemetry().tracing().contextPropagation(); + this.httpServerConfig = config; } private static Map lowercaseHeaderNames() { diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerRefreshTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerRefreshTest.java index 4963bdf3a3..ceeac307b0 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerRefreshTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerRefreshTest.java @@ -20,7 +20,7 @@ class UndertowHttpServerRefreshTest { void servesRequestsWithHandlerReplacedByGraphRefresh() throws Exception { var handler = new AtomicReference(exchange -> exchange.setStatusCode(418)); var handlerValue = (ValueOf) handler::get; - var server = new UndertowHttpServer("test", handlerValue, null, config(), null); + var server = new UndertowHttpServer("test", null, handlerValue, null, config(), null); server.init(); try { diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java index 463dc92fca..536a39d766 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java @@ -120,7 +120,7 @@ protected HttpServer httpServer(ValueOf config, Http return new UndertowHttpServer( "test", valueOf(new UndertowConfig() {}), - valueOf(new KoraVirtualThreadPerConnectionDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(valueOf(new UndertowConfig() {}), (ValueOf) config, telemetry, httpServerRouter))), + valueOf(new KoraVirtualThreadPerConnectionDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(valueOf(new UndertowConfig() {}), config.get(), httpServerRouter, telemetry))), null, config, null From b3f9c07eea6967123ddb75735a0445e440763e72 Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Sat, 26 Sep 2026 22:20:06 +0300 Subject: [PATCH 7/9] Cleanup --- .../server/undertow/UndertowHttpServer.java | 14 +- .../UndertowHttpServerFactoryModule.java | 4 +- .../KoraRequestProcessingHttpHandler.java | 140 ++---------------- .../undertow/UndertowHttpServerTest.java | 2 +- 4 files changed, 20 insertions(+), 140 deletions(-) diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java index 1f288bf16b..298f63b2e0 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServer.java @@ -30,7 +30,7 @@ public class UndertowHttpServer implements HttpServer, ReadinessProbe { private final AtomicReference state = new AtomicReference<>(HttpServerState.INIT); private final ValueOf httpHandler; - private final ValueOf config; + private final ValueOf httpServerConfig; private final GracefulShutdownHandler gracefulShutdown; private final String name; private final XnioWorker xnioWorker; @@ -43,10 +43,10 @@ public UndertowHttpServer(String name, ValueOf undertowConfig, ValueOf httpHandler, XnioWorker xnioWorker, - ValueOf config, + ValueOf httpServerConfig, @Nullable Configurer configurer) { this.httpHandler = httpHandler; - this.config = config; + this.httpServerConfig = httpServerConfig; this.name = name; this.xnioWorker = xnioWorker; this.configurer = configurer; @@ -71,9 +71,9 @@ public void init() { logger.info(data, "HTTP Server {} (Undertow) started in {}", name, TimeUtils.tookForLogging(started)); } catch (Exception e) { if (e.getCause() instanceof BindException be) { - throw new IllegalStateException("HTTP server '%s' (Undertow) failed to start on port '%s': port is already in use; stop the other process or configure a different port".formatted(name, config.get().port()), be); + throw new IllegalStateException("HTTP server '%s' (Undertow) failed to start on port '%s': port is already in use; stop the other process or configure a different port".formatted(name, httpServerConfig.get().port()), be); } else { - throw new IllegalStateException("HTTP server '%s' (Undertow) failed to start on port '%s': %s; check server config, handler initialization, and network binding".formatted(name, config.get().port(), e.getMessage()), e); + throw new IllegalStateException("HTTP server '%s' (Undertow) failed to start on port '%s': %s; check server config, handler initialization, and network binding".formatted(name, httpServerConfig.get().port(), e.getMessage()), e); } } } @@ -84,7 +84,7 @@ public void release() { this.state.set(HttpServerState.SHUTDOWN); final long started = TimeUtils.started(); this.gracefulShutdown.shutdown(); - final Duration shutdownAwait = this.config.get().shutdownWait(); + final Duration shutdownAwait = this.httpServerConfig.get().shutdownWait(); try { logger.debug("HTTP Server {} (Undertow) awaiting graceful shutdown...", this.name); if (!this.gracefulShutdown.awaitShutdown(shutdownAwait.toMillis())) { @@ -102,7 +102,7 @@ public void release() { } private Undertow createServer() { - var config = this.config.get(); + var config = this.httpServerConfig.get(); var undertow = Undertow.builder() .setHandler(this.gracefulShutdown) .addHttpListener(config.port(), "0.0.0.0") diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java index 3dc7ccc8e0..0a361cece9 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/UndertowHttpServerFactoryModule.java @@ -30,9 +30,9 @@ public UndertowHttpServerFactoryModule(String name, String configPath) { public UndertowHttpServer server(XnioWorker worker, ValueOf undertowConfig, @Tag(Tag.Factory.class) ValueOf httpHandler, - @Tag(Tag.Factory.class) ValueOf config, + @Tag(Tag.Factory.class) ValueOf httpServerConfig, @Tag(Tag.Factory.class) @Nullable Configurer configurer) { - return new UndertowHttpServer(this.name, undertowConfig, httpHandler, worker, config, configurer); + return new UndertowHttpServer(this.name, undertowConfig, httpHandler, worker, httpServerConfig, configurer); } @DefaultComponent diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java index e3968992ae..fd6e819cc6 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java @@ -41,9 +41,7 @@ import java.io.OutputStream; import java.nio.ByteBuffer; import java.util.Arrays; -import java.util.HashMap; import java.util.Iterator; -import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Semaphore; @@ -56,18 +54,6 @@ public final class KoraRequestProcessingHttpHandler implements HttpHandler { private static final W3CTraceContextPropagator PROPAGATOR = W3CTraceContextPropagator.getInstance(); - private static final HttpString TRACE_PARENT = new HttpString("traceparent"); - private static final HttpString TRACE_STATE = new HttpString("tracestate"); - - /** - * Kora normalizes every header name to lower case, while Undertow's {@link Headers} cache is keyed by the - * canonical spelling ({@code Content-Type}). Looking a lower case name up via {@link HttpString#tryFromString} - * therefore always misses that cache and allocates a fresh {@link HttpString} plus its backing byte array for - * every header of every response. This table restores the cache hit: {@link HttpString} hashes and compares - * case-insensitively, so the canonical instances behave exactly like freshly built lower case ones. - */ - private static final Map HEADER_NAMES = lowercaseHeaderNames(); - private final HttpServerConfig httpServerConfig; private final HttpServerTelemetry telemetry; private final HttpServerRouter httpServerRouter; @@ -75,120 +61,14 @@ public final class KoraRequestProcessingHttpHandler implements HttpHandler { private final boolean contextPropagationEnabled; public KoraRequestProcessingHttpHandler(ValueOf undertowConfig, - HttpServerConfig config, + HttpServerConfig httpServerConfig, HttpServerRouter httpServerRouter, HttpServerTelemetry telemetry) { this.telemetry = telemetry; this.httpServerRouter = httpServerRouter; this.telemetryEnabled = !(telemetry instanceof NoopHttpServerTelemetry); - this.contextPropagationEnabled = this.telemetryEnabled && config.telemetry().tracing().contextPropagation(); - this.httpServerConfig = config; - } - - private static Map lowercaseHeaderNames() { - var names = new HashMap(256); - names.put("accept", Headers.ACCEPT); - names.put("accept-charset", Headers.ACCEPT_CHARSET); - names.put("accept-encoding", Headers.ACCEPT_ENCODING); - names.put("accept-language", Headers.ACCEPT_LANGUAGE); - names.put("accept-ranges", Headers.ACCEPT_RANGES); - names.put("age", Headers.AGE); - names.put("allow", Headers.ALLOW); - names.put("authentication-info", Headers.AUTHENTICATION_INFO); - names.put("authorization", Headers.AUTHORIZATION); - names.put("cache-control", Headers.CACHE_CONTROL); - names.put("connection", Headers.CONNECTION); - names.put("content-disposition", Headers.CONTENT_DISPOSITION); - names.put("content-encoding", Headers.CONTENT_ENCODING); - names.put("content-language", Headers.CONTENT_LANGUAGE); - names.put("content-length", Headers.CONTENT_LENGTH); - names.put("content-location", Headers.CONTENT_LOCATION); - names.put("content-md5", Headers.CONTENT_MD5); - names.put("content-range", Headers.CONTENT_RANGE); - names.put("content-security-policy", Headers.CONTENT_SECURITY_POLICY); - names.put("content-transfer-encoding", Headers.CONTENT_TRANSFER_ENCODING); - names.put("content-transfer-encoding", Headers.CONTENT_TRANSFER_ENCODING); - names.put("content-type", Headers.CONTENT_TYPE); - names.put("cookie", Headers.COOKIE); - names.put("cookie2", Headers.COOKIE2); - names.put("date", Headers.DATE); - names.put("etag", Headers.ETAG); - names.put("expect", Headers.EXPECT); - names.put("expires", Headers.EXPIRES); - names.put("forwarded", Headers.FORWARDED); - names.put("from", Headers.FROM); - names.put("host", Headers.HOST); - names.put("if-match", Headers.IF_MATCH); - names.put("if-modified-since", Headers.IF_MODIFIED_SINCE); - names.put("if-none-match", Headers.IF_NONE_MATCH); - names.put("if-range", Headers.IF_RANGE); - names.put("if-unmodified-since", Headers.IF_UNMODIFIED_SINCE); - names.put("last-modified", Headers.LAST_MODIFIED); - names.put("location", Headers.LOCATION); - names.put("max-forwards", Headers.MAX_FORWARDS); - names.put("origin", Headers.ORIGIN); - names.put("pragma", Headers.PRAGMA); - names.put("proxy-authenticate", Headers.PROXY_AUTHENTICATE); - names.put("proxy-authorization", Headers.PROXY_AUTHORIZATION); - names.put("range", Headers.RANGE); - names.put("referer", Headers.REFERER); - names.put("referrer-policy", Headers.REFERRER_POLICY); - names.put("refresh", Headers.REFRESH); - names.put("retry-after", Headers.RETRY_AFTER); - names.put("sec-websocket-accept", Headers.SEC_WEB_SOCKET_ACCEPT); - names.put("sec-websocket-extensions", Headers.SEC_WEB_SOCKET_EXTENSIONS); - names.put("sec-websocket-key", Headers.SEC_WEB_SOCKET_KEY); - names.put("sec-websocket-key1", Headers.SEC_WEB_SOCKET_KEY1); - names.put("sec-websocket-key2", Headers.SEC_WEB_SOCKET_KEY2); - names.put("sec-websocket-location", Headers.SEC_WEB_SOCKET_LOCATION); - names.put("sec-websocket-origin", Headers.SEC_WEB_SOCKET_ORIGIN); - names.put("sec-websocket-protocol", Headers.SEC_WEB_SOCKET_PROTOCOL); - names.put("sec-websocket-version", Headers.SEC_WEB_SOCKET_VERSION); - names.put("secure_protocol", Headers.SECURE_PROTOCOL); - names.put("server", Headers.SERVER); - names.put("servlet-engine", Headers.SERVLET_ENGINE); - names.put("set-cookie", Headers.SET_COOKIE); - names.put("set-cookie2", Headers.SET_COOKIE2); - names.put("ssl_cipher", Headers.SSL_CIPHER); - names.put("ssl_cipher_usekeysize", Headers.SSL_CIPHER_USEKEYSIZE); - names.put("ssl_client_cert", Headers.SSL_CLIENT_CERT); - names.put("ssl_session_id", Headers.SSL_SESSION_ID); - names.put("status", Headers.STATUS); - names.put("strict-transport-security", Headers.STRICT_TRANSPORT_SECURITY); - names.put("te", Headers.TE); - names.put("trailer", Headers.TRAILER); - names.put("transfer-encoding", Headers.TRANSFER_ENCODING); - names.put("upgrade", Headers.UPGRADE); - names.put("user-agent", Headers.USER_AGENT); - names.put("vary", Headers.VARY); - names.put("via", Headers.VIA); - names.put("warning", Headers.WARNING); - names.put("www-authenticate", Headers.WWW_AUTHENTICATE); - names.put("x-content-length", Headers.X_CONTENT_LENGTH); - names.put("x-content-type-options", Headers.X_CONTENT_TYPE_OPTIONS); - names.put("x-disable-push", Headers.X_DISABLE_PUSH); - names.put("x-forwarded-for", Headers.X_FORWARDED_FOR); - names.put("x-forwarded-host", Headers.X_FORWARDED_HOST); - names.put("x-forwarded-port", Headers.X_FORWARDED_PORT); - names.put("x-forwarded-proto", Headers.X_FORWARDED_PROTO); - names.put("x-forwarded-server", Headers.X_FORWARDED_SERVER); - names.put("x-frame-options", Headers.X_FRAME_OPTIONS); - names.put("x-xss-protection", Headers.X_XSS_PROTECTION); - names.put("traceparent", TRACE_PARENT); - names.put("tracestate", TRACE_STATE); - return Map.copyOf(names); - } - - /** - * @return interned {@link HttpString} for an already lower cased header name, or null if the name can not be - * encoded as a header name at all. - */ - @Nullable - private static HttpString headerName(String lowercaseName) { - var cached = HEADER_NAMES.get(lowercaseName); - return cached != null - ? cached - : HttpString.tryFromString(lowercaseName); + this.contextPropagationEnabled = this.telemetryEnabled && httpServerConfig.telemetry().tracing().contextPropagation(); + this.httpServerConfig = httpServerConfig; } @Override @@ -228,9 +108,6 @@ private ProcessedResponse processRequest(HttpServerExchange exchange) { .where(Observation.VALUE, observation) .call(() -> { HttpServerResponse response; - if (httpServerConfig.headerServerNameEnabled()) { - exchange.getResponseHeaders().put(Headers.SERVER, "Kora"); - } try { var httpServerRequest = observation.observeRequest(invocation.routedRequest()); response = invocation.proceed(httpServerRequest); @@ -375,7 +252,6 @@ private void prepareExchange(HttpServerExchange exchange, ProcessedResponse resp exchange.addExchangeCompleteListener(response); } exchange.setStatusCode(response.code()); - responseHeaders.put(Headers.SERVER, "Kora"); var contentType = response.contentType(); setHeaders(responseHeaders, response.headers(), contentType); if (contentType != null) { @@ -701,13 +577,17 @@ private void finishException(Throwable failure) { } } - private static void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { + private void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { + if (this.httpServerConfig.headerServerNameEnabled()) { + responseHeaders.put(Headers.SERVER, "Kora"); + } + for (var header : headers) { var key = header.getKey(); if (isReservedHeader(key, contentType)) { continue; } - var name = headerName(key); + var name = HttpString.tryFromString(key); if (name == null) { logger.warn("HTTP response header with unsupported name was skipped: {}", key); continue; @@ -751,7 +631,7 @@ public String get(HeaderMap headers, String key) { @Override public void set(HeaderMap headers, String key, String value) { - var name = headerName(key); + var name = HttpString.tryFromString(key); if (name != null) { headers.add(name, value); } diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java index 536a39d766..5643091f29 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java @@ -122,7 +122,7 @@ protected HttpServer httpServer(ValueOf config, Http valueOf(new UndertowConfig() {}), valueOf(new KoraVirtualThreadPerConnectionDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(valueOf(new UndertowConfig() {}), config.get(), httpServerRouter, telemetry))), null, - config, + (ValueOf) config, null ); } From bddaa27e31873113758096f2fed32c4cef11e565 Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Sat, 26 Sep 2026 22:39:19 +0300 Subject: [PATCH 8/9] Added(http-server): - RawHttpClient in the kit's test fixtures: a small socket client for keep-alive, pipelining, reading headers separately, and abrupt disconnects, which OkHttp can't do. The kit also gets a port() accessor. - DefaultFullHttpBodyTest in http-common (28 tests): the direct-buffer fix, at sizes around the 1 KiB copy buffer. --- .../http/server/common/HttpServerTestKit.java | 486 ++++++++++++++++++ .../http/server/common/RawHttpClient.java | 154 ++++++ .../undertow/UndertowHttpServerTest.java | 76 +++ 3 files changed, 716 insertions(+) create mode 100644 http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/RawHttpClient.java diff --git a/http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/HttpServerTestKit.java b/http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/HttpServerTestKit.java index 76d5e3929d..9c28fa86dd 100644 --- a/http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/HttpServerTestKit.java +++ b/http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/HttpServerTestKit.java @@ -37,6 +37,11 @@ import org.jetbrains.annotations.NotNull; import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.*; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.AdditionalAnswers; import org.mockito.ArgumentMatchers; import org.mockito.Mockito; @@ -51,10 +56,16 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Optional; +import java.util.SplittableRandom; import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; +import java.util.stream.Stream; import static io.koraframework.http.common.HttpMethod.GET; import static io.koraframework.http.common.HttpMethod.POST; @@ -387,6 +398,477 @@ public void close() throws IOException { //todo request body tests } + /** + * Server-agnostic response body contract: bytes are delivered intact whatever the size, declared length and write + * pattern, the body is closed exactly once on every path, failures after the status line truncate the response, + * and a slow or vanished client never makes the server buffer the whole response. + */ + @Nested + public class ResponseBodyTest { + + enum WritePattern { + SINGLE_WRITE, + RANDOM_SLICES, + BYTE_BY_BYTE, + WITH_EMPTY_WRITES + } + + enum BufferKind { + HEAP, + HEAP_SLICE, + READ_ONLY, + DIRECT + } + + enum Failure { + THROW_BEFORE_FIRST_BYTE, + THROW_AFTER_FIRST_BYTES, + THROW_AFTER_LARGE_PART, + SHORT_SMALL_BODY_OF_DECLARED_LENGTH, + SHORT_LARGE_BODY_OF_DECLARED_LENGTH + } + + static Stream streamingBodies() { + // common buffer boundaries of HTTP servers: 8, 16 and 64 KiB + var sizes = new int[]{0, 1, 8 * 1024 - 1, 8 * 1024 + 1, 16 * 1024, 64 * 1024 - 1, 64 * 1024, 64 * 1024 + 1, 200 * 1024 + 3, 1024 * 1024 + 7}; + var arguments = new ArrayList(); + for (var size : sizes) { + for (var declaredLength : new boolean[]{true, false}) { + for (var pattern : WritePattern.values()) { + if (pattern == WritePattern.BYTE_BY_BYTE && size > 200 * 1024 + 3) { + continue; + } + arguments.add(Arguments.of(size, declaredLength, pattern)); + } + } + } + return arguments.stream(); + } + + static Stream fullBodies() { + var arguments = new ArrayList(); + for (var size : new int[]{0, 1, 1023, 64 * 1024 + 1, 1024 * 1024 + 3}) { + for (var kind : BufferKind.values()) { + arguments.add(Arguments.of(size, kind)); + } + } + return arguments.stream(); + } + + @ParameterizedTest(name = "{0} bytes, declared length: {1}, {2}") + @MethodSource("streamingBodies") + void streamingBodyIsDeliveredIntact(int size, boolean declaredLength, WritePattern pattern) throws Exception { + var data = payload(size, size); + var body = new TrackingBody(declaredLength ? size : -1, os -> write(os, data, pattern, size)); + startServer(handler(GET, "/body", _ -> HttpServerResponse.of(200, body))); + + try (var response = client.newCall(request("/body").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().bytes()).isEqualTo(data); + var contentLength = response.header("Content-Length"); + if (declaredLength) { + assertThat(contentLength).isEqualTo(Integer.toString(size)); + } else if (contentLength == null) { + assertThat(response.header("Transfer-Encoding")).isEqualToIgnoringCase("chunked"); + } else { + assertThat(contentLength).isEqualTo(Integer.toString(size)); + } + } + body.assertClosedOnce(); + } + + @ParameterizedTest(name = "{0} bytes, {1} buffer") + @MethodSource("fullBodies") + void fullContentBodyIsSentWithContentLength(int size, BufferKind kind) throws Exception { + var data = payload(size, 31L * size + kind.ordinal()); + startServer(handler(GET, "/full", _ -> HttpServerResponse.of(200, HttpBody.octetStream(buffer(data, kind))))); + + // twice: the same body instance must not be consumed by the first response + for (var i = 0; i < 2; i++) { + try (var response = client.newCall(request("/full").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.header("Content-Length")).isEqualTo(Integer.toString(size)); + assertThat(response.body().bytes()).isEqualTo(data); + } + } + } + + @ParameterizedTest(name = "declared length {0}") + @ValueSource(longs = {-1, 0, 100, 64 * 1024 + 1, 10 * 1024 * 1024}) + void headResponseNeverProducesBody(long declaredLength) throws Exception { + var body = new TrackingBody(declaredLength, os -> os.write(new byte[1024])); + startServer(handler("HEAD", "/head", _ -> HttpServerResponse.of(200, body))); + + try (var response = client.newCall(request("/head").head().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().bytes()).isEmpty(); + if (declaredLength >= 0) { + assertThat(response.header("Content-Length")).isEqualTo(Long.toString(declaredLength)); + } + } + body.assertClosedOnce(); + assertThat(body.writes).hasValue(0); + } + + @ParameterizedTest + @EnumSource(Failure.class) + void bodyFailureIsReportedAndServerKeepsServing(Failure failure) throws Exception { + var body = new TrackingBody(switch (failure) { + case SHORT_SMALL_BODY_OF_DECLARED_LENGTH -> 1000; + case SHORT_LARGE_BODY_OF_DECLARED_LENGTH -> 300 * 1024; + default -> -1; + }, os -> { + switch (failure) { + case THROW_BEFORE_FIRST_BYTE -> throw new IllegalStateException("boom"); + case THROW_AFTER_FIRST_BYTES -> { + os.write(new byte[10]); + os.flush(); + throw new IllegalStateException("boom"); + } + case THROW_AFTER_LARGE_PART -> { + os.write(new byte[100 * 1024]); + throw new IllegalStateException("boom"); + } + case SHORT_SMALL_BODY_OF_DECLARED_LENGTH -> os.write(new byte[500]); + case SHORT_LARGE_BODY_OF_DECLARED_LENGTH -> os.write(new byte[200 * 1024]); + } + }); + startServer( + handler(GET, "/failing", _ -> HttpServerResponse.of(200, body)), + handler(GET, "/ok", _ -> HttpServerResponse.of(200, HttpBody.plaintext("ok"))) + ); + + if (failure == Failure.THROW_BEFORE_FIRST_BYTE) { + try (var response = client.newCall(request("/failing").get().build()).execute()) { + assertThat(response.code()).isEqualTo(500); + assertThat(response.body().string()).isEqualTo("boom"); + } + } else { + // once the body has started the client must see the response truncated, never complete + assertThatThrownBy(() -> { + try (var response = client.newCall(request("/failing").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + response.body().bytes(); + } + }).isInstanceOf(IOException.class); + } + body.assertClosedOnce(); + + try (var response = client.newCall(request("/ok").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().string()).isEqualTo("ok"); + } + } + + @ParameterizedTest(name = "{0} bytes") + @ValueSource(ints = {1024, 1024 * 1024}) + void bodyCloseFailureDoesNotBreakDeliveredResponse(int size) throws Exception { + var data = payload(size, 11); + var closes = new AtomicInteger(); + var body = new HttpBodyOutput() { + @Override + public long contentLength() { + return -1; + } + + @Override + public String contentType() { + return "application/octet-stream"; + } + + @Override + public void write(OutputStream os) throws IOException { + os.write(data); + } + + @Override + public void close() throws IOException { + closes.incrementAndGet(); + throw new IOException("close failed"); + } + }; + startServer(handler(GET, "/close-failure", _ -> HttpServerResponse.of(200, body))); + + for (var i = 0; i < 2; i++) { + try (var response = client.newCall(request("/close-failure").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().bytes()).isEqualTo(data); + } + } + awaitValue(closes, 2); + } + + @Test + void keepAliveConnectionIsReusedAcrossAllResponseKinds() throws Exception { + var small = "small".getBytes(StandardCharsets.UTF_8); + var medium = payload(10 * 1024, 1); + var unknownLength = payload(200 * 1024 + 5, 2); + var knownLength = payload(300 * 1024 + 9, 3); + startServer( + handler(GET, "/small", _ -> HttpServerResponse.of(200, HttpBody.octetStream(small))), + handler(GET, "/medium", _ -> HttpServerResponse.of(200, HttpBodyOutput.octetStream(os -> os.write(medium)))), + handler(GET, "/unknown-length", _ -> HttpServerResponse.of(200, HttpBodyOutput.octetStream(os -> write(os, unknownLength, WritePattern.RANDOM_SLICES, 2)))), + handler(GET, "/known-length", _ -> HttpServerResponse.of(200, HttpBodyOutput.octetStream(knownLength.length, os -> write(os, knownLength, WritePattern.RANDOM_SLICES, 3)))), + handler(GET, "/empty", _ -> HttpServerResponse.of(200)), + handler(GET, "/error", _ -> { + throw new IllegalStateException("boom"); + }), + handler("HEAD", "/head", _ -> HttpServerResponse.of(200, HttpBodyOutput.octetStream(knownLength.length, os -> os.write(knownLength)))) + ); + + try (var raw = new RawHttpClient(port())) { + for (var round = 0; round < 2; round++) { + assertRawBody(raw.exchange("GET", "/small"), 200, small); + assertRawBody(raw.exchange("GET", "/medium"), 200, medium); + assertRawBody(raw.exchange("GET", "/unknown-length"), 200, unknownLength); + assertRawBody(raw.exchange("GET", "/known-length"), 200, knownLength); + assertRawBody(raw.exchange("GET", "/empty"), 200, new byte[0]); + assertRawBody(raw.exchange("GET", "/error"), 500, "boom".getBytes(StandardCharsets.UTF_8)); + var head = raw.exchange("HEAD", "/head"); + assertRawBody(head, 200, new byte[0]); + assertThat(head.header("Content-Length")).isEqualTo(Integer.toString(knownLength.length)); + } + } + } + + @Test + void pipelinedRequestsAreAnsweredInOrder() throws Exception { + var small = "small".getBytes(StandardCharsets.UTF_8); + var medium = payload(10 * 1024, 4); + var large = payload(300 * 1024 + 1, 5); + startServer( + handler(GET, "/small", _ -> HttpServerResponse.of(200, HttpBody.octetStream(small))), + handler(GET, "/medium", _ -> HttpServerResponse.of(200, HttpBodyOutput.octetStream(os -> os.write(medium)))), + handler(GET, "/large", _ -> HttpServerResponse.of(200, HttpBodyOutput.octetStream(os -> write(os, large, WritePattern.RANDOM_SLICES, 5)))) + ); + + try (var raw = new RawHttpClient(port())) { + raw.send( + RawHttpClient.request("GET", "/small"), + RawHttpClient.request("GET", "/large"), + RawHttpClient.request("GET", "/medium"), + RawHttpClient.request("GET", "/large"), + RawHttpClient.request("GET", "/small") + ); + assertRawBody(raw.readResponse(false), 200, small); + assertRawBody(raw.readResponse(false), 200, large); + assertRawBody(raw.readResponse(false), 200, medium); + assertRawBody(raw.readResponse(false), 200, large); + assertRawBody(raw.readResponse(false), 200, small); + } + } + + @Test + void slowClientDoesNotMakeServerBufferWholeResponse() throws Exception { + var chunk = payload(16 * 1024, 6); + var total = 32L * 1024 * 1024; + var produced = new AtomicLong(); + var body = new TrackingBody(-1, os -> { + while (produced.get() < total) { + os.write(chunk); + produced.addAndGet(chunk.length); + } + }); + startServer(handler(GET, "/slow", _ -> HttpServerResponse.of(200, body))); + + try (var raw = new RawHttpClient(port(), 16 * 1024)) { + raw.send(RawHttpClient.request("GET", "/slow")); + var head = raw.readHead(); + + // the client stops reading: the producer has to stall once socket and server buffers are full + Thread.sleep(250); + var stalledAt = produced.get(); + Thread.sleep(250); + assertThat(stalledAt).isLessThan(total); + assertThat(produced.get() - stalledAt).isLessThanOrEqualTo(1024 * 1024); + + var received = raw.readBody(head); + assertThat(received).hasSize((int) total); + for (var offset = 0; offset < received.length; offset += chunk.length) { + assertThat(Arrays.equals(received, offset, offset + chunk.length, chunk, 0, chunk.length)) + .as("chunk at offset %d", offset) + .isTrue(); + } + } + body.assertClosedOnce(); + } + + @Test + void clientDisconnectMidStreamStopsProducerAndClosesBody() throws Exception { + var chunk = payload(16 * 1024, 8); + var limit = 256L * 1024 * 1024; + var produced = new AtomicLong(); + var producerFailure = new AtomicReference(); + var body = new TrackingBody(-1, os -> { + try { + while (produced.get() < limit) { + os.write(chunk); + produced.addAndGet(chunk.length); + } + } catch (IOException e) { + producerFailure.set(e); + throw e; + } + }); + startServer( + handler(GET, "/endless", _ -> HttpServerResponse.of(200, body)), + handler(GET, "/ok", _ -> HttpServerResponse.of(200, HttpBody.plaintext("ok"))) + ); + + try (var raw = new RawHttpClient(port())) { + raw.send(RawHttpClient.request("GET", "/endless")); + raw.readHead(); + raw.readBytes(128 * 1024); + raw.abort(); + } + + body.assertClosedOnce(); + assertThat(producerFailure.get()).isInstanceOf(IOException.class); + assertThat(produced.get()).isLessThan(limit); + try (var response = client.newCall(request("/ok").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().string()).isEqualTo("ok"); + } + } + + @Test + void concurrentStreamsDoNotMixContent() throws Exception { + var streams = 24; + startServer(handler(GET, "/stream/{id}", request -> { + var id = Integer.parseInt(request.pathParams().get("id")); + var data = payload(256 * 1024 + id * 1013, id); + return HttpServerResponse.of(200, HttpBodyOutput.octetStream(os -> write(os, data, WritePattern.RANDOM_SLICES, id))); + })); + + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var results = new ArrayList>(); + for (var id = 0; id < streams; id++) { + var path = "/stream/" + id; + results.add(executor.submit(() -> { + try (var response = client.newCall(request(path).get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + return response.body().bytes(); + } + })); + } + for (var id = 0; id < streams; id++) { + assertThat(results.get(id).get(30, TimeUnit.SECONDS)) + .as("stream %d", id) + .isEqualTo(payload(256 * 1024 + id * 1013, id)); + } + } + } + } + + /** + * Response body that counts writes and closes, see {@link ResponseBodyTest}. + */ + protected static final class TrackingBody implements HttpBodyOutput { + + private final long contentLength; + private final HttpBodyOutput.HttpBodyWriter writer; + private final AtomicInteger closes = new AtomicInteger(); + private final CountDownLatch closed = new CountDownLatch(1); + public final AtomicInteger writes = new AtomicInteger(); + + public TrackingBody(long contentLength, HttpBodyOutput.HttpBodyWriter writer) { + this.contentLength = contentLength; + this.writer = writer; + } + + @Override + public long contentLength() { + return this.contentLength; + } + + @Override + public @Nullable String contentType() { + return "application/octet-stream"; + } + + @Override + public void write(OutputStream os) throws IOException { + this.writes.incrementAndGet(); + this.writer.write(os); + } + + @Override + public void close() { + this.closes.incrementAndGet(); + this.closed.countDown(); + } + + public void assertClosedOnce() throws InterruptedException { + assertThat(this.closed.await(5, TimeUnit.SECONDS)).as("body closed").isTrue(); + // a second close would come from another completion path shortly after the first one + Thread.sleep(20); + assertThat(this.closes).as("body close count").hasValue(1); + } + } + + protected static byte[] payload(int size, long seed) { + var bytes = new byte[size]; + new SplittableRandom(seed).nextBytes(bytes); + return bytes; + } + + private static void assertRawBody(RawHttpClient.Response response, int code, byte[] body) { + assertThat(response.code()).isEqualTo(code); + assertThat(response.body()).isEqualTo(body); + } + + private static void awaitValue(AtomicInteger counter, int expected) throws InterruptedException { + var deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos(); + while (counter.get() < expected && System.nanoTime() < deadline) { + Thread.sleep(10); + } + assertThat(counter).hasValue(expected); + } + + private static ByteBuffer buffer(byte[] data, ResponseBodyTest.BufferKind kind) { + return switch (kind) { + case HEAP -> ByteBuffer.wrap(data); + case HEAP_SLICE -> { + var padded = new byte[data.length + 20]; + System.arraycopy(data, 0, padded, 10, data.length); + yield ByteBuffer.wrap(padded, 10, data.length).slice(); + } + case READ_ONLY -> ByteBuffer.wrap(data).asReadOnlyBuffer(); + case DIRECT -> ByteBuffer.allocateDirect(data.length).put(data).flip(); + }; + } + + private static void write(OutputStream os, byte[] data, ResponseBodyTest.WritePattern pattern, long seed) throws IOException { + switch (pattern) { + case SINGLE_WRITE -> os.write(data); + case BYTE_BY_BYTE -> { + for (var b : data) { + os.write(b); + } + } + case RANDOM_SLICES -> { + // written from a larger array at a non-zero offset, in sizes up to 48 KiB + var padded = new byte[data.length + 32]; + System.arraycopy(data, 0, padded, 16, data.length); + var random = new SplittableRandom(seed); + for (var offset = 0; offset < data.length; ) { + var length = Math.min(data.length - offset, 1 + random.nextInt(48 * 1024)); + os.write(padded, 16 + offset, length); + offset += length; + } + } + case WITH_EMPTY_WRITES -> { + var half = data.length / 2; + os.write(new byte[0]); + os.write(data, 0, 0); + os.write(data, 0, half); + os.write(data, half, 0); + os.write(data, half, data.length - half); + os.write(new byte[0]); + } + } + } + @Test void testHelloWorld() throws IOException, InterruptedException { var httpResponse = HttpServerResponse.of(200, HttpBody.plaintext("hello world")); @@ -1070,6 +1552,10 @@ protected Request.Builder request(String path) { return request(this.httpServer.port(), path); } + protected int port() { + return this.httpServer.port(); + } + protected Request.Builder request(int port, String path) { return new Request.Builder() .url("http://localhost:%d%s".formatted(port, path)); diff --git a/http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/RawHttpClient.java b/http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/RawHttpClient.java new file mode 100644 index 0000000000..a05f377bbf --- /dev/null +++ b/http/http-server-common/src/testFixtures/java/io/koraframework/http/server/common/RawHttpClient.java @@ -0,0 +1,154 @@ +package io.koraframework.http.server.common; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; + +/** + * Minimal HTTP/1.1 client over one socket: keep-alive, pipelining, reading headers and body separately + * and abrupt disconnects, none of which OkHttp exposes. + */ +public final class RawHttpClient implements AutoCloseable { + + public record Response(int code, Map headers, byte[] body) { + public String header(String name) { + return this.headers.get(name.toLowerCase(Locale.ROOT)); + } + } + + public record Head(int code, Map headers) { + public String header(String name) { + return this.headers.get(name.toLowerCase(Locale.ROOT)); + } + } + + private final Socket socket; + private final InputStream in; + private final OutputStream out; + + public RawHttpClient(int port) throws IOException { + this(port, 0); + } + + public RawHttpClient(int port, int receiveBufferSize) throws IOException { + this.socket = new Socket(); + if (receiveBufferSize > 0) { + this.socket.setReceiveBufferSize(receiveBufferSize); + } + this.socket.setSoTimeout(10_000); + this.socket.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port)); + this.in = new BufferedInputStream(this.socket.getInputStream()); + this.out = this.socket.getOutputStream(); + } + + public static String request(String method, String path) { + return method + " " + path + " HTTP/1.1\r\nHost: localhost\r\n\r\n"; + } + + public void send(String... requests) throws IOException { + this.out.write(String.join("", requests).getBytes(StandardCharsets.ISO_8859_1)); + this.out.flush(); + } + + public Response exchange(String method, String path) throws IOException { + this.send(request(method, path)); + return this.readResponse(method.equals("HEAD")); + } + + public Response readResponse(boolean headRequest) throws IOException { + var head = this.readHead(); + var body = headRequest || head.code() == 204 || head.code() == 304 + ? new byte[0] + : this.readBody(head); + return new Response(head.code(), head.headers(), body); + } + + public Head readHead() throws IOException { + var statusLine = this.readLine(); + var code = Integer.parseInt(statusLine.split(" ", 3)[1]); + var headers = new HashMap(); + for (var line = this.readLine(); !line.isEmpty(); line = this.readLine()) { + var colon = line.indexOf(':'); + headers.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT), line.substring(colon + 1).trim()); + } + return new Head(code, headers); + } + + public byte[] readBody(Head head) throws IOException { + var contentLength = head.header("content-length"); + if (contentLength != null) { + return this.in.readNBytes(Integer.parseInt(contentLength)); + } + if ("chunked".equalsIgnoreCase(head.header("transfer-encoding"))) { + return this.readChunked(); + } + return this.in.readAllBytes(); + } + + public byte[] readBytes(int count) throws IOException { + var bytes = this.in.readNBytes(count); + if (bytes.length != count) { + throw new EOFException("Expected " + count + " bytes, got " + bytes.length); + } + return bytes; + } + + /** + * Closes with RST instead of FIN, so the server sees the peer disappear mid-response. + */ + public void abort() throws IOException { + this.socket.setSoLinger(true, 0); + this.socket.close(); + } + + @Override + public void close() throws IOException { + this.socket.close(); + } + + private byte[] readChunked() throws IOException { + var body = new ByteArrayOutputStream(); + while (true) { + var sizeLine = this.readLine(); + var extension = sizeLine.indexOf(';'); + var size = Integer.parseInt((extension < 0 ? sizeLine : sizeLine.substring(0, extension)).trim(), 16); + if (size == 0) { + // trailers until the empty line + while (!this.readLine().isEmpty()) { + } + return body.toByteArray(); + } + body.write(this.readBytes(size)); + if (!this.readLine().isEmpty()) { + throw new IOException("Malformed chunk terminator"); + } + } + } + + private String readLine() throws IOException { + var line = new ByteArrayOutputStream(64); + while (true) { + var b = this.in.read(); + if (b < 0) { + throw new EOFException("Connection closed while reading a line"); + } + if (b == '\r') { + if (this.in.read() != '\n') { + throw new IOException("Malformed line ending"); + } + return line.toString(StandardCharsets.ISO_8859_1); + } + line.write(b); + } + } +} diff --git a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java index 5643091f29..ddc9a9cfce 100644 --- a/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java +++ b/http/http-server-undertow/src/test/java/io/koraframework/http/server/undertow/UndertowHttpServerTest.java @@ -5,13 +5,17 @@ import io.koraframework.http.server.common.HttpServerConfig; import io.koraframework.http.server.common.HttpServerTestKit; import io.koraframework.http.common.body.HttpBodyOutput; +import io.koraframework.http.server.common.RawHttpClient; import io.koraframework.http.server.common.request.HttpServerRequestHandlerImpl; import io.koraframework.http.server.common.response.HttpServerResponse; import io.koraframework.http.server.common.router.HttpServerRouter; import io.koraframework.http.server.common.telemetry.HttpServerTelemetry; import io.koraframework.http.server.undertow.handler.KoraRequestProcessingHttpHandler; import io.koraframework.http.server.undertow.handler.KoraVirtualThreadPerConnectionDispatchHttpHandler; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import java.io.IOException; import java.io.OutputStream; @@ -19,6 +23,7 @@ import java.util.Arrays; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import static org.assertj.core.api.Assertions.assertThat; @@ -115,6 +120,77 @@ public void close() { assertThat(closeThread.get().getName()).contains("XNIO").contains("I/O"); } + /** + * Undertow specifics of {@link KoraRequestProcessingHttpHandler} on top of {@link HttpServerTestKit.ResponseBodyTest}: + * bodies of unknown length up to 64 KiB are buffered and sent with a Content-Length, larger ones are streamed + * chunked in 16 KiB pieces with at most 4 chunks pending on a slow client. + */ + @Nested + class UndertowResponseBody { + + static final int SMALL_BODY_THRESHOLD = 64 * 1024; + static final int STREAM_CHUNK_SIZE = 16 * 1024; + static final int MAX_PENDING_CHUNKS = 4; + + @ParameterizedTest(name = "{0} bytes, declared length: {1}") + @CsvSource({ + "0, false", "1, false", "65535, false", "65536, false", "65537, false", "81920, false", "1048583, false", + "0, true", "65536, true", "65537, true", "1048583, true" + }) + void bodyIsBufferedUpToThresholdAndChunkedAboveIt(int size, boolean declaredLength) throws Exception { + var data = payload(size, size); + var body = new TrackingBody(declaredLength ? size : -1, os -> { + // small writes, so crossing the threshold happens in the middle of a write sequence + for (var offset = 0; offset < data.length; offset += 1000) { + os.write(data, offset, Math.min(1000, data.length - offset)); + } + }); + startServer(HttpServerRequestHandlerImpl.get("/body", _ -> HttpServerResponse.of(200, body))); + + try (var response = client.newCall(request("/body").get().build()).execute()) { + assertThat(response.code()).isEqualTo(200); + assertThat(response.body().bytes()).isEqualTo(data); + if (declaredLength || size <= SMALL_BODY_THRESHOLD) { + assertThat(response.header("Content-Length")).isEqualTo(Integer.toString(size)); + assertThat(response.header("Transfer-Encoding")).isNull(); + } else { + assertThat(response.header("Content-Length")).isNull(); + assertThat(response.header("Transfer-Encoding")).isEqualToIgnoringCase("chunked"); + } + } + body.assertClosedOnce(); + } + + @Test + void slowClientKeepsAtMostPendingChunksInMemory() throws Exception { + var chunk = payload(STREAM_CHUNK_SIZE, 6); + var total = 32L * 1024 * 1024; + var produced = new AtomicLong(); + var body = new TrackingBody(-1, os -> { + while (produced.get() < total) { + os.write(chunk); + produced.addAndGet(chunk.length); + } + }); + startServer(HttpServerRequestHandlerImpl.get("/slow", _ -> HttpServerResponse.of(200, body))); + + try (var raw = new RawHttpClient(port(), 16 * 1024)) { + raw.send(RawHttpClient.request("GET", "/slow")); + var head = raw.readHead(); + assertThat(head.header("Transfer-Encoding")).isEqualToIgnoringCase("chunked"); + + Thread.sleep(250); + var stalledAt = produced.get(); + Thread.sleep(250); + assertThat(stalledAt).isLessThan(total); + assertThat(produced.get()).isLessThanOrEqualTo(stalledAt + (long) MAX_PENDING_CHUNKS * STREAM_CHUNK_SIZE); + + assertThat(raw.readBody(head)).hasSize((int) total); + } + body.assertClosedOnce(); + } + } + @Override protected HttpServer httpServer(ValueOf config, HttpServerRouter httpServerRouter, HttpServerTelemetry telemetry) { return new UndertowHttpServer( From 924858f1995202a1a381f70cd8f9bf33f3074510 Mon Sep 17 00:00:00 2001 From: Anton Kurako Date: Sun, 27 Sep 2026 13:19:54 +0300 Subject: [PATCH 9/9] Cleanup propagation --- .../server/common/telemetry/HttpServerTelemetryConfig.java | 4 ---- .../undertow/handler/KoraRequestProcessingHttpHandler.java | 2 +- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java b/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java index ac8ca372be..253b186d87 100644 --- a/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java +++ b/http/http-server-common/src/main/java/io/koraframework/http/server/common/telemetry/HttpServerTelemetryConfig.java @@ -53,10 +53,6 @@ interface HttpServerMetricsConfig extends TelemetryConfig.MetricsConfig { } @ConfigMapper interface HttpServerTracingConfig extends TelemetryConfig.TracingConfig { - default boolean contextPropagation() { - return true; - } - default boolean tracePathFull() { return true; } diff --git a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java index fd6e819cc6..36c027a43a 100644 --- a/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java +++ b/http/http-server-undertow/src/main/java/io/koraframework/http/server/undertow/handler/KoraRequestProcessingHttpHandler.java @@ -67,7 +67,7 @@ public KoraRequestProcessingHttpHandler(ValueOf undertowConfig, this.telemetry = telemetry; this.httpServerRouter = httpServerRouter; this.telemetryEnabled = !(telemetry instanceof NoopHttpServerTelemetry); - this.contextPropagationEnabled = this.telemetryEnabled && httpServerConfig.telemetry().tracing().contextPropagation(); + this.contextPropagationEnabled = this.telemetryEnabled; this.httpServerConfig = httpServerConfig; }