Handle streaming response errors (#595)

micronaut-projects · Nov 20, 2023 · 3fe286c · 3fe286c
1 parent 48ab7ba
commit 3fe286c
Show file tree

Hide file tree

Showing 5 changed files with 128 additions and 8 deletions.
diff --git a/http-server-jetty/src/test/groovy/io/micronaut/servlet/jetty/JettyErrorSpec.groovy b/http-server-jetty/src/test/groovy/io/micronaut/servlet/jetty/JettyErrorSpec.groovy
@@ -12,8 +12,10 @@ import io.micronaut.http.annotation.Produces
 import io.micronaut.http.annotation.Status
 import io.micronaut.http.client.HttpClient
 import io.micronaut.http.client.annotation.Client
+import io.micronaut.http.client.exceptions.HttpClientResponseException
 import io.micronaut.test.extensions.spock.annotation.MicronautTest
 import jakarta.inject.Inject
+import reactor.core.publisher.Flux
 import spock.lang.Specification
 
 @MicronautTest
@@ -35,14 +37,49 @@ class JettyErrorSpec extends Specification {
         result.header(HttpHeaders.CONTENT_TYPE).startsWith(MediaType.TEXT_PLAIN)
     }
 
+    void "error that occurs with streaming response results in client receiving an error"() {
+        when:
+        client.toBlocking().exchange("/errors/stream-immediate", String)
+
+        then:
+        HttpClientResponseException ex = thrown()
+        ex.status == HttpStatus.INTERNAL_SERVER_ERROR
+        ex.response.body.orElseThrow() == "Internal Server Error: Immediate error"
+    }
+
+    void "error that occurs with streaming response after data sent results in client receiving incomplete data"() {
+        when:
+        client.toBlocking().exchange("/errors/stream-delayed", Integer[].class)
+
+        then:
+        HttpClientResponseException ex = thrown()
+        ex.status == HttpStatus.OK
+        ex.message.contains("Unexpected end-of-input")
+    }
+
     @Requires(property = "spec.name", value = "JettyErrorSpec")
     @Controller("/errors")
     static class ErrorController {
 
         @Get("/local")
         @Produces(MediaType.APPLICATION_PDF)
         String localHandler() {
-            throw new AnotherException("bad things");
+            throw new AnotherException("bad things")
+        }
+
+        @Get("/stream-immediate")
+        Flux<String> streamingImmediateError() {
+            return Flux.error(new IllegalStateException("Immediate error"))
+        }
+
+        @Get("/stream-delayed")
+        Flux<Integer> streamingDelayedError() {
+            return Flux.range(1, 5).map(data -> {
+                if (data == 3) {
+                    throw new IllegalStateException("Delayed error")
+                }
+                return data
+            })
         }
 
         @Error

diff --git a/http-server-tomcat/src/test/groovy/io/micronaut/servlet/tomcat/TomcatErrorSpec.groovy b/http-server-tomcat/src/test/groovy/io/micronaut/servlet/tomcat/TomcatErrorSpec.groovy
@@ -12,8 +12,10 @@ import io.micronaut.http.annotation.Produces
 import io.micronaut.http.annotation.Status
 import io.micronaut.http.client.HttpClient
 import io.micronaut.http.client.annotation.Client
+import io.micronaut.http.client.exceptions.HttpClientResponseException
 import io.micronaut.test.extensions.spock.annotation.MicronautTest
 import jakarta.inject.Inject
+import reactor.core.publisher.Flux
 import spock.lang.Specification
 
 @MicronautTest
@@ -35,14 +37,49 @@ class TomcatErrorSpec extends Specification {
         result.header(HttpHeaders.CONTENT_TYPE).startsWith(MediaType.TEXT_PLAIN)
     }
 
+    void "error that occurs with streaming response results in client receiving an error"() {
+        when:
+        client.toBlocking().exchange("/errors/stream-immediate", String)
+
+        then:
+        HttpClientResponseException ex = thrown()
+        ex.status == HttpStatus.INTERNAL_SERVER_ERROR
+        ex.response.body.orElseThrow() == "Internal Server Error: Immediate error"
+    }
+
+    void "error that occurs with streaming response after data sent results in client receiving incomplete data"() {
+        when:
+        client.toBlocking().exchange("/errors/stream-delayed", Integer[].class)
+
+        then:
+        HttpClientResponseException ex = thrown()
+        ex.status == HttpStatus.OK
+        ex.message.contains("Unexpected end-of-input")
+    }
+
     @Requires(property = "spec.name", value = "TomcatErrorSpec")
     @Controller("/errors")
     static class ErrorController {
 
         @Get("/local")
         @Produces(MediaType.APPLICATION_PDF)
         String localHandler() {
-            throw new AnotherException("bad things");
+            throw new AnotherException("bad things")
+        }
+
+        @Get("/stream-immediate")
+        Flux<String> streamingImmediateError() {
+            return Flux.error(new IllegalStateException("Immediate error"))
+        }
+
+        @Get("/stream-delayed")
+        Flux<Integer> streamingDelayedError() {
+            return Flux.range(1, 5).map(data -> {
+                if (data == 3) {
+                    throw new IllegalStateException("Delayed error")
+                }
+                return data
+            })
         }
 
         @Error

diff --git a/http-server-undertow/src/test/groovy/io/micronaut/servlet/undertow/UndertowErrorSpec.groovy b/http-server-undertow/src/test/groovy/io/micronaut/servlet/undertow/UndertowErrorSpec.groovy
@@ -12,8 +12,10 @@ import io.micronaut.http.annotation.Produces
 import io.micronaut.http.annotation.Status
 import io.micronaut.http.client.HttpClient
 import io.micronaut.http.client.annotation.Client
+import io.micronaut.http.client.exceptions.HttpClientResponseException
 import io.micronaut.test.extensions.spock.annotation.MicronautTest
 import jakarta.inject.Inject
+import reactor.core.publisher.Flux
 import spock.lang.Specification
 
 @MicronautTest
@@ -35,14 +37,49 @@ class UndertowErrorSpec extends Specification {
         result.header(HttpHeaders.CONTENT_TYPE).startsWith(MediaType.TEXT_PLAIN)
     }
 
+    void "error that occurs with streaming response results in client receiving an error"() {
+        when:
+        client.toBlocking().exchange("/errors/stream-immediate", String)
+
+        then:
+        HttpClientResponseException ex = thrown()
+        ex.status == HttpStatus.INTERNAL_SERVER_ERROR
+        ex.response.body.orElseThrow() == "Internal Server Error: Immediate error"
+    }
+
+    void "error that occurs with streaming response after data sent results in client receiving incomplete data"() {
+        when:
+        client.toBlocking().exchange("/errors/stream-delayed", Integer[].class)
+
+        then:
+        HttpClientResponseException ex = thrown()
+        ex.status == HttpStatus.OK
+        ex.message.contains("Unexpected end-of-input")
+    }
+
     @Requires(property = "spec.name", value = "UndertowErrorSpec")
     @Controller("/errors")
     static class ErrorController {
 
         @Get("/local")
         @Produces(MediaType.APPLICATION_PDF)
         String localHandler() {
-            throw new AnotherException("bad things");
+            throw new AnotherException("bad things")
+        }
+
+        @Get("/stream-immediate")
+        Flux<String> streamingImmediateError() {
+            return Flux.error(new IllegalStateException("Immediate error"))
+        }
+
+        @Get("/stream-delayed")
+        Flux<Integer> streamingDelayedError() {
+            return Flux.range(1, 5).map(data -> {
+                if (data == 3) {
+                    throw new IllegalStateException("Delayed error")
+                }
+                return data
+            })
         }
 
         @Error

diff --git a/servlet-core/src/main/java/io/micronaut/servlet/http/ServletHttpHandler.java b/servlet-core/src/main/java/io/micronaut/servlet/http/ServletHttpHandler.java
@@ -420,7 +420,9 @@ private void encodeResponse(ServletExchange<REQ, RES> exchange,
                 } else {
                     // stream case
                     if (exchange.getRequest().isAsyncSupported()) {
-                        Mono.from(servletResponse.stream(publisher)).subscribe(responsePublisherCallback);
+                        Mono.from(servletResponse.stream(publisher)).subscribe(responsePublisherCallback, throwable -> {
+                            responsePublisherCallback.accept(null);
+                        });
                         return;
                     } else {
                         // fallback to blocking

diff --git a/servlet-engine/src/main/java/io/micronaut/servlet/engine/DefaultServletHttpResponse.java b/servlet-engine/src/main/java/io/micronaut/servlet/engine/DefaultServletHttpResponse.java
@@ -35,15 +35,17 @@
 import io.micronaut.http.cookie.Cookie;
 import io.micronaut.http.exceptions.HttpStatusException;
 import io.micronaut.servlet.http.ServletHttpResponse;
+import jakarta.servlet.ServletOutputStream;
+import jakarta.servlet.WriteListener;
+import jakarta.servlet.http.HttpServletResponse;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 
-import jakarta.servlet.ServletOutputStream;
-import jakarta.servlet.WriteListener;
-import jakarta.servlet.http.HttpServletResponse;
 import java.io.BufferedWriter;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -67,6 +69,8 @@
 @Internal
 public class DefaultServletHttpResponse<B> implements ServletHttpResponse<HttpServletResponse, B> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(DefaultServletHttpResponse.class);
+
     private static final byte[] EMPTY_ARRAY = "[]".getBytes();
 
     private final ConversionService conversionService;
@@ -199,7 +203,10 @@ public void onError(Throwable t) {
                     if (t instanceof HttpStatusException) {
                         maybeReportErrorDownstream(t);
                     } else {
-                        emitter.error(t);
+                        if (LOG.isWarnEnabled()) {
+                            LOG.warn("Reactive response received an error after some data has already been written. This error cannot be forwarded to the client.", t);
+                        }
+                        maybeReportErrorDownstream(new HttpStatusException(HttpStatus.INTERNAL_SERVER_ERROR, HttpStatus.INTERNAL_SERVER_ERROR.getReason() + ": " + t.getMessage()));
                     }
                     subscription.cancel();
                 }