diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs index 594c9bcf..f486baab 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs @@ -42,7 +42,9 @@ public interface IKpiHistoryRepository /// /// Bulk-deletes rows whose is strictly older - /// than (retention purge). + /// than (retention purge). Implementations slice the + /// purge into time-windowed DELETE batches so a large catch-up (e.g. after an + /// outage) never runs as a single lock/log-heavy transaction (arch-review 04, P4). /// /// UTC cut-off; rows captured before this instant are deleted. /// Cancellation token. diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs index e6a9fc09..421e764f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs @@ -64,9 +64,22 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository /// public async Task PurgeOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) { - // Set-based delete — no entity materialisation; returns the rows affected. - return await _context.KpiSamples - .Where(s => s.CapturedAtUtc < before) - .ExecuteDeleteAsync(cancellationToken); + // Time-sliced batches: each DELETE covers at most one hour of samples, capping the + // lock/log footprint per statement (arch-review 04, P4 — steady state is ~1 day of + // rows/day; after an outage the catch-up would otherwise be one giant transaction). + var total = 0; + var floor = await _context.KpiSamples.Where(s => s.CapturedAtUtc < before) + .MinAsync(s => (DateTime?)s.CapturedAtUtc, cancellationToken); + while (floor is not null && floor < before) + { + var ceiling = floor.Value.AddHours(1) < before ? floor.Value.AddHours(1) : before; + total += await _context.KpiSamples + .Where(s => s.CapturedAtUtc < ceiling) + .ExecuteDeleteAsync(cancellationToken); + floor = await _context.KpiSamples.Where(s => s.CapturedAtUtc < before) + .MinAsync(s => (DateTime?)s.CapturedAtUtc, cancellationToken); + } + + return total; } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs index cfdfc85c..d7323e25 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs @@ -1,3 +1,5 @@ +using System.Data.Common; +using Microsoft.EntityFrameworkCore.Diagnostics; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories; @@ -154,4 +156,79 @@ public class KpiHistoryRepositoryTests Assert.Equal(2, remaining.Count); Assert.Equal(new[] { 3d, 4d }, remaining.Select(p => p.Value).ToArray()); } + + [Fact] + public async Task PurgeOlderThanAsync_MultiWindow_IssuesMoreThanOneDelete_AndReturnsTotal() + { + // Rows span several days, each in its own hour-bucket, all older than the + // cutoff. The batched purge must slice them into per-hour DELETE windows + // (arch-review 04, P4) rather than one giant DELETE — so more than one + // DELETE statement is issued, and the return value still equals the total. + var counter = new DeleteCountingInterceptor(); + await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter); + var repo = new KpiHistoryRepository(ctx); + + await repo.RecordSamplesAsync(new[] + { + // Three distinct hour-buckets strictly older than the cutoff — purged. + Sample("NotificationOutbox", "queueDepth", "Global", null, value: 1, capturedAtUtc: Base.AddDays(-5)), + Sample("NotificationOutbox", "queueDepth", "Global", null, value: 2, capturedAtUtc: Base.AddDays(-4)), + Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddDays(-3)), + // Newer than the cutoff — kept, and never scanned into a delete window. + Sample("NotificationOutbox", "queueDepth", "Global", null, value: 4, capturedAtUtc: Base.AddDays(-1)), + Sample("NotificationOutbox", "queueDepth", "Global", null, value: 5, capturedAtUtc: Base), + }); + + // Only count DELETEs the purge itself issues (RecordSamples above emits none). + counter.Reset(); + + var deleted = await repo.PurgeOlderThanAsync(Base.AddDays(-2)); + + // All three old rows gone; the batched total is the sum across windows. + Assert.Equal(3, deleted); + Assert.True(counter.DeleteCount > 1, + $"expected the batched purge to issue more than one DELETE, got {counter.DeleteCount}"); + + var remaining = await repo.GetRawSeriesAsync( + "NotificationOutbox", "queueDepth", "Global", scopeKey: null, + fromUtc: Base.AddDays(-30), toUtc: Base.AddDays(30)); + Assert.Equal(new[] { 4d, 5d }, remaining.Select(p => p.Value).ToArray()); + } + + /// + /// EF command interceptor that counts the DELETE statements actually issued + /// to the store, so a test can prove the purge is sliced into multiple + /// windowed DELETEs rather than one mega-statement. Threads both the sync and + /// async non-query entry points (ExecuteDeleteAsync routes through the + /// async path). + /// + private sealed class DeleteCountingInterceptor : DbCommandInterceptor + { + public int DeleteCount { get; private set; } + + public void Reset() => DeleteCount = 0; + + private void CountIfDelete(DbCommand command) + { + if (command.CommandText.Contains("DELETE", StringComparison.OrdinalIgnoreCase)) + { + DeleteCount++; + } + } + + public override InterceptionResult NonQueryExecuting( + DbCommand command, CommandEventData eventData, InterceptionResult result) + { + CountIfDelete(command); + return base.NonQueryExecuting(command, eventData, result); + } + + public override ValueTask> NonQueryExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult result, + CancellationToken cancellationToken = default) + { + CountIfDelete(command); + return base.NonQueryExecutingAsync(command, eventData, result, cancellationToken); + } + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SqliteTestHelper.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SqliteTestHelper.cs index 8c8647f5..77e362c1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SqliteTestHelper.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SqliteTestHelper.cs @@ -87,6 +87,25 @@ public static class SqliteTestHelper return context; } + /// + /// In-memory SQLite context with EF command interceptors attached — lets a + /// test observe the actual SQL a repository emits (e.g. count DELETE statements + /// issued by a batched purge). + /// + public static ScadaBridgeDbContext CreateInMemoryContext(params IInterceptor[] interceptors) + { + var options = new DbContextOptionsBuilder() + .UseSqlite("DataSource=:memory:") + .ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning)) + .AddInterceptors(interceptors) + .Options; + + var context = new SqliteTestDbContext(options); + context.Database.OpenConnection(); + context.Database.EnsureCreated(); + return context; + } + public static ScadaBridgeDbContext CreateFileContext(string dbPath) { var options = new DbContextOptionsBuilder()