From 7777449fad7ffc1bb8afd5506dadcff5dfbd4678 Mon Sep 17 00:00:00 2001 From: LouisCuvelier Date: Fri, 7 Aug 2026 14:29:53 +0200 Subject: [PATCH 1/3] Exit ephemeral runner when its job assignment is lost An ephemeral runner that failed to acquire an assigned job (404/409/422) returned to its message loop. The service consumes an ephemeral runner's registration when it assigns the job, so the runner stayed alive and deregistered, never to be assigned work again. Only a restart recovered it, and nothing surfaced the failure: the process kept reporting "Listening for Jobs". Mirror the existing handling of a lost acknowledge and exit with success. The finally block then deletes the session and the local config, so the supervising process can register a fresh runner. Co-Authored-By: Claude Opus 5 (1M context) --- src/Runner.Listener/Runner.cs | 12 ++++ src/Test/L0/Listener/RunnerL0.cs | 94 ++++++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+) diff --git a/src/Runner.Listener/Runner.cs b/src/Runner.Listener/Runner.cs index 31c87130257..3061675c28a 100644 --- a/src/Runner.Listener/Runner.cs +++ b/src/Runner.Listener/Runner.cs @@ -741,6 +741,18 @@ ex is TaskOrchestrationJobAlreadyAcquiredException || // HTTP status 409 ex is TaskOrchestrationJobUnprocessableException) // HTTP status 422 { Trace.Info($"Skipping message Job. {ex.Message}"); + + // The service consumes an ephemeral runner's registration when it assigns + // the job. Once that assignment is lost there is no session left to listen + // on, so skipping would leave the runner alive but deregistered, never to + // be assigned work again. + if (settings.Ephemeral) + { + Trace.Info("Ephemeral runner lost its job assignment. Exiting runner."); + runOnceJobCompleted = true; + return Constants.Runner.ReturnCode.Success; + } + await _acquireJobThrottler.IncrementAndWaitAsync(messageQueueLoopTokenSource.Token); continue; } diff --git a/src/Test/L0/Listener/RunnerL0.cs b/src/Test/L0/Listener/RunnerL0.cs index fcf442244f3..122a5562934 100644 --- a/src/Test/L0/Listener/RunnerL0.cs +++ b/src/Test/L0/Listener/RunnerL0.cs @@ -1056,6 +1056,100 @@ public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnAckno } } + [Fact] + [Trait("Level", "L0")] + [Trait("Category", "Runner")] + public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnJobAlreadyAcquired() + { + 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" }), + 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.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())) + .ThrowsAsync(new TaskOrchestrationJobAlreadyAcquiredException("Job message already acquired '999'. job assignment is invalid: MissingKey")); + + _credentialManager.Setup(x => x.LoadCredentials(true)).Returns(new VssCredentials()); + + _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); + } + + _runServer.Verify(x => x.GetJobMessageAsync("999", "github", It.IsAny()), Times.Once()); + _jobDispatcher.Verify(x => x.Run(It.IsAny(), It.IsAny()), Times.Never()); + _acquireJobThrottler.Verify(x => x.IncrementAndWaitAsync(It.IsAny()), 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")] From 43a75d969e9b36f6f61e3e56f31c727396354cf4 Mon Sep 17 00:00:00 2001 From: LouisCuvelier Date: Fri, 7 Aug 2026 14:55:23 +0200 Subject: [PATCH 2/3] Cover all three lost-assignment errors and fix the skip log The catch handles 404, 409 and 422, so test the ephemeral exit for all three rather than 409 alone, and add the persistent-runner counterpart asserting skip-and-retry still holds on the acquire path. Also stop logging "Skipping message Job." before a branch that exits instead of skipping. Co-Authored-By: Claude Opus 5 (1M context) --- src/Runner.Listener/Runner.cs | 5 +- src/Test/L0/Listener/RunnerL0.cs | 110 ++++++++++++++++++++++++++++++- 2 files changed, 109 insertions(+), 6 deletions(-) diff --git a/src/Runner.Listener/Runner.cs b/src/Runner.Listener/Runner.cs index 3061675c28a..ea1d1e1bcb2 100644 --- a/src/Runner.Listener/Runner.cs +++ b/src/Runner.Listener/Runner.cs @@ -740,19 +740,18 @@ ex is TaskOrchestrationJobNotFoundException || // HTTP status 404 ex is TaskOrchestrationJobAlreadyAcquiredException || // HTTP status 409 ex is TaskOrchestrationJobUnprocessableException) // HTTP status 422 { - Trace.Info($"Skipping message Job. {ex.Message}"); - // The service consumes an ephemeral runner's registration when it assigns // the job. Once that assignment is lost there is no session left to listen // on, so skipping would leave the runner alive but deregistered, never to // be assigned work again. if (settings.Ephemeral) { - Trace.Info("Ephemeral runner lost its job assignment. Exiting runner."); + Trace.Info($"Ephemeral runner lost its job assignment. Exiting runner. {ex.Message}"); runOnceJobCompleted = true; return Constants.Runner.ReturnCode.Success; } + Trace.Info($"Skipping message Job. {ex.Message}"); await _acquireJobThrottler.IncrementAndWaitAsync(messageQueueLoopTokenSource.Token); continue; } diff --git a/src/Test/L0/Listener/RunnerL0.cs b/src/Test/L0/Listener/RunnerL0.cs index 122a5562934..82d6f953211 100644 --- a/src/Test/L0/Listener/RunnerL0.cs +++ b/src/Test/L0/Listener/RunnerL0.cs @@ -1056,10 +1056,13 @@ public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnAckno } } - [Fact] + [Theory] + [InlineData(typeof(TaskOrchestrationJobNotFoundException))] // HTTP status 404 + [InlineData(typeof(TaskOrchestrationJobAlreadyAcquiredException))] // HTTP status 409 + [InlineData(typeof(TaskOrchestrationJobUnprocessableException))] // HTTP status 422 [Trait("Level", "L0")] [Trait("Category", "Runner")] - public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnJobAlreadyAcquired() + public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnLostJobAssignment(Type acquireException) { using (var hc = new TestHostContext(this)) { @@ -1121,7 +1124,7 @@ public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnJobAl }); _runServer.Setup(x => x.GetJobMessageAsync("999", "github", It.IsAny())) - .ThrowsAsync(new TaskOrchestrationJobAlreadyAcquiredException("Job message already acquired '999'. job assignment is invalid: MissingKey")); + .ThrowsAsync((Exception)Activator.CreateInstance(acquireException, "Job assignment is invalid: MissingKey")); _credentialManager.Setup(x => x.LoadCredentials(true)).Returns(new VssCredentials()); @@ -1150,6 +1153,107 @@ public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnJobAl } } + [Fact] + [Trait("Level", "L0")] + [Trait("Category", "Runner")] + public async Task TestRunnerJobRequestMessageFromRunServiceContinuesOnLostJobAssignmentForPersistentRunner() + { + 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" }), + MessageId = 4234, + MessageType = JobRequestMessageTypes.RunnerJobRequest + }; + + var messages = new Queue(); + messages.Enqueue(message); + var signalThrottled = 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.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())) + .ThrowsAsync(new TaskOrchestrationJobAlreadyAcquiredException("Job assignment is invalid: MissingKey")); + _acquireJobThrottler.Setup(x => x.IncrementAndWaitAsync(It.IsAny())) + .Returns(Task.CompletedTask) + .Callback(() => + { + signalThrottled.Release(); + }); + + _credentialManager.Setup(x => x.LoadCredentials(true)).Returns(new VssCredentials()); + + _configStore.Setup(x => x.IsServiceConfigured()).Returns(false); + + //Act + var command = new CommandSettings(hc, new string[] { "run" }); + Task runnerTask = runner.ExecuteCommand(command); + + //Assert + //the runner skips the job and keeps listening, so it only stops when we shut it down + if (!await signalThrottled.WaitAsync(2000)) + { + Assert.Fail($"{nameof(_acquireJobThrottler.Object.IncrementAndWaitAsync)} 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); + _runServer.Verify(x => x.GetJobMessageAsync("999", "github", It.IsAny()), Times.Once()); + _acquireJobThrottler.Verify(x => x.IncrementAndWaitAsync(It.IsAny()), Times.Once()); + _jobDispatcher.Verify(x => x.Run(It.IsAny(), It.IsAny()), Times.Never()); + _configurationManager.Verify(x => x.DeleteLocalRunnerConfig(), Times.Never()); + } + } + [Fact] [Trait("Level", "L0")] [Trait("Category", "Runner")] From 5e7b5e7fbab0d9377e7b01af474831e24d2884c0 Mon Sep 17 00:00:00 2001 From: LouisCuvelier Date: Fri, 7 Aug 2026 16:11:01 +0200 Subject: [PATCH 3/3] Restrict the ephemeral exit to a lost assignment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 422 means the job is unprocessable, not that the assignment moved, so exiting on it would delete the local config of a runner that could still serve — and run-helper.sh reads exit 0 as final, so the host would never come back. Exit on 404 and 409 only. Set skipSessionDeletion before returning: the registration is already consumed, so DeleteSessionAsync stalls on a call the service rejects, and its failure escapes the finally whenever runOnce and settings.Ephemeral disagree. Write the reason to the terminal, since a trace-only exit leaves the operator with a bare exit 0. On the tests: park the message-loop mock instead of dequeuing an empty queue, so a regression surfaces as the assertion rather than "Queue empty"; name the host context per theory row, which otherwise overwrite each other's trace file; and cover the continue path for 422, for a persistent runner and for --once, pinning both predicates. Co-Authored-By: Claude Opus 5 (1M context) --- src/Runner.Listener/Runner.cs | 16 ++- src/Test/L0/Listener/RunnerL0.cs | 240 +++++++++++++------------------ 2 files changed, 108 insertions(+), 148 deletions(-) diff --git a/src/Runner.Listener/Runner.cs b/src/Runner.Listener/Runner.cs index ea1d1e1bcb2..2cd9e9aea47 100644 --- a/src/Runner.Listener/Runner.cs +++ b/src/Runner.Listener/Runner.cs @@ -740,13 +740,19 @@ ex is TaskOrchestrationJobNotFoundException || // HTTP status 404 ex is TaskOrchestrationJobAlreadyAcquiredException || // HTTP status 409 ex is TaskOrchestrationJobUnprocessableException) // HTTP status 422 { - // The service consumes an ephemeral runner's registration when it assigns - // the job. Once that assignment is lost there is no session left to listen - // on, so skipping would leave the runner alive but deregistered, never to - // be assigned work again. - if (settings.Ephemeral) + // 404 and 409 mean the assignment is gone, and the service consumes an + // ephemeral runner's registration when it assigns the job, so there is no + // session left to listen on: skipping would leave the runner alive but + // deregistered, never assigned work again. 422 says the job itself is + // unprocessable, not that the assignment moved, so it keeps skipping. + if (settings.Ephemeral && + (ex is TaskOrchestrationJobNotFoundException || ex is TaskOrchestrationJobAlreadyAcquiredException)) { + _term.WriteLine("The job assigned to this ephemeral runner is no longer available. Cleaning up local configuration."); Trace.Info($"Ephemeral runner lost its job assignment. Exiting runner. {ex.Message}"); + // The registration is already gone, so deleting the session would only + // stall on a call the service is bound to reject. + skipSessionDeletion = true; runOnceJobCompleted = true; return Constants.Runner.ReturnCode.Success; } diff --git a/src/Test/L0/Listener/RunnerL0.cs b/src/Test/L0/Listener/RunnerL0.cs index 82d6f953211..8c7cede6f51 100644 --- a/src/Test/L0/Listener/RunnerL0.cs +++ b/src/Test/L0/Listener/RunnerL0.cs @@ -1056,80 +1056,93 @@ public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnAckno } } + // Shared arrange for the RunService acquire-path tests. The message loop parks instead of + // dequeuing an empty queue: a bare Dequeue surfaces a regression as "Queue empty" thrown + // from the mock, which hides the assertion that was meant to catch it. + private (Queue Messages, TaskCompletionSource Drained) ArrangeRunServiceRunner(TestHostContext hc, Runner.Listener.Runner runner, bool ephemeral) + { + 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 = ephemeral, + ServerUrl = "https://github.com", + }; + + var messages = new Queue(); + messages.Enqueue(new TaskAgentMessage() + { + Body = JsonUtility.ToString(new RunnerJobRequestRef() { BillingOwnerId = "github", RunnerRequestId = "999", RunServiceUrl = "https://run-service.com" }), + MessageId = 4234, + MessageType = JobRequestMessageTypes.RunnerJobRequest + }); + + // Completed once the runner is back asking for work, which means the message loop has + // already re-checked its cancellation token. Signalling any earlier races the shutdown. + var drained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + _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) + { + drained.TrySetResult(true); + await Task.Delay(Timeout.Infinite, token); + } + + return messages.Dequeue(); + }); + _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(() => + { + + }); + _credentialManager.Setup(x => x.LoadCredentials(true)).Returns(new VssCredentials()); + _configStore.Setup(x => x.IsServiceConfigured()).Returns(false); + + return (messages, drained); + } + [Theory] [InlineData(typeof(TaskOrchestrationJobNotFoundException))] // HTTP status 404 [InlineData(typeof(TaskOrchestrationJobAlreadyAcquiredException))] // HTTP status 409 - [InlineData(typeof(TaskOrchestrationJobUnprocessableException))] // HTTP status 422 [Trait("Level", "L0")] [Trait("Category", "Runner")] public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnLostJobAssignment(Type acquireException) { - using (var hc = new TestHostContext(this)) + // Name the context per row: TestHostContext derives its trace file from + // [CallerMemberName] and deletes any existing one, so rows would overwrite each other. + using (var hc = new TestHostContext(this, $"{nameof(TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnLostJobAssignment)}_{acquireException.Name}")) { //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" }), - 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.DeleteSessionAsync()) - .Returns(Task.CompletedTask); - _messageListener.Setup(x => x.DeleteMessageAsync(It.IsAny())) - .Returns(Task.CompletedTask); - _jobNotification.Setup(x => x.StartClient(It.IsAny())) - .Callback(() => - { - - }); + ArrangeRunServiceRunner(hc, runner, ephemeral: true); _runServer.Setup(x => x.GetJobMessageAsync("999", "github", It.IsAny())) .ThrowsAsync((Exception)Activator.CreateInstance(acquireException, "Job assignment is invalid: MissingKey")); - _credentialManager.Setup(x => x.LoadCredentials(true)).Returns(new VssCredentials()); - - _configStore.Setup(x => x.IsServiceConfigured()).Returns(false); - //Act var command = new CommandSettings(hc, new string[] { "run" }); Task runnerTask = runner.ExecuteCommand(command); @@ -1139,117 +1152,58 @@ public async Task TestEphemeralRunnerJobRequestMessageFromRunServiceExitsOnLostJ 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); - } + Assert.Equal(Constants.Runner.ReturnCode.Success, await runnerTask); _runServer.Verify(x => x.GetJobMessageAsync("999", "github", It.IsAny()), Times.Once()); _jobDispatcher.Verify(x => x.Run(It.IsAny(), It.IsAny()), Times.Never()); _acquireJobThrottler.Verify(x => x.IncrementAndWaitAsync(It.IsAny()), Times.Never()); - _messageListener.Verify(x => x.DeleteSessionAsync(), Times.Once()); + // The registration is already consumed, so deleting the session would only stall. + _messageListener.Verify(x => x.DeleteSessionAsync(), Times.Never()); _messageListener.Verify(x => x.DeleteMessageAsync(It.IsAny()), Times.Once()); _configurationManager.Verify(x => x.DeleteLocalRunnerConfig(), Times.Once()); } } - [Fact] + [Theory] + // 422 says the job is unprocessable, not that the assignment moved: an ephemeral runner keeps listening. + [InlineData(true, false, typeof(TaskOrchestrationJobUnprocessableException))] + // A persistent runner keeps skipping whatever the service answers. + [InlineData(false, false, typeof(TaskOrchestrationJobNotFoundException))] + [InlineData(false, false, typeof(TaskOrchestrationJobAlreadyAcquiredException))] + // --once is a client-side flag, so the registration is not consumed and the runner must + // keep listening. This pins settings.Ephemeral as the predicate over the in-scope runOnce. + [InlineData(false, true, typeof(TaskOrchestrationJobAlreadyAcquiredException))] [Trait("Level", "L0")] [Trait("Category", "Runner")] - public async Task TestRunnerJobRequestMessageFromRunServiceContinuesOnLostJobAssignmentForPersistentRunner() + public async Task TestRunnerJobRequestMessageFromRunServiceContinuesOnLostJobAssignment(bool ephemeral, bool runOnce, Type acquireException) { - using (var hc = new TestHostContext(this)) + using (var hc = new TestHostContext(this, $"{nameof(TestRunnerJobRequestMessageFromRunServiceContinuesOnLostJobAssignment)}_{ephemeral}_{runOnce}_{acquireException.Name}")) { //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" }), - MessageId = 4234, - MessageType = JobRequestMessageTypes.RunnerJobRequest - }; - - var messages = new Queue(); - messages.Enqueue(message); - var signalThrottled = 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.DeleteSessionAsync()) - .Returns(Task.CompletedTask); - _messageListener.Setup(x => x.DeleteMessageAsync(It.IsAny())) - .Returns(Task.CompletedTask); - _jobNotification.Setup(x => x.StartClient(It.IsAny())) - .Callback(() => - { - - }); + var arrange = ArrangeRunServiceRunner(hc, runner, ephemeral: ephemeral); _runServer.Setup(x => x.GetJobMessageAsync("999", "github", It.IsAny())) - .ThrowsAsync(new TaskOrchestrationJobAlreadyAcquiredException("Job assignment is invalid: MissingKey")); - _acquireJobThrottler.Setup(x => x.IncrementAndWaitAsync(It.IsAny())) - .Returns(Task.CompletedTask) - .Callback(() => - { - signalThrottled.Release(); - }); - - _credentialManager.Setup(x => x.LoadCredentials(true)).Returns(new VssCredentials()); - - _configStore.Setup(x => x.IsServiceConfigured()).Returns(false); + .ThrowsAsync((Exception)Activator.CreateInstance(acquireException, "Job assignment is invalid: MissingKey")); //Act - var command = new CommandSettings(hc, new string[] { "run" }); + var command = new CommandSettings(hc, runOnce ? new string[] { "run", "--once" } : new string[] { "run" }); Task runnerTask = runner.ExecuteCommand(command); //Assert - //the runner skips the job and keeps listening, so it only stops when we shut it down - if (!await signalThrottled.WaitAsync(2000)) - { - Assert.Fail($"{nameof(_acquireJobThrottler.Object.IncrementAndWaitAsync)} was not invoked."); - } + //the runner skips the job and goes back to listening, so it stops only once we shut it down + await Task.WhenAny(arrange.Drained.Task, runnerTask, Task.Delay(30000)); + Assert.True(arrange.Drained.Task.IsCompletedSuccessfully, $"the runner did not go back to listening. {runnerTask.Exception?.ToString()}"); hc.ShutdownRunner(ShutdownReason.UserCancelled); - await Task.WhenAny(runnerTask, Task.Delay(2000)); + await Task.WhenAny(runnerTask, Task.Delay(30000)); Assert.True(runnerTask.IsCompleted, $"{nameof(runner.ExecuteCommand)} timed out."); Assert.True(runnerTask.IsCanceled); _runServer.Verify(x => x.GetJobMessageAsync("999", "github", It.IsAny()), Times.Once()); _acquireJobThrottler.Verify(x => x.IncrementAndWaitAsync(It.IsAny()), Times.Once()); _jobDispatcher.Verify(x => x.Run(It.IsAny(), It.IsAny()), Times.Never()); + // Skipping must still drain the message, or Broker redelivers it forever. + _messageListener.Verify(x => x.DeleteMessageAsync(It.IsAny()), Times.Once()); _configurationManager.Verify(x => x.DeleteLocalRunnerConfig(), Times.Never()); } }