diff --git a/src/Runner.Listener/Runner.cs b/src/Runner.Listener/Runner.cs index e20eb6256e5..4c27c36190c 100644 --- a/src/Runner.Listener/Runner.cs +++ b/src/Runner.Listener/Runner.cs @@ -687,7 +687,7 @@ private async Task RunAsync(RunnerSettings settings, bool runOnce = false, else { var messageRef = StringUtil.ConvertFromJson(message.Body); - + // Acknowledge (best-effort) if (messageRef.ShouldAcknowledge) // Temporary feature flag { @@ -695,6 +695,12 @@ private async Task RunAsync(RunnerSettings settings, bool runOnce = false, { await _listener.AcknowledgeMessageAsync(messageRef.RunnerRequestId, messageQueueLoopTokenSource.Token); } + catch (RunnerRequestJobNotFoundException) when (settings.Ephemeral) + { + Trace.Info($"Acknowledge returned job-not-found for ephemeral runner request '{messageRef.RunnerRequestId}'. Exiting runner."); + runOnceJobCompleted = true; + return Constants.Runner.ReturnCode.Success; + } catch (Exception ex) { Trace.Error($"Best-effort acknowledge failed for request '{messageRef.RunnerRequestId}'"); diff --git a/src/Sdk/WebApi/WebApi/BrokerHttpClient.cs b/src/Sdk/WebApi/WebApi/BrokerHttpClient.cs index d4a05526290..256ff60a085 100644 --- a/src/Sdk/WebApi/WebApi/BrokerHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/BrokerHttpClient.cs @@ -256,11 +256,24 @@ public async Task AcknowledgeRunnerRequestAsync( default: break; } + + if (IsAcknowledgeJobNotFound(result.StatusCode, brokerError)) + { + throw new RunnerRequestJobNotFoundException(brokerError.Message); + } } throw new Exception($"Failed to acknowledge runner request. Request to {requestUri} failed with status: {result.StatusCode}. Error message {result.Error}"); } + private static bool IsAcknowledgeJobNotFound(HttpStatusCode statusCode, BrokerError brokerError) + { + return statusCode == HttpStatusCode.NotFound && + brokerError?.StatusCode == (int)HttpStatusCode.NotFound && + !string.Equals(brokerError.ErrorKind, BrokerErrorKind.RunnerNotFound, StringComparison.Ordinal) && + string.Equals(brokerError.Message, "Job not found", StringComparison.OrdinalIgnoreCase); + } + private static bool TryParseErrorBody(string errorBody, out BrokerError error) { if (!string.IsNullOrEmpty(errorBody)) diff --git a/src/Sdk/WebApi/WebApi/Exceptions/RunnerRequestJobNotFoundException.cs b/src/Sdk/WebApi/WebApi/Exceptions/RunnerRequestJobNotFoundException.cs new file mode 100644 index 00000000000..d432f65e9bb --- /dev/null +++ b/src/Sdk/WebApi/WebApi/Exceptions/RunnerRequestJobNotFoundException.cs @@ -0,0 +1,23 @@ +using System; + +namespace GitHub.Services.WebApi +{ + [Serializable] + public sealed class RunnerRequestJobNotFoundException : Exception + { + public RunnerRequestJobNotFoundException() + : base() + { + } + + public RunnerRequestJobNotFoundException(String message) + : base(message) + { + } + + public RunnerRequestJobNotFoundException(String message, Exception innerException) + : base(message, innerException) + { + } + } +} diff --git a/src/Test/L0/Listener/RunnerL0.cs b/src/Test/L0/Listener/RunnerL0.cs index b9aa4c88017..ba7f42e8843 100644 --- a/src/Test/L0/Listener/RunnerL0.cs +++ b/src/Test/L0/Listener/RunnerL0.cs @@ -871,6 +871,202 @@ public async Task TestRunnerJobRequestMessageFromRunService() } } + [Fact] + [Trait("Level", "L0")] + [Trait("Category", "Runner")] + public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnAcknowledgeJobNotFound() + { + using (var hc = new TestHostContext(this)) + { + //Arrange + var runner = new Runner.Listener.Runner(); + hc.SetSingleton(_configurationManager.Object); + hc.SetSingleton(_jobNotification.Object); + hc.SetSingleton(_messageListener.Object); + hc.SetSingleton(_promptManager.Object); + hc.SetSingleton(_runnerServer.Object); + hc.SetSingleton(_configStore.Object); + hc.SetSingleton(_updater.Object); + hc.SetSingleton(_credentialManager.Object); + hc.EnqueueInstance(_acquireJobThrottler.Object); + hc.EnqueueInstance(_runServer.Object); + hc.EnqueueInstance(_jobDispatcher.Object); + + runner.Initialize(hc); + var settings = new RunnerSettings + { + PoolId = 43242, + AgentId = 5678, + Ephemeral = true, + ServerUrl = "https://github.com", + }; + + var message = new TaskAgentMessage() + { + Body = JsonUtility.ToString(new RunnerJobRequestRef() { BillingOwnerId = "github", RunnerRequestId = "999", RunServiceUrl = "https://run-service.com", ShouldAcknowledge = true }), + MessageId = 4234, + MessageType = JobRequestMessageTypes.RunnerJobRequest + }; + + var messages = new Queue(); + messages.Enqueue(message); + _configurationManager.Setup(x => x.LoadSettings()) + .Returns(settings); + _configurationManager.Setup(x => x.IsConfigured()) + .Returns(true); + _messageListener.Setup(x => x.CreateSessionAsync(It.IsAny())) + .Returns(Task.FromResult(CreateSessionResult.Success)); + _messageListener.Setup(x => x.GetNextMessageAsync(It.IsAny())) + .Returns(async (CancellationToken token) => + { + if (0 == messages.Count) + { + await Task.Delay(2000, token); + } + + return messages.Dequeue(); + }); + _messageListener.Setup(x => x.AcknowledgeMessageAsync("999", It.IsAny())) + .ThrowsAsync(new RunnerRequestJobNotFoundException("Job not found")); + _messageListener.Setup(x => x.DeleteSessionAsync()) + .Returns(Task.CompletedTask); + _messageListener.Setup(x => x.DeleteMessageAsync(It.IsAny())) + .Returns(Task.CompletedTask); + _jobNotification.Setup(x => x.StartClient(It.IsAny())) + .Callback(() => + { + + }); + + _configStore.Setup(x => x.IsServiceConfigured()).Returns(false); + + //Act + var command = new CommandSettings(hc, new string[] { "run" }); + Task runnerTask = runner.ExecuteCommand(command); + + //Assert + await Task.WhenAny(runnerTask, Task.Delay(30000)); + + Assert.True(runnerTask.IsCompleted, $"{nameof(runner.ExecuteCommand)} timed out."); + Assert.True(!runnerTask.IsFaulted, runnerTask.Exception?.ToString()); + if (runnerTask.IsCompleted) + { + Assert.Equal(Constants.Runner.ReturnCode.Success, await runnerTask); + } + + _messageListener.Verify(x => x.AcknowledgeMessageAsync("999", It.IsAny()), Times.Once()); + _jobDispatcher.Verify(x => x.Run(It.IsAny(), It.IsAny()), Times.Never()); + _runServer.Verify(x => x.GetJobMessageAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never()); + _credentialManager.Verify(x => x.LoadCredentials(true), Times.Never()); + _messageListener.Verify(x => x.DeleteSessionAsync(), Times.Once()); + _messageListener.Verify(x => x.DeleteMessageAsync(It.IsAny()), Times.Once()); + _configurationManager.Verify(x => x.DeleteLocalRunnerConfig(), Times.Once()); + } + } + + [Fact] + [Trait("Level", "L0")] + [Trait("Category", "Runner")] + [Trait("Level", "L0")] + [Trait("Category", "Runner")] + public async Task TestRunnerJobRequestMessageFromRunServiceContinuesOnAcknowledgeJobNotFoundForPersistentRunner() + { + using (var hc = new TestHostContext(this)) + { + //Arrange + var runner = new Runner.Listener.Runner(); + hc.SetSingleton(_configurationManager.Object); + hc.SetSingleton(_jobNotification.Object); + hc.SetSingleton(_messageListener.Object); + hc.SetSingleton(_promptManager.Object); + hc.SetSingleton(_runnerServer.Object); + hc.SetSingleton(_configStore.Object); + hc.SetSingleton(_updater.Object); + hc.SetSingleton(_credentialManager.Object); + hc.EnqueueInstance(_acquireJobThrottler.Object); + hc.EnqueueInstance(_runServer.Object); + hc.EnqueueInstance(_jobDispatcher.Object); + + runner.Initialize(hc); + var settings = new RunnerSettings + { + PoolId = 43242, + AgentId = 5678, + Ephemeral = false, + ServerUrl = "https://github.com", + }; + + var message = new TaskAgentMessage() + { + Body = JsonUtility.ToString(new RunnerJobRequestRef() { BillingOwnerId = "github", RunnerRequestId = "999", RunServiceUrl = "https://run-service.com", ShouldAcknowledge = true }), + MessageId = 4234, + MessageType = JobRequestMessageTypes.RunnerJobRequest + }; + + var messages = new Queue(); + messages.Enqueue(message); + var signalWorkerStarted = new SemaphoreSlim(0, 1); + _configurationManager.Setup(x => x.LoadSettings()) + .Returns(settings); + _configurationManager.Setup(x => x.IsConfigured()) + .Returns(true); + _messageListener.Setup(x => x.CreateSessionAsync(It.IsAny())) + .Returns(Task.FromResult(CreateSessionResult.Success)); + _messageListener.Setup(x => x.GetNextMessageAsync(It.IsAny())) + .Returns(async (CancellationToken token) => + { + if (0 == messages.Count) + { + await Task.Delay(2000, token); + } + + return messages.Dequeue(); + }); + _messageListener.Setup(x => x.AcknowledgeMessageAsync("999", It.IsAny())) + .ThrowsAsync(new RunnerRequestJobNotFoundException("Job not found")); + _messageListener.Setup(x => x.DeleteSessionAsync()) + .Returns(Task.CompletedTask); + _messageListener.Setup(x => x.DeleteMessageAsync(It.IsAny())) + .Returns(Task.CompletedTask); + _jobNotification.Setup(x => x.StartClient(It.IsAny())) + .Callback(() => + { + + }); + _runServer.Setup(x => x.GetJobMessageAsync("999", "github", It.IsAny())) + .Returns(Task.FromResult(CreateJobRequestMessage("test"))); + _credentialManager.Setup(x => x.LoadCredentials(true)).Returns(new VssCredentials()); + _jobDispatcher.Setup(x => x.Run(It.IsAny(), false)) + .Callback(() => + { + signalWorkerStarted.Release(); + }); + + _configStore.Setup(x => x.IsServiceConfigured()).Returns(false); + + //Act + var command = new CommandSettings(hc, new string[] { "run" }); + Task runnerTask = runner.ExecuteCommand(command); + + //Assert + if (!await signalWorkerStarted.WaitAsync(2000)) + { + Assert.Fail($"{nameof(_jobDispatcher.Object.Run)} was not invoked."); + } + + hc.ShutdownRunner(ShutdownReason.UserCancelled); + await Task.WhenAny(runnerTask, Task.Delay(2000)); + + Assert.True(runnerTask.IsCompleted, $"{nameof(runner.ExecuteCommand)} timed out."); + Assert.True(runnerTask.IsCanceled); + _messageListener.Verify(x => x.AcknowledgeMessageAsync("999", It.IsAny()), Times.Once()); + _jobDispatcher.Verify(x => x.Run(It.IsAny(), false), Times.Once()); + _runServer.Verify(x => x.GetJobMessageAsync("999", "github", It.IsAny()), Times.Once()); + _credentialManager.Verify(x => x.LoadCredentials(true), Times.Once()); + _configurationManager.Verify(x => x.DeleteLocalRunnerConfig(), Times.Never()); + } + } + [Fact] [Trait("Level", "L0")] [Trait("Category", "Runner")] diff --git a/src/Test/L0/Sdk/RSWebApi/BrokerHttpClientL0.cs b/src/Test/L0/Sdk/RSWebApi/BrokerHttpClientL0.cs new file mode 100644 index 00000000000..0af35a72618 --- /dev/null +++ b/src/Test/L0/Sdk/RSWebApi/BrokerHttpClientL0.cs @@ -0,0 +1,75 @@ +using System; +using System.Net; +using System.Net.Http; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using GitHub.DistributedTask.WebApi; +using GitHub.Runner.Sdk; +using GitHub.Services.WebApi; +using Xunit; + +namespace GitHub.Actions.RunService.WebApi.Tests; + +public sealed class BrokerHttpClientL0 +{ + [Fact] + public async Task AcknowledgeRunnerRequestAsyncThrowsRunnerRequestJobNotFoundException() + { + using var client = CreateClient( + HttpStatusCode.NotFound, + new BrokerError + { + Source = "actions-broker-listener", + StatusCode = (int)HttpStatusCode.NotFound, + Message = "Job not found", + }); + + await Assert.ThrowsAsync(() => + client.AcknowledgeRunnerRequestAsync("runner-request", Guid.NewGuid(), "2.0.0", TaskAgentStatus.Online, cancellationToken: CancellationToken.None)); + } + + [Fact] + public async Task AcknowledgeRunnerRequestAsyncKeepsRunnerNotFoundClassification() + { + using var client = CreateClient( + HttpStatusCode.NotFound, + new BrokerError + { + Source = "actions-broker-listener", + ErrorKind = BrokerErrorKind.RunnerNotFound, + StatusCode = (int)HttpStatusCode.NotFound, + Message = "Runner not found", + }); + + await Assert.ThrowsAsync(() => + client.AcknowledgeRunnerRequestAsync("runner-request", Guid.NewGuid(), "2.0.0", TaskAgentStatus.Online, cancellationToken: CancellationToken.None)); + } + + private static BrokerHttpClient CreateClient(HttpStatusCode statusCode, BrokerError brokerError) + { + return new BrokerHttpClient( + new Uri("https://broker.actions.githubusercontent.com/"), + new StaticResponseHandler(new HttpResponseMessage(statusCode) + { + Content = new StringContent(JsonUtility.ToString(brokerError), Encoding.UTF8, "application/json"), + }), + disposeHandler: true); + } + + private sealed class StaticResponseHandler : HttpMessageHandler + { + private readonly HttpResponseMessage _response; + + public StaticResponseHandler(HttpResponseMessage response) + { + _response = response; + } + + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + _response.RequestMessage = request; + return Task.FromResult(_response); + } + } +}