From fd7bf0e7aceac3175fa2eda008d7389fc5432c14 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Tue, 21 Apr 2026 12:29:12 +0200 Subject: [PATCH] =?UTF-8?q?feat(backend):=20M5.d.4=20index-reconcile=20tim?= =?UTF-8?q?er=20+=20release-gate=20=E2=80=94=20M5=20complete?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the M5.d dual-write loop with a 2h safety-net timer and the per-index-table PK/RK consistency release-gate tests required by Plan §M5-Pflicht. Components: - ISignalRepository.QueryByTimestampAtOrAfterAsync + IDecisionTransitionRepository.QueryByTimestampAtOrAfterAsync: new cross-partition scans filtered on the Azure Tables system-managed `Timestamp` property. Bounded to maxResults=50000 with a warn-log on cap-reached so silent truncation can't masquerade as success. - IndexReconcileTimer: TimerTrigger on "0 0 */2 * * *" (same cadence as MaintenanceFunction). Scans the last 4h (window deliberately 2× the cadence so every primary row gets re-checked at least twice before falling out of range), builds envelopes via the existing IndexReconcileEnvelopeFactory, and hands them to the producer — reusing the M5.d.2 flag-gate + queue-failure-swallow semantics instead of duplicating them. - Testable core: RunReconcileAsync(cutoffUtc, CT) split out from the TimerTrigger entry so unit tests can drive with deterministic cutoffs without an Azure Functions host. - AdminConfigurationService.GetConfigurationAsync is now virtual — enables Moq-based mocking in timer flag-gate tests. Low-risk prod change (no existing caller relies on final). Release gate (Plan §M5-Pflicht line 1395): IndexDualWriteConsistencyTests verifies per-index-table that the captured index row's back-ref fields (TenantId, SessionId, StepIndex / SessionSignalOrdinal) plug directly into the primary repos' key builders to reconstruct the primary (PartitionKey, RowKey). The resubmit-idempotency test simulates a timer-triggered replay of the same primaries and asserts all duplicated rows collapse onto identical Index (PK, RK) pairs — which is exactly what the real TableIndexRepository's UpsertReplace does at the storage layer. Left deferred (not blocking M5 close): - SessionsByStage stale-row pruning: stage-X→Y transitions still leave an orphan row in {tenantId}_X. Either query-side dedupe (per SessionId, max-Ticks RK) or a follow-up pruner handles it. - Actual EnableIndexDualWrite=true flip: stays default-false until post-deploy verification runs. Producer + timer both no-op in the meantime, so the change is zero-risk on merge. - Cross-partition-scan perf: fine for current primary-table sizes; revisit if prod metrics after pilot show degradation. Tests (+10 in Functions): - 4 IndexReconcileTimerTests: flag-off short-circuits queries, flag-on with primary rows enqueues all envelopes, empty-window skips enqueue, ReconcileWindow=4h invariant. - 6 IndexDualWriteConsistencyTests: 5 per-index-table PK/RK back-ref (SignalsByKind + all 4 transition-derived indices) + 1 resubmit idempotency ensuring all duplicated rows collapse onto the same index keys. M5 status: M5.a + M5.b (+ 3 sub-commits) + M5.c (+ 3 sub-commits) + M5.d (+ 4 sub-commits) all complete. M5.d loop: Primary commit (M5.a) → Envelope enqueue (M5.d.2, flag-gated) → Queue → Handler (M5.d.3) → Index-Tables (M5.d.1) + 2h safety-net Timer (M5.d.4) reconciles missed envelopes. Full Functions suite: 692 green (prev 682 + 10 new). Full solution build green. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../IndexDualWriteConsistencyTests.cs | 285 ++++++++++++++++++ .../IndexReconcileTimerTests.cs | 156 ++++++++++ .../TableDecisionTransitionRepository.cs | 28 ++ .../TableStorage/TableSignalRepository.cs | 30 ++ .../Functions/Admin/IndexReconcileTimer.cs | 100 ++++++ .../Services/AdminConfigurationService.cs | 8 +- .../IDecisionTransitionRepository.cs | 15 + .../DataAccess/ISignalRepository.cs | 15 + 8 files changed, 635 insertions(+), 2 deletions(-) create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/IndexDualWriteConsistencyTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/IndexReconcileTimerTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/Functions/Admin/IndexReconcileTimer.cs diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IndexDualWriteConsistencyTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IndexDualWriteConsistencyTests.cs new file mode 100644 index 00000000..36be3e64 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IndexDualWriteConsistencyTests.cs @@ -0,0 +1,285 @@ +using AutopilotMonitor.Functions.DataAccess.TableStorage; +using AutopilotMonitor.Functions.Services.Indexing; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using Microsoft.Extensions.Logging.Abstractions; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// M5 release-gate (Plan §M5-Pflicht line 1395): per-index-table PK/RK consistency +/// between primary rows and their derived index rows. For each index table, walk a +/// canonical primary row through the Factory + Handler pipeline and verify that the +/// captured index record carries enough back-ref information to reconstruct the +/// primary's (PartitionKey, RowKey) via / +/// key builders. +/// +/// Also asserts pipeline idempotency: running the same primary row through twice +/// produces identical upsert keys — the consumer's UpsertReplace batches stay +/// deduplicating even under timer-triggered resubmits. +/// +/// +public class IndexDualWriteConsistencyTests +{ + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + + private static readonly DateTime Occurred = + new(2026, 4, 21, 14, 7, 33, DateTimeKind.Utc); + + private static (IndexReconcileHandler handler, FakeIndexTableRepository repo) Sut() + { + var repo = new FakeIndexTableRepository(); + var handler = new IndexReconcileHandler(repo, NullLogger.Instance); + return (handler, repo); + } + + // ============================================================ SignalsByKind ↔ Signals + + [Fact] + public async Task SignalsByKind_row_carries_back_ref_to_primary_Signals_row() + { + var primary = new SignalRecord + { + TenantId = TenantId, + SessionId = SessionId, + SessionSignalOrdinal = 42, + Kind = "EspTerminalFailure", + OccurredAtUtc = Occurred, + SourceOrigin = "EspAndHelloTrackerAdapter", + }; + + var (handler, repo) = Sut(); + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromSignal(primary)); + + var index = Assert.Single(repo.SignalsByKind); + + // Back-ref fields on the index row must reconstruct the primary PK/RK. + var reconstructedPk = TableSignalRepository.BuildPartitionKey(index.TenantId, index.SessionId); + var reconstructedRk = TableSignalRepository.BuildRowKey(index.SessionSignalOrdinal); + + Assert.Equal( + TableSignalRepository.BuildPartitionKey(primary.TenantId, primary.SessionId), + reconstructedPk); + Assert.Equal( + TableSignalRepository.BuildRowKey(primary.SessionSignalOrdinal), + reconstructedRk); + } + + // ============================================================ SessionsByTerminal ↔ DecisionTransitions + + [Fact] + public async Task SessionsByTerminal_row_carries_back_ref_to_primary_DecisionTransitions_row() + { + var primary = TerminalTransition(); + + var (handler, repo) = Sut(); + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary)); + + var index = Assert.Single(repo.SessionsByTerminal); + + var reconstructedPk = TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId); + var reconstructedRk = TableDecisionTransitionRepository.BuildRowKey(index.StepIndex); + + Assert.Equal( + TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId), + reconstructedPk); + Assert.Equal( + TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex), + reconstructedRk); + } + + [Fact] + public async Task SessionsByStage_row_carries_back_ref_to_primary_DecisionTransitions_row() + { + var primary = new DecisionTransitionRecord + { + TenantId = TenantId, + SessionId = SessionId, + StepIndex = 3, + FromStage = "AwaitingEsp", + ToStage = "EspInProgress", + Taken = true, + IsTerminal = false, + OccurredAtUtc = Occurred, + }; + + var (handler, repo) = Sut(); + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary)); + + var index = Assert.Single(repo.SessionsByStage); + + var reconstructedPk = TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId); + var reconstructedRk = TableDecisionTransitionRepository.BuildRowKey(index.StepIndex); + + Assert.Equal( + TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId), + reconstructedPk); + Assert.Equal( + TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex), + reconstructedRk); + } + + [Fact] + public async Task DeadEndsByReason_row_carries_back_ref_to_primary_DecisionTransitions_row() + { + var primary = new DecisionTransitionRecord + { + TenantId = TenantId, + SessionId = SessionId, + StepIndex = 5, + FromStage = "EspInProgress", + ToStage = "EspInProgress", + Taken = false, + IsTerminal = false, + DeadEndReason = "hybrid_reboot_gate_blocking", + OccurredAtUtc = Occurred, + }; + + var (handler, repo) = Sut(); + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary)); + + var index = Assert.Single(repo.DeadEndsByReason); + + Assert.Equal( + TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId), + TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId)); + Assert.Equal( + TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex), + TableDecisionTransitionRepository.BuildRowKey(index.StepIndex)); + } + + [Fact] + public async Task ClassifierVerdictsByIdLevel_row_carries_back_ref_to_primary_DecisionTransitions_row() + { + var primary = new DecisionTransitionRecord + { + TenantId = TenantId, + SessionId = SessionId, + StepIndex = 7, + FromStage = "EspInProgress", + ToStage = "WhiteGloveSealed", + Taken = true, + IsTerminal = true, + ClassifierVerdictId = "whiteglove-sealing", + ClassifierHypothesisLevel = "Strong", + OccurredAtUtc = Occurred, + }; + + var (handler, repo) = Sut(); + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary)); + + var index = Assert.Single(repo.ClassifierVerdictsByIdLevel); + + Assert.Equal( + TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId), + TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId)); + Assert.Equal( + TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex), + TableDecisionTransitionRepository.BuildRowKey(index.StepIndex)); + } + + // ============================================================ pipeline idempotency + + [Fact] + public async Task Resubmitting_same_primary_rows_produces_same_index_keys() + { + // Simulates the M5.d.4 timer re-enqueueing primary rows it already handled — + // the handler's index-row keys must be key-stable so the real TableIndexRepository's + // UpsertReplace collapses duplicates rather than appending rows. + var (handler, repo) = Sut(); + + var signal = new SignalRecord + { + TenantId = TenantId, + SessionId = SessionId, + SessionSignalOrdinal = 1, + Kind = "DesktopArrived", + OccurredAtUtc = Occurred, + }; + var transition = TerminalTransition(); + + // First pass: initial ingest fan-out. + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromSignal(signal)); + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(transition)); + + // Second pass: 2h-timer reconcile resubmits the same primaries. + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromSignal(signal)); + await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(transition)); + + // The fake repo captures one row per call, but all duplicated rows must map to the + // same (PK, RK) — the real Azure Tables Upsert would collapse them. + Assert.Equal(2, repo.SignalsByKind.Count); + var signalKeys = repo.SignalsByKind + .Select(r => ( + Pk: IndexRowKeys.BuildSignalsByKindPk(r.TenantId, r.SignalKind), + Rk: IndexRowKeys.BuildSignalsByKindRk(r.OccurredAtUtc, r.SessionId, r.SessionSignalOrdinal))) + .Distinct() + .ToList(); + Assert.Single(signalKeys); + + Assert.Equal(2, repo.SessionsByTerminal.Count); + var terminalKeys = repo.SessionsByTerminal + .Select(r => ( + Pk: IndexRowKeys.BuildSessionsByTerminalPk(r.TenantId, r.TerminalStage), + Rk: IndexRowKeys.BuildSessionsByTerminalRk(r.OccurredAtUtc, r.SessionId))) + .Distinct() + .ToList(); + Assert.Single(terminalKeys); + + Assert.Equal(2, repo.SessionsByStage.Count); + var stageKeys = repo.SessionsByStage + .Select(r => ( + Pk: IndexRowKeys.BuildSessionsByStagePk(r.TenantId, r.Stage), + Rk: IndexRowKeys.BuildSessionsByStageRk(r.LastUpdatedUtc, r.SessionId))) + .Distinct() + .ToList(); + Assert.Single(stageKeys); + + Assert.Equal(2, repo.ClassifierVerdictsByIdLevel.Count); + var verdictKeys = repo.ClassifierVerdictsByIdLevel + .Select(r => ( + Pk: IndexRowKeys.BuildClassifierVerdictsByIdLevelPk(r.TenantId, r.ClassifierId, r.HypothesisLevel), + Rk: IndexRowKeys.BuildClassifierVerdictsByIdLevelRk(r.OccurredAtUtc, r.SessionId, r.StepIndex))) + .Distinct() + .ToList(); + Assert.Single(verdictKeys); + } + + // ============================================================ helpers + + private static DecisionTransitionRecord TerminalTransition() + => new() + { + TenantId = TenantId, + SessionId = SessionId, + StepIndex = 7, + FromStage = "EspInProgress", + ToStage = "WhiteGloveSealed", + Taken = true, + IsTerminal = true, + ClassifierVerdictId = "whiteglove-sealing", + ClassifierHypothesisLevel = "Strong", + OccurredAtUtc = Occurred, + }; + + private sealed class FakeIndexTableRepository : IIndexTableRepository + { + public List SessionsByTerminal { get; } = new(); + public List SessionsByStage { get; } = new(); + public List DeadEndsByReason { get; } = new(); + public List ClassifierVerdictsByIdLevel { get; } = new(); + public List SignalsByKind { get; } = new(); + + public Task StoreSessionsByTerminalAsync(IReadOnlyList records, CancellationToken ct = default) + { SessionsByTerminal.AddRange(records); return Task.FromResult(records.Count); } + public Task StoreSessionsByStageAsync(IReadOnlyList records, CancellationToken ct = default) + { SessionsByStage.AddRange(records); return Task.FromResult(records.Count); } + public Task StoreDeadEndsByReasonAsync(IReadOnlyList records, CancellationToken ct = default) + { DeadEndsByReason.AddRange(records); return Task.FromResult(records.Count); } + public Task StoreClassifierVerdictsByIdLevelAsync(IReadOnlyList records, CancellationToken ct = default) + { ClassifierVerdictsByIdLevel.AddRange(records); return Task.FromResult(records.Count); } + public Task StoreSignalsByKindAsync(IReadOnlyList records, CancellationToken ct = default) + { SignalsByKind.AddRange(records); return Task.FromResult(records.Count); } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IndexReconcileTimerTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IndexReconcileTimerTests.cs new file mode 100644 index 00000000..61e0e993 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IndexReconcileTimerTests.cs @@ -0,0 +1,156 @@ +using AutopilotMonitor.Functions.Functions.Admin; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using Microsoft.Extensions.Caching.Memory; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Tests for the 2h safety-net (Plan §2.8, §M5.d.4). +/// Covers the flag-gate short-circuit, the happy path (enqueue every primary row +/// in the window), and the empty-window case. +/// +public class IndexReconcileTimerTests +{ + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + + private static readonly DateTime Cutoff = new(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc); + + // ============================================================ flag-off short-circuit + + [Fact] + public async Task Feature_flag_off_skips_queries_and_enqueue() + { + var adminConfigSvc = BuildAdminConfigService(enableIndexDualWrite: false); + var signalRepo = new Mock(MockBehavior.Strict); + var transitionRepo = new Mock(MockBehavior.Strict); + var producer = new Mock(MockBehavior.Strict); + + var timer = new IndexReconcileTimer( + adminConfigSvc, + signalRepo.Object, + transitionRepo.Object, + producer.Object, + NullLogger.Instance); + + await timer.RunReconcileAsync(Cutoff, default); + + // Strict mocks throw if any unconfigured call happens — so zero setups + no throw + // proves the timer short-circuited before any query / enqueue. + signalRepo.VerifyNoOtherCalls(); + transitionRepo.VerifyNoOtherCalls(); + producer.VerifyNoOtherCalls(); + } + + // ============================================================ happy path + + [Fact] + public async Task Flag_on_with_primary_rows_enqueues_all_envelopes() + { + var adminConfigSvc = BuildAdminConfigService(enableIndexDualWrite: true); + + var signals = new List + { + new() { TenantId = TenantId, SessionId = SessionId, SessionSignalOrdinal = 1, Kind = "EspPhaseChanged", OccurredAtUtc = Cutoff.AddMinutes(5) }, + new() { TenantId = TenantId, SessionId = SessionId, SessionSignalOrdinal = 2, Kind = "DesktopArrived", OccurredAtUtc = Cutoff.AddMinutes(10) }, + }; + var transitions = new List + { + new() { TenantId = TenantId, SessionId = SessionId, StepIndex = 1, Taken = true, FromStage = "A", ToStage = "B", OccurredAtUtc = Cutoff.AddMinutes(7) }, + }; + + var signalRepo = new Mock(); + signalRepo + .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny())) + .ReturnsAsync(signals); + var transitionRepo = new Mock(); + transitionRepo + .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny())) + .ReturnsAsync(transitions); + + List? captured = null; + var producer = new Mock(); + producer + .Setup(p => p.EnqueueBatchAsync(It.IsAny>(), It.IsAny())) + .Callback, CancellationToken>((e, _) => captured = e.ToList()) + .ReturnsAsync(3); + + var timer = new IndexReconcileTimer( + adminConfigSvc, + signalRepo.Object, + transitionRepo.Object, + producer.Object, + NullLogger.Instance); + + await timer.RunReconcileAsync(Cutoff, default); + + Assert.NotNull(captured); + Assert.Equal(3, captured!.Count); + Assert.Equal("Signal", captured[0].SourceKind); + Assert.Equal(1L, captured[0].SessionSignalOrdinal); + Assert.Equal("Signal", captured[1].SourceKind); + Assert.Equal(2L, captured[1].SessionSignalOrdinal); + Assert.Equal("DecisionTransition", captured[2].SourceKind); + Assert.Equal(1, captured[2].StepIndex); + } + + // ============================================================ empty-window: no enqueue + + [Fact] + public async Task Flag_on_empty_window_skips_enqueue() + { + var adminConfigSvc = BuildAdminConfigService(enableIndexDualWrite: true); + + var signalRepo = new Mock(); + signalRepo + .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny())) + .ReturnsAsync(new List()); + var transitionRepo = new Mock(); + transitionRepo + .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny())) + .ReturnsAsync(new List()); + + var producer = new Mock(MockBehavior.Strict); + + var timer = new IndexReconcileTimer( + adminConfigSvc, + signalRepo.Object, + transitionRepo.Object, + producer.Object, + NullLogger.Instance); + + await timer.RunReconcileAsync(Cutoff, default); + + // Empty batch → no producer call (strict mock throws otherwise). + producer.VerifyNoOtherCalls(); + } + + // ============================================================ cutoff window invariant + + [Fact] + public void ReconcileWindow_is_4_hours_and_larger_than_timer_cadence() + { + // The 2h cron fires every 2h but scans a 4h window — so every primary row is + // re-checked at least twice before falling out of range. Protect the invariant. + Assert.Equal(TimeSpan.FromHours(4), IndexReconcileTimer.ReconcileWindow); + } + + // ============================================================ helpers + + private static AdminConfigurationService BuildAdminConfigService(bool enableIndexDualWrite) + { + var mock = new Mock( + Mock.Of(), + NullLogger.Instance, + new MemoryCache(new MemoryCacheOptions())); + + mock.Setup(s => s.GetConfigurationAsync()) + .ReturnsAsync(new AdminConfiguration { EnableIndexDualWrite = enableIndexDualWrite }); + + return mock.Object; + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs index 13874e60..44c091c0 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -100,6 +101,33 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage return results; } + public async Task> QueryByTimestampAtOrAfterAsync( + DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default) + { + var table = _storage.GetTableClient(Constants.TableNames.DecisionTransitions); + var filter = $"Timestamp ge datetime'{cutoffUtc.ToUniversalTime():yyyy-MM-ddTHH:mm:ss.fffffffZ}'"; + + var results = new List(capacity: 128); + var pages = table.QueryAsync( + filter: filter, + maxPerPage: 1000, + cancellationToken: cancellationToken); + + await foreach (var entity in pages.ConfigureAwait(false)) + { + if (results.Count >= maxResults) + { + _logger.LogWarning( + "DecisionTransitions: time-range query reached maxResults cap {Cap} at cutoff {Cutoff:o} — narrow window or bump cap", + maxResults, cutoffUtc); + break; + } + results.Add(FromEntity(entity)); + } + + return results; + } + /// /// Projects an Azure back into a , /// reassembling chunked PayloadJson if present. Internal for mapping tests. diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs index 298d4e7f..c35ffc60 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -101,6 +102,35 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage return results; } + public async Task> QueryByTimestampAtOrAfterAsync( + DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default) + { + var table = _storage.GetTableClient(Constants.TableNames.Signals); + // Timestamp is the Azure Tables system-managed server-side property. Filter shape + // follows OData datetime literal format. + var filter = $"Timestamp ge datetime'{cutoffUtc.ToUniversalTime():yyyy-MM-ddTHH:mm:ss.fffffffZ}'"; + + var results = new List(capacity: 128); + var pages = table.QueryAsync( + filter: filter, + maxPerPage: 1000, + cancellationToken: cancellationToken); + + await foreach (var entity in pages.ConfigureAwait(false)) + { + if (results.Count >= maxResults) + { + _logger.LogWarning( + "Signals: time-range query reached maxResults cap {Cap} at cutoff {Cutoff:o} — narrow window or bump cap", + maxResults, cutoffUtc); + break; + } + results.Add(FromEntity(entity)); + } + + return results; + } + /// /// Projects an Azure back into a , /// reassembling chunked PayloadJson if present. Internal for mapping tests. diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Admin/IndexReconcileTimer.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Admin/IndexReconcileTimer.cs new file mode 100644 index 00000000..96af808d --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Admin/IndexReconcileTimer.cs @@ -0,0 +1,100 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Functions.Services.Indexing; +using AutopilotMonitor.Shared.DataAccess; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Functions.Admin +{ + /// + /// 2h safety-net timer (Plan §2.8, §M5.d.4) that recovers missed index-table writes. + /// Scans the last 4h of primary Signals + DecisionTransitions rows and + /// re-enqueues envelopes onto the telemetry-index-reconcile queue via the + /// same producer the hot ingest path uses. The queue-triggered handler (M5.d.3) + /// then upserts index rows idempotently. + /// + /// Why 2h / 4h: window is deliberately larger than cadence so every row + /// gets re-checked at least twice before falling out of scan range. Queue failures + /// or pod restarts during ingest are recoverable within the overlap. + /// + /// + /// Feature gate: . + /// When false, the scan is short-circuited — no queries, no enqueues. + /// + /// + public sealed class IndexReconcileTimer + { + // Cron "0 0 */2 * * *" = top of every second hour (00:00, 02:00, 04:00 UTC, …) + // — matches the MaintenanceFunction cadence so ops only has two timer heartbeats + // to babysit. + private const string Cron = "0 0 */2 * * *"; + + /// Safety-net window: scan primaries committed in the last N hours. + internal static readonly TimeSpan ReconcileWindow = TimeSpan.FromHours(4); + + private readonly AdminConfigurationService _adminConfig; + private readonly ISignalRepository _signalRepo; + private readonly IDecisionTransitionRepository _transitionRepo; + private readonly IIndexReconcileProducer _producer; + private readonly ILogger _logger; + + public IndexReconcileTimer( + AdminConfigurationService adminConfig, + ISignalRepository signalRepo, + IDecisionTransitionRepository transitionRepo, + IIndexReconcileProducer producer, + ILogger logger) + { + _adminConfig = adminConfig; + _signalRepo = signalRepo; + _transitionRepo = transitionRepo; + _producer = producer; + _logger = logger; + } + + [Function("IndexReconcileTimer")] + public Task Run([TimerTrigger(Cron)] object timer, CancellationToken cancellationToken) + { + var cutoff = DateTime.UtcNow - ReconcileWindow; + return RunReconcileAsync(cutoff, cancellationToken); + } + + /// + /// Testable core. Separated from the TimerTrigger entry so unit tests can drive it + /// with a deterministic cutoff without an Azure Functions host. + /// + internal async Task RunReconcileAsync(DateTime cutoffUtc, CancellationToken cancellationToken) + { + var config = await _adminConfig.GetConfigurationAsync().ConfigureAwait(false); + if (!config.EnableIndexDualWrite) + { + _logger.LogInformation( + "IndexReconcileTimer: EnableIndexDualWrite=false — skipping scan (cutoff {Cutoff:o})", + cutoffUtc); + return; + } + + var signals = await _signalRepo + .QueryByTimestampAtOrAfterAsync(cutoffUtc, cancellationToken: cancellationToken) + .ConfigureAwait(false); + var transitions = await _transitionRepo + .QueryByTimestampAtOrAfterAsync(cutoffUtc, cancellationToken: cancellationToken) + .ConfigureAwait(false); + + var envelopes = IndexReconcileEnvelopeFactory.BuildBatch(signals, transitions); + + _logger.LogInformation( + "IndexReconcileTimer: re-enqueueing {Count} envelope(s) ({S} signals + {T} transitions) since {Cutoff:o}", + envelopes.Count, signals.Count, transitions.Count, cutoffUtc); + + if (envelopes.Count == 0) return; + + // The producer is already flag-gated and swallows queue-side failures. We trust + // its contract here — timer never throws back into the host. + await _producer.EnqueueBatchAsync(envelopes, cancellationToken).ConfigureAwait(false); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs b/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs index 9994e904..c55af9db 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs @@ -30,9 +30,13 @@ namespace AutopilotMonitor.Functions.Services } /// - /// Gets global admin configuration (uses cache with 5-minute TTL) + /// Gets global admin configuration (uses cache with 5-minute TTL). + /// + /// Virtual so it can be mocked via Moq in consumer tests that need controlled + /// flag values (e.g. IndexReconcileTimer flag-gate tests). + /// /// - public async Task GetConfigurationAsync() + public virtual async Task GetConfigurationAsync() { if (_cache.TryGetValue(CacheKey, out AdminConfiguration? cachedConfig) && cachedConfig != null) { diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs index 1abdf10a..77a0b766 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -27,5 +28,19 @@ namespace AutopilotMonitor.Shared.DataAccess /// Task> QueryBySessionAsync( string tenantId, string sessionId, int maxResults = 1000, CancellationToken cancellationToken = default); + + /// + /// Cross-partition scan of all decision transitions whose Azure Tables server-side + /// Timestamp is at or after . Used by the + /// IndexReconcileTimer (Plan §M5.d.4) to re-enqueue the last 4h of primary + /// rows as a safety-net against queue failures. + /// + /// Perf note: Azure Tables has no PK-bound filter here, so this is a scan; + /// bounds memory. If the cap is reached the caller + /// should log + narrow the window rather than trust completeness. + /// + /// + Task> QueryByTimestampAtOrAfterAsync( + DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default); } } diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs index 86107880..edf88c8f 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -29,5 +30,19 @@ namespace AutopilotMonitor.Shared.DataAccess /// Task> QueryBySessionAsync( string tenantId, string sessionId, int maxResults = 1000, CancellationToken cancellationToken = default); + + /// + /// Cross-partition scan of all signals whose Azure Tables server-side Timestamp + /// is at or after . Used by the + /// IndexReconcileTimer (Plan §M5.d.4) to re-enqueue the last 4h of primary + /// rows as a safety-net against queue failures. + /// + /// Perf note: Azure Tables has no PK-bound filter here, so this is a scan; + /// bounds memory. If the cap is reached the caller + /// should log + narrow the window rather than trust completeness. + /// + /// + Task> QueryByTimestampAtOrAfterAsync( + DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default); } }