Fix hang on deserialization exception

src/StreamJsonRpc.Tests/JsonRpcJsonHeadersTests.cs

@@ -189,8 +189,28 @@ public async Task CanPassExceptionFromServer()
 
     protected override void InitializeFormattersAndHandlers()
     {
-        this.clientMessageFormatter = new JsonMessageFormatter { JsonSerializer = { Converters = { new UnserializableTypeConverter() } } };
-        this.serverMessageFormatter = new JsonMessageFormatter { JsonSerializer = { Converters = { new UnserializableTypeConverter() } } };
+        this.clientMessageFormatter = new JsonMessageFormatter
+        {
+            JsonSerializer =
+            {
+                Converters =
+                {
+                    new UnserializableTypeConverter(),
+                    new TypeThrowsWhenDeserializedConverter(),
+                },
+            },
+        };
+        this.serverMessageFormatter = new JsonMessageFormatter
+        {
+            JsonSerializer =
+            {
+                Converters =
+                {
+                    new UnserializableTypeConverter(),
+                    new TypeThrowsWhenDeserializedConverter(),
+                },
+            },
+        };
 
         this.serverMessageHandler = new HeaderDelimitedMessageHandler(this.serverStream, this.serverStream, this.serverMessageFormatter);
         this.clientMessageHandler = new HeaderDelimitedMessageHandler(this.clientStream, this.clientStream, this.clientMessageFormatter);
@@ -221,6 +241,20 @@ public override void WriteJson(JsonWriter writer, object value, JsonSerializer s
         }
     }
 
+    private class TypeThrowsWhenDeserializedConverter : JsonConverter<TypeThrowsWhenDeserialized>
+    {
+        public override TypeThrowsWhenDeserialized ReadJson(JsonReader reader, Type objectType, TypeThrowsWhenDeserialized existingValue, bool hasExistingValue, JsonSerializer serializer)
+        {
+            throw new Exception("This exception is meant to be thrown.");
+        }
+
+        public override void WriteJson(JsonWriter writer, TypeThrowsWhenDeserialized value, JsonSerializer serializer)
+        {
+            writer.WriteStartObject();
+            writer.WriteEndObject();
+        }
+    }
+
     private class StringBase64Converter : JsonConverter
     {
         public override bool CanConvert(Type objectType) => objectType == typeof(string);

src/StreamJsonRpc.Tests/JsonRpcMessagePackLengthTests.cs

@@ -62,7 +62,7 @@ protected override void InitializeFormattersAndHandlers()
 
         var options = MessagePackSerializerOptions.Standard
             .WithResolver(CompositeResolver.Create(
-                new IMessagePackFormatter[] { new UnserializableTypeFormatter() },
+                new IMessagePackFormatter[] { new UnserializableTypeFormatter(), new TypeThrowsWhenDeserializedFormatter() },
                 new IFormatterResolver[] { StandardResolverAllowPrivate.Instance }));
         ((MessagePackFormatter)this.serverMessageFormatter).SetMessagePackSerializerOptions(options);
         ((MessagePackFormatter)this.clientMessageFormatter).SetMessagePackSerializerOptions(options);
@@ -83,4 +83,17 @@ public void Serialize(ref MessagePackWriter writer, CustomSerializedType value,
             writer.Write(value?.Value);
         }
     }
+
+    private class TypeThrowsWhenDeserializedFormatter : IMessagePackFormatter<TypeThrowsWhenDeserialized>
+    {
+        public TypeThrowsWhenDeserialized Deserialize(ref MessagePackReader reader, MessagePackSerializerOptions options)
+        {
+            throw new Exception("This exception is meant to be thrown.");
+        }
+
+        public void Serialize(ref MessagePackWriter writer, TypeThrowsWhenDeserialized value, MessagePackSerializerOptions options)
+        {
+            writer.WriteArrayHeader(0);
+        }
+    }
 }

src/StreamJsonRpc.Tests/JsonRpcTests.cs

@@ -1629,6 +1629,12 @@ public async Task FormatterFatalException()
         await Assert.ThrowsAsync<ConnectionLostException>(() => this.clientRpc.InvokeAsync<string>(nameof(Server.AsyncMethod), "Fail"));
     }
 
+    [Fact]
+    public async Task ReturnTypeThrowsOnDeserialization()
+    {
+        await Assert.ThrowsAnyAsync<Exception>(() => this.clientRpc.InvokeWithCancellationAsync<TypeThrowsWhenDeserialized>(nameof(Server.GetTypeThrowsWhenDeserialized), cancellationToken: this.TimeoutToken)).WithCancellation(this.TimeoutToken);
+    }
+
     protected override void Dispose(bool disposing)
     {
         if (disposing)
@@ -1861,6 +1867,8 @@ public int InstanceMethodWithSingleObjectParameterAndCancellationToken(XAndYFiel
             return fields.x + fields.y;
         }
 
+        public TypeThrowsWhenDeserialized GetTypeThrowsWhenDeserialized() => new TypeThrowsWhenDeserialized();
+
         public int? MethodReturnsNullableInt(int a) => a > 0 ? (int?)a : null;
 
         public int MethodAcceptsNullableArgs(int? a, int? b) => (a.HasValue ? 1 : 0) + (b.HasValue ? 1 : 0);
@@ -2165,6 +2173,10 @@ public class CustomSerializedType
         public string? Value { get; set; }
     }
 
