From c287cb3eaeca554b53386bc41aedd46247ce4f9a Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Fri, 9 Oct 2026 21:33:29 +0200 Subject: [PATCH] Backend: retention and device-history reads fail closed (D-338) - The retention fanout reads the tenant configuration strictly. A storage failure used to yield the synthesized default (90 days), which deletes sessions and prunes rows a Pro-365 or never-delete tenant keeps. Now the tenant is skipped and counted: tenantsFailed on SessionDeletionMaintenanceCompleted, which turns Warning. A row that vanished since the tenant listing is skipped too, instead of being recreated with defaults. - GetDeviceHistoryAsync throws on a read failure instead of returning "no history". Both writers merge into the chain they read and replace the row, so the old behaviour dropped every older ref. The sweep skips such devices and leaves the FTR aggregates untouched for that tick. Co-Authored-By: Claude Opus 5.5 --- .../DeviceJourneyAndFtrTests.cs | 37 ++++++++ .../SessionRetentionFanoutServiceTests.cs | 44 ++++++++- .../SessionDeletionMaintenanceFunction.cs | 1 + .../Deletion/SessionRetentionFanoutService.cs | 22 +++-- .../MaintenanceService.DeviceJourneys.cs | 92 ++++++++++++++----- .../Services/OpsEventService.cs | 14 ++- .../TableStorageService.DeviceHistory.cs | 21 ++--- .../DataAccess/IMetricsRepository.cs | 2 +- 8 files changed, 183 insertions(+), 50 deletions(-) diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/DeviceJourneyAndFtrTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/DeviceJourneyAndFtrTests.cs index 6461b673..71d9c7c7 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/DeviceJourneyAndFtrTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/DeviceJourneyAndFtrTests.cs @@ -1,7 +1,10 @@ +using Azure; using AutopilotMonitor.Functions.Functions.Metrics; using AutopilotMonitor.Functions.Helpers; using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models; +using Moq; using Xunit; namespace AutopilotMonitor.Functions.Tests; @@ -795,4 +798,38 @@ public class DeviceJourneyAndFtrTests Assert.DoesNotContain(selected, c => c.History.SerialKey == "dev-00"); Assert.DoesNotContain(selected, c => c.History.SerialKey == "dev-01"); } + + // ── sweep merge: a failed history read is never "no history" ──────────── + + [Fact] + public async Task SweepMerge_HistoryReadFailure_SkipsTheDevice_AndKeepsOlderRefsOfTheOthers() + { + // A failed point read used to come back as null ("no history"): the sweep merged the + // window into an empty chain and replaced the row, dropping every older ref. + var repo = new Mock(); + repo.Setup(r => r.GetDeviceHistoryAsync(TenantA, "broken")) + .ThrowsAsync(new RequestFailedException(503, "Server Busy")); + repo.Setup(r => r.GetDeviceHistoryAsync(TenantA, "healthy")) + .ReturnsAsync(History(TenantA, Ref("old-1", SessionStatus.Failed, T0.AddDays(-200)))); + var upserted = new List(); + repo.Setup(r => r.UpsertDeviceHistoryAsync(It.IsAny())) + .Callback(upserted.Add) + .ReturnsAsync(true); + + var byDevice = new Dictionary<(string TenantId, string SerialKey), List> + { + [(TenantA, "broken")] = new() { Session("w-1") }, + [(TenantA, "healthy")] = new() { Session("w-2") }, + }; + + var (merged, readFailures, firstError) = await MaintenanceService.MergeWindowIntoDeviceHistoriesAsync( + repo.Object, byDevice, new Dictionary>()); + + Assert.Equal(1, readFailures); + Assert.IsType(firstError); + var written = Assert.Single(upserted); + Assert.Equal("healthy", written.SerialKey); + Assert.Equal(new[] { "old-1", "w-2" }, written.Chain.Select(r => r.SessionId).ToArray()); + Assert.Same(written, Assert.Single(merged)); + } } diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs index dc4830df..aac599b8 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs @@ -2,6 +2,7 @@ using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; +using Azure; using AutopilotMonitor.Functions.Services; using AutopilotMonitor.Functions.Services.Deletion; using AutopilotMonitor.Shared.DataAccess; @@ -66,6 +67,45 @@ public class SessionRetentionFanoutServiceTests harness.Enqueuer.Verify(e => e.EnqueueAsync(TenantB, "b1", It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); } + [Fact] + public async Task RunAsync_skips_tenant_whose_configuration_cannot_be_read() + { + // A config read failure used to yield the synthesized default (90 days) and delete what a + // Pro-365 or never-delete tenant chose to keep. Now the tenant is skipped and counted — + // nothing enqueued, nothing pruned — and the next tenant still runs. + var harness = new Harness(); + harness.WithTenant(TenantA, retentionDays: 365, sessions: new[] { Old("a1", 120) }, planTier: "pro"); + harness.WithTenant(TenantB, retentionDays: 30, sessions: new[] { Old("b1", 45) }); + harness.TenantConfig.Setup(t => t.GetConfigurationIfExistsAsync(TenantA)) + .ThrowsAsync(new RequestFailedException(503, "Server Busy")); + + var result = await harness.RunAsync(); + + Assert.Equal(1, result.TenantsFailed); + Assert.Equal(1, result.TenantsProcessed); + harness.Enqueuer.Verify(e => e.EnqueueAsync(TenantA, It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + harness.MaintenanceRepo.Verify(m => m.DeleteTenantAuditLogsOlderThanAsync(TenantA, It.IsAny()), Times.Never); + harness.MetricsRepo.Verify(m => m.DeleteTenantRuleStatsOlderThanAsync(TenantA, It.IsAny()), Times.Never); + harness.Enqueuer.Verify(e => e.EnqueueAsync(TenantB, "b1", "retention_cutoff", It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); + } + + [Fact] + public async Task RunAsync_skips_tenant_whose_configuration_row_vanished() + { + // The tenant list reads the configuration rows; a row gone by the time of the point read + // (offboarding race) leaves nothing to decide on — no default retention, no recreated row. + var harness = new Harness(); + harness.WithTenant(TenantA, retentionDays: 30, sessions: new[] { Old("a1", 45) }); + harness.TenantConfig.Setup(t => t.GetConfigurationIfExistsAsync(TenantA)) + .ReturnsAsync((TenantConfiguration?)null); + + var result = await harness.RunAsync(); + + Assert.Equal(0, result.TenantsFailed); + harness.Enqueuer.Verify(e => e.EnqueueAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + harness.TenantConfig.Verify(t => t.GetConfigurationAsync(It.IsAny()), Times.Never); + } + [Fact] public async Task RunAsync_enforces_rate_limit_per_tenant_per_run() { @@ -564,7 +604,7 @@ public class SessionRetentionFanoutServiceTests DateTime? proDowngradedUtc = null) { _tenantIds.Add(tenantId); - TenantConfig.Setup(t => t.GetConfigurationAsync(tenantId)) + TenantConfig.Setup(t => t.GetConfigurationIfExistsAsync(tenantId)) .ReturnsAsync(new TenantConfiguration { TenantId = tenantId, @@ -579,7 +619,7 @@ public class SessionRetentionFanoutServiceTests public void WithTenantOverride(string tenantId, int retentionDays, List sessions) { _tenantIds.Add(tenantId); - TenantConfig.Setup(t => t.GetConfigurationAsync(tenantId)) + TenantConfig.Setup(t => t.GetConfigurationIfExistsAsync(tenantId)) .ReturnsAsync(new TenantConfiguration { TenantId = tenantId, DataRetentionDays = retentionDays }); foreach (var s in sessions) s.TenantId = tenantId; MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny(), It.IsAny(), It.IsAny())).ReturnsAsync(sessions); diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Maintenance/SessionDeletionMaintenanceFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Maintenance/SessionDeletionMaintenanceFunction.cs index 7a8b18ae..40c4c300 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Maintenance/SessionDeletionMaintenanceFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Maintenance/SessionDeletionMaintenanceFunction.cs @@ -232,6 +232,7 @@ namespace AutopilotMonitor.Functions.Functions.Maintenance await _opsEvents.RecordSessionDeletionMaintenanceCompletedAsync( killSwitchActive: killSwitchActive, tenantsProcessed: fanoutResult.TenantsProcessed, + tenantsFailed: fanoutResult.TenantsFailed, sessionsEnqueued: fanoutResult.SessionsEnqueued, sessionsSkipped: fanoutResult.SessionsSkipped, rateLimitedTenants: fanoutResult.RateLimitedTenants, diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs index 09f33afc..65bf87ba 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs @@ -89,6 +89,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion public sealed class FanoutResult { public int TenantsProcessed { get; set; } + public int TenantsFailed { get; set; } // tenant skipped on an exception (e.g. unreadable config) — nothing deleted public int SessionsEnqueued { get; set; } // enqueued for cascade public int SessionsSkipped { get; set; } // already locked / poisoned / kill-switch / etc. public int RateLimitedTenants { get; set; } // tenants that hit MaxEnqueuesPerTenantPerRun @@ -150,6 +151,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion } catch (Exception ex) { + result.TenantsFailed++; _logger.LogError(ex, "SessionRetentionFanout failed for tenant {TenantId} — continuing with next tenant", tenantId); } @@ -168,23 +170,31 @@ namespace AutopilotMonitor.Functions.Services.Deletion private async Task RunForTenantAsync(string tenantId, DateTime deadlineUtc, FanoutResult result, CancellationToken cancellationToken) { - var config = await _tenantConfig.GetConfigurationAsync(tenantId).ConfigureAwait(false); + // Strict read: a storage failure throws — RunAsync's per-tenant catch skips and counts + // the tenant — instead of yielding the synthesized default, whose 90 days would delete + // what a Pro-365 or never-delete tenant chose to keep. No row means it vanished since + // the tenant listing (which reads these rows): nothing to decide on, skip. + var config = await _tenantConfig.GetConfigurationIfExistsAsync(tenantId).ConfigureAwait(false); + if (config == null) + { + _logger.LogWarning("SessionRetentionFanout: tenant {TenantId} has no configuration row any more — skipped", tenantId); + return; + } + // Edition retention cap (fail-closed backstop): the stored value is clamped to the // effective edition's cap at read time (Community 90 / Pro 365). A tenant whose // trial expired, or whose stored value predates the cap, is enforced here even though // the stored DataRetentionDays is left untouched. days <= 0 stays the GA-only // "infinite" escape hatch (skipped below, never clamped). var nowUtc = _utcNow(); - var storedDays = config?.DataRetentionDays ?? 90; - var retentionDays = config == null - ? 90 - : TenantEntitlementService.GetEffectiveRetentionDays(config, nowUtc); + var storedDays = config.DataRetentionDays; + var retentionDays = TenantEntitlementService.GetEffectiveRetentionDays(config, nowUtc); if (retentionDays > 0 && retentionDays < storedDays) { _logger.LogWarning( "Tenant {TenantId}: DataRetentionDays={Stored} exceeds the {Edition}-edition cap — enforcing {Effective} days", tenantId, storedDays, - TenantEntitlementService.ResolveEdition(config!, nowUtc), retentionDays); + TenantEntitlementService.ResolveEdition(config, nowUtc), retentionDays); } if (retentionDays <= 0) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/MaintenanceService.DeviceJourneys.cs b/src/Backend/AutopilotMonitor.Functions/Services/MaintenanceService.DeviceJourneys.cs index 0f07791f..bace9849 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/MaintenanceService.DeviceJourneys.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/MaintenanceService.DeviceJourneys.cs @@ -4,6 +4,7 @@ using System.Diagnostics; using System.Linq; using System.Threading.Tasks; using AutopilotMonitor.Functions.Helpers; +using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models; using Microsoft.Extensions.Logging; @@ -113,33 +114,21 @@ namespace AutopilotMonitor.Functions.Services list.Add(session); } - var mergedHistories = new List(byDevice.Count); - foreach (var pair in byDevice) - { - var (tenantId, serialKey) = pair.Key; - var existing = await _metricsRepo.GetDeviceHistoryAsync(tenantId, serialKey); - var refs = pair.Value - .Select(DeviceJourneyCalculator.BuildSessionRef) - .Where(r => r != null) - .Select(r => r!); - - var chain = DeviceJourneyCalculator.MergeChain(existing?.Chain, refs); - if (deletedByTenant.TryGetValue(tenantId, out var deletedIds)) - chain = DeviceJourneyCalculator.RemoveSessionRefs(chain, deletedIds); - if (chain.Count == 0) - continue; - - // Display fields follow the chain's newest entry (BuildDeviceHistoryRow rule); - // pick the window session that IS that entry, falling back to any of the - // device's window sessions (then the existing row's values win inside). - var newestId = chain[chain.Count - 1].SessionId; - var displaySource = pair.Value.FirstOrDefault(s => s.SessionId == newestId) ?? pair.Value[0]; - var history = TableStorageService.BuildDeviceHistoryRow(tenantId, serialKey, chain, existing, displaySource); - await _metricsRepo.UpsertDeviceHistoryAsync(history); - mergedHistories.Add(history); - } + var (mergedHistories, historyReadFailures, firstReadError) = + await MergeWindowIntoDeviceHistoriesAsync(_metricsRepo, byDevice, deletedByTenant); // ---- Phase 3: daily FTR aggregates from the merged chains. ---- + // Only from a complete Phase 2: aggregates built without the skipped devices would + // overwrite complete rows with smaller counts, and the stale reconcile below would + // delete buckets those devices alone fill. + if (historyReadFailures > 0) + { + _logger.LogWarning(firstReadError, + "Device-journey sweep: {Failures} of {Devices} device-history reads failed — those devices wait for the next tick and the FTR aggregates are left unchanged", + historyReadFailures, byDevice.Count); + return; + } + var aggregates = BuildDeviceJourneyAggregates( mergedHistories, excludedSessions, windowStart, windowEnd, DateTime.UtcNow); var saved = 0; @@ -180,6 +169,59 @@ namespace AutopilotMonitor.Functions.Services } } + /// + /// Phase 2 of : merges each device's window refs into + /// its stored chain and replaces the row. internal static so the read-failure rule is pinned + /// by tests: a device whose history read FAILS is skipped and counted, never treated as "no + /// history" — merging the window into an empty chain and replacing the row would drop every + /// older ref. + /// + internal static async Task<(List Merged, int ReadFailures, Exception? FirstReadError)> MergeWindowIntoDeviceHistoriesAsync( + IMetricsRepository metricsRepo, + IReadOnlyDictionary<(string TenantId, string SerialKey), List> byDevice, + IReadOnlyDictionary> deletedByTenant) + { + var merged = new List(byDevice.Count); + var readFailures = 0; + Exception? firstReadError = null; + foreach (var pair in byDevice) + { + var (tenantId, serialKey) = pair.Key; + DeviceHistory? existing; + try + { + existing = await metricsRepo.GetDeviceHistoryAsync(tenantId, serialKey); + } + catch (Exception ex) + { + readFailures++; + firstReadError ??= ex; + continue; + } + + var refs = pair.Value + .Select(DeviceJourneyCalculator.BuildSessionRef) + .Where(r => r != null) + .Select(r => r!); + + var chain = DeviceJourneyCalculator.MergeChain(existing?.Chain, refs); + if (deletedByTenant.TryGetValue(tenantId, out var deletedIds)) + chain = DeviceJourneyCalculator.RemoveSessionRefs(chain, deletedIds); + if (chain.Count == 0) + continue; + + // Display fields follow the chain's newest entry (BuildDeviceHistoryRow rule); + // pick the window session that IS that entry, falling back to any of the + // device's window sessions (then the existing row's values win inside). + var newestId = chain[chain.Count - 1].SessionId; + var displaySource = pair.Value.FirstOrDefault(s => s.SessionId == newestId) ?? pair.Value[0]; + var history = TableStorageService.BuildDeviceHistoryRow(tenantId, serialKey, chain, existing, displaySource); + await metricsRepo.UpsertDeviceHistoryAsync(history); + merged.Add(history); + } + return (merged, readFailures, firstReadError); + } + /// /// Pure aggregation core behind — internal static /// so the FTR contract is pinned by unit tests. A journey counts on the StartedAt date diff --git a/src/Backend/AutopilotMonitor.Functions/Services/OpsEventService.cs b/src/Backend/AutopilotMonitor.Functions/Services/OpsEventService.cs index bfd41e54..6583de04 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/OpsEventService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/OpsEventService.cs @@ -301,15 +301,21 @@ namespace AutopilotMonitor.Functions.Services /// PR6 follow-up F3: replaces the prior LogAuditEntryAsync(null!, ...) call, which /// silently failed because the AuditLogs schema requires a non-null PartitionKey (tenantId). /// + /// + /// Run summary of the retention maintenance. Warning instead of Info when + /// > 0: those tenants were skipped (nothing deleted), so + /// their retention did not run this time. + /// public Task RecordSessionDeletionMaintenanceCompletedAsync( - bool killSwitchActive, int tenantsProcessed, int sessionsEnqueued, + bool killSwitchActive, int tenantsProcessed, int tenantsFailed, int sessionsEnqueued, int sessionsSkipped, int rateLimitedTenants, int blobsTtlGced, int preparingRowsCleared, int strandedQueuedDetected, int durationMs, bool abortedByKillSwitch, bool abortedByBudget, int tenantRowsPruned) - => WriteAsync(OpsEventCategory.Maintenance, OpsEventTypes.SessionDeletionMaintenanceCompleted, OpsEventSeverity.Info, - $"SessionDeletionMaintenance completed in {durationMs}ms — tenants={tenantsProcessed} enqueued={sessionsEnqueued} skipped={sessionsSkipped} tenantRowsPruned={tenantRowsPruned} blobsTtlGced={blobsTtlGced} preparingCleared={preparingRowsCleared} stranded={strandedQueuedDetected} killSwitch={killSwitchActive} abortedByBudget={abortedByBudget}", + => WriteAsync(OpsEventCategory.Maintenance, OpsEventTypes.SessionDeletionMaintenanceCompleted, + tenantsFailed > 0 ? OpsEventSeverity.Warning : OpsEventSeverity.Info, + $"SessionDeletionMaintenance completed in {durationMs}ms — tenants={tenantsProcessed} failed={tenantsFailed} enqueued={sessionsEnqueued} skipped={sessionsSkipped} tenantRowsPruned={tenantRowsPruned} blobsTtlGced={blobsTtlGced} preparingCleared={preparingRowsCleared} stranded={strandedQueuedDetected} killSwitch={killSwitchActive} abortedByBudget={abortedByBudget}", null, "System.Maintenance", new { - killSwitchActive, tenantsProcessed, sessionsEnqueued, + killSwitchActive, tenantsProcessed, tenantsFailed, sessionsEnqueued, sessionsSkipped, rateLimitedTenants, blobsTtlGced, preparingRowsCleared, strandedQueuedDetected, durationMs, abortedByKillSwitch, abortedByBudget, tenantRowsPruned, diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.DeviceHistory.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.DeviceHistory.cs index 3242a567..05a384ff 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.DeviceHistory.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.DeviceHistory.cs @@ -95,22 +95,19 @@ namespace AutopilotMonitor.Functions.Services }; } - /// Point-reads one device's history row; null when the device has none. Takes the NORMALIZED serial (RK encoding is internal). + /// + /// Point-reads one device's history row; null when the device has none. Takes the NORMALIZED + /// serial (RK encoding is internal). A read failure PROPAGATES: both writers merge into the + /// returned chain and replace the row, so a failure taken for "no history" would drop every + /// older ref of the device. + /// public async Task GetDeviceHistoryAsync(string tenantId, string serialKey) { SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); if (string.IsNullOrEmpty(serialKey)) return null; - try - { - var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.DeviceHistories); - var result = await tableClient.GetEntityIfExistsAsync(tenantId, SerialRowKey(serialKey)); - return result.HasValue ? MapToDeviceHistory(result.Value!) : null; - } - catch (Exception ex) - { - _logger.LogError(ex, "Failed to get device history for tenant {TenantId}", tenantId); - return null; - } + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.DeviceHistories); + var result = await tableClient.GetEntityIfExistsAsync(tenantId, SerialRowKey(serialKey)); + return result.HasValue ? MapToDeviceHistory(result.Value!) : null; } /// All device-history rows of one tenant (the sweep's tombstone-cleanup scan; bounded by the tenant's device count). diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/IMetricsRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/IMetricsRepository.cs index 97149029..9ac8ceaa 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/IMetricsRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/IMetricsRepository.cs @@ -175,7 +175,7 @@ namespace AutopilotMonitor.Shared.DataAccess Task DeleteTenantTimeAttributionAggregatesOlderThanAsync(string tenantId, DateTime cutoffDate); // --- F2 Device History / First-Time-Right (insights spec §F2, PR4) --- - /// Point-reads one device's history row by NORMALIZED serial; null when absent. + /// Point-reads one device's history row by NORMALIZED serial; null when absent, throws when the read fails. Task GetDeviceHistoryAsync(string tenantId, string serialKey); /// All device-history rows of one tenant (sweep's tombstone-cleanup scan). Task> GetDeviceHistoriesByTenantAsync(string tenantId);