Beef up exception handling in gRPC code. (#360)

* Beef up exception handling in gRPC code.

* Verify default stacktrace isn't used in exceptions

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 2.4.1
+
+* Plumb stacktraces through request / response stream error handlers.
+* Catch and forward any errors decoding the response.
+
 ## 2.4.0
 
 * Add the ability to bypass CORS preflight requests.

lib/src/client/call.dart

@@ -266,8 +266,8 @@ class ClientCall<Q, R> implements Response {
   }
 
   /// Emit an error response to the user, and tear down this call.
-  void _responseError(GrpcError error) {
+  void _responseError(GrpcError error, [StackTrace stackTrace]) {
-    _responses.addError(error);
+    _responses.addError(error, stackTrace);
     _timeoutTimer?.cancel();
     _requestSubscription?.cancel();
     _responseSubscription.cancel();
@@ -287,8 +287,12 @@ class ClientCall<Q, R> implements Response {
         _responseError(GrpcError.unimplemented('Received data after trailers'));
         return;
       }
+      try {
         _responses.add(_method.responseDeserializer(data.data));
         _hasReceivedResponses = true;
+      } catch (e, s) {
+        _responseError(GrpcError.dataLoss('Error parsing response'), s);
+      }
     } else if (data is GrpcMetadata) {
       if (!_headers.isCompleted) {
         // TODO(jakobr): Parse, and extract common headers.
@@ -319,12 +323,12 @@ class ClientCall<Q, R> implements Response {
 
   /// Handler for response errors. Forward the error to the [_responses] stream,
   /// wrapped if necessary.
-  void _onResponseError(error) {
+  void _onResponseError(error, StackTrace stackTrace) {
     if (error is GrpcError) {
-      _responseError(error);
+      _responseError(error, stackTrace);
       return;
     }
-    _responseError(GrpcError.unknown(error.toString()));
+    _responseError(GrpcError.unknown(error.toString()), stackTrace);
   }
 
   /// Handles closure of the response stream. Verifies that server has sent
@@ -362,12 +366,12 @@ class ClientCall<Q, R> implements Response {
   /// Error handler for the requests stream. Something went wrong while trying
   /// to send the request to the server. Abort the request, and forward the
   /// error to the user code on the [_responses] stream.
-  void _onRequestError(error, [StackTrace stackTrace]) {
+  void _onRequestError(error, StackTrace stackTrace) {
     if (error is! GrpcError) {
       error = GrpcError.unknown(error.toString());
     }
 
-    _responses.addError(error);
+    _responses.addError(error, stackTrace);
     _timeoutTimer?.cancel();
     _responses.close();
     _requestSubscription?.cancel();

lib/src/client/transport/transport.dart

@@ -19,7 +19,7 @@ import '../../shared/message.dart';
 
 typedef void SocketClosedHandler();
 typedef void ActiveStateHandler(bool isActive);
-typedef void ErrorHandler(error);
+typedef void ErrorHandler(error, StackTrace stackTrace);
 
 abstract class GrpcTransportStream {
   Stream<GrpcMessage> get incomingMessages;

lib/src/client/transport/xhr_transport.dart

@@ -68,8 +68,10 @@ class XhrTransportStream implements GrpcTransportStream {
           break;
         case HttpRequest.DONE:
           if (_request.status != 200) {
-            _onError(GrpcError.unavailable(
+            _onError(
-                'XhrConnection status ${_request.status}'));
+                GrpcError.unavailable(
+                    'XhrConnection status ${_request.status}'),
+                StackTrace.current);
           } else {
             _close();
           }
@@ -81,7 +83,8 @@ class XhrTransportStream implements GrpcTransportStream {
       if (_incomingMessages.isClosed) {
         return;
       }
-      _onError(GrpcError.unavailable('XhrConnection connection-error'));
+      _onError(GrpcError.unavailable('XhrConnection connection-error'),
+          StackTrace.current);
       terminate();
     });
 
@@ -113,21 +116,24 @@ class XhrTransportStream implements GrpcTransportStream {
   _onHeadersReceived() {
     final contentType = _request.getResponseHeader(_contentTypeKey);
     if (_request.status != 200) {
-      _onError(
+      _onError(GrpcError.unavailable('XhrConnection status ${_request.status}'),
-          GrpcError.unavailable('XhrConnection status ${_request.status}'));
+          StackTrace.current);
       return;
     }
     if (contentType == null) {
-      _onError(GrpcError.unavailable('XhrConnection missing Content-Type'));
+      _onError(GrpcError.unavailable('XhrConnection missing Content-Type'),
+          StackTrace.current);
       return;
     }
     if (!_checkContentType(contentType)) {
       _onError(
-          GrpcError.unavailable('XhrConnection bad Content-Type $contentType'));
+          GrpcError.unavailable('XhrConnection bad Content-Type $contentType'),
+          StackTrace.current);
       return;
     }
     if (_request.response == null) {
-      _onError(GrpcError.unavailable('XhrConnection request null response'));
+      _onError(GrpcError.unavailable('XhrConnection request null response'),
+          StackTrace.current);
       return;
     }
 

pubspec.yaml

@@ -1,7 +1,7 @@
 name: grpc
 description: Dart implementation of gRPC, a high performance, open-source universal RPC framework.
 
-version: 2.4.0
+version: 2.4.1
 
 author: Dart Team <misc@dartlang.org>
 homepage: https://github.com/dart-lang/grpc-dart

test/client_tests/client_test.dart

@@ -290,6 +290,27 @@ void main() {
     );
   });
 
+  test('Call throws if unable to decode response', () async {
+    const responseValue = 19;
+
+    void handleRequest(StreamMessage message) {
+      harness
+        ..sendResponseHeader()
+        ..sendResponseValue(responseValue)
+        ..sendResponseTrailer();
+    }
+
+    harness.client = TestClient(harness.channel, decode: (bytes) {
+      throw "error decoding";
+    });
+
+    await harness.runFailureTest(
+      clientCall: harness.client.unary(dummyValue),
+      expectedException: GrpcError.dataLoss('Error parsing response'),
+      serverHandlers: [handleRequest],
+    );
+  });
+
   test('Call forwards known response stream errors', () async {
     final expectedException = GrpcError.dataLoss('Oops!');
 

test/client_tests/client_xhr_transport_test.dart

@@ -69,7 +69,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     connection.makeRequest('path', Duration(seconds: 10), metadata,
-        (error) => fail(error.toString()));
+        (error, _) => fail(error.toString()));
 
     verify(connection.latestRequest
         .setRequestHeader('Content-Type', 'application/grpc-web+proto'));
@@ -88,7 +88,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     connection.makeRequest('path', Duration(seconds: 10), metadata,
-        (error) => fail(error.toString()),
+        (error, _) => fail(error.toString()),
         callOptions: WebCallOptions(bypassCorsPreflight: true));
 
     expect(metadata, isEmpty);
@@ -110,7 +110,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     connection.makeRequest('/path', Duration(seconds: 10), metadata,
-        (error) => fail(error.toString()));
+        (error, _) => fail(error.toString()));
 
     expect(metadata, {
       'header_1': 'value_1',
@@ -127,7 +127,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     connection.makeRequest('/path', Duration(seconds: 10), metadata,
-        (error) => fail(error.toString()));
+        (error, _) => fail(error.toString()));
     expect(metadata, {
       'header_1': 'value_1',
       'CONTENT-TYPE': 'application/json+protobuf',
@@ -138,7 +138,7 @@ void main() {
       'content-type': 'application/json+protobuf'
     };
     connection.makeRequest('/path', Duration(seconds: 10), lowerMetadata,
-        (error) => fail(error.toString()));
+        (error, _) => fail(error.toString()));
     expect(lowerMetadata, {
       'header_1': 'value_1',
       'content-type': 'application/json+protobuf',
@@ -151,7 +151,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     connection.makeRequest('path', Duration(seconds: 10), metadata,
-        (error) => fail(error.toString()),
+        (error, _) => fail(error.toString()),
         callOptions: WebCallOptions(withCredentials: true));
 
     expect(metadata, {
@@ -182,7 +182,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     final stream = connection.makeRequest('path', Duration(seconds: 10),
-        metadata, (error) => fail(error.toString()));
+        metadata, (error, _) => fail(error.toString()));
 
     final data = List.filled(10, 0);
     stream.outgoingMessages.add(data);
@@ -202,7 +202,7 @@ void main() {
     final transport = MockXhrClientConnection();
 
     final stream = transport.makeRequest('test_path', Duration(seconds: 10),
-        metadata, (error) => fail(error.toString()));
+        metadata, (error, _) => fail(error.toString()));
 
     stream.incomingMessages.listen((message) {
       expect(message, TypeMatcher<GrpcMetadata>());
@@ -223,7 +223,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     final stream = connection.makeRequest('test_path', Duration(seconds: 10),
-        {}, (error) => fail(error.toString()));
+        {}, (error, _) => fail(error.toString()));
 
     final encodedTrailers = frame(trailers.entries
         .map((e) => '${e.key}:${e.value}')
@@ -254,7 +254,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     final stream = connection.makeRequest('test_path', Duration(seconds: 10),
-        {}, (error) => fail(error.toString()));
+        {}, (error, _) => fail(error.toString()));
 
     final encoded = frame(''.codeUnits);
     encoded[0] = 0x80; // Mark this frame as trailers.
@@ -285,7 +285,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     final stream = connection.makeRequest('test_path', Duration(seconds: 10),
-        metadata, (error) => fail(error.toString()));
+        metadata, (error, _) => fail(error.toString()));
     final data = List<int>.filled(10, 224);
     final encoded = frame(data);
     final encodedString = String.fromCharCodes(encoded);
@@ -315,7 +315,7 @@ void main() {
     final connection = MockXhrClientConnection();
 
     final stream = connection.makeRequest('test_path', Duration(seconds: 10),
-        metadata, (error) => fail(error.toString()));
+        metadata, (error, _) => fail(error.toString()));
 
     final data = <List<int>>[
       List<int>.filled(10, 224),

test/src/client_utils.dart

@@ -69,17 +69,24 @@ class FakeChannel extends ClientChannel {
 typedef ServerMessageHandler = void Function(StreamMessage message);
 
 class TestClient extends Client {
-  static final _$unary =
+  ClientMethod<int, int> _$unary;
-      ClientMethod<int, int>('/Test/Unary', mockEncode, mockDecode);
+  ClientMethod<int, int> _$clientStreaming;
-  static final _$clientStreaming =
+  ClientMethod<int, int> _$serverStreaming;
-      ClientMethod<int, int>('/Test/ClientStreaming', mockEncode, mockDecode);
+  ClientMethod<int, int> _$bidirectional;
-  static final _$serverStreaming =
-      ClientMethod<int, int>('/Test/ServerStreaming', mockEncode, mockDecode);
-  static final _$bidirectional =
-      ClientMethod<int, int>('/Test/Bidirectional', mockEncode, mockDecode);
 
-  TestClient(ClientChannel channel, {CallOptions options})
+  final int Function(List<int> value) decode;
-      : super(channel, options: options);
+
+  TestClient(ClientChannel channel,
+      {CallOptions options, this.decode: mockDecode})
+      : super(channel, options: options) {
+    _$unary = ClientMethod<int, int>('/Test/Unary', mockEncode, decode);
+    _$clientStreaming =
+        ClientMethod<int, int>('/Test/ClientStreaming', mockEncode, decode);
+    _$serverStreaming =
+        ClientMethod<int, int>('/Test/ServerStreaming', mockEncode, decode);
+    _$bidirectional =
+        ClientMethod<int, int>('/Test/Bidirectional', mockEncode, decode);
+  }
 
   ResponseFuture<int> unary(int request, {CallOptions options}) {
     final call =
@@ -201,8 +208,9 @@ class ClientHarness {
     try {
       await future;
       fail('Did not throw');
-    } catch (e) {
+    } catch (e, st) {
       expect(e, exception);
+      expect(st, isNot(equals(StackTrace.current)));
     }
   }