From d0cb7e66f412aafbb7cfab6ad402f5c23d23e418 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Sat, 5 Sep 2026 22:03:07 +0200 Subject: [PATCH] backend: byte-aware table transactions and classified storage failures on the ingest (audit F07) Every entity-group transaction was chunked by count only; 100 payload-heavy rows (Signals with chunked PayloadJson, Events with 30k-char Message + DataJson) could exceed the 4 MiB transaction cap, and the failure surfaced as an opaque 500. The agent halves its batch only on 413 and treats 5xx as "replay later", so such a batch wedged the device forever. - TableTransactionBatcher: 100 actions, 3.5 MB per transaction, 1 MB per entity, never across partitions; an entity that cannot fit throws TableEntityTooLargeException. Used by the event, signal, transition, restore and rules writers; delete-only batches keep their count chunking. - StorageErrors: one classifier for RequestFailedException (transient 408/429/ 5xx, payload-too-large by status or error code, already-exists incl. the Azurite 400 shape); the two IsAlreadyExistsStatus copies use it. - Event writer: per-row fallback only for data-shaped rejections; throttling, outages and oversize batches propagate. Single and batched writer share one entity builder. - IngestTelemetryFunction: payload too large -> 413, transient -> 503 with Retry-After: 5, anything else 500. - TableStorageChunking.MaxPropertyChars is the one 30,000-char bound. Co-Authored-By: Claude Fable 5.1 --- ...IngestStorageFailureClassificationTests.cs | 54 +++++ .../StorageErrorsTests.cs | 62 ++++++ .../TableStorageEventBatchTests.cs | 145 ++++++++++++ .../TableTransactionBatcherTests.cs | 127 +++++++++++ .../TableDecisionTransitionRepository.cs | 15 +- .../TableStorage/TableSignalRepository.cs | 16 +- .../TableStorage/TableStorageChunking.cs | 9 +- .../TableStorage/TableTransactionBatcher.cs | 135 ++++++++++++ .../Ingest/IngestTelemetryFunction.cs | 42 +++- .../Helpers/StorageErrors.cs | 49 +++++ .../Backup/CriticalTableRestoreService.cs | 9 +- .../Services/TableStorageService.Restore.cs | 53 +---- .../Services/TableStorageService.Rules.cs | 45 +--- .../Services/TableStorageService.Sessions.cs | 207 ++++++++---------- 14 files changed, 745 insertions(+), 223 deletions(-) create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/TableStorageEventBatchTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/TableTransactionBatcherTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableTransactionBatcher.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs new file mode 100644 index 00000000..aefdaf43 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStorageFailureClassificationTests.cs @@ -0,0 +1,54 @@ +using System.Net; +using Azure; +using AutopilotMonitor.Functions.Functions.Ingest; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// The agent's uploader halves its batch only on 413 and treats every 5xx as "replay later"; +/// is what keeps an oversize +/// transaction from surfacing as a 500 the device would retry forever. +/// +public class IngestStorageFailureClassificationTests +{ + private static RequestFailedException Failure(int status, string? code = null) + => new RequestFailedException(status, "storage", code, null); + + [Fact] + public void Oversize_transaction_becomes_413_without_retry_after() + { + var (status, _, retryAfter) = IngestTelemetryFunction.ClassifyStorageFailure(Failure(413, "RequestBodyTooLarge")); + Assert.Equal(HttpStatusCode.RequestEntityTooLarge, status); + Assert.Null(retryAfter); + } + + [Fact] + public void Oversize_reported_as_400_with_size_code_still_becomes_413() + { + var (status, _, _) = IngestTelemetryFunction.ClassifyStorageFailure(Failure(400, "EntityTooLarge")); + Assert.Equal(HttpStatusCode.RequestEntityTooLarge, status); + } + + [Theory] + [InlineData(503)] + [InlineData(429)] + [InlineData(500)] + public void Transient_storage_failure_becomes_503_with_retry_after(int storageStatus) + { + var (status, _, retryAfter) = IngestTelemetryFunction.ClassifyStorageFailure(Failure(storageStatus)); + Assert.Equal(HttpStatusCode.ServiceUnavailable, status); + Assert.Equal(IngestTelemetryFunction.StorageRetryAfterSeconds, retryAfter); + } + + [Theory] + [InlineData(404)] + [InlineData(409)] + [InlineData(400)] + public void Anything_else_stays_500(int storageStatus) + { + var (status, _, retryAfter) = IngestTelemetryFunction.ClassifyStorageFailure(Failure(storageStatus, "InvalidInput")); + Assert.Equal(HttpStatusCode.InternalServerError, status); + Assert.Null(retryAfter); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs new file mode 100644 index 00000000..b5e5f6ba --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/StorageErrorsTests.cs @@ -0,0 +1,62 @@ +using Azure; +using AutopilotMonitor.Functions.Helpers; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +public class StorageErrorsTests +{ + private static RequestFailedException Failure(int status, string? code = null) + => new RequestFailedException(status, "storage said no", code, null); + + [Theory] + [InlineData(408)] + [InlineData(429)] + [InlineData(500)] + [InlineData(502)] + [InlineData(503)] + [InlineData(504)] + public void Transient_statuses_are_transient(int status) + => Assert.True(StorageErrors.IsTransient(Failure(status))); + + [Theory] + [InlineData(400)] + [InlineData(404)] + [InlineData(409)] + [InlineData(412)] + [InlineData(413)] + public void Client_side_statuses_are_not_transient(int status) + => Assert.False(StorageErrors.IsTransient(Failure(status))); + + [Fact] + public void Payload_too_large_by_status() + => Assert.True(StorageErrors.IsPayloadTooLarge(Failure(413))); + + [Theory] + [InlineData("RequestBodyTooLarge")] + [InlineData("EntityTooLarge")] + [InlineData("PropertyValueTooLarge")] + public void Payload_too_large_by_error_code_even_with_a_400(string code) + => Assert.True(StorageErrors.IsPayloadTooLarge(Failure(400, code))); + + [Fact] + public void A_plain_400_is_not_payload_too_large() + => Assert.False(StorageErrors.IsPayloadTooLarge(Failure(400, "InvalidInput"))); + + [Fact] + public void Already_exists_is_409_or_the_azurite_400_shape() + { + Assert.True(StorageErrors.IsAlreadyExists(Failure(409))); + Assert.True(StorageErrors.IsAlreadyExists(Failure(400, "EntityAlreadyExists"))); + Assert.False(StorageErrors.IsAlreadyExists(Failure(400, "InvalidInput"))); + Assert.False(StorageErrors.IsAlreadyExists(Failure(412))); + } + + [Fact] + public void Null_is_never_classified() + { + Assert.False(StorageErrors.IsTransient(null!)); + Assert.False(StorageErrors.IsPayloadTooLarge(null!)); + Assert.False(StorageErrors.IsAlreadyExists(null!)); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageEventBatchTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageEventBatchTests.cs new file mode 100644 index 00000000..9f809822 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TableStorageEventBatchTests.cs @@ -0,0 +1,145 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared.Models; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// after audit F07: byte-aware +/// transactions, and a per-row fallback only for data-shaped rejections. Throttling, outages +/// and oversize batches propagate so the ingest can answer 503 / 413 and the agent replays. +/// +public class TableStorageEventBatchTests +{ + private const string TenantId = "11111111-1111-1111-1111-111111111111"; + private const string SessionId = "22222222-2222-2222-2222-222222222222"; + + private static EnrollmentEvent Event(int i, int payloadChars = 0) + { + var data = new Dictionary(); + if (payloadChars > 0) data["blob"] = new string('d', payloadChars); + return new EnrollmentEvent + { + EventId = $"e{i:D4}", + TenantId = TenantId, + SessionId = SessionId, + EventType = "test_event", + Source = "test", + Message = payloadChars > 0 ? new string('m', payloadChars) : $"message {i}", + Timestamp = new DateTime(2026, 9, 5, 12, 0, 0, DateTimeKind.Utc).AddMilliseconds(i), + ReceivedAt = DateTime.UtcNow, + Sequence = i, + Data = data, + }; + } + + [Fact] + public async Task Transient_failure_propagates_without_per_row_fallback() + { + var h = new Harness { BatchBehavior = _ => throw new RequestFailedException(503, "busy", "ServerBusy", null) }; + + await Assert.ThrowsAsync(() => h.Sut.StoreEventsBatchAsync(Enumerable.Range(0, 5).Select(i => Event(i)).ToList())); + + Assert.Single(h.SubmittedBatches); + Assert.Empty(h.PerRowEventUpserts); + } + + [Fact] + public async Task Oversize_batch_propagates_without_per_row_fallback() + { + var h = new Harness { BatchBehavior = _ => throw new RequestFailedException(413, "too big", "RequestBodyTooLarge", null) }; + + await Assert.ThrowsAsync(() => h.Sut.StoreEventsBatchAsync(Enumerable.Range(0, 5).Select(i => Event(i)).ToList())); + + Assert.Empty(h.PerRowEventUpserts); + } + + [Fact] + public async Task Data_shaped_rejection_falls_back_to_per_row_writes() + { + var h = new Harness { BatchBehavior = _ => throw new RequestFailedException(400, "bad row", "InvalidInput", null) }; + var events = Enumerable.Range(0, 5).Select(i => Event(i)).ToList(); + + var stored = await h.Sut.StoreEventsBatchAsync(events); + + Assert.Equal(5, stored.Count); + Assert.Equal(5, h.PerRowEventUpserts.Count); + } + + [Fact] + public async Task Payload_heavy_events_split_into_several_transactions_under_the_byte_budget() + { + // Each event carries ~30k chars in Message and ~30k in DataJson (truncation bound) ≈ 120 KB; + // 40 of them exceed the 3.5 MB budget, so the writer must submit more than one transaction. + var events = Enumerable.Range(0, 40).Select(i => Event(i, payloadChars: 40_000)).ToList(); + var h = new Harness(); + + var stored = await h.Sut.StoreEventsBatchAsync(events); + + Assert.Equal(40, stored.Count); + Assert.True(h.SubmittedBatches.Count >= 2, $"expected ≥2 transactions, got {h.SubmittedBatches.Count}"); + Assert.Equal(40, h.SubmittedBatches.Sum(b => b.Count)); + Assert.Empty(h.PerRowEventUpserts); + } + + [Fact] + public async Task Small_batches_stay_one_transaction_and_dedup_last_wins() + { + var h = new Harness(); + var a = Event(1); a.Message = "first"; + var b = Event(1); b.Message = "second"; // same Timestamp + Sequence → same RowKey + + var stored = await h.Sut.StoreEventsBatchAsync(new List { a, b, Event(2) }); + + Assert.Equal(2, stored.Count); + var batch = Assert.Single(h.SubmittedBatches); + Assert.Equal(2, batch.Count); + Assert.Equal("second", ((TableEntity)batch[0].Entity).GetString("Message")); + } + + private sealed class Harness + { + public List> SubmittedBatches { get; } = new(); + public List<(string Pk, string Rk)> PerRowEventUpserts { get; } = new(); + public Func, Response>>? BatchBehavior { get; set; } + public TableStorageService Sut { get; } + + public Harness() + { + var table = new Mock(); + table + .Setup(c => c.SubmitTransactionAsync(It.IsAny>(), It.IsAny())) + .Returns, CancellationToken>((actions, _) => + { + var snapshot = actions.ToList(); + SubmittedBatches.Add(snapshot); + if (BatchBehavior != null) return Task.FromResult(BatchBehavior(snapshot)); + var inner = (IReadOnlyList)snapshot.Select(_ => new Mock().Object).ToList(); + return Task.FromResult(Response.FromValue(inner, new Mock().Object)); + }); + table + .Setup(c => c.UpsertEntityAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((entity, _, _) => + { + // Event rows live in the composite {tenant}_{session} partition; the + // EventSessionIndex side row uses the bare tenant id. + if (entity.PartitionKey.Contains('_')) + PerRowEventUpserts.Add((entity.PartitionKey, entity.RowKey)); + return Task.FromResult(new Mock().Object); + }); + + var service = new Mock(); + service.Setup(s => s.GetTableClient(It.IsAny())).Returns(table.Object); + Sut = new TableStorageService(service.Object, NullLogger.Instance); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TableTransactionBatcherTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TableTransactionBatcherTests.cs new file mode 100644 index 00000000..82e44265 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TableTransactionBatcherTests.cs @@ -0,0 +1,127 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.DataAccess.TableStorage; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// is the one place that knows all three Azure Tables +/// limits (100 actions, ~4 MiB per transaction, ~1 MiB per entity). These tests pin the split +/// rules; the writers only iterate what it returns. +/// +public class TableTransactionBatcherTests +{ + private static TableEntity Row(string pk, string rk, int payloadChars = 0) + { + var e = new TableEntity(pk, rk) { ["Kind"] = "x", ["Ordinal"] = 1L }; + if (payloadChars > 0) e["PayloadJson"] = new string('p', payloadChars); + return e; + } + + [Fact] + public void Split_caps_at_100_actions_per_transaction_and_preserves_order() + { + var rows = Enumerable.Range(0, 250).Select(i => Row("pk", $"r{i:D4}")).ToList(); + + var batches = TableTransactionBatcher.Split(rows, TableTransactionActionType.UpsertReplace); + + Assert.Equal(new[] { 100, 100, 50 }, batches.Select(b => b.Count).ToArray()); + Assert.All(batches.SelectMany(b => b), a => Assert.Equal(TableTransactionActionType.UpsertReplace, a.ActionType)); + Assert.Equal(rows.Select(r => r.RowKey), batches.SelectMany(b => b).Select(a => a.Entity.RowKey)); + } + + [Fact] + public void Split_starts_a_new_transaction_when_the_byte_budget_would_overflow() + { + // 200k chars ≈ 400 KB each → 8 fit under 3.5 MB, the 9th does not. + var rows = Enumerable.Range(0, 20).Select(i => Row("pk", $"r{i:D2}", payloadChars: 200_000)).ToList(); + + var batches = TableTransactionBatcher.Split(rows, TableTransactionActionType.UpsertReplace); + + Assert.Equal(new[] { 8, 8, 4 }, batches.Select(b => b.Count).ToArray()); + Assert.All(batches, b => Assert.True( + b.Sum(a => TableTransactionBatcher.EstimateEntityBytes((TableEntity)a.Entity)) <= TableTransactionBatcher.TransactionByteBudget)); + } + + [Fact] + public void Split_never_mixes_partitions_in_one_transaction() + { + var rows = new[] { Row("a", "1"), Row("a", "2"), Row("b", "1"), Row("a", "3") }; + + var batches = TableTransactionBatcher.Split(rows, TableTransactionActionType.Add); + + Assert.Equal(3, batches.Count); + Assert.All(batches, b => Assert.Single(b.Select(a => a.Entity.PartitionKey).Distinct())); + } + + [Fact] + public void Split_throws_for_an_entity_that_can_never_fit() + { + var rows = new[] { Row("pk", "ok"), Row("pk", "huge", payloadChars: 600_000) }; + + var ex = Assert.Throws( + () => TableTransactionBatcher.Split(rows, TableTransactionActionType.UpsertReplace)); + + Assert.Equal("pk", ex.PartitionKey); + Assert.Equal("huge", ex.RowKey); + Assert.True(ex.EstimatedBytes > TableTransactionBatcher.EntityByteLimit); + } + + [Fact] + public void Split_honours_a_smaller_action_cap() + { + var rows = Enumerable.Range(0, 7).Select(i => Row("pk", $"r{i}")).ToList(); + + var batches = TableTransactionBatcher.Split(rows, TableTransactionActionType.Delete, maxActions: 3); + + Assert.Equal(new[] { 3, 3, 1 }, batches.Select(b => b.Count).ToArray()); + } + + [Fact] + public void Split_of_nothing_is_nothing() + { + Assert.Empty(TableTransactionBatcher.Split(Array.Empty(), TableTransactionActionType.UpsertReplace)); + } + + [Fact] + public void Estimate_counts_strings_at_utf16_width_binaries_at_base64_growth_and_scalars_flat() + { + var e = new TableEntity("pk", "rk") + { + ["S"] = new string('s', 1000), + ["B"] = new byte[300], + ["I"] = 7, + ["L"] = 7L, + ["D"] = 1.5, + ["F"] = true, + ["T"] = DateTime.UtcNow, + ["G"] = Guid.NewGuid(), + ["N"] = null, + }; + + var estimate = TableTransactionBatcher.EstimateEntityBytes(e); + + var expected = 256 + 2 * 2 + 2 * 2 // envelope + keys + + (2 + 2000) // "S" + + (2 + 400) // "B": 300 bytes → 400 base64 chars + + 6 * (2 + 48) // I L D F T G + + (2 + 0); // N + Assert.Equal(expected, estimate); + } + + [Fact] + public void Estimate_of_a_chunked_payload_counts_every_chunk_and_its_name() + { + var e = new TableEntity("pk", "rk"); + foreach (var kv in TableStorageChunking.ChunkProperty("PayloadJson", new string('x', 75_000))) + e[kv.Key] = kv.Value; + + var estimate = TableTransactionBatcher.EstimateEntityBytes(e); + + Assert.True(estimate > 150_000, $"estimate {estimate} must cover 75k UTF-16 chars"); + Assert.True(estimate < TableTransactionBatcher.EntityByteLimit); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs index 7d4c1563..1b484d00 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs @@ -20,8 +20,6 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage /// public sealed class TableDecisionTransitionRepository : IDecisionTransitionRepository { - /// Azure Table Storage limit per entity-group-transaction. - internal const int TransactionChunkSize = 100; private readonly TableStorageService _storage; private readonly ILogger _logger; @@ -60,18 +58,13 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage "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 < deduped.Count; offset += TransactionChunkSize) + // Byte-aware batching (100 actions / 3.5 MB per transaction / 1 MB per entity); + // storage failures propagate classified — the ingest maps them to 413 / 503 / 500. + foreach (var actions in TableTransactionBatcher.Split(deduped, TableTransactionActionType.UpsertReplace)) { cancellationToken.ThrowIfCancellationRequested(); - - var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList(); - var actions = chunk - .Select(e => new TableTransactionAction( - TableTransactionActionType.UpsertReplace, e)) - .ToList(); - await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false); - committed += chunk.Count; + committed += actions.Count; } _logger.LogDebug( diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs index b4b35abb..23fd035a 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs @@ -20,9 +20,6 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage /// public sealed class TableSignalRepository : ISignalRepository { - /// Azure Table Storage limit per entity-group-transaction. - internal const int TransactionChunkSize = 100; - private readonly TableStorageService _storage; private readonly ILogger _logger; @@ -59,18 +56,13 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage "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 < deduped.Count; offset += TransactionChunkSize) + // Byte-aware batching (100 actions / 3.5 MB per transaction / 1 MB per entity); + // storage failures propagate classified — the ingest maps them to 413 / 503 / 500. + foreach (var actions in TableTransactionBatcher.Split(deduped, TableTransactionActionType.UpsertReplace)) { cancellationToken.ThrowIfCancellationRequested(); - - var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList(); - var actions = chunk - .Select(e => new TableTransactionAction( - TableTransactionActionType.UpsertReplace, e)) - .ToList(); - await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false); - committed += chunk.Count; + committed += actions.Count; } _logger.LogDebug( diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableStorageChunking.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableStorageChunking.cs index 28d80852..212c3b09 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableStorageChunking.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableStorageChunking.cs @@ -13,14 +13,19 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage /// public static class TableStorageChunking { - private const int DefaultMaxChunkSize = 30_000; + /// + /// Largest string kept in one property: 30,000 UTF-16 chars stay under the 64 KiB + /// per-property limit with headroom. The event writer truncates to the same bound; + /// guards the entity and transaction totals. + /// + public const int MaxPropertyChars = 30_000; /// /// Splits a string value into chunked properties if it exceeds maxChunkSize. /// Small values: { "Prop": value } /// Large values: { "Prop_0": chunk0, "Prop_1": chunk1, "Prop_ChunkCount": "2" } /// - public static Dictionary ChunkProperty(string propertyName, string value, int maxChunkSize = DefaultMaxChunkSize) + public static Dictionary ChunkProperty(string propertyName, string value, int maxChunkSize = MaxPropertyChars) { if (string.IsNullOrEmpty(value) || value.Length <= maxChunkSize) return new Dictionary { { propertyName, value ?? "" } }; diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableTransactionBatcher.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableTransactionBatcher.cs new file mode 100644 index 00000000..f115ea3e --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableTransactionBatcher.cs @@ -0,0 +1,135 @@ +using System; +using System.Collections.Generic; +using Azure.Data.Tables; + +namespace AutopilotMonitor.Functions.DataAccess.TableStorage +{ + /// + /// Splits entities into entity-group transactions that respect all three Azure Tables + /// limits at once: at most 100 actions per transaction, a transaction payload under 4 MiB + /// and a single entity under 1 MiB. Every other writer used to chunk by count only, which + /// let 100 payload-heavy rows (Signals with chunked PayloadJson, Events with 30k-char + /// Message + DataJson) exceed the 4 MiB cap and surface as an opaque 500 the agent + /// retries forever (audit 2026-09-05 F07). + /// + /// Sizes are estimates on the safe side (UTF-16 width for strings, base64 growth for + /// binaries, a fixed envelope per entity); the budget leaves headroom for the OData + /// batch framing. A batch never spans partitions: a PartitionKey change starts a new + /// transaction, so callers may pass a pre-grouped sequence and rely on the order. + /// + /// + internal static class TableTransactionBatcher + { + /// Azure Tables hard limit on actions per entity-group transaction. + public const int MaxActionsPerTransaction = 100; + + /// Conservative payload budget per transaction (service limit: 4 MiB). + public const int TransactionByteBudget = 3_500_000; + + /// Conservative size guard per entity (service limit: 1 MiB). + public const int EntityByteLimit = 1_000_000; + + private const int EntityEnvelopeBytes = 256; + private const int ScalarPropertyBytes = 48; + + /// + /// Groups into transactions of + /// actions. Preserves order; starts a new transaction on the action cap, the byte + /// budget or a PartitionKey change. Throws + /// for an entity that can never fit on its own — splitting cannot fix that, the caller + /// has to shrink or reject the row. + /// + public static List> Split( + IEnumerable entities, + TableTransactionActionType actionType, + int maxActions = MaxActionsPerTransaction, + int byteBudget = TransactionByteBudget) + { + if (entities == null) throw new ArgumentNullException(nameof(entities)); + if (maxActions < 1 || maxActions > MaxActionsPerTransaction) + throw new ArgumentOutOfRangeException(nameof(maxActions)); + if (byteBudget < 1) throw new ArgumentOutOfRangeException(nameof(byteBudget)); + + var batches = new List>(); + var current = new List(); + var currentBytes = 0; + string? currentPartition = null; + + foreach (var entity in entities) + { + var bytes = EstimateEntityBytes(entity); + if (bytes > EntityByteLimit) + throw new TableEntityTooLargeException(entity.PartitionKey, entity.RowKey, bytes); + + var partitionChanged = currentPartition != null + && !string.Equals(currentPartition, entity.PartitionKey, StringComparison.Ordinal); + if (current.Count > 0 + && (partitionChanged || current.Count >= maxActions || currentBytes + bytes > byteBudget)) + { + batches.Add(current); + current = new List(); + currentBytes = 0; + } + + current.Add(new TableTransactionAction(actionType, entity)); + currentBytes += bytes; + currentPartition = entity.PartitionKey; + } + + if (current.Count > 0) batches.Add(current); + return batches; + } + + /// + /// Upper-bound estimate of the serialized size of one entity: UTF-16 width for strings + /// (property names included), base64 growth for binaries, a fixed cost for every + /// scalar, plus the per-entity envelope. + /// + public static int EstimateEntityBytes(TableEntity entity) + { + if (entity == null) throw new ArgumentNullException(nameof(entity)); + + long total = EntityEnvelopeBytes + + 2L * (entity.PartitionKey?.Length ?? 0) + + 2L * (entity.RowKey?.Length ?? 0); + + foreach (var kv in entity) + { + // TableEntity enumerates the keys as properties too; they are counted above. + if (kv.Key == "PartitionKey" || kv.Key == "RowKey") continue; + + total += 2L * kv.Key.Length; + total += kv.Value switch + { + null => 0, + string s => 2L * s.Length, + byte[] b => (b.Length * 4L + 2) / 3, + BinaryData bd => (bd.ToMemory().Length * 4L + 2) / 3, + _ => ScalarPropertyBytes, + }; + } + + return total > int.MaxValue ? int.MaxValue : (int)total; + } + } + + /// + /// A single entity exceeds what Azure Tables accepts in one row. Permanent by nature: no + /// retry and no smaller batch can make it fit. The ingest maps it to HTTP 413 so the agent + /// isolates the item instead of replaying the batch. + /// + public sealed class TableEntityTooLargeException : Exception + { + public string PartitionKey { get; } + public string RowKey { get; } + public int EstimatedBytes { get; } + + public TableEntityTooLargeException(string partitionKey, string rowKey, int estimatedBytes) + : base($"Entity {partitionKey}/{rowKey} is about {estimatedBytes} bytes, above the {TableTransactionBatcher.EntityByteLimit}-byte entity limit") + { + PartitionKey = partitionKey; + RowKey = rowKey; + EstimatedBytes = estimatedBytes; + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs index 5705e638..82aea935 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs @@ -1,5 +1,7 @@ using System.IO; using System.Net; +using Azure; +using AutopilotMonitor.Functions.DataAccess.TableStorage; using AutopilotMonitor.Functions.Helpers; using AutopilotMonitor.Functions.Security; using AutopilotMonitor.Functions.Services; @@ -303,6 +305,24 @@ namespace AutopilotMonitor.Functions.Functions.Ingest SignalRMessages = outcome.SignalRMessages, }; } + catch (TableEntityTooLargeException ex) + { + // One item can never fit a table row. 413 makes the agent halve the batch and, + // once the item is alone, quarantine it — a 500 would be replayed forever. + _logger.LogWarning(ex, "IngestTelemetry: telemetry item {Partition}/{Row} exceeds the entity limit ({Bytes} bytes)", + ex.PartitionKey, ex.RowKey, ex.EstimatedBytes); + return AsOutput(await WriteErrorAsync(req, HttpStatusCode.RequestEntityTooLarge, + "A telemetry item exceeds the storage entity limit")); + } + catch (RequestFailedException ex) + { + var (status, message, retryAfter) = ClassifyStorageFailure(ex); + if (status == HttpStatusCode.InternalServerError) + _logger.LogError(ex, "IngestTelemetry: storage request failed with status {Status} ({Code})", ex.Status, ex.ErrorCode); + else + _logger.LogWarning(ex, "IngestTelemetry: storage request failed with status {Status} ({Code}) — answering {Http}", ex.Status, ex.ErrorCode, (int)status); + return AsOutput(await WriteErrorAsync(req, status, message, retryAfter)); + } catch (Exception ex) { _logger.LogError(ex, "IngestTelemetry: unhandled exception"); @@ -310,6 +330,24 @@ namespace AutopilotMonitor.Functions.Functions.Ingest } } + /// Seconds the agent should wait before replaying a batch after a storage outage. + internal const int StorageRetryAfterSeconds = 5; + + /// + /// Maps a storage failure onto the status the agent's uploader understands: 413 makes it + /// halve the batch, 503 (with Retry-After) makes it replay later; the agent treats + /// 500 as transient too but never shrinks on it, so an oversize batch surfacing as 500 + /// would wedge the device (audit 2026-09-05 F07). + /// + internal static (HttpStatusCode Status, string Message, int? RetryAfterSeconds) ClassifyStorageFailure(RequestFailedException ex) + { + if (StorageErrors.IsPayloadTooLarge(ex)) + return (HttpStatusCode.RequestEntityTooLarge, "Telemetry batch exceeds the storage transaction limit; retry with a smaller batch", null); + if (StorageErrors.IsTransient(ex)) + return (HttpStatusCode.ServiceUnavailable, "Storage temporarily unavailable; retry later", StorageRetryAfterSeconds); + return (HttpStatusCode.InternalServerError, "Internal server error", null); + } + private static IngestEventsOutput AsOutput(HttpResponseData response) => new IngestEventsOutput { @@ -645,9 +683,11 @@ namespace AutopilotMonitor.Functions.Functions.Ingest } private static async Task WriteErrorAsync( - HttpRequestData req, HttpStatusCode status, string message) + HttpRequestData req, HttpStatusCode status, string message, int? retryAfterSeconds = null) { var response = req.CreateResponse(status); + if (retryAfterSeconds is int seconds) + response.Headers.Add("Retry-After", seconds.ToString()); await response.WriteAsJsonAsync(new IngestEventsResponse { Success = false, diff --git a/src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs b/src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs new file mode 100644 index 00000000..96bd828c --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Helpers/StorageErrors.cs @@ -0,0 +1,49 @@ +using System; +using Azure; + +namespace AutopilotMonitor.Functions.Helpers; + +/// +/// One place that classifies from Azure Storage. Writers +/// decide "retry / split / give up" on these predicates and the ingest maps them to the HTTP +/// status the agent understands: 503 for transient (replay the batch later), 413 for payload +/// too large (halve the batch), anything else 500. The transient list mirrors +/// ' HTTP set so both stay visibly aligned. +/// +public static class StorageErrors +{ + private static readonly int[] TransientStatuses = [408, 429, 500, 502, 503, 504]; + + private static readonly string[] PayloadTooLargeCodes = + [ + "RequestBodyTooLarge", + "EntityTooLarge", + "PropertyValueTooLarge", + ]; + + /// Throttling, timeouts and service-side failures — a later retry can succeed. + public static bool IsTransient(RequestFailedException ex) + => ex != null && Array.IndexOf(TransientStatuses, ex.Status) >= 0; + + /// + /// The transaction or entity exceeds a size limit. HTTP 413 in production; some backends + /// report the size codes with a 400, so the error code is checked as well. + /// + public static bool IsPayloadTooLarge(RequestFailedException ex) + { + if (ex == null) return false; + if (ex.Status == 413) return true; + return ex.ErrorCode != null && Array.IndexOf(PayloadTooLargeCodes, ex.ErrorCode) >= 0; + } + + /// + /// A conditional Add hit an existing row. Production Azure Tables answers 409; the Azurite + /// emulator and older service versions answer 400 with the same error code. + /// + public static bool IsAlreadyExists(RequestFailedException ex) + { + if (ex == null) return false; + if (ex.Status == 409) return true; + return ex.Status == 400 && string.Equals(ex.ErrorCode, "EntityAlreadyExists", StringComparison.Ordinal); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Backup/CriticalTableRestoreService.cs b/src/Backend/AutopilotMonitor.Functions/Services/Backup/CriticalTableRestoreService.cs index 196558ad..d588559b 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Backup/CriticalTableRestoreService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Backup/CriticalTableRestoreService.cs @@ -12,6 +12,7 @@ using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models.Backup; using AutopilotMonitor.Shared.Models.Deletion; using Azure; +using AutopilotMonitor.Functions.Helpers; using Azure.Data.Tables; using Microsoft.Extensions.Logging; @@ -237,7 +238,7 @@ namespace AutopilotMonitor.Functions.Services.Backup { await tableClient.AddEntityAsync(entity, handlerCts.Token).ConfigureAwait(false); } - catch (RequestFailedException ex) when (IsAlreadyExistsStatus(ex)) + catch (RequestFailedException ex) when (StorageErrors.IsAlreadyExists(ex)) { throw new BackupTerminalException( "CurrentRowChanged", @@ -484,11 +485,5 @@ namespace AutopilotMonitor.Functions.Services.Backup return Convert.ToHexString(hash).ToLowerInvariant(); } - private static bool IsAlreadyExistsStatus(RequestFailedException ex) - { - if (ex.Status == 409) return true; - if (ex.Status == 400 && string.Equals(ex.ErrorCode, "EntityAlreadyExists", StringComparison.Ordinal)) return true; - return false; - } } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Restore.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Restore.cs index 97e0ff05..451b9e4a 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Restore.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Restore.cs @@ -7,6 +7,8 @@ using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Azure; +using AutopilotMonitor.Functions.DataAccess.TableStorage; +using AutopilotMonitor.Functions.Helpers; using Azure.Data.Tables; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models.Deletion; @@ -104,22 +106,16 @@ namespace AutopilotMonitor.Functions.Services foreach (var group in rows.GroupBy(r => r.Pk, StringComparer.Ordinal)) { - var rowsInGroup = group.ToList(); - for (var i = 0; i < rowsInGroup.Count; i += RestoreBatchActionLimit) + var entities = group.Select(dump => ConvertDumpToEntity(dump, tableName)).ToList(); + foreach (var actions in TableTransactionBatcher.Split(entities, TableTransactionActionType.Add)) { - var chunk = rowsInGroup.Skip(i).Take(RestoreBatchActionLimit).ToList(); - var actions = chunk - .Select(dump => new TableTransactionAction( - TableTransactionActionType.Add, - ConvertDumpToEntity(dump, tableName))) - .ToList(); - + var chunk = actions.Select(a => (TableEntity)a.Entity).ToList(); try { await tableClient.SubmitTransactionAsync(actions, cancellationToken); restored += chunk.Count; } - catch (RequestFailedException ex) when (IsAlreadyExistsStatus(ex)) + catch (RequestFailedException ex) when (StorageErrors.IsAlreadyExists(ex)) { // Azure rolls back the entire transaction when any Add hits an existing // row. Fall back to per-row AddEntity so we can distinguish "row already @@ -130,15 +126,15 @@ namespace AutopilotMonitor.Functions.Services // retries after partial-failed restore). The `mode` parameter is kept for // API readability but no longer changes behaviour here. _ = mode; // intentional: relaxation documented above - foreach (var dump in chunk) + foreach (var entity in chunk) { cancellationToken.ThrowIfCancellationRequested(); try { - await tableClient.AddEntityAsync(ConvertDumpToEntity(dump, tableName), cancellationToken); + await tableClient.AddEntityAsync(entity, cancellationToken); restored++; } - catch (RequestFailedException rfe) when (IsAlreadyExistsStatus(rfe)) + catch (RequestFailedException rfe) when (StorageErrors.IsAlreadyExists(rfe)) { // Mirror the batch-level relaxation: some Azure deployments and // the Azurite emulator surface duplicate-row conflict as HTTP 400 @@ -245,10 +241,6 @@ namespace AutopilotMonitor.Functions.Services return entity; } - // Azure Tables batch transactions cap at 100 actions per submission, all sharing a - // PartitionKey. Mirrors the const in the deletion partial. - private const int RestoreBatchActionLimit = 100; - /// /// Bulk-restore helper for the critical-table backup feature (plan §PR0). Symmetric to /// but uses @@ -275,37 +267,16 @@ namespace AutopilotMonitor.Functions.Services foreach (var group in rows.GroupBy(r => r.Pk, StringComparer.Ordinal)) { - var rowsInGroup = group.ToList(); - for (var i = 0; i < rowsInGroup.Count; i += RestoreBatchActionLimit) + var entities = group.Select(dump => ConvertDumpToEntity(dump, tableName)).ToList(); + foreach (var actions in TableTransactionBatcher.Split(entities, TableTransactionActionType.UpsertReplace)) { - var chunk = rowsInGroup.Skip(i).Take(RestoreBatchActionLimit).ToList(); - var actions = chunk - .Select(dump => new TableTransactionAction( - TableTransactionActionType.UpsertReplace, - ConvertDumpToEntity(dump, tableName))) - .ToList(); - await tableClient.SubmitTransactionAsync(actions, cancellationToken); - upserted += chunk.Count; + upserted += actions.Count; } } return upserted; } - /// - /// True when the Azure SDK exception encodes a "row already exists" outcome — the only - /// case where we treat the failed Add as Skipped during restore. Production Azure Tables - /// reports this as HTTP 409 Conflict (ErrorCode EntityAlreadyExists), but the - /// Azurite emulator and some older service versions report HTTP 400 Bad Request with the - /// same ErrorCode. Anything else propagates so we don't silently swallow genuine errors. - /// - private static bool IsAlreadyExistsStatus(RequestFailedException ex) - { - if (ex.Status == 409) return true; - if (ex.Status == 400 && string.Equals(ex.ErrorCode, "EntityAlreadyExists", StringComparison.Ordinal)) - return true; - return false; - } } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Rules.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Rules.cs index 72f0294b..0822ff73 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Rules.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Rules.cs @@ -1067,19 +1067,10 @@ namespace AutopilotMonitor.Functions.Services entitiesToUpsert[rowKey] = entity; } - // Batch write (max 100 per batch, all same PK) + // Batch write (byte-aware, all same PK) var upsertList = entitiesToUpsert.Values.ToList(); - for (int i = 0; i < upsertList.Count; i += 100) - { - var batch = upsertList.Skip(i).Take(100) - .Select(e => new TableTransactionAction(TableTransactionActionType.UpsertReplace, e)) - .ToList(); - - if (batch.Count > 0) - { - await tableClient.SubmitTransactionAsync(batch); - } - } + foreach (var batch in TableTransactionBatcher.Split(upsertList, TableTransactionActionType.UpsertReplace)) + await tableClient.SubmitTransactionAsync(batch); _logger.LogInformation("Upserted {Count} software inventory entries for tenant {TenantId}", upsertList.Count, tenantId); } @@ -1201,18 +1192,9 @@ namespace AutopilotMonitor.Functions.Services }); } - // Batch write (max 100 per batch, all same PK) - for (int i = 0; i < entities.Count; i += 100) - { - var batch = entities.Skip(i).Take(100) - .Select(e => new TableTransactionAction(TableTransactionActionType.UpsertReplace, e)) - .ToList(); - - if (batch.Count > 0) - { - await tableClient.SubmitTransactionAsync(batch); - } - } + // Batch write (byte-aware, all same PK) + foreach (var batch in TableTransactionBatcher.Split(entities, TableTransactionActionType.UpsertReplace)) + await tableClient.SubmitTransactionAsync(batch); _logger.LogInformation("Imported {Count} CPE seed mapping entries from GitHub JSON into VulnerabilityCache table", entities.Count); return entities.Count; @@ -1270,18 +1252,9 @@ namespace AutopilotMonitor.Functions.Services }); } - // Batch write (max 100 per batch, all same PK) - for (int i = 0; i < entities.Count; i += 100) - { - var batch = entities.Skip(i).Take(100) - .Select(e => new TableTransactionAction(TableTransactionActionType.UpsertReplace, e)) - .ToList(); - - if (batch.Count > 0) - { - await tableClient.SubmitTransactionAsync(batch); - } - } + // Batch write (byte-aware, all same PK) + foreach (var batch in TableTransactionBatcher.Split(entities, TableTransactionActionType.UpsertReplace)) + await tableClient.SubmitTransactionAsync(batch); _logger.LogInformation("Imported {Count} CPE community mapping entries from GitHub JSON into VulnerabilityCache table", entities.Count); return entities.Count; diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs index fc25b1bb..30075775 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs @@ -1,4 +1,5 @@ using Azure; +using AutopilotMonitor.Functions.DataAccess.TableStorage; using Azure.Data.Tables; using AutopilotMonitor.Functions.Helpers; using AutopilotMonitor.Functions.Pagination; @@ -428,10 +429,10 @@ namespace AutopilotMonitor.Functions.Services }; /// - /// Azure Table Storage limits string properties to 64KB (32K UTF-16 chars). - /// Truncate to 30,000 chars to leave buffer for multi-byte characters. + /// Azure Table Storage limits string properties to 64KB (32K UTF-16 chars). Events are + /// truncated rather than chunked; the bound is shared with . /// - private const int MaxTableStorageStringLength = 30000; + private const int MaxTableStorageStringLength = TableStorageChunking.MaxPropertyChars; private string TruncateForTableStorage(string value, string propertyName, string eventId) { @@ -847,6 +848,60 @@ namespace AutopilotMonitor.Functions.Services /// /// Stores an event /// + /// RowKey of an event row: device event time (ms) plus the agent sequence. + private static string BuildEventRowKey(EnrollmentEvent evt) + => $"{evt.Timestamp:yyyyMMddHHmmssfff}_{evt.Sequence:D10}"; + + /// + /// Projects an event onto its Events row. Shared by the single and the batched writer so + /// the two can never drift. PartitionKey {TenantId}_{SessionId}, RowKey + /// . + /// + private TableEntity BuildEventEntity(EnrollmentEvent evt) + { + var entity = new TableEntity($"{evt.TenantId}_{evt.SessionId}", BuildEventRowKey(evt)) + { + ["EventId"] = evt.EventId, + ["SessionId"] = evt.SessionId, + ["TenantId"] = evt.TenantId, + // "Timestamp" is a reserved system property — supplied values are ignored + // and any row rewrite (storage migration) resets it. The sanitized agent + // event time therefore lives in its own column. + [BusinessTimestamp.OccurredUtcColumn] = EnsureUtc(evt.Timestamp), + ["EventType"] = evt.EventType ?? string.Empty, + ["Severity"] = (int)evt.Severity, + ["Source"] = evt.Source ?? string.Empty, + ["Phase"] = (int)evt.Phase, + ["Message"] = TruncateForTableStorage(evt.Message ?? string.Empty, "Message", evt.EventId), + ["Sequence"] = evt.Sequence, + ["DataJson"] = TruncateForTableStorage( + evt.Data != null && evt.Data.Count > 0 + ? JsonConvert.SerializeObject(evt.Data) + : string.Empty, + "DataJson", evt.EventId), + ["ReceivedAt"] = evt.ReceivedAt, + ["TimestampClamped"] = evt.TimestampClamped + }; + + // P14: request-level device-clock send time; column stays unset for + // events from agents that pre-date the X-Send-Time-Utc header. + if (evt.SentAt.HasValue) + entity["SentAt"] = EnsureUtc(evt.SentAt.Value); + + if (evt.OriginalTimestamp.HasValue) + entity["OriginalTimestamp"] = EnsureUtc(evt.OriginalTimestamp.Value); + + // Codex follow-up #3: forward-link columns. Null when absent on the agent + // payload (pre-#3 events, or events emitted outside the reducer pipeline); + // skip the entity setter in that case so the column stays unset. + if (evt.CausedByTransitionStepIndex.HasValue) + entity["CausedByTransitionStepIndex"] = evt.CausedByTransitionStepIndex.Value; + if (evt.CausedBySignalOrdinal.HasValue) + entity["CausedBySignalOrdinal"] = evt.CausedBySignalOrdinal.Value; + + return entity; + } + public async Task StoreEventAsync(EnrollmentEvent evt) { SecurityValidator.EnsureValidGuid(evt.TenantId, "TenantId"); @@ -856,52 +911,7 @@ namespace AutopilotMonitor.Functions.Services { var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Events); - // PartitionKey: TenantId_SessionId for efficient querying - // RowKey: Timestamp_Sequence for ordering - var partitionKey = $"{evt.TenantId}_{evt.SessionId}"; - var rowKey = $"{evt.Timestamp:yyyyMMddHHmmssfff}_{evt.Sequence:D10}"; - - var entity = new TableEntity(partitionKey, rowKey) - { - ["EventId"] = evt.EventId, - ["SessionId"] = evt.SessionId, - ["TenantId"] = evt.TenantId, - // "Timestamp" is a reserved system property — supplied values are ignored - // and any row rewrite (storage migration) resets it. The sanitized agent - // event time therefore lives in its own column. - [BusinessTimestamp.OccurredUtcColumn] = EnsureUtc(evt.Timestamp), - ["EventType"] = evt.EventType ?? string.Empty, - ["Severity"] = (int)evt.Severity, - ["Source"] = evt.Source ?? string.Empty, - ["Phase"] = (int)evt.Phase, - ["Message"] = TruncateForTableStorage(evt.Message ?? string.Empty, "Message", evt.EventId), - ["Sequence"] = evt.Sequence, - ["DataJson"] = TruncateForTableStorage( - evt.Data != null && evt.Data.Count > 0 - ? JsonConvert.SerializeObject(evt.Data) - : string.Empty, - "DataJson", evt.EventId), - ["ReceivedAt"] = evt.ReceivedAt, - ["TimestampClamped"] = evt.TimestampClamped - }; - - // P14: request-level device-clock send time; column stays unset for - // events from agents that pre-date the X-Send-Time-Utc header. - if (evt.SentAt.HasValue) - entity["SentAt"] = EnsureUtc(evt.SentAt.Value); - - if (evt.OriginalTimestamp.HasValue) - entity["OriginalTimestamp"] = EnsureUtc(evt.OriginalTimestamp.Value); - - // Codex follow-up #3: forward-link columns. Null when absent on the agent - // payload (pre-#3 events, or events emitted outside the reducer pipeline); - // skip the entity setter in that case so the column stays unset. - if (evt.CausedByTransitionStepIndex.HasValue) - entity["CausedByTransitionStepIndex"] = evt.CausedByTransitionStepIndex.Value; - if (evt.CausedBySignalOrdinal.HasValue) - entity["CausedBySignalOrdinal"] = evt.CausedBySignalOrdinal.Value; - - await tableClient.UpsertEntityAsync(entity); + await tableClient.UpsertEntityAsync(BuildEventEntity(evt)); _logger.LogDebug($"Stored event {evt.EventId}"); return true; @@ -940,84 +950,55 @@ namespace AutopilotMonitor.Functions.Services foreach (var group in groups) { - // Chunk into batches of 100 (Azure Table Storage limit) - var chunks = group.Select((evt, index) => new { evt, index }) - .GroupBy(x => x.index / 100) - .Select(g => g.Select(x => x.evt).ToList()); + // Dedup by RowKey within the partition. The RowKey is {Timestamp:ms}_{Sequence:D10}; + // two events sharing the same millisecond AND Sequence map to one row and would + // collapse under UpsertReplace anyway — but Azure rejects the WHOLE transaction + // with InvalidDuplicateRow if both appear in one batch. Keeping the last + // occurrence (matches UpsertReplace last-wins) lets the batch succeed cleanly. + var deduped = group + .GroupBy(BuildEventRowKey) + .Select(g => g.Last()) + .ToList(); - foreach (var rawChunk in chunks) + // Byte-aware batching: 100 actions, 3.5 MB per transaction, 1 MB per entity. + // An entity that can never fit throws TableEntityTooLargeException — permanent, + // the ingest maps it to 413 so the agent isolates the item. + var eventsByEntity = new Dictionary(ReferenceEqualityComparer.Instance); + foreach (var evt in deduped) + eventsByEntity[BuildEventEntity(evt)] = evt; + + foreach (var actions in TableTransactionBatcher.Split(eventsByEntity.Keys, TableTransactionActionType.UpsertReplace)) { - // Dedup by RowKey within the chunk. The RowKey is {Timestamp:ms}_{Sequence:D10}; - // two events sharing the same millisecond AND Sequence map to one row and would - // collapse under UpsertReplace anyway — but Azure rejects the WHOLE transaction - // with InvalidDuplicateRow if both appear in one batch (then we degrade to slow - // per-event writes and track a noisy TableTransactionFailedException). Keeping the - // last occurrence (matches UpsertReplace last-wins) lets the batch succeed cleanly. - var chunk = rawChunk - .GroupBy(evt => $"{evt.Timestamp:yyyyMMddHHmmssfff}_{evt.Sequence:D10}") - .Select(g => g.Last()) - .ToList(); - + var chunk = actions.Select(a => eventsByEntity[(TableEntity)a.Entity]).ToList(); try { - var actions = chunk.Select(evt => - { - var partitionKey = $"{evt.TenantId}_{evt.SessionId}"; - var rowKey = $"{evt.Timestamp:yyyyMMddHHmmssfff}_{evt.Sequence:D10}"; - - var entity = new TableEntity(partitionKey, rowKey) - { - ["EventId"] = evt.EventId, - ["SessionId"] = evt.SessionId, - ["TenantId"] = evt.TenantId, - // Reserved-name caveat: see StoreEventAsync. - [BusinessTimestamp.OccurredUtcColumn] = EnsureUtc(evt.Timestamp), - ["EventType"] = evt.EventType ?? string.Empty, - ["Severity"] = (int)evt.Severity, - ["Source"] = evt.Source ?? string.Empty, - ["Phase"] = (int)evt.Phase, - ["Message"] = TruncateForTableStorage(evt.Message ?? string.Empty, "Message", evt.EventId), - ["Sequence"] = evt.Sequence, - ["DataJson"] = TruncateForTableStorage( - evt.Data != null && evt.Data.Count > 0 - ? JsonConvert.SerializeObject(evt.Data) - : string.Empty, - "DataJson", evt.EventId), - ["ReceivedAt"] = evt.ReceivedAt, - ["TimestampClamped"] = evt.TimestampClamped - }; - - // P14: see StoreEventAsync. - if (evt.SentAt.HasValue) - entity["SentAt"] = EnsureUtc(evt.SentAt.Value); - - if (evt.OriginalTimestamp.HasValue) - entity["OriginalTimestamp"] = EnsureUtc(evt.OriginalTimestamp.Value); - - // Codex follow-up #3: forward-link columns (see StoreEventAsync). - if (evt.CausedByTransitionStepIndex.HasValue) - entity["CausedByTransitionStepIndex"] = evt.CausedByTransitionStepIndex.Value; - if (evt.CausedBySignalOrdinal.HasValue) - entity["CausedBySignalOrdinal"] = evt.CausedBySignalOrdinal.Value; - - return new TableTransactionAction(TableTransactionActionType.UpsertReplace, entity); - }).ToList(); - await tableClient.SubmitTransactionAsync(actions); storedEvents.AddRange(chunk); _logger.LogDebug($"Batch stored {chunk.Count} events for partition {group.Key}"); } - catch (Exception ex) + catch (RequestFailedException ex) when (StorageErrors.IsTransient(ex) || StorageErrors.IsPayloadTooLarge(ex)) { - // Batch failed - fall back to individual writes for this chunk - _logger.LogWarning(ex, $"Batch write failed for {chunk.Count} events, falling back to individual writes"); + // Throttling / outage / oversize: per-row retries would only multiply the + // pressure (100 upserts × SDK retries against a partition that just said + // no). Propagate; the ingest answers 503 or 413 and the agent replays the + // batch — every row is an idempotent UpsertReplace. + _logger.LogWarning(ex, + "Batch write of {Count} events for {Partition} failed with status {Status} ({Code}) — propagating, no per-row fallback", + chunk.Count, group.Key, ex.Status, ex.ErrorCode); + throw; + } + catch (RequestFailedException ex) + { + // Data-shaped rejection (one bad row spoils the transaction): fall back to + // per-row writes so the good rows land and the bad one is isolated. + _logger.LogWarning(ex, + "Batch write of {Count} events for {Partition} rejected with status {Status} ({Code}) — falling back to individual writes", + chunk.Count, group.Key, ex.Status, ex.ErrorCode); foreach (var evt in chunk) { if (await StoreEventAsync(evt)) - { storedEvents.Add(evt); - } } } }