gsadmin/AutopilotMonitor
1
0
Fork 0
mirror of https://github.com/okieselbach/AutopilotMonitor.git synced 2026-10-24 12:42:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(queue): extract QueueClientFactory + QueuePollingWorker base (audit D2)

Browse Source 
Eliminates the copy-paste queue scaffolding flagged as finding D2: identical
Managed-Identity/connection-string QueueClient bootstrap hand-rolled across 13
sites, plus a poll-loop/poison/safe-delete skeleton duplicated in 5 workers.

- QueueClientFactory: resolves the MI/connstring model once; consumed by all 5
  producers, the offboarding enqueuer, the backup producer and the poison probe.
- QueuePollingWorkerBase: shared bootstrap + poll loop + poison-move + safe-delete
  + queue-create (ExecuteAsync/ProcessOneAsync sealed to prevent drift).
- QueuePollingWorker<TEnvelope>: deserialize -> validate -> handler flow with an
  opt-in in-flight heartbeat.
- 3 vanilla workers (Index/Analyze/Vuln) collapse to ctor + HandleAsync.
- SessionDeletion/Offboarding use BatchSize/UseHeartbeat/ShouldPauseAsync +
  Before/AfterPoisonMoveAsync hooks (kill-switch, CAS state transition, OpsEvent
  enrichment, Failed-state-before-poison all preserved bit-for-bit).
- Backup derives from the non-generic base and overrides ProcessOneAsync (its
  lease + CAS state machine is intentionally bespoke).

Producers get the factory only -- no shared base, their business logic is
genuinely divergent. Test seams unchanged; the static SessionDeletionWorker.
MaxDequeueCount reference moves to QueuePollingWorkerBase.DefaultMaxDequeueCount.

