From ad345f5869e5522da21eba157c2babb56c1e2e84 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Mon, 8 Jun 2026 17:09:46 +0200 Subject: [PATCH] fix(ingest): dedup duplicate RowKeys before batch transaction in V2 ingest repos A telemetry batch containing two rows with the same RowKey made Azure Table Storage reject the entire entity-group-transaction with HTTP 400 / InvalidDuplicateRow, surfacing as HTTP 500 on POST IngestTelemetry. Observed as a 23-minute spike (36x 500) from a single session replaying an overlapping SessionSignalOrdinal; self-resolved when the agent stopped retrying. The three V2 ingest write paths (Signals, DecisionTransitions, Index) built the transaction without deduping duplicate RowKeys and had no fallback (unlike the Events path). Add a shared TableBatchDedup.ByRowKey helper that collapses duplicate (PartitionKey, RowKey) entities last-write-wins (idempotent UpsertReplace) and logs a WARN with the dropped count, so bad agents become visible in logs instead of failing the batch. SubmitTransaction stays fail-loud for any other batch error. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../TableBatchDedupTests.cs | 101 ++++++++++++++++++ .../TableStorage/TableBatchDedup.cs | 43 ++++++++ .../TableDecisionTransitionRepository.cs | 20 ++-- .../TableStorage/TableIndexRepository.cs | 13 ++- .../TableStorage/TableSignalRepository.cs | 21 ++-- 5 files changed, 181 insertions(+), 17 deletions(-) create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/TableBatchDedupTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableBatchDedup.cs diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TableBatchDedupTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TableBatchDedupTests.cs new file mode 100644 index 00000000..11d065f7 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TableBatchDedupTests.cs @@ -0,0 +1,101 @@ +using System; +using System.Linq; +using AutopilotMonitor.Functions.DataAccess.TableStorage; +using AutopilotMonitor.Shared.Models; +using Azure.Data.Tables; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Covers — the guard that stops an agent replaying an +/// overlapping ordinal/step from failing an entire ingest entity-group-transaction with Azure +/// Tables' HTTP 400 / InvalidDuplicateRow. Exercised through the three real ingest +/// projections (Signals, DecisionTransitions, Index) so the (PK, RowKey) contract is asserted +/// against the keys those repos actually emit. +/// +public class TableBatchDedupTests +{ + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + + private static SignalRecord Signal(long ordinal, string kind) => new() + { + TenantId = TenantId, + SessionId = SessionId, + SessionSignalOrdinal = ordinal, + SessionTraceOrdinal = ordinal, + Kind = kind, + KindSchemaVersion = 1, + OccurredAtUtc = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc), + SourceOrigin = "test", + PayloadJson = "{}", + }; + + [Fact] + public void ByRowKey_collapses_duplicate_signal_ordinal_keeping_last_write() + { + var first = TableSignalRepository.ToEntity(Signal(6, "first")); + var second = TableSignalRepository.ToEntity(Signal(6, "second")); // same RowKey as first + var other = TableSignalRepository.ToEntity(Signal(7, "other")); + + var (deduped, dropped) = TableBatchDedup.ByRowKey(new[] { first, second, other }); + + Assert.Equal(1, dropped); + Assert.Equal(2, deduped.Count); + + var collapsed = deduped.Single(e => e.RowKey == TableSignalRepository.BuildRowKey(6)); + Assert.Equal("second", collapsed.GetString("Kind")); // last-wins + } + + [Fact] + public void ByRowKey_collapses_duplicate_transition_stepIndex() + { + DecisionTransitionRecord Transition(int step, string toStage) => new() + { + TenantId = TenantId, + SessionId = SessionId, + StepIndex = step, + SessionTraceOrdinal = step, + SignalOrdinalRef = step, + OccurredAtUtc = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc), + Trigger = "t", + FromStage = "A", + ToStage = toStage, + ReducerVersion = "1", + PayloadJson = "{}", + }; + + var first = TableDecisionTransitionRepository.ToEntity(Transition(3, "Stage1")); + var second = TableDecisionTransitionRepository.ToEntity(Transition(3, "Stage2")); + + var (deduped, dropped) = TableBatchDedup.ByRowKey(new[] { first, second }); + + Assert.Equal(1, dropped); + var collapsed = Assert.Single(deduped); + Assert.Equal("Stage2", collapsed.GetString("ToStage")); // last-wins + } + + [Fact] + public void ByRowKey_is_a_passthrough_when_no_duplicates_and_orders_by_rowkey() + { + var ten = TableSignalRepository.ToEntity(Signal(10, "ten")); + var nine = TableSignalRepository.ToEntity(Signal(9, "nine")); + + var (deduped, dropped) = TableBatchDedup.ByRowKey(new[] { ten, nine }); + + Assert.Equal(0, dropped); + Assert.Equal(2, deduped.Count); + // D19 padding → RowKey lex order matches numeric order: 9 before 10. + Assert.Equal(TableSignalRepository.BuildRowKey(9), deduped[0].RowKey); + Assert.Equal(TableSignalRepository.BuildRowKey(10), deduped[1].RowKey); + } + + [Fact] + public void ByRowKey_handles_empty_input() + { + var (deduped, dropped) = TableBatchDedup.ByRowKey(Array.Empty()); + + Assert.Empty(deduped); + Assert.Equal(0, dropped); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableBatchDedup.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableBatchDedup.cs new file mode 100644 index 00000000..1f1bc38d --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableBatchDedup.cs @@ -0,0 +1,43 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using Azure.Data.Tables; + +namespace AutopilotMonitor.Functions.DataAccess.TableStorage +{ + /// + /// Collapses entities sharing a (PartitionKey, RowKey) within a single batch down to the + /// last occurrence (idempotent → + /// last-write-wins). Azure Table Storage rejects the entire entity-group-transaction + /// with HTTP 400 / InvalidDuplicateRow if any two actions target the same RowKey, so a + /// duplicate-keyed record (e.g. an agent replaying an overlapping signal ordinal) would + /// otherwise fail the whole ingest batch. Callers must group by PartitionKey before submitting; + /// the (PK, RK) key here keeps the helper correct even if that precondition is ever relaxed. + /// + internal static class TableBatchDedup + { + /// + /// Returns the de-duplicated entities (unique by (PartitionKey, RowKey), last-wins) plus the + /// number of rows dropped. Output is PK/RowKey-ordered for deterministic chunk boundaries. + /// + internal static (List Deduped, int Dropped) ByRowKey(IEnumerable entities) + { + var byKey = new Dictionary<(string Pk, string Rk), TableEntity>(); + var dropped = 0; + + foreach (var e in entities) + { + var key = (e.PartitionKey, e.RowKey); + if (byKey.ContainsKey(key)) dropped++; + byKey[key] = e; // last-wins + } + + var deduped = byKey.Values + .OrderBy(e => e.PartitionKey, StringComparer.Ordinal) + .ThenBy(e => e.RowKey, StringComparer.Ordinal) + .ToList(); + + return (deduped, dropped); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs index 44c091c0..a7501f12 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs @@ -51,17 +51,23 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage foreach (var group in records.GroupBy(r => (r.TenantId, r.SessionId))) { - var ordered = group.OrderBy(r => r.StepIndex).ToList(); + // Collapse duplicate RowKeys (D10(StepIndex)) before building the transaction — + // Azure Tables rejects the whole batch with InvalidDuplicateRow otherwise. Dedup + // also yields RowKey ordering, so the previous OrderBy(StepIndex) is redundant. + var (deduped, dropped) = TableBatchDedup.ByRowKey(group.Select(ToEntity)); + if (dropped > 0) + _logger.LogWarning( + "DecisionTransitions: dropped {Dropped} duplicate-RowKey row(s) (last-wins) for {Tenant}_{Session} — agent likely replayed overlapping StepIndex", + dropped, group.Key.TenantId, group.Key.SessionId); - for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize) + for (var offset = 0; offset < deduped.Count; offset += TransactionChunkSize) { cancellationToken.ThrowIfCancellationRequested(); - var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList(); + var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList(); var actions = chunk - .Select(r => new TableTransactionAction( - TableTransactionActionType.UpsertReplace, - ToEntity(r))) + .Select(e => new TableTransactionAction( + TableTransactionActionType.UpsertReplace, e)) .ToList(); await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false); @@ -70,7 +76,7 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage _logger.LogDebug( "DecisionTransitions: committed {Count} rows for {Tenant}_{Session}", - ordered.Count, group.Key.TenantId, group.Key.SessionId); + deduped.Count, group.Key.TenantId, group.Key.SessionId); } return committed; diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs index 44474624..bab7303c 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs @@ -84,13 +84,20 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage foreach (var group in entities.GroupBy(e => e.PartitionKey)) { - var ordered = group.OrderBy(e => e.RowKey, StringComparer.Ordinal).ToList(); + // Collapse duplicate RowKeys before building the transaction — Azure Tables + // rejects the whole batch with InvalidDuplicateRow otherwise. Dedup also yields + // RowKey ordering, replacing the previous explicit OrderBy. + var (deduped, dropped) = TableBatchDedup.ByRowKey(group); + if (dropped > 0) + _logger.LogWarning( + "{Table}: dropped {Dropped} duplicate-RowKey index row(s) (last-wins) for partition {Pk}", + tableName, dropped, group.Key); - for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize) + for (var offset = 0; offset < deduped.Count; offset += TransactionChunkSize) { cancellationToken.ThrowIfCancellationRequested(); - var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList(); + var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList(); var actions = chunk .Select(e => new TableTransactionAction(TableTransactionActionType.UpsertReplace, e)) .ToList(); diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs index c35ffc60..b59f1001 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs @@ -49,17 +49,24 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage foreach (var group in records.GroupBy(r => (r.TenantId, r.SessionId))) { - var ordered = group.OrderBy(r => r.SessionSignalOrdinal).ToList(); + // Collapse duplicate RowKeys (D19(SessionSignalOrdinal)) before building the + // transaction — Azure Tables rejects the whole batch with InvalidDuplicateRow + // otherwise (e.g. an agent replaying an overlapping ordinal). Dedup also yields + // RowKey ordering, so the previous OrderBy(ordinal) is no longer needed. + var (deduped, dropped) = TableBatchDedup.ByRowKey(group.Select(ToEntity)); + if (dropped > 0) + _logger.LogWarning( + "Signals: dropped {Dropped} duplicate-RowKey row(s) (last-wins) for {Tenant}_{Session} — agent likely replayed overlapping SessionSignalOrdinal", + dropped, group.Key.TenantId, group.Key.SessionId); - for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize) + for (var offset = 0; offset < deduped.Count; offset += TransactionChunkSize) { cancellationToken.ThrowIfCancellationRequested(); - var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList(); + var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList(); var actions = chunk - .Select(r => new TableTransactionAction( - TableTransactionActionType.UpsertReplace, - ToEntity(r))) + .Select(e => new TableTransactionAction( + TableTransactionActionType.UpsertReplace, e)) .ToList(); await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false); @@ -68,7 +75,7 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage _logger.LogDebug( "Signals: committed {Count} rows for {Tenant}_{Session}", - ordered.Count, group.Key.TenantId, group.Key.SessionId); + deduped.Count, group.Key.TenantId, group.Key.SessionId); } return committed;