diff --git a/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs b/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs index 8fa3ddd91..377b147f0 100644 --- a/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs +++ b/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs @@ -12,6 +12,11 @@ public class LlmQueueToProposalWorker : BackgroundService { private const string QueueNameLlm = "llm"; private const string QueueNameCaptureTriage = "capture-triage"; + // The claim was released back to Pending before the host stopped; nothing was abandoned. + private const string OutcomeReleasedOnShutdown = "released_on_shutdown"; + // The release write failed, so the row stays Processing for the recovery sweep: the unhealthy path. + private const string OutcomeShutdownReleaseFailed = "shutdown_release_failed"; + private const string OutcomeCancelledRequeued = "cancelled_requeued"; // Floor on the stuck-recovery lease so an aggressively-low ProcessingLeaseSeconds can never sweep a // request that was claimed seconds ago and is still legitimately in flight. Mirrors the floor the @@ -367,7 +372,11 @@ private async Task ProcessSingleItemAsync(Guid itemId, DateTimeOffset expectedUp // marked Failed and requeued purely because the host was stopping. catch (OperationCanceledException) when (ct.IsCancellationRequested) { - await ReleaseClaimOnShutdownAsync(item); + if (await ReleaseClaimOnShutdownAsync(item)) + { + RecordWorkerOutcome(OutcomeReleasedOnShutdown); + } + throw; } catch (Exception ex) @@ -403,14 +412,14 @@ private async Task ProcessSingleItemAsync(Guid itemId, DateTimeOffset expectedUp /// Processing already is their queued state and returning one to Pending would drop it back to /// "untriaged in the inbox" — the wrong state, and one no worker drains. /// - private async Task ReleaseClaimOnShutdownAsync(LlmRequest item) + private async Task ReleaseClaimOnShutdownAsync(LlmRequest item) { // The failure path may already have moved the row on (Failed, or Pending/Processing once // HandleFailureWithRetryAsync completed its own interrupted transition); only a still-claimed row // is ours to release. if (item.Status != RequestStatus.Processing) { - return; + return false; } try @@ -423,7 +432,7 @@ private async Task ReleaseClaimOnShutdownAsync(LlmRequest item) var shutdownItem = await shutdownUnitOfWork.LlmQueue.GetByIdAsync(item.Id, CancellationToken.None); if (shutdownItem == null || shutdownItem.Status != RequestStatus.Processing) { - return; + return false; } shutdownItem.ReleaseClaim(); @@ -436,6 +445,7 @@ private async Task ReleaseClaimOnShutdownAsync(LlmRequest item) _logger.LogInformation( "Queue item {ItemId} released back to Pending during shutdown; no retry charged", item.Id); + return true; } catch (Exception ex) { @@ -446,6 +456,10 @@ private async Task ReleaseClaimOnShutdownAsync(LlmRequest item) "Failed to release queue item {ItemId} during shutdown; it stays Processing for the recovery sweep. {ExceptionSummary}", item.Id, SensitiveDataRedactor.SummarizeException(ex)); + // Only a genuinely failed release write is the unhealthy outcome; a row that had already + // moved on (the early returns above) has nothing to release and emits nothing here. + RecordWorkerOutcome(OutcomeShutdownReleaseFailed); + return false; } } @@ -741,11 +755,14 @@ private async Task HandleFailureWithRetryAsync( { try { - await CompleteRetryTransitionOnShutdownAsync(item.Id, retryAsProcessing); - _logger.LogInformation( - "Queue item {ItemId} requeued for retry attempt {RetryCount} during shutdown", - item.Id, - item.RetryCount + 1); + if (await CompleteRetryTransitionOnShutdownAsync(item.Id, retryAsProcessing)) + { + RecordWorkerOutcome(OutcomeCancelledRequeued); + _logger.LogInformation( + "Queue item {ItemId} requeued for retry attempt {RetryCount} during shutdown", + item.Id, + item.RetryCount + 1); + } } catch (Exception ex) { @@ -800,18 +817,19 @@ private async Task HandleFailureWithRetryInFreshScopeAsync( retryAsProcessing); } - private async Task CompleteRetryTransitionOnShutdownAsync(Guid itemId, bool retryAsProcessing) + private async Task CompleteRetryTransitionOnShutdownAsync(Guid itemId, bool retryAsProcessing) { using var scope = _scopeFactory.CreateScope(); var shutdownUnitOfWork = scope.ServiceProvider.GetRequiredService(); var shutdownItem = await shutdownUnitOfWork.LlmQueue.GetByIdAsync(itemId, CancellationToken.None); if (shutdownItem == null) { - return; + return false; } ApplyRetryTransition(shutdownItem, retryAsProcessing); await shutdownUnitOfWork.SaveChangesAsync(CancellationToken.None); + return true; } /// @@ -868,5 +886,13 @@ private static void RecordWorkerProcessingMetrics(double durationMs, string outc new KeyValuePair(TaskdeckTelemetryTags.Outcome, outcome)); } + private static void RecordWorkerOutcome(string outcome) + { + TaskdeckTelemetry.WorkerItemsProcessed.Add( + 1, + new KeyValuePair(TaskdeckTelemetryTags.WorkerName, nameof(LlmQueueToProposalWorker)), + new KeyValuePair(TaskdeckTelemetryTags.Outcome, outcome)); + } + private readonly record struct WorkerBatchItem(Guid ItemId, bool IsCaptureTriage, DateTimeOffset? ExpectedUpdatedAt = null); } diff --git a/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs b/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs index fcc004d8c..ef143346d 100644 --- a/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs +++ b/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs @@ -1,9 +1,11 @@ +using System.Diagnostics.Metrics; using System.Reflection; using FluentAssertions; using Microsoft.Extensions.DependencyInjection; using Moq; using Microsoft.Extensions.Logging.Abstractions; using Taskdeck.Api.Workers; +using Taskdeck.Api.Telemetry; using Taskdeck.Application.DTOs; using Taskdeck.Application.Interfaces; using Taskdeck.Application.Services; @@ -15,6 +17,15 @@ namespace Taskdeck.Api.Tests; +// The outcome listener below observes the process-global WorkerItemsProcessed counter, which +// the resilience and transcript-worker test classes also drive; this collection runs alone. +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class WorkerTelemetryCollection +{ + public const string Name = "Worker telemetry global state"; +} + +[Collection(WorkerTelemetryCollection.Name)] public class LlmQueueToProposalWorkerTests { #region Helper factories @@ -121,6 +132,47 @@ private static async Task InvokeProcessBatchAsync( await (Task)task!; } + private static MeterListener ListenForWorkerOutcomes(List outcomes) + { + var listener = new MeterListener + { + InstrumentPublished = (instrument, meterListener) => + { + if (instrument == TaskdeckTelemetry.WorkerItemsProcessed) + { + meterListener.EnableMeasurementEvents(instrument); + } + } + }; + listener.SetMeasurementEventCallback((instrument, _, tags, _) => + { + string? outcome = null; + var thisWorker = false; + foreach (var tag in tags) + { + if (tag.Key == TaskdeckTelemetryTags.Outcome && tag.Value is string value) + { + outcome = value; + } + else if (tag.Key == TaskdeckTelemetryTags.WorkerName && tag.Value is string name) + { + thisWorker = name == nameof(LlmQueueToProposalWorker); + } + } + + if (thisWorker && outcome is not null) + { + lock (outcomes) + { + outcomes.Add(outcome); + } + } + }); + listener.Start(); + listener.EnableMeasurementEvents(TaskdeckTelemetry.WorkerItemsProcessed); + return listener; + } + #endregion #region Happy path: pending item processed to completion @@ -246,6 +298,8 @@ public async Task ProcessBatch_ProposalLaneCallerCancellation_PropagatesAndRelea { var item = CreatePendingItem(); var queueRepo = new FakeLlmQueueRepository([item]); + var outcomes = new List(); + using var listener = ListenForWorkerOutcomes(outcomes); using var cts = new CancellationTokenSource(); var planner = new FakeAutomationPlannerService { @@ -270,6 +324,7 @@ await Assert.ThrowsAnyAsync( // the processing lease expires. item.Status.Should().Be(RequestStatus.Pending); item.RetryCount.Should().Be(0, "a shutdown mid-item must not consume the retry budget"); + outcomes.Should().Equal("released_on_shutdown"); } [Fact] @@ -303,6 +358,35 @@ await Assert.ThrowsAnyAsync( "the release must be persisted with CancellationToken.None, not the cancelled caller token"); } + [Fact] + public async Task ProcessBatch_ProposalLaneCallerCancellation_WhenReleaseFails_DoesNotEmitSuccessfulOutcome() + { + var item = CreatePendingItem(); + var queueRepo = new FakeLlmQueueRepository([item]); + var unitOfWork = new FakeUnitOfWork(queueRepo) + { + OnSaveChanges = _ => throw new InvalidOperationException("release write failed") + }; + var outcomes = new List(); + using var listener = ListenForWorkerOutcomes(outcomes); + using var cts = new CancellationTokenSource(); + var planner = new FakeAutomationPlannerService + { + ResultFactory = _ => + { + cts.Cancel(); + throw new OperationCanceledException(cts.Token); + } + }; + using var sp = BuildServiceProvider(queueRepo, planner, unitOfWork: unitOfWork); + var worker = CreateWorker(sp.GetRequiredService(), DefaultSettings(retryBackoff: [0])); + + await Assert.ThrowsAnyAsync( + () => InvokeProcessBatchAsync(worker, cts.Token)); + + outcomes.Should().NotContain("released_on_shutdown"); + } + [Fact] public async Task ProcessBatch_ProposalLaneShutdownRelease_UsesFreshScopeWithoutFlushingPlannerState() { @@ -605,6 +689,8 @@ public async Task ProcessBatch_CancellationDuringRetryBackoff_PersistsTheRequeue var item = CreatePendingItem(); var queueRepo = new FakeLlmQueueRepository([item]); var unitOfWork = new FakeUnitOfWork(queueRepo); + var outcomes = new List(); + using var listener = ListenForWorkerOutcomes(outcomes); using var cts = new CancellationTokenSource(); // Cancel from inside the FIRST write (the MarkAsFailed commit) so cancellation lands squarely in // the backoff wait that follows -- deterministic, unlike a timer race, and it keeps the token @@ -632,6 +718,41 @@ await Assert.ThrowsAnyAsync( unitOfWork.SaveChangesTokens.Should().HaveCount(2, "the MarkAsFailed commit and the requeue commit"); unitOfWork.SaveChangesTokens[1].IsCancellationRequested.Should().BeFalse( "the requeue must be persisted with CancellationToken.None; the caller token is already cancelled"); + outcomes.Should().Equal("cancelled_requeued"); + } + + [Fact] + public async Task ProcessBatch_CancellationDuringRetryBackoff_WhenRequeueFails_DoesNotEmitSuccessfulOutcome() + { + var item = CreatePendingItem(); + var queueRepo = new FakeLlmQueueRepository([item]); + var unitOfWork = new FakeUnitOfWork(queueRepo); + var outcomes = new List(); + using var listener = ListenForWorkerOutcomes(outcomes); + using var cts = new CancellationTokenSource(); + unitOfWork.OnSaveChanges = saveNumber => + { + if (saveNumber == 1) + { + cts.Cancel(); + } + else if (saveNumber == 2) + { + throw new InvalidOperationException("requeue write failed"); + } + }; + var planner = new FakeAutomationPlannerService + { + ResultFactory = _ => Result.Failure(ErrorCodes.UnexpectedError, "transient failure") + }; + using var sp = BuildServiceProvider(queueRepo, planner, unitOfWork: unitOfWork); + var worker = CreateWorker(sp.GetRequiredService(), + DefaultSettings(maxRetries: 3, retryBackoff: [5])); + + await Assert.ThrowsAnyAsync( + () => InvokeProcessBatchAsync(worker, cts.Token)); + + outcomes.Should().NotContain("cancelled_requeued"); } [Fact]