mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 04:32:32 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
@@ -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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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 })
|
||||
{
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -1,325 +1,283 @@
|
||||
using AutopilotMonitor.Functions.DataAccess.TableStorage;
|
||||
using AutopilotMonitor.Shared.DataAccess;
|
||||
using static AutopilotMonitor.Functions.DataAccess.TableStorage.PerPartitionFanOutMerge;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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: "<set-by-caller>", 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<OpsEventEntry> 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<string, List<string>> _rows;
|
||||
private readonly Func<string, bool> _matches;
|
||||
public List<(string Partition, RowKeyBound? Bound, int Limit)> Calls { get; } = new();
|
||||
|
||||
public Store(Dictionary<string, List<string>> rows, Func<string, bool>? matches = null)
|
||||
{
|
||||
_rows = rows;
|
||||
_matches = matches ?? (_ => true);
|
||||
}
|
||||
|
||||
public IEnumerable<string> Partitions => _rows.Keys;
|
||||
|
||||
public PartitionReader<string> 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);
|
||||
};
|
||||
|
||||
/// <summary>The full newest-first order the pages must reproduce.</summary>
|
||||
public List<string> 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<List<List<string>>> WalkAsync(Store store, int pageSize)
|
||||
{
|
||||
var pages = new List<List<string>>();
|
||||
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<OpsEventEntry>(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<string, PartitionContinuation>(),
|
||||
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<string, PartitionContinuation>(),
|
||||
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<string> { Rk(i, "_" + Guid.Empty.ToString("N")), Rk(i + 1000, "_" + Guid.Empty.ToString("N")) }));
|
||||
|
||||
var (items, conts) = MergeAndAdvance(
|
||||
fetched,
|
||||
new Dictionary<string, PartitionContinuation>(),
|
||||
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<string, PartitionContinuation>(),
|
||||
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<string, PartitionContinuation>
|
||||
{
|
||||
["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<PartitionFetchResult<OpsEventEntry>> SimulateFetch(
|
||||
Dictionary<string, PartitionContinuation> continuations, int pageSize, bool isFirstPage)
|
||||
{
|
||||
var fetched = new List<PartitionFetchResult<OpsEventEntry>>();
|
||||
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<OpsEventEntry>(part, rows));
|
||||
}
|
||||
return fetched;
|
||||
}
|
||||
|
||||
const int pageSize = 5;
|
||||
var seen = new List<OpsEventEntry>();
|
||||
var conts = new Dictionary<string, PartitionContinuation>();
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
+119
-120
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Per-partition continuation anchored on RowKey (not Azure's opaque token):
|
||||
/// <c>LastRowKey</c> = the oldest RowKey we have already returned for this
|
||||
/// partition. Next fetch filters <c>RowKey gt LastRowKey</c> — 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Continuation shape: only <em>active</em> 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 <c>net::ERR_FAILED</c> on Next click).
|
||||
/// Now the token shrinks page-over-page as partitions exhaust.
|
||||
/// Each partition reads at most <c>pageSize + 1</c> rows after the watermark. More rows exist
|
||||
/// exactly when more than <c>pageSize</c> were read in total: if no partition reached its limit,
|
||||
/// every partition is exhausted.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class PerPartitionFanOutMerge
|
||||
{
|
||||
internal sealed record PartitionContinuation(string? LastRowKey);
|
||||
/// <summary>The (RowKey, Partition) of the last row a page returned.</summary>
|
||||
internal sealed record Watermark(string RowKey, string Partition);
|
||||
|
||||
internal sealed record PartitionFetchResult<T>(
|
||||
string Partition,
|
||||
IReadOnlyList<(string RowKey, T Item)> Items);
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
public static (List<T> Items, Dictionary<string, PartitionContinuation> NextContinuations)
|
||||
MergeAndAdvance<T>(
|
||||
IReadOnlyList<PartitionFetchResult<T>> fetched,
|
||||
IReadOnlyDictionary<string, PartitionContinuation> priorContinuations,
|
||||
int pageSize,
|
||||
Func<T, DateTime> timestampSelector)
|
||||
/// <summary>Lower RowKey bound of one partition's read.</summary>
|
||||
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<string, PartitionContinuation>(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;
|
||||
/// <summary>
|
||||
/// Delegate that reads up to <c>limit</c> rows of one partition, newest first (RowKey
|
||||
/// ascending), restricted by <c>bound</c> when it is set. It must return every matching row
|
||||
/// up to the limit — a short read is taken as "partition exhausted".
|
||||
/// </summary>
|
||||
internal delegate Task<List<(string RowKey, T Item)>> PartitionReader<T>(
|
||||
string partition, RowKeyBound? bound, int limit, CancellationToken cancellationToken);
|
||||
|
||||
public static string EncodeMultiContinuation(IReadOnlyDictionary<string, PartitionContinuation> continuations)
|
||||
/// <summary>
|
||||
/// The rows of <paramref name="partition"/> 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.
|
||||
/// </summary>
|
||||
internal static RowKeyBound? BoundAfter(Watermark? watermark, string partition)
|
||||
{
|
||||
var doc = new Dictionary<string, object?>
|
||||
if (watermark == null) return null;
|
||||
var inclusive = string.CompareOrdinal(partition, watermark.Partition) > 0;
|
||||
return new RowKeyBound(watermark.RowKey, inclusive);
|
||||
}
|
||||
|
||||
public static async Task<RawPage<T>> FetchPageAsync<T>(
|
||||
IEnumerable<string> partitions,
|
||||
int pageSize,
|
||||
string? continuation,
|
||||
PartitionReader<T> 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<T>.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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Pure merge step: orders the rows read from all partitions by (RowKey, Partition), returns
|
||||
/// the first <paramref name="pageSize"/> and a continuation when rows are left over.
|
||||
/// </summary>
|
||||
internal static RawPage<T> Merge<T>(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<T>(page.Select(r => r.Item).ToList(), next);
|
||||
}
|
||||
|
||||
private const int WireFormatVersion = 3;
|
||||
|
||||
public static string EncodeWatermark(Watermark watermark)
|
||||
{
|
||||
var json = JsonSerializer.Serialize(new Dictionary<string, object>
|
||||
{
|
||||
["v"] = WireFormatVersion,
|
||||
["c"] = continuations.ToDictionary(
|
||||
kv => kv.Key,
|
||||
kv => (object?)new Dictionary<string, object?>
|
||||
{
|
||||
["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<string, PartitionContinuation> DecodeMultiContinuation(string? raw)
|
||||
/// <summary>Decodes a watermark; null for anything that is not a v3 watermark.</summary>
|
||||
public static Watermark? DecodeWatermark(string? raw)
|
||||
{
|
||||
var result = new Dictionary<string, PartitionContinuation>(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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+16
-41
@@ -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<RawPage<OpsEventEntry>> FanOutAcrossCategoriesAsync(
|
||||
private Task<RawPage<OpsEventEntry>> 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<OpsEventEntry>(new List<OpsEventEntry>(), 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<TableEntity>(filter: filter, maxPerPage: pageSize))
|
||||
return PerPartitionFanOutMerge.FetchPageAsync<OpsEventEntry>(
|
||||
AllCategories, pageSize, continuation,
|
||||
async (category, bound, limit, ct) =>
|
||||
{
|
||||
fetched.Add((e.RowKey, MapToEntry(e)));
|
||||
if (fetched.Count >= pageSize) break;
|
||||
}
|
||||
return new PerPartitionFanOutMerge.PartitionFetchResult<OpsEventEntry>(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<OpsEventEntry>(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<TableEntity>(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<string>
|
||||
{
|
||||
$"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)
|
||||
|
||||
+18
-54
@@ -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<AuditLogEntry>(new List<AuditLogEntry>(), 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<TableEntity>(filter: tenantFilter, maxPerPage: pageSize))
|
||||
return await PerPartitionFanOutMerge.FetchPageAsync<AuditLogEntry>(
|
||||
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<AuditLogEntry>(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<AuditLogEntry>(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<TableEntity>(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<string>
|
||||
{
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user