diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr4cCodexFixesTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr4cCodexFixesTests.cs index b04f6621..22fa3894 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr4cCodexFixesTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr4cCodexFixesTests.cs @@ -12,6 +12,7 @@ using Azure.Storage.Queues; using Azure.Storage.Queues.Models; using AutopilotMonitor.Functions.Services; using AutopilotMonitor.Functions.Services.Deletion; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models; @@ -268,7 +269,7 @@ public class SessionDeletionPr4cCodexFixesTests ManifestId = "STALE-OLD-MANIFEST-ID", // ≠ row.PendingDeletionManifestId Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(staleEnvelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(staleEnvelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); // Default GetSessionRowAsync setup returns PendingDeletionManifestId = ManifestId // (a fresh active cascade); the envelope's STALE-OLD-MANIFEST-ID must NOT touch state. diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionWorkerTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionWorkerTests.cs index 21282ceb..5cc12025 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionWorkerTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionWorkerTests.cs @@ -11,6 +11,7 @@ using Azure.Storage.Queues.Models; using AutopilotMonitor.Functions.Services; using AutopilotMonitor.Functions.Services.Deletion; using AutopilotMonitor.Functions.Services.Notifications; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models; using AutopilotMonitor.Shared.Models.Deletion; @@ -59,7 +60,7 @@ public class SessionDeletionWorkerTests TenantId = TenantId, SessionId = SessionId, ManifestId = ManifestId, Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); await harness.RunForAsync(TimeSpan.FromMilliseconds(500)); @@ -103,7 +104,7 @@ public class SessionDeletionWorkerTests TenantId = TenantId, SessionId = SessionId, ManifestId = ManifestId, Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); const int capObservation = 50; var sample = string.Join(",", Enumerable.Range(0, capObservation) @@ -153,7 +154,7 @@ public class SessionDeletionWorkerTests TenantId = TenantId, SessionId = SessionId, ManifestId = ManifestId, Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); harness.BlobMock.Setup(b => b.DownloadDeletionProgressAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) @@ -281,7 +282,7 @@ public class SessionDeletionWorkerTests TenantId = TenantId, SessionId = SessionId, ManifestId = ManifestId, Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); // 50 sample entries × ~200-char composite keys = ~10 KB blob; the worker must defend. const int progressSampleSize = 50; @@ -334,7 +335,7 @@ public class SessionDeletionWorkerTests TenantId = TenantId, SessionId = SessionId, ManifestId = ManifestId, Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); // Track call order: the CAS must come BEFORE the poison-queue send. var callOrder = new List(); @@ -378,7 +379,7 @@ public class SessionDeletionWorkerTests TenantId = TenantId, SessionId = SessionId, ManifestId = ManifestId, Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); // First attempt: Running→Poisoned WrongState (current=Queued). harness.StorageMock.Setup(s => s.CasSetSessionDeletionStateAsync( @@ -431,7 +432,7 @@ public class SessionDeletionWorkerTests TenantId = TenantId, SessionId = SessionId, ManifestId = ManifestId, Reason = "admin_delete", EnqueuedAt = DateTime.UtcNow, }; - harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: SessionDeletionWorker.MaxDequeueCount + 1); + harness.EnqueueMessage(JsonConvert.SerializeObject(envelope), dequeueCount: QueuePollingWorkerBase.DefaultMaxDequeueCount + 1); harness.StorageMock.Setup(s => s.CasSetSessionDeletionStateAsync( TenantId, SessionId, diff --git a/src/Backend/AutopilotMonitor.Functions/Program.cs b/src/Backend/AutopilotMonitor.Functions/Program.cs index a220ae97..436b3b25 100644 --- a/src/Backend/AutopilotMonitor.Functions/Program.cs +++ b/src/Backend/AutopilotMonitor.Functions/Program.cs @@ -134,6 +134,9 @@ builder.Services.AddMemoryCache(); builder.Services.AddHttpClient(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +// Shared QueueClient bootstrap (code-quality audit D2): single Managed-Identity/connection-string +// resolution consumed by every queue producer / worker / probe. +builder.Services.AddSingleton(); // Cascade-deletion read surface implemented by the TableStorageService partial (PR1). // No production caller in PR1; producer (PR3) + worker (PR4) consume it later. builder.Services.AddSingleton( diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs index 968e406a..ce1cca53 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs @@ -1,255 +1,34 @@ using System; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; -using Azure.Storage.Queues; -using Azure.Storage.Queues.Models; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; -using Newtonsoft.Json; namespace AutopilotMonitor.Functions.Services.Analyze { /// - /// Background worker for the analyze-on-enrollment-end queue. Replaces the - /// previous in-function fire-and-forget Task.Run that ran the rule engine after - /// session-terminal events — that pattern could be killed mid-flight by Azure Functions - /// scale-in, leaving rule results un-persisted. - /// - /// Mirrors IndexReconcileQueueWorker instead of using a Functions QueueTrigger - /// because the trigger binding requires a Functions-host-specific connection app-setting - /// (<Connection>__queueServiceUri) that the rest of the project's storage - /// access does not need. Pure DI BackgroundService driven by the existing - /// AzureStorageAccountName (Managed Identity) or AzureTableStorageConnectionString - /// fallback — no bespoke binding glue. - /// - /// - /// Platform parity to QueueTrigger: - /// - /// Receive batch of up to messages with - /// visibility-timeout per message. - /// Handler success → DeleteMessageAsync. - /// Handler exception → no delete; message becomes visible again after the timeout, - /// DequeueCount increments. Identical to platform-retry semantics. - /// After failed attempts the message is moved to - /// analyze-on-enrollment-end-poison and deleted from the main queue. - /// Empty receive → sleep for and retry. - /// - /// + /// Background worker for the analyze-on-enrollment-end queue. Replaces the previous + /// in-function fire-and-forget Task.Run that ran the rule engine after session-terminal + /// events — that pattern could be killed mid-flight by Azure Functions scale-in, leaving rule + /// results un-persisted. Uses the shared poll loop + /// with vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison). /// - public sealed class AnalyzeOnEnrollmentEndQueueWorker : BackgroundService + public sealed class AnalyzeOnEnrollmentEndQueueWorker : QueuePollingWorker { - /// Max messages received per poll. Storage Queue caps batch-receive at 32. - internal const int BatchSize = 32; - - /// - /// Visibility timeout per received message. Rule evaluation reads all session events - /// + N rules, then writes 0..N rule-result rows + per-rule stat rows; large sessions - /// can take several seconds. 5 minutes leaves plenty of headroom and matches the - /// IndexReconcile worker. - /// - internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5); - - /// Idle sleep between empty receives. - internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10); - - /// Cool-down after an unhandled poll-loop exception. - internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30); - - /// Match the QueueTrigger platform default: 5 attempts, then move to poison. - internal const int MaxDequeueCount = 5; - - private const string PoisonQueueSuffix = "-poison"; - - private readonly QueueClient _mainQueue; - private readonly QueueClient _poisonQueue; private readonly AnalyzeOnEnrollmentEndHandler _handler; - private readonly ILogger _logger; public AnalyzeOnEnrollmentEndQueueWorker( - IConfiguration configuration, + QueueClientFactory queueFactory, AnalyzeOnEnrollmentEndHandler handler, ILogger logger) + : base(queueFactory, Constants.QueueNames.AnalyzeOnEnrollmentEnd, logger) { _handler = handler ?? throw new ArgumentNullException(nameof(handler)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var mainUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.AnalyzeOnEnrollmentEnd}"); - var poisonUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.AnalyzeOnEnrollmentEnd}{PoisonQueueSuffix}"); - var credential = new DefaultAzureCredential(); - _mainQueue = new QueueClient(mainUri, credential, options); - _poisonQueue = new QueueClient(poisonUri, credential, options); - _logger.LogInformation( - "AnalyzeOnEnrollmentEndQueueWorker initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _mainQueue = new QueueClient( - connectionString, Constants.QueueNames.AnalyzeOnEnrollmentEnd, options); - _poisonQueue = new QueueClient( - connectionString, Constants.QueueNames.AnalyzeOnEnrollmentEnd + PoisonQueueSuffix, options); - _logger.LogInformation("AnalyzeOnEnrollmentEndQueueWorker initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - await TryCreateQueueAsync(_mainQueue, "main", stoppingToken).ConfigureAwait(false); - await TryCreateQueueAsync(_poisonQueue, "poison", stoppingToken).ConfigureAwait(false); - - _logger.LogInformation("AnalyzeOnEnrollmentEndQueueWorker: poll loop started (queue {Name})", - _mainQueue.Name); - - while (!stoppingToken.IsCancellationRequested) - { - try - { - var batch = await _mainQueue - .ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken) - .ConfigureAwait(false); - - if (batch?.Value is null || batch.Value.Length == 0) - { - await Task.Delay(PollInterval, stoppingToken).ConfigureAwait(false); - continue; - } - - foreach (var msg in batch.Value) - { - if (stoppingToken.IsCancellationRequested) break; - await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _logger.LogError(ex, - "AnalyzeOnEnrollmentEndQueueWorker: poll-loop error — backing off {Backoff}", - ErrorBackoff); - try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); } - catch (OperationCanceledException) { break; } - } - } - - _logger.LogInformation("AnalyzeOnEnrollmentEndQueueWorker: poll loop stopped"); - } - - private async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) - { - if (msg.DequeueCount > MaxDequeueCount) - { - await MoveToPoisonAsync(msg, ct).ConfigureAwait(false); - return; - } - - AnalyzeOnEnrollmentEndEnvelope? envelope; - try - { - envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); - } - catch (JsonException ex) - { - _logger.LogWarning(ex, - "AnalyzeOnEnrollmentEndQueueWorker: malformed envelope JSON — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - if (envelope is null) - { - _logger.LogWarning( - "AnalyzeOnEnrollmentEndQueueWorker: null envelope after deserialization — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - try - { - await _handler.HandleAsync(envelope, ct).ConfigureAwait(false); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - throw; - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "AnalyzeOnEnrollmentEndQueueWorker: handler failed (msg {Id}, dequeue {N}) — visibility-timeout retry", - msg.MessageId, msg.DequeueCount); - } - } - - private async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct) - { - try - { - await _poisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - _logger.LogWarning( - "AnalyzeOnEnrollmentEndQueueWorker: moved message {Id} to poison queue after {N} failed attempts", - msg.MessageId, msg.DequeueCount - 1); - } - catch (Exception ex) - { - _logger.LogError(ex, - "AnalyzeOnEnrollmentEndQueueWorker: poison move failed for message {Id} (will retry)", - msg.MessageId); - } - } - - private async Task SafeDeleteAsync(QueueMessage msg, CancellationToken ct) - { - try - { - await _mainQueue.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "AnalyzeOnEnrollmentEndQueueWorker: delete failed for message {Id} (will reappear after visibility-timeout)", - msg.MessageId); - } - } - - private async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct) - { - try - { - await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "AnalyzeOnEnrollmentEndQueueWorker: CreateIfNotExists failed for {Label} queue — will continue, send/receive will retry", - label); - } - } + protected override Task HandleAsync(AnalyzeOnEnrollmentEndEnvelope envelope, CancellationToken ct) + => _handler.HandleAsync(envelope, ct); } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AzureQueueAnalyzeOnEnrollmentEndProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AzureQueueAnalyzeOnEnrollmentEndProducer.cs index df3a1858..460c8233 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AzureQueueAnalyzeOnEnrollmentEndProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AzureQueueAnalyzeOnEnrollmentEndProducer.cs @@ -1,11 +1,10 @@ using System; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; using Azure.Storage.Queues; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -30,40 +29,12 @@ namespace AutopilotMonitor.Functions.Services.Analyze private int _queueEnsured; // 0 = not yet ensured, 1 = CreateIfNotExistsAsync has run public AzureQueueAnalyzeOnEnrollmentEndProducer( - IConfiguration configuration, + QueueClientFactory queueFactory, ILogger logger) { _logger = logger; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - // Match the consumer (BackgroundService) Base64 encoding so messages round-trip. - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var queueUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.AnalyzeOnEnrollmentEnd}"); - _queueClient = new QueueClient(queueUri, new DefaultAzureCredential(), options); - _logger.LogInformation( - "AnalyzeOnEnrollmentEnd producer initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _queueClient = new QueueClient( - connectionString, Constants.QueueNames.AnalyzeOnEnrollmentEnd, options); - _logger.LogInformation("AnalyzeOnEnrollmentEnd producer initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } + // Base64 encoding matches the consumer (BackgroundService) so messages round-trip. + _queueClient = queueFactory.Create(Constants.QueueNames.AnalyzeOnEnrollmentEnd); } public async Task EnqueueAsync(AnalyzeOnEnrollmentEndEnvelope envelope, CancellationToken cancellationToken = default) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/AzureQueueCriticalTableBackupProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/AzureQueueCriticalTableBackupProducer.cs index 8031f25c..c172bf44 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/AzureQueueCriticalTableBackupProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/AzureQueueCriticalTableBackupProducer.cs @@ -1,10 +1,9 @@ using System; using System.Threading; using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; -using Azure.Identity; using Azure.Storage.Queues; -using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -25,39 +24,11 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue private int _queueEnsured; public AzureQueueCriticalTableBackupProducer( - IConfiguration configuration, + QueueClientFactory queueFactory, ILogger logger) { _logger = logger; - - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var queueUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.CriticalTableBackup}"); - _queueClient = new QueueClient(queueUri, new DefaultAzureCredential(), options); - _logger.LogInformation( - "CriticalTableBackup producer initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _queueClient = new QueueClient( - connectionString, Constants.QueueNames.CriticalTableBackup, options); - _logger.LogInformation("CriticalTableBackup producer initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' or 'AzureTableStorageConnectionString'."); - } + _queueClient = queueFactory.Create(Constants.QueueNames.CriticalTableBackup); } public async Task EnqueueAsync(CriticalTableBackupEnvelope envelope, CancellationToken ct = default) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs index 9e0c4377..775a01f9 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs @@ -2,147 +2,69 @@ using System; using System.Threading; using System.Threading.Tasks; using AutopilotMonitor.Functions.DataAccess.TableStorage; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models.Backup; using Azure; -using Azure.Identity; using Azure.Storage.Blobs.Specialized; -using Azure.Storage.Queues; using Azure.Storage.Queues.Models; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Newtonsoft.Json; namespace AutopilotMonitor.Functions.Services.Backup.Queue { /// - /// Background worker for the critical-table-backup-jobs queue. Drains messages, - /// holds the maintenance lease for the duration of the per-job work, and runs + /// Background worker for the critical-table-backup-jobs queue. Drains messages, holds + /// the maintenance lease for the duration of the per-job work, and runs /// with the lease held. /// - /// State-machine boundary (plan Wave17/18/19/22/23): the WORKER owns the lease - /// and JobStatus; the SERVICE is intentionally lease-unaware and stateless. State - /// transitions follow this normative order: Duplicate-Detection → Acquire-Lease → - /// CAS Queued→Running → Run (under lease) → CAS Running→Completed (under lease) → - /// Release Lease → SafeDelete. Service-throw of a retryable exception triggers - /// CAS Running→Queued (so the next dequeue sees Queued, not Running+free-lease) plus - /// rethrow so the outer poll-loop applies VisibilityTimeout retry semantics. + /// Built on (not ) + /// because the per-message lifecycle is a bespoke lease + CAS state machine, not a simple + /// deserialize → handler → delete flow. The base supplies the queue bootstrap, poll loop, + /// safe-delete, queue-creation and poison-move shell; this worker overrides + /// and . /// /// - /// Queue-config differs from short-lived workers (plan Wave6): BatchSize=1, - /// VisibilityTimeout=60min. The per-run service budget is 50min so the 60min visibility - /// gives a 10min cushion — no explicit renewal loop is needed in PR1 (deferred as a - /// hardening follow-up; Renewal-Loop adds ~150 LOC of race-handling that is only - /// load-bearing for runs > 50min). + /// State-machine boundary (plan Wave17/18/19/22/23): the WORKER owns the lease and + /// JobStatus; the SERVICE is intentionally lease-unaware and stateless. State transitions: + /// Duplicate-Detection → Acquire-Lease → CAS Queued→Running → Run (under lease) → + /// CAS Running→Completed (under lease) → Release Lease → SafeDelete. + /// + /// + /// Queue-config differs from short-lived workers (plan Wave6): =1, + /// =60min. The per-run service budget is 50min so the 60min + /// visibility gives a 10min cushion. /// /// - public sealed class CriticalTableBackupQueueWorker : BackgroundService + public sealed class CriticalTableBackupQueueWorker : QueuePollingWorkerBase { - internal const int BatchSize = 1; - internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(60); - internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10); - internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30); - internal const int MaxDequeueCount = 5; - - private const string PoisonQueueSuffix = "-poison"; - - private readonly QueueClient _mainQueue; - private readonly QueueClient _poisonQueue; private readonly ICriticalTableBackupService _backupService; private readonly BlobBackupStore _blobStore; private readonly BackupJobsRepository _jobs; private readonly OpsEventService _opsEvents; - private readonly ILogger _logger; private readonly TimeProvider _clock; public CriticalTableBackupQueueWorker( - IConfiguration configuration, + QueueClientFactory queueFactory, ICriticalTableBackupService backupService, BlobBackupStore blobStore, BackupJobsRepository jobs, OpsEventService opsEvents, ILogger logger, TimeProvider? clock = null) + : base(queueFactory, Constants.QueueNames.CriticalTableBackup, logger, Constants.QueueNames.CriticalTableBackupPoison) { _backupService = backupService ?? throw new ArgumentNullException(nameof(backupService)); _blobStore = blobStore ?? throw new ArgumentNullException(nameof(blobStore)); _jobs = jobs ?? throw new ArgumentNullException(nameof(jobs)); _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _clock = clock ?? TimeProvider.System; - - var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var mainUri = new Uri($"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.CriticalTableBackup}"); - var poisonUri = new Uri($"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.CriticalTableBackup}{PoisonQueueSuffix}"); - var credential = new DefaultAzureCredential(); - _mainQueue = new QueueClient(mainUri, credential, options); - _poisonQueue = new QueueClient(poisonUri, credential, options); - _logger.LogInformation( - "CriticalTableBackupQueueWorker initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _mainQueue = new QueueClient(connectionString, Constants.QueueNames.CriticalTableBackup, options); - _poisonQueue = new QueueClient(connectionString, Constants.QueueNames.CriticalTableBackup + PoisonQueueSuffix, options); - _logger.LogInformation("CriticalTableBackupQueueWorker initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' or 'AzureTableStorageConnectionString'."); - } } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - await TryCreateQueueAsync(_mainQueue, "main", stoppingToken).ConfigureAwait(false); - await TryCreateQueueAsync(_poisonQueue, "poison", stoppingToken).ConfigureAwait(false); + protected override int BatchSize => 1; + protected override TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(60); - _logger.LogInformation("CriticalTableBackupQueueWorker: poll loop started"); - - while (!stoppingToken.IsCancellationRequested) - { - try - { - var batch = await _mainQueue - .ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken) - .ConfigureAwait(false); - - if (batch?.Value is null || batch.Value.Length == 0) - { - await Task.Delay(PollInterval, stoppingToken).ConfigureAwait(false); - continue; - } - - foreach (var msg in batch.Value) - { - if (stoppingToken.IsCancellationRequested) break; - await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _logger.LogError(ex, "CriticalTableBackupQueueWorker: poll-loop error — backing off {Backoff}", ErrorBackoff); - try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); } - catch (OperationCanceledException) { break; } - } - } - - _logger.LogInformation("CriticalTableBackupQueueWorker: poll loop stopped"); - } - - private async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) + protected override async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) { if (msg.DequeueCount > MaxDequeueCount) { @@ -157,14 +79,14 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue } catch (JsonException ex) { - _logger.LogWarning(ex, "CriticalTableBackupQueueWorker: malformed envelope — dropping (msg {Id})", msg.MessageId); + Logger.LogWarning(ex, "{Worker}: malformed envelope — dropping (msg {Id})", WorkerName, msg.MessageId); await SafeDeleteAsync(msg, ct).ConfigureAwait(false); return; } if (envelope is null || string.IsNullOrEmpty(envelope.JobId)) { - _logger.LogWarning("CriticalTableBackupQueueWorker: empty envelope — dropping (msg {Id})", msg.MessageId); + Logger.LogWarning("{Worker}: empty envelope — dropping (msg {Id})", WorkerName, msg.MessageId); await SafeDeleteAsync(msg, ct).ConfigureAwait(false); return; } @@ -178,18 +100,18 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue var (job, jobETag) = await _jobs.GetWithETagAsync(jobId, ct).ConfigureAwait(false); if (job is null) { - _logger.LogWarning( - "CriticalTableBackupQueueWorker: jobId {JobId} not in BackupJobs — dropping orphan message {Id}", - jobId, msg.MessageId); + Logger.LogWarning( + "{Worker}: jobId {JobId} not in BackupJobs — dropping orphan message {Id}", + WorkerName, jobId, msg.MessageId); await SafeDeleteAsync(msg, ct).ConfigureAwait(false); return; } if (IsTerminal(job.State)) { - _logger.LogInformation( - "CriticalTableBackupQueueWorker: jobId {JobId} already terminal ({State}) — dropping reappearance", - jobId, job.State); + Logger.LogInformation( + "{Worker}: jobId {JobId} already terminal ({State}) — dropping reappearance", + WorkerName, jobId, job.State); await SafeDeleteAsync(msg, ct).ConfigureAwait(false); return; } @@ -199,18 +121,18 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue // still-running peer's status (plan Wave9 #1 / Wave10 #4). if (job.State == BackupJobState.Running) { - _logger.LogWarning( - "CriticalTableBackupQueueWorker: jobId {JobId} state=Running on dequeue — deferring visibility, not mutating status", - jobId); + Logger.LogWarning( + "{Worker}: jobId {JobId} state=Running on dequeue — deferring visibility, not mutating status", + WorkerName, jobId); try { - await _mainQueue.UpdateMessageAsync(msg.MessageId, msg.PopReceipt, visibilityTimeout: VisibilityTimeout, cancellationToken: ct).ConfigureAwait(false); + await MainQueue.UpdateMessageAsync(msg.MessageId, msg.PopReceipt, visibilityTimeout: VisibilityTimeout, cancellationToken: ct).ConfigureAwait(false); } catch (Exception ex) { - _logger.LogWarning(ex, - "CriticalTableBackupQueueWorker: defer-visibility failed for jobId {JobId} — message will reappear naturally", - jobId); + Logger.LogWarning(ex, + "{Worker}: defer-visibility failed for jobId {JobId} — message will reappear naturally", + WorkerName, jobId); } return; } @@ -228,9 +150,9 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue } catch (LeaseHeldException) { - _logger.LogInformation( - "CriticalTableBackupQueueWorker: maintenance lease held by another op — jobId {JobId} → Skipped", - jobId); + Logger.LogInformation( + "{Worker}: maintenance lease held by another op — jobId {JobId} → Skipped", + WorkerName, jobId); var skippedJob = job; skippedJob.State = BackupJobState.Skipped; skippedJob.LastHeartbeatUtc = _clock.GetUtcNow().UtcDateTime; @@ -245,7 +167,7 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue // Chain a per-handler CTS off the worker's stoppingToken so the renewal loop can // cancel the in-flight backup if the lease is lost mid-run. using var handlerCts = CancellationTokenSource.CreateLinkedTokenSource(ct); - MaintenanceLeaseHolder leaseHolder = new MaintenanceLeaseHolder(lease, handlerCts, _logger); + MaintenanceLeaseHolder leaseHolder = new MaintenanceLeaseHolder(lease, handlerCts, Logger); BackupRunResult? result = null; Exception? retryableException = null; @@ -283,7 +205,7 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue } else { - _logger.LogWarning("CriticalTableBackupQueueWorker: CAS Queued→Running 412 with non-terminal fresh state — leaving message for retry"); + Logger.LogWarning("{Worker}: CAS Queued→Running 412 with non-terminal fresh state — leaving message for retry", WorkerName); } casConflictReturn = true; } @@ -367,9 +289,9 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue } else { - _logger.LogWarning( - "CriticalTableBackupQueueWorker: skipping retry rollback for jobId {JobId} — row no longer Running (now {State})", - jobId, rb.State); + Logger.LogWarning( + "{Worker}: skipping retry rollback for jobId {JobId} — row no longer Running (now {State})", + WorkerName, jobId, rb.State); } } } @@ -429,9 +351,9 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue // Rollback already done UNDER LEASE inside the leased section // (Codex-Hotfix Wave3 #1). Just log + rethrow so the outer poll loop // applies visibility-timeout retry semantics. - _logger.LogWarning(retryableException, - "CriticalTableBackupQueueWorker: jobId {JobId} backup attempt failed — rolled back to Queued for retry (dequeue {N})", - jobId, msg.DequeueCount); + Logger.LogWarning(retryableException, + "{Worker}: jobId {JobId} backup attempt failed — rolled back to Queued for retry (dequeue {N})", + WorkerName, jobId, msg.DequeueCount); throw retryableException; } @@ -458,12 +380,13 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue await SafeDeleteAsync(msg, ct).ConfigureAwait(false); } - // ── Helpers ───────────────────────────────────────────────────────────── + // ── Poison hook ─────────────────────────────────────────────────────────── - private async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct) + protected override async Task BeforePoisonMoveAsync(QueueMessage msg, CancellationToken ct) { - // Plan Wave5 #1 + Wave22 #2: persist Failed BEFORE poison-move so the job - // status never hangs in Running while the message is already poison. + // Plan Wave5 #1 + Wave22 #2: persist Failed BEFORE poison-move so the job status never + // hangs in Running while the message is already poison. Best-effort — on any failure we + // still proceed with the poison move (return true), matching the original semantics. try { var envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); @@ -483,46 +406,9 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue } catch (Exception ex) { - _logger.LogWarning(ex, "CriticalTableBackupQueueWorker: failed-state persistence before poison-move threw — proceeding with poison move anyway"); - } - - try - { - await _poisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - _logger.LogWarning( - "CriticalTableBackupQueueWorker: moved message {Id} to poison queue after {N} failed attempts", - msg.MessageId, msg.DequeueCount - 1); - } - catch (Exception ex) - { - _logger.LogError(ex, "CriticalTableBackupQueueWorker: poison move failed (will retry)"); - } - } - - private async Task SafeDeleteAsync(QueueMessage msg, CancellationToken ct) - { - try - { - await _mainQueue.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "CriticalTableBackupQueueWorker: delete failed for message {Id} — will reappear after visibility-timeout", - msg.MessageId); - } - } - - private async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct) - { - try { await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); } - catch (Exception ex) - { - _logger.LogWarning(ex, - "CriticalTableBackupQueueWorker: CreateIfNotExists failed for {Label} queue — send/receive will retry", - label); + Logger.LogWarning(ex, "{Worker}: failed-state persistence before poison-move threw — proceeding with poison move anyway", WorkerName); } + return true; } /// OpsEvent write must not bubble — backup itself already succeeded (or failed by other means). @@ -531,7 +417,7 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue try { await writeAsync().ConfigureAwait(false); } catch (Exception ex) { - _logger.LogWarning(ex, "CriticalTableBackupQueueWorker: ops-event recording failed — backup state remains authoritative"); + Logger.LogWarning(ex, "{Worker}: ops-event recording failed — backup state remains authoritative", WorkerName); } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs index a021d6e0..0cf435c4 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs @@ -2,12 +2,11 @@ using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; using Azure.Storage.Queues; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models.Deletion; -using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -79,7 +78,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion BlobStorageService blob, AdminConfigurationService adminConfig, IMaintenanceRepository maintenanceRepo, - IConfiguration configuration, + QueueClientFactory queueFactory, ILogger logger) { _storage = storage; @@ -88,27 +87,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion _adminConfig = adminConfig; _maintenanceRepo = maintenanceRepo; _logger = logger; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var queueUri = new Uri($"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.SessionDeletion}"); - _queueClient = new QueueClient(queueUri, new DefaultAzureCredential(), options); - _logger.LogInformation("SessionDeletionProducer initialized with Managed Identity (account: {Account})", storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _queueClient = new QueueClient(connectionString, Constants.QueueNames.SessionDeletion, options); - _logger.LogInformation("SessionDeletionProducer initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } + _queueClient = queueFactory.Create(Constants.QueueNames.SessionDeletion); } /// diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs index 2d9324ae..756fdc47 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs @@ -1,143 +1,67 @@ using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; using Azure.Storage.Queues; using Azure.Storage.Queues.Models; using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models.Deletion; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Newtonsoft.Json; namespace AutopilotMonitor.Functions.Services.Deletion { /// - /// Background poll-loop for the session-deletion queue (plan §5 PR4). Modeled on - /// - /// with four explicit deviations: + /// Background poll-loop for the session-deletion queue (plan §5 PR4). Built on the + /// shared with four explicit deviations: /// /// = 1 (not 32) — a single cascade can be tens of MB /// and minutes of wall-time; one-at-a-time bounds memory and poison blast radius. - /// Kill-switch on entry: every receive checks + /// Kill-switch on entry via : every receive checks /// AdminConfiguration.SessionDeletionKillSwitch; when active the worker idles - /// without dequeuing (envelope stays on the queue, no visibility timeout pressure). - /// Heartbeat: while runs, a sidecar task - /// calls UpdateMessageAsync every to extend the - /// message visibility by . Prevents queue re-delivery - /// from spawning a parallel worker on the same manifest while a step is in flight. - /// Poison OpsEvent: max-dequeue exhaustion records a - /// SessionDeletionPoisoned OpsEvent alongside the poison-queue move so the - /// operator can correlate the poisoned cascade with its manifest (and wire Telegram - /// routing). PR-B audit consolidation: this used to be a per-tenant - /// deletion_poisoned audit but tenant admins should only see the lifecycle - /// endpoints (started / completed / restored), so the signal moved to OpsEvents. - /// The Sessions row's DeletionState is NOT auto-cleared — operator action via - /// restore-from-poisoned (§13, PR4b) is required. + /// without dequeuing. + /// Heartbeat (): the base extends message visibility + /// while runs so re-delivery can't spawn a parallel + /// worker on the same manifest. + /// Poison hooks: transitions + /// Sessions.DeletionState → Poisoned before the move; + /// records a SessionDeletionPoisoned OpsEvent enriched with the handler's last-failure + /// breadcrumb. (PR-B audit consolidation: the signal lives in OpsEvents, not a per-tenant + /// audit; the Sessions row's DeletionState is NOT auto-cleared — operator action via + /// restore-from-poisoned is required.) /// /// - public sealed class SessionDeletionWorker : BackgroundService + public sealed class SessionDeletionWorker : QueuePollingWorker { - /// - /// Explicit deviation from sibling workers — one cascade per receive bounds the worker's - /// memory footprint (a 35MB snapshot blob × 32 messages = 1.1GB) and limits poison-queue - /// blast radius on a corruption signal. Plan §5 PR4. - /// - internal const int BatchSize = 1; - - /// Initial message-visibility window — extended in-flight by the heartbeat task. - internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5); - - /// How often the heartbeat task extends the in-flight message's visibility. - internal static readonly TimeSpan DefaultHeartbeatInterval = TimeSpan.FromSeconds(60); - - /// How much the heartbeat task adds to visibility on each tick. - internal static readonly TimeSpan HeartbeatExtendBy = TimeSpan.FromMinutes(5); - - /// Idle sleep between empty receives or kill-switch-on cycles. - internal static readonly TimeSpan DefaultPollInterval = TimeSpan.FromSeconds(10); - - /// Instance-level overrides — production uses the defaults above, tests inject shorter values via the internal ctor. - private readonly TimeSpan _heartbeatInterval; - private readonly TimeSpan _pollInterval; - - /// Cool-down after an unhandled poll-loop exception. - internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30); - - /// Match the QueueTrigger platform default + sibling workers: 5 attempts → poison. - internal const int MaxDequeueCount = 5; - - private const string PoisonQueueSuffix = "-poison"; - - private readonly QueueClient _mainQueue; - private readonly QueueClient _poisonQueue; private readonly SessionDeletionHandler _handler; private readonly TableStorageService _storage; private readonly AdminConfigurationService _adminConfig; private readonly BlobStorageService _blob; private readonly OpsEventService _opsEvents; - private readonly ILogger _logger; public SessionDeletionWorker( - IConfiguration configuration, + QueueClientFactory queueFactory, SessionDeletionHandler handler, TableStorageService storage, AdminConfigurationService adminConfig, BlobStorageService blob, OpsEventService opsEvents, ILogger logger) + : base(queueFactory, Constants.QueueNames.SessionDeletion, logger) { _handler = handler ?? throw new ArgumentNullException(nameof(handler)); _storage = storage ?? throw new ArgumentNullException(nameof(storage)); _adminConfig = adminConfig ?? throw new ArgumentNullException(nameof(adminConfig)); _blob = blob ?? throw new ArgumentNullException(nameof(blob)); _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - _heartbeatInterval = DefaultHeartbeatInterval; - _pollInterval = DefaultPollInterval; - - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var mainUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.SessionDeletion}"); - var poisonUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.SessionDeletion}{PoisonQueueSuffix}"); - var credential = new DefaultAzureCredential(); - _mainQueue = new QueueClient(mainUri, credential, options); - _poisonQueue = new QueueClient(poisonUri, credential, options); - _logger.LogInformation( - "SessionDeletionWorker initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _mainQueue = new QueueClient( - connectionString, Constants.QueueNames.SessionDeletion, options); - _poisonQueue = new QueueClient( - connectionString, Constants.QueueNames.SessionDeletion + PoisonQueueSuffix, options); - _logger.LogInformation("SessionDeletionWorker initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } } /// - /// Test seam: construct directly with mock instances so the - /// poll loop can be exercised against an in-memory queue without spinning up Azurite. - /// Mirrors 's internal test ctor. + /// Test seam: construct directly with mock instances so the poll + /// loop can be exercised against an in-memory queue without Azurite. Mirrors + /// 's internal test ctor. /// internal SessionDeletionWorker( QueueClient mainQueue, @@ -150,288 +74,147 @@ namespace AutopilotMonitor.Functions.Services.Deletion ILogger logger, TimeSpan? heartbeatInterval = null, TimeSpan? pollInterval = null) + : base(mainQueue, poisonQueue, logger, pollInterval, heartbeatInterval) { - _mainQueue = mainQueue ?? throw new ArgumentNullException(nameof(mainQueue)); - _poisonQueue = poisonQueue ?? throw new ArgumentNullException(nameof(poisonQueue)); _handler = handler ?? throw new ArgumentNullException(nameof(handler)); _storage = storage ?? throw new ArgumentNullException(nameof(storage)); _adminConfig = adminConfig ?? throw new ArgumentNullException(nameof(adminConfig)); _blob = blob ?? throw new ArgumentNullException(nameof(blob)); _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - _heartbeatInterval = heartbeatInterval ?? DefaultHeartbeatInterval; - _pollInterval = pollInterval ?? DefaultPollInterval; } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) + /// + /// Explicit deviation from sibling workers — one cascade per receive bounds the worker's + /// memory footprint (a 35MB snapshot blob × 32 messages = 1.1GB) and limits poison-queue + /// blast radius on a corruption signal. Plan §5 PR4. + /// + protected override int BatchSize => 1; + + protected override bool UseHeartbeat => true; + + protected override async ValueTask ShouldPauseAsync(CancellationToken ct) { - await TryCreateQueueAsync(_mainQueue, "main", stoppingToken).ConfigureAwait(false); - await TryCreateQueueAsync(_poisonQueue, "poison", stoppingToken).ConfigureAwait(false); - - _logger.LogInformation("SessionDeletionWorker: poll loop started (queue {Name})", _mainQueue.Name); - - while (!stoppingToken.IsCancellationRequested) + // Kill-switch entry guard (plan §1 P8 / §9). When active, the worker does not dequeue — + // the message stays visible on the main queue until the operator flips the switch back. + // PR5 finding 1: uncached read so a flip-ON is honored within seconds across all worker + // instances, not minutes (the cache TTL). One extra storage read per poll interval is + // fine — the config row is tiny. + if (await _adminConfig.IsSessionDeletionKillSwitchActiveAsync().ConfigureAwait(false)) { - try - { - // Kill-switch entry guard (plan §1 P8 / §9). When active, the worker does not - // dequeue — the message stays visible on the main queue until the operator - // flips the switch back. Envelope is preserved without losing dequeue budget. - // PR5 finding 1: uncached read so a flip-ON is honored within seconds across - // all worker instances, not minutes (the cache TTL). One extra storage read - // per poll interval (10s default) is fine — config row is tiny. - if (await _adminConfig.IsSessionDeletionKillSwitchActiveAsync().ConfigureAwait(false)) - { - _logger.LogDebug("SessionDeletionWorker: kill-switch active; idling for {Interval}", _pollInterval); - await Task.Delay(_pollInterval, stoppingToken).ConfigureAwait(false); - continue; - } - - var batch = await _mainQueue - .ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken) - .ConfigureAwait(false); - - if (batch?.Value is null || batch.Value.Length == 0) - { - await Task.Delay(_pollInterval, stoppingToken).ConfigureAwait(false); - continue; - } - - foreach (var msg in batch.Value) - { - if (stoppingToken.IsCancellationRequested) break; - await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _logger.LogError(ex, - "SessionDeletionWorker: poll-loop error — backing off {Backoff}", - ErrorBackoff); - try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); } - catch (OperationCanceledException) { break; } - } + Logger.LogDebug("{Worker}: kill-switch active; idling for {Interval}", WorkerName, ResolvedPollInterval); + return true; } - - _logger.LogInformation("SessionDeletionWorker: poll loop stopped"); + return false; } - private async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) + protected override bool TryValidate(SessionDeletionEnvelope envelope) + => !string.IsNullOrEmpty(envelope.TenantId) + && !string.IsNullOrEmpty(envelope.SessionId) + && !string.IsNullOrEmpty(envelope.ManifestId); + + protected override string DescribeForLog(SessionDeletionEnvelope envelope) + => $"tenant={envelope.TenantId} session={envelope.SessionId} manifestId={envelope.ManifestId}"; + + protected override Task HandleAsync(SessionDeletionEnvelope envelope, CancellationToken ct) + => _handler.HandleAsync(envelope, ct); + + // ── Poison hooks ───────────────────────────────────────────────────────── + + protected override async Task BeforePoisonMoveAsync(QueueMessage msg, CancellationToken ct) { - if (msg.DequeueCount > MaxDequeueCount) - { - await MoveToPoisonAsync(msg, ct).ConfigureAwait(false); - return; - } - - SessionDeletionEnvelope? envelope; - try - { - envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); - } - catch (JsonException ex) - { - _logger.LogWarning(ex, - "SessionDeletionWorker: malformed envelope JSON — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - if (envelope is null - || string.IsNullOrEmpty(envelope.TenantId) - || string.IsNullOrEmpty(envelope.SessionId) - || string.IsNullOrEmpty(envelope.ManifestId)) - { - _logger.LogWarning( - "SessionDeletionWorker: envelope missing required fields — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - // Heartbeat: extend visibility every _heartbeatInterval while the handler runs. We - // capture the latest PopReceipt locally because each UpdateMessageAsync returns a - // fresh receipt; the SafeDelete at the end uses whichever receipt was issued last. - using var heartbeatCts = CancellationTokenSource.CreateLinkedTokenSource(ct); - var heartbeatState = new HeartbeatState(msg.MessageId, msg.PopReceipt); - var heartbeatTask = Task.Run(() => HeartbeatLoopAsync(heartbeatState, heartbeatCts.Token), heartbeatCts.Token); - - try - { - await _handler.HandleAsync(envelope, ct).ConfigureAwait(false); - - // Handler succeeded — stop heartbeat and delete using the latest pop-receipt. - heartbeatCts.Cancel(); - try { await heartbeatTask.ConfigureAwait(false); } catch (OperationCanceledException) { } - - await SafeDeleteAsync(msg.MessageId, heartbeatState.PopReceipt, ct).ConfigureAwait(false); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - heartbeatCts.Cancel(); - throw; - } - catch (Exception ex) - { - heartbeatCts.Cancel(); - try { await heartbeatTask.ConfigureAwait(false); } catch (OperationCanceledException) { } - - _logger.LogWarning(ex, - "SessionDeletionWorker: handler failed (tenant={Tenant} session={Session} manifestId={ManifestId} dequeue={N}) — visibility-timeout retry", - envelope.TenantId, envelope.SessionId, envelope.ManifestId, msg.DequeueCount); - } - } - - private async Task HeartbeatLoopAsync(HeartbeatState state, CancellationToken ct) - { - while (!ct.IsCancellationRequested) - { - try { await Task.Delay(_heartbeatInterval, ct).ConfigureAwait(false); } - catch (OperationCanceledException) { return; } - - try - { - var updated = await _mainQueue.UpdateMessageAsync( - state.MessageId, state.PopReceipt, - visibilityTimeout: HeartbeatExtendBy, - cancellationToken: ct).ConfigureAwait(false); - state.PopReceipt = updated.Value.PopReceipt; - _logger.LogDebug( - "SessionDeletionWorker: heartbeat extended visibility for message {Id} by {Extend}", - state.MessageId, HeartbeatExtendBy); - } - catch (OperationCanceledException) { return; } - catch (Exception ex) - { - // Heartbeat failure (e.g. PopReceipt expired because the visibility window - // was missed) is logged and the loop exits. The handler keeps running; on - // its eventual return, SafeDelete will hit a 404 and the message will - // reappear after its current visibility expires — but the handler's own - // ETag-CAS on the progress blob makes a parallel pickup idempotent. - _logger.LogWarning(ex, - "SessionDeletionWorker: heartbeat UpdateMessageAsync failed for message {Id} — heartbeat exiting", - state.MessageId); - return; - } - } - } - - private async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct) - { - // Best-effort: attempt to parse envelope so we can audit the manifest id alongside - // the poison move. If the body is malformed we still move it to the poison queue - // (the audit just lacks the manifestId field). - SessionDeletionEnvelope? envelope = null; - try { envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); } - catch (JsonException) { /* swallow — non-parseable envelopes still get poisoned */ } - // PR4b E1: transition Sessions.DeletionState → Poisoned BEFORE the poison-queue send. // Without this, the row stays at Running (or Queued if poisoned before pickup) and the // restore endpoint cannot dispatch into partial-restore mode. Best-effort: if CAS fails - // (concurrent writer, row gone, state mismatch) we still continue with the poison-queue - // move so the original poison flow completes and an audit trail exists. + // we still proceed with the poison-queue move so the original poison flow completes and + // an audit trail exists (so this hook always returns true). + var envelope = TryDeserialize(msg); if (envelope != null && !string.IsNullOrEmpty(envelope.TenantId) && !string.IsNullOrEmpty(envelope.SessionId)) { await TryTransitionToPoisonedStateAsync(envelope.TenantId, envelope.SessionId, envelope.ManifestId, ct).ConfigureAwait(false); } + return true; + } - try + protected override async Task AfterPoisonMoveAsync(QueueMessage msg, CancellationToken ct) + { + var envelope = TryDeserialize(msg); + if (envelope == null || string.IsNullOrEmpty(envelope.TenantId) || string.IsNullOrEmpty(envelope.SessionId)) { - await _poisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false); - await SafeDeleteAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - _logger.LogWarning( - "SessionDeletionWorker: moved message {Id} to poison queue after {N} failed attempts", - msg.MessageId, msg.DequeueCount - 1); - } - catch (Exception ex) - { - _logger.LogError(ex, - "SessionDeletionWorker: poison move failed for message {Id} (will retry)", - msg.MessageId); return; } - if (envelope != null && !string.IsNullOrEmpty(envelope.TenantId) && !string.IsNullOrEmpty(envelope.SessionId)) + // PR-B audit consolidation: poisoned cascades surface as a global-scope OpsEvent + // (operator-bound, Telegram-routable) rather than a per-tenant audit. Tenant admins + // observe deletion via the lifecycle audits (started / completed / restored). + // + // Codex F4 follow-up: pull the handler's last-failure breadcrumb out of the + // DeletionProgress blob so the OpsEvent carries the root cause (failureType + + // residual sample), not just queue-side metadata. Blob read is best-effort. + string? failureType = null; + string? failureMessage = null; + int? observedResidualCount = null; + string? residualSamplePreviewJson = null; + if (!string.IsNullOrEmpty(envelope.ManifestId)) { - // PR-B audit consolidation: poisoned cascades surface as a global-scope OpsEvent - // (operator-bound, Telegram-routable) rather than a per-tenant audit. Tenant admins - // observe deletion via the lifecycle audits (started / completed / restored). - // - // Codex F4 follow-up: pull the handler's last-failure breadcrumb out of the - // DeletionProgress blob so the OpsEvent carries the root cause (failureType + - // residual sample), not just queue-side metadata. Blob read is best-effort — - // a missing blob, transient 5xx, or pre-followup progress without LastFailure* - // fields still produces a useful OpsEvent, just without root-cause data. - string? failureType = null; - string? failureMessage = null; - int? observedResidualCount = null; - string? residualSamplePreviewJson = null; - if (!string.IsNullOrEmpty(envelope.ManifestId)) - { - try - { - var (progress, _) = await _blob.DownloadDeletionProgressAsync( - envelope.TenantId, envelope.SessionId, envelope.ManifestId!, ct) - .ConfigureAwait(false); - failureType = string.IsNullOrEmpty(progress.LastFailureType) ? null : progress.LastFailureType; - failureMessage = string.IsNullOrEmpty(progress.LastFailureMessage) ? null : progress.LastFailureMessage; - // Codex F2 round-3: this is the verifier's OBSERVED count - // (CascadeVerificationService caps at MaxResidualSampleSize per table AND - // short-circuits after the first failing table — the real residual count - // may be larger). For pre-followup progress blobs lacking the field, fall - // back to the sample-array length so the OpsEvent still carries a number. - observedResidualCount = progress.LastObservedResidualCount - ?? ExtractResidualSampleArrayLength(progress.LastResidualSampleJson); - // Codex F2 round-2: shrink the residual sample further before embedding - // it in the OpsEvent. The OpsEvents table truncates Details at 4096 chars - // (TableOpsEventRepository.cs); the full 50-entry sample plus the other - // payload fields can blow past that and leave the JSON mid-string-corrupt - // — operators end up reading garbled raw text instead of structured data. - // The Session Cleanup admin page already shows the full sample by reading - // the progress blob directly, so the OpsEvent only needs a preview. - residualSamplePreviewJson = ShrinkResidualSampleForOpsEvent(progress.LastResidualSampleJson); - } - catch (Exception blobEx) - { - _logger.LogWarning(blobEx, - "SessionDeletionWorker: failed to read DeletionProgress for poison enrichment (tenant={Tenant} session={Session} manifestId={Manifest}) — OpsEvent will lack root-cause data", - envelope.TenantId, envelope.SessionId, envelope.ManifestId); - } - } - try { - await _opsEvents.RecordSessionDeletionPoisonedAsync( - envelope.TenantId, - envelope.SessionId, - envelope.ManifestId ?? string.Empty, - envelope.Reason ?? string.Empty, - msg.MessageId, - (int)(msg.DequeueCount - 1), - failureType, - failureMessage, - observedResidualCount, - residualSamplePreviewJson).ConfigureAwait(false); + var (progress, _) = await _blob.DownloadDeletionProgressAsync( + envelope.TenantId, envelope.SessionId, envelope.ManifestId!, ct) + .ConfigureAwait(false); + failureType = string.IsNullOrEmpty(progress.LastFailureType) ? null : progress.LastFailureType; + failureMessage = string.IsNullOrEmpty(progress.LastFailureMessage) ? null : progress.LastFailureMessage; + // Codex F2 round-3: this is the verifier's OBSERVED count (capped per table AND + // short-circuited after the first failing table — the real residual count may be + // larger). For pre-followup progress blobs lacking the field, fall back to the + // sample-array length so the OpsEvent still carries a number. + observedResidualCount = progress.LastObservedResidualCount + ?? ExtractResidualSampleArrayLength(progress.LastResidualSampleJson); + // Codex F2 round-2: shrink the residual sample further before embedding it in the + // OpsEvent. The OpsEvents table truncates Details at 4096 chars; the full sample + // can blow past that and corrupt the JSON. The Session Cleanup admin page shows + // the full sample via the progress blob, so the OpsEvent only needs a preview. + residualSamplePreviewJson = ShrinkResidualSampleForOpsEvent(progress.LastResidualSampleJson); } - catch (Exception opsEx) + catch (Exception blobEx) { - _logger.LogError(opsEx, - "SessionDeletionWorker: SessionDeletionPoisoned OpsEvent write failed for tenant={Tenant} session={Session}", - envelope.TenantId, envelope.SessionId); + Logger.LogWarning(blobEx, + "{Worker}: failed to read DeletionProgress for poison enrichment (tenant={Tenant} session={Session} manifestId={Manifest}) — OpsEvent will lack root-cause data", + WorkerName, envelope.TenantId, envelope.SessionId, envelope.ManifestId); } } + + try + { + await _opsEvents.RecordSessionDeletionPoisonedAsync( + envelope.TenantId, + envelope.SessionId, + envelope.ManifestId ?? string.Empty, + envelope.Reason ?? string.Empty, + msg.MessageId, + (int)(msg.DequeueCount - 1), + failureType, + failureMessage, + observedResidualCount, + residualSamplePreviewJson).ConfigureAwait(false); + } + catch (Exception opsEx) + { + Logger.LogError(opsEx, + "{Worker}: SessionDeletionPoisoned OpsEvent write failed for tenant={Tenant} session={Session}", + WorkerName, envelope.TenantId, envelope.SessionId); + } + } + + private static SessionDeletionEnvelope? TryDeserialize(QueueMessage msg) + { + try { return JsonConvert.DeserializeObject(msg.Body.ToString()); } + catch (JsonException) { return null; } } /// - /// Back-compat fallback used only when + /// Back-compat fallback used only when /// is missing (progress blobs written before the field was introduced). Parses the JSON - /// sample array length and uses it as a best-effort substitute for the observed count. - /// Returns null on parse errors or missing data. + /// sample array length and uses it as a best-effort substitute. Returns null on parse errors. /// internal static int? ExtractResidualSampleArrayLength(string? residualSampleJson) { @@ -454,21 +237,9 @@ namespace AutopilotMonitor.Functions.Services.Deletion /// /// Trims the progress-blob's residual sample for safe embedding in the /// SessionDeletionPoisoned OpsEvent's Details column. Three-layer defense - /// against the 4096-char Azure Table truncation discovered by Codex review: - /// - /// Take at most - /// entries (drop tail, the rest stays in the progress blob for the admin UI). - /// Trim each table / pk / rk to - /// - /// with a trailing … marker — a single composite key can be 200+ chars and - /// would otherwise consume most of the budget on its own. - /// If the serialized JSON still exceeds - /// , - /// drop trailing entries until it fits. Worst case (all entries too long) returns - /// "[]"; operators still get failureType + observedResidualCount and can - /// fetch the full sample from the progress blob. - /// - /// Returns null when there's nothing to embed or the input cannot be parsed. + /// against the 4096-char Azure Table truncation: cap entry count, trim each key field, then + /// drop trailing entries until the serialized JSON fits the budget. Returns null when there's + /// nothing to embed or the input cannot be parsed. /// internal static string? ShrinkResidualSampleForOpsEvent(string? fullSampleJson) { @@ -483,7 +254,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion return null; } entries = new List(); - var max = AutopilotMonitor.Shared.Models.Deletion.DeletionProgressConstants.OpsEventResidualSamplePreviewSize; + var max = DeletionProgressConstants.OpsEventResidualSamplePreviewSize; foreach (var element in doc.RootElement.EnumerateArray()) { if (entries.Count >= max) break; @@ -501,10 +272,10 @@ namespace AutopilotMonitor.Functions.Services.Deletion return null; } - // Total-budget pass: serialize, then drop trailing entries until under the cap. - // Worst case (one over-budget entry) ends at "[]" — still valid JSON, and the full - // sample remains in the progress blob. - var budget = AutopilotMonitor.Shared.Models.Deletion.DeletionProgressConstants.OpsEventResidualPreviewBudgetChars; + // Total-budget pass: serialize, then drop trailing entries until under the cap. Worst + // case (one over-budget entry) ends at "[]" — still valid JSON; the full sample remains + // in the progress blob. + var budget = DeletionProgressConstants.OpsEventResidualPreviewBudgetChars; while (true) { var serialized = SerializeResidualPreview(entries); @@ -535,10 +306,10 @@ namespace AutopilotMonitor.Functions.Services.Deletion private static string TrimResidualKeyField(string? raw) { if (string.IsNullOrEmpty(raw)) return string.Empty; - var max = AutopilotMonitor.Shared.Models.Deletion.DeletionProgressConstants.OpsEventResidualKeyMaxChars; + var max = DeletionProgressConstants.OpsEventResidualKeyMaxChars; if (raw!.Length <= max) return raw; - // -1 makes room for the trailing ellipsis so the marked-truncated string fits in - // the same budget as a max-length untrimmed one. + // -1 makes room for the trailing ellipsis so the marked-truncated string fits in the + // same budget as a max-length untrimmed one. return raw.Substring(0, max - 1) + "…"; } @@ -555,74 +326,32 @@ namespace AutopilotMonitor.Functions.Services.Deletion public string Rk; } - private async Task SafeDeleteAsync(QueueMessage msg, CancellationToken ct) - => await SafeDeleteAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - - private async Task SafeDeleteAsync(string messageId, string popReceipt, CancellationToken ct) - { - try - { - await _mainQueue.DeleteMessageAsync(messageId, popReceipt, ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "SessionDeletionWorker: delete failed for message {Id} (will reappear after visibility-timeout)", - messageId); - } - } - - private async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct) - { - try - { - await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "SessionDeletionWorker: CreateIfNotExists failed for {Label} queue — will continue, send/receive will retry", - label); - } - } - /// /// PR4b E1 + PR4c F5: CAS Sessions.DeletionState to Poisoned with manifestId pre-check. - /// Tries Running → Poisoned first (the common case — Worker successfully CAS'd - /// Queued→Running on first pickup and the handler then failed N times). Falls back to - /// Queued → Poisoned for the rare case where the cascade poisoned before the - /// Worker ever successfully transitioned to Running (handler always threw at the - /// state-acquire step). Logs warning on persistent CAS failure but does NOT throw — the - /// poison-queue move + audit must still complete so the operator has an observable trail - /// even if the row state is somehow inconsistent. - /// - /// PR4c F5: before issuing any CAS, read the Sessions row's - /// PendingDeletionManifestId and compare to the envelope's manifestId. If - /// they don't match, the envelope is stale (from a prior cascade attempt) — do NOT - /// touch the state of a possibly-fresh active cascade. The poison-queue + audit still - /// fire in the caller so the stale message is removed from the main queue. - /// + /// Tries Running → Poisoned first (the common case), falls back to + /// Queued → Poisoned (poisoned before pickup ever succeeded). Logs warning on + /// persistent CAS failure but does NOT throw — the poison-queue move + audit must still + /// complete. PR4c F5: before any CAS, read the row's PendingDeletionManifestId and + /// compare to the envelope's manifestId; a stale envelope must not flip a fresh + /// active cascade. /// private async Task TryTransitionToPoisonedStateAsync(string tenantId, string sessionId, string? manifestId, CancellationToken ct) { - // PR4c F5: pre-check that the envelope's manifestId matches the row's - // PendingDeletionManifestId. A stale envelope from a prior cascade must not flip a - // fresh active cascade to Poisoned. var row = await _storage.GetSessionRowAsync(tenantId, sessionId, ct).ConfigureAwait(false); if (row == null) { - _logger.LogInformation( - "SessionDeletionWorker: Sessions row gone for tenant={Tenant} session={Session} (cascade already tombstoned by another worker?) — skipping state transition", - tenantId, sessionId); + Logger.LogInformation( + "{Worker}: Sessions row gone for tenant={Tenant} session={Session} (cascade already tombstoned by another worker?) — skipping state transition", + WorkerName, tenantId, sessionId); return; } var currentPending = row.GetString("PendingDeletionManifestId"); if (!string.Equals(currentPending, manifestId, StringComparison.Ordinal)) { - _logger.LogWarning( - "SessionDeletionWorker: stale poison envelope for tenant={Tenant} session={Session} — envelope.ManifestId={Envelope} but row.PendingDeletionManifestId={Pending}; skipping state transition (active cascade is a different manifest)", - tenantId, sessionId, manifestId, currentPending); + Logger.LogWarning( + "{Worker}: stale poison envelope for tenant={Tenant} session={Session} — envelope.ManifestId={Envelope} but row.PendingDeletionManifestId={Pending}; skipping state transition (active cascade is a different manifest)", + WorkerName, tenantId, sessionId, manifestId, currentPending); return; } @@ -635,9 +364,9 @@ namespace AutopilotMonitor.Functions.Services.Deletion ct).ConfigureAwait(false); if (cas.Outcome == TableStorageService.SessionDeletionStateCasOutcome.Updated) { - _logger.LogInformation( - "SessionDeletionWorker: CAS Running→Poisoned succeeded for tenant={Tenant} session={Session} manifestId={ManifestId}", - tenantId, sessionId, manifestId); + Logger.LogInformation( + "{Worker}: CAS Running→Poisoned succeeded for tenant={Tenant} session={Session} manifestId={ManifestId}", + WorkerName, tenantId, sessionId, manifestId); return; } @@ -653,24 +382,17 @@ namespace AutopilotMonitor.Functions.Services.Deletion ct).ConfigureAwait(false); if (cas.Outcome == TableStorageService.SessionDeletionStateCasOutcome.Updated) { - _logger.LogInformation( - "SessionDeletionWorker: CAS Queued→Poisoned succeeded for tenant={Tenant} session={Session} manifestId={ManifestId}", - tenantId, sessionId, manifestId); + Logger.LogInformation( + "{Worker}: CAS Queued→Poisoned succeeded for tenant={Tenant} session={Session} manifestId={ManifestId}", + WorkerName, tenantId, sessionId, manifestId); return; } } // Persistent CAS failure — log and continue so the poison-queue move + audit still fire. - _logger.LogWarning( - "SessionDeletionWorker: could not transition Sessions.DeletionState to Poisoned for tenant={Tenant} session={Session} manifestId={ManifestId} outcome={Outcome} currentState={State}; restore endpoint may reject with 409 \"active_cascade\" until operator manually intervenes.", - tenantId, sessionId, manifestId, cas.Outcome, cas.CurrentState); - } - - private sealed class HeartbeatState - { - public string MessageId { get; } - public string PopReceipt { get; set; } - public HeartbeatState(string messageId, string popReceipt) { MessageId = messageId; PopReceipt = popReceipt; } + Logger.LogWarning( + "{Worker}: could not transition Sessions.DeletionState to Poisoned for tenant={Tenant} session={Session} manifestId={ManifestId} outcome={Outcome} currentState={State}; restore endpoint may reject with 409 \"active_cascade\" until operator manually intervenes.", + WorkerName, tenantId, sessionId, manifestId, cas.Outcome, cas.CurrentState); } } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Indexing/AzureQueueIndexReconcileProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Indexing/AzureQueueIndexReconcileProducer.cs index b018d95c..19001b05 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Indexing/AzureQueueIndexReconcileProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Indexing/AzureQueueIndexReconcileProducer.cs @@ -2,12 +2,11 @@ using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; using Azure.Storage.Queues; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -38,43 +37,15 @@ namespace AutopilotMonitor.Functions.Services.Indexing private int _queueEnsured; // 0 = not yet ensured, 1 = CreateIfNotExistsAsync has run public AzureQueueIndexReconcileProducer( - IConfiguration configuration, + QueueClientFactory queueFactory, AdminConfigurationService adminConfig, ILogger logger) { _adminConfig = adminConfig; _logger = logger; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - // Match Azure Functions QueueTrigger default (Base64) so the M5.d.3 consumer - // binding can decode our messages without an explicit QueueMessageEncoding override. - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var queueUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.TelemetryIndexReconcile}"); - _queueClient = new QueueClient(queueUri, new DefaultAzureCredential(), options); - _logger.LogInformation( - "IndexReconcile producer initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _queueClient = new QueueClient( - connectionString, Constants.QueueNames.TelemetryIndexReconcile, options); - _logger.LogInformation("IndexReconcile producer initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } + // Base64 matches the Azure Functions QueueTrigger default so the M5.d.3 consumer + // binding can decode our messages without an explicit override. + _queueClient = queueFactory.Create(Constants.QueueNames.TelemetryIndexReconcile); } public async Task EnqueueBatchAsync( diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs index 6bb876d5..8e6215ca 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs @@ -1,273 +1,34 @@ using System; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; -using Azure.Storage.Queues; -using Azure.Storage.Queues.Models; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; -using Newtonsoft.Json; namespace AutopilotMonitor.Functions.Services.Indexing { /// /// Background worker for the telemetry-index-reconcile queue (Plan §2.8, §M5.d.3). - /// - /// Replaces the earlier IndexReconcileFunction QueueTrigger because the trigger - /// binding required a Functions-host-specific connection app-setting - /// (<Connection>__queueServiceUri) that the rest of the project's storage - /// access does not need. This worker matches the - /// , and - /// pattern: pure DI service driven by the existing - /// AzureStorageAccountName (Managed Identity) or AzureTableStorageConnectionString - /// fallback — no bespoke binding glue. - /// - /// - /// Plattform-Parität zum QueueTrigger: - /// - /// Receive batch of up to messages with - /// visibility-timeout per message. - /// Handler success → DeleteMessageAsync. - /// Handler exception → no delete; message becomes visible again after the - /// timeout, DequeueCount increments. Identical to platform-retry semantics. - /// After failed attempts the message is moved to - /// telemetry-index-reconcile-poison and deleted from the main queue. Same - /// behaviour as the QueueTrigger platform. - /// Empty receive → sleep for and retry. - /// Unrecoverable poll-loop exception (e.g. transient Storage outage) → log + back - /// off for before resuming. - /// - /// + /// Replaces the earlier IndexReconcileFunction QueueTrigger (whose binding required a + /// Functions-host-specific connection app-setting the project's storage access does not need) + /// with the shared poll loop. Vanilla defaults: + /// batch 32, 5-min visibility, 5 attempts → poison. /// - public sealed class IndexReconcileQueueWorker : BackgroundService + public sealed class IndexReconcileQueueWorker : QueuePollingWorker { - /// Max messages received per poll. Storage Queue caps batch-receive at 32. - internal const int BatchSize = 32; - - /// - /// Visibility timeout per received message — long enough to cover the slowest - /// expected path (5 index-table writes). - /// - internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5); - - /// Idle sleep between empty receives. - internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10); - - /// Cool-down after an unhandled poll-loop exception. - internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30); - - /// - /// Match the QueueTrigger platform default: 5 attempts, then move to poison. - /// (Storage Queue's DequeueCount increments on each visibility-timeout reappear.) - /// - internal const int MaxDequeueCount = 5; - - private const string PoisonQueueSuffix = "-poison"; - - private readonly QueueClient _mainQueue; - private readonly QueueClient _poisonQueue; private readonly IndexReconcileHandler _handler; - private readonly ILogger _logger; public IndexReconcileQueueWorker( - IConfiguration configuration, + QueueClientFactory queueFactory, IndexReconcileHandler handler, ILogger logger) + : base(queueFactory, Constants.QueueNames.TelemetryIndexReconcile, logger) { _handler = handler ?? throw new ArgumentNullException(nameof(handler)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - - // Match the producer's binding base64 encoding so messages round-trip cleanly. - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var mainUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.TelemetryIndexReconcile}"); - var poisonUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.TelemetryIndexReconcile}{PoisonQueueSuffix}"); - var credential = new DefaultAzureCredential(); - _mainQueue = new QueueClient(mainUri, credential, options); - _poisonQueue = new QueueClient(poisonUri, credential, options); - _logger.LogInformation( - "IndexReconcileQueueWorker initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _mainQueue = new QueueClient( - connectionString, Constants.QueueNames.TelemetryIndexReconcile, options); - _poisonQueue = new QueueClient( - connectionString, Constants.QueueNames.TelemetryIndexReconcile + PoisonQueueSuffix, options); - _logger.LogInformation("IndexReconcileQueueWorker initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - // Ensure both queues exist before the first receive — the producer creates the - // main queue on first send too, but having the poison queue ready avoids losing - // the first poison-move attempt. - await TryCreateQueueAsync(_mainQueue, "main", stoppingToken).ConfigureAwait(false); - await TryCreateQueueAsync(_poisonQueue, "poison", stoppingToken).ConfigureAwait(false); - - _logger.LogInformation("IndexReconcileQueueWorker: poll loop started (queue {Name})", - _mainQueue.Name); - - while (!stoppingToken.IsCancellationRequested) - { - try - { - var batch = await _mainQueue - .ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken) - .ConfigureAwait(false); - - if (batch?.Value is null || batch.Value.Length == 0) - { - await Task.Delay(PollInterval, stoppingToken).ConfigureAwait(false); - continue; - } - - foreach (var msg in batch.Value) - { - if (stoppingToken.IsCancellationRequested) break; - await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _logger.LogError(ex, - "IndexReconcileQueueWorker: poll-loop error — backing off {Backoff}", - ErrorBackoff); - try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); } - catch (OperationCanceledException) { break; } - } - } - - _logger.LogInformation("IndexReconcileQueueWorker: poll loop stopped"); - } - - private async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) - { - // Plattform-Parität: nach MaxDequeueCount Versuchen Move-to-poison + Delete-from-main. - // DequeueCount > 5 bedeutet: 5 erfolglose Verarbeitungs-Versuche sind durch, der 6. - // Receive ist jetzt — den schicken wir nicht mehr an den Handler. - if (msg.DequeueCount > MaxDequeueCount) - { - await MoveToPoisonAsync(msg, ct).ConfigureAwait(false); - return; - } - - IndexReconcileEnvelope? envelope; - try - { - envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); - } - catch (JsonException ex) - { - // Malformed JSON is permanent; drop directly so it doesn't retry-loop and waste - // dequeue attempts. Same call as the old QueueTrigger function. - _logger.LogWarning(ex, - "IndexReconcileQueueWorker: malformed envelope JSON — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - if (envelope is null) - { - _logger.LogWarning( - "IndexReconcileQueueWorker: null envelope after deserialization — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - try - { - await _handler.HandleAsync(envelope, ct).ConfigureAwait(false); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - // Shutdown — let the message become visible again so the next worker run picks it up. - throw; - } - catch (Exception ex) - { - // Don't delete: visibility-timeout will make the message visible again, DequeueCount - // increments, eventual move-to-poison after MaxDequeueCount. Identical retry shape - // to the QueueTrigger platform. - _logger.LogWarning(ex, - "IndexReconcileQueueWorker: handler failed (msg {Id}, dequeue {N}) — visibility-timeout retry", - msg.MessageId, msg.DequeueCount); - } - } - - private async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct) - { - try - { - await _poisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - _logger.LogWarning( - "IndexReconcileQueueWorker: moved message {Id} to poison queue after {N} failed attempts", - msg.MessageId, msg.DequeueCount - 1); - } - catch (Exception ex) - { - // If poison-enqueue itself fails, leave the message — next visibility-timeout - // will retry the move. Avoids losing data on transient outages. - _logger.LogError(ex, - "IndexReconcileQueueWorker: poison move failed for message {Id} (will retry)", - msg.MessageId); - } - } - - private async Task SafeDeleteAsync(QueueMessage msg, CancellationToken ct) - { - try - { - await _mainQueue.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "IndexReconcileQueueWorker: delete failed for message {Id} (will reappear after visibility-timeout)", - msg.MessageId); - } - } - - private async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct) - { - try - { - await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "IndexReconcileQueueWorker: CreateIfNotExists failed for {Label} queue — will continue, send/receive will retry", - label); - } - } + protected override Task HandleAsync(IndexReconcileEnvelope envelope, CancellationToken ct) + => _handler.HandleAsync(envelope, ct); } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Monitoring/AzurePoisonQueueProbe.cs b/src/Backend/AutopilotMonitor.Functions/Services/Monitoring/AzurePoisonQueueProbe.cs index d03d9f83..dd9355b0 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Monitoring/AzurePoisonQueueProbe.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Monitoring/AzurePoisonQueueProbe.cs @@ -3,52 +3,29 @@ using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; using Azure; -using Azure.Identity; using Azure.Storage.Queues; -using Microsoft.Extensions.Configuration; +using AutopilotMonitor.Functions.Services.Queueing; namespace AutopilotMonitor.Functions.Services.Monitoring { /// /// Azure Storage Queue implementation of . - /// Mirrors the connection-resolution pattern used by the existing queue workers - /// (AnalyzeOnEnrollmentEndQueueWorker, IndexReconcileQueueWorker, - /// VulnerabilityCorrelateQueueWorker): prefer Managed Identity via - /// AzureStorageAccountName, fall back to AzureTableStorageConnectionString. - /// QueueClient instances are cached per queue name — they are thread-safe and - /// cheap to keep around for the lifetime of the Functions host. + /// Builds clients via the shared (Managed Identity or + /// connection-string fallback). QueueClient instances are cached per queue name — they are + /// thread-safe and cheap to keep around for the lifetime of the Functions host. + /// + /// Probes only read GetPropertiesAsync message counts, so they request the factory's + /// non-Base64 client (encoding is irrelevant for a metadata read). + /// /// public sealed class AzurePoisonQueueProbe : IPoisonQueueProbe { private readonly ConcurrentDictionary _clients = new(); - private readonly Func _clientFactory; + private readonly QueueClientFactory _queueFactory; - public AzurePoisonQueueProbe(IConfiguration configuration) + public AzurePoisonQueueProbe(QueueClientFactory queueFactory) { - if (configuration is null) throw new ArgumentNullException(nameof(configuration)); - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var credential = new DefaultAzureCredential(); - _clientFactory = queueName => - { - var uri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{queueName}"); - return new QueueClient(uri, credential); - }; - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _clientFactory = queueName => new QueueClient(connectionString, queueName); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } + _queueFactory = queueFactory ?? throw new ArgumentNullException(nameof(queueFactory)); } public async Task GetApproximateMessageCountAsync(string queueName, CancellationToken ct) @@ -56,7 +33,7 @@ namespace AutopilotMonitor.Functions.Services.Monitoring if (string.IsNullOrWhiteSpace(queueName)) throw new ArgumentException("Queue name must not be empty.", nameof(queueName)); - var client = _clients.GetOrAdd(queueName, _clientFactory); + var client = _clients.GetOrAdd(queueName, name => _queueFactory.Create(name, base64: false)); try { diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/AzureQueueTenantOffboardingEnqueuer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/AzureQueueTenantOffboardingEnqueuer.cs index 1b24231c..70cfbf63 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/AzureQueueTenantOffboardingEnqueuer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/AzureQueueTenantOffboardingEnqueuer.cs @@ -1,11 +1,10 @@ using System; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; using Azure.Storage.Queues; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models.Offboarding; -using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -26,39 +25,11 @@ namespace AutopilotMonitor.Functions.Services.Offboarding private int _queueEnsured; public AzureQueueTenantOffboardingEnqueuer( - IConfiguration configuration, + QueueClientFactory queueFactory, ILogger logger) { _logger = logger; - - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var queueUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.TenantOffboarding}"); - _queueClient = new QueueClient(queueUri, new DefaultAzureCredential(), options); - _logger.LogInformation( - "TenantOffboarding enqueuer initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _queueClient = new QueueClient( - connectionString, Constants.QueueNames.TenantOffboarding, options); - _logger.LogInformation("TenantOffboarding enqueuer initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } + _queueClient = queueFactory.Create(Constants.QueueNames.TenantOffboarding); } /// Test seam: bind directly to a (possibly mocked) . diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs index 48e2cb55..6fb667dd 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs @@ -1,96 +1,45 @@ using System; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; using Azure.Storage.Queues; using Azure.Storage.Queues.Models; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models.Offboarding; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Newtonsoft.Json; namespace AutopilotMonitor.Functions.Services.Offboarding { /// - /// Background poll-loop for the tenant-offboarding queue (Plan §7.3). Modeled on - /// : + /// Background poll-loop for the tenant-offboarding queue (Plan §7.3). Built on the + /// shared : /// /// = 1 — one tenant cascade per receive bounds memory - /// (per-tenant enumerate + per-session enqueue is long-running) and limits the - /// blast radius of a poison. - /// Heartbeat extends visibility every while - /// the handler runs so concurrent re-delivery can't spawn a parallel offboarding - /// for the same tenant. - /// Max-dequeue → poison queue + the handler's own - /// Failed-state transition (the handler emits TenantOffboardingFailed when - /// it fails closed; the worker emits one too on dequeue exhaustion so an envelope - /// that never reaches the handler still surfaces). + /// (per-tenant enumerate + per-session enqueue is long-running) and limits the blast + /// radius of a poison. + /// Heartbeat () extends visibility while the handler runs so + /// concurrent re-delivery can't spawn a parallel offboarding for the same tenant. + /// Max-dequeue → poison queue, but persists the + /// handler's Failed-state transition FIRST and vetoes the move if that fails — otherwise a + /// transient failure would leave the message in poison while the Tenant hangs at + /// InProgress. /// /// - public sealed class TenantOffboardingWorker : BackgroundService + public sealed class TenantOffboardingWorker : QueuePollingWorker { - internal const int BatchSize = 1; - - internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5); - internal static readonly TimeSpan DefaultHeartbeatInterval = TimeSpan.FromSeconds(60); - internal static readonly TimeSpan HeartbeatExtendBy = TimeSpan.FromMinutes(5); - internal static readonly TimeSpan DefaultPollInterval = TimeSpan.FromSeconds(10); - internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30); - - internal const int MaxDequeueCount = 5; - private const string PoisonQueueSuffix = "-poison"; - - private readonly QueueClient _mainQueue; - private readonly QueueClient _poisonQueue; private readonly TenantOffboardingHandler _handler; private readonly OpsEventService _opsEvents; - private readonly ILogger _logger; - - private readonly TimeSpan _heartbeatInterval; - private readonly TimeSpan _pollInterval; public TenantOffboardingWorker( - IConfiguration configuration, + QueueClientFactory queueFactory, TenantOffboardingHandler handler, OpsEventService opsEvents, ILogger logger) + : base(queueFactory, Constants.QueueNames.TenantOffboarding, logger, Constants.QueueNames.TenantOffboardingPoison) { _handler = handler ?? throw new ArgumentNullException(nameof(handler)); _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - _heartbeatInterval = DefaultHeartbeatInterval; - _pollInterval = DefaultPollInterval; - - var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var mainUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.TenantOffboarding}"); - var poisonUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.TenantOffboardingPoison}"); - var credential = new DefaultAzureCredential(); - _mainQueue = new QueueClient(mainUri, credential, options); - _poisonQueue = new QueueClient(poisonUri, credential, options); - _logger.LogInformation( - "TenantOffboardingWorker initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _mainQueue = new QueueClient(connectionString, Constants.QueueNames.TenantOffboarding, options); - _poisonQueue = new QueueClient(connectionString, Constants.QueueNames.TenantOffboardingPoison, options); - _logger.LogInformation("TenantOffboardingWorker initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } } /// Test seam — inject mocked queues + shorter intervals for unit tests. @@ -102,165 +51,39 @@ namespace AutopilotMonitor.Functions.Services.Offboarding ILogger logger, TimeSpan? heartbeatInterval = null, TimeSpan? pollInterval = null) + : base(mainQueue, poisonQueue, logger, pollInterval, heartbeatInterval) { - _mainQueue = mainQueue ?? throw new ArgumentNullException(nameof(mainQueue)); - _poisonQueue = poisonQueue ?? throw new ArgumentNullException(nameof(poisonQueue)); _handler = handler ?? throw new ArgumentNullException(nameof(handler)); _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - _heartbeatInterval = heartbeatInterval ?? DefaultHeartbeatInterval; - _pollInterval = pollInterval ?? DefaultPollInterval; } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - await TryCreateQueueAsync(_mainQueue, "main", stoppingToken).ConfigureAwait(false); - await TryCreateQueueAsync(_poisonQueue, "poison", stoppingToken).ConfigureAwait(false); + protected override int BatchSize => 1; - _logger.LogInformation("TenantOffboardingWorker: poll loop started (queue {Name})", _mainQueue.Name); + protected override bool UseHeartbeat => true; - while (!stoppingToken.IsCancellationRequested) - { - try - { - var batch = await _mainQueue - .ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken) - .ConfigureAwait(false); + protected override bool TryValidate(TenantOffboardingEnvelope envelope) + => !string.IsNullOrEmpty(envelope.TenantId) + && !string.IsNullOrEmpty(envelope.HistoryRowKey); - if (batch?.Value is null || batch.Value.Length == 0) - { - await Task.Delay(_pollInterval, stoppingToken).ConfigureAwait(false); - continue; - } + protected override string DescribeForLog(TenantOffboardingEnvelope envelope) + => $"tenant={envelope.TenantId} history={envelope.HistoryRowKey}"; - foreach (var msg in batch.Value) - { - if (stoppingToken.IsCancellationRequested) break; - await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _logger.LogError(ex, - "TenantOffboardingWorker: poll-loop error — backing off {Backoff}", - ErrorBackoff); - try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); } - catch (OperationCanceledException) { break; } - } - } + protected override Task HandleAsync(TenantOffboardingEnvelope envelope, CancellationToken ct) + => _handler.HandleAsync(envelope, ct); - _logger.LogInformation("TenantOffboardingWorker: poll loop stopped"); - } - - private async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) - { - if (msg.DequeueCount > MaxDequeueCount) - { - await MoveToPoisonAsync(msg, ct).ConfigureAwait(false); - return; - } - - TenantOffboardingEnvelope? envelope; - try - { - envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); - } - catch (JsonException ex) - { - _logger.LogWarning(ex, - "TenantOffboardingWorker: malformed envelope JSON — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - return; - } - - if (envelope is null - || string.IsNullOrEmpty(envelope.TenantId) - || string.IsNullOrEmpty(envelope.HistoryRowKey)) - { - _logger.LogWarning( - "TenantOffboardingWorker: envelope missing required fields — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - return; - } - - using var heartbeatCts = CancellationTokenSource.CreateLinkedTokenSource(ct); - var heartbeatState = new HeartbeatState(msg.MessageId, msg.PopReceipt); - var heartbeatTask = Task.Run(() => HeartbeatLoopAsync(heartbeatState, heartbeatCts.Token), heartbeatCts.Token); - - try - { - await _handler.HandleAsync(envelope, ct).ConfigureAwait(false); - - heartbeatCts.Cancel(); - try { await heartbeatTask.ConfigureAwait(false); } catch (OperationCanceledException) { } - - await SafeDeleteAsync(msg.MessageId, heartbeatState.PopReceipt, ct).ConfigureAwait(false); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - heartbeatCts.Cancel(); - throw; - } - catch (Exception ex) - { - heartbeatCts.Cancel(); - try { await heartbeatTask.ConfigureAwait(false); } catch (OperationCanceledException) { } - - _logger.LogWarning(ex, - "TenantOffboardingWorker: handler failed (tenant={Tenant} history={History} dequeue={N}) — visibility-timeout retry", - envelope.TenantId, envelope.HistoryRowKey, msg.DequeueCount); - } - } - - private async Task HeartbeatLoopAsync(HeartbeatState state, CancellationToken ct) - { - while (!ct.IsCancellationRequested) - { - try { await Task.Delay(_heartbeatInterval, ct).ConfigureAwait(false); } - catch (OperationCanceledException) { return; } - - try - { - var updated = await _mainQueue.UpdateMessageAsync( - state.MessageId, state.PopReceipt, - visibilityTimeout: HeartbeatExtendBy, - cancellationToken: ct).ConfigureAwait(false); - state.PopReceipt = updated.Value.PopReceipt; - _logger.LogDebug( - "TenantOffboardingWorker: heartbeat extended visibility for message {Id} by {Extend}", - state.MessageId, HeartbeatExtendBy); - } - catch (OperationCanceledException) { return; } - catch (Exception ex) - { - _logger.LogWarning(ex, - "TenantOffboardingWorker: heartbeat UpdateMessageAsync failed for message {Id} — exiting heartbeat", - state.MessageId); - return; - } - } - } - - private async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct) + protected override async Task BeforePoisonMoveAsync(QueueMessage msg, CancellationToken ct) { TenantOffboardingEnvelope? envelope = null; try { envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); } - catch (JsonException) { /* still poison the malformed envelope below */ } + catch (JsonException) { /* still poison the malformed envelope */ } // Review-Fix (second-pass) Finding 1: the durable Failed-state transition MUST land // BEFORE the poison-queue send + main-queue delete. Otherwise a transient storage / // OpsEvent / pointer-CAS failure on the transition would leave the message in poison - // but the Tenant stuck at InProgress. By doing the transition first we keep the - // message visible on the main queue when the transition fails — the worker will - // retry on the next dequeue. The transition itself is idempotent - // (MarkEnvelopeFailedFromPoisonAsync no-ops on terminal History.Status) so retries - // are safe. + // but the Tenant stuck at InProgress. By vetoing the poison move on transition failure + // we keep the message visible on the main queue — the worker retries on the next + // dequeue. The transition is idempotent (MarkEnvelopeFailedFromPoisonAsync no-ops on a + // terminal History.Status), so retries are safe. if (envelope != null && !string.IsNullOrEmpty(envelope.TenantId) && !string.IsNullOrEmpty(envelope.HistoryRowKey)) @@ -268,78 +91,25 @@ namespace AutopilotMonitor.Functions.Services.Offboarding try { await _handler.MarkEnvelopeFailedFromPoisonAsync( - envelope, (int)(msg.DequeueCount - 1), ct); + envelope, (int)(msg.DequeueCount - 1), ct).ConfigureAwait(false); } catch (Exception ex) { // Critical: leave the message visible so the next dequeue can retry the - // transition. We do NOT proceed to the poison send / SafeDelete here — - // otherwise the operator would see a queue with nothing in it and a tenant - // hanging on InProgress. - _logger.LogError(ex, - "TenantOffboardingWorker: MarkEnvelopeFailedFromPoisonAsync failed for tenant={Tenant} history={History} — message left visible for retry (durable Failed state has higher priority than poison move)", - envelope.TenantId, envelope.HistoryRowKey); - return; + // transition. We do NOT proceed to the poison send / SafeDelete — otherwise the + // operator would see an empty queue and a tenant hanging on InProgress. + Logger.LogError(ex, + "{Worker}: MarkEnvelopeFailedFromPoisonAsync failed for tenant={Tenant} history={History} — message left visible for retry (durable Failed state has higher priority than poison move)", + WorkerName, envelope.TenantId, envelope.HistoryRowKey); + return false; } } // Failed state is now durable (or envelope was malformed and couldn't transition). - // Best-effort poison move + SafeDelete; transient failure here leaves the message - // visible but the next pickup will see History.Status=Failed and SafeDelete cleanly. - try - { - await _poisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false); - await SafeDeleteAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - _logger.LogWarning( - "TenantOffboardingWorker: moved message {Id} to poison queue after {N} failed attempts", - msg.MessageId, msg.DequeueCount - 1); - } - catch (Exception ex) - { - _logger.LogError(ex, - "TenantOffboardingWorker: poison move failed for message {Id} (will retry; durable Failed state already written)", - msg.MessageId); - } - } - - private async Task SafeDeleteAsync(string messageId, string popReceipt, CancellationToken ct) - { - try - { - await _mainQueue.DeleteMessageAsync(messageId, popReceipt, ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "TenantOffboardingWorker: DeleteMessageAsync failed for message {Id} — message will reappear after visibility expires", - messageId); - } - } - - private async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct) - { - try - { - await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "TenantOffboardingWorker: CreateIfNotExists failed on {Label} queue {Name}", - label, queue.Name); - } - } - - private sealed class HeartbeatState - { - public string MessageId { get; } - public string PopReceipt { get; set; } - - public HeartbeatState(string messageId, string popReceipt) - { - MessageId = messageId; - PopReceipt = popReceipt; - } + // The base proceeds with a best-effort poison move + SafeDelete; transient failure + // there leaves the message visible but the next pickup sees History.Status=Failed and + // SafeDeletes cleanly. + return true; } } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs new file mode 100644 index 00000000..b081efee --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs @@ -0,0 +1,91 @@ +using System; +using Azure.Identity; +using Azure.Storage.Queues; +using Microsoft.Extensions.Configuration; + +namespace AutopilotMonitor.Functions.Services.Queueing +{ + /// + /// Single source of truth for building instances against the + /// project's storage account. Every queue producer / worker / probe used to hand-roll the + /// same AzureStorageAccountName (Managed Identity) vs + /// AzureTableStorageConnectionString (fallback) branch — ~25 duplicated lines per + /// site across 13 sites (code-quality audit 2026-05-29, finding D2). This factory resolves + /// the credential model once in its constructor and hands out clients on demand. + /// + /// Registered as a singleton; the resolved credential / connection string are captured in + /// closures so each call is allocation-cheap and never + /// re-reads configuration. + /// + /// + public sealed class QueueClientFactory + { + private const string PoisonQueueSuffix = "-poison"; + + private readonly Func _builder; + + public QueueClientFactory(IConfiguration configuration) + { + if (configuration is null) throw new ArgumentNullException(nameof(configuration)); + + var storageAccountName = configuration["AzureStorageAccountName"]; + var connectionString = configuration["AzureTableStorageConnectionString"]; + + if (!string.IsNullOrEmpty(storageAccountName)) + { + // Managed Identity (production). Resolve the credential once — DefaultAzureCredential + // caches tokens internally and is safe to share across all queue clients. + var credential = new DefaultAzureCredential(); + _builder = (queueName, options) => + { + var uri = new Uri( + $"https://{storageAccountName}.queue.core.windows.net/{queueName}"); + return new QueueClient(uri, credential, options); + }; + } + else if (!string.IsNullOrEmpty(connectionString)) + { + _builder = (queueName, options) => new QueueClient(connectionString, queueName, options); + } + else + { + throw new InvalidOperationException( + "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); + } + } + + /// + /// Builds a for . + /// + /// + /// When true (default) the client encodes message bodies as Base64 — required for + /// round-trip compatibility with the Azure Functions QueueTrigger default and with the + /// producer/worker pairs that consume these queues. Pass false for read-only probes + /// (e.g. GetPropertiesAsync message-count checks) where encoding is irrelevant. + /// + public QueueClient Create(string queueName, bool base64 = true) + { + if (string.IsNullOrWhiteSpace(queueName)) + throw new ArgumentException("Queue name must not be empty.", nameof(queueName)); + + var options = base64 + ? new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 } + : new QueueClientOptions(); + return _builder(queueName, options); + } + + /// + /// Builds the main + poison pair for a worker. When + /// is null the poison queue defaults to + /// + "-poison" (the convention every worker + /// previously hard-coded inline). + /// + public (QueueClient Main, QueueClient Poison) CreatePair( + string mainQueueName, string? poisonQueueName = null, bool base64 = true) + { + var main = Create(mainQueueName, base64); + var poison = Create(poisonQueueName ?? mainQueueName + PoisonQueueSuffix, base64); + return (main, poison); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorker.cs new file mode 100644 index 00000000..a40ac84e --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorker.cs @@ -0,0 +1,236 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Azure.Storage.Queues; +using Azure.Storage.Queues.Models; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Functions.Services.Queueing +{ + /// + /// The common-case queue worker: deserialize each message into a , + /// validate it, hand it to , and delete on success. Adds the optional + /// in-flight heartbeat (visibility extension) used by long-running cascade workers. + /// + /// Subclasses override (required) and optionally + /// / , + /// , , + /// , and the poison-move hooks. + /// + /// + /// The JSON message contract for this queue. + public abstract class QueuePollingWorker : QueuePollingWorkerBase + where TEnvelope : class + { + private readonly TimeSpan? _heartbeatIntervalOverride; + + protected QueuePollingWorker( + QueueClientFactory queueFactory, + string queueName, + ILogger logger, + string? poisonQueueName = null) + : base(queueFactory, queueName, logger, poisonQueueName) + { + } + + protected QueuePollingWorker( + QueueClient mainQueue, + QueueClient poisonQueue, + ILogger logger, + TimeSpan? pollIntervalOverride = null, + TimeSpan? heartbeatIntervalOverride = null) + : base(mainQueue, poisonQueue, logger, pollIntervalOverride) + { + _heartbeatIntervalOverride = heartbeatIntervalOverride; + } + + // ── Heartbeat tunables ───────────────────────────────────────────────────── + + /// + /// When true, a sidecar task extends the in-flight message's visibility every + /// while runs — preventing queue + /// re-delivery from spawning a parallel worker on the same long-running item. + /// + protected virtual bool UseHeartbeat => false; + + /// How often the heartbeat task extends the in-flight message's visibility. + protected virtual TimeSpan HeartbeatInterval => TimeSpan.FromSeconds(60); + + /// How much the heartbeat task adds to visibility on each tick. + protected virtual TimeSpan HeartbeatExtendBy => TimeSpan.FromMinutes(5); + + private TimeSpan ResolvedHeartbeatInterval => _heartbeatIntervalOverride ?? HeartbeatInterval; + + // ── Required / optional overrides ─────────────────────────────────────────── + + /// Process a validated envelope. Throw to trigger visibility-timeout retry. + protected abstract Task HandleAsync(TEnvelope envelope, CancellationToken ct); + + /// + /// Validate a deserialized envelope before dispatch. Return false to drop the message + /// (permanently malformed). Default accepts any non-null envelope. + /// + protected virtual bool TryValidate(TEnvelope envelope) => true; + + /// Optional log context appended to handler-failure messages (e.g. tenant/session ids). + protected virtual string DescribeForLog(TEnvelope envelope) => string.Empty; + + // ── Per-message flow ──────────────────────────────────────────────────────── + + protected sealed override async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) + { + // Platform parity: after MaxDequeueCount attempts, move to poison. DequeueCount > max + // means the prior attempts are exhausted; this receive is not handed to the handler. + if (msg.DequeueCount > MaxDequeueCount) + { + await MoveToPoisonAsync(msg, ct).ConfigureAwait(false); + return; + } + + TEnvelope? envelope; + try + { + envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); + } + catch (JsonException ex) + { + // Malformed JSON is permanent — drop directly so it doesn't burn dequeue attempts. + Logger.LogWarning(ex, "{Worker}: malformed envelope JSON — dropping (msg {Id})", WorkerName, msg.MessageId); + await SafeDeleteAsync(msg, ct).ConfigureAwait(false); + return; + } + + if (envelope is null) + { + Logger.LogWarning("{Worker}: null envelope after deserialization — dropping (msg {Id})", WorkerName, msg.MessageId); + await SafeDeleteAsync(msg, ct).ConfigureAwait(false); + return; + } + + if (!TryValidate(envelope)) + { + Logger.LogWarning("{Worker}: envelope failed validation (missing required fields) — dropping (msg {Id})", WorkerName, msg.MessageId); + await SafeDeleteAsync(msg, ct).ConfigureAwait(false); + return; + } + + if (UseHeartbeat) + await DispatchWithHeartbeatAsync(msg, envelope, ct).ConfigureAwait(false); + else + await DispatchAsync(msg, msg.PopReceipt, envelope, ct).ConfigureAwait(false); + } + + private async Task DispatchAsync(QueueMessage msg, string popReceipt, TEnvelope envelope, CancellationToken ct) + { + try + { + await HandleAsync(envelope, ct).ConfigureAwait(false); + await SafeDeleteAsync(msg.MessageId, popReceipt, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + // Shutdown — leave the message visible for the next worker run. + throw; + } + catch (Exception ex) + { + // Don't delete: visibility-timeout makes the message visible again, DequeueCount + // increments, eventual move-to-poison. Identical retry shape to the platform. + LogHandlerFailure(ex, msg, envelope); + } + } + + private async Task DispatchWithHeartbeatAsync(QueueMessage msg, TEnvelope envelope, CancellationToken ct) + { + // Capture the latest PopReceipt locally — each UpdateMessageAsync returns a fresh one; + // the SafeDelete at the end uses whichever receipt was issued last. + using var heartbeatCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + var state = new HeartbeatState(msg.MessageId, msg.PopReceipt); + var heartbeatTask = Task.Run(() => HeartbeatLoopAsync(state, heartbeatCts.Token), heartbeatCts.Token); + + try + { + await HandleAsync(envelope, ct).ConfigureAwait(false); + + heartbeatCts.Cancel(); + try { await heartbeatTask.ConfigureAwait(false); } catch (OperationCanceledException) { } + + await SafeDeleteAsync(state.MessageId, state.PopReceipt, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + heartbeatCts.Cancel(); + throw; + } + catch (Exception ex) + { + heartbeatCts.Cancel(); + try { await heartbeatTask.ConfigureAwait(false); } catch (OperationCanceledException) { } + + LogHandlerFailure(ex, msg, envelope); + } + } + + private async Task HeartbeatLoopAsync(HeartbeatState state, CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + try { await Task.Delay(ResolvedHeartbeatInterval, ct).ConfigureAwait(false); } + catch (OperationCanceledException) { return; } + + try + { + var updated = await MainQueue.UpdateMessageAsync( + state.MessageId, state.PopReceipt, + visibilityTimeout: HeartbeatExtendBy, + cancellationToken: ct).ConfigureAwait(false); + state.PopReceipt = updated.Value.PopReceipt; + Logger.LogDebug( + "{Worker}: heartbeat extended visibility for message {Id} by {Extend}", + WorkerName, state.MessageId, HeartbeatExtendBy); + } + catch (OperationCanceledException) { return; } + catch (Exception ex) + { + // Heartbeat failure (e.g. PopReceipt expired) is logged and the loop exits. The + // handler keeps running; on its return, SafeDelete may 404 and the message will + // reappear after its visibility expires — handlers guard against parallel pickup + // with their own idempotency (ETag-CAS on progress state). + Logger.LogWarning(ex, + "{Worker}: heartbeat UpdateMessageAsync failed for message {Id} — heartbeat exiting", + WorkerName, state.MessageId); + return; + } + } + } + + private void LogHandlerFailure(Exception ex, QueueMessage msg, TEnvelope envelope) + { + var detail = DescribeForLog(envelope); + if (string.IsNullOrEmpty(detail)) + { + Logger.LogWarning(ex, + "{Worker}: handler failed (msg {Id}, dequeue {N}) — visibility-timeout retry", + WorkerName, msg.MessageId, msg.DequeueCount); + } + else + { + Logger.LogWarning(ex, + "{Worker}: handler failed (msg {Id}, dequeue {N}, {Detail}) — visibility-timeout retry", + WorkerName, msg.MessageId, msg.DequeueCount, detail); + } + } + + private sealed class HeartbeatState + { + public string MessageId { get; } + public string PopReceipt { get; set; } + public HeartbeatState(string messageId, string popReceipt) + { + MessageId = messageId; + PopReceipt = popReceipt; + } + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs new file mode 100644 index 00000000..85779c42 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs @@ -0,0 +1,245 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Azure.Storage.Queues; +using Azure.Storage.Queues.Models; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Services.Queueing +{ + /// + /// Shared poll-loop skeleton for the project's Storage-Queue + /// workers (code-quality audit 2026-05-29, finding D2). Replaces the QueueTrigger binding — + /// which needs a Functions-host-specific <Connection>__queueServiceUri app-setting + /// the rest of the project's storage access does not use — with a pure DI poll loop that + /// reproduces the platform's retry semantics: + /// + /// Receive up to messages with . + /// Handler success → DeleteMessageAsync. + /// Handler failure → no delete; the message reappears after the visibility timeout and + /// DequeueCount increments (identical to platform retry). + /// After failed attempts → move to the -poison queue. + /// Empty receive → idle for . + /// Unhandled poll-loop exception → log + back off for . + /// + /// + /// This non-generic base owns everything that does not depend on the envelope type: queue + /// bootstrap (via ), the poll loop, poison-move, safe-delete, + /// and queue creation. The typical worker derives from + /// (deserialize → validate → handler). A worker + /// with a bespoke per-message lifecycle (e.g. lease + CAS state machine) derives from this + /// base directly and implements . + /// + /// + public abstract class QueuePollingWorkerBase : BackgroundService + { + private readonly TimeSpan? _pollIntervalOverride; + + /// Main work queue. Protected so bespoke overrides can drive it directly. + protected QueueClient MainQueue { get; } + + /// Poison sibling of . + protected QueueClient PoisonQueue { get; } + + protected ILogger Logger { get; } + + /// Production bootstrap: build the main + poison client pair from the shared factory. + protected QueuePollingWorkerBase( + QueueClientFactory queueFactory, + string queueName, + ILogger logger, + string? poisonQueueName = null) + { + if (queueFactory is null) throw new ArgumentNullException(nameof(queueFactory)); + Logger = logger ?? throw new ArgumentNullException(nameof(logger)); + (MainQueue, PoisonQueue) = queueFactory.CreatePair(queueName, poisonQueueName); + } + + /// + /// Test seam: inject pre-built (possibly mocked) queues so the poll loop can run against an + /// in-memory queue without Azurite. lets tests use a + /// short idle interval instead of the production default. + /// + protected QueuePollingWorkerBase( + QueueClient mainQueue, + QueueClient poisonQueue, + ILogger logger, + TimeSpan? pollIntervalOverride = null) + { + MainQueue = mainQueue ?? throw new ArgumentNullException(nameof(mainQueue)); + PoisonQueue = poisonQueue ?? throw new ArgumentNullException(nameof(poisonQueue)); + Logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _pollIntervalOverride = pollIntervalOverride; + } + + // ── Tunables (override per worker) ─────────────────────────────────────── + + /// Max messages received per poll. Storage Queue caps batch-receive at 32. + protected virtual int BatchSize => 32; + + /// Visibility timeout per received message — must cover the slowest handler path. + protected virtual TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(5); + + /// Idle sleep between empty receives (and pause cycles). + protected virtual TimeSpan PollInterval => TimeSpan.FromSeconds(10); + + /// Cool-down after an unhandled poll-loop exception. + protected virtual TimeSpan ErrorBackoff => TimeSpan.FromSeconds(30); + + /// + /// Canonical max-attempts value, matching the QueueTrigger platform default. Exposed as a + /// const so tests can drive a message past the threshold without hard-coding the number. + /// + public const int DefaultMaxDequeueCount = 5; + + /// Match the QueueTrigger platform default: 5 attempts, then move to poison. + protected virtual int MaxDequeueCount => DefaultMaxDequeueCount; + + /// Effective poll interval — the test override wins over the virtual default. + protected TimeSpan ResolvedPollInterval => _pollIntervalOverride ?? PollInterval; + + /// Log-friendly worker name. Defaults to the concrete type name. + protected virtual string WorkerName => GetType().Name; + + // ── Loop ───────────────────────────────────────────────────────────────── + + protected sealed override async Task ExecuteAsync(CancellationToken stoppingToken) + { + // Ensure both queues exist before the first receive — the producer creates the main + // queue on first send too, but having the poison queue ready avoids losing the first + // poison-move attempt. + await TryCreateQueueAsync(MainQueue, "main", stoppingToken).ConfigureAwait(false); + await TryCreateQueueAsync(PoisonQueue, "poison", stoppingToken).ConfigureAwait(false); + + Logger.LogInformation("{Worker}: poll loop started (queue {Name})", WorkerName, MainQueue.Name); + + while (!stoppingToken.IsCancellationRequested) + { + try + { + // Optional entry guard (e.g. kill-switch). When paused the worker does not + // dequeue — messages stay visible without burning dequeue budget. + if (await ShouldPauseAsync(stoppingToken).ConfigureAwait(false)) + { + await Task.Delay(ResolvedPollInterval, stoppingToken).ConfigureAwait(false); + continue; + } + + var batch = await MainQueue + .ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken) + .ConfigureAwait(false); + + if (batch?.Value is null || batch.Value.Length == 0) + { + await Task.Delay(ResolvedPollInterval, stoppingToken).ConfigureAwait(false); + continue; + } + + foreach (var msg in batch.Value) + { + if (stoppingToken.IsCancellationRequested) break; + await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + Logger.LogError(ex, "{Worker}: poll-loop error — backing off {Backoff}", WorkerName, ErrorBackoff); + try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); } + catch (OperationCanceledException) { break; } + } + } + + Logger.LogInformation("{Worker}: poll loop stopped", WorkerName); + } + + /// + /// Optional entry guard evaluated once per poll cycle before receiving. Default never pauses. + /// Override to honor a kill-switch. + /// + protected virtual ValueTask ShouldPauseAsync(CancellationToken ct) => new(false); + + /// Process a single received message. Implemented by or a bespoke worker. + protected abstract Task ProcessOneAsync(QueueMessage msg, CancellationToken ct); + + // ── Shared helpers ───────────────────────────────────────────────────────── + + /// + /// Moves a message that exhausted its dequeue budget to the poison queue and deletes it + /// from the main queue. can veto the move (leaving the + /// message visible for retry); runs after a successful move. + /// + protected async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct) + { + if (!await BeforePoisonMoveAsync(msg, ct).ConfigureAwait(false)) + { + // Hook vetoed the move (e.g. a durable terminal-state transition failed and must + // be retried before we lose the message). Leave it visible for the next dequeue. + return; + } + + try + { + await PoisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false); + await SafeDeleteAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); + Logger.LogWarning( + "{Worker}: moved message {Id} to poison queue after {N} failed attempts", + WorkerName, msg.MessageId, msg.DequeueCount - 1); + } + catch (Exception ex) + { + // If poison-enqueue itself fails, leave the message — next visibility-timeout + // retries the move. Avoids losing data on transient outages. + Logger.LogError(ex, "{Worker}: poison move failed for message {Id} (will retry)", WorkerName, msg.MessageId); + return; + } + + await AfterPoisonMoveAsync(msg, ct).ConfigureAwait(false); + } + + /// + /// Runs before the poison-queue send. Return false to abort the move and leave the + /// message visible (used when a durable terminal-state transition must succeed first). + /// Default proceeds with the move. The message body is available for envelope re-parsing. + /// + protected virtual Task BeforePoisonMoveAsync(QueueMessage msg, CancellationToken ct) => Task.FromResult(true); + + /// Runs after a successful poison move (e.g. emit an OpsEvent). Default no-op. + protected virtual Task AfterPoisonMoveAsync(QueueMessage msg, CancellationToken ct) => Task.CompletedTask; + + protected Task SafeDeleteAsync(QueueMessage msg, CancellationToken ct) + => SafeDeleteAsync(msg.MessageId, msg.PopReceipt, ct); + + protected async Task SafeDeleteAsync(string messageId, string popReceipt, CancellationToken ct) + { + try + { + await MainQueue.DeleteMessageAsync(messageId, popReceipt, ct).ConfigureAwait(false); + } + catch (Exception ex) + { + Logger.LogWarning(ex, + "{Worker}: delete failed for message {Id} (will reappear after visibility-timeout)", + WorkerName, messageId); + } + } + + protected async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct) + { + try + { + await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); + } + catch (Exception ex) + { + Logger.LogWarning(ex, + "{Worker}: CreateIfNotExists failed for {Label} queue — will continue, send/receive will retry", + WorkerName, label); + } + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/AzureQueueVulnerabilityCorrelateProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/AzureQueueVulnerabilityCorrelateProducer.cs index a7d8f87e..14bde1b3 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/AzureQueueVulnerabilityCorrelateProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/AzureQueueVulnerabilityCorrelateProducer.cs @@ -1,11 +1,10 @@ using System; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; using Azure.Storage.Queues; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -30,40 +29,12 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability private int _queueEnsured; // 0 = not yet ensured, 1 = CreateIfNotExistsAsync has run public AzureQueueVulnerabilityCorrelateProducer( - IConfiguration configuration, + QueueClientFactory queueFactory, ILogger logger) { _logger = logger; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - // Match the consumer (BackgroundService) Base64 encoding so messages round-trip. - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var queueUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.VulnerabilityCorrelate}"); - _queueClient = new QueueClient(queueUri, new DefaultAzureCredential(), options); - _logger.LogInformation( - "VulnerabilityCorrelate producer initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _queueClient = new QueueClient( - connectionString, Constants.QueueNames.VulnerabilityCorrelate, options); - _logger.LogInformation("VulnerabilityCorrelate producer initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } + // Base64 encoding matches the consumer (BackgroundService) so messages round-trip. + _queueClient = queueFactory.Create(Constants.QueueNames.VulnerabilityCorrelate); } public async Task EnqueueAsync(VulnerabilityCorrelateEnvelope envelope, CancellationToken cancellationToken = default) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs index b8110c20..ce923fa3 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs @@ -1,254 +1,35 @@ using System; using System.Threading; using System.Threading.Tasks; -using Azure.Identity; -using Azure.Storage.Queues; -using Azure.Storage.Queues.Models; +using AutopilotMonitor.Functions.Services.Queueing; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; -using Newtonsoft.Json; namespace AutopilotMonitor.Functions.Services.Vulnerability { /// /// Background worker for the vulnerability-correlate queue. Replaces the previous /// in-function fire-and-forget Task.Run in EventIngestProcessor that ran - /// vulnerability correlation after a shutdown software_inventory_analysis event — - /// that pattern could be killed mid-flight by Azure Functions scale-in, leaving sessions - /// without a stored report (manual "Analyze Now" rescan was the only recovery). - /// - /// Mirrors AnalyzeOnEnrollmentEndQueueWorker instead of using a Functions - /// QueueTrigger: pure DI driven by the existing - /// AzureStorageAccountName (Managed Identity) or AzureTableStorageConnectionString - /// fallback — no bespoke binding glue. - /// - /// - /// Platform parity to QueueTrigger: - /// - /// Receive batch of up to messages with - /// per message. - /// Handler success → DeleteMessageAsync. - /// Handler exception → no delete; message becomes visible again after the timeout, - /// DequeueCount increments. Identical to platform-retry semantics. - /// After failed attempts the message is moved to - /// vulnerability-correlate-poison and deleted from the main queue. - /// Empty receive → sleep for and retry. - /// - /// + /// vulnerability correlation after a shutdown software_inventory_analysis event — that + /// pattern could be killed mid-flight by Azure Functions scale-in, leaving sessions without a + /// stored report. Uses the shared poll loop with + /// vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison). /// - public sealed class VulnerabilityCorrelateQueueWorker : BackgroundService + public sealed class VulnerabilityCorrelateQueueWorker : QueuePollingWorker { - /// Max messages received per poll. Storage Queue caps batch-receive at 32. - internal const int BatchSize = 32; - - /// - /// Visibility timeout per received message. Correlation involves CVE/KEV lookups - /// against external data and can take several seconds; the inner correlation has its - /// own 60s timeout. 5 minutes leaves plenty of headroom for retries and matches - /// the Analyze + IndexReconcile workers. - /// - internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5); - - /// Idle sleep between empty receives. - internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10); - - /// Cool-down after an unhandled poll-loop exception. - internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30); - - /// Match the QueueTrigger platform default: 5 attempts, then move to poison. - internal const int MaxDequeueCount = 5; - - private const string PoisonQueueSuffix = "-poison"; - - private readonly QueueClient _mainQueue; - private readonly QueueClient _poisonQueue; private readonly VulnerabilityCorrelateHandler _handler; - private readonly ILogger _logger; public VulnerabilityCorrelateQueueWorker( - IConfiguration configuration, + QueueClientFactory queueFactory, VulnerabilityCorrelateHandler handler, ILogger logger) + : base(queueFactory, Constants.QueueNames.VulnerabilityCorrelate, logger) { _handler = handler ?? throw new ArgumentNullException(nameof(handler)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - - var options = new QueueClientOptions - { - MessageEncoding = QueueMessageEncoding.Base64, - }; - - var storageAccountName = configuration["AzureStorageAccountName"]; - var connectionString = configuration["AzureTableStorageConnectionString"]; - - if (!string.IsNullOrEmpty(storageAccountName)) - { - var mainUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.VulnerabilityCorrelate}"); - var poisonUri = new Uri( - $"https://{storageAccountName}.queue.core.windows.net/{Constants.QueueNames.VulnerabilityCorrelate}{PoisonQueueSuffix}"); - var credential = new DefaultAzureCredential(); - _mainQueue = new QueueClient(mainUri, credential, options); - _poisonQueue = new QueueClient(poisonUri, credential, options); - _logger.LogInformation( - "VulnerabilityCorrelateQueueWorker initialized with Managed Identity (account: {Account})", - storageAccountName); - } - else if (!string.IsNullOrEmpty(connectionString)) - { - _mainQueue = new QueueClient( - connectionString, Constants.QueueNames.VulnerabilityCorrelate, options); - _poisonQueue = new QueueClient( - connectionString, Constants.QueueNames.VulnerabilityCorrelate + PoisonQueueSuffix, options); - _logger.LogInformation("VulnerabilityCorrelateQueueWorker initialized with connection string"); - } - else - { - throw new InvalidOperationException( - "Queue Storage not configured. Set either 'AzureStorageAccountName' (for Managed Identity) or 'AzureTableStorageConnectionString'."); - } } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - await TryCreateQueueAsync(_mainQueue, "main", stoppingToken).ConfigureAwait(false); - await TryCreateQueueAsync(_poisonQueue, "poison", stoppingToken).ConfigureAwait(false); - - _logger.LogInformation("VulnerabilityCorrelateQueueWorker: poll loop started (queue {Name})", - _mainQueue.Name); - - while (!stoppingToken.IsCancellationRequested) - { - try - { - var batch = await _mainQueue - .ReceiveMessagesAsync(BatchSize, VisibilityTimeout, stoppingToken) - .ConfigureAwait(false); - - if (batch?.Value is null || batch.Value.Length == 0) - { - await Task.Delay(PollInterval, stoppingToken).ConfigureAwait(false); - continue; - } - - foreach (var msg in batch.Value) - { - if (stoppingToken.IsCancellationRequested) break; - await ProcessOneAsync(msg, stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _logger.LogError(ex, - "VulnerabilityCorrelateQueueWorker: poll-loop error — backing off {Backoff}", - ErrorBackoff); - try { await Task.Delay(ErrorBackoff, stoppingToken).ConfigureAwait(false); } - catch (OperationCanceledException) { break; } - } - } - - _logger.LogInformation("VulnerabilityCorrelateQueueWorker: poll loop stopped"); - } - - private async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) - { - if (msg.DequeueCount > MaxDequeueCount) - { - await MoveToPoisonAsync(msg, ct).ConfigureAwait(false); - return; - } - - VulnerabilityCorrelateEnvelope? envelope; - try - { - envelope = JsonConvert.DeserializeObject(msg.Body.ToString()); - } - catch (JsonException ex) - { - _logger.LogWarning(ex, - "VulnerabilityCorrelateQueueWorker: malformed envelope JSON — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - if (envelope is null) - { - _logger.LogWarning( - "VulnerabilityCorrelateQueueWorker: null envelope after deserialization — dropping (msg {Id})", - msg.MessageId); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - return; - } - - try - { - await _handler.HandleAsync(envelope, ct).ConfigureAwait(false); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - throw; - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "VulnerabilityCorrelateQueueWorker: handler failed (msg {Id}, dequeue {N}) — visibility-timeout retry", - msg.MessageId, msg.DequeueCount); - } - } - - private async Task MoveToPoisonAsync(QueueMessage msg, CancellationToken ct) - { - try - { - await _poisonQueue.SendMessageAsync(msg.Body.ToString(), ct).ConfigureAwait(false); - await SafeDeleteAsync(msg, ct).ConfigureAwait(false); - _logger.LogWarning( - "VulnerabilityCorrelateQueueWorker: moved message {Id} to poison queue after {N} failed attempts", - msg.MessageId, msg.DequeueCount - 1); - } - catch (Exception ex) - { - _logger.LogError(ex, - "VulnerabilityCorrelateQueueWorker: poison move failed for message {Id} (will retry)", - msg.MessageId); - } - } - - private async Task SafeDeleteAsync(QueueMessage msg, CancellationToken ct) - { - try - { - await _mainQueue.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "VulnerabilityCorrelateQueueWorker: delete failed for message {Id} (will reappear after visibility-timeout)", - msg.MessageId); - } - } - - private async Task TryCreateQueueAsync(QueueClient queue, string label, CancellationToken ct) - { - try - { - await queue.CreateIfNotExistsAsync(cancellationToken: ct).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "VulnerabilityCorrelateQueueWorker: CreateIfNotExists failed for {Label} queue — will continue, send/receive will retry", - label); - } - } + protected override Task HandleAsync(VulnerabilityCorrelateEnvelope envelope, CancellationToken ct) + => _handler.HandleAsync(envelope, ct); } }