fix(deployment): pending-deployment supersede/purge/stage deletes never read ConfigurationJson; drop stale deferred-TODO comment (review 2026-10-21 08/P6, 08/C2)
This commit is contained in:
+73
-24
@@ -339,12 +339,19 @@ public class DeploymentManagerRepository : IDeploymentManagerRepository
|
|||||||
// Supersession: a newer deploy for the same instance replaces any prior pending
|
// Supersession: a newer deploy for the same instance replaces any prior pending
|
||||||
// row. Safe because the per-instance operation lock serializes same-instance
|
// row. Safe because the per-instance operation lock serializes same-instance
|
||||||
// deploys. SaveChanges is deferred to the caller, matching the other Add methods.
|
// deploys. SaveChanges is deferred to the caller, matching the other Add methods.
|
||||||
var prior = await _dbContext.Set<PendingDeployment>()
|
//
|
||||||
|
// Only the prior rows' KEYS are read: each row carries the whole flattened
|
||||||
|
// ConfigurationJson (tens of KB), and materialising it just to delete it was the
|
||||||
|
// dominant cost of the supersede. The delete itself stays a tracked Remove (not
|
||||||
|
// ExecuteDelete) so it still commits atomically with the new row in the caller's
|
||||||
|
// SaveChanges.
|
||||||
|
var priorIds = await _dbContext.Set<PendingDeployment>()
|
||||||
.Where(p => p.InstanceId == pending.InstanceId)
|
.Where(p => p.InstanceId == pending.InstanceId)
|
||||||
|
.Select(p => p.Id)
|
||||||
.ToListAsync(cancellationToken);
|
.ToListAsync(cancellationToken);
|
||||||
if (prior.Count > 0)
|
foreach (var id in priorIds)
|
||||||
{
|
{
|
||||||
_dbContext.Set<PendingDeployment>().RemoveRange(prior);
|
RemovePendingByKey(id);
|
||||||
}
|
}
|
||||||
await _dbContext.Set<PendingDeployment>().AddAsync(pending, cancellationToken);
|
await _dbContext.Set<PendingDeployment>().AddAsync(pending, cancellationToken);
|
||||||
}
|
}
|
||||||
@@ -395,16 +402,61 @@ public class DeploymentManagerRepository : IDeploymentManagerRepository
|
|||||||
public async Task<int> PurgeExpiredPendingDeploymentsAsync(DateTimeOffset nowUtc, CancellationToken cancellationToken = default)
|
public async Task<int> PurgeExpiredPendingDeploymentsAsync(DateTimeOffset nowUtc, CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
// Self-contained TTL maintenance op: commits its own delete and returns the count.
|
// Self-contained TTL maintenance op: commits its own delete and returns the count.
|
||||||
var expired = await _dbContext.Set<PendingDeployment>()
|
// A set-based DELETE — the expired rows' ConfigurationJson is never read.
|
||||||
.Where(p => p.ExpiresAtUtc <= nowUtc)
|
return await DeletePendingWhereAsync(p => p.ExpiresAtUtc <= nowUtc, cancellationToken);
|
||||||
.ToListAsync(cancellationToken);
|
}
|
||||||
if (expired.Count == 0)
|
|
||||||
|
/// <summary>
|
||||||
|
/// Deletes every pending row matching <paramref name="predicate"/> immediately, without
|
||||||
|
/// materialising the rows (each carries its full flattened <c>ConfigurationJson</c>).
|
||||||
|
/// A relational provider issues one set-based <c>DELETE</c>; a non-relational provider
|
||||||
|
/// (the EF in-memory provider some host-level tests run on) cannot translate
|
||||||
|
/// <c>ExecuteDelete</c>, so it falls back to load-and-remove with its own save.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="predicate">Rows to delete.</param>
|
||||||
|
/// <param name="cancellationToken">Cancellation token.</param>
|
||||||
|
/// <returns>The number of rows deleted.</returns>
|
||||||
|
private async Task<int> DeletePendingWhereAsync(
|
||||||
|
System.Linq.Expressions.Expression<Func<PendingDeployment, bool>> predicate,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
var query = _dbContext.Set<PendingDeployment>().Where(predicate);
|
||||||
|
if (_dbContext.Database.IsRelational())
|
||||||
|
{
|
||||||
|
return await query.ExecuteDeleteAsync(cancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
|
var rows = await query.ToListAsync(cancellationToken);
|
||||||
|
if (rows.Count == 0)
|
||||||
{
|
{
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
_dbContext.Set<PendingDeployment>().RemoveRange(expired);
|
_dbContext.Set<PendingDeployment>().RemoveRange(rows);
|
||||||
await _dbContext.SaveChangesAsync(cancellationToken);
|
await _dbContext.SaveChangesAsync(cancellationToken);
|
||||||
return expired.Count;
|
return rows.Count;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Marks the pending row with primary key <paramref name="id"/> for deletion on the next
|
||||||
|
/// <c>SaveChanges</c> without loading it: an already-tracked instance is removed in place,
|
||||||
|
/// otherwise a key-only stub is attached and removed (EF deletes by key).
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="id">Primary key of the pending row.</param>
|
||||||
|
private void RemovePendingByKey(int id)
|
||||||
|
{
|
||||||
|
var set = _dbContext.Set<PendingDeployment>();
|
||||||
|
var tracked = set.Local.FirstOrDefault(p => p.Id == id);
|
||||||
|
if (tracked != null)
|
||||||
|
{
|
||||||
|
set.Remove(tracked);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
var stub = new PendingDeployment(
|
||||||
|
string.Empty, 0, string.Empty, string.Empty, string.Empty,
|
||||||
|
default, default) { Id = id };
|
||||||
|
set.Attach(stub);
|
||||||
|
set.Remove(stub);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
@@ -467,18 +519,15 @@ public class DeploymentManagerRepository : IDeploymentManagerRepository
|
|||||||
CancellationToken cancellationToken = default)
|
CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
// Treat createdAtUtc as "now". FIRST remove any EXPIRED pending rows for this instance
|
// Treat createdAtUtc as "now". FIRST remove any EXPIRED pending rows for this instance
|
||||||
// (ExpiresAtUtc <= now). Pending rows are only TTL-purged (the periodic purge is still a
|
// (ExpiresAtUtc <= now). Pending rows are only TTL-purged — PendingDeploymentPurgeActor
|
||||||
// deferred TODO), so an EXPIRED-but-unpurged row would otherwise (a) read as "a deploy is
|
// sweeps them on its PendingDeploymentPurgeInterval cadence — so between sweeps an
|
||||||
// in flight" and block staging — handing the node an expired token (HTTP 404) and leaving
|
// EXPIRED-but-unpurged row would otherwise (a) read as "a deploy is in flight" and block
|
||||||
// it unhealed — and (b) collide on the DeploymentId UNIQUE index when a reconcile re-stages
|
// staging — handing the node an expired token (HTTP 404) and leaving it unhealed — and
|
||||||
// the snapshot's own DeploymentId. Dropping expired rows first fixes both.
|
// (b) collide on the DeploymentId UNIQUE index when a reconcile re-stages the snapshot's
|
||||||
var expired = await _dbContext.Set<PendingDeployment>()
|
// own DeploymentId. Dropping expired rows first fixes both. The delete is set-based and
|
||||||
.Where(p => p.InstanceId == instanceId && p.ExpiresAtUtc <= createdAtUtc)
|
// immediate (an expired row is dead either way), so its JSON is never read.
|
||||||
.ToListAsync(cancellationToken);
|
await DeletePendingWhereAsync(
|
||||||
if (expired.Count > 0)
|
p => p.InstanceId == instanceId && p.ExpiresAtUtc <= createdAtUtc, cancellationToken);
|
||||||
{
|
|
||||||
_dbContext.Set<PendingDeployment>().RemoveRange(expired);
|
|
||||||
}
|
|
||||||
|
|
||||||
// THEN insert-if-absent against still-LIVE rows only. A live pending row means a genuine
|
// THEN insert-if-absent against still-LIVE rows only. A live pending row means a genuine
|
||||||
// in-flight deploy (or a concurrent reconcile) already owns the slot — do NOT supersede it;
|
// in-flight deploy (or a concurrent reconcile) already owns the slot — do NOT supersede it;
|
||||||
@@ -494,9 +543,9 @@ public class DeploymentManagerRepository : IDeploymentManagerRepository
|
|||||||
await _dbContext.Set<PendingDeployment>().AddAsync(pending, cancellationToken);
|
await _dbContext.Set<PendingDeployment>().AddAsync(pending, cancellationToken);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Self-contained: one SaveChanges flushes the expired-row cleanup and, when staged, the new
|
// Self-contained: the expired-row cleanup has already run, so the same-DeploymentId insert
|
||||||
// row together (EF orders the delete before the same-DeploymentId insert to satisfy the
|
// below cannot collide with it on the unique index. Returns true only when a fresh row was
|
||||||
// unique index). Returns true only when a fresh row was staged.
|
// staged.
|
||||||
await _dbContext.SaveChangesAsync(cancellationToken);
|
await _dbContext.SaveChangesAsync(cancellationToken);
|
||||||
return !liveExists;
|
return !liveExists;
|
||||||
}
|
}
|
||||||
|
|||||||
+73
-1
@@ -18,11 +18,12 @@ public class PendingDeploymentRepositoryTests : IDisposable
|
|||||||
{
|
{
|
||||||
private readonly ScadaBridgeDbContext _context;
|
private readonly ScadaBridgeDbContext _context;
|
||||||
private readonly DeploymentManagerRepository _repository;
|
private readonly DeploymentManagerRepository _repository;
|
||||||
|
private readonly SqlReaderCommandRecorder _recorder = new();
|
||||||
|
|
||||||
/// <summary>Initializes the test class with an in-memory context and repository.</summary>
|
/// <summary>Initializes the test class with an in-memory context and repository.</summary>
|
||||||
public PendingDeploymentRepositoryTests()
|
public PendingDeploymentRepositoryTests()
|
||||||
{
|
{
|
||||||
_context = SqliteTestHelper.CreateInMemoryContext();
|
_context = SqliteTestHelper.CreateInMemoryContext(_recorder);
|
||||||
_repository = new DeploymentManagerRepository(_context);
|
_repository = new DeploymentManagerRepository(_context);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -344,4 +345,75 @@ public class PendingDeploymentRepositoryTests : IDisposable
|
|||||||
Assert.Null(await _repository.GetPendingDeploymentByIdAsync("stale-b"));
|
Assert.Null(await _repository.GetPendingDeploymentByIdAsync("stale-b"));
|
||||||
Assert.NotNull(await _repository.GetPendingDeploymentByIdAsync("dep3"));
|
Assert.NotNull(await _repository.GetPendingDeploymentByIdAsync("dep3"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── review 2026-10-21 08/P6: pending deletes never materialise ConfigurationJson ──
|
||||||
|
|
||||||
|
/// <summary>True when the change tracker holds a pending row whose JSON is <paramref name="json"/>.</summary>
|
||||||
|
private bool TracksPendingJson(string json) =>
|
||||||
|
_context.ChangeTracker.Entries<PendingDeployment>().Any(e => e.Entity.ConfigurationJson == json);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Supersession must delete the prior row by key, without loading its (tens-of-KB)
|
||||||
|
/// flattened config into the change tracker.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task AddPendingDeployment_Supersede_DoesNotMaterialisePriorConfigJson()
|
||||||
|
{
|
||||||
|
var instanceId = await SeedInstanceAsync("InstP6a");
|
||||||
|
await _repository.AddPendingDeploymentAsync(NewPending("old", instanceId, "{\"prior\":true}"));
|
||||||
|
await _repository.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
_recorder.Reset();
|
||||||
|
|
||||||
|
await _repository.AddPendingDeploymentAsync(NewPending("new", instanceId, "{\"v\":2}"));
|
||||||
|
|
||||||
|
Assert.False(TracksPendingJson("{\"prior\":true}"));
|
||||||
|
Assert.False(_recorder.AnySelectReads("ConfigurationJson"));
|
||||||
|
await _repository.SaveChangesAsync();
|
||||||
|
Assert.Null(await _repository.GetPendingDeploymentByIdAsync("old"));
|
||||||
|
Assert.NotNull(await _repository.GetPendingDeploymentByIdAsync("new"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>The TTL purge is a set-based delete: nothing is loaded into the tracker.</summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task PurgeExpiredPendingDeployments_DoesNotMaterialiseRows()
|
||||||
|
{
|
||||||
|
var instanceId = await SeedInstanceAsync("InstP6b");
|
||||||
|
var now = DateTimeOffset.UtcNow;
|
||||||
|
await _repository.AddPendingDeploymentAsync(new PendingDeployment(
|
||||||
|
"exp", instanceId, "rev", "{\"expired\":true}", "tok", now.AddMinutes(-20), now.AddMinutes(-10)));
|
||||||
|
await _repository.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
_recorder.Reset();
|
||||||
|
|
||||||
|
var purged = await _repository.PurgeExpiredPendingDeploymentsAsync(now);
|
||||||
|
|
||||||
|
Assert.Equal(1, purged);
|
||||||
|
Assert.False(_recorder.AnySelectReads("ConfigurationJson"));
|
||||||
|
Assert.Null(await _repository.GetPendingDeploymentByIdAsync("exp"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>StagePendingIfAbsent drops expired rows without reading their config.</summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task StagePendingIfAbsent_ExpiredCleanup_DoesNotMaterialiseExpiredConfigJson()
|
||||||
|
{
|
||||||
|
var instanceId = await SeedInstanceAsync("InstP6c");
|
||||||
|
var now = DateTimeOffset.UtcNow;
|
||||||
|
await _repository.AddPendingDeploymentAsync(new PendingDeployment(
|
||||||
|
"dep-x", instanceId, "rev", "{\"expired\":true}", "tok", now.AddMinutes(-20), now.AddMinutes(-10)));
|
||||||
|
await _repository.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
_recorder.Reset();
|
||||||
|
|
||||||
|
var staged = await _repository.StagePendingIfAbsentAsync(
|
||||||
|
instanceId, "dep-x", "rev2", "{\"fresh\":true}", "tok2", now, now.AddMinutes(5));
|
||||||
|
|
||||||
|
Assert.True(staged);
|
||||||
|
Assert.False(_recorder.AnySelectReads("ConfigurationJson"));
|
||||||
|
var row = await _repository.GetPendingDeploymentByIdAsync("dep-x");
|
||||||
|
Assert.Equal("{\"fresh\":true}", row!.ConfigurationJson);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,46 @@
|
|||||||
|
using System.Data.Common;
|
||||||
|
using Microsoft.EntityFrameworkCore.Diagnostics;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Records the text of every reader command a context executes, so a query-shape test can
|
||||||
|
/// assert what a repository method actually reads (e.g. that a delete never SELECTs a fat
|
||||||
|
/// JSON column).
|
||||||
|
/// </summary>
|
||||||
|
public sealed class SqlReaderCommandRecorder : DbCommandInterceptor
|
||||||
|
{
|
||||||
|
private readonly List<string> _commands = new();
|
||||||
|
|
||||||
|
/// <summary>The text of every reader command recorded since the last <see cref="Reset"/>.</summary>
|
||||||
|
public IReadOnlyList<string> Commands => _commands;
|
||||||
|
|
||||||
|
/// <summary>Clears all recorded commands.</summary>
|
||||||
|
public void Reset() => _commands.Clear();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// True when any recorded reader command is a SELECT that projects <paramref name="column"/>.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="column">Column name to look for.</param>
|
||||||
|
/// <returns>Whether a recorded SELECT read the column.</returns>
|
||||||
|
public bool AnySelectReads(string column) =>
|
||||||
|
_commands.Any(c => c.TrimStart().StartsWith("SELECT", StringComparison.OrdinalIgnoreCase)
|
||||||
|
&& c.Contains(column, StringComparison.Ordinal));
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public override InterceptionResult<DbDataReader> ReaderExecuting(
|
||||||
|
DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
|
||||||
|
{
|
||||||
|
_commands.Add(command.CommandText);
|
||||||
|
return base.ReaderExecuting(command, eventData, result);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public override ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(
|
||||||
|
DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result,
|
||||||
|
CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
_commands.Add(command.CommandText);
|
||||||
|
return base.ReaderExecutingAsync(command, eventData, result, cancellationToken);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user