+    public class TypeThrowsWhenDeserialized
+    {
+    }
+
     [DataContract]
     public class XAndYFields
     {

src/StreamJsonRpc/JsonRpc.cs

@@ -2163,6 +2163,7 @@ private async Task ReadAndHandleRequestsAsync()
         private async Task HandleRpcAsync(JsonRpcMessage rpc)
         {
             Requires.NotNull(rpc, nameof(rpc));
+            OutstandingCallData? data = null;
             try
             {
                 if (rpc is JsonRpcRequest request)
@@ -2221,7 +2222,6 @@ private async Task HandleRpcAsync(JsonRpcMessage rpc)
                     JsonRpcResult? result = resultOrError as JsonRpcResult;
                     JsonRpcError? error = resultOrError as JsonRpcError;
 
-                    OutstandingCallData? data = null;
                     lock (this.dispatcherMapLock)
                     {
                         if (this.resultDispatcherMap.TryGetValue(resultOrError.RequestId, out data))
@@ -2260,6 +2260,7 @@ private async Task HandleRpcAsync(JsonRpcMessage rpc)
                         // Complete the caller's request with the response asynchronously so it doesn't delay handling of other JsonRpc messages.
                         await TaskScheduler.Default.SwitchTo(alwaysYield: true);
                         data.CompletionHandler(rpc);
+                        data = null; // avoid invoking again if we throw later
                     }
                 }
                 else
@@ -2279,6 +2280,9 @@ private async Task HandleRpcAsync(JsonRpcMessage rpc)
 
                 // Fatal error. Raise disconnected event.
                 this.OnJsonRpcDisconnected(eventArgs);
+
+                // If we extracted this callback from the collection already, take care to complete it to avoid hanging our client.
+                data?.CompletionHandler(null);
             }
         }
 

src/StreamJsonRpc/MessagePackFormatter.cs

@@ -11,6 +11,7 @@ namespace StreamJsonRpc
     using System.IO;
     using System.IO.Pipelines;
     using System.Reflection;
+    using System.Runtime.ExceptionServices;
     using System.Runtime.Serialization;
     using MessagePack;
     using MessagePack.Formatters;
@@ -1330,12 +1331,6 @@ public override bool TryGetArgumentByNameOrIndex(string? name, int position, Typ
                     value = MessagePackSerializer.Deserialize(typeHint ?? typeof(object), ref reader, this.formatter.userDataSerializationOptions);
                     return true;
                 }
-                catch (NotSupportedException)
-                {
-                    // This block can be removed after https://github.com/neuecc/MessagePack-CSharp/pull/633 is applied.
-                    value = null;
-                    return false;
-                }
                 catch (MessagePackSerializationException)
                 {
                     value = null;
@@ -1354,6 +1349,8 @@ private class JsonRpcResult : Protocol.JsonRpcResult, IJsonRpcMessageBufferManag
         {
             private readonly MessagePackSerializerOptions serializerOptions;
 
+            private Exception? resultDeserializationException;
+
             internal JsonRpcResult(MessagePackSerializerOptions serializerOptions)
             {
                 this.serializerOptions = serializerOptions ?? throw new ArgumentNullException(nameof(serializerOptions));
@@ -1372,6 +1369,11 @@ void IJsonRpcMessageBufferManager.DeserializationComplete(JsonRpcMessage message
 
             public override T GetResult<T>()
             {
+                if (this.resultDeserializationException != null)
+                {
+                    ExceptionDispatchInfo.Capture(this.resultDeserializationException).Throw();
+                }
+
                 return this.MsgPackResult.IsEmpty
                     ? (T)this.Result!
                     : MessagePackSerializer.Deserialize<T>(this.MsgPackResult, this.serializerOptions);
@@ -1382,8 +1384,16 @@ protected internal override void SetExpectedResultType(Type resultType)
                 Verify.Operation(!this.MsgPackResult.IsEmpty, "Result is no longer available or has already been deserialized.");
 
                 var reader = new MessagePackReader(this.MsgPackResult);
-                this.Result = MessagePackSerializer.Deserialize(resultType, ref reader, this.serializerOptions);
-                this.MsgPackResult = default;
+                try
+                {
+                    this.Result = MessagePackSerializer.Deserialize(resultType, ref reader, this.serializerOptions);
+                    this.MsgPackResult = default;
+                }
+                catch (MessagePackSerializationException ex)
+                {
+                    // This was a best effort anyway. We'll throw again later at a more convenient time for JsonRpc.
+                    this.resultDeserializationException = ex;
+                }
             }
         }