Repository: grpc/grpc-java
Issue: Exceptions thrown in StreamObserver.onNext() may be swallowed

For unary calls it appears the exception is basically guaranteed to be swallowed.
See #1638 for some discussion. Basically we try to fail the call with a Status, but if the call already completed successfully (which is especially likely for unary calls) the exception is thrown away.


Changed files:
core/src/main/java/io/grpc/internal/ClientCallImpl.java, core/src/test/java/io/grpc/internal/ClientCallImplTest.java

diff --git a/core/src/main/java/io/grpc/internal/ClientCallImpl.java b/core/src/main/java/io/grpc/internal/ClientCallImpl.java
index 366d0f7b3..baa117507 100644
--- a/core/src/main/java/io/grpc/internal/ClientCallImpl.java
+++ b/core/src/main/java/io/grpc/internal/ClientCallImpl.java
@@ -414,8 +414,10 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT>
 
             observer.onHeaders(headers);
           } catch (Throwable t) {
-            stream.cancel(Status.CANCELLED.withCause(t).withDescription("Failed to read headers"));
-            return;
+            Status status =
+                Status.CANCELLED.withCause(t).withDescription("Failed to read headers");
+            stream.cancel(status);
+            close(status, new Metadata());
           }
         }
       });
@@ -437,13 +439,28 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT>
               message.close();
             }
           } catch (Throwable t) {
-            stream.cancel(Status.CANCELLED.withCause(t).withDescription("Failed to read message."));
-            return;
+            Status status =
+                Status.CANCELLED.withCause(t).withDescription("Failed to read message.");
+            stream.cancel(status);
+            close(status, new Metadata());
           }
         }
       });
     }
 
+    /**
+     * Must be called from application thread.
+     */
+    private void close(Status status, Metadata trailers) {
+      closed = true;
+      cancelListenersShouldBeRemoved = true;
+      try {
+        observer.onClose(status, trailers);
+      } finally {
+        removeContextListenerAndCancelDeadlineFuture();
+      }
+    }
+
     @Override
     public void closed(Status status, Metadata trailers) {
       Deadline deadline = effectiveDeadline();
@@ -462,13 +479,11 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT>
       callExecutor.execute(new ContextRunnable(context) {
         @Override
         public final void runInContext() {
-          try {
-            closed = true;
-            cancelListenersShouldBeRemoved = true;
-            observer.onClose(savedStatus, savedTrailers);
-          } finally {
-            removeContextListenerAndCancelDeadlineFuture();
+          if (closed) {
+            // We intentionally don't keep the status or metadata from the server.
+            return;
           }
+          close(savedStatus, savedTrailers);
         }
       });
     }
@@ -478,7 +493,14 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT>
       callExecutor.execute(new ContextRunnable(context) {
         @Override
         public final void runInContext() {
-          observer.onReady();
+          try {
+            observer.onReady();
+          } catch (Throwable t) {
+            Status status =
+                Status.CANCELLED.withCause(t).withDescription("Failed to call onReady.");
+            stream.cancel(status);
+            close(status, new Metadata());
+          }
         }
       });
     }
diff --git a/core/src/test/java/io/grpc/internal/ClientCallImplTest.java b/core/src/test/java/io/grpc/internal/ClientCallImplTest.java
index 8a73fb0b7..ed0773082 100644
--- a/core/src/test/java/io/grpc/internal/ClientCallImplTest.java
+++ b/core/src/test/java/io/grpc/internal/ClientCallImplTest.java
@@ -43,6 +43,7 @@ import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
@@ -75,6 +76,7 @@ import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
+import org.mockito.Matchers;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
@@ -82,8 +84,11 @@ import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Set;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -146,6 +151,103 @@ public class ClientCallImplTest {
     Context.ROOT.attach();
   }
 
