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) <noreply@anthropic.com>
2026-10-24 12:42:38 +00:00 · 2026-06-08 17:09:46 +02:00
parent 291789cd54
commit ad345f5869
5 changed files with 181 additions and 17 deletions
@@ -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;
+
+/// <summary>
+/// Covers <see cref="TableBatchDedup.ByRowKey"/> — the guard that stops an agent replaying an
+/// overlapping ordinal/step from failing an entire ingest entity-group-transaction with Azure
+/// Tables' HTTP 400 / <c>InvalidDuplicateRow</c>. Exercised through the three real ingest
+/// projections (Signals, DecisionTransitions, Index) so the (PK, RowKey) contract is asserted
+/// against the keys those repos actually emit.
+/// </summary>
+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<TableEntity>());
+
+        Assert.Empty(deduped);
+        Assert.Equal(0, dropped);
+    }
+}
@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Azure.Data.Tables;
+
+namespace AutopilotMonitor.Functions.DataAccess.TableStorage
+{
+    /// <summary>
+    /// Collapses entities sharing a (PartitionKey, RowKey) within a single batch down to the
+    /// last occurrence (idempotent <see cref="TableTransactionActionType.UpsertReplace"/> →
+    /// last-write-wins). Azure Table Storage rejects the <em>entire</em> entity-group-transaction
+    /// with HTTP 400 / <c>InvalidDuplicateRow</c> 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.
+    /// </summary>
+    internal static class TableBatchDedup
+    {
+        /// <summary>
+        /// 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.
+        /// </summary>
+        internal static (List<TableEntity> Deduped, int Dropped) ByRowKey(IEnumerable<TableEntity> 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);
+        }
+    }
+}
@@ -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;
@@ -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();
@@ -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;