mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 12:42:38 +00:00
fix(backend): PR3 codex round-2 follow-ups (F2 ETag-CAS, F4 Preparing-resume)
Round 2 of codex feedback flagged that the first follow-up wasn't tight
enough:
== F2 round 2 (HIGH): TOCTOU race still possible on Sessions row ==
The first fix only preserved DeletionState if it was non-empty on the
INITIAL read. The race remained: T1 StoreSessionAsync reads row →
DeletionState=None; T2 producer CAS-es None → Preparing; T3
StoreSessionAsync writes Replace → silently clobbers the lock. Codex
specifically asked for ETag/Retry oder Merge.
Fix: replace the unconditional UpsertEntityAsync with an ETag-bound CAS
write loop (bounded retry = 5 attempts). Each iteration re-reads the
CURRENT row to capture (a) the fresh ETag and (b) the live cascade-lock
columns, stamps the cascade-lock columns from that FRESH read onto our
entity (so the producer's intervening write isn't clobbered), then tries
UpdateEntity(etag, Replace). On 412/409 conflict: retry. Falls back to
AddEntity when the row was deleted between reads (tombstone race). The
preserved-fields extract from the initial read still runs as a
best-effort hint; the per-iteration fresh read is what actually wins
the race.
Replace mode is kept (not Merge) so registration can still clear stale
fields like FailureReason on a successful re-register. Only the
cascade-lock columns get the cross-iteration preservation treatment.
== F4 round 2 (MEDIUM): stranded Preparing with progress blob ==
If the prior producer call CAS'd None → Preparing AND uploaded snapshot
+ progress blobs AND audited deletion_started, then crashed before the
CAS Preparing → Queued step — the row stays in Preparing forever. Plan
§10 GC only clears Preparing-WITHOUT-progress-blob (intentional, to
avoid racing in-flight producers); the resume path I added before only
covered the Queued case. So this would sit until manual intervention.
Fix: in the WrongState=Preparing branch with non-empty
PendingDeletionManifestId, probe the snapshot blob via the new
BlobStorageService.DeletionSnapshotExistsAsync helper. If it exists,
the prior crash happened POST-upload — safe to finish the CAS
Preparing → Queued + send the queue message (resume from step 6). If
absent, the prior crash was PRE-upload — bail out as AlreadyInFlight
and let the §10 GC clean up after 1h.
Probe failure (e.g. transient blob storage outage) also falls back to
AlreadyInFlight rather than risk a half-completed recovery. Extracted
the queue-message-resume call into a shared helper
(ResumeBySendingQueueMessageAsync) so both Queued-resume and
Preparing-resume go through one audit-marker code path
(Reason="resume").
== Tests ==
- SessionDeletionPr3CodexFixesRound2Tests (4 new):
- Stranded Preparing-with-blob → resumes (CAS Preparing→Queued runs,
queue message sent, returns Enqueued + Reason=resume)
- Preparing without blob → AlreadyInFlight, no CAS attempt
- Snapshot existence probe throws → AlreadyInFlight, no CAS attempt
- Resume CAS loses to concurrent writer → AlreadyInFlight, no queue
message
- Backend suite: 1349/1349 green.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
+229
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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<string?>(), It.IsAny<CancellationToken>()),
|
||||
Times.Once);
|
||||
// And re-sent the queue message with the EXISTING manifestId.
|
||||
Assert.Single(harness.QueueMessages);
|
||||
var envelope = JsonConvert.DeserializeObject<SessionDeletionEnvelope>(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<string>(), It.IsAny<string>(),
|
||||
SessionDeletionState.Preparing, SessionDeletionState.Queued,
|
||||
It.IsAny<string?>(), It.IsAny<CancellationToken>()),
|
||||
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<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.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<TableStorageService> Storage { get; }
|
||||
public Mock<DeletionManifestBuilder> Builder { get; }
|
||||
public Mock<BlobStorageService> Blob { get; }
|
||||
public Mock<AdminConfigurationService> AdminConfig { get; }
|
||||
public Mock<IMaintenanceRepository> Maintenance { get; }
|
||||
public Mock<QueueClient> Queue { get; }
|
||||
public List<string> QueueMessages { get; } = new List<string>();
|
||||
public SessionDeletionProducer Sut { get; }
|
||||
|
||||
public ProducerHarness()
|
||||
{
|
||||
Storage = new Mock<TableStorageService>(
|
||||
Mock.Of<TableServiceClient>(), NullLogger<TableStorageService>.Instance);
|
||||
|
||||
Builder = new Mock<DeletionManifestBuilder>(
|
||||
Mock.Of<ISessionDeletionInventoryReader>(), NullLogger<DeletionManifestBuilder>.Instance);
|
||||
|
||||
Blob = new Mock<BlobStorageService>(
|
||||
new Azure.Storage.Blobs.BlobServiceClient("UseDevelopmentStorage=true"),
|
||||
NullLogger<BlobStorageService>.Instance, false);
|
||||
|
||||
AdminConfig = new Mock<AdminConfigurationService>(
|
||||
Mock.Of<IConfigRepository>(),
|
||||
NullLogger<AdminConfigurationService>.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<IMaintenanceRepository>();
|
||||
|
||||
Queue = new Mock<QueueClient>();
|
||||
Queue.Setup(q => q.SendMessageAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.Returns<string, CancellationToken>((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<Response>().Object));
|
||||
});
|
||||
Queue.Setup(q => q.CreateIfNotExistsAsync(It.IsAny<IDictionary<string, string>>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((Response?)null);
|
||||
|
||||
Sut = new SessionDeletionProducer(
|
||||
Storage.Object, Builder.Object, Blob.Object,
|
||||
AdminConfig.Object, Maintenance.Object, Queue.Object,
|
||||
NullLogger<SessionDeletionProducer>.Instance);
|
||||
}
|
||||
|
||||
public void SetWrongState(string currentState, string? currentManifestId)
|
||||
{
|
||||
Storage.Setup(s => s.CasSetSessionDeletionStateAsync(
|
||||
It.IsAny<string>(), It.IsAny<string>(),
|
||||
SessionDeletionState.None, SessionDeletionState.Preparing,
|
||||
It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(new TableStorageService.SessionDeletionStateCasResult
|
||||
{
|
||||
Outcome = TableStorageService.SessionDeletionStateCasOutcome.WrongState,
|
||||
CurrentState = currentState,
|
||||
CurrentManifestId = currentManifestId,
|
||||
});
|
||||
}
|
||||
|
||||
public void SetSnapshotExists(bool exists)
|
||||
{
|
||||
Blob.Setup(b => b.DeletionSnapshotExistsAsync(
|
||||
It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(exists);
|
||||
}
|
||||
|
||||
public void SetCas2Updated(string newState)
|
||||
{
|
||||
Storage.Setup(s => s.CasSetSessionDeletionStateAsync(
|
||||
It.IsAny<string>(), It.IsAny<string>(),
|
||||
SessionDeletionState.Preparing, SessionDeletionState.Queued,
|
||||
It.IsAny<string?>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(new TableStorageService.SessionDeletionStateCasResult
|
||||
{
|
||||
Outcome = TableStorageService.SessionDeletionStateCasOutcome.Updated,
|
||||
CurrentState = newState,
|
||||
});
|
||||
}
|
||||
|
||||
public void SetCas2WrongState(string fromState, string actualCurrentState)
|
||||
{
|
||||
Storage.Setup(s => s.CasSetSessionDeletionStateAsync(
|
||||
It.IsAny<string>(), It.IsAny<string>(),
|
||||
SessionDeletionState.Preparing, SessionDeletionState.Queued,
|
||||
It.IsAny<string?>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(new TableStorageService.SessionDeletionStateCasResult
|
||||
{
|
||||
Outcome = TableStorageService.SessionDeletionStateCasOutcome.WrongState,
|
||||
CurrentState = actualCurrentState,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -327,6 +327,36 @@ namespace AutopilotMonitor.Functions.Services
|
||||
return response.Value.ETag;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
public virtual async Task<bool> 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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Test seam: the production override calls Azure Blob Storage's HEAD;
|
||||
/// tests subclass to return canned true/false without spinning up an SDK client.
|
||||
/// </summary>
|
||||
protected internal virtual async Task<bool> 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";
|
||||
|
||||
|
||||
+80
-22
@@ -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
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a SessionDeletionEnvelope with the SUPPLIED ManifestId (the stranded one we're
|
||||
/// recovering) and returns <see cref="SessionDeletionEnqueueOutcome.Enqueued"/> with
|
||||
/// <c>Reason="resume"</c>. Used by both the Queued-resume and Preparing-resume paths so
|
||||
/// the recovery is consistent and the audit log shows a unified ":resume" marker.
|
||||
/// </summary>
|
||||
private async Task<SessionDeletionEnqueueResult> 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)
|
||||
|
||||
@@ -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<TableEntity>(
|
||||
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);
|
||||
|
||||
Reference in New Issue
Block a user