Support retries for long running RPCs. With this We will optionally retry a failed long running file upload/download. (#1147)

* Support retries for long running RPCs

* .

* .

* .

* .

* Update halibut

* .

* .

* .

* .

* .

* .

* .

* .

* .

* .

* .

* .

* .

* .

* .

Author: LukeButters

source/Octopus.Tentacle.Client.Tests/RpcCallRetryHandlerFixture.cs

@@ -7,8 +7,10 @@
 using FluentAssertions;
 using Halibut;
 using Halibut.Exceptions;
+using Halibut.Transport;
 using NUnit.Framework;
 using Octopus.Tentacle.Client.Retries;
+using Octopus.Tentacle.CommonTestUtils.Assertions;
 using Polly.Timeout;
 
 namespace Octopus.Tentacle.Client.Tests
@@ -687,5 +689,184 @@ private TimeSpan GetMinTimeoutDuration(RpcCallRetryHandler handler)
         {
             return handler.RetryTimeout - handler.RetryIfRemainingDurationAtLeast - retryBackoffBuffer;
         }
+
+        [Test]
+        public async Task MinimumAttempts_WhenSetTo1_DoesNotRetryAfterTimeoutExceeded()
+        {
+            var expectedResult = Guid.NewGuid();
+            var callCount = 0;
+            var onRetryActionCalled = false;
+            var onTimeoutActionCalled = false;
+
+            // Short timeout to ensure it's exceeded, but minimumAttemptsForInterruptedLongRunningCalls = 1
+            var handler = new RpcCallRetryHandler(TimeSpan.FromSeconds(1), minimumAttemptsForInterruptedLongRunningCalls: 1);
+
+            var result = await handler.ExecuteWithRetries(
+                async ct =>
+                {
+                    callCount++;
+                    
+                    // Sleep for 3 seconds to simulate a long-running operation
+                    // must exceed retry duration
+                    await Task.Delay(TimeSpan.FromSeconds(3), ct);
+                    
+                    // Succeed on the first (and only) attempt since minimumAttemptsForInterruptedLongRunningCalls = 1
+                    if (callCount == 1)
+                    {
+                        return expectedResult;
+                    }
+                    
+                    throw new HalibutClientException($"An error has occurred on attempt {callCount}");
+                },
+                onRetryAction: async (_, _, _, _, _, _) =>
+                {
+                    await Task.CompletedTask;
+                    onRetryActionCalled = true;
+                },
+                onTimeoutAction: async (_, _, _, _) =>
+                {
+                    await Task.CompletedTask;
+                    onTimeoutActionCalled = true;
+                },
+                CancellationToken.None);
+
+            // With minimumAttemptsForInterruptedLongRunningCalls = 1, we should only make one attempt (no retries)
+            result.Should().Be(expectedResult);
+            callCount.Should().Be(1);
+            onRetryActionCalled.Should().BeFalse();
+            onTimeoutActionCalled.Should().BeFalse();
+        }
+
+        [Test]
+        public async Task MinimumAttempts_WhenSetTo2_MakesOneRetryEvenAfterTimeoutExceeded()
+        {
+            var expectedResult = Guid.NewGuid();
+            var callCount = 0;
+            var onRetryActionCalled = false;
+            var onTimeoutActionCalled = false;
+
+            // Short timeout to ensure it's exceeded, but minimumAttemptsForInterruptedLongRunningCalls = 2
+            var handler = new RpcCallRetryHandler(TimeSpan.FromSeconds(5), minimumAttemptsForInterruptedLongRunningCalls: 2);
+
+            var result = await handler.ExecuteWithRetries(
+                async ct =>
+                {
+                    callCount++;
+                    
+                    // Delay 2 seconds to ensure the ct doesn't get canceled.
+                    await Task.Delay(TimeSpan.FromSeconds(2), ct);
+                    
+                    // Fail on first attempt, succeed on second
+                    if (callCount == 1)
+                    {
+                        throw new HalibutClientException($"An error has occurred on attempt {callCount}");
+                    }
+                    
+                    return expectedResult;
+                },
+                onRetryAction: async (_, _, _, _, _, _) =>
+                {
+                    await Task.CompletedTask;
+                    onRetryActionCalled = true;
+                },
+                onTimeoutAction: async (_, _, _, _) =>
+                {
+                    await Task.CompletedTask;
+                    onTimeoutActionCalled = true;
+                },
+                CancellationToken.None);
+
+            // With minimumAttemptsForInterruptedLongRunningCalls = 2, we should make 2 attempts (1 initial + 1 retry)
+            result.Should().Be(expectedResult);
+            callCount.Should().Be(2);
+            onRetryActionCalled.Should().BeTrue();
+            onTimeoutActionCalled.Should().BeFalse(); // Should succeed before timeout
+        }
+
+        [Test]
+        public async Task MinimumAttempts_WhenSetTo3_MakesTwoRetriesEvenAfterTimeoutExceeded()
+        {
+            var expectedResult = Guid.NewGuid();
+            var callCount = 0;
+            var retryCount = 0;
+            var onTimeoutActionCalled = false;
+
+            // Short timeout to ensure it's exceeded, but minimumAttemptsForInterruptedLongRunningCalls = 3
+            var handler = new RpcCallRetryHandler(TimeSpan.FromSeconds(8), minimumAttemptsForInterruptedLongRunningCalls: 3);
+
+            var result = await handler.ExecuteWithRetries(
+                async ct =>
+                {
+                    callCount++;
+                    
+                    // Always add 2 second delay as requested
+                    await Task.Delay(TimeSpan.FromSeconds(2), ct);
+                    
+                    // Fail on first two attempts, succeed on third
+                    if (callCount <= 2)
+                    {
+                        throw new HalibutClientException($"An error has occurred on attempt {callCount}");
+                    }
+                    
+                    return expectedResult;
+                },
+                onRetryAction: async (_, _, _, _, _, _) =>
+                {
+                    await Task.CompletedTask;
+                    retryCount++;
+                },
+                onTimeoutAction: async (_, _, _, _) =>
+                {
+                    await Task.CompletedTask;
+                    onTimeoutActionCalled = true;
+                },
+                CancellationToken.None);
+
+            // With minimumAttemptsForInterruptedLongRunningCalls = 3, we should make 3 attempts (1 initial + 2 retries)
+            result.Should().Be(expectedResult);
+            callCount.Should().Be(3);
+            retryCount.Should().Be(2);
+            onTimeoutActionCalled.Should().BeFalse(); // Should succeed before timeout
+        }
+        
+        /// <summary>
+        /// Connection error means the tentacle did not get the request, which means the tentacle is considered to be offline.
+        /// This shows that we won't exceed the retry duration attempting to connect to an offline tentacle to meet the
+        /// minimumAttemptsForInterruptedLongRunningCalls count.
+        /// </summary>
+        [Test]
+        public async Task WhenConfiguredToMakeAMinimumNumberOfAttempts_AndTheFirstAttemptExceedsTheRetryDuration_AndTheFailureIsAConnectingFailure_ARetryIsNotMade()
+        {
+            var callCount = 0;
+
+            // Short timeout to ensure it's exceeded, but minimumAttemptsForInterruptedLongRunningCalls = 2
+            var handler = new RpcCallRetryHandler(TimeSpan.FromSeconds(2), minimumAttemptsForInterruptedLongRunningCalls: 9999);
+            
+            var exception = await AssertThrowsAny.Exception(async () =>
+                await handler.ExecuteWithRetries(
+                    async ct =>
+                    {
+                        callCount++;
+                        
+                        // Must exceed retry duration
+                        await Task.Delay(TimeSpan.FromSeconds(5), ct);
+                        
+                        if(callCount == -1) return Guid.NewGuid(); // Never called used to make the typing work.
+
+                        throw new HalibutClientException("", new Exception(), ConnectionState.Connecting);
+                    },
+                    onRetryAction: async (_, _, _, _, _, _) =>
+                    {
+                        await Task.CompletedTask;
+                    },
+                    onTimeoutAction: async (_, _, _, _) =>
+                    {
+                        await Task.CompletedTask;
+                    },
+                    CancellationToken.None));
+            exception.Should().BeAssignableTo<HalibutClientException>();
+            
+            callCount.Should().Be(1);
+        }
     }
 }

