From 18311a89c6a2ce620028390f9ae9515382c5853b Mon Sep 17 00:00:00 2001 From: Chris0Jeky Date: Sun, 6 Sep 2026 05:40:04 +0100 Subject: [PATCH 1/2] Add shutdown outcome metrics to LLM worker --- .../Workers/LlmQueueToProposalWorker.cs | 42 +++++--- .../LlmQueueToProposalWorkerTests.cs | 99 +++++++++++++++++++ 2 files changed, 130 insertions(+), 11 deletions(-) diff --git a/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs b/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs index 8fa3ddd91..b7a597f6e 100644 --- a/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs +++ b/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs @@ -12,6 +12,8 @@ public class LlmQueueToProposalWorker : BackgroundService { private const string QueueNameLlm = "llm"; private const string QueueNameCaptureTriage = "capture-triage"; + private const string OutcomeAbandonedShutdown = "abandoned_shutdown"; + 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 +369,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(OutcomeAbandonedShutdown); + } + throw; } catch (Exception ex) @@ -403,14 +409,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 +429,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 +442,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 +453,7 @@ 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)); + return false; } } @@ -741,11 +749,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 +811,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 +880,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..0886cb360 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; @@ -121,6 +123,34 @@ 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, _) => + { + foreach (var tag in tags) + { + if (tag.Key == TaskdeckTelemetryTags.Outcome && tag.Value is string outcome) + { + outcomes.Add(outcome); + break; + } + } + }); + listener.Start(); + listener.EnableMeasurementEvents(TaskdeckTelemetry.WorkerItemsProcessed); + return listener; + } + #endregion #region Happy path: pending item processed to completion @@ -246,6 +276,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 +302,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("abandoned_shutdown"); } [Fact] @@ -303,6 +336,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("abandoned_shutdown"); + } + [Fact] public async Task ProcessBatch_ProposalLaneShutdownRelease_UsesFreshScopeWithoutFlushingPlannerState() { @@ -605,6 +667,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 +696,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] From 2f6531cea9f0581561036b23def6234d98b05aa4 Mon Sep 17 00:00:00 2001 From: Chris0Jeky Date: Sun, 6 Sep 2026 12:52:16 +0100 Subject: [PATCH 2/2] fix(worker): shutdown outcomes named for what happened (released_on_shutdown, shutdown_release_failed); telemetry listener filtered by worker, synchronized, and run in a non-parallel collection (review HIGH-1, MEDIUM-2) --- .../Workers/LlmQueueToProposalWorker.cs | 10 +++++-- .../LlmQueueToProposalWorkerTests.cs | 30 ++++++++++++++++--- 2 files changed, 34 insertions(+), 6 deletions(-) diff --git a/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs b/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs index b7a597f6e..377b147f0 100644 --- a/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs +++ b/backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs @@ -12,7 +12,10 @@ public class LlmQueueToProposalWorker : BackgroundService { private const string QueueNameLlm = "llm"; private const string QueueNameCaptureTriage = "capture-triage"; - private const string OutcomeAbandonedShutdown = "abandoned_shutdown"; + // 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 @@ -371,7 +374,7 @@ private async Task ProcessSingleItemAsync(Guid itemId, DateTimeOffset expectedUp { if (await ReleaseClaimOnShutdownAsync(item)) { - RecordWorkerOutcome(OutcomeAbandonedShutdown); + RecordWorkerOutcome(OutcomeReleasedOnShutdown); } throw; @@ -453,6 +456,9 @@ 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; } } diff --git a/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs b/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs index 0886cb360..ef143346d 100644 --- a/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs +++ b/backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs @@ -17,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 @@ -137,12 +146,25 @@ private static MeterListener ListenForWorkerOutcomes(List outcomes) }; listener.SetMeasurementEventCallback((instrument, _, tags, _) => { + string? outcome = null; + var thisWorker = false; foreach (var tag in tags) { - if (tag.Key == TaskdeckTelemetryTags.Outcome && tag.Value is string outcome) + 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); - break; } } }); @@ -302,7 +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("abandoned_shutdown"); + outcomes.Should().Equal("released_on_shutdown"); } [Fact] @@ -362,7 +384,7 @@ public async Task ProcessBatch_ProposalLaneCallerCancellation_WhenReleaseFails_D await Assert.ThrowsAnyAsync( () => InvokeProcessBatchAsync(worker, cts.Token)); - outcomes.Should().NotContain("abandoned_shutdown"); + outcomes.Should().NotContain("released_on_shutdown"); } [Fact]