2168/2168 backend tests green, 0 warnings.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-06-05 14:56:00 +02:00
parent 06cfcfd748
commit 467ee07810
19 changed files with 905 additions and 1818 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr4cCodexFixesTests.cs
+2 -1
View File
@@ -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.
src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionWorkerTests.cs
+8 -7
View File
@@ -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<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
@@ -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<string>();
@@ -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,
src/Backend/AutopilotMonitor.Functions/Program.cs
+3
View File
@@ -134,6 +134,9 @@ builder.Services.AddMemoryCache();
builder.Services.AddHttpClient();
builder.Services.AddSingleton<ResiliencePolicies>();
builder.Services.AddSingleton<TableStorageService>();
// Shared QueueClient bootstrap (code-quality audit D2): single Managed-Identity/connection-string
// resolution consumed by every queue producer / worker / probe.
builder.Services.AddSingleton<AutopilotMonitor.Functions.Services.Queueing.QueueClientFactory>();
// 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<AutopilotMonitor.Functions.Services.Deletion.ISessionDeletionInventoryReader>(
src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs
+11 -232
View File
@@ -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
{
/// <summary>
/// Background worker for the <c>analyze-on-enrollment-end</c> 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.
/// <para>
/// Mirrors <c>IndexReconcileQueueWorker</c> instead of using a Functions <c>QueueTrigger</c>
/// because the trigger binding requires a Functions-host-specific connection app-setting
/// (<c>&lt;Connection&gt;__queueServiceUri</c>) that the rest of the project's storage
/// access does not need. Pure DI BackgroundService driven by the existing
/// <c>AzureStorageAccountName</c> (Managed Identity) or <c>AzureTableStorageConnectionString</c>
/// fallback — no bespoke binding glue.
/// </para>
/// <para>
/// <b>Platform parity to QueueTrigger:</b>
/// <list type="bullet">
/// <item>Receive batch of up to <see cref="BatchSize"/> messages with
/// <see cref="VisibilityTimeout"/> visibility-timeout per message.</item>
/// <item>Handler success → <c>DeleteMessageAsync</c>.</item>
/// <item>Handler exception → no delete; message becomes visible again after the timeout,
/// <c>DequeueCount</c> increments. Identical to platform-retry semantics.</item>
/// <item>After <see cref="MaxDequeueCount"/> failed attempts the message is moved to
/// <c>analyze-on-enrollment-end-poison</c> and deleted from the main queue.</item>
/// <item>Empty receive → sleep for <see cref="PollInterval"/> and retry.</item>
/// </list>
/// </para>
/// Background worker for the <c>analyze-on-enrollment-end</c> queue. Replaces the previous
/// in-function fire-and-forget <c>Task.Run</c> 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 <see cref="QueuePollingWorker{TEnvelope}"/> poll loop
/// with vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison).
/// </summary>
public sealed class AnalyzeOnEnrollmentEndQueueWorker : BackgroundService
public sealed class AnalyzeOnEnrollmentEndQueueWorker : QueuePollingWorker<AnalyzeOnEnrollmentEndEnvelope>
{
/// <summary>Max messages received per poll. Storage Queue caps batch-receive at 32.</summary>
internal const int BatchSize = 32;
/// <summary>
/// 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.
/// </summary>
internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5);
/// <summary>Idle sleep between empty receives.</summary>
internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10);
/// <summary>Cool-down after an unhandled poll-loop exception.</summary>
internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30);
/// <summary>Match the QueueTrigger platform default: 5 attempts, then move to poison.</summary>
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<AnalyzeOnEnrollmentEndQueueWorker> _logger;
public AnalyzeOnEnrollmentEndQueueWorker(
IConfiguration configuration,
QueueClientFactory queueFactory,
AnalyzeOnEnrollmentEndHandler handler,
ILogger<AnalyzeOnEnrollmentEndQueueWorker> 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<AnalyzeOnEnrollmentEndEnvelope>(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);
}
}
src/Backend/AutopilotMonitor.Functions/Services/Analyze/AzureQueueAnalyzeOnEnrollmentEndProducer.cs
+4 -33
View File
@@ -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<AzureQueueAnalyzeOnEnrollmentEndProducer> 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)
src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/AzureQueueCriticalTableBackupProducer.cs
+3 -32
View File
@@ -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<AzureQueueCriticalTableBackupProducer> 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)
src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs
+57 -171
View File
@@ -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
{
/// <summary>
/// Background worker for the <c>critical-table-backup-jobs</c> queue. Drains messages,
/// holds the maintenance lease for the duration of the per-job work, and runs
/// Background worker for the <c>critical-table-backup-jobs</c> queue. Drains messages, holds
/// the maintenance lease for the duration of the per-job work, and runs
/// <see cref="ICriticalTableBackupService.RunBackupUnderLeaseAsync"/> with the lease held.
/// <para>
/// <b>State-machine boundary (plan Wave17/18/19/22/23):</b> 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 <see cref="QueuePollingWorkerBase"/> (not <see cref="QueuePollingWorker{TEnvelope}"/>)
/// 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
/// <see cref="ProcessOneAsync"/> and <see cref="BeforePoisonMoveAsync"/>.
/// </para>
/// <para>
/// <b>Queue-config differs from short-lived workers</b> (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 &gt; 50min).
/// <b>State-machine boundary (plan Wave17/18/19/22/23):</b> 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.
/// </para>
/// <para>
/// <b>Queue-config differs from short-lived workers</b> (plan Wave6): <see cref="BatchSize"/>=1,
/// <see cref="VisibilityTimeout"/>=60min. The per-run service budget is 50min so the 60min
/// visibility gives a 10min cushion.
/// </para>
/// </summary>
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<CriticalTableBackupQueueWorker> _logger;
private readonly TimeProvider _clock;
public CriticalTableBackupQueueWorker(
IConfiguration configuration,
QueueClientFactory queueFactory,
ICriticalTableBackupService backupService,
BlobBackupStore blobStore,
BackupJobsRepository jobs,
OpsEventService opsEvents,
ILogger<CriticalTableBackupQueueWorker> 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<bool> 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<CriticalTableBackupEnvelope>(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;
}
/// <summary>OpsEvent write must not bubble — backup itself already succeeded (or failed by other means).</summary>
@@ -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);
}
}
src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs
+3 -24
View File
@@ -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<SessionDeletionProducer> 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);
}
/// <summary>
src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs
+157 -435
View File
@@ -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
{
/// <summary>
/// Background poll-loop for the <c>session-deletion</c> queue (plan §5 PR4). Modeled on
/// <see cref="AutopilotMonitor.Functions.Services.Analyze.AnalyzeOnEnrollmentEndQueueWorker"/>
/// with four explicit deviations:
/// Background poll-loop for the <c>session-deletion</c> queue (plan §5 PR4). Built on the
/// shared <see cref="QueuePollingWorker{TEnvelope}"/> with four explicit deviations:
/// <list type="bullet">
/// <item><see cref="BatchSize"/> = <b>1</b> (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.</item>
/// <item><b>Kill-switch on entry</b>: every receive checks
/// <item><b>Kill-switch on entry</b> via <see cref="ShouldPauseAsync"/>: every receive checks
/// <c>AdminConfiguration.SessionDeletionKillSwitch</c>; when active the worker idles
/// without dequeuing (envelope stays on the queue, no visibility timeout pressure).</item>
/// <item><b>Heartbeat</b>: while <see cref="SessionDeletionHandler"/> runs, a sidecar task
/// calls <c>UpdateMessageAsync</c> every <see cref="_heartbeatInterval"/> to extend the
/// message visibility by <see cref="HeartbeatExtendBy"/>. Prevents queue re-delivery
/// from spawning a parallel worker on the same manifest while a step is in flight.</item>
/// <item><b>Poison OpsEvent</b>: max-dequeue exhaustion records a
/// <c>SessionDeletionPoisoned</c> 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
/// <c>deletion_poisoned</c> audit but tenant admins should only see the lifecycle
/// endpoints (started / completed / restored), so the signal moved to OpsEvents.
/// The Sessions row's <c>DeletionState</c> is NOT auto-cleared — operator action via
/// restore-from-poisoned (§13, PR4b) is required.</item>
/// without dequeuing.</item>
/// <item><b>Heartbeat</b> (<see cref="UseHeartbeat"/>): the base extends message visibility
/// while <see cref="SessionDeletionHandler"/> runs so re-delivery can't spawn a parallel
/// worker on the same manifest.</item>
/// <item><b>Poison hooks</b>: <see cref="BeforePoisonMoveAsync"/> transitions
/// <c>Sessions.DeletionState → Poisoned</c> before the move; <see cref="AfterPoisonMoveAsync"/>
/// records a <c>SessionDeletionPoisoned</c> 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 <c>DeletionState</c> is NOT auto-cleared — operator action via
/// restore-from-poisoned is required.)</item>
/// </list>
/// </summary>
public sealed class SessionDeletionWorker : BackgroundService
public sealed class SessionDeletionWorker : QueuePollingWorker<SessionDeletionEnvelope>
{
/// <summary>
/// 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.
/// </summary>
internal const int BatchSize = 1;
/// <summary>Initial message-visibility window — extended in-flight by the heartbeat task.</summary>
internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5);
/// <summary>How often the heartbeat task extends the in-flight message's visibility.</summary>
internal static readonly TimeSpan DefaultHeartbeatInterval = TimeSpan.FromSeconds(60);
/// <summary>How much the heartbeat task adds to visibility on each tick.</summary>
internal static readonly TimeSpan HeartbeatExtendBy = TimeSpan.FromMinutes(5);
/// <summary>Idle sleep between empty receives or kill-switch-on cycles.</summary>
internal static readonly TimeSpan DefaultPollInterval = TimeSpan.FromSeconds(10);
/// <summary>Instance-level overrides — production uses the defaults above, tests inject shorter values via the internal ctor.</summary>
private readonly TimeSpan _heartbeatInterval;
private readonly TimeSpan _pollInterval;
/// <summary>Cool-down after an unhandled poll-loop exception.</summary>
internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30);
/// <summary>Match the QueueTrigger platform default + sibling workers: 5 attempts → poison.</summary>
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<SessionDeletionWorker> _logger;
public SessionDeletionWorker(
IConfiguration configuration,
QueueClientFactory queueFactory,
SessionDeletionHandler handler,
TableStorageService storage,
AdminConfigurationService adminConfig,
BlobStorageService blob,
OpsEventService opsEvents,
ILogger<SessionDeletionWorker> 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'.");
}
}
/// <summary>
/// Test seam: construct directly with mock <see cref="QueueClient"/> instances so the
/// poll loop can be exercised against an in-memory queue without spinning up Azurite.
/// Mirrors <see cref="SessionDeletionProducer"/>'s internal test ctor.
/// Test seam: construct directly with mock <see cref="QueueClient"/> instances so the poll
/// loop can be exercised against an in-memory queue without Azurite. Mirrors
/// <see cref="SessionDeletionProducer"/>'s internal test ctor.
/// </summary>
internal SessionDeletionWorker(
QueueClient mainQueue,
@@ -150,288 +74,147 @@ namespace AutopilotMonitor.Functions.Services.Deletion
ILogger<SessionDeletionWorker> 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)
/// <summary>
/// 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.
/// </summary>
protected override int BatchSize => 1;
protected override bool UseHeartbeat => true;
protected override async ValueTask<bool> 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<bool> BeforePoisonMoveAsync(QueueMessage msg, CancellationToken ct)
{
if (msg.DequeueCount > MaxDequeueCount)
{
await MoveToPoisonAsync(msg, ct).ConfigureAwait(false);
return;
}
SessionDeletionEnvelope? envelope;
try
{
envelope = JsonConvert.DeserializeObject<SessionDeletionEnvelope>(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<SessionDeletionEnvelope>(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<SessionDeletionEnvelope>(msg.Body.ToString()); }
catch (JsonException) { return null; }
}
/// <summary>
/// Back-compat fallback used only when <see cref="AutopilotMonitor.Shared.Models.Deletion.DeletionProgress.LastObservedResidualCount"/>
/// Back-compat fallback used only when <see cref="DeletionProgress.LastObservedResidualCount"/>
/// 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.
/// </summary>
internal static int? ExtractResidualSampleArrayLength(string? residualSampleJson)
{
@@ -454,21 +237,9 @@ namespace AutopilotMonitor.Functions.Services.Deletion
/// <summary>
/// Trims the progress-blob's residual sample for safe embedding in the
/// <c>SessionDeletionPoisoned</c> OpsEvent's <c>Details</c> column. Three-layer defense
/// against the 4096-char Azure Table truncation discovered by Codex review:
/// <list type="number">
/// <item>Take at most <see cref="AutopilotMonitor.Shared.Models.Deletion.DeletionProgressConstants.OpsEventResidualSamplePreviewSize"/>
/// entries (drop tail, the rest stays in the progress blob for the admin UI).</item>
/// <item>Trim each <c>table</c> / <c>pk</c> / <c>rk</c> to
/// <see cref="AutopilotMonitor.Shared.Models.Deletion.DeletionProgressConstants.OpsEventResidualKeyMaxChars"/>
/// with a trailing <c>…</c> marker — a single composite key can be 200+ chars and
/// would otherwise consume most of the budget on its own.</item>
/// <item>If the serialized JSON still exceeds
/// <see cref="AutopilotMonitor.Shared.Models.Deletion.DeletionProgressConstants.OpsEventResidualPreviewBudgetChars"/>,
/// drop trailing entries until it fits. Worst case (all entries too long) returns
/// <c>"[]"</c>; operators still get failureType + observedResidualCount and can
/// fetch the full sample from the progress blob.</item>
/// </list>
/// 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.
/// </summary>
internal static string? ShrinkResidualSampleForOpsEvent(string? fullSampleJson)
{
@@ -483,7 +254,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion
return null;
}
entries = new List<ResidualPreviewEntry>();
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);
}
}
/// <summary>
/// PR4b E1 + PR4c F5: CAS Sessions.DeletionState to Poisoned with manifestId pre-check.
/// Tries <c>Running → Poisoned</c> first (the common case — Worker successfully CAS'd
/// Queued→Running on first pickup and the handler then failed N times). Falls back to
/// <c>Queued → Poisoned</c> 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.
/// <para>
/// <b>PR4c F5</b>: before issuing any CAS, read the Sessions row's
/// <c>PendingDeletionManifestId</c> and compare to the envelope's <c>manifestId</c>. 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.
/// </para>
/// Tries <c>Running → Poisoned</c> first (the common case), falls back to
/// <c>Queued → Poisoned</c> (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 <c>PendingDeletionManifestId</c> and
/// compare to the envelope's <c>manifestId</c>; a stale envelope must not flip a fresh
/// active cascade.
/// </summary>
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);
}
}
}
src/Backend/AutopilotMonitor.Functions/Services/Indexing/AzureQueueIndexReconcileProducer.cs
+5 -34
View File
@@ -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<AzureQueueIndexReconcileProducer> 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<int> EnqueueBatchAsync(
src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs
+10 -249
View File
@@ -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
{
/// <summary>
/// Background worker for the <c>telemetry-index-reconcile</c> queue (Plan §2.8, §M5.d.3).
/// <para>
/// Replaces the earlier <c>IndexReconcileFunction</c> QueueTrigger because the trigger
/// binding required a Functions-host-specific connection app-setting
/// (<c>&lt;Connection&gt;__queueServiceUri</c>) that the rest of the project's storage
/// access does not need. This worker matches the
/// <see cref="AzureQueueIndexReconcileProducer"/>, <see cref="TableStorageService"/> and
/// <see cref="BlobStorageService"/> pattern: pure DI service driven by the existing
/// <c>AzureStorageAccountName</c> (Managed Identity) or <c>AzureTableStorageConnectionString</c>
/// fallback — no bespoke binding glue.
/// </para>
/// <para>
/// <b>Plattform-Parität zum QueueTrigger:</b>
/// <list type="bullet">
/// <item>Receive batch of up to <see cref="BatchSize"/> messages with
/// <see cref="VisibilityTimeout"/> visibility-timeout per message.</item>
/// <item>Handler success → <c>DeleteMessageAsync</c>.</item>
/// <item>Handler exception → no delete; message becomes visible again after the
/// timeout, <c>DequeueCount</c> increments. Identical to platform-retry semantics.</item>
/// <item>After <see cref="MaxDequeueCount"/> failed attempts the message is moved to
/// <c>telemetry-index-reconcile-poison</c> and deleted from the main queue. Same
/// behaviour as the QueueTrigger platform.</item>
/// <item>Empty receive → sleep for <see cref="PollInterval"/> and retry.</item>
/// <item>Unrecoverable poll-loop exception (e.g. transient Storage outage) → log + back
/// off for <see cref="ErrorBackoff"/> before resuming.</item>
/// </list>
/// </para>
/// Replaces the earlier <c>IndexReconcileFunction</c> QueueTrigger (whose binding required a
/// Functions-host-specific connection app-setting the project's storage access does not need)
/// with the shared <see cref="QueuePollingWorker{TEnvelope}"/> poll loop. Vanilla defaults:
/// batch 32, 5-min visibility, 5 attempts → poison.
/// </summary>
public sealed class IndexReconcileQueueWorker : BackgroundService
public sealed class IndexReconcileQueueWorker : QueuePollingWorker<IndexReconcileEnvelope>
{
/// <summary>Max messages received per poll. Storage Queue caps batch-receive at 32.</summary>
internal const int BatchSize = 32;
/// <summary>
/// Visibility timeout per received message — long enough to cover the slowest
/// expected <see cref="IndexReconcileHandler.HandleAsync"/> path (5 index-table writes).
/// </summary>
internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5);
/// <summary>Idle sleep between empty receives.</summary>
internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10);
/// <summary>Cool-down after an unhandled poll-loop exception.</summary>
internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30);
/// <summary>
/// Match the QueueTrigger platform default: 5 attempts, then move to poison.
/// (Storage Queue's <c>DequeueCount</c> increments on each visibility-timeout reappear.)
/// </summary>
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<IndexReconcileQueueWorker> _logger;
public IndexReconcileQueueWorker(
IConfiguration configuration,
QueueClientFactory queueFactory,
IndexReconcileHandler handler,
ILogger<IndexReconcileQueueWorker> 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<IndexReconcileEnvelope>(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);
}
}
src/Backend/AutopilotMonitor.Functions/Services/Monitoring/AzurePoisonQueueProbe.cs
+12 -35
View File
@@ -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
{
/// <summary>
/// Azure Storage Queue implementation of <see cref="IPoisonQueueProbe"/>.
/// Mirrors the connection-resolution pattern used by the existing queue workers
/// (AnalyzeOnEnrollmentEndQueueWorker, IndexReconcileQueueWorker,
/// VulnerabilityCorrelateQueueWorker): prefer Managed Identity via
/// <c>AzureStorageAccountName</c>, fall back to <c>AzureTableStorageConnectionString</c>.
/// 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 <see cref="QueueClientFactory"/> (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.
/// <para>
/// Probes only read <c>GetPropertiesAsync</c> message counts, so they request the factory's
/// non-Base64 client (encoding is irrelevant for a metadata read).
/// </para>
/// </summary>
public sealed class AzurePoisonQueueProbe : IPoisonQueueProbe
{
private readonly ConcurrentDictionary<string, QueueClient> _clients = new();
private readonly Func<string, QueueClient> _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<long> 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
{
src/Backend/AutopilotMonitor.Functions/Services/Offboarding/AzureQueueTenantOffboardingEnqueuer.cs
+3 -32
View File
@@ -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<AzureQueueTenantOffboardingEnqueuer> 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);
}
/// <summary>Test seam: bind directly to a (possibly mocked) <see cref="QueueClient"/>.</summary>
src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs
+41 -271
View File
@@ -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
{
/// <summary>
/// Background poll-loop for the <c>tenant-offboarding</c> queue (Plan §7.3). Modeled on
/// <see cref="Deletion.SessionDeletionWorker"/>:
/// Background poll-loop for the <c>tenant-offboarding</c> queue (Plan §7.3). Built on the
/// shared <see cref="QueuePollingWorker{TEnvelope}"/>:
/// <list type="bullet">
/// <item><see cref="BatchSize"/> = <b>1</b> — one tenant cascade per receive bounds memory
/// (per-tenant enumerate + per-session enqueue is long-running) and limits the
/// blast radius of a poison.</item>
/// <item>Heartbeat extends visibility every <see cref="DefaultHeartbeatInterval"/> while
/// the handler runs so concurrent re-delivery can't spawn a parallel offboarding
/// for the same tenant.</item>
/// <item>Max-dequeue <see cref="MaxDequeueCount"/> → poison queue + the handler's own
/// Failed-state transition (the handler emits <c>TenantOffboardingFailed</c> when
/// it fails closed; the worker emits one too on dequeue exhaustion so an envelope
/// that never reaches the handler still surfaces).</item>
/// (per-tenant enumerate + per-session enqueue is long-running) and limits the blast
/// radius of a poison.</item>
/// <item>Heartbeat (<see cref="UseHeartbeat"/>) extends visibility while the handler runs so
/// concurrent re-delivery can't spawn a parallel offboarding for the same tenant.</item>
/// <item>Max-dequeue → poison queue, but <see cref="BeforePoisonMoveAsync"/> 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.</item>
/// </list>
/// </summary>
public sealed class TenantOffboardingWorker : BackgroundService
public sealed class TenantOffboardingWorker : QueuePollingWorker<TenantOffboardingEnvelope>
{
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<TenantOffboardingWorker> _logger;
private readonly TimeSpan _heartbeatInterval;
private readonly TimeSpan _pollInterval;
public TenantOffboardingWorker(
IConfiguration configuration,
QueueClientFactory queueFactory,
TenantOffboardingHandler handler,
OpsEventService opsEvents,
ILogger<TenantOffboardingWorker> 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'.");
}
}
/// <summary>Test seam — inject mocked queues + shorter intervals for unit tests.</summary>
@@ -102,165 +51,39 @@ namespace AutopilotMonitor.Functions.Services.Offboarding
ILogger<TenantOffboardingWorker> 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<TenantOffboardingEnvelope>(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<bool> BeforePoisonMoveAsync(QueueMessage msg, CancellationToken ct)
{
TenantOffboardingEnvelope? envelope = null;
try { envelope = JsonConvert.DeserializeObject<TenantOffboardingEnvelope>(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;
}
}
}
src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueClientFactory.cs
+91
View File
@@ -0,0 +1,91 @@
using System;
using Azure.Identity;
using Azure.Storage.Queues;
using Microsoft.Extensions.Configuration;
namespace AutopilotMonitor.Functions.Services.Queueing
{
/// <summary>
/// Single source of truth for building <see cref="QueueClient"/> instances against the
/// project's storage account. Every queue producer / worker / probe used to hand-roll the
/// same <c>AzureStorageAccountName</c> (Managed Identity) vs
/// <c>AzureTableStorageConnectionString</c> (fallback) branch — ~25 duplicated lines per
/// site across 13 sites (code-quality audit 2026-05-29, finding D2). This factory resolves
/// the credential model <b>once</b> in its constructor and hands out clients on demand.
/// <para>
/// Registered as a singleton; the resolved credential / connection string are captured in
/// closures so each <see cref="Create(string, bool)"/> call is allocation-cheap and never
/// re-reads configuration.
/// </para>
/// </summary>
public sealed class QueueClientFactory
{
private const string PoisonQueueSuffix = "-poison";
private readonly Func<string, QueueClientOptions, QueueClient> _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'.");
}
}
/// <summary>
/// Builds a <see cref="QueueClient"/> for <paramref name="queueName"/>.
/// </summary>
/// <param name="base64">
/// When <c>true</c> (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 <c>false</c> for read-only probes
/// (e.g. <c>GetPropertiesAsync</c> message-count checks) where encoding is irrelevant.
/// </param>
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);
}
/// <summary>
/// Builds the main + poison <see cref="QueueClient"/> pair for a worker. When
/// <paramref name="poisonQueueName"/> is null the poison queue defaults to
/// <paramref name="mainQueueName"/> + <c>"-poison"</c> (the convention every worker
/// previously hard-coded inline).
/// </summary>
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);
}
}
}
src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorker.cs
+236
View File
@@ -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
{
/// <summary>
/// The common-case queue worker: deserialize each message into a <typeparamref name="TEnvelope"/>,
/// validate it, hand it to <see cref="HandleAsync"/>, and delete on success. Adds the optional
/// in-flight <b>heartbeat</b> (visibility extension) used by long-running cascade workers.
/// <para>
/// Subclasses override <see cref="HandleAsync"/> (required) and optionally
/// <see cref="QueuePollingWorkerBase.BatchSize"/> / <see cref="QueuePollingWorkerBase.VisibilityTimeout"/>,
/// <see cref="TryValidate"/>, <see cref="QueuePollingWorkerBase.ShouldPauseAsync"/>,
/// <see cref="UseHeartbeat"/>, and the poison-move hooks.
/// </para>
/// </summary>
/// <typeparam name="TEnvelope">The JSON message contract for this queue.</typeparam>
public abstract class QueuePollingWorker<TEnvelope> : 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 ─────────────────────────────────────────────────────
/// <summary>
/// When true, a sidecar task extends the in-flight message's visibility every
/// <see cref="HeartbeatInterval"/> while <see cref="HandleAsync"/> runs — preventing queue
/// re-delivery from spawning a parallel worker on the same long-running item.
/// </summary>
protected virtual bool UseHeartbeat => false;
/// <summary>How often the heartbeat task extends the in-flight message's visibility.</summary>
protected virtual TimeSpan HeartbeatInterval => TimeSpan.FromSeconds(60);
/// <summary>How much the heartbeat task adds to visibility on each tick.</summary>
protected virtual TimeSpan HeartbeatExtendBy => TimeSpan.FromMinutes(5);
private TimeSpan ResolvedHeartbeatInterval => _heartbeatIntervalOverride ?? HeartbeatInterval;
// ── Required / optional overrides ───────────────────────────────────────────
/// <summary>Process a validated envelope. Throw to trigger visibility-timeout retry.</summary>
protected abstract Task HandleAsync(TEnvelope envelope, CancellationToken ct);
/// <summary>
/// Validate a deserialized envelope before dispatch. Return <c>false</c> to drop the message
/// (permanently malformed). Default accepts any non-null envelope.
/// </summary>
protected virtual bool TryValidate(TEnvelope envelope) => true;
/// <summary>Optional log context appended to handler-failure messages (e.g. tenant/session ids).</summary>
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<TEnvelope>(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;
}
}
}
}
src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs
+245
View File
@@ -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
{
/// <summary>
/// Shared poll-loop skeleton for the project's Storage-Queue <see cref="BackgroundService"/>
/// workers (code-quality audit 2026-05-29, finding D2). Replaces the QueueTrigger binding —
/// which needs a Functions-host-specific <c>&lt;Connection&gt;__queueServiceUri</c> 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:
/// <list type="bullet">
/// <item>Receive up to <see cref="BatchSize"/> messages with <see cref="VisibilityTimeout"/>.</item>
/// <item>Handler success → <c>DeleteMessageAsync</c>.</item>
/// <item>Handler failure → no delete; the message reappears after the visibility timeout and
/// <c>DequeueCount</c> increments (identical to platform retry).</item>
/// <item>After <see cref="MaxDequeueCount"/> failed attempts → move to the <c>-poison</c> queue.</item>
/// <item>Empty receive → idle for <see cref="PollInterval"/>.</item>
/// <item>Unhandled poll-loop exception → log + back off for <see cref="ErrorBackoff"/>.</item>
/// </list>
/// <para>
/// This non-generic base owns everything that does not depend on the envelope type: queue
/// bootstrap (via <see cref="QueueClientFactory"/>), the poll loop, poison-move, safe-delete,
/// and queue creation. The typical worker derives from
/// <see cref="QueuePollingWorker{TEnvelope}"/> (deserialize → validate → handler). A worker
/// with a bespoke per-message lifecycle (e.g. lease + CAS state machine) derives from this
/// base directly and implements <see cref="ProcessOneAsync"/>.
/// </para>
/// </summary>
public abstract class QueuePollingWorkerBase : BackgroundService
{
private readonly TimeSpan? _pollIntervalOverride;
/// <summary>Main work queue. Protected so bespoke <see cref="ProcessOneAsync"/> overrides can drive it directly.</summary>
protected QueueClient MainQueue { get; }
/// <summary>Poison sibling of <see cref="MainQueue"/>.</summary>
protected QueueClient PoisonQueue { get; }
protected ILogger Logger { get; }
/// <summary>Production bootstrap: build the main + poison client pair from the shared factory.</summary>
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);
}
/// <summary>
/// Test seam: inject pre-built (possibly mocked) queues so the poll loop can run against an
/// in-memory queue without Azurite. <paramref name="pollIntervalOverride"/> lets tests use a
/// short idle interval instead of the production default.
/// </summary>
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) ───────────────────────────────────────
/// <summary>Max messages received per poll. Storage Queue caps batch-receive at 32.</summary>
protected virtual int BatchSize => 32;
/// <summary>Visibility timeout per received message — must cover the slowest handler path.</summary>
protected virtual TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(5);
/// <summary>Idle sleep between empty receives (and pause cycles).</summary>
protected virtual TimeSpan PollInterval => TimeSpan.FromSeconds(10);
/// <summary>Cool-down after an unhandled poll-loop exception.</summary>
protected virtual TimeSpan ErrorBackoff => TimeSpan.FromSeconds(30);
/// <summary>
/// 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.
/// </summary>
public const int DefaultMaxDequeueCount = 5;
/// <summary>Match the QueueTrigger platform default: 5 attempts, then move to poison.</summary>
protected virtual int MaxDequeueCount => DefaultMaxDequeueCount;
/// <summary>Effective poll interval — the test override wins over the virtual default.</summary>
protected TimeSpan ResolvedPollInterval => _pollIntervalOverride ?? PollInterval;
/// <summary>Log-friendly worker name. Defaults to the concrete type name.</summary>
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);
}
/// <summary>
/// Optional entry guard evaluated once per poll cycle before receiving. Default never pauses.
/// Override to honor a kill-switch.
/// </summary>
protected virtual ValueTask<bool> ShouldPauseAsync(CancellationToken ct) => new(false);
/// <summary>Process a single received message. Implemented by <see cref="QueuePollingWorker{TEnvelope}"/> or a bespoke worker.</summary>
protected abstract Task ProcessOneAsync(QueueMessage msg, CancellationToken ct);
// ── Shared helpers ─────────────────────────────────────────────────────────
/// <summary>
/// Moves a message that exhausted its dequeue budget to the poison queue and deletes it
/// from the main queue. <see cref="BeforePoisonMoveAsync"/> can veto the move (leaving the
/// message visible for retry); <see cref="AfterPoisonMoveAsync"/> runs after a successful move.
/// </summary>
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);
}
/// <summary>
/// Runs before the poison-queue send. Return <c>false</c> 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.
/// </summary>
protected virtual Task<bool> BeforePoisonMoveAsync(QueueMessage msg, CancellationToken ct) => Task.FromResult(true);
/// <summary>Runs after a successful poison move (e.g. emit an OpsEvent). Default no-op.</summary>
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);
}
}
}
}
src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/AzureQueueVulnerabilityCorrelateProducer.cs
+4 -33
View File
@@ -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<AzureQueueVulnerabilityCorrelateProducer> 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)
src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs
+10 -229
View File
@@ -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
{
/// <summary>
/// Background worker for the <c>vulnerability-correlate</c> queue. Replaces the previous
/// in-function fire-and-forget <c>Task.Run</c> in <c>EventIngestProcessor</c> that ran
/// vulnerability correlation after a shutdown <c>software_inventory_analysis</c> 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).
/// <para>
/// Mirrors <c>AnalyzeOnEnrollmentEndQueueWorker</c> instead of using a Functions
/// <c>QueueTrigger</c>: pure DI <see cref="BackgroundService"/> driven by the existing
/// <c>AzureStorageAccountName</c> (Managed Identity) or <c>AzureTableStorageConnectionString</c>
/// fallback — no bespoke binding glue.
/// </para>
/// <para>
/// <b>Platform parity to QueueTrigger:</b>
/// <list type="bullet">
/// <item>Receive batch of up to <see cref="BatchSize"/> messages with
/// <see cref="VisibilityTimeout"/> per message.</item>
/// <item>Handler success → <c>DeleteMessageAsync</c>.</item>
/// <item>Handler exception → no delete; message becomes visible again after the timeout,
/// <c>DequeueCount</c> increments. Identical to platform-retry semantics.</item>
/// <item>After <see cref="MaxDequeueCount"/> failed attempts the message is moved to
/// <c>vulnerability-correlate-poison</c> and deleted from the main queue.</item>
/// <item>Empty receive → sleep for <see cref="PollInterval"/> and retry.</item>
/// </list>
/// </para>
/// vulnerability correlation after a shutdown <c>software_inventory_analysis</c> event — that
/// pattern could be killed mid-flight by Azure Functions scale-in, leaving sessions without a
/// stored report. Uses the shared <see cref="QueuePollingWorker{TEnvelope}"/> poll loop with
/// vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison).
/// </summary>
public sealed class VulnerabilityCorrelateQueueWorker : BackgroundService
public sealed class VulnerabilityCorrelateQueueWorker : QueuePollingWorker<VulnerabilityCorrelateEnvelope>
{
/// <summary>Max messages received per poll. Storage Queue caps batch-receive at 32.</summary>
internal const int BatchSize = 32;
/// <summary>
/// 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.
/// </summary>
internal static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5);
/// <summary>Idle sleep between empty receives.</summary>
internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(10);
/// <summary>Cool-down after an unhandled poll-loop exception.</summary>
internal static readonly TimeSpan ErrorBackoff = TimeSpan.FromSeconds(30);
/// <summary>Match the QueueTrigger platform default: 5 attempts, then move to poison.</summary>
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<VulnerabilityCorrelateQueueWorker> _logger;
public VulnerabilityCorrelateQueueWorker(
IConfiguration configuration,
QueueClientFactory queueFactory,
VulnerabilityCorrelateHandler handler,
ILogger<VulnerabilityCorrelateQueueWorker> 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<VulnerabilityCorrelateEnvelope>(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);
}
}