fix(cv-qa): protect import lifecycle races and terminal receipts

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<NoContentResult>(await Controller(discarding).DiscardRun(id));
+        }
+        finally { gate.Release.TrySetResult(); }
+        Assert.IsType<ConflictObjectResult>(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<OkObjectResult>(await Controller(applying).ApplyReviewed(id, Review));
+        }
+        finally { gate.Release.TrySetResult(); }
+        Assert.IsType<ConflictObjectResult>(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<AppPaths>();
+        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<OkObjectResult>(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<OkObjectResult>(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<NoContentResult>(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<int> 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<ClaimsPrincipal>())).ReturnsAsync(user);
+        users.Setup(u => u.UpdateAsync(user)).Returns(async () =>
+        {
+            await db.SaveChangesAsync();
+            return IdentityResult.Success;
+        });
+        return new(users.Object, Mock.Of<ISummarizerService>(), 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<DbDataReader> 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;
+        }
+    }
+}

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); }
     }

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();
     }
 

JobTrackerApi/Data/JobTrackerContext.cs

@@ -290,6 +290,8 @@ namespace JobTrackerApi.Data
         modelBuilder.Entity<CvExtractionRun>()
             .HasIndex(x => new { x.OwnerUserId, x.StartedAtUtc });
 
+        modelBuilder.Entity<CvExtractionRun>().Property(x => x.Status).IsConcurrencyToken();
+
         modelBuilder.Entity<CvExtractionRun>()
             .HasOne(x => x.Artifact)
             .WithMany()

JobTrackerApi/Migrations/JobTrackerContextModelSnapshot.cs

@@ -1230,6 +1230,7 @@ namespace JobTrackerApi.Migrations
                         .HasColumnType("TEXT");
 
                     b.Property<string>("Status")
+                        .IsConcurrencyToken()
                         .IsRequired()
                         .HasColumnType("TEXT");
 

JobTrackerApi/Services/CvImportRetentionHostedService.cs

@@ -53,6 +53,8 @@ public sealed class CvImportRetentionHostedService(IServiceScopeFactory scopes,
             catch (IOException) { continue; }
             catch (UnauthorizedAccessException) { continue; }
             artifact.StoragePath = "";
+            for (var attempt = 0; ; attempt++)
+            {
                 run.DocumentJson = null;
                 run.RawExtractedText = null;
                 run.NormalizedText = null;
@@ -63,6 +65,14 @@ public sealed class CvImportRetentionHostedService(IServiceScopeFactory scopes,
                     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++;
         }
         await db.SaveChangesAsync(token);

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)

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.