source/Octopus.Tentacle.Client/Execution/RpcCallExecutor.cs

@@ -84,7 +84,7 @@ public async Task<T> ExecuteWithRetries<T>(
 
                             onErrorAction?.Invoke(lastException);
 
-                            var remainingDurationInSeconds = (int)(totalRetryDuration - elapsedDuration).TotalSeconds;
+                            var remainingDurationInSeconds = Math.Max((int)(totalRetryDuration - elapsedDuration).TotalSeconds, 0);
                             logger.Info($"An error occurred communicating with Tentacle. This action will be retried after {(int)sleepDuration.TotalSeconds} seconds. Retry attempt {retryCount}. Retries will be performed for up to {remainingDurationInSeconds} seconds.");
                             logger.Verbose(lastException);
                         },

source/Octopus.Tentacle.Client/Execution/RpcCallExecutorFactory.cs

@@ -6,9 +6,9 @@ namespace Octopus.Tentacle.Client.Execution
 {
     internal static class RpcCallExecutorFactory
     {
-        internal static RpcCallExecutor Create(TimeSpan retryDuration, ITentacleClientObserver tentacleClientObserver)
+        internal static RpcCallExecutor Create(TimeSpan retryDuration, ITentacleClientObserver tentacleClientObserver, int? minimumAttemptsForInterruptedLongRunningCalls = null)
         {
-            var rpcCallRetryHandler = new RpcCallRetryHandler(retryDuration);
+            var rpcCallRetryHandler = new RpcCallRetryHandler(retryDuration, minimumAttemptsForInterruptedLongRunningCalls);
             var rpcCallExecutor = new RpcCallExecutor(rpcCallRetryHandler, tentacleClientObserver);
 
             return rpcCallExecutor;

source/Octopus.Tentacle.Client/Retries/RpcCallRetryHandler.cs

@@ -24,13 +24,27 @@ internal class RpcCallRetryHandler
 
         public delegate Task OnTimeoutAction(TimeSpan retryTimeout, TimeSpan elapsedDuration, int retryCount, CancellationToken cancellationToken);
 
-        public RpcCallRetryHandler(TimeSpan retryTimeout)
+        public RpcCallRetryHandler(TimeSpan retryTimeout, int? minimumAttemptsForInterruptedLongRunningCalls = null)
         {
             RetryTimeout = retryTimeout;
+            MinimumAttemptsForInterruptedLongRunningCalls = minimumAttemptsForInterruptedLongRunningCalls;
         }
 
         public TimeSpan RetryTimeout { get; }
 
+        /// <summary>
+        /// This is the minimum number of attempts that will be made to long-running calls,
+        /// that are interrupted while executing.
+        /// For example, if a long-running file upload is interrupted after ten minutes of transfering,
+        /// and this is set to 2, we will make another attempt to upload the file even if the RetryTimeout
+        /// is exceeded.
+        /// If this is set to 9999, and a polling tentacle did not collect the file upload request from the queue,
+        /// we will not continue to make attempts until we have made 9999 attempts. This is because the RPC call
+        /// is not considered to be interrupted as the RPC call never started. This means we won't try 9999 times
+        /// to send a request to a tentacle that is not connected.
+        /// </summary>
+        public int? MinimumAttemptsForInterruptedLongRunningCalls { get; }
+
         public TimeSpan RetryIfRemainingDurationAtLeast { get; } = TimeSpan.FromSeconds(1);
 
         public async Task<T> ExecuteWithRetries<T>(
@@ -43,9 +57,18 @@ public async Task<T> ExecuteWithRetries<T>(
             var started = new Stopwatch();
             var nextSleepDuration = TimeSpan.Zero;
             var totalRetryCount = 0;
+            var totalAttemptCount = 0;
+            bool shouldExecuteNextRetryUnderTimeout = true;
 
             async Task OnRetryAction(Exception exception, TimeSpan sleepDuration, int retryCount, Context context)
             {
+                // If tentacle was online (by virtue of NOT getting a connecting exception) AND we have been told
+                // to make a min number of attempts, then our next attempt (if any) should be done without a timeout.
+                // However, if tentacle was to be offline, then we would want to revert to retries being done under a timeout
+                // so that we limit how long we try to communicate with a tentacle offline.
+                // We must re-evaluate this each failed attempt, since the failure type can change on each attempt.
+                shouldExecuteNextRetryUnderTimeout = !(MinimumAttemptsForInterruptedLongRunningCalls.HasValue && !exception.IsConnectionException());
+                
                 if (lastException == null)
                 {
                     lastException = exception;
@@ -62,7 +85,7 @@ async Task OnRetryAction(Exception exception, TimeSpan sleepDuration, int retryC
                 var elapsedDuration = started.Elapsed;
                 var remainingRetryDuration = RetryTimeout - elapsedDuration - sleepDuration;
 
-                if (ShouldRetryWithRemainingDuration(remainingRetryDuration))
+                if (ShouldRetry(remainingRetryDuration, totalAttemptCount, !shouldExecuteNextRetryUnderTimeout))
                 {
                     totalRetryCount = retryCount;
 
@@ -96,25 +119,37 @@ async Task<T> ExecuteAction(CancellationToken ct)
             {
                 if (isInitialAction)
                 {
+                    totalAttemptCount++;
                     isInitialAction = false;
                     return await action(ct).ConfigureAwait(false);
                 }
 
                 var remainingRetryDuration = RetryTimeout - started.Elapsed - nextSleepDuration;
 
-                if (!ShouldRetryWithRemainingDuration(remainingRetryDuration))
+                if (!ShouldRetry(remainingRetryDuration, totalAttemptCount, !shouldExecuteNextRetryUnderTimeout))
                 {
-                    // We are short circuiting as the retry duration has elapsed
+                    // We are short circuiting as the retry duration has elapsed and minimum attempts have been satisfied
                     await OnTimeoutAction(null, RetryTimeout, null, null).ConfigureAwait(false);
                     throw new TimeoutRejectedException("The delegate executed asynchronously through TimeoutPolicy did not complete within the timeout.");
                 }
 
-                var timeoutPolicy = policyBuilder
-                    // Ensure the remaining retry time excludes the elapsed time
-                    .WithRetryTimeout(remainingRetryDuration)
-                    .BuildTimeoutPolicy();
-
-                return await timeoutPolicy.ExecuteAsync(action, ct).ConfigureAwait(false);
+                totalAttemptCount++;
+                
+                if (!shouldExecuteNextRetryUnderTimeout)
+                {
+                    // Retry a long-running operation with no timeout
+                    return await action(ct);
+                }
+                else
+                {
+                    var timeoutPolicy = policyBuilder
+                        // Ensure the remaining retry time excludes the elapsed time
+                        .WithRetryTimeout(remainingRetryDuration)
+                        .BuildTimeoutPolicy();
+                
+                
+                    return await timeoutPolicy.ExecuteAsync(action, ct).ConfigureAwait(false);
+                }
             }
 
             try
@@ -135,8 +170,12 @@ async Task<T> ExecuteAction(CancellationToken ct)
                 throw;
             }
 
-            bool ShouldRetryWithRemainingDuration(TimeSpan remainingRetryDuration)
+            bool ShouldRetry(TimeSpan remainingRetryDuration, int currentAttemptCount, bool retryRegardlessOfTimeout)
             {
+                if (MinimumAttemptsForInterruptedLongRunningCalls.HasValue
+                    && currentAttemptCount < MinimumAttemptsForInterruptedLongRunningCalls.Value
+                    && retryRegardlessOfTimeout) return true;
+                
                 return remainingRetryDuration > RetryIfRemainingDurationAtLeast;
             }
         }

source/Octopus.Tentacle.Client/Retries/RpcCallRetryPolicyBuilder.cs

@@ -66,16 +66,5 @@ async Task DefaultOnRetryAction(Exception exception, TimeSpan sleepDuration, int
                 await Task.CompletedTask;
             }
         }
-
-        public AsyncPolicy Build()
-        {
-            var timeoutPolicy = BuildTimeoutPolicy();
-
-            var handleAndRetryPolicy = BuildRetryPolicy();
-
-            var policyWrap = Policy.WrapAsync(timeoutPolicy, handleAndRetryPolicy);
-
-            return policyWrap!;
-        }
     }
 }

source/Octopus.Tentacle.Client/TentacleClient.cs

@@ -79,6 +79,10 @@ internal TentacleClient(
 
         public TimeSpan OnCancellationAbandonCompleteScriptAfter { get; set; } = TimeSpan.FromMinutes(1);
 
+        // Created on the fly since, most of the time we don't need this executor.
+        RpcCallExecutor FileTransferRpcCallExecutor => 
+            RpcCallExecutorFactory.Create(this.clientOptions.RpcRetrySettings.RetryDuration, this.tentacleClientObserver, clientOptions.MinimumAttemptsForInterruptedLongRunningCalls);
+
         public async Task<UploadResult> UploadFile(string fileName, string path, DataStream package, ITentacleClientTaskLog logger, CancellationToken cancellationToken)
         {
             var operationMetricsBuilder = ClientOperationMetricsBuilder.Start();
@@ -94,7 +98,7 @@ async Task<UploadResult> UploadFileAction(CancellationToken ct)
 
             try
             {
-                return await rpcCallExecutor.Execute(
+                return await FileTransferRpcCallExecutor.Execute(
                     retriesEnabled: clientOptions.RpcRetrySettings.RetriesEnabled,
                     RpcCall.Create<IFileTransferService>(nameof(IFileTransferService.UploadFile)),
                     UploadFileAction,
@@ -129,7 +133,7 @@ async Task<DataStream> DownloadFileAction(CancellationToken ct)
 
             try
             {
-                return await rpcCallExecutor.Execute(
+                return await FileTransferRpcCallExecutor.Execute(
                     retriesEnabled: clientOptions.RpcRetrySettings.RetriesEnabled,
                     RpcCall.Create<IFileTransferService>(nameof(IFileTransferService.DownloadFile)),
                     DownloadFileAction,

source/Octopus.Tentacle.Client/TentacleClientOptions.cs

@@ -6,10 +6,13 @@ namespace Octopus.Tentacle.Client
     public class TentacleClientOptions
     {
         public RpcRetrySettings RpcRetrySettings { get; }
+        
+        public int? MinimumAttemptsForInterruptedLongRunningCalls { get; }
 
-        public TentacleClientOptions(RpcRetrySettings rpcRetrySettings)
+        public TentacleClientOptions(RpcRetrySettings rpcRetrySettings, int? minimumAttemptsForInterruptedLongRunningCalls = null)
         {
             RpcRetrySettings = rpcRetrySettings;
+            MinimumAttemptsForInterruptedLongRunningCalls = minimumAttemptsForInterruptedLongRunningCalls;
         }
     }
 }