+  @Test
+  public void exceptionInOnMessageTakesPrecedenceOverServer() {
+    DelayedExecutor executor = new DelayedExecutor();
+    ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
+        method,
+        executor,
+        CallOptions.DEFAULT,
+        provider,
+        deadlineCancellationExecutor);
+    call.start(callListener, new Metadata());
+    verify(stream).start(listenerArgumentCaptor.capture());
+    final ClientStreamListener streamListener = listenerArgumentCaptor.getValue();
+    streamListener.headersRead(new Metadata());
+
+    RuntimeException failure = new RuntimeException("bad");
+    doThrow(failure).when(callListener).onMessage(any(Void.class));
+
+    /*
+     * In unary calls, the server closes the call right after responding, so the onClose call is
+     * queued to run.  When messageRead is called, an exception will occur and attempt to cancel the
+     * stream.  However, since the server closed it "first" the second exception is lost leading to
+     * the call being counted as successful.
+     */
+    streamListener.messageRead(new ByteArrayInputStream(new byte[]{}));
+    streamListener.closed(Status.OK, new Metadata());
+    executor.release();
+
+    verify(callListener).onClose(statusArgumentCaptor.capture(), Matchers.isA(Metadata.class));
+    assertThat(statusArgumentCaptor.getValue().getCode()).isEqualTo(Status.Code.CANCELLED);
+    assertThat(statusArgumentCaptor.getValue().getCause()).isSameAs(failure);
+    verify(stream).cancel(statusArgumentCaptor.getValue());
+  }
+
+  @Test
+  public void exceptionInOnHeadersTakesPrecedenceOverServer() {
+    DelayedExecutor executor = new DelayedExecutor();
+    ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
+        method,
+        executor,
+        CallOptions.DEFAULT,
+        provider,
+        deadlineCancellationExecutor);
+    call.start(callListener, new Metadata());
+    verify(stream).start(listenerArgumentCaptor.capture());
+    final ClientStreamListener streamListener = listenerArgumentCaptor.getValue();
+
+    RuntimeException failure = new RuntimeException("bad");
+    doThrow(failure).when(callListener).onHeaders(any(Metadata.class));
+
+    /*
+     * In unary calls, the server closes the call right after responding, so the onClose call is
+     * queued to run.  When headersRead is called, an exception will occur and attempt to cancel the
+     * stream.  However, since the server closed it "first" the second exception is lost leading to
+     * the call being counted as successful.
+     */
+    streamListener.headersRead(new Metadata());
+    streamListener.closed(Status.OK, new Metadata());
+    executor.release();
+
+    verify(callListener).onClose(statusArgumentCaptor.capture(), Matchers.isA(Metadata.class));
+    assertThat(statusArgumentCaptor.getValue().getCode()).isEqualTo(Status.Code.CANCELLED);
+    assertThat(statusArgumentCaptor.getValue().getCause()).isSameAs(failure);
+    verify(stream).cancel(statusArgumentCaptor.getValue());
+  }
+
+  @Test
+  public void exceptionInOnReadyTakesPrecedenceOverServer() {
+    DelayedExecutor executor = new DelayedExecutor();
+    ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
+        method,
+        executor,
+        CallOptions.DEFAULT,
+        provider,
+        deadlineCancellationExecutor);
+    call.start(callListener, new Metadata());
+    verify(stream).start(listenerArgumentCaptor.capture());
+    final ClientStreamListener streamListener = listenerArgumentCaptor.getValue();
+
+    RuntimeException failure = new RuntimeException("bad");
+    doThrow(failure).when(callListener).onReady();
+
+    /*
+     * In unary calls, the server closes the call right after responding, so the onClose call is
+     * queued to run.  When onReady is called, an exception will occur and attempt to cancel the
+     * stream.  However, since the server closed it "first" the second exception is lost leading to
+     * the call being counted as successful.
+     */
+    streamListener.onReady();
+    streamListener.closed(Status.OK, new Metadata());
+    executor.release();
+
+    verify(callListener).onClose(statusArgumentCaptor.capture(), Matchers.isA(Metadata.class));
+    assertThat(statusArgumentCaptor.getValue().getCode()).isEqualTo(Status.Code.CANCELLED);
+    assertThat(statusArgumentCaptor.getValue().getCause()).isSameAs(failure);
+    verify(stream).cancel(statusArgumentCaptor.getValue());
+  }
+
   @Test
   public void advertisedEncodingsAreSent() {
     ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
@@ -668,5 +770,19 @@ public class ClientCallImplTest {
     assertTrue("timeout: " + timeout + " ns", timeout >= from);
   }
 
+  private static final class DelayedExecutor implements Executor {
+    private final BlockingQueue<Runnable> commands = new LinkedBlockingQueue<Runnable>();
+
+    @Override
+    public void execute(Runnable command) {
+      commands.add(command);
+    }
+
+    void release() {
+      while (!commands.isEmpty()) {
+        commands.poll().run();
+      }
+    }
+  }
 }