From 89974a257e4192b3d93779000a7df6998afc4051 Mon Sep 17 00:00:00 2001 From: cesnimda Date: Tue, 27 Oct 2026 00:27:57 +0100 Subject: [PATCH] fix(cv-qa): protect import lifecycle races and terminal receipts --- .../CvImportLifecycleConcurrencyTests.cs | 201 ++++++++++++++++++ .../Controllers/ProfileCvController.Import.cs | 2 + .../Controllers/ProfileCvController.cs | 8 +- JobTrackerApi/Data/JobTrackerContext.cs | 2 + .../JobTrackerContextModelSnapshot.cs | 1 + .../CvImportRetentionHostedService.cs | 26 ++- JobTrackerApi/Services/UserOperationStore.cs | 6 +- docs/cv-integration-qa/MARIA_DB_AND_RD21.md | 15 ++ 8 files changed, 250 insertions(+), 11 deletions(-) create mode 100644 JobTrackerApi.MariaDb.Tests/CvImportLifecycleConcurrencyTests.cs create mode 100644 docs/cv-integration-qa/MARIA_DB_AND_RD21.md diff --git a/JobTrackerApi.MariaDb.Tests/CvImportLifecycleConcurrencyTests.cs b/JobTrackerApi.MariaDb.Tests/CvImportLifecycleConcurrencyTests.cs new file mode 100644 index 0000000..e19d24a --- /dev/null +++ b/JobTrackerApi.MariaDb.Tests/CvImportLifecycleConcurrencyTests.cs @@ -0,0 +1,201 @@ +using System.Data.Common; +using System.Security.Claims; +using JobTrackerApi.Controllers; +using JobTrackerApi.Data; +using JobTrackerApi.Models; +using JobTrackerApi.Services; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Identity; +using Microsoft.AspNetCore.Mvc; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Microsoft.Extensions.DependencyInjection; +using Moq; +using Xunit; + +namespace JobTrackerApi.MariaDb.Tests; + +public class CvImportLifecycleConcurrencyTests +{ + private const string Owner = "rd21-owner"; + private static readonly ProfileCvController.ApplyReviewedRequest Review = new(0, new[] { new CvImportDecision("skills.0", "use") }); + + [Fact] + public async Task Acknowledged_discard_prevents_an_apply_that_already_read_pending() + { + await using var database = await MariaDbDatabase.CreateAsync(); + var id = await Seed(database); + var gate = new RunReadGate(); + await using var applying = database.Context(Owner, gate); + var apply = Controller(applying).ApplyReviewed(id, Review); + try + { + await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15)); + await using var discarding = database.Context(); + Assert.IsType(await Controller(discarding).DiscardRun(id)); + } + finally { gate.Release.TrySetResult(); } + Assert.IsType(await apply.WaitAsync(TimeSpan.FromSeconds(15))); + await applying.SaveChangesAsync(); + await using var verify = database.Context(); + Assert.Equal("discarded", (await verify.CvExtractionRuns.SingleAsync()).Status); + Assert.Empty(await verify.CareerProfileVersions.ToListAsync()); + Assert.Empty(await verify.CareerSkills.ToListAsync()); + Assert.Null((await verify.Users.SingleAsync()).CurrentCvExtractionRunId); + } + + [Fact] + public async Task Applied_review_prevents_a_discard_that_already_read_pending() + { + await using var database = await MariaDbDatabase.CreateAsync(); + var id = await Seed(database); + var gate = new RunReadGate(); + await using var discarding = database.Context(Owner, gate); + var discard = Controller(discarding).DiscardRun(id); + try + { + await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15)); + await using var applying = database.Context(); + Assert.IsType(await Controller(applying).ApplyReviewed(id, Review)); + } + finally { gate.Release.TrySetResult(); } + Assert.IsType(await discard.WaitAsync(TimeSpan.FromSeconds(15))); + await using var verify = database.Context(); + Assert.Equal("applied", (await verify.CvExtractionRuns.SingleAsync()).Status); + Assert.Equal("C#", (await verify.CareerSkills.SingleAsync()).Name); + Assert.Single(await verify.CareerProfileVersions.ToListAsync()); + } + + [Fact] + public async Task Retention_with_a_stale_pending_snapshot_preserves_applied_result_and_replay() + { + await using var database = await MariaDbDatabase.CreateAsync(); + var paths = database.App.Services.GetRequiredService(); + var source = Path.Combine(paths.CvArtifactsRoot, "lifecycle-synthetic.pdf"); + await File.WriteAllTextAsync(source, "Synthetic source"); + var id = await Seed(database, source); + var gate = new RunReadGate(); + await using var sweeping = database.Context(Owner, gate); + var sweep = CvImportRetentionHostedService.SweepAsync(sweeping, paths, DateTimeOffset.UtcNow, default); + string? candidate = null; + string? receipt = null; + try + { + await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15)); + await using var applying = database.Context(); + Assert.IsType(await Controller(applying).ApplyReviewed(id, Review)); + var saved = await applying.CvExtractionRuns.AsNoTracking().SingleAsync(); + candidate = saved.StructuredProfileJson; + receipt = saved.ReviewJson; + } + finally { gate.Release.TrySetResult(); } + await sweep.WaitAsync(TimeSpan.FromSeconds(15)); + await using var verify = database.Context(); + var run = await verify.CvExtractionRuns.SingleAsync(); + Assert.Equal("applied", run.Status); + Assert.Equal(candidate, run.StructuredProfileJson); + Assert.Equal(receipt, run.ReviewJson); + Assert.False(File.Exists(source)); + Assert.Equal("", (await verify.CvUploadArtifacts.SingleAsync()).StoragePath); + Assert.IsType(await Controller(verify).ApplyReviewed(id, Review)); + Assert.Single(await verify.CareerProfileVersions.ToListAsync()); + Assert.Equal("C#", (await verify.CareerSkills.SingleAsync()).Name); + } + + [Fact] + public async Task Cancellation_acknowledgement_preserves_a_discarded_review() + { + await using var database = await MariaDbDatabase.CreateAsync(); + var id = await Seed(database); + await using var db = database.Context(); + var operation = Operation(id, OperationStatuses.Running); + operation.CancellationRequestedAtUtc = DateTime.UtcNow; + db.UserOperations.Add(operation); + await db.SaveChangesAsync(); + Assert.IsType(await Controller(db).DiscardRun(id)); + var store = new UserOperationStore(db, TimeProvider.System); + Assert.Equal(1, await store.AcknowledgeCancellationAsync(operation.Id, "lease", default)); + await using var verify = database.Context(); + Assert.Equal("discarded", (await verify.CvExtractionRuns.SingleAsync()).Status); + Assert.Equal(OperationStatuses.Cancelled, (await verify.UserOperations.SingleAsync()).Status); + } + + [Theory] + [InlineData("discarded")] + [InlineData("expired")] + [InlineData("applied")] + public async Task Retry_cannot_reopen_a_terminal_review(string status) + { + await using var database = await MariaDbDatabase.CreateAsync(); + var id = await Seed(database); + await using var db = database.Context(); + var run = await db.CvExtractionRuns.SingleAsync(); + run.Status = status; + var operation = Operation(id, OperationStatuses.Cancelled); + db.UserOperations.Add(operation); + await db.SaveChangesAsync(); + Assert.False(await new UserOperationStore(db, TimeProvider.System).RetryAsync(operation.Id, default)); + await using var verify = database.Context(); + Assert.Equal(status, (await verify.CvExtractionRuns.SingleAsync()).Status); + Assert.Equal(OperationStatuses.Cancelled, (await verify.UserOperations.SingleAsync()).Status); + } + + private static UserOperation Operation(int runId, string status) => new() + { + Id = Guid.NewGuid(), OwnerUserId = Owner, TaskType = CvProcessingQueue.TaskType, + SubjectType = CvProcessingQueue.SubjectType, SubjectId = runId.ToString(System.Globalization.CultureInfo.InvariantCulture), + IdempotencyKey = "lifecycle", Status = status, LeaseToken = "lease", + CreatedAtUtc = DateTime.UtcNow, AvailableAtUtc = DateTime.UtcNow, + }; + + private static async Task Seed(MariaDbDatabase database, string? source = null) + { + await using var db = database.Context(); + db.Users.Add(new() { Id = Owner, UserName = "lifecycle-synthetic" }); + var run = new CvExtractionRun + { + OwnerUserId = Owner, Trigger = "document_import", ParserVersion = ProfileCvController.DocumentParserVersion, + Status = "pending_review", StartedAtUtc = DateTimeOffset.UtcNow.AddDays(-8), + RawExtractedText = "C#", DocumentJson = "{}", + StructuredProfileJson = StructuredCvProfileJson.SerializePersisted(new() { Skills = { "C#" } }), + Artifact = source is null ? null : new() { OwnerUserId = Owner, StoragePath = source, OriginalFileName = "synthetic.pdf" }, + }; + db.CvExtractionRuns.Add(run); + await db.SaveChangesAsync(); + return run.Id; + } + + private static ProfileCvController Controller(JobTrackerContext db) + { + var user = db.Users.Single(u => u.Id == Owner); + var users = TestHostFactory.CreateUserManager(); + users.Setup(u => u.GetUserAsync(It.IsAny())).ReturnsAsync(user); + users.Setup(u => u.UpdateAsync(user)).Returns(async () => + { + await db.SaveChangesAsync(); + return IdentityResult.Success; + }); + return new(users.Object, Mock.Of(), db, null!) + { ControllerContext = new() { HttpContext = new DefaultHttpContext() } }; + } + + // Pause after the server has executed the SELECT, so the result contains the old row. + private sealed class RunReadGate : DbCommandInterceptor + { + public TaskCompletionSource Entered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _entered; + + public override async ValueTask ReaderExecutedAsync(DbCommand command, CommandExecutedEventData eventData, DbDataReader result, CancellationToken cancellationToken = default) + { + if (command.CommandText.TrimStart().StartsWith("SELECT", StringComparison.OrdinalIgnoreCase) + && command.CommandText.Contains("`CvExtractionRuns`", StringComparison.Ordinal) + && Interlocked.Exchange(ref _entered, 1) == 0) + { + Entered.TrySetResult(); + await Release.Task.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken); + } + return result; + } + } +} diff --git a/JobTrackerApi/Controllers/ProfileCvController.Import.cs b/JobTrackerApi/Controllers/ProfileCvController.Import.cs index 26fbb75..a49ac77 100644 --- a/JobTrackerApi/Controllers/ProfileCvController.Import.cs +++ b/JobTrackerApi/Controllers/ProfileCvController.Import.cs @@ -239,6 +239,8 @@ public sealed partial class ProfileCvController }, HttpContext.RequestAborted); } catch (DbUpdateConcurrencyException) { return Conflict(new { code = "revision_conflict", message = "The career profile changed. Refresh the comparison and review your decisions." }); } + // MariaDB can report a changed run as a snapshot conflict before EF sees zero affected rows. + catch (DbUpdateException ex) when (ex.InnerException is MySqlConnector.MySqlException { Number: 1020 }) { return Conflict(new { code = "revision_conflict", message = "The import changed. Refresh the comparison and review your decisions." }); } catch (ArgumentException ex) { return BadRequest(ex.Message); } catch (CareerProjectionException ex) { return BadRequest(ex.Message); } } diff --git a/JobTrackerApi/Controllers/ProfileCvController.cs b/JobTrackerApi/Controllers/ProfileCvController.cs index c540a62..318dd4d 100644 --- a/JobTrackerApi/Controllers/ProfileCvController.cs +++ b/JobTrackerApi/Controllers/ProfileCvController.cs @@ -311,6 +311,7 @@ public sealed partial class ProfileCvController : ControllerBase }, HttpContext.RequestAborted); } catch (DbUpdateConcurrencyException) { return Conflict(new { code = "revision_conflict", message = "The career profile changed. Review the import again before accepting." }); } + catch (DbUpdateException ex) when (ex.InnerException is MySqlConnector.MySqlException { Number: 1020 }) { return Conflict(new { code = "revision_conflict", message = "The import changed. Review it again before accepting." }); } catch (CareerProjectionException ex) { return BadRequest(ex.Message); } } @@ -325,7 +326,12 @@ public sealed partial class ProfileCvController : ControllerBase if (run.Status != "pending_review") return Conflict("This extraction run is not awaiting review."); run.Status = "discarded"; - await _db.SaveChangesAsync(HttpContext.RequestAborted); + try { await _db.SaveChangesAsync(HttpContext.RequestAborted); } + catch (DbUpdateException ex) when (ex is DbUpdateConcurrencyException || ex.InnerException is MySqlConnector.MySqlException { Number: 1020 }) + { + _db.Entry(run).State = EntityState.Detached; + return Conflict("This extraction run changed. Refresh before discarding it."); + } return NoContent(); } diff --git a/JobTrackerApi/Data/JobTrackerContext.cs b/JobTrackerApi/Data/JobTrackerContext.cs index 899926e..2922b1e 100644 --- a/JobTrackerApi/Data/JobTrackerContext.cs +++ b/JobTrackerApi/Data/JobTrackerContext.cs @@ -290,6 +290,8 @@ namespace JobTrackerApi.Data modelBuilder.Entity() .HasIndex(x => new { x.OwnerUserId, x.StartedAtUtc }); + modelBuilder.Entity().Property(x => x.Status).IsConcurrencyToken(); + modelBuilder.Entity() .HasOne(x => x.Artifact) .WithMany() diff --git a/JobTrackerApi/Migrations/JobTrackerContextModelSnapshot.cs b/JobTrackerApi/Migrations/JobTrackerContextModelSnapshot.cs index f572f3e..8d6f6bb 100644 --- a/JobTrackerApi/Migrations/JobTrackerContextModelSnapshot.cs +++ b/JobTrackerApi/Migrations/JobTrackerContextModelSnapshot.cs @@ -1230,6 +1230,7 @@ namespace JobTrackerApi.Migrations .HasColumnType("TEXT"); b.Property("Status") + .IsConcurrencyToken() .IsRequired() .HasColumnType("TEXT"); diff --git a/JobTrackerApi/Services/CvImportRetentionHostedService.cs b/JobTrackerApi/Services/CvImportRetentionHostedService.cs index c7a265d..c21b614 100644 --- a/JobTrackerApi/Services/CvImportRetentionHostedService.cs +++ b/JobTrackerApi/Services/CvImportRetentionHostedService.cs @@ -53,15 +53,25 @@ public sealed class CvImportRetentionHostedService(IServiceScopeFactory scopes, catch (IOException) { continue; } catch (UnauthorizedAccessException) { continue; } artifact.StoragePath = ""; - run.DocumentJson = null; - run.RawExtractedText = null; - run.NormalizedText = null; - if (run.Status != "applied") + for (var attempt = 0; ; attempt++) { - run.StructuredProfileJson = null; - if (run.Status is "pending_review" or "queued" or "running") run.Status = "expired"; - run.CompletedAtUtc ??= now; - run.ErrorMessage = "The source document expired. Upload it again to start a new review."; + run.DocumentJson = null; + run.RawExtractedText = null; + run.NormalizedText = null; + if (run.Status != "applied") + { + run.StructuredProfileJson = null; + if (run.Status is "pending_review" or "queued" or "running") run.Status = "expired"; + run.CompletedAtUtc ??= now; + run.ErrorMessage = "The source document expired. Upload it again to start a new review."; + } + try { await db.SaveChangesAsync(token); break; } + catch (DbUpdateConcurrencyException ex) when (attempt < 2 && ex.Entries.All(e => ReferenceEquals(e.Entity, run))) + { + // Retry source cleanup against the new status; never replay canonical profile writes. + await db.Entry(run).ReloadAsync(token); + if (db.Entry(run).State == EntityState.Detached) break; + } } removed++; } diff --git a/JobTrackerApi/Services/UserOperationStore.cs b/JobTrackerApi/Services/UserOperationStore.cs index b418fe9..9e56b99 100644 --- a/JobTrackerApi/Services/UserOperationStore.cs +++ b/JobTrackerApi/Services/UserOperationStore.cs @@ -355,6 +355,8 @@ public sealed class UserOperationStore(JobTrackerContext db, TimeProvider timePr await using var transaction = await BeginTransactionAsync(cancellationToken); var operation = await db.UserOperations.FirstOrDefaultAsync(item => item.Id == operationId, cancellationToken); if (operation is null || operation.Status is not (OperationStatuses.Failed or OperationStatuses.Cancelled)) return false; + if (operation.TaskType == CvProcessingQueue.TaskType && operation.SubjectType == CvProcessingQueue.SubjectType + && await SynchronizeCvRunAsync(operation, OperationStatuses.Queued, null, UtcNow, cancellationToken) != 1) return false; operation.Status = OperationStatuses.Queued; operation.AttemptCount = 0; operation.AvailableAtUtc = UtcNow; @@ -371,7 +373,6 @@ public sealed class UserOperationStore(JobTrackerContext db, TimeProvider timePr var existingNotification = await db.UserNotifications.FirstOrDefaultAsync(item => item.OperationId == operationId, cancellationToken); if (existingNotification is not null) db.UserNotifications.Remove(existingNotification); await db.SaveChangesAsync(cancellationToken); - await SynchronizeCvRunAsync(operation, OperationStatuses.Queued, null, UtcNow, cancellationToken); if (transaction is not null) await transaction.CommitAsync(cancellationToken); return true; } @@ -475,7 +476,8 @@ public sealed class UserOperationStore(JobTrackerContext db, TimeProvider timePr : (DateTimeOffset?)null; return db.CvExtractionRuns.IgnoreQueryFilters() - .Where(run => run.Id == runId && run.OwnerUserId == operation.OwnerUserId && run.Status != "expired" && run.Status != "applied") + .Where(run => run.Id == runId && run.OwnerUserId == operation.OwnerUserId + && (run.Status == "queued" || run.Status == "running" || run.Status == "pending_review" || run.Status == "failed" || run.Status == "cancelled")) .ExecuteUpdateAsync(setters => setters .SetProperty(run => run.Status, runStatus) .SetProperty(run => run.ErrorMessage, message) diff --git a/docs/cv-integration-qa/MARIA_DB_AND_RD21.md b/docs/cv-integration-qa/MARIA_DB_AND_RD21.md new file mode 100644 index 0000000..d2fafd3 --- /dev/null +++ b/docs/cv-integration-qa/MARIA_DB_AND_RD21.md @@ -0,0 +1,15 @@ +# MariaDB and RD21 — 2026-10-27 + +The disposable real MariaDB lane passed **79/79**, zero skips, with successful startup, locked restore, test and owned-container cleanup (`.artifacts/cv-integration-qa/lifecycle-green-01/run.json`). The entry baseline was 70/70; the first mapping checkpoint added two provider cases and lifecycle qualification added seven. These are successive totals, not additive runs. Final source-freeze qualification is reported separately in REGRESSION_RESULTS.md. + +Canonical writes retain the existing Career Profile transaction, expected revision, relational projections, history and durable import receipt. No historical CV, CV override, submitted package or frozen letter is rewritten. Mapping-only multi-record imports into an empty profile cover contacts/links, two jobs, two education records, languages, certification/project and long-tail sections. Actual readback verifies child rows, source identifiers, incomplete date precision and rejected fields. Exact replay produces no extra revision. Injected projection failure rolls all changes back. + +## Reproduced lifecycle races + +Independent contexts use finite database-command barriers after SELECT; tests do not substitute an in-memory provider. The retained RED attempt had seven failures and 72 passes. Apply after a competing discard actually raised MariaDB 1020 (`record has changed since last read`) rather than silently committing. Reverse discard/apply, stale retention, cancel acknowledgement and retries of discarded/expired/applied jobs also failed their assertions. + +The run status is now an EF concurrency token, with matching snapshot metadata and **no physical column or migration**. A losing discard returns 409. Apply handles EF stale writes and narrowly identified MariaDB 1020 after the established transaction rollback. Retention retries each conflicting run at most twice and recomputes cleanup from reloaded state; it preserves applied candidate/receipt data while expiring the source file. Cancellation cannot overwrite a discarded terminal state. Retry claims only an eligible mutable CV run atomically before queueing; terminal runs cannot reopen. + +The GREEN cases confirm no canonical write after discard, no discard after apply, retained applied receipts and exact replay after retention, terminal cancellation preservation and rejected terminal retries. Existing owner/tenant and first/existing-profile revision tests remain enabled. A displayed success still requires confirmed durable application. + +This qualifies the exercised disposable MariaDB behavior. It does not establish hosted CI, production database behavior, power-loss recovery or server filesystem/backup integrity. Historical RD21 evidence is preserved; current-source focused races complement rather than replace its full regression lanes.