Add JsonRpc outbound request timeout (#1471)

AArnott · Copilot · web-flow · commit 67708eb9457e · 2026-06-23T10:42:25.000-07:00
* Add outbound request timeout support

Add a JsonRpc-wide OutboundRequestTimeout that applies to response-expected outbound calls, turns timeout-origin failures into TimeoutException, preserves caller cancellation tokens, and exercises the behavior across direct, proxy, and cancellation-strategy tests.

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;

* Address PR feedback on outbound timeouts

Translate timeout-driven send-path cancellations into TimeoutException, align most timeout tests with ExpectedTimeout, and keep the custom cancellation-strategy timeout at 100ms because the larger shared timeout hangs on net472 in that scenario.

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;

* Stabilize outbound-timeout tests on Windows CI

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;

* Address PR timeout race and atomicity feedback

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;

---------

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;
diff --git a/src/StreamJsonRpc/JsonRpc.cs b/src/StreamJsonRpc/JsonRpc.cs
@@ -126,6 +126,13 @@ public class JsonRpc : IDisposableObservable, IJsonRpcFormatterCallbacks, IJsonR
     [DebuggerBrowsable(DebuggerBrowsableState.Never)]
     private bool cancelLocallyInvokedMethodsWhenConnectionIsClosed;
 
+    /// <summary>
+    /// Backing field for the <see cref="OutboundRequestTimeout"/> property.
+    /// Stores timeout ticks, with 0 representing <see langword="null"/>.
+    /// </summary>
+    [DebuggerBrowsable(DebuggerBrowsableState.Never)]
+    private long outboundRequestTimeoutTicks;
+
     /// <summary>
     /// Backing field for the <see cref="SynchronizationContext"/> property.
     /// </summary>
@@ -590,6 +597,47 @@ public bool CancelLocallyInvokedMethodsWhenConnectionIsClosed
         }
     }
 
+    /// <summary>
+    /// Gets or sets the timeout to apply to each outbound invocation that expects a response.
+    /// </summary>
+    /// <value>
+    /// The timeout, or <see langword="null"/> to disable this behavior.
+    /// The default value is <see langword="null"/>.
+    /// </value>
+    /// <remarks>
+    /// <para>
+    /// This timeout is applied to outbound method invocations (such as <see cref="InvokeWithCancellationAsync(string, IReadOnlyList{object?}?, CancellationToken)"/> and proxy method calls),
+    /// but not notifications.
+    /// </para>
+    /// <para>
+    /// When a timeout triggers, a notification is sent to the server to request cancellation
+    /// and the local invocation is immediately canceled allowing the caller to handle the timeout
+    /// as a <see cref="TimeoutException"/>.
+    /// </para>
+    /// </remarks>
+    /// <exception cref="ArgumentOutOfRangeException">Thrown if a non-positive timeout value is assigned.</exception>
+    public TimeSpan? OutboundRequestTimeout
+    {
+        get
+        {
+            long timeoutTicks = Interlocked.Read(ref this.outboundRequestTimeoutTicks);
+            return timeoutTicks == 0 ? null : TimeSpan.FromTicks(timeoutTicks);
+        }
+
+        set
+        {
+            if (value is TimeSpan timeout)
+            {
+                Requires.Range(timeout > TimeSpan.Zero, nameof(value), Resources.PositiveTimeSpanRequired);
+                Interlocked.Exchange(ref this.outboundRequestTimeoutTicks, timeout.Ticks);
+            }
+            else
+            {
+                Interlocked.Exchange(ref this.outboundRequestTimeoutTicks, 0);
+            }
+        }
+    }
+
     /// <summary>
     /// Gets or sets the <see cref="System.Diagnostics.TraceSource"/> used to trace JSON-RPC messages and events.
     /// </summary>
@@ -2120,9 +2168,19 @@ private JsonRpcError CreateCancellationResponse(JsonRpcRequest request)
         Requires.NotNull(request, nameof(request));
         Assumes.NotNull(request.Method);
 
