WIP

src/Runner.Common/BrokerServer.cs

@@ -5,6 +5,7 @@ using System.Threading.Tasks;
 using GitHub.Actions.RunService.WebApi;
 using GitHub.DistributedTask.Pipelines;
 using GitHub.DistributedTask.WebApi;
+using GitHub.Runner.Common.Util;
 using GitHub.Runner.Sdk;
 using GitHub.Services.Common;
 using Sdk.RSWebApi.Contracts;
@@ -18,6 +19,7 @@ namespace GitHub.Runner.Common
         Task ConnectAsync(Uri serverUrl, VssCredentials credentials);
 
         Task<TaskAgentMessage> GetRunnerMessageAsync(CancellationToken token, TaskAgentStatus status, string version);
+        Task<bool> DeleteRunnerMessageAsync(string messageID, CancellationToken token);
     }
 
     public sealed class BrokerServer : RunnerService, IBrokerServer
@@ -48,9 +50,20 @@ namespace GitHub.Runner.Common
         {
             CheckConnection();
             var jobMessage = RetryRequest<TaskAgentMessage>(
-                async () => await _brokerHttpClient.GetRunnerMessageAsync(version, status, cancellationToken), cancellationToken);
+                async () => await _brokerHttpClient.GetRunnerMessageAsync(version, status, cancellationToken), cancellationToken
+                );
 
             return jobMessage;
         }
+
+        public Task<bool> DeleteRunnerMessageAsync(string messageID, CancellationToken cancellationToken)
+        {
+            CheckConnection();
+
+            return RetryRequest(
+                async () => await _brokerHttpClient.DeleteRunnerMessageAsync(messageID, cancellationToken),
+                cancellationToken
+            );
+        }
     }
 }

src/Runner.Listener/BrokerMessageListener.cs

@@ -170,7 +170,17 @@ namespace GitHub.Runner.Listener
 
         public async Task DeleteMessageAsync(TaskAgentMessage message)
         {
-            await Task.CompletedTask;
+            Trace.Entering();
+
+            if (MessageUtil.IsRunServiceJob(message.MessageType))
+            {
+                var messageRef = StringUtil.ConvertFromJson<RunnerJobRequestRef>(message.Body);
+
+                using (var cs = new CancellationTokenSource(TimeSpan.FromSeconds(30)))
+                {
+                    await _brokerServer.DeleteRunnerMessageAsync(messageRef.RunnerRequestId, cs.Token);
+                }
+            }
         }
 
         private bool IsGetNextMessageExceptionRetriable(Exception ex)

src/Runner.Listener/Runner.cs

@@ -392,6 +392,7 @@ namespace GitHub.Runner.Listener
                     {
                         TaskAgentMessage message = null;
                         bool skipMessageDeletion = false;
+                        bool useBrokerDeletion = false;
                         try
                         {
                             Task<TaskAgentMessage> getNextMessage = _listener.GetNextMessageAsync(messageQueueLoopTokenSource.Token);
@@ -600,7 +601,15 @@ namespace GitHub.Runner.Listener
                             {
                                 try
                                 {
-                                    await _listener.DeleteMessageAsync(message);
+                                    if (useBrokerDeletion)
+                                    {
+                                        // await _listener.DeleteMessageAsync(brokerDeletionParamsMessageID, brokerDeletionParamsToken);
+                                        await _listener.DeleteMessageAsync(message);
+                                    }
+                                    else
+                                    {
+                                        await _listener.DeleteMessageAsync(message);
+                                    }
                                 }
                                 catch (Exception ex)
                                 {

src/Sdk/WebApi/WebApi/BrokerHttpClient.cs

@@ -88,5 +88,28 @@ namespace GitHub.Actions.RunService.WebApi
 
             throw new Exception($"Failed to get job message: {result.Error}");
         }
+
+        public async Task<bool> DeleteRunnerMessageAsync(
+            string messageID,
+            CancellationToken cancellationToken = default
+        )
+        {
+            var requestUri = new Uri(Client.BaseAddress, "message");
+
+            List<KeyValuePair<string, string>> queryParams = new List<KeyValuePair<string, string>>();
+
+            if (messageID != null)
+            {
+                queryParams.Add("messageID", messageID);
+            }
+
+            var result = await SendAsync<TaskAgentMessage>(
+                new HttpMethod("DELETE"),
+                requestUri: requestUri,
+                queryParameters: queryParams,
+                cancellationToken: cancellationToken);
+
+            return (result.IsSuccess);
+        }
     }
 }