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-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/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-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..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 @@ -1,8 +1,8 @@ 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; 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/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..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; @@ -40,12 +40,13 @@ public class UndertowHttpServer implements HttpServer, ReadinessProbe { private volatile Undertow undertow; 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; @@ -70,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); } } } @@ -83,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())) { @@ -101,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 28b32e316d..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 @@ -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; @@ -28,20 +28,22 @@ 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) ValueOf httpServerConfig, @Tag(Tag.Factory.class) @Nullable Configurer configurer) { - return new UndertowHttpServer(this.name, httpHandler, worker, config, configurer); + return new UndertowHttpServer(this.name, undertowConfig, httpHandler, worker, httpServerConfig, 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()); - var handler = (HttpHandler) new KoraRequestProcessingHttpHandler(httpServerConfig, httpServerRouter, telemetry); - handler = new KoraVirtualThreadDispatchHttpHandler(this.name, handler); + 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 ad47134ac2..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 @@ -1,29 +1,37 @@ 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; 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; 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.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.UndertowMessages; -import io.undertow.io.BufferWritableOutputStream; +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; import io.undertow.util.HeaderValues; 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; @@ -32,54 +40,73 @@ 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); + private static final ByteBuffer BODY_EMPTY = ByteBuffer.allocate(0); + + private static final W3CTraceContextPropagator PROPAGATOR = W3CTraceContextPropagator.getInstance(); private final HttpServerConfig httpServerConfig; private final HttpServerTelemetry telemetry; private final HttpServerRouter httpServerRouter; + private final boolean telemetryEnabled; + private final boolean contextPropagationEnabled; - public KoraRequestProcessingHttpHandler(HttpServerConfig httpServerConfig, + public KoraRequestProcessingHttpHandler(ValueOf undertowConfig, + HttpServerConfig httpServerConfig, HttpServerRouter httpServerRouter, HttpServerTelemetry telemetry) { - this.httpServerConfig = httpServerConfig; this.telemetry = telemetry; this.httpServerRouter = httpServerRouter; + this.telemetryEnabled = !(telemetry instanceof NoopHttpServerTelemetry); + this.contextPropagationEnabled = this.telemetryEnabled; + this.httpServerConfig = httpServerConfig; } @Override public void handleRequest(HttpServerExchange exchange) { - var rootCtx = W3CTraceContextPropagator.getInstance().extract(Context.root(), exchange.getRequestHeaders(), HttpServerExchangeMapGetter.INSTANCE); - ScopedValue + 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 = this.errorResponse(exchange, NoopHttpServerObservation.INSTANCE, Context.root(), null, e); + } + if (!processedResponse.streamingStarted()) { + exchange.getIoThread().execute(processedResponse); + } + } + + private ProcessedResponse processRequest(HttpServerExchange exchange) { + 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) - .run(() -> { - MDC.clear(); + .call(() -> { 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(); - }); - ScopedValue + return ScopedValue .where(OpentelemetryContext.VALUE, ctx) .where(Observation.VALUE, observation) - .run(() -> { + .call(() -> { HttpServerResponse response; try { var httpServerRequest = observation.observeRequest(invocation.routedRequest()); @@ -87,118 +114,496 @@ public void handleRequest(HttpServerExchange exchange) { } 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 this.errorResponse(exchange, observation, ctx, null, e); } - return; } - this.sendResponse(observation, exchange, response); + return this.prepareResponse(exchange, 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 prepareResponse(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(this, exchange, observation, context, response.code(), response.headers(), null, null, -1, null, null, false); } - try (body) { + + var declaredLength = body.contentLength(); + if (headRequest) { + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), body.contentType(), BODY_EMPTY, declaredLength, body, null, false); + } + + AdaptiveBodyOutputStream output = null; + try { var contentType = body.contentType(); - this.setHeaders(exchange.getResponseHeaders(), headers, contentType); - if (contentType != null) { - exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, contentType); + var content = body.getFullContentIfAvailable(); + if (content != null) { + var contentLength = declaredLength >= 0 ? declaredLength : content.remaining(); + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); } - var contentLength = body.contentLength(); - if (contentLength >= 0) { - exchange.setResponseContentLength(contentLength); + + 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(); + return streamingResponse; } - try (var os = exchange.getOutputStream()) { - try { - var full = body.getFullContentIfAvailable(); - if (full != null) { - this.writeBuffer(exchange, os, full); - return; - } - 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; - } + + content = output.fullContent(); + var contentLength = declaredLength >= 0 ? declaredLength : content.remaining(); + return new ProcessedResponse(this, exchange, observation, context, response.code(), response.headers(), contentType, content, contentLength, body, null, false); + } catch (Throwable 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(this, exchange, observation, context, response.code(), response.headers(), body.contentType(), content, contentLength, body, e, false); + } + return this.errorResponse(exchange, observation, context, body, e); + } + } + + private ProcessedResponse errorResponse(HttpServerExchange exchange, HttpServerObservation observation, Context context, @Nullable HttpBodyOutput bodyToClose, Throwable error) { + var message = Objects.requireNonNullElse(error.getMessage(), "Unknown error"); + 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(), + errorBody.contentType(), + content, + content.remaining(), + bodyToClose, + null, + false + ); + } + + private void sendResponse(HttpServerExchange exchange, ProcessedResponse response) { + var observation = response.observation(); + try { + this.prepareExchange(exchange, response); + var content = response.content(); + if (content == null) { + closeBody(observation, response.body()); + exchange.endExchange(); + return; + } + exchange.setResponseContentLength(response.contentLength()); + // 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); + closeBody(observation, response.body()); + exchange.endExchange(); + try { + exchange.getConnection().close(); + } catch (IOException closeException) { + e.addSuppressed(closeException); + } + logger.warn("HTTP response send failed", e); } } - private void setHeaders(HeaderMap responseHeaders, HttpHeaders headers, @Nullable String contentType) { - for (var header : headers) { - var key = header.getKey(); - if (key.equals("server")) { - continue; + private void sendStreamingResponse(HttpServerExchange exchange, ProcessedResponse response, AsyncBodyPipe pipe) { + try { + this.prepareExchange(exchange, response); + if (response.contentLength() >= 0) { + exchange.setResponseContentLength(response.contentLength()); } - if (key.equals("content-type") && contentType != null) { - continue; + pipe.attach(exchange.getResponseSender()); + } catch (Throwable e) { + pipe.abort(e); + exchange.endExchange(); + logger.warn("HTTP streaming response setup failed", e); + } + } + + private void prepareExchange(HttpServerExchange exchange, ProcessedResponse response) { + var responseHeaders = exchange.getResponseHeaders(); + if (this.contextPropagationEnabled) { + PROPAGATOR.inject(response.context(), responseHeaders, HttpServerExchangeMapGetter.INSTANCE); + } + if (this.telemetryEnabled) { + exchange.addExchangeCompleteListener(response); + } + exchange.setStatusCode(response.code()); + var contentType = response.contentType(); + setHeaders(responseHeaders, response.headers(), contentType); + if (contentType != null) { + responseHeaders.put(Headers.CONTENT_TYPE, contentType); + } + } + + private static void closeBody(HttpServerObservation observation, @Nullable HttpBody body) { + if (body == null) { + return; + } + try { + body.close(); + } catch (IOException e) { + observation.observeError(e); + } + } + + /** + * 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, + @Nullable String contentType, + @Nullable ByteBuffer content, + long contentLength, + @Nullable HttpBodyOutput body, + @Nullable Throwable sendFailure, + 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; } - if (key.equals("content-length")) { - continue; + 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; + private int size; + private @Nullable AsyncBodyPipe pipe; + + private AdaptiveBodyOutputStream(KoraRequestProcessingHttpHandler handler, + HttpServerExchange exchange, + 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 (key.equals("transfer-encoding")) { - continue; + 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); + } + + @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; } - responseHeaders.addAll(HttpString.tryFromString(key), header.getValue()); + 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 = EMPTY_BUFFER; + 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 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; + 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; } - if (buffer.hasArray()) { - outputStream.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining()); - return; + + private void start() { + // 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)); + } + + private void attach(Sender sender) { + this.sender = sender; + this.scheduleDrain(); } - try (var pooled = exchange.getConnection().getByteBufferPool().getArrayBackedPool().allocate()) { - if (pooled == null) { - throw UndertowMessages.MESSAGES.failedToAllocateResource(); + + 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); } - 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); + 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) { + if (this.httpServerConfig.headerServerNameEnabled()) { + responseHeaders.put(Headers.SERVER, "Kora"); + } + + for (var header : headers) { + var key = header.getKey(); + if (isReservedHeader(key, contentType)) { + continue; + } + var name = HttpString.tryFromString(key); + if (name == null) { + logger.warn("HTTP response header with unsupported name was skipped: {}", key); + continue; + } + 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(); @@ -226,7 +631,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 = HttpString.tryFromString(key); + if (name != null) { + headers.add(name, value); + } } } } 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/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 90765bb64b..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 @@ -4,20 +4,201 @@ 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.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.KoraVirtualThreadDispatchHttpHandler; +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; +import java.nio.charset.StandardCharsets; +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; 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"); + } + + @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"); + } + + /** + * 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( "test", - valueOf(new KoraVirtualThreadDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(config.get(), httpServerRouter, telemetry))), + valueOf(new UndertowConfig() {}), + valueOf(new KoraVirtualThreadPerConnectionDispatchHttpHandler("uvt", new KoraRequestProcessingHttpHandler(valueOf(new UndertowConfig() {}), config.get(), httpServerRouter, telemetry))), null, - config, + (ValueOf) config, null ); }