+        CancellationToken effectiveOutboundCancellationToken = cancellationToken;
+        CancellationTokenSource? timeoutCancellationSource = null;
+        if (request.IsResponseExpected && this.OutboundRequestTimeout is TimeSpan outboundRequestTimeout)
+        {
+            timeoutCancellationSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            timeoutCancellationSource.CancelAfter(outboundRequestTimeout);
+            effectiveOutboundCancellationToken = timeoutCancellationSource.Token;
+        }
+
         try
         {
-            using (CancellationTokenExtensions.CombinedCancellationToken cts = this.DisconnectedToken.CombineWith(cancellationToken))
+            using (timeoutCancellationSource)
+            using (CancellationTokenExtensions.CombinedCancellationToken cts = this.DisconnectedToken.CombineWith(effectiveOutboundCancellationToken))
             {
                 if (!request.IsResponseExpected)
                 {
@@ -2161,10 +2219,10 @@ private JsonRpcError CreateCancellationResponse(JsonRpcRequest request)
                                 JsonRpcEventSource.Instance.ReceivedNoResponse(request.RequestId.NumberIfPossibleForEvent);
                             }
 
-                            if (cancellationToken.IsCancellationRequested)
+                            if (effectiveOutboundCancellationToken.IsCancellationRequested)
                             {
                                 // Consider lost connection to be result of task canceled and set state to canceled.
-                                tcs.TrySetCanceled(cancellationToken);
+                                tcs.TrySetCanceled(effectiveOutboundCancellationToken);
                             }
                             else
                             {
@@ -2180,7 +2238,7 @@ private JsonRpcError CreateCancellationResponse(JsonRpcRequest request)
 
                             if (error.Error?.Code == JsonRpcErrorCode.RequestCanceled)
                             {
-                                tcs.TrySetCanceled(cancellationToken.IsCancellationRequested ? cancellationToken : CancellationToken.None);
+                                tcs.TrySetCanceled(effectiveOutboundCancellationToken.IsCancellationRequested ? effectiveOutboundCancellationToken : CancellationToken.None);
                             }
                             else
                             {
@@ -2241,22 +2299,49 @@ private JsonRpcError CreateCancellationResponse(JsonRpcRequest request)
                 // Arrange for sending a cancellation message if canceled while we're waiting for a response.
                 try
                 {
-                    using (cancellationToken.Register(this.cancelPendingOutboundRequestAction!, request.RequestId, useSynchronizationContext: false))
+                    using (effectiveOutboundCancellationToken.Register(this.cancelPendingOutboundRequestAction!, request.RequestId, useSynchronizationContext: false))
                     {
-                        // This task will be completed when the Response object comes back from the other end of the pipe
-                        return await tcs.Task.ConfigureAwait(false);
+                        // This task will be completed when the Response object comes back from the other end of the pipe.
+                        try
+                        {
+                            if (timeoutCancellationSource is null)
+                            {
+                                return await tcs.Task.ConfigureAwait(false);
+                            }
+
+                            Task completedTask = await Task.WhenAny(tcs.Task, Task.Delay(Timeout.Infinite, effectiveOutboundCancellationToken)).ConfigureAwait(false);
+                            if (completedTask == tcs.Task)
+                            {
+                                return await tcs.Task.ConfigureAwait(false);
+                            }
+
+                            effectiveOutboundCancellationToken.ThrowIfCancellationRequested();
+                            throw Assumes.NotReachable();
+                        }
+                        catch (OperationCanceledException ex) when (timeoutCancellationSource?.IsCancellationRequested is true && !cancellationToken.IsCancellationRequested)
+                        {
+                            throw new TimeoutException(Resources.FormatOutboundInvocationTimedOut(nameof(this.OutboundRequestTimeout)), ex);
+                        }
                     }
                 }
                 finally
                 {
-                    if (cancellationToken.IsCancellationRequested)
+                    if (effectiveOutboundCancellationToken.IsCancellationRequested)
                     {
                         this.CancellationStrategy?.OutboundRequestEnded(request.RequestId);
                     }
                 }
             }
         }
-        catch (OperationCanceledException ex) when (this.DisconnectedToken.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
+        catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested && ex.CancellationToken != cancellationToken)
+        {
+            throw new OperationCanceledException(ex.Message, ex, cancellationToken);
+        }
+        catch (OperationCanceledException ex) when (timeoutCancellationSource?.IsCancellationRequested is true && !cancellationToken.IsCancellationRequested)
+        {
+            throw new TimeoutException(Resources.FormatOutboundInvocationTimedOut(nameof(this.OutboundRequestTimeout)), ex);
+        }
+        catch (OperationCanceledException ex) when (this.DisconnectedToken.IsCancellationRequested && !effectiveOutboundCancellationToken.IsCancellationRequested)
         {
             throw new ConnectionLostException(Resources.ConnectionDropped, ex);
         }
diff --git a/src/StreamJsonRpc/Resources.resx b/src/StreamJsonRpc/Resources.resx
@@ -256,6 +256,9 @@
   <data name="PositiveIntegerRequired" xml:space="preserve">
     <value>A positive integer is required.</value>
   </data>
+  <data name="PositiveTimeSpanRequired" xml:space="preserve">
+    <value>A positive TimeSpan is required.</value>
+  </data>
   <data name="ReachedEndOfStream" xml:space="preserve">
     <value>Reached end of stream.</value>
   </data>
@@ -321,6 +324,9 @@
   <data name="TaskWasCancelled" xml:space="preserve">
     <value>The task was cancelled.</value>
   </data>
+  <data name="OutboundInvocationTimedOut" xml:space="preserve">
+    <value>The outbound JSON-RPC invocation was canceled automatically after reaching the timeout configured by {0}. The server may still be processing the request.</value>
+  </data>
   <data name="TextEncoderNotApplicable" xml:space="preserve">
     <value>Text encoding is not supported because the formatter "{0}" does not implement "{1}".</value>
     <comment>{0} and {1} are CLR type names.</comment>
diff --git a/test/StreamJsonRpc.Tests/CustomCancellationStrategyTests.cs b/test/StreamJsonRpc.Tests/CustomCancellationStrategyTests.cs
@@ -91,6 +91,18 @@ public async Task CancelRequest_ServerMethodThrows()
         await this.mockStrategy.OutboundRequestEndedInvoked.WaitAsync(this.TimeoutToken);
     }
 
+    [Fact]
+    public async Task CancelRequest_WhenOutboundRequestTimesOut()
+    {
+        this.clientRpc.OutboundRequestTimeout = TimeSpan.FromSeconds(1);
+        Task invokeTask = this.clientRpc.InvokeWithCancellationAsync(nameof(Server.NoticeCancellationAsync), new object?[] { false }, cancellationToken: CancellationToken.None);
+        await this.server.MethodEntered.WaitAsync(this.TimeoutToken);
+
+        TimeoutException ex = await Assert.ThrowsAsync<TimeoutException>(() => invokeTask);
+        Assert.Contains(nameof(JsonRpc.OutboundRequestTimeout), ex.Message, StringComparison.Ordinal);
+        Assert.True(this.mockStrategy.OutboundRequestEndedCalled);
+    }
+
     [Fact]
     public async Task UncanceledRequest_GetsNoClientSideInvocations()
     {
@@ -114,11 +126,9 @@ public async Task OutboundCancellationStartAndRequestFinishOverlap()
         var completingTask = await Task.WhenAny(invokeTask, this.server.MethodEntered.WaitAsync(TestContext.Current.CancellationToken)).WithCancellation(this.TimeoutToken);
         await completingTask;  // rethrow an exception if there is one.
 
+        this.mockStrategy.BlockCancelOutboundRequest = true;
         this.mockStrategy.AllowCancelOutboundRequestToExit.Reset();
         cts.Cancel();
-
-        // This may be invoked, but if the product doesn't invoke it, that's ok too.
-        ////await this.mockStrategy.OutboundRequestEndedInvoked.WaitAsync(this.TimeoutToken);
     }
 
     protected virtual void InitializeFormattersAndHandlers()
@@ -170,10 +180,16 @@ internal MockCancellationStrategy(CustomCancellationStrategyTests owner, ITestOu
 
         internal bool CancelRequestMade { get; private set; }
 
-        internal AsyncAutoResetEvent OutboundRequestEndedInvoked { get; } = new AsyncAutoResetEvent();
+        internal bool OutboundRequestEndedCalled { get; private set; }
+
+        internal AsyncManualResetEvent CancelOutboundRequestInvoked { get; } = new AsyncManualResetEvent();
+
+        internal AsyncManualResetEvent OutboundRequestEndedInvoked { get; } = new AsyncManualResetEvent();
 
         internal ManualResetEventSlim AllowCancelOutboundRequestToExit { get; } = new ManualResetEventSlim(initialState: true);
 
+        internal bool BlockCancelOutboundRequest { get; set; }
+
         public void CancelOutboundRequest(RequestId requestId)
         {
             this.logger.WriteLine($"{nameof(this.CancelOutboundRequest)}({requestId})");
@@ -192,20 +208,25 @@ public void CancelOutboundRequest(RequestId requestId)
 
             cts?.Cancel();
             this.CancelRequestMade = true;
+            this.CancelOutboundRequestInvoked.Set();
 
-            // Wait for the out of order invocation to happen if it's possible,
-            // so the OutboundCancellationStartAndRequestFinishOverlap test can catch it.
-            // Otherwise timeout, which is necessary to avoid a test hang when the product DOES work,
-            // since it shouldn't allow OutboundRequestEnded to execute before this method exits.
-            if (!this.AllowCancelOutboundRequestToExit.Wait(ExpectedTimeout))
+            if (this.BlockCancelOutboundRequest)
             {
-                this.logger.WriteLine("Timed out waiting for " + nameof(this.AllowCancelOutboundRequestToExit) + " to be signaled (good thing).");
+                // Wait for the out of order invocation to happen if it's possible,
+                // so the OutboundCancellationStartAndRequestFinishOverlap test can catch it.
+                // Otherwise timeout, which is necessary to avoid a test hang when the product DOES work,
+                // since it shouldn't allow OutboundRequestEnded to execute before this method exits.
+                if (!this.AllowCancelOutboundRequestToExit.Wait(ExpectedTimeout))
+                {
+                    this.logger.WriteLine("Timed out waiting for " + nameof(this.AllowCancelOutboundRequestToExit) + " to be signaled (good thing).");
+                }
             }
         }
 
         public void OutboundRequestEnded(RequestId requestId)
         {
             this.logger.WriteLine($"{nameof(this.OutboundRequestEnded)}({requestId}) invoked.");
+            this.OutboundRequestEndedCalled = true;
             lock (this.endedRequestIds)
             {
                 this.endedRequestIds.Add(requestId);
diff --git a/test/StreamJsonRpc.Tests/JsonRpcClient20InteropTests.cs b/test/StreamJsonRpc.Tests/JsonRpcClient20InteropTests.cs
@@ -273,6 +273,16 @@ public async Task CancelMessageSentAndResponseCompletes()
         }
     }
 
+    [Fact]
+    public async Task NotifyAsync_IgnoresOutboundRequestTimeout()
+    {
+        this.clientRpc.OutboundRequestTimeout = TimeSpan.FromMilliseconds(1);
+        await this.clientRpc.NotifyAsync("test");
+        JToken request = await this.ReceiveAsync();
+        Assert.Equal("test", request["method"]?.ToString());
+        Assert.Null(request["id"]);
+    }
+
     [Fact]
     public async Task ErrorResponseIncludesCallstack()
     {
diff --git a/test/StreamJsonRpc.Tests/JsonRpcDelegatedDispatchAndSendTests.cs b/test/StreamJsonRpc.Tests/JsonRpcDelegatedDispatchAndSendTests.cs
@@ -62,6 +62,24 @@ public async Task DelegatedDispatcherCanDispatchInReverseOrderBasedOnTopLevelPro
         }
     }
 
+    [Fact]
+    public async Task InvokeAsync_UsesOutboundRequestTimeoutWhenSendAsyncTimesOut()
+    {
+        var streams = Nerdbank.FullDuplexStream.CreateStreams();
+        using var clientRpc = new BlockingSendJsonRpc(new HeaderDelimitedMessageHandler(streams.Item1));
+        using var serverRpc = new DelegatedJsonRpc(new HeaderDelimitedMessageHandler(streams.Item2), this.server);
+
+        clientRpc.StartListening();
+        serverRpc.StartListening();
+        clientRpc.OutboundRequestTimeout = ExpectedTimeout;
+        clientRpc.BlockRequestSend = true;
+        Task<int> invokeTask = clientRpc.InvokeAsync<int>(nameof(Server.GetCallCountAsync));
+        await clientRpc.RequestSendBlocked.WaitAsync(this.TimeoutToken);
+
+        TimeoutException ex = await Assert.ThrowsAsync<TimeoutException>(() => invokeTask);
+        Assert.Contains(nameof(JsonRpc.OutboundRequestTimeout), ex.Message, StringComparison.Ordinal);
+    }
+
 #pragma warning disable CA1801 // use all parameters
     public class Server
     {
@@ -158,5 +176,28 @@ protected override ValueTask SendAsync(JsonRpcMessage message, CancellationToken
         }
     }
 
+    private sealed class BlockingSendJsonRpc : DelegatedJsonRpc
+    {
+        public BlockingSendJsonRpc(IJsonRpcMessageHandler handler)
+            : base(handler)
+        {
+        }
+
+        public bool BlockRequestSend { get; set; }
+
+        public AsyncAutoResetEvent RequestSendBlocked { get; } = new AsyncAutoResetEvent();
+
+        protected override async ValueTask SendAsync(JsonRpcMessage message, CancellationToken cancellationToken)
+        {
+            if (this.BlockRequestSend && message is JsonRpcRequest)
+            {
+                this.RequestSendBlocked.Set();
+                await Task.Delay(Timeout.Infinite, cancellationToken);
+            }
+
+            await base.SendAsync(message, cancellationToken);
+        }
+    }
+
 #pragma warning restore CA1801 // use all parameters
 }
