Merge pull request #50361 from geoand/#50336

When an error occurs during a streaming response, call `HttpServerResponse.reset()`

Author: geoand

extensions/resteasy-reactive/rest-servlet/runtime/src/main/java/io/quarkus/resteasy/reactive/server/servlet/runtime/ServletRequestContext.java

@@ -724,4 +724,9 @@ public void handle(Void event) {
         });
         return this;
     }
+
+    @Override
+    public void reset() {
+        context.response().reset();
+    }
 }

extensions/resteasy-reactive/rest/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/stream/ErrorDuringStreamingTestCase.java

@@ -0,0 +1,124 @@
+package io.quarkus.resteasy.reactive.server.test.stream;
+
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+import java.util.stream.IntStream;
+
+import jakarta.inject.Inject;
+import jakarta.ws.rs.DefaultValue;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+
+import org.jboss.resteasy.reactive.RestQuery;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.restassured.RestAssured;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.subscription.MultiEmitter;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.http.HttpClosedException;
+import io.vertx.core.http.HttpMethod;
+import io.vertx.core.impl.NoStackTraceException;
+
+public class ErrorDuringStreamingTestCase {
+
+    @RegisterExtension
+    static QuarkusUnitTest runner = new QuarkusUnitTest()
+            .withApplicationRoot(jar -> jar.addClasses(Resource.class));
+
+    @Inject
+    Vertx vertx;
+
+    @Test
+    public void noFailure() throws ExecutionException, InterruptedException, TimeoutException {
+        HttpClient client = null;
+        try {
+            AtomicLong count = new AtomicLong();
+            CompletableFuture<Object> latch = new CompletableFuture<>();
+            client = vertx.createHttpClient();
+            sendRequest(client, "/test", latch, b -> count.getAndIncrement());
+            latch.get(10, TimeUnit.SECONDS);
+            Assertions.assertEquals(2 * 100, count.get());
+        } finally {
+            if (client != null) {
+                client.close().toCompletionStage().toCompletableFuture().get();
+            }
+        }
+    }
+
+    @Test
+    public void failure() throws InterruptedException, TimeoutException, ExecutionException {
+        HttpClient client = null;
+        AtomicLong count = new AtomicLong();
+        try {
+            CompletableFuture<Object> latch = new CompletableFuture<>();
+            client = vertx.createHttpClient();
+            sendRequest(client, "/test?fail=true", latch, b -> count.getAndIncrement());
+            latch.get(10, TimeUnit.SECONDS);
+            fail("The client should have failed as the server reset the connection");
+        } catch (ExecutionException e) {
+            Assertions.assertInstanceOf(HttpClosedException.class, e.getCause());
+            Assertions.assertEquals(100, count.get());
+        } finally {
+            if (client != null) {
+                client.close().toCompletionStage().toCompletableFuture().get();
+            }
+        }
+    }
+
+    private void sendRequest(HttpClient client, String requestURI, CompletableFuture<Object> latch,
+            Consumer<Buffer> bodyConsumer) {
+        Handler<Throwable> failure = latch::completeExceptionally;
+        client.request(HttpMethod.GET, RestAssured.port, "localhost", requestURI)
+                .onFailure(failure)
+                .onSuccess(new Handler<>() {
+                    @Override
+                    public void handle(HttpClientRequest event) {
+                        event.connect().onFailure(failure)
+                                .onSuccess(response -> {
+                                    response
+                                            .handler(bodyConsumer::accept)
+                                            .exceptionHandler(latch::completeExceptionally)
+                                            .end(latch::complete);
+                                });
+
+                    }
+                });
+    }
+
+    @Path("test")
+    public static class Resource {
+
+        @GET
+        public Multi<String> stream(@RestQuery @DefaultValue("false") boolean fail) {
+            return Multi.createFrom().emitter(emitter -> {
+                emit(emitter);
+                if (fail) {
+                    throw new NoStackTraceException("dummy");
+                } else {
+                    emit(emitter);
+                    emitter.complete();
+                }
+            });
+        }
+
+        private static void emit(MultiEmitter<? super String> emitter) {
+            IntStream.range(0, 100).forEach(i -> {
+                emitter.emit(String.valueOf(i));
+            });
+        }
+    }
+}

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/handlers/PublisherResponseHandler.java

@@ -246,6 +246,9 @@ protected void handleException(ResteasyReactiveRequestContext requestContext, Th
             // it will appear to be an SSE value, which is incorrect, so we should only log it and close the connection
             if (requestContext.serverResponse().headWritten()) {
                 log.error("Exception in SSE server handling, impossible to send it to client", t);
+                // HTTP chunked encoding sends an indeterminate number of chunks, but it has to end with an end chunk of zero size to indicate successful transmission.
+                // reset() will cause this last chunk to not be sent, even though every other chunk was sent, and so clients can detect the error
+                requestContext.serverResponse().reset();
             } else {
                 // we can go through the abort chain
                 requestContext.resume(t, true);

independent-projects/resteasy-reactive/server/runtime/src/main/java/org/jboss/resteasy/reactive/server/spi/ServerHttpResponse.java

@@ -50,4 +50,6 @@ public interface ServerHttpResponse extends StreamingResponse<ServerHttpResponse
     boolean isWriteQueueFull();
 
     ServerHttpResponse addDrainHandler(Runnable onDrain);
+
+    void reset();
 }

independent-projects/resteasy-reactive/server/vertx/src/main/java/org/jboss/resteasy/reactive/server/vertx/VertxResteasyReactiveRequestContext.java

@@ -531,6 +531,11 @@ public void handle(Void event) {
         return this;
     }
 
+    @Override
+    public void reset() {
+        response.reset();
+    }
+
     @Override
     public boolean isWriteQueueFull() {
         return response.writeQueueFull();