Fix cancellation of RPC methods

Fixes #1750

* Start by adding a test that demonstrates the error. Give a 5ms cancellation to `BasicConsumeAsync`, with a much longer delay via a hacked RabbitMQ. If running in debug mode, you will see the same `task canceled` exception, but it does not propagate to the test itself.
* Set cancellation correctly for TaskCompletionSource in AsyncRpcContinuation
* Handle continuation cancellation and timeouts correctly
* Refactor repeated code
* Add wait on `RegisteredAsync` to see how the changes in #1750 affect the order of operations.
* Check to ensure no connection shutdown in `TestBasicConsumeCancellation_GH1750`
* @danielmarbach noticed an improvement. Do not elide `await` here.
* Extend GHA timeout for Windows builds, yuck (see actions/runner-images#7320)

.github/workflows/build-test.yaml

@@ -67,7 +67,7 @@ jobs:
         id: install-start-rabbitmq
         run: ${{ github.workspace }}\.ci\windows\gha-setup.ps1
       - name: Integration Tests
-        timeout-minutes: 25
+        timeout-minutes: 45
         run: |
           Start-Job -Verbose -ScriptBlock { & "${{ github.workspace }}\.ci\windows\toxiproxy\toxiproxy-server.exe" | Out-File -LiteralPath $env:APPDATA\RabbitMQ\log\toxiproxy-log.txt }; `
           dotnet test `
@@ -113,7 +113,7 @@ jobs:
         id: install-start-rabbitmq
         run: ${{ github.workspace }}\.ci\windows\gha-setup.ps1
       - name: Sequential Integration Tests
-        timeout-minutes: 25
+        timeout-minutes: 45
         run: dotnet test `
             --environment 'RABBITMQ_LONG_RUNNING_TESTS=true' `
             --environment "RABBITMQ_RABBITMQCTL_PATH=${{ steps.install-start-rabbitmq.outputs.path }}" `

projects/RabbitMQ.Client/ConsumerDispatching/ConsumerDispatcherChannelBase.cs

@@ -83,61 +83,65 @@ internal ConsumerDispatcherChannelBase(Impl.Channel channel, ushort concurrency)
 
         public ushort Concurrency => _concurrency;
 
-        public ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken)
+        public async ValueTask HandleBasicConsumeOkAsync(IAsyncBasicConsumer consumer, string consumerTag, CancellationToken cancellationToken)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+
             if (false == _disposed && false == _quiesce)
             {
-                AddConsumer(consumer, consumerTag);
-                WorkStruct work = WorkStruct.CreateConsumeOk(consumer, consumerTag);
-                return _writer.WriteAsync(work, cancellationToken);
-            }
-            else
-            {
-                return default;
+                try
+                {
+                    AddConsumer(consumer, consumerTag);
+                    WorkStruct work = WorkStruct.CreateConsumeOk(consumer, consumerTag);
+                    await _writer.WriteAsync(work, cancellationToken)
+                        .ConfigureAwait(false);
+                }
+                catch
+                {
+                    _ = GetAndRemoveConsumer(consumerTag);
+                    throw;
+                }
             }
         }
 
-        public ValueTask HandleBasicDeliverAsync(string consumerTag, ulong deliveryTag, bool redelivered,
+        public async ValueTask HandleBasicDeliverAsync(string consumerTag, ulong deliveryTag, bool redelivered,
             string exchange, string routingKey, IReadOnlyBasicProperties basicProperties, RentedMemory body,
             CancellationToken cancellationToken)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+
             if (false == _disposed && false == _quiesce)
             {
                 IAsyncBasicConsumer consumer = GetConsumerOrDefault(consumerTag);
                 var work = WorkStruct.CreateDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, body);
-                return _writer.WriteAsync(work, cancellationToken);
-            }
-            else
-            {
-                return default;
+                await _writer.WriteAsync(work, cancellationToken)
+                    .ConfigureAwait(false);
             }
         }
 
-        public ValueTask HandleBasicCancelOkAsync(string consumerTag, CancellationToken cancellationToken)
+        public async ValueTask HandleBasicCancelOkAsync(string consumerTag, CancellationToken cancellationToken)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+
             if (false == _disposed && false == _quiesce)
             {
                 IAsyncBasicConsumer consumer = GetAndRemoveConsumer(consumerTag);
                 WorkStruct work = WorkStruct.CreateCancelOk(consumer, consumerTag);
-                return _writer.WriteAsync(work, cancellationToken);
-            }
-            else
-            {
-                return default;
+                await _writer.WriteAsync(work, cancellationToken)
+                    .ConfigureAwait(false);
             }
         }
 
-        public ValueTask HandleBasicCancelAsync(string consumerTag, CancellationToken cancellationToken)
+        public async ValueTask HandleBasicCancelAsync(string consumerTag, CancellationToken cancellationToken)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+
             if (false == _disposed && false == _quiesce)
             {
                 IAsyncBasicConsumer consumer = GetAndRemoveConsumer(consumerTag);
                 WorkStruct work = WorkStruct.CreateCancel(consumer, consumerTag);
-                return _writer.WriteAsync(work, cancellationToken);
-            }
-            else
-            {
-                return default;
+                await _writer.WriteAsync(work, cancellationToken)
+                    .ConfigureAwait(false);
             }
         }