diff --git a/test/StreamJsonRpc.Tests/JsonRpcProxyGenerationTests.cs b/test/StreamJsonRpc.Tests/JsonRpcProxyGenerationTests.cs
@@ -548,6 +548,18 @@ public async Task CallMethod_CancellationToken_int()
         await Assert.ThrowsAnyAsync<OperationCanceledException>(() => task);
     }
 
+    [Fact]
+    public async Task CallMethod_UsesJsonRpcOutboundRequestTimeout()
+    {
+        this.clientJsonRpc.OutboundRequestTimeout = ExpectedTimeout;
+        this.server.ResumeMethod.Reset();
+        Task task = this.clientRpc.HeavyWorkAsync(CancellationToken.None);
+        await this.server.MethodEntered.WaitAsync(TestContext.Current.CancellationToken).WithCancellation(this.TimeoutToken);
+
+        TimeoutException ex = await Assert.ThrowsAsync<TimeoutException>(() => task);
+        Assert.Contains(nameof(JsonRpc.OutboundRequestTimeout), ex.Message, StringComparison.Ordinal);
+    }
+
     [Fact]
     public async Task CallMethod_intCancellationToken_int()
     {
diff --git a/test/StreamJsonRpc.Tests/JsonRpcTests.cs b/test/StreamJsonRpc.Tests/JsonRpcTests.cs

Original file line number	Diff line number	Diff line change
`@@ -273,6 +273,16 @@ public async Task CancelMessageSentAndResponseCompletes()`
`273`	`273`	`}`
`274`	`274`	`}`
`275`	`275`
	`276`	`+ [Fact]`
	`277`	`+ public async Task NotifyAsync_IgnoresOutboundRequestTimeout()`
	`278`	`+ {`
	`279`	`+ this.clientRpc.OutboundRequestTimeout = TimeSpan.FromMilliseconds(1);`
	`280`	`+ await this.clientRpc.NotifyAsync("test");`
	`281`	`+ JToken request = await this.ReceiveAsync();`
	`282`	`+ Assert.Equal("test", request["method"]?.ToString());`
	`283`	`+ Assert.Null(request["id"]);`
	`284`	`+ }`
	`285`	`+`
`276`	`286`	`[Fact]`
`277`	`287`	`public async Task ErrorResponseIncludesCallstack()`
`278`	`288`	`{`