Respect Dequeue Async CT

Author: LukeButters

source/Halibut.Tests/LocalExecutionModeFixture.cs

@@ -29,10 +29,6 @@ public async Task SimpleLocalExecutionExample()
         {
             var services = GetDelegateServiceFactory();
             var timeoutsAndLimits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();
-            timeoutsAndLimits = new HalibutTimeoutsAndLimits();
-
-            // Use a shared queue factory so client and worker share the same queue
-            var queueFactory = new PendingRequestQueueFactoryAsync(timeoutsAndLimits, new LogFactory());
 
             var logFactory = new CachingLogFactory(new TestContextLogCreator("", LogLevel.Trace));
 
@@ -44,22 +40,21 @@ public async Task SimpleLocalExecutionExample()
 
             await using var client = new HalibutRuntimeBuilder()
                 .WithServerCertificate(Certificates.Octopus)
-                .WithPendingRequestQueueFactory(RedisFactory())
+                .WithPendingRequestQueueFactory(RedisFactory(preSharedGuid, disposables, log, logFactory))
                 .WithHalibutTimeoutsAndLimits(timeoutsAndLimits)
                 .Build();
 
             await using var worker = new HalibutRuntimeBuilder()
                 .WithServerCertificate(Certificates.TentaclePolling)
                 .WithServiceFactory(services)
-                .WithPendingRequestQueueFactory(RedisFactory())
+                .WithPendingRequestQueueFactory(RedisFactory(preSharedGuid, disposables, log, logFactory))
                 .WithHalibutTimeoutsAndLimits(timeoutsAndLimits)
                 .Build();
 
             // Start worker polling for local://test-worker
             using var workerCts = new CancellationTokenSource();
             var pollingTask = Task.Run(async () =>
             {
-                //await Task.Delay(TimeSpan.FromSeconds(10));
                 await worker.PollLocalAsync(new Uri("local://test-worker"), workerCts.Token);
             }, workerCts.Token);
 
@@ -72,23 +67,29 @@ public async Task SimpleLocalExecutionExample()
 
             await workerCts.CancelAsync();
 
-            Func<QueueMessageSerializer, IPendingRequestQueueFactory> RedisFactory()
+            await pollingTask;
+        }
+
+        Func<QueueMessageSerializer, IPendingRequestQueueFactory> RedisFactory(
+            Guid preSharedGuid, 
+            DisposableCollection disposables,
+            TestContextLogCreator log,
+            CachingLogFactory logFactory)
+        {
+            return msgSer =>
             {
-                return msgSer =>
-                {
-                    var redisFacade = RedisFacadeBuilder.CreateRedisFacade(prefix: preSharedGuid);
-                    disposables.AddAsyncDisposable(redisFacade);
-                    var watchForRedisLosingAllItsData = new WatchForRedisLosingAllItsData(redisFacade, log.CreateNewForPrefix("watcher"));
-                    disposables.AddAsyncDisposable(watchForRedisLosingAllItsData);
+                var redisFacade = RedisFacadeBuilder.CreateRedisFacade(prefix: preSharedGuid);
+                disposables.AddAsyncDisposable(redisFacade);
+                var watchForRedisLosingAllItsData = new WatchForRedisLosingAllItsData(redisFacade, log.CreateNewForPrefix("watcher"));
+                disposables.AddAsyncDisposable(watchForRedisLosingAllItsData);
 
-                    return new RedisPendingRequestQueueFactory(msgSer,
-                        new InMemoryStoreDataStreamsForDistributedQueues(),
-                        watchForRedisLosingAllItsData,
-                        new HalibutRedisTransport(redisFacade),
-                        new HalibutTimeoutsAndLimitsForTestsBuilder().Build(),
-                        logFactory);
-                };
-            }
+                return new RedisPendingRequestQueueFactory(msgSer,
+                    new InMemoryStoreDataStreamsForDistributedQueues(),
+                    watchForRedisLosingAllItsData,
+                    new HalibutRedisTransport(redisFacade),
+                    new HalibutTimeoutsAndLimitsForTestsBuilder().Build(),
+                    logFactory);
+            };
         }
 
         static DelegateServiceFactory GetDelegateServiceFactory()

source/Halibut/Queue/Redis/RedisPendingRequestQueue.cs

@@ -380,7 +380,7 @@ async Task<bool> TryClearRequestFromQueue(RedisPendingRequest redisPending)
             // It will kill the TCP connection, which will force re-connect (in perhaps a backoff function)
             // This could result in connecting to a node that is actually connected to redis. It could also
             // cause a cascade of failure from high load.
-            var pending = await DequeueNextAsync();
+            var pending = await DequeueNextAsync(cancellationToken);
             if (pending == null) return null;
 
             var pendingRequest = pending.Value.Item1;
@@ -491,9 +491,9 @@ public async Task ApplyResponse(ResponseMessage response, Guid requestActivityId
             }
         }
 
-        async Task<(RequestMessage, RequestDataStreamsTransferProgress)?> DequeueNextAsync()
+        async Task<(RequestMessage, RequestDataStreamsTransferProgress)?> DequeueNextAsync(CancellationToken cancellationToken)
         {
-            await using var cts = new CancelOnDisposeCancellationToken(queueToken);
+            await using var cts = new CancelOnDisposeCancellationToken(queueToken, cancellationToken);
             try
             {
                 hasItemsForEndpoint.Reset();