Add TryGetTokenForDataLossDetection method to avoid unnecessary CancellationTokenSource creation and Timer creation on every RPC (#689)

* Add TryGetTokenForDataLossDetection method to avoid unnecessary CancellationTokenSource creation

- Added TryGetTokenForDataLossDetection() method to IWatchForRedisLosingAllItsData interface
- Implemented the method in WatchForRedisLosingAllItsData to return token immediately if monitoring is active
- Updated RedisPendingRequestQueue to use try method first, falling back to GetTokenForDataLossDetection
- Refactored DequeueAsync to use centralized DataLossCancellationToken method instead of duplicating logic
- Updated test implementations (RedisNeverLosesData and CancellableDataLossWatchForRedisLosingAllItsData)
- Improves performance by avoiding CancellationTokenSource allocation when monitoring is already ready
- Maintains backward compatibility with existing GetTokenForDataLossDetection method
- Reduces code duplication by centralizing data loss cancellation token logic

Resolves TODO comment in WatchForRedisLosingAllItsData.cs

* .

* Add happy path test

Author: LukeButters

source/Halibut.Tests/Queue/Redis/RedisDataLoseDetection/WatchForRedisLosingAllItsDataFixture.cs

@@ -24,6 +24,18 @@ public async Task WhenTheConnectionToRedisCanNotBeCreated_WhenAskingForALostData
             await AssertException.Throws<TaskCanceledException>(watcher.GetTokenForDataLossDetection(TimeSpan.FromSeconds(1), CancellationToken));
         }
 
+        [Test]
+        public async Task WhenTheConnectionToRedisCanBeEstablished_AndSomeoneHasAlreadyGotTheCancellationToken_TryGetCancellationTokenReturnsTheCT()
+        {
+            await using var redisFacade = RedisFacadeBuilder.CreateRedisFacade();
+            await using var watcher = new WatchForRedisLosingAllItsData(redisFacade, HalibutLog, watchInterval:TimeSpan.FromSeconds(1));
+
+
+            await watcher.GetTokenForDataLossDetection(TimeSpan.FromSeconds(5), CancellationToken);
+
+            watcher.TryGetTokenForDataLossDetection().Should().NotBeNull();
+        }
+        
         [Test]
         public async Task WhenTheConnectionToRedisIsInitiallyDown_WhenAskingForALostDataCancellationToken_AndTheConnectionToRedisReturns_TheCancellationTokenIsReturned()
         {

source/Halibut.Tests/Queue/Redis/Utils/CancellableDataLossWatchForRedisLosingAllItsData.cs

@@ -37,5 +37,14 @@ public async Task<CancellationToken> GetTokenForDataLossDetection(TimeSpan timeT
             return await TaskCompletionSource.Task;
 #pragma warning restore VSTHRD003
         }
+
+        public CancellationToken? TryGetTokenForDataLossDetection()
+        {
+            if (TaskCompletionSource.Task.IsCompleted && TaskCompletionSource.Task.Status == TaskStatus.RanToCompletion)
+            {
+                return TaskCompletionSource.Task.Result;
+            }
+            return null;
+        }
     }
 }

source/Halibut.Tests/Queue/Redis/Utils/RedisNeverLosesData.cs

@@ -19,6 +19,11 @@ public Task<CancellationToken> GetTokenForDataLossDetection(TimeSpan timeToWait,
             return Task.FromResult(CancellationToken.None);
         }
 
+        public CancellationToken? TryGetTokenForDataLossDetection()
+        {
+            return CancellationToken.None;
+        }
+
         public ValueTask DisposeAsync()
         {
             return ValueTask.CompletedTask;

source/Halibut/Queue/Redis/RedisDataLossDetection/IWatchForRedisLosingAllItsData.cs

@@ -14,5 +14,12 @@ public interface IWatchForRedisLosingAllItsData : IAsyncDisposable
         /// <param name="cancellationToken"></param>
         /// <returns>A cancellation token which is triggered when data loss occurs.</returns>
         Task<CancellationToken> GetTokenForDataLossDetection(TimeSpan timeToWait, CancellationToken cancellationToken);
+        
+        /// <summary>
+        /// Tries to get a cancellation token for data loss detection if monitoring is already active.
+        /// This method returns immediately without waiting.
+        /// </summary>
+        /// <returns>A cancellation token if monitoring is active, null if monitoring is not yet ready.</returns>
+        CancellationToken? TryGetTokenForDataLossDetection();
     }
 }

