feat(audit): flip AuditLog clustered key to (OccurredAtUtc, EventId) (ScadaBridge#111)

Variant B2 of the #78 benchmark: migration FlipAuditLogClusteredKey drops the
twelve non-clustered indexes, re-adds PK_AuditLog as (OccurredAtUtc, EventId)
on ps_AuditLog_Month, adds the aligned unique UX_AuditLog_EventId_Occurred
(EventId, OccurredAtUtc) for the idempotency probe, and rebuilds the twelve
indexes unchanged. Guarded (re-runnable) Up, inverse-rebuild Down; production
scripts docs/plans/sql/FlipAuditLogClusteredKey{,.rollback}.sql.

The partition-switch purge staging table now reads the live clustered key
order from sys.index_columns, so it switches either shape (old binaries
hard-code the old order: ship with the binaries, both centrals stopped).

Evidence: 300k rows, page 1 90,532 -> 34 logical reads (ordered backward
clustered scan under Top), keyset page 90,532 -> 46, probe 14 -> 14;
FlipAuditLogClusteredKeyMigrationTests pins indexes, grants, Down/Up, the
staging key (mutation: hard-coded old key fails SWITCH) and the read ratio.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-09 22:50:35 -05:00
parent 559e02f5c8
commit 5e7d5389d4
14 changed files with 3032 additions and 43 deletions
@@ -39,9 +39,10 @@ public class AuditLogEntityTypeConfigurationTests : IDisposable
[Fact]
public void Configure_MapsToAuditLogTable_WithCompositePrimaryKey()
{
// Composite PK {EventId, OccurredAtUtc} is required by the partitioned
// AuditLog table — the clustered key must include the partition column
// (OccurredAtUtc) so each row can be located in its partition.
// Composite PK {OccurredAtUtc, EventId} (FlipAuditLogClusteredKey, ScadaBridge#111):
// the clustered key must include the partition column (OccurredAtUtc), and leading
// with it lets the Audit Log page's newest-first TOP-N / keyset read an ordered range
// instead of the whole partitioned table.
var entity = _context.Model.FindEntityType(typeof(AuditLogRow));
Assert.NotNull(entity);
@@ -51,12 +52,15 @@ public class AuditLogEntityTypeConfigurationTests : IDisposable
Assert.NotNull(pk);
var pkPropertyNames = pk!.Properties.Select(p => p.Name).ToArray();
Assert.Equal(new[] { nameof(AuditLogRow.EventId), nameof(AuditLogRow.OccurredAtUtc) }, pkPropertyNames);
Assert.Equal(new[] { nameof(AuditLogRow.OccurredAtUtc), nameof(AuditLogRow.EventId) }, pkPropertyNames);
}
/// <summary>Verifies that no non-aligned EventId index is declared and uniqueness rides the aligned clustered key.</summary>
/// <summary>
/// Verifies that no single-column (non-aligned) EventId index is declared, and that EventId
/// uniqueness rides the aligned unique (EventId, OccurredAtUtc) index.
/// </summary>
[Fact]
public void Configure_DeclaresNoNonAlignedEventIdIndex_UniquenessRidesTheAlignedClusteredKey()
public void Configure_DeclaresNoNonAlignedEventIdIndex_UniquenessRidesTheAlignedEventIdIndex()
{
// AlignAuditLogEventIdUniqueness: the standalone non-aligned
// UX_AuditLog_EventId is gone. EventId uniqueness now rides the clustered
@@ -69,6 +73,10 @@ public class AuditLogEntityTypeConfigurationTests : IDisposable
//
// Re-declaring a single-column unique index here would silently reinstate
// the SWITCH incompatibility, so this test pins its absence.
//
// FlipAuditLogClusteredKey (ScadaBridge#111) moved OccurredAtUtc to the front of the
// clustered key, so the EventId-leading seek now comes from UX_AuditLog_EventId_Occurred:
// unique, and keyed (EventId, OccurredAtUtc) so it stays partition-aligned.
var entity = _context.Model.FindEntityType(typeof(AuditLogRow));
Assert.NotNull(entity);
@@ -76,11 +84,11 @@ public class AuditLogEntityTypeConfigurationTests : IDisposable
entity!.GetIndexes(),
i => i.GetDatabaseName() == "UX_AuditLog_EventId");
var pk = entity.FindPrimaryKey();
Assert.NotNull(pk);
var ux = Assert.Single(entity.GetIndexes(), i => i.GetDatabaseName() == "UX_AuditLog_EventId_Occurred");
Assert.True(ux.IsUnique);
Assert.Equal(
new[] { nameof(AuditLogRow.EventId), nameof(AuditLogRow.OccurredAtUtc) },
pk!.Properties.Select(p => p.Name).ToArray());
ux.Properties.Select(p => p.Name).ToArray());
}
/// <summary>Verifies that the configuration maps the expected number of properties.</summary>
@@ -189,8 +197,9 @@ public class AuditLogEntityTypeConfigurationTests : IDisposable
"IX_AuditLog_Site_Occurred",
"IX_AuditLog_Target_Occurred",
"IX_AuditLog_Unchained", // hash chain (ScadaBridge#41)
// UX_AuditLog_EventId is intentionally absent — dropped by
// AlignAuditLogEventIdUniqueness; the aligned clustered PK carries it.
"UX_AuditLog_EventId_Occurred", // FlipAuditLogClusteredKey (ScadaBridge#111)
// UX_AuditLog_EventId (single-column, non-aligned) is intentionally absent —
// dropped by AlignAuditLogEventIdUniqueness.
};
Assert.Equal(expected, indexNames);
@@ -0,0 +1,315 @@
using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
using Microsoft.EntityFrameworkCore.Migrations.Operations;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using Xunit.Abstractions;
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Migrations;
/// <summary>
/// SQL Server tests for the <c>FlipAuditLogClusteredKey</c> migration (ScadaBridge#111,
/// variant B2 of <c>docs/plans/2026-11-09-auditlog-clustered-key-benchmark.md</c>): the new
/// clustered key and the aligned unique EventId index, every other index and grant preserved,
/// a faithful Down, a re-runnable Up, a partition-switch purge that works on BOTH key shapes,
/// and the read win itself — the Audit Log page's page-1 and keyset queries read a small
/// fraction of the pages they read on the old key.
/// </summary>
/// <remarks>
/// The tests in this class migrate the fixture database down and back up, so they own their
/// fixture (one database per test class) and every test leaves the database fully migrated.
/// </remarks>
public sealed class FlipAuditLogClusteredKeyMigrationTests : IClassFixture<MsSqlMigrationFixture>
{
private const string FlipMigration = "20261110033732_FlipAuditLogClusteredKey";
// The migration immediately before the flip, so "Down" reverts exactly this one.
private static readonly string PreviousMigration = MigrationBefore(FlipMigration);
private static readonly string[] NonClusteredIndexes =
[
"IX_AuditLog_OccurredAtUtc", "IX_AuditLog_Site_Occurred", "IX_AuditLog_CorrelationId",
"IX_AuditLog_Channel_Status_Occurred", "IX_AuditLog_Target_Occurred", "IX_AuditLog_Execution",
"IX_AuditLog_ParentExecution", "IX_AuditLog_Node_Occurred", "IX_AuditLog_Kind_Occurred",
"IX_AuditLog_Actor_Occurred", "IX_AuditLog_Unchained", "IX_AuditLog_ChainSeq",
];
private readonly MsSqlMigrationFixture _fixture;
private readonly ITestOutputHelper _output;
/// <summary>Creates the test class with its own migrated SQL Server database.</summary>
/// <param name="fixture">The per-class migration fixture.</param>
/// <param name="output">Test output (records the measured logical reads).</param>
public FlipAuditLogClusteredKeyMigrationTests(MsSqlMigrationFixture fixture, ITestOutputHelper output)
{
_fixture = fixture;
_output = output;
}
/// <summary>
/// After the full chain: clustered PK (OccurredAtUtc, EventId), the unique aligned
/// (EventId, OccurredAtUtc) index, all twelve non-clustered indexes with unchanged keys and
/// filters, everything on ps_AuditLog_Month, and every append-only/chain grant intact.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[SkippableFact]
public async Task Up_FlipsTheClusteredKey_AddsTheAlignedUniqueIndex_AndKeepsEveryIndexAndGrant()
{
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
var indexes = await ReadIndexesAsync();
Assert.Equal("OccurredAtUtc, EventId", indexes["PK_AuditLog"].Keys);
Assert.True(indexes["PK_AuditLog"].Clustered);
Assert.Equal("EventId, OccurredAtUtc", indexes["UX_AuditLog_EventId_Occurred"].Keys);
Assert.True(indexes["UX_AuditLog_EventId_Occurred"].Unique);
Assert.Equal(NonClusteredIndexes.Length + 2, indexes.Count);
Assert.All(indexes.Values, i => Assert.Equal("ps_AuditLog_Month", i.DataSpace));
AssertTwelveNonClusteredIndexes(indexes);
Assert.Equal(ExpectedGrants, await ReadGrantsAsync());
}
/// <summary>
/// Down restores the old key and drops the B2 index; Up re-flips; re-running Up's SQL on an
/// already-flipped table is a no-op; and a partition-switch purge succeeds on each shape
/// (the staging table's key follows the live table's).
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[SkippableFact]
public async Task DownUp_RoundTrips_IsReRunnable_AndThePartitionSwitchPurgesEitherShape()
{
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
await MigrateToAsync(PreviousMigration);
var down = await ReadIndexesAsync();
Assert.Equal("EventId, OccurredAtUtc", down["PK_AuditLog"].Keys);
Assert.False(down.ContainsKey("UX_AuditLog_EventId_Occurred"));
AssertTwelveNonClusteredIndexes(down);
Assert.Equal(ExpectedGrants, await ReadGrantsAsync());
// Old shape: the purge's staging table must take the old key order.
await AssertSwitchOutPurgesAsync(new DateTime(2027, 3, 1, 0, 0, 0, DateTimeKind.Utc));
await MigrateToAsync(FlipMigration);
var up = await ReadIndexesAsync();
Assert.Equal("OccurredAtUtc, EventId", up["PK_AuditLog"].Keys);
// Re-run Up's SQL against the flipped table: guarded, so nothing changes.
await using (var conn = _fixture.OpenConnection())
{
foreach (var op in new FlipAuditLogClusteredKey().UpOperations.OfType<SqlOperation>())
{
await using var cmd = new SqlCommand(op.Sql, conn) { CommandTimeout = 300 };
await cmd.ExecuteNonQueryAsync();
}
}
Assert.Equal(up.Select(kv => (kv.Key, kv.Value)).OrderBy(x => x.Key),
(await ReadIndexesAsync()).Select(kv => (kv.Key, kv.Value)).OrderBy(x => x.Key));
// New shape: same purge, now with the time-leading staging key.
await AssertSwitchOutPurgesAsync(new DateTime(2027, 4, 1, 0, 0, 0, DateTimeKind.Utc));
Assert.Equal(ExpectedGrants, await ReadGrantsAsync());
}
/// <summary>
/// The read win, measured with the session's logical-read counter around the repository's real query: page 1
/// (newest 50, no filter) and a keyset page read far fewer AuditLog pages on the flipped
/// key than on the old one over the same rows.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[SkippableFact]
public async Task PageOneAndKeysetPage_ReadAFractionOfThePages_OnTheFlippedKey()
{
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
await SeedRowsAsync(new DateTime(2027, 6, 1, 0, 0, 0, DateTimeKind.Utc), rows: 20_000);
var (flippedPage1, flippedKeyset) = await MeasurePageReadsAsync();
await MigrateToAsync(PreviousMigration);
try
{
var (oldPage1, oldKeyset) = await MeasurePageReadsAsync();
_output.WriteLine(
$"logical reads, 20k rows - page 1: {flippedPage1} flipped vs {oldPage1} old; keyset: {flippedKeyset} flipped vs {oldKeyset} old");
Assert.True(flippedPage1 * 10 < oldPage1,
$"page 1: {flippedPage1} logical reads on (OccurredAtUtc, EventId) vs {oldPage1} on (EventId, OccurredAtUtc)");
Assert.True(flippedKeyset * 10 < oldKeyset,
$"keyset: {flippedKeyset} logical reads on (OccurredAtUtc, EventId) vs {oldKeyset} on (EventId, OccurredAtUtc)");
}
finally
{
await MigrateToAsync(FlipMigration);
}
}
private static readonly string[] ExpectedGrants =
[
"scadabridge_audit_chainer|SELECT|GRANT|",
"scadabridge_audit_chainer|UPDATE|GRANT|ChainSeq",
"scadabridge_audit_chainer|UPDATE|GRANT|RowHash",
"scadabridge_audit_purger|DELETE|GRANT|",
"scadabridge_audit_purger|SELECT|GRANT|",
"scadabridge_audit_writer|DELETE|DENY|",
"scadabridge_audit_writer|INSERT|GRANT|",
"scadabridge_audit_writer|SELECT|GRANT|",
"scadabridge_audit_writer|UPDATE|DENY|",
];
private static void AssertTwelveNonClusteredIndexes(IReadOnlyDictionary<string, IndexShape> indexes)
{
Assert.Equal("OccurredAtUtc DESC", indexes["IX_AuditLog_OccurredAtUtc"].Keys);
Assert.Equal("Status", indexes["IX_AuditLog_OccurredAtUtc"].Included);
Assert.Equal("SourceSiteId, OccurredAtUtc DESC", indexes["IX_AuditLog_Site_Occurred"].Keys);
Assert.Equal("CorrelationId", indexes["IX_AuditLog_CorrelationId"].Keys);
Assert.Equal("([CorrelationId] IS NOT NULL)", indexes["IX_AuditLog_CorrelationId"].Filter);
Assert.Equal("Category, Status, OccurredAtUtc DESC", indexes["IX_AuditLog_Channel_Status_Occurred"].Keys);
Assert.Equal("Target, OccurredAtUtc DESC", indexes["IX_AuditLog_Target_Occurred"].Keys);
Assert.Equal("([Target] IS NOT NULL)", indexes["IX_AuditLog_Target_Occurred"].Filter);
Assert.Equal("ExecutionId", indexes["IX_AuditLog_Execution"].Keys);
Assert.Equal("ParentExecutionId", indexes["IX_AuditLog_ParentExecution"].Keys);
Assert.Equal("SourceNode, OccurredAtUtc", indexes["IX_AuditLog_Node_Occurred"].Keys);
Assert.Equal("Kind, OccurredAtUtc DESC", indexes["IX_AuditLog_Kind_Occurred"].Keys);
Assert.Equal("Actor, OccurredAtUtc DESC", indexes["IX_AuditLog_Actor_Occurred"].Keys);
Assert.Equal("OccurredAtUtc, EventId", indexes["IX_AuditLog_Unchained"].Keys);
Assert.Equal("([ChainSeq] IS NULL)", indexes["IX_AuditLog_Unchained"].Filter);
Assert.Equal("ChainSeq", indexes["IX_AuditLog_ChainSeq"].Keys);
Assert.Equal("([ChainSeq] IS NOT NULL)", indexes["IX_AuditLog_ChainSeq"].Filter);
Assert.All(NonClusteredIndexes, n => Assert.False(indexes[n].Clustered, n));
}
private async Task AssertSwitchOutPurgesAsync(DateTime month)
{
await SeedRowsAsync(month, rows: 50);
await using var ctx = CreateContext();
var purged = await new AuditLogRepository(ctx).SwitchOutPartitionAsync(month);
Assert.Equal(50, purged);
await using var conn = _fixture.OpenConnection();
await using var cmd = new SqlCommand(
"SELECT COUNT_BIG(*) FROM dbo.AuditLog WHERE OccurredAtUtc >= @s AND OccurredAtUtc < @e;", conn);
cmd.Parameters.AddWithValue("@s", month);
cmd.Parameters.AddWithValue("@e", month.AddMonths(1));
Assert.Equal(0L, (long)(await cmd.ExecuteScalarAsync())!);
}
private async Task<(long Page1, long Keyset)> MeasurePageReadsAsync()
{
await using var conn = new SqlConnection(_fixture.ConnectionString);
await conn.OpenAsync();
// EF reuses the already-open connection, so the session's cumulative logical-read
// counter (sys.dm_exec_sessions) brackets exactly the repository's query. The probe
// itself reads no AuditLog pages.
await using var ctx = new ScadaBridgeDbContext(
new DbContextOptionsBuilder<ScadaBridgeDbContext>().UseSqlServer(conn).Options);
var repo = new AuditLogRepository(ctx);
var before = await SessionLogicalReadsAsync(conn);
var page1 = await repo.QueryAsync(new AuditLogQueryFilter(), new AuditLogPaging(50));
var page1Reads = await SessionLogicalReadsAsync(conn) - before;
Assert.Equal(50, page1.Count);
var cursor = page1[^1];
before = await SessionLogicalReadsAsync(conn);
var page2 = await repo.QueryAsync(
new AuditLogQueryFilter(),
new AuditLogPaging(50, cursor.OccurredAtUtc.UtcDateTime, cursor.EventId));
var keysetReads = await SessionLogicalReadsAsync(conn) - before;
Assert.Equal(50, page2.Count);
Assert.True(page2[0].OccurredAtUtc <= cursor.OccurredAtUtc);
return (page1Reads, keysetReads);
}
private static async Task<long> SessionLogicalReadsAsync(SqlConnection conn)
{
await using var cmd = new SqlCommand("SELECT logical_reads FROM sys.dm_exec_sessions WHERE session_id = @@SPID;", conn);
return Convert.ToInt64(await cmd.ExecuteScalarAsync());
}
private async Task SeedRowsAsync(DateTime month, int rows)
{
// Set-based seed with random EventIds (as minted at the sites) spread over the month;
// DetailsJson carries kind/status so the persisted computed columns materialise.
await using var conn = _fixture.OpenConnection();
await using var cmd = new SqlCommand(@"
WITH n AS (SELECT TOP (@rows) ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS i
FROM sys.all_objects a CROSS JOIN sys.all_objects b)
INSERT INTO dbo.AuditLog (EventId, OccurredAtUtc, Actor, Action, Outcome, Category, Target, SourceNode, CorrelationId, DetailsJson)
SELECT NEWID(), DATEADD(SECOND, CAST(i * (2419200.0 / @rows) AS int), @month), N'flip-test', 'ApiOutbound.ApiCall', 'Success',
'ApiOutbound', N'erp', 'central-a', NULL,
N'{""kind"":""ApiCall"",""status"":""Delivered"",""sourceSiteId"":""flip"",""pad"":""' + REPLICATE(N'x', 400) + N'""}'
FROM n;", conn) { CommandTimeout = 300 };
cmd.Parameters.AddWithValue("@rows", rows);
cmd.Parameters.Add(new SqlParameter("@month", System.Data.SqlDbType.DateTime2) { Scale = 7, Value = month });
await cmd.ExecuteNonQueryAsync();
}
private async Task MigrateToAsync(string migration)
{
await using var ctx = new ScadaBridgeDbContext(new DbContextOptionsBuilder<ScadaBridgeDbContext>()
.UseSqlServer(_fixture.ConnectionString, sql => sql.CommandTimeout(600)).Options);
await ctx.GetService<IMigrator>().MigrateAsync(migration);
}
private static string MigrationBefore(string migration)
{
using var ctx = new ScadaBridgeDbContext(new DbContextOptionsBuilder<ScadaBridgeDbContext>()
.UseSqlServer("Server=unused;Database=unused").Options);
var all = ctx.Database.GetMigrations().ToList();
return all[all.IndexOf(migration) - 1];
}
private ScadaBridgeDbContext CreateContext()
=> new(new DbContextOptionsBuilder<ScadaBridgeDbContext>().UseSqlServer(_fixture.ConnectionString).Options);
private async Task<Dictionary<string, IndexShape>> ReadIndexesAsync()
{
await using var conn = _fixture.OpenConnection();
await using var cmd = new SqlCommand(@"
SELECT i.name, i.type, i.is_unique, ds.name, i.filter_definition,
STUFF((SELECT ', ' + c.name + CASE WHEN ic.is_descending_key = 1 THEN ' DESC' ELSE '' END
FROM sys.index_columns ic JOIN sys.columns c ON c.object_id = ic.object_id AND c.column_id = ic.column_id
WHERE ic.object_id = i.object_id AND ic.index_id = i.index_id AND ic.key_ordinal > 0
ORDER BY ic.key_ordinal FOR XML PATH('')), 1, 2, ''),
STUFF((SELECT ', ' + c.name
FROM sys.index_columns ic JOIN sys.columns c ON c.object_id = ic.object_id AND c.column_id = ic.column_id
WHERE ic.object_id = i.object_id AND ic.index_id = i.index_id AND ic.is_included_column = 1
ORDER BY ic.index_column_id FOR XML PATH('')), 1, 2, '')
FROM sys.indexes i JOIN sys.data_spaces ds ON ds.data_space_id = i.data_space_id
WHERE i.object_id = OBJECT_ID('dbo.AuditLog') AND i.type IN (1, 2);", conn);
var result = new Dictionary<string, IndexShape>(StringComparer.Ordinal);
await using var reader = await cmd.ExecuteReaderAsync();
while (await reader.ReadAsync())
{
result[reader.GetString(0)] = new IndexShape(
Clustered: reader.GetByte(1) == 1,
Unique: reader.GetBoolean(2),
DataSpace: reader.GetString(3),
Filter: reader.IsDBNull(4) ? null : reader.GetString(4),
Keys: reader.GetString(5),
Included: reader.IsDBNull(6) ? null : reader.GetString(6));
}
return result;
}
private async Task<string[]> ReadGrantsAsync()
{
await using var conn = _fixture.OpenConnection();
await using var cmd = new SqlCommand(@"
SELECT pr.name, p.permission_name, p.state_desc, COL_NAME(p.major_id, p.minor_id)
FROM sys.database_permissions p JOIN sys.database_principals pr ON pr.principal_id = p.grantee_principal_id
WHERE p.major_id = OBJECT_ID('dbo.AuditLog') AND p.class = 1;", conn);
var list = new List<string>();
await using var reader = await cmd.ExecuteReaderAsync();
while (await reader.ReadAsync())
{
list.Add($"{reader.GetString(0)}|{reader.GetString(1)}|{reader.GetString(2)}|{(reader.IsDBNull(3) ? "" : reader.GetString(3))}");
}
return list.Order(StringComparer.Ordinal).ToArray();
}
private sealed record IndexShape(bool Clustered, bool Unique, string DataSpace, string? Filter, string Keys, string? Included);
}