diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr3CodexFixesRound2Tests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr3CodexFixesRound2Tests.cs new file mode 100644 index 00000000..b8c8ea8d --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionPr3CodexFixesRound2Tests.cs @@ -0,0 +1,229 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using Azure.Storage.Queues; +using Azure.Storage.Queues.Models; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Functions.Services.Deletion; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using AutopilotMonitor.Shared.Models.Deletion; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Newtonsoft.Json; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Round-2 codex follow-ups on the PR3 follow-up commit. Pin behaviour for +/// (F2) the residual TOCTOU race that the previous fix didn't fully close, and +/// (F4) the stranded-Preparing-with-progress-blob recovery path. +/// +public class SessionDeletionPr3CodexFixesRound2Tests +{ + private const string TenantId = "11111111-1111-1111-1111-111111111111"; + private const string SessionId = "22222222-2222-2222-2222-222222222222"; + + // ============================================================ F4 round 2: Preparing-resume ==== + + [Fact] + public async Task EnqueueAsync_resumes_stranded_Preparing_when_snapshot_blob_exists() + { + // Codex round-2 F4: producer crashed after upload(snapshot)+upload(progress)+audit but + // before CAS Preparing→Queued. Row is in Preparing, snapshot exists, no queue message. + // §10 GC won't clean this up (it only handles Preparing-WITHOUT-progress-blob). The + // producer must finish the resume itself. + const string strandedManifestId = "STRANDED-PREPARING-WITH-BLOB"; + var harness = new ProducerHarness(); + harness.SetWrongState(SessionDeletionState.Preparing, strandedManifestId); + harness.SetSnapshotExists(true); + harness.SetCas2Updated(SessionDeletionState.Queued); + + var result = await harness.Sut.EnqueueAsync( + TenantId, SessionId, "admin_delete", + new DeletionActor { Type = "admin", Actor = "alice@example.com" }); + + Assert.Equal(SessionDeletionEnqueueOutcome.Enqueued, result.Outcome); + Assert.Equal(strandedManifestId, result.ManifestId); + Assert.Equal(SessionDeletionState.Preparing, result.ExistingState); + Assert.Equal("resume", result.Reason); + + // Producer ran the missing CAS Preparing→Queued. + harness.Storage.Verify(s => s.CasSetSessionDeletionStateAsync( + TenantId, SessionId, + SessionDeletionState.Preparing, SessionDeletionState.Queued, + It.IsAny(), It.IsAny()), + Times.Once); + // And re-sent the queue message with the EXISTING manifestId. + Assert.Single(harness.QueueMessages); + var envelope = JsonConvert.DeserializeObject(harness.QueueMessages[0]); + Assert.Equal(strandedManifestId, envelope!.ManifestId); + Assert.Contains("resume", envelope.Reason); + } + + [Fact] + public async Task EnqueueAsync_does_NOT_resume_Preparing_when_snapshot_blob_is_missing() + { + // No snapshot blob → an earlier producer crash (during build/upload). PR6 GC is the + // safe path here: it'll roll Preparing → None after 1h. Re-trying the missing CAS now + // could race a still-in-flight producer. + var harness = new ProducerHarness(); + harness.SetWrongState(SessionDeletionState.Preparing, "PARTIAL-PREPARING"); + harness.SetSnapshotExists(false); + + var result = await harness.Sut.EnqueueAsync( + TenantId, SessionId, "admin_delete", + new DeletionActor { Type = "admin", Actor = "alice@example.com" }); + + Assert.Equal(SessionDeletionEnqueueOutcome.AlreadyInFlight, result.Outcome); + Assert.Empty(harness.QueueMessages); + // Did NOT attempt the CAS Preparing→Queued. + harness.Storage.Verify(s => s.CasSetSessionDeletionStateAsync( + It.IsAny(), It.IsAny(), + SessionDeletionState.Preparing, SessionDeletionState.Queued, + It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task EnqueueAsync_falls_back_to_AlreadyInFlight_when_blob_existence_probe_throws() + { + // If the blob-exists probe throws (Storage outage, throttling), bail out cleanly + // rather than risk a half-completed recovery. Operator gets AlreadyInFlight; PR6 GC + // will eventually clean up. + var harness = new ProducerHarness(); + harness.SetWrongState(SessionDeletionState.Preparing, "PROBE-FAIL-MANIFEST"); + harness.Blob + .Setup(b => b.DeletionSnapshotExistsAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RequestFailedException(503, "Service Unavailable")); + + var result = await harness.Sut.EnqueueAsync( + TenantId, SessionId, "admin_delete", + new DeletionActor { Type = "admin", Actor = "alice@example.com" }); + + Assert.Equal(SessionDeletionEnqueueOutcome.AlreadyInFlight, result.Outcome); + Assert.Empty(harness.QueueMessages); + } + + [Fact] + public async Task EnqueueAsync_returns_AlreadyInFlight_when_resume_CAS_loses_to_concurrent_writer() + { + // Snapshot exists, but the CAS Preparing→Queued races with another producer that won + // (state is now something other than Preparing). Producer reports AlreadyInFlight + // cleanly without sending a duplicate queue message. + var harness = new ProducerHarness(); + harness.SetWrongState(SessionDeletionState.Preparing, "RACE-PREPARING"); + harness.SetSnapshotExists(true); + harness.SetCas2WrongState(SessionDeletionState.Queued, "OTHER-PRODUCER-WON"); + + var result = await harness.Sut.EnqueueAsync( + TenantId, SessionId, "admin_delete", + new DeletionActor { Type = "admin", Actor = "alice@example.com" }); + + Assert.Equal(SessionDeletionEnqueueOutcome.AlreadyInFlight, result.Outcome); + Assert.Empty(harness.QueueMessages); + } + + // ============================================================ Harness ==== + + private sealed class ProducerHarness + { + public Mock Storage { get; } + public Mock Builder { get; } + public Mock Blob { get; } + public Mock AdminConfig { get; } + public Mock Maintenance { get; } + public Mock Queue { get; } + public List QueueMessages { get; } = new List(); + public SessionDeletionProducer Sut { get; } + + public ProducerHarness() + { + Storage = new Mock( + Mock.Of(), NullLogger.Instance); + + Builder = new Mock( + Mock.Of(), NullLogger.Instance); + + Blob = new Mock( + new Azure.Storage.Blobs.BlobServiceClient("UseDevelopmentStorage=true"), + NullLogger.Instance, false); + + AdminConfig = new Mock( + Mock.Of(), + NullLogger.Instance, + new Microsoft.Extensions.Caching.Memory.MemoryCache(new Microsoft.Extensions.Caching.Memory.MemoryCacheOptions())); + AdminConfig.Setup(a => a.GetConfigurationAsync()) + .ReturnsAsync(new AdminConfiguration { SessionDeletionKillSwitch = false }); + + Maintenance = new Mock(); + + Queue = new Mock(); + Queue.Setup(q => q.SendMessageAsync(It.IsAny(), It.IsAny())) + .Returns((body, _) => + { + QueueMessages.Add(body); + var receipt = QueuesModelFactory.SendReceipt("msg", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(7), "pop", DateTimeOffset.UtcNow); + return Task.FromResult(Response.FromValue(receipt, new Mock().Object)); + }); + Queue.Setup(q => q.CreateIfNotExistsAsync(It.IsAny>(), It.IsAny())) + .ReturnsAsync((Response?)null); + + Sut = new SessionDeletionProducer( + Storage.Object, Builder.Object, Blob.Object, + AdminConfig.Object, Maintenance.Object, Queue.Object, + NullLogger.Instance); + } + + public void SetWrongState(string currentState, string? currentManifestId) + { + Storage.Setup(s => s.CasSetSessionDeletionStateAsync( + It.IsAny(), It.IsAny(), + SessionDeletionState.None, SessionDeletionState.Preparing, + It.IsAny(), It.IsAny())) + .ReturnsAsync(new TableStorageService.SessionDeletionStateCasResult + { + Outcome = TableStorageService.SessionDeletionStateCasOutcome.WrongState, + CurrentState = currentState, + CurrentManifestId = currentManifestId, + }); + } + + public void SetSnapshotExists(bool exists) + { + Blob.Setup(b => b.DeletionSnapshotExistsAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(exists); + } + + public void SetCas2Updated(string newState) + { + Storage.Setup(s => s.CasSetSessionDeletionStateAsync( + It.IsAny(), It.IsAny(), + SessionDeletionState.Preparing, SessionDeletionState.Queued, + It.IsAny(), It.IsAny())) + .ReturnsAsync(new TableStorageService.SessionDeletionStateCasResult + { + Outcome = TableStorageService.SessionDeletionStateCasOutcome.Updated, + CurrentState = newState, + }); + } + + public void SetCas2WrongState(string fromState, string actualCurrentState) + { + Storage.Setup(s => s.CasSetSessionDeletionStateAsync( + It.IsAny(), It.IsAny(), + SessionDeletionState.Preparing, SessionDeletionState.Queued, + It.IsAny(), It.IsAny())) + .ReturnsAsync(new TableStorageService.SessionDeletionStateCasResult + { + Outcome = TableStorageService.SessionDeletionStateCasOutcome.WrongState, + CurrentState = actualCurrentState, + }); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs b/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs index f4c17919..f2792115 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/BlobStorageService.cs @@ -327,6 +327,36 @@ namespace AutopilotMonitor.Functions.Services return response.Value.ETag; } + /// + /// Lightweight existence check for a cascade-deletion snapshot blob. Used by the + /// producer's resume-from-Preparing path to distinguish "the prior producer crashed + /// before uploading the snapshot" (skip resume, let GC clean up) from "snapshot is + /// there, only the CAS Preparing→Queued failed" (safe to resume). + /// + public virtual async Task DeletionSnapshotExistsAsync( + string tenantId, string sessionId, string manifestId, CancellationToken cancellationToken = default) + { + if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("tenantId is required", nameof(tenantId)); + if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("sessionId is required", nameof(sessionId)); + if (string.IsNullOrEmpty(manifestId)) throw new ArgumentException("manifestId is required", nameof(manifestId)); + + var blobName = BuildSnapshotBlobName(tenantId, sessionId, manifestId); + return await DeletionSnapshotBlobExistsAsync(blobName, cancellationToken); + } + + /// + /// Test seam: the production override calls Azure Blob Storage's HEAD; + /// tests subclass to return canned true/false without spinning up an SDK client. + /// + protected internal virtual async Task DeletionSnapshotBlobExistsAsync( + string blobName, CancellationToken cancellationToken) + { + var containerClient = _blobServiceClient.GetBlobContainerClient(DeletionManifestsContainer); + var blobClient = containerClient.GetBlobClient(blobName); + var response = await blobClient.ExistsAsync(cancellationToken); + return response.Value; + } + private static string BuildSnapshotBlobName(string tenantId, string sessionId, string manifestId) => $"{tenantId}/{sessionId}/{manifestId}.snapshot.json.gz"; diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs index 78e13c3b..01aa7c09 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionProducer.cs @@ -192,33 +192,59 @@ namespace AutopilotMonitor.Functions.Services.Deletion // worker picks it up. Idempotent — if the queue already has a message, // duplicates are handled by the worker via the manifest's CAS-bound // progress blob (Plan §5 PR3 step 4 + §16-R12). - await EnsureQueueExistsAsync(cancellationToken).ConfigureAwait(false); - var resumeEnvelope = new SessionDeletionEnvelope + return await ResumeBySendingQueueMessageAsync( + tenantId, sessionId, cas1.CurrentManifestId!, cas1.CurrentState!, reason, cancellationToken).ConfigureAwait(false); + } + if (cas1.CurrentState == SessionDeletionState.Preparing && !string.IsNullOrEmpty(cas1.CurrentManifestId)) + { + // Codex round-2 follow-up F4: a prior producer crashed AFTER uploading + // the snapshot + progress blobs but BEFORE the CAS Preparing→Queued + // step. Plan §10 GC only clears Preparing rows without a progress blob, + // so this case would otherwise sit forever. Probe the snapshot blob; + // if present, attempt the missing CAS Preparing→Queued + queue send + // to finish the resume. If absent (the prior crash was earlier in the + // build/upload phase), bail out with AlreadyInFlight and let the + // maintenance GC clean up. + var snapshotExists = false; + try { - TenantId = tenantId, - SessionId = sessionId, - ManifestId = cas1.CurrentManifestId!, - Reason = reason + ":resume", - EnqueuedAt = DateTime.UtcNow, - }; - await _queueClient.SendMessageAsync(JsonConvert.SerializeObject(resumeEnvelope), cancellationToken).ConfigureAwait(false); - _logger.LogInformation( - "SessionDeletionProducer re-enqueued (resume) for stranded Queued: tenant={TenantId} session={SessionId} manifestId={ManifestId}", - tenantId, sessionId, cas1.CurrentManifestId); - return new SessionDeletionEnqueueResult + snapshotExists = await _blob.DeletionSnapshotExistsAsync( + tenantId, sessionId, cas1.CurrentManifestId!, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) { - Outcome = SessionDeletionEnqueueOutcome.Enqueued, - ManifestId = cas1.CurrentManifestId, - ExistingState = cas1.CurrentState, - Reason = "resume", - }; + _logger.LogWarning(ex, + "DeletionSnapshotExistsAsync probe failed while attempting Preparing-resume; falling back to AlreadyInFlight. tenant={TenantId} session={SessionId} manifestId={ManifestId}", + tenantId, sessionId, cas1.CurrentManifestId); + } + if (snapshotExists) + { + var resumeCas = await _storage.CasSetSessionDeletionStateAsync( + tenantId, sessionId, + fromState: SessionDeletionState.Preparing, + toState: SessionDeletionState.Queued, + newManifestId: null, + cancellationToken).ConfigureAwait(false); + if (resumeCas.Outcome == TableStorageService.SessionDeletionStateCasOutcome.Updated) + { + _logger.LogInformation( + "Preparing-resume: snapshot blob present, CAS Preparing→Queued succeeded. tenant={TenantId} session={SessionId} manifestId={ManifestId}", + tenantId, sessionId, cas1.CurrentManifestId); + return await ResumeBySendingQueueMessageAsync( + tenantId, sessionId, cas1.CurrentManifestId!, SessionDeletionState.Preparing, reason, cancellationToken).ConfigureAwait(false); + } + _logger.LogInformation( + "Preparing-resume: CAS Preparing→Queued lost (outcome={Outcome}, currentState={CurrentState}); reporting AlreadyInFlight", + resumeCas.Outcome, resumeCas.CurrentState); + // Fall through to AlreadyInFlight below. + } } if (SessionDeletionState.IsLocked(cas1.CurrentState)) { - // Preparing or Running — caller decides UX (admin → 409 Conflict, - // maintenance → skip + audit). Preparing without progress blob is GC'd - // after 1h by PR6's maintenance function; Running is the worker actively - // processing. + // Preparing without a snapshot blob (or Running) — caller decides UX + // (admin → 409 Conflict, maintenance → skip + audit). Preparing without + // progress blob is GC'd after 1h by PR6's maintenance function; Running + // is the worker actively processing. return new SessionDeletionEnqueueResult { Outcome = SessionDeletionEnqueueOutcome.AlreadyInFlight, @@ -335,6 +361,38 @@ namespace AutopilotMonitor.Functions.Services.Deletion }; } + /// + /// Sends a SessionDeletionEnvelope with the SUPPLIED ManifestId (the stranded one we're + /// recovering) and returns with + /// Reason="resume". Used by both the Queued-resume and Preparing-resume paths so + /// the recovery is consistent and the audit log shows a unified ":resume" marker. + /// + private async Task ResumeBySendingQueueMessageAsync( + string tenantId, string sessionId, string existingManifestId, string fromState, + string originalReason, CancellationToken cancellationToken) + { + await EnsureQueueExistsAsync(cancellationToken).ConfigureAwait(false); + var resumeEnvelope = new SessionDeletionEnvelope + { + TenantId = tenantId, + SessionId = sessionId, + ManifestId = existingManifestId, + Reason = originalReason + ":resume", + EnqueuedAt = DateTime.UtcNow, + }; + await _queueClient.SendMessageAsync(JsonConvert.SerializeObject(resumeEnvelope), cancellationToken).ConfigureAwait(false); + _logger.LogInformation( + "SessionDeletionProducer re-enqueued (resume) from {FromState}: tenant={TenantId} session={SessionId} manifestId={ManifestId}", + fromState, tenantId, sessionId, existingManifestId); + return new SessionDeletionEnqueueResult + { + Outcome = SessionDeletionEnqueueOutcome.Enqueued, + ManifestId = existingManifestId, + ExistingState = fromState, + Reason = "resume", + }; + } + private async Task EnsureQueueExistsAsync(CancellationToken cancellationToken) { if (System.Threading.Interlocked.CompareExchange(ref _queueEnsured, 1, 0) == 0) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs index 315e5b23..518a8972 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs @@ -454,17 +454,69 @@ namespace AutopilotMonitor.Functions.Services if (!string.IsNullOrEmpty(existingIndexRowKey)) entity["IndexRowKey"] = existingIndexRowKey; - // PR3: preserve cascade-delete state-machine columns through the UpsertEntity (Replace). - // Without this, an agent re-register during Preparing/Queued/Running would silently - // clear the lock — letting the agent's subsequent writes race the in-flight cascade. - // Empty/null on first registration is fine; the columns are written by the producer's - // CAS, not by the agent path. - if (!string.IsNullOrEmpty(existingDeletionState)) - entity["DeletionState"] = existingDeletionState; - if (!string.IsNullOrEmpty(existingPendingDeletionManifestId)) - entity["PendingDeletionManifestId"] = existingPendingDeletionManifestId; + // PR3 (codex round 2 follow-up): ETag-bound CAS write loop. The plain + // UpsertEntity (Replace) had a TOCTOU race against the cascade-delete producer: + // (T1) StoreSessionAsync reads row → DeletionState=None, captures preserved fields. + // (T2) Producer concurrently CAS-es None → Preparing. + // (T3) StoreSessionAsync writes Replace → silently clobbers DeletionState=Preparing. + // ETag-bound UpdateEntity makes T3 fail with 412 instead, the loop re-reads the + // fresh row (now DeletionState=Preparing), re-stamps the cascade-lock columns + // onto our entity, and retries. The initial-read extract above is preserved as a + // best-effort hint; the per-iteration fresh read is what actually wins the race. + const int MaxStoreSessionCasAttempts = 5; + for (var casAttempt = 0; ; casAttempt++) + { + Azure.ETag? freshEtag = null; + string? freshDeletionState = null; + string? freshPendingDeletionManifestId = null; + try + { + var freshResponse = await tableClient.GetEntityAsync( + registration.TenantId, registration.SessionId, + select: new[] { "DeletionState", "PendingDeletionManifestId" }); + freshEtag = freshResponse.Value.ETag; + freshDeletionState = freshResponse.Value.GetString("DeletionState"); + freshPendingDeletionManifestId = freshResponse.Value.GetString("PendingDeletionManifestId"); + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + // Row was deleted between the initial extract above and now (cascade + // tombstone happened, or maintenance cleanup raced). Fall through to + // AddEntity — re-creates the row from registration data only. + } - await tableClient.UpsertEntityAsync(entity); + // Stamp cascade-lock columns from the FRESH read so the Replace below + // doesn't silently clear them. This kicks in even when the initial read + // saw None and the producer transitioned to Preparing in between. + if (!string.IsNullOrEmpty(freshDeletionState)) + entity["DeletionState"] = freshDeletionState; + else if (!string.IsNullOrEmpty(existingDeletionState)) + entity["DeletionState"] = existingDeletionState; + if (!string.IsNullOrEmpty(freshPendingDeletionManifestId)) + entity["PendingDeletionManifestId"] = freshPendingDeletionManifestId; + else if (!string.IsNullOrEmpty(existingPendingDeletionManifestId)) + entity["PendingDeletionManifestId"] = existingPendingDeletionManifestId; + + try + { + if (freshEtag.HasValue) + { + await tableClient.UpdateEntityAsync(entity, freshEtag.Value, TableUpdateMode.Replace); + } + else + { + await tableClient.AddEntityAsync(entity); + } + break; + } + catch (RequestFailedException ex) when ((ex.Status == 412 || ex.Status == 409) && casAttempt < MaxStoreSessionCasAttempts - 1) + { + _logger.LogDebug( + "StoreSessionAsync ETag CAS conflict (status={Status}, attempt={Attempt}); retrying for tenant={TenantId} session={SessionId}", + ex.Status, casAttempt + 1, registration.TenantId, registration.SessionId); + continue; + } + } // Dual-write: upsert into SessionsIndex for time-sorted listing await UpsertSessionIndexAsync(entity, startedAt);