source/Halibut/Queue/Redis/RedisDataLossDetection/WatchForRedisLosingAllItsData.cs

@@ -43,7 +43,7 @@ public WatchForRedisLosingAllItsData(RedisFacade redisFacade, ILog log, TimeSpan
             var _ = Task.Run(async () => await KeepWatchingForDataLoss(cts.Token));
         }
 
-        private TaskCompletionSource<CancellationToken> taskCompletionSource = new TaskCompletionSource<CancellationToken>();
+        TaskCompletionSource<CancellationToken> taskCompletionSource = new TaskCompletionSource<CancellationToken>();
 
         /// <summary>
         /// Will cause the caller to wait until we are connected to redis and so can detect datalose.
@@ -53,14 +53,30 @@ public WatchForRedisLosingAllItsData(RedisFacade redisFacade, ILog log, TimeSpan
         /// <returns>A cancellation token which is triggered when data lose occurs.</returns>
         public async Task<CancellationToken> GetTokenForDataLossDetection(TimeSpan timeToWait, CancellationToken cancellationToken)
         {
-            if (taskCompletionSource.Task.IsCompleted)
+            var localCopyOfTaskCompletionSource = taskCompletionSource;
+            if (localCopyOfTaskCompletionSource.Task.IsCompleted)
             {
-                return await taskCompletionSource.Task;
+                return await localCopyOfTaskCompletionSource.Task;
             }
 
             await using var cts = new CancelOnDisposeCancellationToken(cancellationToken);
             cts.CancelAfter(timeToWait);
-            return await taskCompletionSource.Task.WaitAsync(cts.Token);
+            return await localCopyOfTaskCompletionSource.Task.WaitAsync(cts.Token);
+        }
+
+        /// <summary>
+        /// Tries to get a cancellation token for data loss detection if monitoring is already active.
+        /// This method returns immediately without waiting.
+        /// </summary>
+        /// <returns>A cancellation token if monitoring is active, null if monitoring is not yet ready.</returns>
+        public CancellationToken? TryGetTokenForDataLossDetection()
+        {
+            var localCopyOfTaskCompletionSource = taskCompletionSource;
+            if (localCopyOfTaskCompletionSource.Task.IsCompleted && localCopyOfTaskCompletionSource.Task.Status == TaskStatus.RanToCompletion)
+            {
+                return localCopyOfTaskCompletionSource.Task.Result;
+            }
+            return null;
         }
 
         private async Task KeepWatchingForDataLoss(CancellationToken cancellationToken)

source/Halibut/Queue/Redis/RedisPendingRequestQueue.cs

@@ -117,6 +117,14 @@ public RedisPendingRequestQueue(
 
         async Task<CancellationToken> DataLossCancellationToken(CancellationToken? cancellationToken)
         {
+            // Try to get the token immediately if monitoring is already active
+            var token = watchForRedisLosingAllItsData.TryGetTokenForDataLossDetection();
+            if (token.HasValue)
+            {
+                return token.Value;
+            }
+            
+            // Fall back to waiting for the token if not immediately available
             await using var cts = new CancelOnDisposeCancellationToken(queueCts.Token, cancellationToken ?? CancellationToken.None);
             return await watchForRedisLosingAllItsData.GetTokenForDataLossDetection(TimeSpan.FromSeconds(30), cts.Token);
         }
@@ -383,7 +391,7 @@ async Task<bool> TryClearRequestFromQueue(RedisPendingRequest redisPending)
             {
                 // There is a chance the data loss occured after we got the data but before here.
                 // In that case we will just time out because of the lack of heart beats.
-                var dataLossCT = await watchForRedisLosingAllItsData.GetTokenForDataLossDetection(TimeSpan.FromSeconds(30), queueToken);
+                var dataLossCT = await DataLossCancellationToken(cancellationToken);
 
                 disposables.AddAsyncDisposable(new NodeHeartBeatSender(
                     endpoint,