From ffb09e60fab127684828ab18902bff5c09c4ea38 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Mon, 28 Sep 2026 20:47:45 +0200 Subject: [PATCH] Global audit and ops-event pages continue from one RowKey watermark The cross-partition pages (audit log over all tenants, ops events over all categories) kept one continuation entry per partition that still had rows. On the tenant fan-out the nextLink outgrew the host's URL limit: the follow-up request was rejected with HTTP 414 before the function ran, so Next never loaded a second page for a Global Admin. The continuation is now the (RowKey, Partition) of the last returned row. Every page reads each partition through BoundedFanOut (pageSize + 1 rows after the watermark) and merges by (RowKey, Partition), which is the display order in both tables. Partitions after the watermark's partition continue with `RowKey ge`, the others with `RowKey gt`, so equal ops RowKeys across categories survive a page cut. The merge no longer sorts by the resolved OccurredUtc while advancing by RowKey, which could skip rows where the two disagree. The fan-out was unbounded (Task.WhenAll over every tenant); it now runs with the shared cross-tenant concurrency ceiling like the session list. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../AuditLogDeletionExclusionTests.cs | 6 +- .../AuditLogFieldFilterTests.cs | 2 +- .../BusinessTimestampFilterClauseTests.cs | 2 +- .../OpsEventFieldFilterTests.cs | 3 +- .../PerPartitionFanOutMergeTests.cs | 514 ++++++++---------- .../TableStorage/PerPartitionFanOutMerge.cs | 239 ++++---- .../TableStorage/TableOpsEventRepository.cs | 57 +- .../TableStorageService.Maintenance.cs | 72 +-- 8 files changed, 396 insertions(+), 499 deletions(-) diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogDeletionExclusionTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogDeletionExclusionTests.cs index 53473da8..6562e13b 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogDeletionExclusionTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogDeletionExclusionTests.cs @@ -31,9 +31,9 @@ public class AuditLogDeletionExclusionTests public void Global_fanout_filter_includes_deletion_exclusion_only_when_requested() { var without = TableStorageService.BuildAuditLogFilterWithRowKeyBound( - Tenant.ToString(), null, null, lastRowKey: null, excludeDeletions: false); + Tenant.ToString(), null, null, rowKeyBound: null, excludeDeletions: false); var with = TableStorageService.BuildAuditLogFilterWithRowKeyBound( - Tenant.ToString(), null, null, lastRowKey: null, excludeDeletions: true); + Tenant.ToString(), null, null, rowKeyBound: null, excludeDeletions: true); Assert.DoesNotContain("deletion_started", without); Assert.Contains("Action ne 'deletion_started'", with); @@ -47,7 +47,7 @@ public class AuditLogDeletionExclusionTests // noise leak back into one of the two views. var tenant = TableStorageService.BuildAuditLogFilter(Tenant.ToString(), null, null, excludeDeletions: true); var global = TableStorageService.BuildAuditLogFilterWithRowKeyBound( - Tenant.ToString(), null, null, lastRowKey: null, excludeDeletions: true); + Tenant.ToString(), null, null, rowKeyBound: null, excludeDeletions: true); var clause = TableStorageService.DeletionExclusionClause(); Assert.Contains(clause, tenant); diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogFieldFilterTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogFieldFilterTests.cs index 0ce96bba..7ab733cd 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogFieldFilterTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/AuditLogFieldFilterTests.cs @@ -51,7 +51,7 @@ public class AuditLogFieldFilterTests var filters = new AuditLogQueryFilters { Action = "device_blocked", EntityType = "Device" }; var f = TableStorageService.BuildAuditLogFilterWithRowKeyBound( - Tenant, null, null, lastRowKey: null, excludeDeletions: false, filters); + Tenant, null, null, rowKeyBound: null, excludeDeletions: false, filters); Assert.Contains("Action eq 'device_blocked'", f); Assert.Contains("EntityType eq 'Device'", f); diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/BusinessTimestampFilterClauseTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/BusinessTimestampFilterClauseTests.cs index 483ba0d5..06be0c98 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/BusinessTimestampFilterClauseTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/BusinessTimestampFilterClauseTests.cs @@ -152,7 +152,7 @@ public class BusinessTimestampFilterClauseTests var plain = TableStorageService.BuildAuditLogFilter(Tenant, from, to); var fanOut = TableStorageService.BuildAuditLogFilterWithRowKeyBound( - Tenant, from, to, lastRowKey: "!0123_x", excludeDeletions: false); + Tenant, from, to, rowKeyBound: new PerPartitionFanOutMerge.RowKeyBound("!0123_x", Inclusive: false), excludeDeletions: false); foreach (var f in new[] { plain!, fanOut }) { diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/OpsEventFieldFilterTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/OpsEventFieldFilterTests.cs index 73ed4108..34ba66b9 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/OpsEventFieldFilterTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/OpsEventFieldFilterTests.cs @@ -175,7 +175,8 @@ public class OpsEventFieldFilterTests // unfiltered rows — the exact bug this test exists to prevent. var filters = new OpsEventQueryFilters { EventType = "AgentEmergencyBreak", MinSeverity = "Error" }; - var fanOut = TableOpsEventRepository.BuildFilterWithRowKeyBound("Agent", null, null, "0009", filters); + var fanOut = TableOpsEventRepository.BuildFilterWithRowKeyBound( + "Agent", null, null, new PerPartitionFanOutMerge.RowKeyBound("0009", Inclusive: false), filters); Assert.Contains("EventType eq 'AgentEmergencyBreak'", fanOut); Assert.Contains("(Severity eq 'Error' or Severity eq 'Critical')", fanOut); diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/PerPartitionFanOutMergeTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/PerPartitionFanOutMergeTests.cs index e01b8a91..47e130b1 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/PerPartitionFanOutMergeTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/PerPartitionFanOutMergeTests.cs @@ -1,325 +1,283 @@ using AutopilotMonitor.Functions.DataAccess.TableStorage; -using AutopilotMonitor.Shared.DataAccess; using static AutopilotMonitor.Functions.DataAccess.TableStorage.PerPartitionFanOutMerge; namespace AutopilotMonitor.Functions.Tests; /// -/// Cross-partition merge used by both the global all-category ops-events page -/// and the global cross-tenant audit-logs page. Without this fan-out, Azure -/// pages cross-partition queries by (PK asc, RK asc) — so page 1 would be -/// entirely the alphabetically-first partition, masking newer items from -/// later-alphabet partitions. The merge picks top-pageSize by timestamp across -/// all per-partition fetches. +/// Cross-partition paging behind the global all-category ops-events page and the global +/// cross-tenant audit-log page. Every page fans out one read per partition and merges by +/// (RowKey, Partition); the continuation is the watermark of the last returned row. The tests +/// drive the helper against an in-memory partition store that honours the RowKey bound exactly +/// like the OData clause does. /// public class PerPartitionFanOutMergeTests { - private static OpsEventEntry Entry(string partition, DateTime ts, string id) - => new() { Category = partition, Id = id, Timestamp = ts, EventType = "synthetic" }; - - private static (string RowKey, OpsEventEntry Item) Row(DateTime ts, string id) + // Reverse-tick RowKey like the storage layer writes it: a smaller string is a newer row. + private static string Rk(int minutesAgo, string suffix = "") { - // Mimics the storage-layer revtick rowkey: smaller string = newer timestamp. - var rk = $"{(DateTime.MaxValue.Ticks - ts.Ticks):D19}_{id}"; - return (rk, Entry(partition: "", ts, id)); + var t0 = new DateTime(2026, 9, 28, 12, 0, 0, DateTimeKind.Utc); + return $"{DateTime.MaxValue.Ticks - t0.AddMinutes(-minutesAgo).Ticks:D19}{suffix}"; } - private static PartitionFetchResult Fetched( - string partition, params (string RowKey, OpsEventEntry Item)[] rows) + private sealed class Store { - var aligned = rows - .Select(r => (r.RowKey, Item: new OpsEventEntry + private readonly Dictionary> _rows; + private readonly Func _matches; + public List<(string Partition, RowKeyBound? Bound, int Limit)> Calls { get; } = new(); + + public Store(Dictionary> rows, Func? matches = null) + { + _rows = rows; + _matches = matches ?? (_ => true); + } + + public IEnumerable Partitions => _rows.Keys; + + public PartitionReader Reader => (partition, bound, limit, _) => + { + lock (Calls) Calls.Add((partition, bound, limit)); + var read = _rows[partition] + .Where(_matches) + .Where(rk => bound == null + || (bound.Inclusive + ? string.CompareOrdinal(rk, bound.RowKey) >= 0 + : string.CompareOrdinal(rk, bound.RowKey) > 0)) + .OrderBy(rk => rk, StringComparer.Ordinal) + .Take(limit) + .Select(rk => (rk, $"{partition}/{rk}")) + .ToList(); + return Task.FromResult(read); + }; + + /// The full newest-first order the pages must reproduce. + public List Expected() => _rows + .SelectMany(kv => kv.Value.Where(_matches).Select(rk => (Partition: kv.Key, RowKey: rk))) + .OrderBy(r => r.RowKey, StringComparer.Ordinal) + .ThenBy(r => r.Partition, StringComparer.Ordinal) + .Select(r => $"{r.Partition}/{r.RowKey}") + .ToList(); + } + + private static async Task>> WalkAsync(Store store, int pageSize) + { + var pages = new List>(); + string? continuation = null; + do + { + var page = await FetchPageAsync(store.Partitions, pageSize, continuation, store.Reader); + pages.Add(page.Items.ToList()); + continuation = page.NextRawToken; + if (pages.Count > 1000) Assert.Fail("Walk did not terminate"); + } + while (continuation != null); + return pages; + } + + // ────────── Order ────────────────────────────────────────────────────── + + [Fact] + public async Task First_page_is_newest_across_partitions_not_alphabetically_first() + { + var store = new Store(new() + { + ["Agent"] = new() { Rk(5) }, + ["Security"] = new() { Rk(0) }, + ["Tenant"] = new() { Rk(10) }, + }); + + var page = await FetchPageAsync(store.Partitions, 3, null, store.Reader); + + Assert.Equal(new[] { $"Security/{Rk(0)}", $"Agent/{Rk(5)}", $"Tenant/{Rk(10)}" }, page.Items); + Assert.Null(page.NextRawToken); + } + + [Fact] + public async Task First_page_reads_every_partition_unbounded_with_one_row_of_lookahead() + { + var store = new Store(new() { ["A"] = new() { Rk(1) }, ["B"] = new(), ["C"] = new() { Rk(2) } }); + + await FetchPageAsync(store.Partitions, 15, null, store.Reader); + + Assert.Equal(new[] { "A", "B", "C" }, store.Calls.Select(c => c.Partition).OrderBy(p => p)); + Assert.All(store.Calls, c => Assert.Null(c.Bound)); + Assert.All(store.Calls, c => Assert.Equal(16, c.Limit)); + } + + // ────────── Watermark and bounds ─────────────────────────────────────── + + [Fact] + public void Bound_is_exclusive_up_to_the_watermark_partition_and_inclusive_after_it() + { + var w = new Watermark("0000000000000000042", "Maintenance"); + + Assert.Null(BoundAfter(null, "Agent")); + Assert.Equal(new RowKeyBound(w.RowKey, Inclusive: false), BoundAfter(w, "Agent")); + Assert.Equal(new RowKeyBound(w.RowKey, Inclusive: false), BoundAfter(w, "Maintenance")); + Assert.Equal(new RowKeyBound(w.RowKey, Inclusive: true), BoundAfter(w, "Security")); + } + + [Fact] + public void Bound_renders_as_an_escaped_odata_rowkey_clause() + { + Assert.Equal("RowKey gt '!0042_a'", new RowKeyBound("!0042_a", false).ToODataClause()); + Assert.Equal("RowKey ge 'it''s'", new RowKeyBound("it's", true).ToODataClause()); + } + + [Fact] + public async Task Equal_rowkeys_in_different_partitions_all_survive_a_page_boundary() + { + // Ops-event RowKeys are bare reverse ticks, so two categories can share one. With a + // RowKey-only watermark the second row would be skipped after the cut. + var shared = Rk(3); + var store = new Store(new() + { + ["Agent"] = new() { shared }, + ["Security"] = new() { shared }, + ["Tenant"] = new() { shared, Rk(4) }, + }); + + var pages = await WalkAsync(store, pageSize: 1); + + Assert.Equal(store.Expected(), pages.SelectMany(p => p)); + Assert.Equal(4, pages.Count); + } + + // ────────── Exact end of list ────────────────────────────────────────── + + [Fact] + public async Task No_continuation_when_the_rows_exactly_fill_the_page() + { + var store = new Store(new() { ["A"] = new() { Rk(1), Rk(2), Rk(3) }, ["B"] = new() }); + + var page = await FetchPageAsync(store.Partitions, 3, null, store.Reader); + + Assert.Equal(3, page.Items.Count); + Assert.Null(page.NextRawToken); + } + + [Fact] + public async Task Continuation_when_one_row_is_left_over() + { + var store = new Store(new() { ["A"] = new() { Rk(1), Rk(2), Rk(3), Rk(4) } }); + + var page = await FetchPageAsync(store.Partitions, 3, null, store.Reader); + + Assert.Equal(3, page.Items.Count); + Assert.NotNull(page.NextRawToken); + } + + // ────────── Full walks ───────────────────────────────────────────────── + + [Theory] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + [InlineData(5)] + [InlineData(15)] + [InlineData(100)] + public async Task Walk_returns_every_row_once_in_global_order_and_never_an_empty_page(int pageSize) + { + // Uneven partitions: one empty, one dominant, interleaved timestamps and a tie. + var store = new Store(new() + { + ["00000000-0000-0000-0000-000000000000"] = new() { Rk(0, "_g0"), Rk(40, "_g1") }, + ["1a"] = new() { Rk(1, "_a"), Rk(2, "_a"), Rk(3, "_a"), Rk(4, "_a"), Rk(5, "_a"), Rk(6, "_a"), Rk(30, "_a") }, + ["2b"] = new(), + ["3c"] = new() { Rk(2, "_c"), Rk(7, "_c"), Rk(8, "_c") }, + ["4d"] = new() { Rk(9, "_a") }, + ["5e"] = new() { Rk(9, "_a"), Rk(50, "_e") }, + }); + + var pages = await WalkAsync(store, pageSize); + + var expected = store.Expected(); + Assert.Equal(expected, pages.SelectMany(p => p)); + Assert.All(pages, p => Assert.NotEmpty(p)); + Assert.All(pages.Take(pages.Count - 1), p => Assert.Equal(pageSize, p.Count)); + } + + [Fact] + public async Task Walk_with_a_server_side_filter_returns_exactly_the_matching_rows() + { + // The reader applies the filter (like `Action ne 'deletion_started'`), so excluded rows + // never take a page slot and never become a watermark. + var store = new Store( + new() { - Id = r.Item.Id, - Category = partition, - EventType = r.Item.EventType, - Timestamp = r.Item.Timestamp, - })) - .ToArray(); - return new PartitionFetchResult(partition, aligned); - } + ["A"] = new() { Rk(1, "_keep"), Rk(2, "_drop"), Rk(3, "_drop"), Rk(4, "_keep") }, + ["B"] = new() { Rk(2, "_drop"), Rk(5, "_keep"), Rk(6, "_drop") }, + }, + matches: rk => rk.EndsWith("_keep", StringComparison.Ordinal)); - // ────────── Merge picks newest-across-partitions, not within-partition ── + var pages = await WalkAsync(store, pageSize: 2); - [Fact] - public void First_page_returns_newest_across_partitions_not_alphabetically_first() - { - // Without the merge, Azure's (PK asc, RK asc) ordering would surface - // "Agent" partition items first (alphabetically), even though "Security" - // has a newer event. The merge must pick by timestamp, not partition. - var t0 = DateTime.UtcNow; - var fetched = new[] - { - Fetched("Agent", Row(t0.AddMinutes(-5), "agent-1")), - Fetched("Security", Row(t0, "sec-1")), - Fetched("Tenant", Row(t0.AddMinutes(-10), "tenant-1")), - }; - - var (items, _) = MergeAndAdvance( - fetched, - new Dictionary(), - pageSize: 3, - timestampSelector: e => e.Timestamp); - - Assert.Equal(3, items.Count); - Assert.Equal("sec-1", items[0].Id); // newest first - Assert.Equal("agent-1", items[1].Id); - Assert.Equal("tenant-1", items[2].Id); + Assert.Equal(store.Expected(), pages.SelectMany(p => p)); + Assert.Equal(3, store.Expected().Count); } [Fact] - public void Empty_fetch_drops_partition_from_continuation_map() + public async Task Empty_store_yields_one_empty_page_without_continuation() { - // Wire-format v2 invariant: exhausted partitions are absent from the - // continuation, not encoded with `x:true`. Consumers treat "missing" - // as exhausted on subsequent pages — that's what shrinks the audit - // fan-out token from 30+ KB to a few hundred bytes. - var t0 = DateTime.UtcNow; - var fetched = new[] - { - Fetched("Agent"), // empty fetch - Fetched("Security", Row(t0, "sec-1")), - }; + var store = new Store(new() { ["A"] = new(), ["B"] = new() }); - var (_, conts) = MergeAndAdvance( - fetched, - new Dictionary(), - pageSize: 5, - timestampSelector: e => e.Timestamp); + var page = await FetchPageAsync(store.Partitions, 15, null, store.Reader); - Assert.False(conts.ContainsKey("Agent")); - Assert.True(conts.ContainsKey("Security")); + Assert.Empty(page.Items); + Assert.Null(page.NextRawToken); } - // ────────── Continuation advance: avoid losing leftovers when a partition loses the cut ── + // ────────── Token ────────────────────────────────────────────────────── [Fact] - public void Continuation_keeps_position_for_partitions_whose_items_lost_the_merge_cut() + public async Task Continuation_size_does_not_grow_with_the_number_of_partitions() { - // Page 1: pageSize=3. Agent fetched 5 items, Security fetched 5 items, but - // Security's are all newer. Merge takes 3 from Security; Agent's 5 items - // all lost the cut. Agent's continuation MUST stay at its previous - // position so the items get re-fetched and compete on page 2. - var t0 = DateTime.UtcNow; - var fetched = new[] - { - Fetched("Agent", - Row(t0.AddMinutes(-100), "a1"), - Row(t0.AddMinutes(-101), "a2"), - Row(t0.AddMinutes(-102), "a3"), - Row(t0.AddMinutes(-103), "a4"), - Row(t0.AddMinutes(-104), "a5")), - Fetched("Security", - Row(t0, "s1"), - Row(t0.AddMinutes(-1), "s2"), - Row(t0.AddMinutes(-2), "s3"), - Row(t0.AddMinutes(-3), "s4"), - Row(t0.AddMinutes(-4), "s5")), - }; + // The per-partition map this replaced grew with every partition holding rows until the + // nextLink failed with HTTP 414. Every partition here has rows that lose the merge cut. + static Store Tenants(int count) => new(Enumerable.Range(0, count).ToDictionary( + i => Guid.Parse($"00000000-0000-0000-0000-{i:D12}").ToString(), + i => new List { Rk(i, "_" + Guid.Empty.ToString("N")), Rk(i + 1000, "_" + Guid.Empty.ToString("N")) })); - var (items, conts) = MergeAndAdvance( - fetched, - new Dictionary(), - pageSize: 3, - timestampSelector: e => e.Timestamp); + var few = await FetchPageAsync(Tenants(3).Partitions, 1, null, Tenants(3).Reader); + var many = await FetchPageAsync(Tenants(500).Partitions, 1, null, Tenants(500).Reader); - // Security wins all three slots. - Assert.Equal(new[] { "s1", "s2", "s3" }, items.Select(i => i.Id).ToArray()); - - // Agent had no items returned and a non-empty fetch → keep prior continuation - // (null = restart). The partition stays in the map (active with no rk), - // so the next page will re-fetch its items and they'll be re-considered - // at a lower floor. - Assert.True(conts.ContainsKey("Agent")); - Assert.Null(conts["Agent"].LastRowKey); - - Assert.True(conts.ContainsKey("Security")); - Assert.NotNull(conts["Security"].LastRowKey); + Assert.NotNull(many.NextRawToken); + Assert.Equal(few.NextRawToken!.Length, many.NextRawToken!.Length); + Assert.True(many.NextRawToken.Length < 200, $"raw continuation is {many.NextRawToken.Length} characters"); } [Fact] - public void Continuation_advances_to_oldest_returned_rowkey_per_partition() + public void Watermark_round_trips() { - // When merge takes some-but-not-all from a partition, advance only past the - // items we actually returned. The leftover (older) items in that fetch - // will be re-fetched and re-considered on the next page. - var t0 = DateTime.UtcNow; - var (rkNew, _) = Row(t0, "x-new"); - var (rkMid, _) = Row(t0.AddMinutes(-5), "x-mid"); - var (rkOld, _) = Row(t0.AddMinutes(-10), "x-old"); - var fetched = new[] - { - Fetched("Agent", - (rkNew, Entry("Agent", t0, "x-new")), - (rkMid, Entry("Agent", t0.AddMinutes(-5), "x-mid")), - (rkOld, Entry("Agent", t0.AddMinutes(-10), "x-old"))), - // A second partition whose only event is between Agent's mid and old - // items, ensuring the merge ends after Agent's mid item. - Fetched("Security", - Row(t0.AddMinutes(-7), "s-only")), - }; + var w = new Watermark("!2516116980007133310_a4bb51498cad468f89431b15f1e701f7", "tenant-a"); - var (items, conts) = MergeAndAdvance( - fetched, - new Dictionary(), - pageSize: 3, - timestampSelector: e => e.Timestamp); - - // Top 3 by timestamp desc: x-new, x-mid, s-only. x-old is dropped. - Assert.Equal(new[] { "x-new", "x-mid", "s-only" }, items.Select(i => i.Id).ToArray()); - - // Agent's continuation advances to rkMid (the oldest Agent rowkey we returned). - // x-old will be re-fetched and re-considered next page (filter: RowKey gt rkMid). - Assert.Equal(rkMid, conts["Agent"].LastRowKey); - } - - // ────────── Codec round-trip ──────────────────────────────────────────── - - [Fact] - public void Encode_then_decode_round_trips_active_partitions() - { - var original = new Dictionary - { - ["Agent"] = new PartitionContinuation("0123456789012345678_aaa"), - ["Consent"] = new PartitionContinuation("0987654321098765432_bbb"), - ["Tenant"] = new PartitionContinuation(null), // active, restart from top - }; - - var encoded = EncodeMultiContinuation(original); - var decoded = DecodeMultiContinuation(encoded); - - Assert.Equal(3, decoded.Count); - Assert.Equal("0123456789012345678_aaa", decoded["Agent"].LastRowKey); - Assert.Equal("0987654321098765432_bbb", decoded["Consent"].LastRowKey); - Assert.Null(decoded["Tenant"].LastRowKey); + Assert.Equal(w, DecodeWatermark(EncodeWatermark(w))); } [Fact] - public void Decode_silently_drops_legacy_v1_exhausted_entries() + public void Decode_rejects_anything_that_is_not_a_v3_watermark() { - // Wire format v1 (pre-fix) explicitly encoded `x:true` for exhausted - // partitions; v2 just omits them. Tokens that were already in flight - // when the deploy went out must continue to page through cleanly — - // those legacy `x:true` entries are dropped silently so the - // consumer's "missing = exhausted" rule applies. - var legacyJson = """{"v":1,"c":{"Agent":{"rk":"0123456789012345678_aaa","x":false},"Security":{"rk":null,"x":true},"Consent":{"rk":"0987654321098765432_bbb","x":false}}}"""; - var legacyToken = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(legacyJson)); + static string B64(string json) => Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(json)); - var decoded = DecodeMultiContinuation(legacyToken); - - Assert.Equal(2, decoded.Count); - Assert.True(decoded.ContainsKey("Agent")); - Assert.True(decoded.ContainsKey("Consent")); - Assert.False(decoded.ContainsKey("Security")); // dropped: was x:true in v1 wire + Assert.Null(DecodeWatermark(null)); + Assert.Null(DecodeWatermark("")); + Assert.Null(DecodeWatermark("not-base64-!@#")); + Assert.Null(DecodeWatermark(B64("this is not json"))); + Assert.Null(DecodeWatermark(B64("""{"v":2,"c":{"Agent":{"rk":"0123"}}}"""))); + Assert.Null(DecodeWatermark(B64("""{"v":3,"rk":"","p":"Agent"}"""))); + Assert.Null(DecodeWatermark(B64("""{"v":3,"rk":"0123"}"""))); + Assert.Null(DecodeWatermark(B64("""{"v":"3","rk":"0123","p":"Agent"}"""))); } [Fact] - public void Encode_omits_partition_state_field_to_keep_token_compact() + public async Task A_token_from_an_older_deploy_ends_the_list_without_reading_storage() { - // Sanity-check the wire compactness gain: 200 partitions of v1 with - // `x:false` per entry produced ~30 KB after base64; v2 should be - // dramatically smaller because we drop the redundant flag and only - // encode active entries. - var many = Enumerable.Range(0, 200) - .ToDictionary(i => $"tenant-{i:D3}", i => new PartitionContinuation($"!1234567890123456789_{i:D8}")); + var store = new Store(new() { ["Agent"] = new() { Rk(1) } }); + var v2 = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes("""{"v":2,"c":{"Agent":{"rk":null}}}""")); - var encoded = EncodeMultiContinuation(many); + var page = await FetchPageAsync(store.Partitions, 15, v2, store.Reader); - // 200 entries × (~30 chars name + ~30 chars rk + JSON overhead) → JSON - // ~12 KB → base64 ~16 KB. The legacy `x:false` flag adds another - // ~10 bytes per entry (~2 KB total) which we now save. The hard - // assertion below is on absolute size: we must stay well below the - // ~30 KB that broke audit Next clicks. - Assert.True(encoded.Length < 20_000, - $"Encoded continuation should stay under 20 KB even at 200 partitions; got {encoded.Length} bytes"); - } - - [Fact] - public void Decode_returns_empty_for_malformed_or_empty_continuation() - { - Assert.Empty(DecodeMultiContinuation(null)); - Assert.Empty(DecodeMultiContinuation("")); - Assert.Empty(DecodeMultiContinuation("not-base64-!@#")); - Assert.Empty(DecodeMultiContinuation("dGhpcyBpcyBub3QganNvbg==")); // valid base64, not JSON - } - - // ────────── End-to-end: drain across pages preserves all items, in order ── - - [Fact] - public void Multi_page_drain_returns_all_items_globally_newest_first_no_duplicates() - { - var t0 = new DateTime(2026, 5, 5, 12, 0, 0, DateTimeKind.Utc); - - var allItems = new List<(string Part, OpsEventEntry Item, string Rk)>(); - var partitions = new[] { "Agent", "Security", "Tenant" }; - var minute = 0; - foreach (var part in partitions) - { - for (int i = 0; i < 4; i++) - { - minute++; - var ts = t0.AddMinutes(-minute); - var (rk, _) = Row(ts, $"{part}-{i}"); - allItems.Add((part, new OpsEventEntry - { - Category = part, - Id = $"{part}-{i}", - Timestamp = ts, - EventType = "synthetic", - }, rk)); - } - } - - IReadOnlyList> SimulateFetch( - Dictionary continuations, int pageSize, bool isFirstPage) - { - var fetched = new List>(); - foreach (var part in partitions) - { - // Mirrors the consumer rule: on first page query everything; on - // subsequent pages only partitions still in the continuation. - if (!isFirstPage && !continuations.ContainsKey(part)) continue; - continuations.TryGetValue(part, out var prior); - var lastRk = prior?.LastRowKey; - - var rows = allItems - .Where(t => t.Part == part) - .Where(t => lastRk == null || string.Compare(t.Rk, lastRk, StringComparison.Ordinal) > 0) - .OrderBy(t => t.Rk, StringComparer.Ordinal) - .Take(pageSize) - .Select(t => (t.Rk, t.Item)) - .ToList(); - fetched.Add(new PartitionFetchResult(part, rows)); - } - return fetched; - } - - const int pageSize = 5; - var seen = new List(); - var conts = new Dictionary(); - var loopGuard = 0; - bool isFirstPage = true; - - while (true) - { - var fetched = SimulateFetch(conts, pageSize, isFirstPage); - var (page, nextConts) = MergeAndAdvance(fetched, conts, pageSize, e => e.Timestamp); - seen.AddRange(page); - conts = nextConts; - isFirstPage = false; - - // Map only carries active partitions now — empty == drain complete. - if (conts.Count == 0) break; - if (page.Count == 0) break; - if (++loopGuard > 50) Assert.Fail("Drain loop did not terminate"); - } - - Assert.Equal(12, seen.Count); - Assert.Equal(12, seen.Select(e => e.Id).Distinct().Count()); - for (int i = 1; i < seen.Count; i++) - { - Assert.True(seen[i - 1].Timestamp >= seen[i].Timestamp, - $"Out-of-order at index {i}: {seen[i - 1].Id}({seen[i - 1].Timestamp}) before {seen[i].Id}({seen[i].Timestamp})"); - } + Assert.Empty(page.Items); + Assert.Null(page.NextRawToken); + Assert.Empty(store.Calls); } } diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/PerPartitionFanOutMerge.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/PerPartitionFanOutMerge.cs index 35563edd..d9b3d7d3 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/PerPartitionFanOutMerge.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/PerPartitionFanOutMerge.cs @@ -3,159 +3,158 @@ using System.Collections.Generic; using System.Linq; using System.Text; using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Helpers; +using AutopilotMonitor.Shared.Pagination; namespace AutopilotMonitor.Functions.DataAccess.TableStorage { /// - /// Generic helpers driving cross-partition merge-pagination for global views - /// where Azure-Tables' native (PK asc, RK asc) ordering would otherwise - /// surface partitions one-at-a-time instead of globally newest-first. - /// Used by ops-events (categories as partitions) and global audit logs - /// (tenants as partitions). + /// Newest-first paging across the partitions of one table: global ops events (categories as + /// partitions) and the global audit log (tenants as partitions). Azure Tables pages a + /// cross-partition query in (PK asc, RK asc) order, which would surface one partition at a time, + /// so every page runs one PartitionKey query per partition and merges the results. /// /// /// - /// Per-partition continuation anchored on RowKey (not Azure's opaque token): - /// LastRowKey = the oldest RowKey we have already returned for this - /// partition. Next fetch filters RowKey gt LastRowKey — natural - /// ordering within the partition (revtick → newest-first) means this skips - /// exactly what we showed and re-considers everything older. + /// Invariant: every partition uses the same reverse-tick RowKey scheme, so the global order is + /// (RowKey asc, Partition asc) and that order is the display order. The merge sorts by exactly + /// that key and never by a resolved timestamp column — a merge key that disagrees with the + /// RowKey inside one partition would let the watermark skip rows. /// /// - /// Why not Azure's continuation token? Cross-partition merge can drop items - /// that were fetched but didn't survive the top-pageSize cut. If we advanced - /// an Azure cursor past those leftovers, we'd lose them forever. With a - /// RowKey bound we simply don't advance the continuation for partitions whose - /// fetched items lost the merge — they get re-fetched next call and compete - /// again at a lower floor. + /// The continuation is the (RowKey, Partition) of the last returned row, so its size does not + /// depend on the number of partitions. The earlier per-partition continuation map kept one entry + /// for every partition with rows and outgrew the host's URL limit (HTTP 414) on the tenant + /// fan-out. /// /// - /// Continuation shape: only active partitions are encoded. Missing - /// from the map = exhausted (or never had any data). For 200-tenant audit - /// fan-outs that previously kept exhausted entries around, the encoded - /// token would balloon past 30 KB and trip URL-length limits before - /// reaching the function (browser net::ERR_FAILED on Next click). - /// Now the token shrinks page-over-page as partitions exhaust. + /// Each partition reads at most pageSize + 1 rows after the watermark. More rows exist + /// exactly when more than pageSize were read in total: if no partition reached its limit, + /// every partition is exhausted. /// /// internal static class PerPartitionFanOutMerge { - internal sealed record PartitionContinuation(string? LastRowKey); + /// The (RowKey, Partition) of the last row a page returned. + internal sealed record Watermark(string RowKey, string Partition); - internal sealed record PartitionFetchResult( - string Partition, - IReadOnlyList<(string RowKey, T Item)> Items); - - /// - /// Pure merge-and-advance step. Given pre-fetched per-partition items - /// (already newest-first within each list) and the previous per-partition - /// continuation map, returns the top-pageSize merged items plus the - /// next continuation map. Partitions with nothing left to return are - /// dropped from the map entirely — consumers treat "missing" as - /// exhausted on subsequent calls (see class remarks). - /// - public static (List Items, Dictionary NextContinuations) - MergeAndAdvance( - IReadOnlyList> fetched, - IReadOnlyDictionary priorContinuations, - int pageSize, - Func timestampSelector) + /// Lower RowKey bound of one partition's read. + internal sealed record RowKeyBound(string RowKey, bool Inclusive) { - var allFetched = fetched - .SelectMany(r => r.Items.Select(t => (part: r.Partition, rk: t.RowKey, item: t.Item))) - .OrderByDescending(t => timestampSelector(t.item)) - .ToList(); - - var merged = allFetched.Take(pageSize).ToList(); - var returnedItems = merged.Select(t => t.item).ToList(); - - var newContinuations = new Dictionary(StringComparer.Ordinal); - foreach (var r in fetched) - { - var returnedFromPart = merged.Where(m => m.part == r.Partition).ToList(); - if (returnedFromPart.Count > 0) - { - // Advance to the oldest RowKey we returned for this partition. - // Revtick scheme: larger RowKey string = older timestamp, so - // OrdinalCompare-max gives us the oldest of the returned set. - var oldestRk = returnedFromPart - .Select(m => m.rk) - .OrderBy(rk => rk, StringComparer.Ordinal) - .Last(); - newContinuations[r.Partition] = new PartitionContinuation(oldestRk); - } - else if (r.Items.Count == 0) - { - // No items returned from this partition's query → exhausted. - // Drop from map entirely; the consumer's "missing = exhausted" - // rule (when continuation was provided) does the right thing - // and keeps the encoded token small. - } - else - { - // Fetched some but they all lost the merge cut to other partitions. - // Keep continuation where it was — they'll be re-fetched and likely - // win on the next page when the floor drops. Wasted bandwidth, but - // correct (no item silently dropped). - priorContinuations.TryGetValue(r.Partition, out var prior); - newContinuations[r.Partition] = prior ?? new PartitionContinuation(null); - } - } - - return (returnedItems, newContinuations); + public string ToODataClause() => $"RowKey {(Inclusive ? "ge" : "gt")} '{RowKey.Replace("'", "''")}'"; } - // Wire format v1 (legacy): every partition encoded with `{rk, x:bool}`. - // Wire format v2 (current): only ACTIVE partitions, `{rk}` only. Decoder - // accepts both and silently drops v1 `x:true` entries so old tokens - // emitted by previous deploys still page through cleanly. - private const int WireFormatVersion = 2; + /// + /// Delegate that reads up to limit rows of one partition, newest first (RowKey + /// ascending), restricted by bound when it is set. It must return every matching row + /// up to the limit — a short read is taken as "partition exhausted". + /// + internal delegate Task> PartitionReader( + string partition, RowKeyBound? bound, int limit, CancellationToken cancellationToken); - public static string EncodeMultiContinuation(IReadOnlyDictionary continuations) + /// + /// The rows of that come strictly after the watermark in + /// (RowKey, Partition) order. A partition that sorts after the watermark's partition may still + /// hold rows with the watermark's own RowKey; one that sorts before (or is the same) may not. + /// + internal static RowKeyBound? BoundAfter(Watermark? watermark, string partition) { - var doc = new Dictionary + if (watermark == null) return null; + var inclusive = string.CompareOrdinal(partition, watermark.Partition) > 0; + return new RowKeyBound(watermark.RowKey, inclusive); + } + + public static async Task> FetchPageAsync( + IEnumerable partitions, + int pageSize, + string? continuation, + PartitionReader readPartition, + CancellationToken cancellationToken = default) + { + if (pageSize < 1) throw new ArgumentOutOfRangeException(nameof(pageSize)); + + Watermark? watermark = null; + if (!string.IsNullOrEmpty(continuation)) + { + watermark = DecodeWatermark(continuation); + // The HMAC wrapper already rejected foreign tokens; one that passes it and still does + // not decode was minted by an older deploy. End the list rather than restart it. + if (watermark == null) return RawPage.Empty; + } + + var limit = pageSize + 1; + var rows = await BoundedFanOut.RunAsync(partitions, BoundedFanOut.CrossTenantConcurrency, async (partition, ct) => + { + var read = await readPartition(partition, BoundAfter(watermark, partition), limit, ct).ConfigureAwait(false); + return read.Select(r => (Partition: partition, r.RowKey, r.Item)).ToList(); + }, cancellationToken).ConfigureAwait(false); + + return Merge(rows, pageSize); + } + + /// + /// Pure merge step: orders the rows read from all partitions by (RowKey, Partition), returns + /// the first and a continuation when rows are left over. + /// + internal static RawPage Merge(IEnumerable<(string Partition, string RowKey, T Item)> rows, int pageSize) + { + var ordered = rows + .OrderBy(r => r.RowKey, StringComparer.Ordinal) + .ThenBy(r => r.Partition, StringComparer.Ordinal) + .ToList(); + var page = ordered.Take(pageSize).ToList(); + + string? next = null; + if (ordered.Count > pageSize) + { + var last = page[page.Count - 1]; + next = EncodeWatermark(new Watermark(last.RowKey, last.Partition)); + } + return new RawPage(page.Select(r => r.Item).ToList(), next); + } + + private const int WireFormatVersion = 3; + + public static string EncodeWatermark(Watermark watermark) + { + var json = JsonSerializer.Serialize(new Dictionary { ["v"] = WireFormatVersion, - ["c"] = continuations.ToDictionary( - kv => kv.Key, - kv => (object?)new Dictionary - { - ["rk"] = kv.Value.LastRowKey, - }), - }; - var json = JsonSerializer.Serialize(doc); + ["rk"] = watermark.RowKey, + ["p"] = watermark.Partition, + }); return Convert.ToBase64String(Encoding.UTF8.GetBytes(json)); } - public static Dictionary DecodeMultiContinuation(string? raw) + /// Decodes a watermark; null for anything that is not a v3 watermark. + public static Watermark? DecodeWatermark(string? raw) { - var result = new Dictionary(StringComparer.Ordinal); - if (string.IsNullOrEmpty(raw)) return result; - + if (string.IsNullOrEmpty(raw)) return null; try { - var bytes = Convert.FromBase64String(raw); - using var doc = JsonDocument.Parse(bytes); - if (!doc.RootElement.TryGetProperty("c", out var entries)) return result; - foreach (var prop in entries.EnumerateObject()) - { - // v1 back-compat: skip entries explicitly marked exhausted. - // v2 never writes them, but old in-flight tokens may. - if (prop.Value.TryGetProperty("x", out var xEl) && xEl.ValueKind == JsonValueKind.True) - continue; + using var doc = JsonDocument.Parse(Convert.FromBase64String(raw)); + var root = doc.RootElement; + if (root.ValueKind != JsonValueKind.Object) return null; + if (!root.TryGetProperty("v", out var v) || v.ValueKind != JsonValueKind.Number + || !v.TryGetInt32(out var version) || version != WireFormatVersion) + return null; + if (!root.TryGetProperty("rk", out var rk) || rk.ValueKind != JsonValueKind.String) return null; + if (!root.TryGetProperty("p", out var p) || p.ValueKind != JsonValueKind.String) return null; - string? rk = prop.Value.TryGetProperty("rk", out var rkEl) && rkEl.ValueKind == JsonValueKind.String - ? rkEl.GetString() - : null; - result[prop.Name] = new PartitionContinuation(rk); - } + var rowKey = rk.GetString(); + var partition = p.GetString(); + return string.IsNullOrEmpty(rowKey) || partition == null ? null : new Watermark(rowKey, partition); } - catch + catch (FormatException) { - // Malformed continuation → treat as empty (start over). The wire- - // layer fingerprint already rejects tampered tokens before reaching here. + return null; + } + catch (JsonException) + { + return null; } - return result; } } } diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableOpsEventRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableOpsEventRepository.cs index 6d934d11..1cb9a84b 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableOpsEventRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableOpsEventRepository.cs @@ -115,60 +115,35 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage // partitions would just trade certainty for cost — the coverage test guards the list instead. internal static readonly string[] AllCategories = OpsEventCategory.All; - private async Task> FanOutAcrossCategoriesAsync( + private Task> FanOutAcrossCategoriesAsync( DateTime? dateFrom, DateTime? dateTo, int pageSize, string? continuation, OpsEventQueryFilters? filters = null) { - var continuations = PerPartitionFanOutMerge.DecodeMultiContinuation(continuation); - // First-page request fans out across every category; subsequent - // pages restrict to whatever is still active in the continuation - // map. With only 6 fixed categories this never grew to a problem - // size, but the convention now matches the audit fan-out so the - // shared fan-out helper has one rule. - var isFirstPage = string.IsNullOrEmpty(continuation); - - var activeCats = AllCategories - .Where(cat => isFirstPage || continuations.ContainsKey(cat)) - .ToList(); - if (activeCats.Count == 0) - return new RawPage(new List(), null); - - var fetchTasks = activeCats.Select(async cat => - { - continuations.TryGetValue(cat, out var catContinuation); - var filter = BuildFilterWithRowKeyBound(cat, dateFrom, dateTo, catContinuation?.LastRowKey, filters); - - var fetched = new List<(string RowKey, OpsEventEntry Item)>(); - await foreach (var e in _table.QueryAsync(filter: filter, maxPerPage: pageSize)) + return PerPartitionFanOutMerge.FetchPageAsync( + AllCategories, pageSize, continuation, + async (category, bound, limit, ct) => { - fetched.Add((e.RowKey, MapToEntry(e))); - if (fetched.Count >= pageSize) break; - } - return new PerPartitionFanOutMerge.PartitionFetchResult(cat, fetched); - }).ToList(); - - var results = await Task.WhenAll(fetchTasks); - - var (items, nextContinuations) = PerPartitionFanOutMerge.MergeAndAdvance( - results, continuations, pageSize, e => e.Timestamp); - - // Map only carries active partitions; empty = pagination complete. - string? nextRawToken = nextContinuations.Count > 0 - ? PerPartitionFanOutMerge.EncodeMultiContinuation(nextContinuations) - : null; - return new RawPage(items, nextRawToken); + var filter = BuildFilterWithRowKeyBound(category, dateFrom, dateTo, bound, filters); + var read = new List<(string RowKey, OpsEventEntry Item)>(); + await foreach (var e in _table.QueryAsync(filter: filter, maxPerPage: limit, cancellationToken: ct)) + { + read.Add((e.RowKey, MapToEntry(e))); + if (read.Count >= limit) break; + } + return read; + }); } internal static string BuildFilterWithRowKeyBound( - string category, DateTime? dateFrom, DateTime? dateTo, string? lastRowKey, + string category, DateTime? dateFrom, DateTime? dateTo, PerPartitionFanOutMerge.RowKeyBound? rowKeyBound, OpsEventQueryFilters? filters = null) { var clauses = new List { $"PartitionKey eq '{category.Replace("'", "''")}'", }; - if (!string.IsNullOrEmpty(lastRowKey)) - clauses.Add($"RowKey gt '{lastRowKey!.Replace("'", "''")}'"); + if (rowKeyBound != null) + clauses.Add(rowKeyBound.ToODataClause()); if (dateFrom.HasValue) clauses.Add(BusinessTimestamp.OpsDateFromClause(ToUtc(dateFrom.Value))); if (dateTo.HasValue) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs index 12e3ec0e..a62c6449 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs @@ -1,5 +1,6 @@ using Azure; using Azure.Data.Tables; +using AutopilotMonitor.Functions.DataAccess.TableStorage; using AutopilotMonitor.Functions.Helpers; using AutopilotMonitor.Functions.Pagination; using AutopilotMonitor.Functions.Security; @@ -196,9 +197,7 @@ namespace AutopilotMonitor.Functions.Services DateTime? dateFrom, DateTime? dateTo, int pageSize, string? continuation, bool excludeDeletions = false, AuditLogQueryFilters? filters = null) { - // Cross-tenant: per-tenant fan-out + merge by RowKey. Without this - // fan-out Azure pages by (PK asc, RK asc) cross-partition, surfacing - // tenants alphabetically rather than newest-first globally. + // Cross-tenant: per-tenant fan-out + merge by RowKey (see PerPartitionFanOutMerge). return FetchAllAuditLogsPageAsync(dateFrom, dateTo, pageSize, continuation, excludeDeletions, filters); } @@ -268,16 +267,6 @@ namespace AutopilotMonitor.Functions.Services if (pageSize < 1) throw new ArgumentOutOfRangeException(nameof(pageSize)); try { - var continuations = AutopilotMonitor.Functions.DataAccess.TableStorage.PerPartitionFanOutMerge - .DecodeMultiContinuation(continuation); - // First-page request (no continuation) fans out across every - // tenant in the catalog. Subsequent pages restrict to tenants - // still in the continuation map — anything dropped by - // MergeAndAdvance (because that partition exhausted on a - // prior page) stays dropped, which is what shrinks the - // wire-format token from 30+ KB to a few hundred bytes. - var isFirstPage = string.IsNullOrEmpty(continuation); - // Tenants come from TenantConfiguration (1 row per tenant — cheap). // PLUS the synthetic global-tenant partition (Constants.AuditGlobalTenantId) // where platform-action audits are written from TenantOffboardFunction, @@ -293,45 +282,20 @@ namespace AutopilotMonitor.Functions.Services tenantIds.Add(tenantId); } - var activeTenantIds = tenantIds - .Where(t => isFirstPage || continuations.ContainsKey(t)) - .ToList(); - if (activeTenantIds.Count == 0) - return new RawPage(new List(), null); - var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.AuditLogs); - - // Per-tenant fetch in parallel: filter `PartitionKey eq tenantId` plus - // `RowKey gt LastRowKey` (revtick scheme → strictly older than what - // we already returned for this tenant), plus the optional date window. - var fetchTasks = activeTenantIds.Select(async tid => - { - continuations.TryGetValue(tid, out var prior); - var tenantFilter = BuildAuditLogFilterWithRowKeyBound(tid, dateFrom, dateTo, prior?.LastRowKey, excludeDeletions, filters); - var fetched = new List<(string RowKey, AuditLogEntry Item)>(); - await foreach (var e in tableClient.QueryAsync(filter: tenantFilter, maxPerPage: pageSize)) + return await PerPartitionFanOutMerge.FetchPageAsync( + tenantIds, pageSize, continuation, + async (tenantId, bound, limit, ct) => { - fetched.Add((e.RowKey, MapToAuditLogEntry(e))); - if (fetched.Count >= pageSize) break; - } - return new AutopilotMonitor.Functions.DataAccess.TableStorage.PerPartitionFanOutMerge - .PartitionFetchResult(tid, fetched); - }).ToList(); - - var results = await Task.WhenAll(fetchTasks); - - var (items, nextContinuations) = AutopilotMonitor.Functions.DataAccess.TableStorage.PerPartitionFanOutMerge - .MergeAndAdvance(results, continuations, pageSize, e => e.Timestamp); - - // MergeAndAdvance only puts active partitions into the map now - // (exhausted ones are dropped, see PerPartitionFanOutMerge.cs). - // An empty map therefore means "every partition is done" and we - // emit no continuation, ending the pagination cleanly. - string? nextRawToken = nextContinuations.Count > 0 - ? AutopilotMonitor.Functions.DataAccess.TableStorage.PerPartitionFanOutMerge - .EncodeMultiContinuation(nextContinuations) - : null; - return new RawPage(items, nextRawToken); + var filter = BuildAuditLogFilterWithRowKeyBound(tenantId, dateFrom, dateTo, bound, excludeDeletions, filters); + var read = new List<(string RowKey, AuditLogEntry Item)>(); + await foreach (var e in tableClient.QueryAsync(filter: filter, maxPerPage: limit, cancellationToken: ct)) + { + read.Add((e.RowKey, MapToAuditLogEntry(e))); + if (read.Count >= limit) break; + } + return read; + }); } catch (Exception ex) { @@ -382,8 +346,8 @@ namespace AutopilotMonitor.Functions.Services } internal static string BuildAuditLogFilterWithRowKeyBound( - string tenantId, DateTime? dateFrom, DateTime? dateTo, string? lastRowKey, bool excludeDeletions, - AuditLogQueryFilters? filters = null) + string tenantId, DateTime? dateFrom, DateTime? dateTo, PerPartitionFanOutMerge.RowKeyBound? rowKeyBound, + bool excludeDeletions, AuditLogQueryFilters? filters = null) { var clauses = new List { @@ -393,8 +357,8 @@ namespace AutopilotMonitor.Functions.Services if (excludeDeletions) clauses.Add(DeletionExclusionClause()); AppendAuditFieldFilters(clauses, filters); - if (!string.IsNullOrEmpty(lastRowKey)) - clauses.Add($"RowKey gt '{lastRowKey!.Replace("'", "''")}'"); + if (rowKeyBound != null) + clauses.Add(rowKeyBound.ToODataClause()); if (dateFrom.HasValue) clauses.Add(BusinessTimestamp.AuditDateFromClause(ToUtc(dateFrom.Value))); if (dateTo.HasValue)