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);
- }
}
}
}