Delay without exception is cheaper

LukeButters · LukeButters · commit bf969d11b191 · 2025-09-19T16:23:40.000+10:00
diff --git a/source/Halibut.Tests/Queue/Redis/RedisPendingRequestQueueFixture.cs b/source/Halibut.Tests/Queue/Redis/RedisPendingRequestQueueFixture.cs
@@ -971,6 +971,8 @@ public async Task WhenUsingTheRedisQueue_ASimpleEchoServiceCanBeCalled(ClientAnd
                     await echo.CountBytesAsync(DataStream.FromString("hello"));
                 }
             }
+            
+            Logger.Fatal("YYYYYY\n\n\nYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYTotal {S}\n\n\n\n", MessageSerialiserAndDataStreamStorage.totalDecompress);
         }
         
         [Test]
diff --git a/source/Halibut.Tests/Util/DelayWithoutExceptionTest.cs b/source/Halibut.Tests/Util/DelayWithoutExceptionTest.cs
@@ -0,0 +1,33 @@
+// Copyright 2012-2013 Octopus Deploy Pty. Ltd.
+// 
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// 
+//   http://www.apache.org/licenses/LICENSE-2.0
+// 
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Halibut.Util;
+using NUnit.Framework;
+
+namespace Halibut.Tests.Util
+{
+    public class DelayWithoutExceptionTest : BaseTest
+    {
+        [Test]
+        public async Task DelayWithoutException_ShouldNotThrow()
+        {
+            using var cts = new CancellationTokenSource();
+            cts.CancelAfter(10);
+            await DelayWithoutException.Delay(TimeSpan.FromDays(1), cts.Token);
+        }
+    }
+}
diff --git a/source/Halibut/Queue/QueueMessageSerializer.cs b/source/Halibut/Queue/QueueMessageSerializer.cs
@@ -43,8 +43,12 @@ public QueueMessageSerializer(Func<StreamCapturingJsonSerializer> createStreamCa
                 stream = WrapInMessageSerialisationStreams(messageStreamWrappers, stream, wrappedStreamDisposables);
 
                 // TODO instead store 
-                using (var zip = new DeflateStream(stream, CompressionMode.Compress, true)) {
-                    using (var jsonTextWriter = new BsonDataWriter(zip) { CloseOutput = false })
+                
+                using (var zip = new DeflateStream(stream, CompressionMode.Compress, true))
+                using (var buf = new BufferedStream(zip))
+                {
+                    
+                    using (var jsonTextWriter = new BsonDataWriter(buf) { CloseOutput = false })
                     {
                         var streamCapturingSerializer = createStreamCapturingSerializer();
                         streamCapturingSerializer.Serializer.Serialize(jsonTextWriter, new MessageEnvelope<T>(message));
@@ -188,7 +192,8 @@ public async Task<byte[]> PrepareBytesFromWire(byte[] responseBytes)
             stream = WrapStreamInMessageDeserialisationStreams(messageStreamWrappers, stream, disposables);
             
             using var deflateStream = new DeflateStream(stream, CompressionMode.Decompress, true);
-            using (var bson = new BsonDataReader(deflateStream) { CloseInput = false })
+            using var buf = new BufferedStream(deflateStream);
+            using (var bson = new BsonDataReader(buf) { CloseInput = false })
             {
                 var streamCapturingSerializer = createStreamCapturingSerializer();
                 var result = streamCapturingSerializer.Serializer.Deserialize<MessageEnvelope<T>>(bson);
diff --git a/source/Halibut/Queue/QueuedDataStreams/HeartBeatDrivenDataStreamProgressReporter.cs b/source/Halibut/Queue/QueuedDataStreams/HeartBeatDrivenDataStreamProgressReporter.cs
@@ -16,7 +16,7 @@ public class HeartBeatDrivenDataStreamProgressReporter : IAsyncDisposable, IGetN
     {
         readonly ImmutableDictionary<Guid, IDataStreamWithFileUploadProgress> dataStreamsToReportProgressOn;
 
-        readonly ConcurrentBag<Guid> completedDataStreams = new();
+        readonly HashSet<Guid> completedDataStreams = new();
 
         HeartBeatDrivenDataStreamProgressReporter(ImmutableDictionary<Guid, IDataStreamWithFileUploadProgress> dataStreamsToReportProgressOn)
         {
@@ -30,7 +30,11 @@ public async Task HeartBeatReceived(HeartBeatMessage heartBeatMessage, Cancellat
             
             foreach (var keyValuePair in heartBeatMessage.DataStreamProgress)
             {
-                if(completedDataStreams.Contains(keyValuePair.Key)) continue;
+                lock (completedDataStreams)
+                {
+                    if(completedDataStreams.Contains(keyValuePair.Key)) continue;
+                }
+                
                 
                 if (dataStreamsToReportProgressOn.TryGetValue(keyValuePair.Key, out var dataStreamWithTransferProgress))
                 {
@@ -39,7 +43,10 @@ public async Task HeartBeatReceived(HeartBeatMessage heartBeatMessage, Cancellat
                     
                     if (dataStreamWithTransferProgress.Length == keyValuePair.Value)
                     {
-                        completedDataStreams.Add(keyValuePair.Key);
+                        lock (completedDataStreams)
+                        {
+                            completedDataStreams.Add(keyValuePair.Key);
+                        }
                     }
                 }
             }
@@ -57,13 +64,21 @@ public async ValueTask DisposeAsync()
             // this object is disposable and on dispose we note that file will no longer be uploading. Which
             // for the normal percentage based file transfer progress will result in marking the DataStreams as 100% uploaded.
             // If we don't do this at the end of a successful call we may find DataStream progress is reported as less than 100%.
+            var localCopyCompletedDataStreams = new List<Guid>();
+            
+            // Because of where this is used, it is hard to be sure this object won't be used while disposing,
+            // so take a copy of streams we have already completed.
+            lock (completedDataStreams)
+            {
+                localCopyCompletedDataStreams.AddRange(completedDataStreams);
+            }
             foreach (var keyValuePair in dataStreamsToReportProgressOn)
             {
-                if (!completedDataStreams.Contains(keyValuePair.Key))
+                if (!localCopyCompletedDataStreams.Contains(keyValuePair.Key))
                 {
                     var progress = keyValuePair.Value.DataStreamTransferProgress;
+                    // Thus may be called twice if HeartBeats are received while disposing.
                     await progress.NoLongerUploading(CancellationToken.None);
-                    completedDataStreams.Add(keyValuePair.Key);
                 }
             }
         }
diff --git a/source/Halibut/Queue/Redis/Cancellation/DelayBeforeSubscribingToRequestCancellation.cs b/source/Halibut/Queue/Redis/Cancellation/DelayBeforeSubscribingToRequestCancellation.cs
@@ -1,6 +1,7 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using Halibut.Util;
 
 namespace Halibut.Queue.Redis.Cancellation
 {
@@ -15,14 +16,15 @@ public DelayBeforeSubscribingToRequestCancellation(TimeSpan delay)
         
         public async Task WaitBeforeHeartBeatSendingOrReceiving(CancellationToken cancellationToken)
         {
-            try
-            {
-                await Task.Delay(Delay, cancellationToken);
-            }
-            catch
-            {
-                // If only Delay had an option to not throw.
-            }
+            await DelayWithoutException.Delay(Delay, cancellationToken);
+            // try
+            // {
+            //     await DelayWithoutException.Delay(Delay, cancellationToken);
+            // }
+            // catch
+            // {
+            //     // If only Delay had an option to not throw.
+            // }
         }
     }
 }
diff --git a/source/Halibut/Queue/Redis/Cancellation/WatchForRequestCancellation.cs b/source/Halibut/Queue/Redis/Cancellation/WatchForRequestCancellation.cs
@@ -62,7 +62,7 @@ async Task WatchForCancellation(
                 // Also poll to see if the request is cancelled since we can miss the publication.
                 while (!token.IsCancellationRequested)
                 {
-                    await Try.IgnoringError(async () => await Task.Delay(TimeSpan.FromSeconds(60), token));
+                    await DelayWithoutException.Delay(TimeSpan.FromSeconds(60), token);
                     
                     if(token.IsCancellationRequested) return;
                     
diff --git a/source/Halibut/Queue/Redis/MessageStorage/MessageSerialiserAndDataStreamStorage.cs b/source/Halibut/Queue/Redis/MessageStorage/MessageSerialiserAndDataStreamStorage.cs
@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.IO;
 using System.Linq;
 using System.Threading;
@@ -70,8 +71,17 @@ public MessageSerialiserAndDataStreamStorage(QueueMessageSerializer queueMessage
             this.storeDataStreamsForDistributedQueues = storeDataStreamsForDistributedQueues;
         }
 
+        // public static long total = 0;
         public async Task<(RedisStoredMessage, HeartBeatDrivenDataStreamProgressReporter)> PrepareRequest(RequestMessage request, CancellationToken cancellationToken)
         {
+            // var sw = Stopwatch.StartNew();
+            // for (int i = 0; i < 1000; i++)
+            // {
+            //     await queueMessageSerializer.PrepareMessageForWireTransferAndForQueue(request);
+            // }
+            // sw.Stop();
+            // Interlocked.Add(ref total, (long) sw.Elapsed.TotalMilliseconds);
+            
             var (jsonRequestMessage, dataStreams) = await queueMessageSerializer.PrepareMessageForWireTransferAndForQueue(request);
             SwitchDataStreamsToNotReportProgress(dataStreams);
             var dataStreamProgressReporter = HeartBeatDrivenDataStreamProgressReporter.CreateForDataStreams(dataStreams);
@@ -118,11 +128,6 @@ static void SwitchDataStreamsToNotReportProgress(IReadOnlyList<DataStream> dataS
         public async Task<RedisStoredMessage> PrepareResponseForStorageInRedis(Guid activityId, ResponseBytesAndDataStreams response, CancellationToken cancellationToken)
         {
             var responseBytesToStoreInRedis = await queueMessageSerializer.PrepareBytesFromWire(response.ResponseBytes);
-
-            if ("hello".Length == 0)
-            {
-                await queueMessageSerializer.ConvertStoredResponseToResponseMessage<ResponseMessage>(responseBytesToStoreInRedis);
-            }
             
             
             var dataStreamMetadata = await storeDataStreamsForDistributedQueues.StoreDataStreams(response.DataStreams, cancellationToken);
@@ -134,9 +139,18 @@ public async Task<RedisStoredMessage> PrepareResponseForStorageInRedis(Guid acti
             
             return new RedisStoredMessage(responseBytesToStoreInRedis, dataStreamMetadata);
         }
-        
+
+        public static long totalDecompress = 0;
         public async Task<ResponseMessage> ReadResponseFromRedisStoredMessage(RedisStoredMessage storedMessage, CancellationToken cancellationToken)
         {
+            // var sw = Stopwatch.StartNew();
+            // for (int i = 0; i < 1000; i++)
+            // {
+            //     await queueMessageSerializer.ConvertStoredResponseToResponseMessage<ResponseMessage>(storedMessage.Message);
+            // }
+            // sw.Stop();
+            // Interlocked.Add(ref totalDecompress, (long) sw.Elapsed.TotalMilliseconds);
+            
             var (response, dataStreams) = await queueMessageSerializer.ConvertStoredResponseToResponseMessage<ResponseMessage>(storedMessage.Message);
             
             var rehydratableDataStreams = BuildUpRehydratableDataStreams(dataStreams, out _);
diff --git a/source/Halibut/Queue/Redis/NodeHeartBeat/HeartBeatInitialDelay.cs b/source/Halibut/Queue/Redis/NodeHeartBeat/HeartBeatInitialDelay.cs
@@ -1,6 +1,7 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using Halibut.Util;
 
 namespace Halibut.Queue.Redis.NodeHeartBeat
 {
@@ -17,7 +18,7 @@ public async Task WaitBeforeHeartBeatSendingOrReceiving(CancellationToken cancel
         {
             try
             {
-                await Task.Delay(InitialDelay, cancellationToken);
+                await DelayWithoutException.Delay(InitialDelay, cancellationToken);
             }
             catch
             {
diff --git a/source/Halibut/Queue/Redis/NodeHeartBeat/NodeHeartBeatSender.cs b/source/Halibut/Queue/Redis/NodeHeartBeat/NodeHeartBeatSender.cs
@@ -74,7 +74,7 @@ async Task SendPulsesWhileProcessingRequest(
                     log.WriteException(EventType.Diagnostic, "Failed to send heartbeat for {0} node, request {1}, switching to panic mode with {2} second intervals", ex, nodeSendingPulsesType, requestActivityId, delayBetweenPulse.TotalSeconds);
                 }
                 
-                await Try.IgnoringError(async () => await Task.Delay(delayBetweenPulse, cancellationToken));
+                await DelayWithoutException.Delay(delayBetweenPulse, cancellationToken);
             }
             
             log.Write(EventType.Diagnostic, "Heartbeat pulse loop ended for {0} node, request {1}", nodeSendingPulsesType, requestActivityId);
diff --git a/source/Halibut/Queue/Redis/NodeHeartBeat/NodeHeartBeatWatcher.cs b/source/Halibut/Queue/Redis/NodeHeartBeat/NodeHeartBeatWatcher.cs
@@ -136,7 +136,7 @@ static async Task<NodeWatcherResult> WatchForPulsesFromNode(Uri endpoint,
                             TimeSpan.FromSeconds(30),
                             maxTimeBetweenHeartBeetsBeforeNodeIsAssumedToBeOffline - timeSinceLastHeartBeat + TimeSpan.FromSeconds(1));
 
-                        await Try.IgnoringError(async () => await Task.Delay(timeToWait, watchCancellationToken));
+                        await DelayWithoutException.Delay(timeToWait, watchCancellationToken);
                     }
 
                     log.Write(EventType.Diagnostic, "{0} node watcher cancelled, request {1}", watchingForPulsesFrom, requestActivityId);
diff --git a/source/Halibut/Queue/Redis/RedisDataLossDetection/WatchForRedisLosingAllItsData.cs b/source/Halibut/Queue/Redis/RedisDataLossDetection/WatchForRedisLosingAllItsData.cs
@@ -130,8 +130,8 @@ async Task WatchForDataLoss(CancellationToken cancellationToken)
 
                 await Try.IgnoringError(async () =>
                 {
-                    if (!hasSetKey) await Task.Delay(SetupErrorBackoffDelay, cancellationToken);
-                    else await Task.Delay(DataLossCheckInterval, cancellationToken);
+                    if (!hasSetKey) await DelayWithoutException.Delay(SetupErrorBackoffDelay, cancellationToken);
+                    else await DelayWithoutException.Delay(DataLossCheckInterval, cancellationToken);
                 });
 
             }
diff --git a/source/Halibut/Queue/Redis/RedisHelpers/RedisFacade.cs b/source/Halibut/Queue/Redis/RedisHelpers/RedisFacade.cs
@@ -90,7 +90,7 @@ async Task<T> ExecuteWithRetry<T>(Func<Task<T>> operation, CancellationToken can
                 catch (Exception ex) when (stopwatch.Elapsed < MaxDurationToRetryFor && !combinedToken.IsCancellationRequested)
                 {
                     log?.Write(EventType.Diagnostic, $"Redis operation failed, retrying in {retryDelay.TotalSeconds}s: {ex.Message}");
-                    await Task.Delay(retryDelay, combinedToken);
+                    await DelayWithoutException.Delay(retryDelay, combinedToken);
                 }
             }
         }
@@ -115,7 +115,7 @@ async Task ExecuteWithRetry(Func<Task> operation, CancellationToken cancellation
                 catch (Exception ex) when (stopwatch.Elapsed < MaxDurationToRetryFor && !combinedToken.IsCancellationRequested)
                 {
                     log?.Write(EventType.Diagnostic, $"Redis operation failed, retrying in {retryDelay.TotalSeconds}s: {ex.Message}");
-                    await Task.Delay(retryDelay, combinedToken);
+                    await DelayWithoutException.Delay(retryDelay, combinedToken);
                 }
             }
         }
@@ -181,7 +181,7 @@ public async Task<IAsyncDisposable> SubscribeToChannel(string channelName, Func<
                 catch (Exception ex)
                 {
                     log?.WriteException(EventType.Diagnostic, "Failed to subscribe to Redis channel {0}, retrying in 2 seconds", ex, channelName);
-                    await Try.IgnoringError(async () => await Task.Delay(2000, cancellationToken));
+                    await DelayWithoutException.Delay(TimeSpan.FromSeconds(1), cancellationToken);
                 }
             }
         }
@@ -375,6 +375,8 @@ async Task SetTtlForKeyRaw(RedisKey key, TimeSpan ttl, CancellationToken cancell
             await ExecuteWithRetry(async () =>
             {
                 var database = Connection.GetDatabase();
+                // TODO: Does fire and forget reduce load?
+                // TODO: experiment without this
                 await database.KeyExpireAsync(key, ttl);
             }, cancellationToken);
         }
diff --git a/source/Halibut/Queue/Redis/RedisPendingRequest.cs b/source/Halibut/Queue/Redis/RedisPendingRequest.cs
@@ -54,7 +54,7 @@ public async Task WaitUntilComplete(Func<Task> checkIfPendingRequestWasCollected
         {
             log.Write(EventType.MessageExchange, "Request {0} was queued", request);
 
-            var pendingRequestPickupTimeout = Try.IgnoringError(async () => await Task.Delay(request.Destination.PollingRequestQueueTimeout, cancellationToken));
+            var pendingRequestPickupTimeout = Try.IgnoringError(async () => await DelayWithoutException.Delay(request.Destination.PollingRequestQueueTimeout, cancellationToken));
             var responseWaiterTask = responseWaiter.WaitAsync(cancellationToken);
             
             await Task.WhenAny(pendingRequestPickupTimeout, responseWaiterTask);
diff --git a/source/Halibut/Queue/Redis/RedisPendingRequestQueue.cs b/source/Halibut/Queue/Redis/RedisPendingRequestQueue.cs
@@ -535,7 +535,7 @@ await Task.WhenAny(
                     log.WriteException(EventType.Error, "Error occured dequeuing from the queue", ex);
                     // It is very likely a queue error means every tentacle will return an error.
                     // Add a random delay to help avoid every client coming back at exactly the same time.
-                    await Task.Delay(TimeSpan.FromSeconds(new Random().Next(15)), cts.Token);
+                    await DelayWithoutException.Delay(TimeSpan.FromSeconds(new Random().Next(15)), cts.Token);
                 }
                 throw;
             }
diff --git a/source/Halibut/Queue/Redis/ResponseMessageTransfer/PollAndSubscribeToResponse.cs b/source/Halibut/Queue/Redis/ResponseMessageTransfer/PollAndSubscribeToResponse.cs
@@ -75,7 +75,7 @@ async Task WaitForResponse(CancellationToken token)
                 {
                     var delay = pollBackoffStrategy.GetSleepPeriod();
                     log.Write(EventType.Diagnostic, "Waiting {0} seconds before next poll for response - Endpoint: {1}, ActivityId: {2}", delay.TotalSeconds, endpoint, activityId);
-                    await Try.IgnoringError(async () => await Task.Delay(delay, token));
+                    await DelayWithoutException.Delay(delay, token);
                     if(token.IsCancellationRequested) break;
                     log.Write(EventType.Diagnostic, "Done waiting going to poll for response - Endpoint: {0}, ActivityId: {1}", endpoint, activityId);
                     
diff --git a/source/Halibut/Util/DelayWithoutException.cs b/source/Halibut/Util/DelayWithoutException.cs
@@ -0,0 +1,28 @@
+// Copyright 2012-2013 Octopus Deploy Pty. Ltd.
+// 
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// 
+//   http://www.apache.org/licenses/LICENSE-2.0
+// 
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Halibut.Util
+{
+    public static class DelayWithoutException
+    {
+        public static Task Delay(TimeSpan timeSpan, CancellationToken cancellationToken)
+        {
+            return Task.Delay(timeSpan, cancellationToken).ContinueWith(t => { }, TaskContinuationOptions.ExecuteSynchronously);
+        }
+    }
+}
diff --git a/source/Halibut/Util/Try.cs b/source/Halibut/Util/Try.cs
@@ -47,16 +47,9 @@ public static void IgnoringError(Action tryThisAction)
             }
         }
         
-        public static async Task IgnoringError(Func<Task> tryThisAction)
+        public static Task IgnoringError(Func<Task> tryThisAction)
         {
-            try
-            {
-                await tryThisAction();
-            }
-            catch
-            {
-                // ignored
-            }
+            return tryThisAction().ContinueWith(t => { }, TaskContinuationOptions.ExecuteSynchronously);
         }
     }
     

Original file line number	Diff line number	Diff line change
`@@ -971,6 +971,8 @@ public async Task WhenUsingTheRedisQueue_ASimpleEchoServiceCanBeCalled(ClientAnd`
`971`	`971`	`await echo.CountBytesAsync(DataStream.FromString("hello"));`
`972`	`972`	`}`
`973`	`973`	`}`
	`974`	`+`
	`975`	`+ Logger.Fatal("YYYYYY\n\n\nYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYTotal {S}\n\n\n\n", MessageSerialiserAndDataStreamStorage.totalDecompress);`
`974`	`976`	`}`
`975`	`977`
`976`	`978`	`[Test]`
Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@ public class HeartBeatDrivenDataStreamProgressReporter : IAsyncDisposable, IGetN`
`16`	`16`	`{`
`17`	`17`	`readonly ImmutableDictionary<Guid, IDataStreamWithFileUploadProgress> dataStreamsToReportProgressOn;`
`18`	`18`
`19`		`- readonly ConcurrentBag<Guid> completedDataStreams = new();`
	`19`	`+ readonly HashSet<Guid> completedDataStreams = new();`
`20`	`20`
`21`	`21`	`HeartBeatDrivenDataStreamProgressReporter(ImmutableDictionary<Guid, IDataStreamWithFileUploadProgress> dataStreamsToReportProgressOn)`
`22`	`22`	`{`
`@@ -30,7 +30,11 @@ public async Task HeartBeatReceived(HeartBeatMessage heartBeatMessage, Cancellat`
`30`	`30`
`31`	`31`	`foreach (var keyValuePair in heartBeatMessage.DataStreamProgress)`
`32`	`32`	`{`
`33`		`- if(completedDataStreams.Contains(keyValuePair.Key)) continue;`
	`33`	`+ lock (completedDataStreams)`
	`34`	`+ {`
	`35`	`+ if(completedDataStreams.Contains(keyValuePair.Key)) continue;`
	`36`	`+ }`
	`37`	`+`
`34`	`38`
`35`	`39`	`if (dataStreamsToReportProgressOn.TryGetValue(keyValuePair.Key, out var dataStreamWithTransferProgress))`
`36`	`40`	`{`
`@@ -39,7 +43,10 @@ public async Task HeartBeatReceived(HeartBeatMessage heartBeatMessage, Cancellat`
`39`	`43`
`40`	`44`	`if (dataStreamWithTransferProgress.Length == keyValuePair.Value)`
`41`	`45`	`{`
`42`		`- completedDataStreams.Add(keyValuePair.Key);`
	`46`	`+ lock (completedDataStreams)`
	`47`	`+ {`
	`48`	`+ completedDataStreams.Add(keyValuePair.Key);`
	`49`	`+ }`
`43`	`50`	`}`
`44`	`51`	`}`
`45`	`52`	`}`
`@@ -57,13 +64,21 @@ public async ValueTask DisposeAsync()`
`57`	`64`	`// this object is disposable and on dispose we note that file will no longer be uploading. Which`
`58`	`65`	`// for the normal percentage based file transfer progress will result in marking the DataStreams as 100% uploaded.`
`59`	`66`	`// If we don't do this at the end of a successful call we may find DataStream progress is reported as less than 100%.`
	`67`	`+ var localCopyCompletedDataStreams = new List<Guid>();`
	`68`	`+`
	`69`	`+ // Because of where this is used, it is hard to be sure this object won't be used while disposing,`
	`70`	`+ // so take a copy of streams we have already completed.`
	`71`	`+ lock (completedDataStreams)`
	`72`	`+ {`
	`73`	`+ localCopyCompletedDataStreams.AddRange(completedDataStreams);`
	`74`	`+ }`
`60`	`75`	`foreach (var keyValuePair in dataStreamsToReportProgressOn)`
`61`	`76`	`{`
`62`		`- if (!completedDataStreams.Contains(keyValuePair.Key))`
	`77`	`+ if (!localCopyCompletedDataStreams.Contains(keyValuePair.Key))`
`63`	`78`	`{`
`64`	`79`	`var progress = keyValuePair.Value.DataStreamTransferProgress;`
	`80`	`+ // Thus may be called twice if HeartBeats are received while disposing.`
`65`	`81`	`await progress.NoLongerUploading(CancellationToken.None);`
`66`		`- completedDataStreams.Add(keyValuePair.Key);`
`67`	`82`	`}`
`68`	`83`	`}`
`69`	`84`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Threading;`
`3`	`3`	`using System.Threading.Tasks;`
	`4`	`+using Halibut.Util;`
`4`	`5`
`5`	`6`	`namespace Halibut.Queue.Redis.Cancellation`
`6`	`7`	`{`
`@@ -15,14 +16,15 @@ public DelayBeforeSubscribingToRequestCancellation(TimeSpan delay)`
`15`	`16`
`16`	`17`	`public async Task WaitBeforeHeartBeatSendingOrReceiving(CancellationToken cancellationToken)`
`17`	`18`	`{`
`18`		`- try`
`19`		`- {`
`20`		`- await Task.Delay(Delay, cancellationToken);`
`21`		`- }`
`22`		`- catch`
`23`		`- {`
`24`		`- // If only Delay had an option to not throw.`
`25`		`- }`
	`19`	`+ await DelayWithoutException.Delay(Delay, cancellationToken);`
	`20`	`+ // try`
	`21`	`+ // {`
	`22`	`+ // await DelayWithoutException.Delay(Delay, cancellationToken);`
	`23`	`+ // }`
	`24`	`+ // catch`
	`25`	`+ // {`
	`26`	`+ // // If only Delay had an option to not throw.`
	`27`	`+ // }`
`26`	`28`	`}`
`27`	`29`	`}`
`28`	`30`	`}`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ async Task WatchForCancellation(`
`62`	`62`	`// Also poll to see if the request is cancelled since we can miss the publication.`
`63`	`63`	`while (!token.IsCancellationRequested)`
`64`	`64`	`{`
`65`		`- await Try.IgnoringError(async () => await Task.Delay(TimeSpan.FromSeconds(60), token));`
	`65`	`+ await DelayWithoutException.Delay(TimeSpan.FromSeconds(60), token);`
`66`	`66`
`67`	`67`	`if(token.IsCancellationRequested) return;`
`68`	`68`
Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,7 @@ async Task SendPulsesWhileProcessingRequest(`
`74`	`74`	`log.WriteException(EventType.Diagnostic, "Failed to send heartbeat for {0} node, request {1}, switching to panic mode with {2} second intervals", ex, nodeSendingPulsesType, requestActivityId, delayBetweenPulse.TotalSeconds);`
`75`	`75`	`}`
`76`	`76`
`77`		`- await Try.IgnoringError(async () => await Task.Delay(delayBetweenPulse, cancellationToken));`
	`77`	`+ await DelayWithoutException.Delay(delayBetweenPulse, cancellationToken);`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`log.Write(EventType.Diagnostic, "Heartbeat pulse loop ended for {0} node, request {1}", nodeSendingPulsesType, requestActivityId);`
Original file line number	Diff line number	Diff line change
`@@ -136,7 +136,7 @@ static async Task<NodeWatcherResult> WatchForPulsesFromNode(Uri endpoint,`
`136`	`136`	`TimeSpan.FromSeconds(30),`
`137`	`137`	`maxTimeBetweenHeartBeetsBeforeNodeIsAssumedToBeOffline - timeSinceLastHeartBeat + TimeSpan.FromSeconds(1));`
`138`	`138`
`139`		`- await Try.IgnoringError(async () => await Task.Delay(timeToWait, watchCancellationToken));`
	`139`	`+ await DelayWithoutException.Delay(timeToWait, watchCancellationToken);`
`140`	`140`	`}`
`141`	`141`
`142`	`142`	`log.Write(EventType.Diagnostic, "{0} node watcher cancelled, request {1}", watchingForPulsesFrom, requestActivityId);`