From 27ef527e80f07f14f22f55eb2d91f7f115fcbc30 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Thu, 25 Jun 2026 10:54:32 +0200 Subject: [PATCH] fix(backend): server-bound retention fanout read to per-run cap The retention fanout materialized the entire eligible-session backlog every run (GetSessionsOlderThanAsync returned a full List) but only advanced MaxEnqueuesPerTenantPerRun=100 per tenant. A large backlog was reread and rematerialized each run, so maintenance got slower as the backlog grew while only 100 sessions progressed. Cap the read at the per-run dispatch limit: GetSessionsOlderThanAsync gains a maxResults parameter that caps maxPerPage and stops enumerating once the cap is reached. The fanout fetches MaxEnqueuesPerTenantPerRun+1 (the +1 is a cheap "more remaining" probe) instead of the whole backlog. Read cost is now proportional to the cap, not the backlog; per-run progress is unchanged. The exact eligible count is no longer known without a full scan, so the audit now records EligibleThisRun + a MoreRemaining flag instead of the exact total. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../OffboardingSessionEnumeratorTests.cs | 2 +- .../SessionRetentionFanoutServiceTests.cs | 31 +++++++++++++++---- .../TableMaintenanceRepository.cs | 4 +-- .../Deletion/SessionRetentionFanoutService.cs | 26 ++++++++++++---- .../TableStorageService.Maintenance.cs | 21 ++++++++++--- .../DataAccess/IMaintenanceRepository.cs | 7 ++++- 6 files changed, 70 insertions(+), 21 deletions(-) diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/OffboardingSessionEnumeratorTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/OffboardingSessionEnumeratorTests.cs index 4c93a3d4..70042a4e 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/OffboardingSessionEnumeratorTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/OffboardingSessionEnumeratorTests.cs @@ -109,7 +109,7 @@ public class OffboardingSessionEnumeratorTests public Task> GetAuditLogsPageAsync(string tenantId, DateTime? dateFrom, DateTime? dateTo, int pageSize, string? continuation, bool excludeDeletions = false, AuditLogQueryFilters? filters = null) => throw new NotImplementedException(); public Task> GetAllAuditLogsPageAsync(DateTime? dateFrom, DateTime? dateTo, int pageSize, string? continuation, bool excludeDeletions = false, AuditLogQueryFilters? filters = null) => throw new NotImplementedException(); public Task DeleteAuditLogsOlderThanAsync(DateTime cutoffUtc) => throw new NotImplementedException(); - public Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate) => throw new NotImplementedException(); + public Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue) => throw new NotImplementedException(); public Task> GetSessionsByDateRangeAsync(DateTime startDate, DateTime endDate, string? tenantId = null) => throw new NotImplementedException(); public Task> GetStalledSessionsAsync(string tenantId, DateTime cutoffTime) => throw new NotImplementedException(); public Task> GetAgentSilentSessionsAsync(string tenantId, DateTime silenceCutoff, DateTime hardCutoff) => throw new NotImplementedException(); diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs index eebb6272..6658f613 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs @@ -50,9 +50,9 @@ public class SessionRetentionFanoutServiceTests harness.WithTenant(TenantB, retentionDays: 120, sessions: new[] { Old("b1", 45), Old("b2", 200) }); // The repo only returns sessions older than each cutoff. Wire each tenant's mock to honor that. - harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny())) + harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny(), It.IsAny())) .ReturnsAsync(new List { Summary(TenantA, "a1") }); - harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantB, It.Is(d => d <= DateTime.UtcNow.AddDays(-120)))) + harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantB, It.Is(d => d <= DateTime.UtcNow.AddDays(-120)), It.IsAny())) .ReturnsAsync(new List { Summary(TenantB, "b2") }); var result = await harness.Sut.RunAsync(CancellationToken.None); @@ -84,6 +84,25 @@ public class SessionRetentionFanoutServiceTests Times.Never); } + [Fact] + public async Task RunAsync_reads_session_backlog_server_bounded_to_cap_plus_one() + { + // Codex HIGH fix: the fanout must NOT materialize the whole backlog. It reads at most + // MaxEnqueuesPerTenantPerRun+1 (the +1 is the "more remaining" probe) so a large backlog + // is no longer reread+rematerialized every run while only 100 sessions advance. + var harness = new Harness(); + harness.WithTenant(TenantA, retentionDays: 30, sessions: new[] { Old("s1", 60) }); + + await harness.Sut.RunAsync(CancellationToken.None); + + harness.MaintenanceRepo.Verify( + m => m.GetSessionsOlderThanAsync( + TenantA, + It.IsAny(), + SessionRetentionFanoutService.MaxEnqueuesPerTenantPerRun + 1), + Times.Once); + } + [Fact] public async Task RunAsync_skips_tenant_with_DataRetentionDays_zero() { @@ -94,7 +113,7 @@ public class SessionRetentionFanoutServiceTests Assert.Equal(0, result.SessionsEnqueued); // GetSessionsOlderThanAsync must not have been called for that tenant. - harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny()), Times.Never); + harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny(), It.IsAny()), Times.Never); } // ────────────────────────────────────────────────────────────────────────── PR6 follow-up F2 ─ @@ -228,7 +247,7 @@ public class SessionRetentionFanoutServiceTests harness.Enqueuer.Verify(e => e.EnqueueAsync(TenantB, It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); - harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantB, It.IsAny()), Times.Never); + harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantB, It.IsAny(), It.IsAny()), Times.Never); } // ─────────────────────────────────────────────────────────────────────────────────────────────── @@ -322,7 +341,7 @@ public class SessionRetentionFanoutServiceTests _tenantIds.Add(tenantId); TenantConfig.Setup(t => t.GetConfigurationAsync(tenantId)) .ReturnsAsync(new TenantConfiguration { TenantId = tenantId, DataRetentionDays = retentionDays }); - MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny())) + MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny(), It.IsAny())) .ReturnsAsync(new List(WithTenantId(tenantId, sessions))); } @@ -332,7 +351,7 @@ public class SessionRetentionFanoutServiceTests TenantConfig.Setup(t => t.GetConfigurationAsync(tenantId)) .ReturnsAsync(new TenantConfiguration { TenantId = tenantId, DataRetentionDays = retentionDays }); foreach (var s in sessions) s.TenantId = tenantId; - MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny())).ReturnsAsync(sessions); + MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny(), It.IsAny())).ReturnsAsync(sessions); } private static IEnumerable WithTenantId(string tenantId, IEnumerable sessions) diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableMaintenanceRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableMaintenanceRepository.cs index f0dd4a8c..1ed832ef 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableMaintenanceRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableMaintenanceRepository.cs @@ -51,8 +51,8 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage public Task DeleteAuditLogsOlderThanAsync(DateTime cutoffUtc) => _storage.DeleteAuditLogsOlderThanAsync(cutoffUtc); - public Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate) - => _storage.GetSessionsOlderThanAsync(tenantId, cutoffDate); + public Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue) + => _storage.GetSessionsOlderThanAsync(tenantId, cutoffDate, maxResults); public Task> GetSessionsByDateRangeAsync(DateTime startDate, DateTime endDate, string? tenantId = null) => _storage.GetSessionsByDateRangeAsync(startDate, endDate, tenantId); diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs index 39a9c26f..6cb86c71 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs @@ -145,7 +145,15 @@ namespace AutopilotMonitor.Functions.Services.Deletion } var cutoffUtc = DateTime.UtcNow.AddDays(-retentionDays); - var oldSessions = await _maintenanceRepo.GetSessionsOlderThanAsync(tenantId, cutoffUtc).ConfigureAwait(false); + + // Server-bounded read: fetch one more than the per-run dispatch cap. The loop below + // only ever advances MaxEnqueuesPerTenantPerRun sessions, so loading the whole backlog + // every run is wasted I/O that grows with the backlog. The "+1" is a cheap probe that + // lets us tell "exactly cap eligible, done" from "cap+ eligible, more deferred" without + // scanning the rest — it is observability only and is never dispatched. + const int fetchLimit = MaxEnqueuesPerTenantPerRun + 1; + var oldSessions = await _maintenanceRepo.GetSessionsOlderThanAsync(tenantId, cutoffUtc, fetchLimit).ConfigureAwait(false); + bool moreRemaining = oldSessions.Count > MaxEnqueuesPerTenantPerRun; if (oldSessions.Count == 0) { @@ -163,9 +171,12 @@ namespace AutopilotMonitor.Functions.Services.Deletion if (processed >= MaxEnqueuesPerTenantPerRun) { + // We only fetched cap+1, so we know "more remain" but not the exact backlog + // size (that would require the full scan this change removes). Report it as a + // floor — the next run drains the following batch. _logger.LogInformation( - "Tenant {TenantId}: hit rate limit ({Limit}/run) — deferring remaining {Remaining} sessions to next run", - tenantId, MaxEnqueuesPerTenantPerRun, oldSessions.Count - processed); + "Tenant {TenantId}: hit rate limit ({Limit}/run) — at least one more batch deferred to next run", + tenantId, MaxEnqueuesPerTenantPerRun); result.RateLimitedTenants++; break; } @@ -223,15 +234,18 @@ namespace AutopilotMonitor.Functions.Services.Deletion { { "RetentionDays", retentionDays.ToString() }, { "CutoffUtc", cutoffUtc.ToString("o") }, - { "Eligible", oldSessions.Count.ToString() }, + // Read is capped at MaxEnqueuesPerTenantPerRun+1, so this is a floor, not the + // exact backlog size. MoreRemaining=true ⇒ at least one further batch is pending. + { "EligibleThisRun", Math.Min(oldSessions.Count, MaxEnqueuesPerTenantPerRun).ToString() }, + { "MoreRemaining", moreRemaining.ToString() }, { "Enqueued", enqueued.ToString() }, { "Skipped", skipped.ToString() }, { "AbortedByKillSwitch", result.AbortedByKillSwitch.ToString() }, }).ConfigureAwait(false); _logger.LogInformation( - "Tenant {TenantId}: retention fanout — cutoff={Cutoff:o} eligible={Eligible} enqueued={Enqueued} skipped={Skipped}", - tenantId, cutoffUtc, oldSessions.Count, enqueued, skipped); + "Tenant {TenantId}: retention fanout — cutoff={Cutoff:o} eligibleThisRun={Eligible} moreRemaining={More} enqueued={Enqueued} skipped={Skipped}", + tenantId, cutoffUtc, Math.Min(oldSessions.Count, MaxEnqueuesPerTenantPerRun), moreRemaining, enqueued, skipped); } /// diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs index 40bea5cc..2717c9d6 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs @@ -439,27 +439,38 @@ namespace AutopilotMonitor.Functions.Services // ===== DATA RETENTION METHODS ===== /// - /// Gets sessions older than a specific date for a tenant + /// Gets sessions older than a specific date for a tenant, capped at . + /// The retention fanout only advances a fixed number of sessions per run, so the read is + /// server-bounded to that cap instead of materializing the whole backlog every run. /// - public async Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate) + public async Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue) { SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + if (maxResults <= 0) return new List(); try { var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions); - // Query sessions for this tenant older than cutoff date + // Query sessions for this tenant older than cutoff date. var filter = $"PartitionKey eq '{tenantId}' and StartedAt lt datetime'{cutoffDate:yyyy-MM-ddTHH:mm:ss}Z'"; - var query = tableClient.QueryAsync(filter: filter); + + // Bound the server read: cap page size to maxResults and stop enumerating once we + // have that many. Azure Tables has no server-side OrderBy, so this returns the first + // maxResults rows in RowKey order — the same ordering as the previous full scan, it + // just stops reading once the cap is reached. Unbounded callers pass int.MaxValue. + int? maxPerPage = maxResults == int.MaxValue ? (int?)null : maxResults; + var query = tableClient.QueryAsync(filter: filter, maxPerPage: maxPerPage); var sessions = new List(); await foreach (var entity in query) { sessions.Add(MapToSessionSummary(entity)); + if (sessions.Count >= maxResults) + break; } - _logger.LogInformation($"Found {sessions.Count} sessions older than {cutoffDate:yyyy-MM-dd} for tenant {tenantId}"); + _logger.LogInformation($"Found {sessions.Count} sessions older than {cutoffDate:yyyy-MM-dd} for tenant {tenantId} (cap={maxResults})"); return sessions; } catch (Exception ex) diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/IMaintenanceRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/IMaintenanceRepository.cs index 58b51326..b03bdd33 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/IMaintenanceRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/IMaintenanceRepository.cs @@ -57,7 +57,12 @@ namespace AutopilotMonitor.Shared.DataAccess Task DeleteAuditLogsOlderThanAsync(DateTime cutoffUtc); // --- Data Retention Queries --- - Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate); + /// + /// Sessions older than for a tenant, capped at + /// (server-bounded read). The retention fanout passes its + /// per-run dispatch cap so it never materializes a backlog it cannot process in one run. + /// + Task> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue); Task> GetSessionsByDateRangeAsync(DateTime startDate, DateTime endDate, string? tenantId = null); Task> GetStalledSessionsAsync(string tenantId, DateTime cutoffTime); Task> GetAgentSilentSessionsAsync(string tenantId, DateTime silenceCutoff, DateTime hardCutoff);