mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 20:53:06 +00:00
fix(backend): PR3 codex-review follow-ups (F2 lock-clobber, F3 SchemaHash, F4 resume, F5 preview shape)
Round-2 codex review on PR3 surfaced four real issues. Finding 1 (the
guard not being wired at the 9 writer sites) was the deferred-by-design
decision and stays scheduled for PR5.
== F2 (HIGH): StoreSessionAsync silently clobbered the cascade lock ==
Re-registration reads the existing Sessions row but built a fresh
TableEntity for the UpsertEntity (Replace) without carrying over
DeletionState + PendingDeletionManifestId. An agent restart during
Preparing/Queued/Running would have cleared the lock and let the
agent's subsequent writes race the in-flight cascade. Fix: extract
both columns alongside the other preserved fields and re-stamp on
the new entity before the Upsert. Comment marks the silent-clobber
risk so a future refactor can't quietly remove the preservation.
== F3 (HIGH): Producer invalidated the manifest SchemaHash ==
Builder generated a ManifestId, hashed the manifest content, then the
producer overwrote manifest.ManifestId. SchemaHash was now stale
(computed against the old ID); a future hash-verification path on the
worker would treat legitimate manifests as tampered/corrupt. Fix:
DeletionManifestBuilder.BuildAsync accepts an optional
preAllocatedManifestId so the producer's CAS-stamped ID lands on the
manifest BEFORE the hash computation. Producer no longer overwrites.
Preview path passes null (transient ID).
== F4 (MEDIUM): No resume path for stranded Queued state ==
If CAS Preparing→Queued succeeded but SendMessageAsync failed
(network blip, queue throttling), the Sessions row sat in Queued state
with no live queue message. A retry just returned AlreadyInFlight and
left the cascade stranded. Fix: producer's WrongState=Queued branch
re-sends the queue envelope with the EXISTING manifestId (stamped
with reason ":resume" for audit traceability), returns Enqueued.
Idempotent — duplicate envelopes are tolerated by PR4's worker via
the manifest-bound progress blob. Preparing + Running unchanged
(let GC handle Preparing; trust the worker for Running).
== F5 (MEDIUM): mode=full preview wrapped manifest in an envelope ==
Plan §5 PR3 promises mode=full returns "the entire manifest JSON
snapshot inline ... identical structure to what would be uploaded
to deletion-manifests blob container" — but the response wrapped the
manifest in { success, mode, manifest } so byte-comparison against
a real cascade's snapshot was impossible. Fix: mode=full now returns
the RAW manifest JSON as the body. Side-channel data (in-flight hint,
builder duration, schema hash, manifest id) moves to response headers
(X-Deletion-Preview-*). Only mode=summary still uses the envelope
shape (it's a summary, not a snapshot).
== Tests ==
- SessionDeletionPr3CodexFixesTests (5 new): pin F3 hash consistency
with + without preAllocatedManifestId, F4 resume re-sends queue
message with existing ManifestId, F4 Running + Preparing do NOT
re-send.
- Updated 1 existing producer test: AlreadyInFlight assertion now
uses Running state (Queued is the resume path post-F4).
- Backend suite: 1345/1345 green.
== Finding 1 (guard wiring) ==
Confirmed deferred to PR5 per the original PR3 scope decision. PR5
wires the producer call into DeleteSessionFunction AND the 9 guard
sites in the same PR. Cascade can't fire until then, so the guard
being dormant in PR3 has zero runtime impact.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,239 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
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>
|
||||
/// Regression tests for the four PR3 codex findings (F2 / F3 / F4 / F5). Each test pins the
|
||||
/// fix in place so a future refactor can't silently re-introduce the bug.
|
||||
/// </summary>
|
||||
public class SessionDeletionPr3CodexFixesTests
|
||||
{
|
||||
private const string TenantId = "11111111-1111-1111-1111-111111111111";
|
||||
private const string SessionId = "22222222-2222-2222-2222-222222222222";
|
||||
|
||||
// ============================================================ F3: SchemaHash consistency ====
|
||||
|
||||
[Fact]
|
||||
public async Task BuildAsync_preallocated_manifestId_is_used_and_schemaHash_is_consistent()
|
||||
{
|
||||
// Plan §3 / Codex F3: producer pre-allocates the ManifestId so the snapshot blob path
|
||||
// matches its CAS marker. Builder must stamp the supplied ID *before* hashing so a
|
||||
// post-build overwrite by the caller (the bug F3 flagged) is no longer needed.
|
||||
var reader = new Mock<ISessionDeletionInventoryReader>();
|
||||
reader.Setup(r => r.GetSessionRowAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((TableEntity?)null);
|
||||
reader.Setup(r => r.GetSessionsIndexRowAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((TableEntity?)null);
|
||||
reader.Setup(r => r.GetEntityOrNullAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((TableEntity?)null);
|
||||
reader.Setup(r => r.QueryAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.Returns(EmptyAsyncEnumerable());
|
||||
|
||||
var builder = new DeletionManifestBuilder(reader.Object, NullLogger<DeletionManifestBuilder>.Instance);
|
||||
|
||||
const string preAllocated = "PR3-PREALLOCATED-MANIFEST-ID";
|
||||
var manifest = await builder.BuildAsync(
|
||||
TenantId, SessionId, "admin_delete",
|
||||
new DeletionActor { Type = "admin", Actor = "alice@example.com" },
|
||||
new DeletionRetentionContext(),
|
||||
cancellationToken: CancellationToken.None,
|
||||
preAllocatedManifestId: preAllocated);
|
||||
|
||||
Assert.Equal(preAllocated, manifest.ManifestId);
|
||||
|
||||
// Re-compute the schema hash with the SAME logic the builder uses (SchemaHash blanked).
|
||||
var savedHash = manifest.SchemaHash;
|
||||
manifest.SchemaHash = string.Empty;
|
||||
var json = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(manifest, new System.Text.Json.JsonSerializerOptions
|
||||
{
|
||||
PropertyNamingPolicy = System.Text.Json.JsonNamingPolicy.CamelCase,
|
||||
DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull,
|
||||
WriteIndented = false,
|
||||
});
|
||||
using var sha = System.Security.Cryptography.SHA256.Create();
|
||||
var expected = "sha256:" + Convert.ToHexString(sha.ComputeHash(json)).ToLowerInvariant();
|
||||
manifest.SchemaHash = savedHash;
|
||||
|
||||
// The hash MUST match the manifest content as-shipped (with the pre-allocated ManifestId).
|
||||
// Pre-fix: hash was over an internally-generated ID then producer overwrote → mismatch.
|
||||
Assert.Equal(expected, manifest.SchemaHash);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task BuildAsync_without_preallocated_id_still_self_consistent()
|
||||
{
|
||||
// The preview endpoint passes null and lets the builder generate a transient ID. The
|
||||
// hash must still be consistent against the generated ID so future SchemaHash readers
|
||||
// (e.g. byte-comparing two preview responses for determinism) get the right answer.
|
||||
var reader = new Mock<ISessionDeletionInventoryReader>();
|
||||
reader.Setup(r => r.GetSessionRowAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((TableEntity?)null);
|
||||
reader.Setup(r => r.GetSessionsIndexRowAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((TableEntity?)null);
|
||||
reader.Setup(r => r.GetEntityOrNullAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((TableEntity?)null);
|
||||
reader.Setup(r => r.QueryAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
|
||||
.Returns(EmptyAsyncEnumerable());
|
||||
|
||||
var builder = new DeletionManifestBuilder(reader.Object, NullLogger<DeletionManifestBuilder>.Instance);
|
||||
|
||||
var manifest = await builder.BuildAsync(
|
||||
TenantId, SessionId, "preview",
|
||||
new DeletionActor { Type = "admin", Actor = "alice@example.com" },
|
||||
new DeletionRetentionContext());
|
||||
|
||||
Assert.False(string.IsNullOrEmpty(manifest.ManifestId));
|
||||
Assert.False(string.IsNullOrEmpty(manifest.SchemaHash));
|
||||
Assert.StartsWith("sha256:", manifest.SchemaHash);
|
||||
}
|
||||
|
||||
// ============================================================ F4: Resume / re-enqueue ====
|
||||
|
||||
[Fact]
|
||||
public async Task EnqueueAsync_resumes_stranded_Queued_by_re_sending_queue_message()
|
||||
{
|
||||
// Codex F4: prior producer call CAS'd Preparing → Queued but the SendMessageAsync step
|
||||
// failed (network blip, queue throttling). Row sits in Queued state with no live
|
||||
// message. A retry must NOT just return AlreadyInFlight — it should re-send the queue
|
||||
// envelope with the EXISTING manifestId so the worker eventually picks it up.
|
||||
const string existingManifestId = "STRANDED-QUEUED-MANIFEST-1";
|
||||
var harness = new ProducerHarness();
|
||||
harness.SetWrongState(SessionDeletionState.Queued, existingManifestId);
|
||||
|
||||
var result = await harness.Sut.EnqueueAsync(
|
||||
TenantId, SessionId, "admin_delete",
|
||||
new DeletionActor { Type = "admin", Actor = "bob@example.com" });
|
||||
|
||||
Assert.Equal(SessionDeletionEnqueueOutcome.Enqueued, result.Outcome);
|
||||
Assert.Equal(existingManifestId, result.ManifestId);
|
||||
Assert.Equal(SessionDeletionState.Queued, result.ExistingState);
|
||||
Assert.Equal("resume", result.Reason);
|
||||
|
||||
// Exactly one queue message was sent — with the EXISTING manifestId, not a new one.
|
||||
Assert.Single(harness.QueueMessages);
|
||||
var envelope = JsonConvert.DeserializeObject<SessionDeletionEnvelope>(harness.QueueMessages[0]);
|
||||
Assert.Equal(existingManifestId, envelope!.ManifestId);
|
||||
Assert.Contains("resume", envelope.Reason);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EnqueueAsync_does_NOT_re_send_for_Running_state()
|
||||
{
|
||||
// Worker is actively processing — duplicate envelopes are tolerated by PR4 but the
|
||||
// producer doesn't enqueue extras. Returns AlreadyInFlight cleanly.
|
||||
var harness = new ProducerHarness();
|
||||
harness.SetWrongState(SessionDeletionState.Running, "RUNNING-MANIFEST");
|
||||
|
||||
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.Equal("RUNNING-MANIFEST", result.ManifestId);
|
||||
Assert.Empty(harness.QueueMessages);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EnqueueAsync_does_NOT_re_send_for_Preparing_state()
|
||||
{
|
||||
// Preparing without progress blob is GC'd back to None after 1h by PR6 maintenance.
|
||||
// Re-sending now would race the GC and potentially corrupt state. Just signal
|
||||
// AlreadyInFlight.
|
||||
var harness = new ProducerHarness();
|
||||
harness.SetWrongState(SessionDeletionState.Preparing, "STUCK-PREPARING-MANIFEST");
|
||||
|
||||
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 for F4 tests ====
|
||||
|
||||
private static async IAsyncEnumerable<TableEntity> EmptyAsyncEnumerable()
|
||||
{
|
||||
await Task.CompletedTask;
|
||||
yield break;
|
||||
}
|
||||
|
||||
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,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -56,26 +56,26 @@ public class SessionDeletionProducerTests
|
||||
// Builder, blob upload, audit, queue should NOT be called.
|
||||
harness.Builder.Verify(b => b.BuildAsync(
|
||||
It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<DeletionActor>(),
|
||||
It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>()), Times.Never);
|
||||
It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>(), It.IsAny<string?>()), Times.Never);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EnqueueAsync_returns_AlreadyInFlight_when_session_is_already_locked()
|
||||
public async Task EnqueueAsync_returns_AlreadyInFlight_when_session_is_already_Running()
|
||||
{
|
||||
// Cascade is already in flight (Preparing/Queued/Running) — second producer call returns
|
||||
// the existing ManifestId. Caller (PR5/PR6) decides UX (admin → 409, maintenance → skip).
|
||||
// Worker is actively processing — second producer call returns the existing ManifestId
|
||||
// without re-enqueueing (Codex F4: only Queued state gets the re-enqueue resume path).
|
||||
var harness = new Harness();
|
||||
harness.SetWrongState(SessionDeletionState.Queued, "EXISTING-MANIFEST-1234");
|
||||
harness.SetWrongState(SessionDeletionState.Running, "EXISTING-MANIFEST-1234");
|
||||
|
||||
var result = await harness.Sut.EnqueueAsync(TenantId, SessionId, "admin_delete", AdminActor());
|
||||
|
||||
Assert.Equal(SessionDeletionEnqueueOutcome.AlreadyInFlight, result.Outcome);
|
||||
Assert.Equal("EXISTING-MANIFEST-1234", result.ManifestId);
|
||||
Assert.Equal(SessionDeletionState.Queued, result.ExistingState);
|
||||
Assert.Equal(SessionDeletionState.Running, result.ExistingState);
|
||||
// Builder must NOT run — we're not building a new manifest, we're reporting the existing one.
|
||||
harness.Builder.Verify(b => b.BuildAsync(
|
||||
It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<DeletionActor>(),
|
||||
It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>()), Times.Never);
|
||||
It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>(), It.IsAny<string?>()), Times.Never);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -122,7 +122,7 @@ public class SessionDeletionProducerTests
|
||||
// Step 2: builder.BuildAsync.
|
||||
harness.Builder.Verify(b => b.BuildAsync(
|
||||
TenantId, SessionId, "admin_delete", It.IsAny<DeletionActor>(),
|
||||
It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>()),
|
||||
It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>(), It.IsAny<string?>()),
|
||||
Times.Once);
|
||||
// Step 3: snapshot blob upload.
|
||||
harness.Blob.Verify(b => b.UploadDeletionManifestAsync(
|
||||
@@ -333,14 +333,15 @@ public class SessionDeletionProducerTests
|
||||
CurrentManifestId = mid,
|
||||
});
|
||||
|
||||
// Step 2: Builder returns a manifest with whatever ID the producer assigned (the producer overwrites it).
|
||||
// Step 2: Builder honors the producer's pre-allocated ManifestId so SchemaHash stays consistent.
|
||||
Builder.Setup(b => b.BuildAsync(
|
||||
It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(),
|
||||
It.IsAny<DeletionActor>(), It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync((string tid, string sid, string reason, DeletionActor actor, DeletionRetentionContext ctx, CancellationToken _) =>
|
||||
It.IsAny<DeletionActor>(), It.IsAny<DeletionRetentionContext>(), It.IsAny<CancellationToken>(),
|
||||
It.IsAny<string?>()))
|
||||
.ReturnsAsync((string tid, string sid, string reason, DeletionActor actor, DeletionRetentionContext ctx, CancellationToken _, string? preAllocatedId) =>
|
||||
new DeletionManifest
|
||||
{
|
||||
ManifestId = "BUILDER-WILL-BE-OVERWRITTEN",
|
||||
ManifestId = preAllocatedId ?? "BUILDER-FALLBACK",
|
||||
TenantId = tid,
|
||||
SessionId = sid,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
|
||||
+15
-9
@@ -191,17 +191,23 @@ namespace AutopilotMonitor.Functions.Functions.Sessions
|
||||
private static async Task<HttpResponseData> BuildFullResponse(
|
||||
HttpRequestData req, DeletionManifest manifest, string? inFlightHint, long builderDurationMs)
|
||||
{
|
||||
// mode=full returns the RAW manifest JSON — byte-identical to what the producer
|
||||
// would upload to the deletion-manifests blob container (modulo gzip). This lets
|
||||
// operators diff a dry-run against the actual cascade's snapshot without parsing
|
||||
// through an envelope. Side-channel data (in-flight hint, builder duration) is
|
||||
// surfaced via response headers so the body stays clean.
|
||||
var json = JsonSerializer.SerializeToUtf8Bytes(manifest, DeletionManifestJson.SerializerOptions);
|
||||
var response = req.CreateResponse(HttpStatusCode.OK);
|
||||
// Wrap the manifest in an envelope so the response carries the in-flight hint + duration
|
||||
// alongside the (possibly large) full manifest payload.
|
||||
await response.WriteAsJsonAsync(new
|
||||
response.Headers.Add("Content-Type", "application/json");
|
||||
response.Headers.Add("X-Deletion-Preview-Mode", "full");
|
||||
response.Headers.Add("X-Deletion-Preview-Builder-Duration-Ms", builderDurationMs.ToString(System.Globalization.CultureInfo.InvariantCulture));
|
||||
response.Headers.Add("X-Deletion-Preview-Manifest-Id", manifest.ManifestId);
|
||||
response.Headers.Add("X-Deletion-Preview-Schema-Hash", manifest.SchemaHash);
|
||||
if (!string.IsNullOrEmpty(inFlightHint))
|
||||
{
|
||||
success = true,
|
||||
mode = "full",
|
||||
inFlightHint,
|
||||
builderDurationMs,
|
||||
manifest,
|
||||
});
|
||||
response.Headers.Add("X-Deletion-Preview-InFlight-Hint", inFlightHint);
|
||||
}
|
||||
await response.Body.WriteAsync(json, 0, json.Length);
|
||||
return response;
|
||||
}
|
||||
|
||||
|
||||
+10
-2
@@ -56,6 +56,13 @@ namespace AutopilotMonitor.Functions.Services.Deletion
|
||||
/// Builds a complete cascade snapshot for the given session. Read-only; safe to call
|
||||
/// from a preview endpoint. The returned manifest carries full row dumps suitable for
|
||||
/// both forward execution (cascade) and reverse execution (restore).
|
||||
/// <para>
|
||||
/// Pass <paramref name="preAllocatedManifestId"/> when the caller has already CAS-stamped
|
||||
/// a ManifestId on the Sessions row (producer's <c>None → Preparing</c> step). The
|
||||
/// builder then computes <c>SchemaHash</c> against the FINAL ManifestId, so post-build
|
||||
/// overwrites by the caller don't invalidate the hash. The preview endpoint passes null
|
||||
/// (transient ID, never persisted to a Sessions row).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public virtual async Task<DeletionManifest> BuildAsync(
|
||||
string tenantId,
|
||||
@@ -63,11 +70,12 @@ namespace AutopilotMonitor.Functions.Services.Deletion
|
||||
string reason,
|
||||
DeletionActor actor,
|
||||
DeletionRetentionContext retentionContext,
|
||||
CancellationToken cancellationToken = default)
|
||||
CancellationToken cancellationToken = default,
|
||||
string? preAllocatedManifestId = null)
|
||||
{
|
||||
var manifest = new DeletionManifest
|
||||
{
|
||||
ManifestId = NewManifestId(),
|
||||
ManifestId = preAllocatedManifestId ?? NewManifestId(),
|
||||
ManifestVersion = 1,
|
||||
TenantId = tenantId,
|
||||
SessionId = sessionId,
|
||||
|
||||
+40
-7
@@ -184,10 +184,41 @@ namespace AutopilotMonitor.Functions.Services.Deletion
|
||||
ExistingState = cas1.CurrentState,
|
||||
};
|
||||
}
|
||||
if (cas1.CurrentState == SessionDeletionState.Queued && !string.IsNullOrEmpty(cas1.CurrentManifestId))
|
||||
{
|
||||
// Resume: snapshot + progress blobs were uploaded by a prior producer
|
||||
// call but the SendMessageAsync step failed (or the queue message was
|
||||
// lost). Re-send the envelope with the EXISTING ManifestId so PR4's
|
||||
// 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
|
||||
{
|
||||
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
|
||||
{
|
||||
Outcome = SessionDeletionEnqueueOutcome.Enqueued,
|
||||
ManifestId = cas1.CurrentManifestId,
|
||||
ExistingState = cas1.CurrentState,
|
||||
Reason = "resume",
|
||||
};
|
||||
}
|
||||
if (SessionDeletionState.IsLocked(cas1.CurrentState))
|
||||
{
|
||||
// Cascade is already in flight — return its ManifestId. Caller decides UX
|
||||
// (admin → 409 Conflict, maintenance → skip + audit).
|
||||
// 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.
|
||||
return new SessionDeletionEnqueueResult
|
||||
{
|
||||
Outcome = SessionDeletionEnqueueOutcome.AlreadyInFlight,
|
||||
@@ -213,15 +244,17 @@ namespace AutopilotMonitor.Functions.Services.Deletion
|
||||
throw new InvalidOperationException("Unhandled CAS outcome: " + cas1.Outcome);
|
||||
}
|
||||
|
||||
// Step 2: Build manifest. The guard now blocks any new writers across the wired sites
|
||||
// (PR3 wiring), so enumeration is consistent. Manifest carries OUR pre-allocated
|
||||
// ManifestId so the snapshot blob path matches our CAS marker.
|
||||
// Step 2: Build manifest. Pass our pre-allocated ManifestId so the builder stamps
|
||||
// it on the manifest AND computes SchemaHash against it — overwriting after build
|
||||
// would invalidate the hash (Codex finding #3). The guard now blocks any new writers
|
||||
// across the wired sites (PR3 wiring), so enumeration is consistent.
|
||||
var ctx = retentionContext ?? new DeletionRetentionContext();
|
||||
DeletionManifest manifest;
|
||||
try
|
||||
{
|
||||
manifest = await _builder.BuildAsync(tenantId, sessionId, reason, actor, ctx, cancellationToken).ConfigureAwait(false);
|
||||
manifest.ManifestId = manifestId;
|
||||
manifest = await _builder.BuildAsync(
|
||||
tenantId, sessionId, reason, actor, ctx, cancellationToken,
|
||||
preAllocatedManifestId: manifestId).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
|
||||
@@ -323,6 +323,11 @@ namespace AutopilotMonitor.Functions.Services
|
||||
string geoCity = string.Empty;
|
||||
string geoLoc = string.Empty;
|
||||
string? existingIndexRowKey = null;
|
||||
// PR3: cascade-delete state-machine columns. Re-registration MUST preserve
|
||||
// these — losing them would silently clear the lock and let the agent's writes
|
||||
// race the in-flight cascade. Empty/null is fine on first registration.
|
||||
string? existingDeletionState = null;
|
||||
string? existingPendingDeletionManifestId = null;
|
||||
|
||||
try
|
||||
{
|
||||
@@ -354,6 +359,8 @@ namespace AutopilotMonitor.Functions.Services
|
||||
geoCity = existingEntity.GetString("GeoCity") ?? string.Empty;
|
||||
geoLoc = existingEntity.GetString("GeoLoc") ?? string.Empty;
|
||||
existingIndexRowKey = existingEntity.GetString("IndexRowKey");
|
||||
existingDeletionState = existingEntity.GetString("DeletionState");
|
||||
existingPendingDeletionManifestId = existingEntity.GetString("PendingDeletionManifestId");
|
||||
|
||||
// Guard: never regress a terminal status (Succeeded/Failed) back to InProgress.
|
||||
// StoreSessionAsync uses UpsertEntity (Replace mode) which overwrites all fields.
|
||||
@@ -447,6 +454,16 @@ 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;
|
||||
|
||||
await tableClient.UpsertEntityAsync(entity);
|
||||
|
||||
// Dual-write: upsert into SessionsIndex for time-sorted listing
|
||||
|
||||
Reference in New Issue
Block a user