handle conflict errors from run service (#2570)

* handle conflict errors from run service

* nit: formatting

* fix formatting

src/Runner.Common/RunServer.cs

@@ -52,7 +52,8 @@ namespace GitHub.Runner.Common
         {
             CheckConnection();
             return RetryRequest<AgentJobRequestMessage>(
-                async () => await _runServiceHttpClient.GetJobMessageAsync(requestUri, id, cancellationToken), cancellationToken);
+                async () => await _runServiceHttpClient.GetJobMessageAsync(requestUri, id, cancellationToken), cancellationToken,
+                shouldRetry: ex => ex is not TaskOrchestrationJobAlreadyAcquiredException);
         }
 
         public Task CompleteJobAsync(Guid planId, Guid jobId, TaskResult result, Dictionary<String, VariableValue> outputs, IList<StepResult> stepResults, CancellationToken cancellationToken)

src/Runner.Common/RunnerService.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using GitHub.Runner.Sdk;
@@ -83,7 +83,8 @@ namespace GitHub.Runner.Common
 
         protected async Task<T> RetryRequest<T>(Func<Task<T>> func,
             CancellationToken cancellationToken,
-            int maxRetryAttemptsCount = 5
+            int maxRetryAttemptsCount = 5,
+            Func<Exception, bool> shouldRetry = null
         )
         {
             var retryCount = 0;
@@ -96,7 +97,7 @@ namespace GitHub.Runner.Common
                     return await func();
                 }
                 // TODO: Add handling of non-retriable exceptions: https://github.com/github/actions-broker/issues/122
-                catch (Exception ex) when (retryCount < maxRetryAttemptsCount)
+                catch (Exception ex) when (retryCount < maxRetryAttemptsCount && (shouldRetry == null || shouldRetry(ex)))
                 {
                     Trace.Error("Catch exception during request");
                     Trace.Error(ex);

src/Runner.Listener/Runner.cs

@@ -549,7 +549,17 @@ namespace GitHub.Runner.Listener
                                     {
                                         var runServer = HostContext.CreateService<IRunServer>();
                                         await runServer.ConnectAsync(new Uri(messageRef.RunServiceUrl), creds);
-                                        jobRequestMessage = await runServer.GetJobMessageAsync(messageRef.RunnerRequestId, messageQueueLoopTokenSource.Token);
+                                        try
+                                        {
+                                            jobRequestMessage =
+                                            await runServer.GetJobMessageAsync(messageRef.RunnerRequestId,
+                                            messageQueueLoopTokenSource.Token);
+                                        }
+                                        catch (TaskOrchestrationJobAlreadyAcquiredException)
+                                        {
+                                            Trace.Info("Job is already acquired, skip this message.");
+                                            continue;
+                                        }
                                     }
 
                                     jobDispatcher.Run(jobRequestMessage, runOnce);

src/Sdk/DTWebApi/WebApi/Exceptions.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Runtime.Serialization;
 using GitHub.Services.Common;
 
@@ -1519,6 +1519,26 @@ namespace GitHub.DistributedTask.WebApi
         }
     }
 
+    [Serializable]
+    [ExceptionMapping("0.0", "3.0", "TaskOrchestrationJobAlreadyAcquiredException", "GitHub.DistributedTask.WebApi.TaskOrchestrationJobAlreadyAcquiredException, GitHub.DistributedTask.WebApi, Version=14.0.0.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a")]
+    public sealed class TaskOrchestrationJobAlreadyAcquiredException : DistributedTaskException
+    {
+        public TaskOrchestrationJobAlreadyAcquiredException(String message)
+            : base(message)
+        {
+        }
+
+        public TaskOrchestrationJobAlreadyAcquiredException(String message, Exception innerException)
+            : base(message, innerException)
+        {
+        }
+
+        private TaskOrchestrationJobAlreadyAcquiredException(SerializationInfo info, StreamingContext context)
+            : base(info, context)
+        {
+        }
+    }
+
     [Serializable]
     [ExceptionMapping("0.0", "3.0", "TaskOrchestrationPlanSecurityException", "GitHub.DistributedTask.WebApi.TaskOrchestrationPlanSecurityException, GitHub.DistributedTask.WebApi, Version=14.0.0.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a")]
     public sealed class TaskOrchestrationPlanSecurityException : DistributedTaskException

src/Sdk/RSWebApi/RunServiceHttpClient.cs

@@ -86,6 +86,8 @@ namespace GitHub.Actions.RunService.WebApi
             {
                 case HttpStatusCode.NotFound:
                     throw new TaskOrchestrationJobNotFoundException($"Job message not found: {messageId}");
+                case HttpStatusCode.Conflict:
+                    throw new TaskOrchestrationJobAlreadyAcquiredException($"Job message already acquired: {messageId}");
                 default:
                     throw new Exception($"Failed to get job message: {result.Error}");
             }