Merge branch 'worktree-agent-a3b474b485c0288de' into arch-review-remediation
This commit is contained in:
Joseph Doherty
@@ -18,8 +18,9 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
/// ingest timestamp into DetailsJson (there is no promoted IngestedAtUtc
|
||||
/// column — the value is a DetailsJson field set via
|
||||
/// <see cref="AuditRowProjection.WithIngestedAtUtc"/>) and inserted idempotently
|
||||
/// via <see cref="IAuditLogRepository.InsertIfNotExistsAsync"/> — duplicates are
|
||||
/// silently swallowed (first-write-wins).
|
||||
/// via <see cref="IAuditLogRepository.InsertManyIfNotExistsAsync"/> — duplicates
|
||||
/// are silently swallowed (first-write-wins), whether they repeat inside one
|
||||
/// packet or across packets.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -28,9 +29,11 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
/// consistent and the site is free to flip its local row to <c>Forwarded</c>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Audit-write failures must NEVER abort the user-facing action. The actor
|
||||
/// wraps each repository call in its own try/catch so a single bad row cannot
|
||||
/// cause the rest of the batch to be lost, and it guards scope/repository
|
||||
/// Audit-write failures must NEVER abort the user-facing action. Each message
|
||||
/// is written as ONE set-based statement (or one transaction, for the cached
|
||||
/// dual-write); if that fails the actor retries the same work row-by-row inside
|
||||
/// per-row try/catch, so a single bad row still cannot cause the rest of the
|
||||
/// batch to be lost. It also guards scope/repository
|
||||
/// resolution so a transient DI fault cannot restart the singleton — those
|
||||
/// catches are what keep this actor alive across handler throws, not the
|
||||
/// supervisor strategy. The <see cref="SupervisorStrategy"/> override returns
|
||||
@@ -50,6 +53,31 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
/// </remarks>
|
||||
public class AuditLogIngestActor : ReceiveActor
|
||||
{
|
||||
/// <summary>
|
||||
/// Overall budget for one ingest message's database work, deliberately
|
||||
/// SHORTER than the gRPC Ask that wraps it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The path used to stack three identical 30 s budgets — the site's Ask
|
||||
/// (<c>CommunicationOptions.NotificationForwardTimeout</c>), the central gRPC
|
||||
/// handler's Ask (<c>SiteStreamGrpcServer.AuditIngestAskTimeout</c>) and the
|
||||
/// ADO.NET command default — so they all expired at the same instant. The
|
||||
/// caller therefore learned nothing except "it took 30 s": no partial ack, no
|
||||
/// distinction between a slow database and a wedged singleton. Making the
|
||||
/// innermost budget strictly smallest means a slow batch is abandoned by the
|
||||
/// actor FIRST, with the accepted-so-far ids still replied, while the outer
|
||||
/// Asks are still waiting.
|
||||
/// </remarks>
|
||||
internal static readonly TimeSpan IngestBudget = TimeSpan.FromSeconds(20);
|
||||
|
||||
/// <summary>
|
||||
/// Per-statement SQL timeout for the ingest write, strictly inside
|
||||
/// <see cref="IngestBudget"/> so a single wedged statement surfaces as a SQL
|
||||
/// timeout the per-batch catch can log, rather than consuming the whole
|
||||
/// actor-level budget and starving the rest of the batch.
|
||||
/// </summary>
|
||||
internal static readonly TimeSpan IngestSqlCommandTimeout = TimeSpan.FromSeconds(15);
|
||||
|
||||
private readonly IServiceProvider? _serviceProvider;
|
||||
private readonly IAuditLogRepository? _injectedRepository;
|
||||
private readonly ILogger<AuditLogIngestActor> _logger;
|
||||
@@ -185,39 +213,75 @@ public class AuditLogIngestActor : ReceiveActor
|
||||
DateTime nowUtc,
|
||||
List<Guid> accepted)
|
||||
{
|
||||
// Stamp the ingest timestamp here, not at the site. Redact BEFORE the
|
||||
// IngestedAtUtc stamp so the redacted copy carries the central-side
|
||||
// ingest timestamp. The redactor is contract-bound to never throw; a null
|
||||
// redactor (test composition root, no IAuditRedactor registered) falls
|
||||
// back to the SafeDefault rather than pass-through, so HTTP header
|
||||
// redaction always runs. IngestedAtUtc is a DetailsJson field on the
|
||||
// canonical record, so stamp it via the projection helper.
|
||||
var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
|
||||
var projected = new List<AuditEvent>(cmd.Events.Count);
|
||||
foreach (var evt in cmd.Events)
|
||||
{
|
||||
try
|
||||
var filtered = safeRedactor.Apply(evt);
|
||||
projected.Add(AuditRowProjection.WithIngestedAtUtc(filtered, nowUtc));
|
||||
}
|
||||
|
||||
// ONE set-based statement for the whole packet (WP2.2) instead of one
|
||||
// IF NOT EXISTS … INSERT round trip per event. A telemetry packet of 200
|
||||
// rows used to cost 200 sequential round trips on the central singleton's
|
||||
// dispatcher; it now costs two. Idempotency is unchanged — duplicates
|
||||
// within the packet collapse first-write-wins, duplicates against
|
||||
// committed rows are eliminated by the anti-semi-join, and a concurrent
|
||||
// writer degrades to the repository's per-row fallback.
|
||||
using var budget = new CancellationTokenSource(IngestBudget);
|
||||
try
|
||||
{
|
||||
await repository
|
||||
.InsertManyIfNotExistsAsync(projected, IngestSqlCommandTimeout, budget.Token)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
// A batch that returned without throwing means every row is now
|
||||
// present — inserted here or already committed by an earlier delivery.
|
||||
// Both count as accepted: the storage state is consistent and the site
|
||||
// is free to flip its local rows to Forwarded.
|
||||
foreach (var evt in cmd.Events)
|
||||
{
|
||||
// Stamp the ingest timestamp here, not at the site. The
|
||||
// repository's duplicate-key hardening already swallows
|
||||
// duplicate-key races, so the same id arriving twice (site
|
||||
// retry, reconciliation) is a silent no-op.
|
||||
// Redact BEFORE the IngestedAtUtc stamp so the redacted
|
||||
// copy carries the central-side ingest timestamp. The redactor
|
||||
// is contract-bound to never throw. A null
|
||||
// redactor (test composition root, no IAuditRedactor
|
||||
// registered) now falls back to the SafeDefault rather than
|
||||
// pass-through, so HTTP header redaction always runs.
|
||||
// IngestedAtUtc is a DetailsJson field on
|
||||
// the canonical record, so stamp it via the projection helper.
|
||||
var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
|
||||
var filtered = safeRedactor.Apply(evt);
|
||||
var ingested = AuditRowProjection.WithIngestedAtUtc(filtered, nowUtc);
|
||||
await repository.InsertIfNotExistsAsync(ingested).ConfigureAwait(false);
|
||||
accepted.Add(evt.EventId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// The batch is a throughput optimisation, NOT a change to the
|
||||
// failure grain. The documented invariant — "a single bad row cannot
|
||||
// cause the rest of the batch to be lost" — is preserved by falling
|
||||
// back to the per-row path here, so a poison row still costs only
|
||||
// itself. Re-running rows the failed batch may already have committed
|
||||
// is safe: every insert is idempotent on EventId.
|
||||
_logger.LogWarning(ex,
|
||||
"Set-based ingest of {Count} audit event(s) failed; falling back to per-row inserts so one bad row does not sink the batch.",
|
||||
cmd.Events.Count);
|
||||
|
||||
for (var i = 0; i < projected.Count; i++)
|
||||
{
|
||||
// Per-row catch — one bad row never sinks the whole batch.
|
||||
// The row stays Pending at the site; the next drain retries.
|
||||
// Bump the central health counter so a
|
||||
// sustained insert-throw failure surfaces on the dashboard.
|
||||
try { failureCounter?.Increment(); }
|
||||
catch { /* counter must never throw — defence in depth */ }
|
||||
_logger.LogError(ex,
|
||||
"Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.",
|
||||
evt.EventId);
|
||||
try
|
||||
{
|
||||
await repository.InsertIfNotExistsAsync(projected[i], budget.Token).ConfigureAwait(false);
|
||||
accepted.Add(cmd.Events[i].EventId);
|
||||
}
|
||||
catch (Exception rowEx)
|
||||
{
|
||||
// Per-row catch — one bad row never sinks the whole batch.
|
||||
// The row stays Pending at the site; the next drain retries.
|
||||
// Bump the central health counter so a sustained insert-throw
|
||||
// failure surfaces on the dashboard.
|
||||
try { failureCounter?.Increment(); }
|
||||
catch { /* counter must never throw — defence in depth */ }
|
||||
_logger.LogError(rowEx,
|
||||
"Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.",
|
||||
cmd.Events[i].EventId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -262,6 +326,24 @@ public class AuditLogIngestActor : ReceiveActor
|
||||
|
||||
var strategy = dbContext.Database.CreateExecutionStrategy();
|
||||
|
||||
// Fast path (WP2.2): the WHOLE packet in ONE transaction — one
|
||||
// set-based audit insert plus one single-statement upsert per entry.
|
||||
// The predecessor opened a transaction PER entry and issued three
|
||||
// statements inside it (IF NOT EXISTS insert, then the two-statement
|
||||
// SiteCalls upsert), so a 50-entry packet cost ~200 round trips and 50
|
||||
// commits. If anything faults, the per-entry loop below re-runs the
|
||||
// packet with its original per-entry isolation, so the documented
|
||||
// "one entry's failure does not abort the others" invariant survives —
|
||||
// it is simply no longer paid for on the healthy path.
|
||||
using var budget = new CancellationTokenSource(IngestBudget);
|
||||
if (await TryIngestCachedBatchAsync(
|
||||
strategy, dbContext, auditRepo, siteCallRepo, redactor, cmd, accepted, budget.Token)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
replyTo.Tell(new IngestCachedTelemetryReply(accepted));
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (var entry in cmd.Entries)
|
||||
{
|
||||
try
|
||||
@@ -330,6 +412,90 @@ public class AuditLogIngestActor : ReceiveActor
|
||||
replyTo.Tell(new IngestCachedTelemetryReply(accepted));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attempts the whole cached-telemetry packet as ONE transaction: a single
|
||||
/// set-based audit insert followed by one monotonic <c>SiteCalls</c> upsert
|
||||
/// per entry. Returns <see langword="true"/> when it committed (and only then
|
||||
/// appends to <paramref name="accepted"/>), <see langword="false"/> when the
|
||||
/// caller should fall back to the per-entry transaction loop.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The batch is all-or-nothing by construction — it is a single transaction —
|
||||
/// which is why a failure MUST fall back rather than report partial success:
|
||||
/// the per-entry loop is what turns one poison entry into one lost entry
|
||||
/// instead of a lost packet. Nothing is appended to
|
||||
/// <paramref name="accepted"/> until the commit returns, so a failed attempt
|
||||
/// leaves the caller's list untouched and the retry starts from a clean slate.
|
||||
/// </remarks>
|
||||
private async Task<bool> TryIngestCachedBatchAsync(
|
||||
Microsoft.EntityFrameworkCore.Storage.IExecutionStrategy strategy,
|
||||
ScadaBridgeDbContext dbContext,
|
||||
IAuditLogRepository auditRepo,
|
||||
ISiteCallAuditRepository siteCallRepo,
|
||||
IAuditRedactor? redactor,
|
||||
IngestCachedTelemetryCommand cmd,
|
||||
List<Guid> accepted,
|
||||
CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
await strategy.ExecuteAsync(async () =>
|
||||
{
|
||||
await using var tx = await dbContext.Database
|
||||
.BeginTransactionAsync(ct)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
// One central-side instant for the whole packet so a join across
|
||||
// the two tables sees matching timestamps (debugging convenience,
|
||||
// not a correctness invariant).
|
||||
var ingestedAt = DateTime.UtcNow;
|
||||
var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
|
||||
|
||||
var auditRows = new List<AuditEvent>(cmd.Entries.Count);
|
||||
foreach (var entry in cmd.Entries)
|
||||
{
|
||||
// Only the AuditLog row's payload columns are redactable;
|
||||
// SiteCalls carries operational state only (status, retry
|
||||
// count) and is left untouched.
|
||||
var filteredAudit = safeRedactor.Apply(entry.Audit);
|
||||
auditRows.Add(AuditRowProjection.WithIngestedAtUtc(filteredAudit, ingestedAt));
|
||||
}
|
||||
|
||||
await auditRepo
|
||||
.InsertManyIfNotExistsAsync(auditRows, IngestSqlCommandTimeout, ct)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
foreach (var entry in cmd.Entries)
|
||||
{
|
||||
await siteCallRepo
|
||||
.UpsertAsync(entry.SiteCall with { IngestedAtUtc = ingestedAt }, ct)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await tx.CommitAsync(ct).ConfigureAwait(false);
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// No health-counter bump here — the per-entry retry is the authority
|
||||
// on whether this packet genuinely failed, and double-counting a
|
||||
// batch that the fallback then writes successfully would make the
|
||||
// dashboard read as a sustained fault during normal contention.
|
||||
_logger.LogWarning(
|
||||
ex,
|
||||
"Batched cached-telemetry dual-write of {Count} entr(ies) failed; falling back to per-entry transactions.",
|
||||
cmd.Entries.Count);
|
||||
return false;
|
||||
}
|
||||
|
||||
foreach (var entry in cmd.Entries)
|
||||
{
|
||||
accepted.Add(entry.Audit.EventId);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fallback handler installed on the single-repository test ctor — that
|
||||
/// ctor has no DbContext and no <see cref="ISiteCallAuditRepository"/>, so
|
||||
|
||||
@@ -17,8 +17,9 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
/// for monthly boundaries whose latest <c>OccurredAtUtc</c> is older
|
||||
/// than <c>DateTime.UtcNow - RetentionDays</c>.</item>
|
||||
/// <item>For each eligible boundary, calls
|
||||
/// <see cref="IAuditLogRepository.SwitchOutPartitionAsync"/> which runs
|
||||
/// the drop-and-rebuild dance around <c>UX_AuditLog_EventId</c>.</item>
|
||||
/// <see cref="IAuditLogRepository.SwitchOutPartitionAsync"/>, a
|
||||
/// metadata-only staging-table switch (WP2.2 removed the index
|
||||
/// drop/rebuild that used to bracket it).</item>
|
||||
/// <item>Publishes <see cref="AuditLogPurgedEvent"/> on the actor-system
|
||||
/// EventStream so the central health collector + ops surfaces
|
||||
/// can subscribe without coupling to this actor.</item>
|
||||
@@ -26,11 +27,10 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Daily cadence.</b> Partition switch is metadata-only but the
|
||||
/// drop-and-rebuild dance briefly removes <c>UX_AuditLog_EventId</c>; running
|
||||
/// more often than necessary trades unique-index rebuild outages for
|
||||
/// negligible freshness wins. The default 24-hour interval matches
|
||||
/// alog.md §10's retention policy.
|
||||
/// <b>Daily cadence.</b> The partition switch is metadata-only, but it still
|
||||
/// takes schema-modification locks on a table the ingest path is writing to;
|
||||
/// running more often than necessary trades contention for negligible freshness
|
||||
/// wins. The default 24-hour interval matches alog.md §10's retention policy.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Continue-on-error.</b> A single boundary that throws (transient SQL
|
||||
|
||||
@@ -10,12 +10,12 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The purge actor is a daily-cadence singleton, not a hot-loop, because
|
||||
/// partition-switch I/O is metadata-only but the drop-and-rebuild dance
|
||||
/// briefly removes the <c>UX_AuditLog_EventId</c> unique index — running
|
||||
/// more often than necessary trades index-rebuild outages for marginal
|
||||
/// freshness gains. Lower this only when an operator can prove they need
|
||||
/// sub-daily purge granularity.
|
||||
/// The purge actor is a daily-cadence singleton, not a hot-loop. The
|
||||
/// partition switch itself is metadata-only (since WP2.2's
|
||||
/// <c>AlignAuditLogEventIdUniqueness</c> it no longer drops and rebuilds a
|
||||
/// non-aligned unique index around it), but it still takes schema-modification
|
||||
/// locks and competes with ingest for the same table. Lower this only when an
|
||||
/// operator can prove they need sub-daily purge granularity.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <see cref="IntervalOverride"/> exists for tests to drop the cadence to
|
||||
@@ -58,15 +58,14 @@ public sealed class AuditLogPurgeOptions
|
||||
|
||||
/// <summary>
|
||||
/// Per-command timeout (in minutes) for the maintenance SQL the purge tick issues —
|
||||
/// both the partition switch-out drop-and-rebuild dance
|
||||
/// both the partition switch-out staging batch
|
||||
/// (<see cref="ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories.IAuditLogRepository.SwitchOutPartitionAsync"/>)
|
||||
/// and each per-channel <c>DELETE TOP</c> batch. Default 30 minutes.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The ADO.NET default command timeout is ~30 seconds. On a large or contended partition the
|
||||
/// SWITCH dance (which briefly drops <c>UX_AuditLog_EventId</c>) can exceed that and abort
|
||||
/// mid-flight — leaving the live table without its idempotency-supporting unique index until a
|
||||
/// later tick's CATCH branch rebuilds it (arch-review 04, S2). A generous maintenance timeout
|
||||
/// SWITCH batch can exceed that and abort mid-flight, leaving an orphaned staging table for the
|
||||
/// next tick's CATCH branch to clean up (arch-review 04, S2). A generous maintenance timeout
|
||||
/// lets the metadata-only switch complete rather than self-locking. Resolved via
|
||||
/// <see cref="ResolvedMaintenanceCommandTimeout"/>, clamped to a 1-minute floor.
|
||||
/// </remarks>
|
||||
|
||||
@@ -34,6 +34,62 @@ public interface IAuditLogRepository
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Set-based form of <see cref="InsertIfNotExistsAsync"/>: inserts every
|
||||
/// event in <paramref name="events"/> that does not already exist, in as few
|
||||
/// round trips as the provider allows, and returns the number of rows
|
||||
/// actually written. First-write-wins idempotency is unchanged — an EventId
|
||||
/// already present is silently skipped.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Duplicate EventIds are tolerated both within and across packets.</b>
|
||||
/// Duplicates inside one call collapse to the first occurrence before the
|
||||
/// statement is built (an append-only row is immutable, so later copies of an
|
||||
/// EventId carry the same content); duplicates against already-committed rows
|
||||
/// are eliminated by the anti-semi-join. A concurrent writer that commits a
|
||||
/// row mid-statement is handled by falling back to the per-row path, so the
|
||||
/// batch is a throughput optimisation and never a correctness dependency.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Default implementation.</b> The interface supplies a per-row loop so a
|
||||
/// test double or an alternative store keeps working unchanged; the EF Core
|
||||
/// implementation overrides it with a genuine set-based statement. Callers on
|
||||
/// the ingest hot path should always prefer this method — the per-row form
|
||||
/// cost one <c>IF NOT EXISTS … INSERT</c> round trip per audit event.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="events">Audit events to insert; may contain duplicates.</param>
|
||||
/// <param name="commandTimeout">
|
||||
/// Optional per-statement timeout. The ingest actor passes a budget strictly
|
||||
/// SHORTER than the gRPC Ask that wraps it, so the reply is produced before
|
||||
/// the caller gives up rather than at the same instant.
|
||||
/// </param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>A task that resolves to the number of rows inserted (duplicates excluded).</returns>
|
||||
async Task<int> InsertManyIfNotExistsAsync(
|
||||
IReadOnlyList<AuditEvent> events,
|
||||
TimeSpan? commandTimeout = null,
|
||||
CancellationToken ct = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(events);
|
||||
|
||||
var inserted = 0;
|
||||
var seen = new HashSet<Guid>(events.Count);
|
||||
foreach (var evt in events)
|
||||
{
|
||||
if (evt is null || !seen.Add(evt.EventId))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
await InsertIfNotExistsAsync(evt, ct).ConfigureAwait(false);
|
||||
inserted++;
|
||||
}
|
||||
|
||||
return inserted;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns up to <see cref="AuditLogPaging.PageSize"/> rows matching
|
||||
/// <paramref name="filter"/>, ordered by <c>(OccurredAtUtc DESC, EventId DESC)</c>.
|
||||
@@ -60,36 +116,32 @@ public interface IAuditLogRepository
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Drop-and-rebuild dance.</b> <c>UX_AuditLog_EventId</c> is intentionally
|
||||
/// non-partition-aligned (it lives on <c>[PRIMARY]</c> so single-column
|
||||
/// EventId uniqueness — required by <see cref="InsertIfNotExistsAsync"/> —
|
||||
/// can be enforced cheaply). SQL Server rejects
|
||||
/// <c>ALTER TABLE … SWITCH PARTITION</c> while a non-aligned unique index
|
||||
/// is present, so the implementation drops the index, creates a staging
|
||||
/// table with byte-identical schema, switches the partition's data into
|
||||
/// staging, drops staging (discarding the rows), and rebuilds the unique
|
||||
/// index. The CATCH branch guarantees the index is rebuilt even on partial
|
||||
/// failure so the table never returns to live traffic without its
|
||||
/// idempotency-supporting index.
|
||||
/// <b>Partition-aligned uniqueness — no index drop.</b> EventId uniqueness is
|
||||
/// enforced by the clustered <c>PK_AuditLog (EventId, OccurredAtUtc)</c>,
|
||||
/// which is aligned on <c>ps_AuditLog_Month(OccurredAtUtc)</c>, so
|
||||
/// <c>ALTER TABLE … SWITCH PARTITION</c> has no non-aligned unique index to
|
||||
/// object to. The implementation creates a staging table with byte-identical
|
||||
/// schema, switches the partition's data into staging, and drops staging
|
||||
/// (discarding the rows). Nothing is dropped from the live table, so there is
|
||||
/// no window in which the idempotency enforcement is absent and no offline
|
||||
/// index rebuild inside the switch transaction.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Outage window.</b> The dance briefly removes the unique index, so
|
||||
/// concurrent <see cref="InsertIfNotExistsAsync"/> calls during the switch
|
||||
/// could in principle race past the IF NOT EXISTS check without the index
|
||||
/// catching the duplicate. This is acceptable for the daily purge cadence
|
||||
/// — the inserts that the IF NOT EXISTS check guards are themselves rare
|
||||
/// enough that a sub-second collision window is operationally negligible,
|
||||
/// and the composite PK still rejects same-(EventId, OccurredAtUtc) rows.
|
||||
/// The predecessor design carried a non-aligned
|
||||
/// <c>UX_AuditLog_EventId</c> on <c>[PRIMARY]</c> that had to be dropped and
|
||||
/// rebuilt around every switch. A defensive guarded <c>DROP INDEX</c> remains
|
||||
/// in the batch so a database restored from a pre-alignment backup still
|
||||
/// purges; it is a one-way cleanup, never rebuilt.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="monthBoundary">Lower-bound datetime of the monthly partition to switch out.</param>
|
||||
/// <param name="commandTimeout">
|
||||
/// Optional per-command timeout for the maintenance SQL (the row-count sample plus the
|
||||
/// drop-and-rebuild dance). When null the provider default applies. The purge actor passes
|
||||
/// staging/switch batch). When null the provider default applies. The purge actor passes
|
||||
/// <see cref="ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeOptions.ResolvedMaintenanceCommandTimeout"/>
|
||||
/// (default 30 min) because the ~30s ADO.NET default can abort the switch mid-dance on a large
|
||||
/// or contended partition, leaving the table without <c>UX_AuditLog_EventId</c> until the next
|
||||
/// tick recovers (arch-review 04, S2).
|
||||
/// (default 30 min) because the ~30s ADO.NET default can abort the switch mid-batch on a large
|
||||
/// or contended partition, leaving an orphaned staging table for the next tick's CATCH to clean
|
||||
/// up (arch-review 04, S2).
|
||||
/// </param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>A task that resolves to the approximate number of rows discarded by the partition switch.</returns>
|
||||
|
||||
+18
-3
@@ -46,10 +46,25 @@ public interface INotificationOutboxRepository
|
||||
Task<Notification?> GetByIdAsync(string notificationId, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Marks <paramref name="n"/> modified and persists it (status transitions).
|
||||
/// Commits internally — this call is its own transaction.
|
||||
/// Persists <paramref name="n"/>'s <b>delivery-state</b> columns —
|
||||
/// <c>Status</c>, <c>RetryCount</c>, <c>LastError</c>, <c>ResolvedTargets</c>,
|
||||
/// <c>LastAttemptAt</c>, <c>NextAttemptAt</c>, <c>DeliveredAt</c>. Commits
|
||||
/// internally — this call is its own transaction.
|
||||
/// </summary>
|
||||
/// <param name="n">The notification to update.</param>
|
||||
/// <remarks>
|
||||
/// <b>Scope is deliberately narrow.</b> Those seven columns are the ONLY
|
||||
/// mutable state a notification has: everything else (identity, type, list,
|
||||
/// subject, body, type data, source/origin attribution, enqueue and creation
|
||||
/// timestamps) is written once at ingest and is immutable by contract. Every
|
||||
/// caller — the dispatcher's per-attempt write and the operator retry/discard
|
||||
/// one-shots — touches only this set. Implementations are therefore free to
|
||||
/// issue a targeted UPDATE of these columns rather than rewriting the whole
|
||||
/// row, which matters because the row carries <c>nvarchar(max)</c> body and
|
||||
/// payload columns and the dispatcher writes on EVERY attempt. A future
|
||||
/// caller that needs to change an immutable column must add its own method
|
||||
/// rather than widening this one.
|
||||
/// </remarks>
|
||||
/// <param name="n">The notification whose delivery state should be persisted.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>A task that completes when the notification has been persisted.</returns>
|
||||
Task UpdateAsync(Notification n, CancellationToken cancellationToken = default);
|
||||
|
||||
+10
-7
@@ -172,15 +172,18 @@ public class AuditLogEntityTypeConfiguration : IEntityTypeConfiguration<AuditLog
|
||||
// ── Keys + indexes ───────────────────────────────────────────────────
|
||||
|
||||
// Composite PK includes OccurredAtUtc — required by the monthly partition scheme
|
||||
// (ps_AuditLog_Month) so the clustered key is partition-aligned. EventId still
|
||||
// needs to be globally unique for InsertIfNotExistsAsync idempotency, so a
|
||||
// separate (non-aligned) unique index is declared on EventId alone.
|
||||
// (ps_AuditLog_Month) so the clustered key is partition-aligned. It is ALSO the
|
||||
// only uniqueness enforcement the ingest path needs: EventId is a GUID minted
|
||||
// once at the emitting site and never re-stamped, so a given EventId always
|
||||
// arrives with the same OccurredAtUtc and can only ever land in one partition.
|
||||
// Uniqueness of the pair is therefore uniqueness of EventId in practice, and the
|
||||
// idempotency probe (WHERE EventId = @id) still seeks the clustered key's leading
|
||||
// column. The predecessor non-aligned UX_AuditLog_EventId on [PRIMARY] was
|
||||
// dropped by AlignAuditLogEventIdUniqueness — it existed only to give
|
||||
// single-column uniqueness and its non-alignment forced an offline drop/rebuild
|
||||
// around every partition-switch purge.
|
||||
builder.HasKey(e => new { e.EventId, e.OccurredAtUtc });
|
||||
|
||||
builder.HasIndex(e => e.EventId)
|
||||
.IsUnique()
|
||||
.HasDatabaseName("UX_AuditLog_EventId");
|
||||
|
||||
// Index names are locked for reconciliation/migration discoverability. The
|
||||
// column SETS migrate to the canonical/computed shape (alog.md §4 semantics
|
||||
// preserved): Channel→Category, Site/Node/Execution/ParentExecution now read
|
||||
|
||||
+2090
File diff suppressed because it is too large
Load Diff
+93
@@ -0,0 +1,93 @@
|
||||
using Microsoft.EntityFrameworkCore.Migrations;
|
||||
|
||||
#nullable disable
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
|
||||
{
|
||||
/// <summary>
|
||||
/// Makes <c>dbo.AuditLog</c>'s EventId uniqueness <b>partition-aligned</b> by
|
||||
/// dropping the non-aligned <c>UX_AuditLog_EventId</c> and leaving the clustered
|
||||
/// <c>PK_AuditLog (EventId, OccurredAtUtc)</c> — already aligned on
|
||||
/// <c>ps_AuditLog_Month(OccurredAtUtc)</c> — as the sole enforcement.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Why.</b> <c>ALTER TABLE … SWITCH PARTITION</c> refuses to run while a
|
||||
/// non-aligned index exists on the table, so the monthly retention purge
|
||||
/// (<c>AuditLogRepository.SwitchOutPartitionAsync</c>) had to DROP
|
||||
/// <c>UX_AuditLog_EventId</c>, switch, and then CREATE it again — an OFFLINE
|
||||
/// whole-table unique-index build, inside the switch transaction, blocking every
|
||||
/// audit writer for its duration. It also opened a window in which the index that
|
||||
/// backs ingest idempotency did not exist at all, and a mid-dance failure could
|
||||
/// leave the live table without it until a later tick's CATCH branch repaired it.
|
||||
/// With alignment there is nothing to drop, so the switch is metadata-only and the
|
||||
/// purge stops competing with ingest.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Why dropping it is safe — EventId is globally unique by construction.</b>
|
||||
/// The composite key enforces uniqueness of the PAIR, not of EventId alone, so in
|
||||
/// principle the same EventId could now be stored twice under two different
|
||||
/// <c>OccurredAtUtc</c> values (in two different partitions). That cannot happen
|
||||
/// here: <c>EventId</c> is a GUID minted ONCE at the emitting site, in the same
|
||||
/// operation that stamps <c>OccurredAtUtc</c>, and both travel together verbatim
|
||||
/// through telemetry and reconciliation — nothing downstream re-stamps either
|
||||
/// field. A given EventId therefore always arrives with the same OccurredAtUtc and
|
||||
/// can only ever map to one partition, which makes pair-uniqueness equivalent to
|
||||
/// EventId-uniqueness for every row this system produces. GUID collision across
|
||||
/// partitions is not a real risk.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The idempotency probe still seeks.</b> Both ingest forms test
|
||||
/// <c>WHERE EventId = @id</c>, which is the LEADING column of the clustered PK, so
|
||||
/// the probe remains an index seek. The cost changes shape rather than order: it
|
||||
/// becomes one seek per partition (the partition column is not in the predicate, so
|
||||
/// SQL Server cannot eliminate partitions) instead of a single seek on a
|
||||
/// non-partitioned index. Against a monthly scheme that is a couple of dozen
|
||||
/// shallow B-tree seeks — cheap, and paid on a path that now issues one statement
|
||||
/// per telemetry packet rather than one per row.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Edition note.</b> The alternative remedy — keeping the non-aligned index and
|
||||
/// rebuilding it with <c>ONLINE = ON</c> outside the switch transaction — requires
|
||||
/// Enterprise (or Azure SQL / Developer) edition; online index rebuild is not
|
||||
/// available on Standard, which this deployment does not guarantee. Alignment
|
||||
/// needs no edition-specific feature and removes the rebuild entirely, so it is
|
||||
/// preferred regardless of edition.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Down is a faithful reverse</b> and recreates the index on <c>[PRIMARY]</c>
|
||||
/// exactly as <c>CollapseAuditLogToCanonical</c> created it. Reverting also
|
||||
/// reinstates the SWITCH incompatibility, so the purge's guarded defensive
|
||||
/// <c>DROP INDEX</c> (retained in <c>SwitchOutPartitionAsync</c> for databases
|
||||
/// restored from pre-alignment backups) would remove it again on the next purge.
|
||||
/// The partition function/scheme (<c>pf_AuditLog_Month</c> /
|
||||
/// <c>ps_AuditLog_Month</c>) and every aligned index are untouched by both
|
||||
/// directions.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public partial class AlignAuditLogEventIdUniqueness : Migration
|
||||
{
|
||||
/// <inheritdoc />
|
||||
protected override void Up(MigrationBuilder migrationBuilder)
|
||||
{
|
||||
// Raw, existence-guarded SQL rather than the scaffolded DropIndex: the
|
||||
// AuditLog table is raw-SQL managed (partition scheme, persisted computed
|
||||
// columns, append-only role grants), so its migrations stay explicit and
|
||||
// re-runnable. The guard also lets this apply cleanly to a database whose
|
||||
// index was already removed by the purge path's defensive cleanup.
|
||||
migrationBuilder.Sql(@"
|
||||
IF EXISTS (SELECT 1 FROM sys.indexes
|
||||
WHERE name = 'UX_AuditLog_EventId' AND object_id = OBJECT_ID('dbo.AuditLog'))
|
||||
DROP INDEX UX_AuditLog_EventId ON dbo.AuditLog;");
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void Down(MigrationBuilder migrationBuilder)
|
||||
{
|
||||
migrationBuilder.Sql(@"
|
||||
IF NOT EXISTS (SELECT 1 FROM sys.indexes
|
||||
WHERE name = 'UX_AuditLog_EventId' AND object_id = OBJECT_ID('dbo.AuditLog'))
|
||||
CREATE UNIQUE NONCLUSTERED INDEX UX_AuditLog_EventId ON dbo.AuditLog (EventId) ON [PRIMARY];");
|
||||
}
|
||||
}
|
||||
}
|
||||
-4
@@ -1806,10 +1806,6 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
|
||||
.HasDatabaseName("IX_AuditLog_CorrelationId")
|
||||
.HasFilter("[CorrelationId] IS NOT NULL");
|
||||
|
||||
b.HasIndex("EventId")
|
||||
.IsUnique()
|
||||
.HasDatabaseName("UX_AuditLog_EventId");
|
||||
|
||||
b.HasIndex("ExecutionId")
|
||||
.HasDatabaseName("IX_AuditLog_Execution");
|
||||
|
||||
|
||||
+241
-28
@@ -1,5 +1,8 @@
|
||||
using System.Data;
|
||||
using System.Text;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.EntityFrameworkCore.Storage;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.Audit;
|
||||
@@ -17,15 +20,32 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
||||
/// </summary>
|
||||
public class AuditLogRepository : IAuditLogRepository
|
||||
{
|
||||
// SQL Server error numbers for duplicate-key violations on
|
||||
// UX_AuditLog_EventId. 2601 is a unique-index violation; 2627 is a
|
||||
// primary-key/unique-constraint violation. The IF NOT EXISTS … INSERT
|
||||
// pattern has a check-then-act race window — two sessions can both pass
|
||||
// the EXISTS check and then both attempt the INSERT — and the loser
|
||||
// surfaces as one of these errors. Idempotency demands we swallow them.
|
||||
// SQL Server error numbers for duplicate-key violations on the
|
||||
// partition-aligned clustered PK_AuditLog (EventId, OccurredAtUtc).
|
||||
// 2601 is a unique-index violation; 2627 is a primary-key/unique-constraint
|
||||
// violation. The IF NOT EXISTS … INSERT pattern has a check-then-act race
|
||||
// window — two sessions can both pass the EXISTS check and then both attempt
|
||||
// the INSERT — and the loser surfaces as one of these errors. Idempotency
|
||||
// demands we swallow them.
|
||||
private const int SqlErrorUniqueIndexViolation = 2601;
|
||||
private const int SqlErrorPrimaryKeyViolation = 2627;
|
||||
|
||||
// Rows per set-based ingest statement. Ten bound parameters per row against
|
||||
// SQL Server's 2,100-parameter ceiling leaves ample headroom at 100 rows
|
||||
// (1,000 parameters) while still collapsing a typical telemetry packet into
|
||||
// a single round trip. Larger chunks buy little — the win is round-trip
|
||||
// elimination, not statement size — and would push plan-cache churn up
|
||||
// (one cached plan per distinct row count).
|
||||
private const int IngestChunkRows = 100;
|
||||
|
||||
// Ordinal-stable column list shared by the single-row and set-based inserts.
|
||||
// The five persisted computed columns (Kind/Status/SourceSiteId/ExecutionId/
|
||||
// ParentExecutionId) plus the non-persisted IngestedAtUtc are derived
|
||||
// server-side from DetailsJson and must NEVER appear here — writing a
|
||||
// computed column is an error.
|
||||
private const string CanonicalColumnList =
|
||||
"EventId, OccurredAtUtc, Actor, Action, Outcome, Category, Target, SourceNode, CorrelationId, DetailsJson";
|
||||
|
||||
private readonly ScadaBridgeDbContext _context;
|
||||
private readonly ILogger<AuditLogRepository> _logger;
|
||||
|
||||
@@ -83,7 +103,7 @@ VALUES
|
||||
{
|
||||
// Two concurrent sessions both passed the IF NOT EXISTS check and
|
||||
// both attempted the INSERT — the loser raises 2601/2627 against
|
||||
// UX_AuditLog_EventId. First-write-wins idempotency is already the
|
||||
// the clustered PK. First-write-wins idempotency is already the
|
||||
// documented contract for this method, so the race outcome is
|
||||
// semantically a no-op. Swallow at Debug; other SqlExceptions
|
||||
// bubble.
|
||||
@@ -95,6 +115,196 @@ VALUES
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<int> InsertManyIfNotExistsAsync(
|
||||
IReadOnlyList<AuditEvent> events,
|
||||
TimeSpan? commandTimeout = null,
|
||||
CancellationToken ct = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(events);
|
||||
|
||||
if (events.Count == 0)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
// De-duplicate WITHIN the packet before the statement is built. The
|
||||
// set-based INSERT … SELECT … WHERE NOT EXISTS only tests rows that are
|
||||
// ALREADY committed, so two copies of one EventId inside a single VALUES
|
||||
// constructor would both pass the anti-semi-join and collide on the
|
||||
// clustered PK — losing the whole statement to a duplicate-key fault.
|
||||
// First-write-wins matches the single-row contract exactly (a later copy
|
||||
// of the same EventId is by definition the same immutable append-only
|
||||
// row), so keeping the first occurrence is not merely convenient, it is
|
||||
// the documented semantics.
|
||||
var seen = new HashSet<Guid>(events.Count);
|
||||
var distinct = new List<AuditEvent>(events.Count);
|
||||
foreach (var evt in events)
|
||||
{
|
||||
if (evt is not null && seen.Add(evt.EventId))
|
||||
{
|
||||
distinct.Add(evt);
|
||||
}
|
||||
}
|
||||
|
||||
var inserted = 0;
|
||||
for (var offset = 0; offset < distinct.Count; offset += IngestChunkRows)
|
||||
{
|
||||
var length = Math.Min(IngestChunkRows, distinct.Count - offset);
|
||||
var chunk = distinct.GetRange(offset, length);
|
||||
|
||||
try
|
||||
{
|
||||
inserted += await InsertChunkAsync(chunk, commandTimeout, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (SqlException ex) when (
|
||||
ex.Number == SqlErrorUniqueIndexViolation
|
||||
|| ex.Number == SqlErrorPrimaryKeyViolation)
|
||||
{
|
||||
// A CONCURRENT writer committed one of this chunk's EventIds
|
||||
// between the anti-semi-join and the insert (the same
|
||||
// check-then-act window the single-row path documents), and the
|
||||
// whole set-based statement rolled back with it. Fall back to
|
||||
// the per-row path so the rows that are genuinely new still land
|
||||
// — the batch is a throughput optimisation, never a correctness
|
||||
// dependency. Every row is retried, including the one that
|
||||
// collided, because InsertIfNotExistsAsync swallows its own
|
||||
// duplicate-key fault as a no-op.
|
||||
_logger.LogDebug(
|
||||
ex,
|
||||
"Set-based audit ingest chunk of {Count} row(s) hit a duplicate-key violation (error {SqlErrorNumber}); falling back to per-row inserts.",
|
||||
chunk.Count,
|
||||
ex.Number);
|
||||
|
||||
foreach (var evt in chunk)
|
||||
{
|
||||
await InsertIfNotExistsAsync(evt, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return inserted;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Executes one set-based idempotent insert: a VALUES table constructor
|
||||
/// anti-semi-joined against the committed rows, so a whole telemetry packet
|
||||
/// costs ONE round trip instead of one <c>IF NOT EXISTS … INSERT</c> per row.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Raw ADO.NET (rather than <c>ExecuteSqlInterpolated</c>) because the
|
||||
/// statement's parameter count varies with the chunk size and every parameter
|
||||
/// needs an explicit <see cref="SqlDbType"/>: the VALUES constructor's column
|
||||
/// types are inferred from the first row's parameters, so leaving a null
|
||||
/// <c>Target</c>/<c>SourceNode</c> untyped would give the derived column the
|
||||
/// wrong type and defeat the seek on the anti-semi-join. The command enlists
|
||||
/// in the DbContext's ambient transaction when one is open — the cached
|
||||
/// telemetry dual-write runs the audit insert and the SiteCalls upsert inside
|
||||
/// a single transaction and both must commit or roll back together.
|
||||
/// </remarks>
|
||||
private async Task<int> InsertChunkAsync(
|
||||
IReadOnlyList<AuditEvent> chunk, TimeSpan? commandTimeout, CancellationToken ct)
|
||||
{
|
||||
var sql = new StringBuilder(256 + (chunk.Count * 64));
|
||||
sql.Append("INSERT INTO dbo.AuditLog (").Append(CanonicalColumnList).Append(")\n");
|
||||
sql.Append("SELECT v.EventId, v.OccurredAtUtc, v.Actor, v.Action, v.Outcome, v.Category, ");
|
||||
sql.Append("v.Target, v.SourceNode, v.CorrelationId, v.DetailsJson\n");
|
||||
sql.Append("FROM (VALUES\n");
|
||||
for (var i = 0; i < chunk.Count; i++)
|
||||
{
|
||||
if (i > 0)
|
||||
{
|
||||
sql.Append(",\n");
|
||||
}
|
||||
|
||||
sql.Append(" (@e").Append(i)
|
||||
.Append(",@t").Append(i)
|
||||
.Append(",@a").Append(i)
|
||||
.Append(",@n").Append(i)
|
||||
.Append(",@o").Append(i)
|
||||
.Append(",@c").Append(i)
|
||||
.Append(",@g").Append(i)
|
||||
.Append(",@s").Append(i)
|
||||
.Append(",@r").Append(i)
|
||||
.Append(",@d").Append(i)
|
||||
.Append(')');
|
||||
}
|
||||
|
||||
sql.Append("\n) AS v (").Append(CanonicalColumnList).Append(")\n");
|
||||
sql.Append("WHERE NOT EXISTS (SELECT 1 FROM dbo.AuditLog x WHERE x.EventId = v.EventId);");
|
||||
|
||||
var conn = _context.Database.GetDbConnection();
|
||||
var openedHere = false;
|
||||
if (conn.State != ConnectionState.Open)
|
||||
{
|
||||
await conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
openedHere = true;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = sql.ToString();
|
||||
cmd.Transaction = _context.Database.CurrentTransaction?.GetDbTransaction();
|
||||
if (commandTimeout is { } timeout)
|
||||
{
|
||||
cmd.CommandTimeout = (int)timeout.TotalSeconds;
|
||||
}
|
||||
|
||||
for (var i = 0; i < chunk.Count; i++)
|
||||
{
|
||||
var evt = chunk[i];
|
||||
|
||||
// Same canonical projection as the single-row path: UTC-kind
|
||||
// OccurredAtUtc, empty Actor collapses to NULL, Outcome/Category
|
||||
// bound as their varchar storage form.
|
||||
var occurred = DateTime.SpecifyKind(evt.OccurredAtUtc.UtcDateTime, DateTimeKind.Utc);
|
||||
object actor = string.IsNullOrEmpty(evt.Actor) ? DBNull.Value : evt.Actor;
|
||||
|
||||
AddParameter(cmd, "@e" + i, SqlDbType.UniqueIdentifier, size: 0, evt.EventId);
|
||||
AddParameter(cmd, "@t" + i, SqlDbType.DateTime2, size: 0, occurred);
|
||||
AddParameter(cmd, "@a" + i, SqlDbType.NVarChar, size: 256, actor);
|
||||
AddParameter(cmd, "@n" + i, SqlDbType.VarChar, size: 64, evt.Action);
|
||||
AddParameter(cmd, "@o" + i, SqlDbType.VarChar, size: 16, evt.Outcome.ToString());
|
||||
AddParameter(cmd, "@c" + i, SqlDbType.VarChar, size: 32, evt.Category);
|
||||
AddParameter(cmd, "@g" + i, SqlDbType.NVarChar, size: 256, evt.Target);
|
||||
AddParameter(cmd, "@s" + i, SqlDbType.VarChar, size: 64, evt.SourceNode);
|
||||
AddParameter(cmd, "@r" + i, SqlDbType.UniqueIdentifier, size: 0, evt.CorrelationId);
|
||||
AddParameter(cmd, "@d" + i, SqlDbType.NVarChar, size: -1, evt.DetailsJson);
|
||||
}
|
||||
|
||||
return await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (openedHere)
|
||||
{
|
||||
await conn.CloseAsync().ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Binds one explicitly-typed parameter. A null CLR value binds as
|
||||
/// <see cref="DBNull"/> while KEEPING its declared <see cref="SqlDbType"/>,
|
||||
/// which is what makes the VALUES constructor's derived column types stable
|
||||
/// regardless of which rows happen to carry nulls.
|
||||
/// </summary>
|
||||
private static void AddParameter(
|
||||
System.Data.Common.DbCommand cmd, string name, SqlDbType type, int size, object? value)
|
||||
{
|
||||
var p = (SqlParameter)cmd.CreateParameter();
|
||||
p.ParameterName = name;
|
||||
p.SqlDbType = type;
|
||||
if (size != 0)
|
||||
{
|
||||
p.Size = size;
|
||||
}
|
||||
|
||||
p.Value = value ?? DBNull.Value;
|
||||
cmd.Parameters.Add(p);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<IReadOnlyList<AuditEvent>> QueryAsync(
|
||||
AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default)
|
||||
@@ -229,13 +439,23 @@ VALUES
|
||||
/// <inheritdoc />
|
||||
public async Task<long> SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default)
|
||||
{
|
||||
// The drop-and-rebuild batch below runs via
|
||||
// The switch batch below runs via
|
||||
// ExecuteSqlRaw with NO EF user-transaction — it carries its own server-side
|
||||
// BEGIN TRANSACTION / TRY-CATCH / ROLLBACK — so the DbContext's retrying
|
||||
// execution strategy (EnableRetryOnFailure) MAY auto-replay the whole batch on
|
||||
// a transient fault. That replay is safe: every step is IF-EXISTS / IF-NOT-EXISTS
|
||||
// guarded and the staging table is GUID-suffixed, so a re-run is idempotent.
|
||||
//
|
||||
// ALIGNED UNIQUENESS (WP2.2): there is no longer an index drop/rebuild around
|
||||
// the SWITCH. EventId uniqueness is now enforced solely by the clustered
|
||||
// PK_AuditLog (EventId, OccurredAtUtc), which is partition-aligned on
|
||||
// ps_AuditLog_Month(OccurredAtUtc) — so ALTER TABLE … SWITCH PARTITION has no
|
||||
// non-aligned index to object to. The former dance dropped UX_AuditLog_EventId,
|
||||
// switched, then rebuilt it OFFLINE inside the same transaction: a whole-table
|
||||
// unique-index build blocking every writer for the duration of the purge, and a
|
||||
// window in which the idempotency-supporting index did not exist at all. Both
|
||||
// are gone. See migration AlignAuditLogEventIdUniqueness for the reasoning.
|
||||
//
|
||||
// Maintenance timeout in whole seconds (ADO.NET CommandTimeout unit). Null leaves the
|
||||
// provider default in place. See AuditLogPurgeOptions.MaintenanceCommandTimeoutMinutes /
|
||||
// arch-review 04 S2 for why the ~30s default is unsafe for the switch-out dance.
|
||||
@@ -270,8 +490,13 @@ VALUES
|
||||
BEGIN TRY
|
||||
BEGIN TRANSACTION;
|
||||
|
||||
-- 1. Drop the non-aligned unique index. ALTER TABLE SWITCH refuses
|
||||
-- to run while it exists.
|
||||
-- 1. Defensive cleanup for databases created before
|
||||
-- AlignAuditLogEventIdUniqueness: the migration drops the
|
||||
-- non-aligned UX_AuditLog_EventId, but a database restored from an
|
||||
-- older backup could still carry it and SWITCH refuses to run while
|
||||
-- a non-aligned unique index exists. Dropping it here is idempotent
|
||||
-- and permanent — the aligned clustered PK is the only uniqueness
|
||||
-- enforcement the ingest path needs, so there is nothing to rebuild.
|
||||
IF EXISTS (SELECT 1 FROM sys.indexes WHERE name = 'UX_AuditLog_EventId' AND object_id = OBJECT_ID('dbo.AuditLog'))
|
||||
DROP INDEX UX_AuditLog_EventId ON dbo.AuditLog;
|
||||
|
||||
@@ -313,31 +538,19 @@ VALUES
|
||||
-- 4. Drop staging — the rows are discarded here. This is the purge.
|
||||
DROP TABLE dbo.[{stagingTableName}];
|
||||
|
||||
-- 5. Rebuild the non-aligned unique index. Live traffic that hit the
|
||||
-- table during steps 1-4 saw composite-PK uniqueness only; from
|
||||
-- here on, single-column EventId uniqueness is restored.
|
||||
CREATE UNIQUE NONCLUSTERED INDEX UX_AuditLog_EventId ON dbo.AuditLog (EventId) ON [PRIMARY];
|
||||
|
||||
COMMIT TRANSACTION;
|
||||
END TRY
|
||||
BEGIN CATCH
|
||||
IF @@TRANCOUNT > 0 ROLLBACK TRANSACTION;
|
||||
|
||||
-- Best-effort staging cleanup. The DROP INDEX in step 1 is now
|
||||
-- rolled back (so the index is back), but the staging table from
|
||||
-- step 2 may or may not survive the rollback depending on the
|
||||
-- failure point. Guard the DROP so a missing staging table doesn't
|
||||
-- mask the original error.
|
||||
-- Best-effort staging cleanup. The staging table from step 2 may or
|
||||
-- may not survive the rollback depending on the failure point. Guard
|
||||
-- the DROP so a missing staging table doesn't mask the original error.
|
||||
-- Nothing else needs repairing: uniqueness lives on the clustered PK,
|
||||
-- which the switch never touches, so a failed purge can no longer
|
||||
-- leave the live table without its idempotency enforcement.
|
||||
IF OBJECT_ID('dbo.[{stagingTableName}]', 'U') IS NOT NULL DROP TABLE dbo.[{stagingTableName}];
|
||||
|
||||
-- Idempotent index rebuild — covers the niche case where ROLLBACK
|
||||
-- failed to restore UX_AuditLog_EventId (or the failure happened
|
||||
-- AFTER the COMMIT, which shouldn't be possible inside this TRY
|
||||
-- but is cheap insurance). Without this, a failed switch could
|
||||
-- leave the live table without its idempotency-supporting index.
|
||||
IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = 'UX_AuditLog_EventId' AND object_id = OBJECT_ID('dbo.AuditLog'))
|
||||
CREATE UNIQUE NONCLUSTERED INDEX UX_AuditLog_EventId ON dbo.AuditLog (EventId) ON [PRIMARY];
|
||||
|
||||
-- Surface the original error to the caller — the purge actor logs
|
||||
-- and continues with the next boundary.
|
||||
THROW;
|
||||
|
||||
+36
-12
@@ -128,6 +128,37 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
|
||||
var groups = samples.GroupBy(s => new SeriesHourKey(
|
||||
s.Source, s.Metric, s.Scope, s.ScopeKey, TruncateToHour(s.CapturedAtUtc)));
|
||||
|
||||
// Preload every rollup row already covering this window in ONE query and
|
||||
// index it by series+hour (WP2.2). The predecessor issued a
|
||||
// FirstOrDefaultAsync existence probe PER (series, hour) group — an N+1
|
||||
// that scaled with the metric catalogue times the lookback: a 3 h re-fold
|
||||
// over ~40 series cost ~120 sequential round trips before a single row was
|
||||
// written. The window is bounded by the caller's small trailing lookback
|
||||
// and the rollup table holds exactly one row per series-hour, so the
|
||||
// preload is a narrow range seek on IX_KpiRollupHourly_Series.
|
||||
//
|
||||
// Deliberately TRACKED (not a projection): the re-fold path mutates the
|
||||
// existing entity in place and relies on the change tracker to emit the
|
||||
// UPDATE. The dictionary's ScopeKey comparison is ordinal where the
|
||||
// previous SQL predicate used the database collation; both sides are
|
||||
// written from the same KpiSample.ScopeKey values, so they are
|
||||
// byte-identical in practice and a residual mismatch degrades to the
|
||||
// already-handled upsert-race path rather than a wrong aggregate.
|
||||
var existingRollups = await _context.KpiRollupHourly
|
||||
.Where(r => r.HourStartUtc >= from && r.HourStartUtc < to)
|
||||
.ToListAsync(cancellationToken);
|
||||
|
||||
var existingByKey = new Dictionary<SeriesHourKey, KpiRollupHourly>(existingRollups.Count);
|
||||
foreach (var row in existingRollups)
|
||||
{
|
||||
existingByKey[new SeriesHourKey(
|
||||
row.Source,
|
||||
row.Metric,
|
||||
row.Scope,
|
||||
row.ScopeKey,
|
||||
DateTime.SpecifyKind(row.HourStartUtc, DateTimeKind.Utc))] = row;
|
||||
}
|
||||
|
||||
foreach (var group in groups)
|
||||
{
|
||||
var key = group.Key;
|
||||
@@ -143,18 +174,11 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
|
||||
var maxValue = group.Max(s => s.Value);
|
||||
var sampleCount = group.Count();
|
||||
|
||||
// Idempotent upsert on the unique series+hour key. The ScopeKey == key.ScopeKey
|
||||
// comparison matches null against the Global-scope rows (IS NULL) exactly as the
|
||||
// UNIQUE IX_KpiRollupHourly_Series index treats a null key as participating.
|
||||
var existing = await _context.KpiRollupHourly.FirstOrDefaultAsync(
|
||||
r => r.Source == key.Source
|
||||
&& r.Metric == key.Metric
|
||||
&& r.Scope == key.Scope
|
||||
&& r.ScopeKey == key.ScopeKey
|
||||
&& r.HourStartUtc == key.HourStartUtc,
|
||||
cancellationToken);
|
||||
|
||||
if (existing is null)
|
||||
// Idempotent upsert on the unique series+hour key, resolved against the
|
||||
// preloaded dictionary. A null ScopeKey keys the Global-scope rows exactly
|
||||
// as the UNIQUE IX_KpiRollupHourly_Series index treats a null key as
|
||||
// participating.
|
||||
if (!existingByKey.TryGetValue(key, out var existing))
|
||||
{
|
||||
_context.KpiRollupHourly.Add(new KpiRollupHourly
|
||||
{
|
||||
|
||||
+74
-7
@@ -140,7 +140,14 @@ VALUES
|
||||
public async Task<IReadOnlyList<Notification>> GetDueAsync(
|
||||
DateTimeOffset now, int batchSize, CancellationToken cancellationToken = default)
|
||||
{
|
||||
// AsNoTracking (WP2.2): the dispatcher mutates each row's delivery state in
|
||||
// memory and persists it through UpdateAsync, which is now a targeted
|
||||
// server-side ExecuteUpdate and needs no change tracker. Tracking a batch
|
||||
// of notifications — each carrying an nvarchar(max) Body and TypeData —
|
||||
// paid for a full snapshot copy per row plus a DetectChanges scan of the
|
||||
// whole batch on every save.
|
||||
return await _context.Notifications
|
||||
.AsNoTracking()
|
||||
.Where(n => n.Status == NotificationStatus.Pending
|
||||
|| (n.Status == NotificationStatus.Retrying
|
||||
&& n.NextAttemptAt != null
|
||||
@@ -150,22 +157,59 @@ VALUES
|
||||
.ToListAsync(cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<Notification?> GetByIdAsync(string notificationId, CancellationToken cancellationToken = default)
|
||||
=> await _context.Notifications.FindAsync(new object[] { notificationId }, cancellationToken);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task UpdateAsync(Notification n, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_context.Notifications.Update(n);
|
||||
await _context.SaveChangesAsync(cancellationToken);
|
||||
ArgumentNullException.ThrowIfNull(n);
|
||||
|
||||
// Targeted server-side UPDATE of the seven mutable delivery-state columns
|
||||
// (WP2.2). The predecessor called DbSet.Update(n) + SaveChanges, which
|
||||
// marks EVERY property modified and rewrites all 21 columns — including
|
||||
// the immutable nvarchar(max) Body/TypeData payloads — on every single
|
||||
// delivery attempt. ExecuteUpdate also bypasses the change tracker
|
||||
// entirely, so it composes with the untracked GetDueAsync read.
|
||||
//
|
||||
// Immutable-by-contract columns (NotificationId, Type, ListName, Subject,
|
||||
// Body, TypeData, Source*, Origin*, SiteEnqueuedAt, CreatedAt) are
|
||||
// deliberately absent — see the interface contract: nothing in the
|
||||
// notification lifecycle ever changes them, and omitting them is what
|
||||
// makes the write narrow.
|
||||
var status = n.Status;
|
||||
var retryCount = n.RetryCount;
|
||||
var lastError = n.LastError;
|
||||
var resolvedTargets = n.ResolvedTargets;
|
||||
var lastAttemptAt = n.LastAttemptAt;
|
||||
var nextAttemptAt = n.NextAttemptAt;
|
||||
var deliveredAt = n.DeliveredAt;
|
||||
|
||||
var notificationId = n.NotificationId;
|
||||
|
||||
await _context.Notifications
|
||||
.Where(row => row.NotificationId == notificationId)
|
||||
.ExecuteUpdateAsync(
|
||||
setters => setters
|
||||
.SetProperty(row => row.Status, status)
|
||||
.SetProperty(row => row.RetryCount, retryCount)
|
||||
.SetProperty(row => row.LastError, lastError)
|
||||
.SetProperty(row => row.ResolvedTargets, resolvedTargets)
|
||||
.SetProperty(row => row.LastAttemptAt, lastAttemptAt)
|
||||
.SetProperty(row => row.NextAttemptAt, nextAttemptAt)
|
||||
.SetProperty(row => row.DeliveredAt, deliveredAt),
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<Notification?> GetByIdAsync(string notificationId, CancellationToken cancellationToken = default)
|
||||
=> await _context.Notifications.FindAsync(new object[] { notificationId }, cancellationToken);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<(IReadOnlyList<Notification> Rows, int TotalCount)> QueryAsync(
|
||||
NotificationOutboxFilter filter, int pageNumber, int pageSize, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var query = _context.Notifications.AsQueryable();
|
||||
// AsNoTracking (WP2.2): this is the Central UI's read-only list page. The
|
||||
// rows are projected to the wire and never saved, so tracking them cost a
|
||||
// snapshot copy of every nvarchar(max) Body in the page for nothing.
|
||||
var query = _context.Notifications.AsNoTracking().AsQueryable();
|
||||
|
||||
if (filter.Status is { } status)
|
||||
{
|
||||
@@ -218,8 +262,14 @@ VALUES
|
||||
|
||||
var totalCount = await query.CountAsync(cancellationToken);
|
||||
|
||||
// NotificationId breaks CreatedAt ties so the OFFSET window is deterministic —
|
||||
// without it two rows sharing a CreatedAt could appear on both pages or on
|
||||
// neither. (This page keeps OFFSET paging rather than the sibling repos'
|
||||
// keyset cursor because its contract surfaces a page number and a total
|
||||
// count, neither of which a keyset cursor can express.)
|
||||
var rows = await query
|
||||
.OrderByDescending(n => n.CreatedAt)
|
||||
.ThenByDescending(n => n.NotificationId)
|
||||
.Skip((pageNumber - 1) * pageSize)
|
||||
.Take(pageSize)
|
||||
.ToListAsync(cancellationToken);
|
||||
@@ -265,7 +315,24 @@ VALUES
|
||||
// One conditional-aggregation pass replaces four sequential COUNT round trips:
|
||||
// each metric is a COUNT(CASE WHEN <predicate> THEN 1 END) over the same scan
|
||||
// (arch-review 04). GroupBy(_ => 1) yields a single group (no rows → no group).
|
||||
//
|
||||
// WP2.2 — the aggregation is now PREDICATE-RESTRICTED instead of scanning the
|
||||
// whole table. Every KPI here is about the live queue (Pending/Retrying), the
|
||||
// parked backlog, or the last delivery interval; the overwhelming bulk of the
|
||||
// Notifications table is historical Delivered and Discarded rows that
|
||||
// contribute to NONE of them. The pre-filter below is the union of the
|
||||
// metric-contributing predicates, which lets the optimizer seek the status
|
||||
// index for the live/parked legs and the filtered
|
||||
// (DeliveredAt) WHERE Status='Delivered' index (WP1.4) for the interval leg,
|
||||
// instead of paying a full scan whose cost grows with retained history.
|
||||
// Mirrors the pre-filter ComputePerSiteKpisAsync/ComputePerNodeKpisAsync
|
||||
// already use — the global snapshot was the odd one out.
|
||||
var counts = await _context.Notifications
|
||||
.Where(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying
|
||||
|| n.Status == NotificationStatus.Parked
|
||||
|| (n.Status == NotificationStatus.Delivered
|
||||
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince))
|
||||
.GroupBy(_ => 1)
|
||||
.Select(g => new
|
||||
{
|
||||
|
||||
+68
-45
@@ -72,49 +72,36 @@ public class SiteCallAuditRepository : ISiteCallAuditRepository
|
||||
var idText = siteCall.TrackedOperationId.Value.ToString("D");
|
||||
var incomingRank = GetRankOrThrow(siteCall.Status);
|
||||
|
||||
// Step 1: insert-if-not-exists. Like AuditLogRepository.InsertIfNotExistsAsync
|
||||
// this is check-then-act so a duplicate-key violation may surface under
|
||||
// concurrent inserts on the same id — caught + logged at Debug.
|
||||
// ONE round trip, UPDATE-first (WP2.2). The predecessor issued an
|
||||
// unconditional IF NOT EXISTS … INSERT and THEN a monotonic UPDATE — two
|
||||
// statements, two round trips, on every single packet, of which the insert
|
||||
// half was wasted work for every packet after the first (the steady state:
|
||||
// a cached call emits Submitted → Forwarded → Attempted → terminal, so
|
||||
// three of four packets hit an existing row).
|
||||
//
|
||||
// SourceNode-stamping: the column is included in the INSERT
|
||||
// column list / VALUES so a fresh row carries the originating node
|
||||
// name (node-a/node-b for site rows). A null SourceNode (legacy hosts
|
||||
// / unstamped reconciled rows) writes NULL straight through.
|
||||
try
|
||||
{
|
||||
await _context.Database.ExecuteSqlInterpolatedAsync(
|
||||
$@"IF NOT EXISTS (SELECT 1 FROM dbo.SiteCalls WHERE TrackedOperationId = {idText})
|
||||
INSERT INTO dbo.SiteCalls
|
||||
(TrackedOperationId, Channel, Target, SourceSite, SourceNode, Status, RetryCount,
|
||||
LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc)
|
||||
VALUES
|
||||
({idText}, {siteCall.Channel}, {siteCall.Target}, {siteCall.SourceSite}, {siteCall.SourceNode}, {siteCall.Status}, {siteCall.RetryCount},
|
||||
{siteCall.LastError}, {siteCall.HttpStatus}, {siteCall.CreatedAtUtc}, {siteCall.UpdatedAtUtc}, {siteCall.TerminalAtUtc}, {siteCall.IngestedAtUtc});",
|
||||
ct);
|
||||
}
|
||||
catch (SqlException ex) when (
|
||||
ex.Number == SqlErrorUniqueIndexViolation
|
||||
|| ex.Number == SqlErrorPrimaryKeyViolation)
|
||||
{
|
||||
_logger.LogDebug(
|
||||
ex,
|
||||
"SiteCallAuditRepository.UpsertAsync swallowed duplicate-key violation (error {SqlErrorNumber}) for TrackedOperationId {TrackedOperationId}; falling through to monotonic update.",
|
||||
ex.Number,
|
||||
idText);
|
||||
}
|
||||
|
||||
// Step 2: monotonic update with a same-rank freshness tiebreaker. The
|
||||
// CASE expression maps the stored Status string to the same rank table
|
||||
// the caller uses. We mutate when EITHER the incoming rank is strictly
|
||||
// greater, OR the incoming rank equals the stored rank AND that rank is
|
||||
// non-terminal (< TerminalRank) AND the incoming UpdatedAtUtc is strictly
|
||||
// newer than the stored one — so a retrying call's Attempted-phase
|
||||
// RetryCount/LastError/HttpStatus stay live instead of freezing at the
|
||||
// first Attempted packet. Terminal ranks are excluded from the
|
||||
// tiebreaker, so a later terminal NEVER overwrites an earlier one; equal
|
||||
// stamps are inert (idempotent replay) and a lower rank is always a no-op.
|
||||
// The combined batch below runs UPDATE first and inserts only when the
|
||||
// UPDATE matched nothing AND the row genuinely does not exist. The
|
||||
// NOT EXISTS re-check is load-bearing: @@ROWCOUNT = 0 is ALSO what a
|
||||
// monotonic REJECTION looks like (a stale or regressive packet against an
|
||||
// existing row), and inserting there would resurrect a row the guard just
|
||||
// refused. Both statements ship in one command text, so this is one
|
||||
// round trip, not two.
|
||||
//
|
||||
// SourceNode-stamping: SourceNode is updated via
|
||||
// Monotonic update semantics are unchanged: mutate when EITHER the
|
||||
// incoming rank is strictly greater, OR the incoming rank equals the
|
||||
// stored rank AND that rank is non-terminal (< TerminalRank) AND the
|
||||
// incoming UpdatedAtUtc is strictly newer than the stored one — so a
|
||||
// retrying call's Attempted-phase RetryCount/LastError/HttpStatus stay
|
||||
// live instead of freezing at the first Attempted packet. Terminal ranks
|
||||
// are excluded from the tiebreaker, so a later terminal NEVER overwrites
|
||||
// an earlier one; equal stamps are inert (idempotent replay) and a lower
|
||||
// rank is always a no-op.
|
||||
//
|
||||
// SourceNode-stamping: the column is included in the INSERT column list /
|
||||
// VALUES so a fresh row carries the originating node name (node-a/node-b
|
||||
// for site rows). A null SourceNode (legacy hosts / unstamped reconciled
|
||||
// rows) writes NULL straight through. On the UPDATE leg SourceNode is
|
||||
// written via
|
||||
// COALESCE(@SourceNode, SourceNode). The operator returns @SourceNode
|
||||
// when it is non-null, otherwise the stored value — so the column
|
||||
// behaves protectively: a later packet that carries a null
|
||||
@@ -128,8 +115,12 @@ VALUES
|
||||
// lifecycle every packet should carry the same SourceNode value (one
|
||||
// execution, one node) so the "overwrite" path is in practice
|
||||
// idempotent.
|
||||
await _context.Database.ExecuteSqlInterpolatedAsync(
|
||||
$@"UPDATE dbo.SiteCalls
|
||||
try
|
||||
{
|
||||
await _context.Database.ExecuteSqlInterpolatedAsync(
|
||||
$@"DECLARE @updated int;
|
||||
|
||||
UPDATE dbo.SiteCalls
|
||||
SET Status = {siteCall.Status},
|
||||
RetryCount = {siteCall.RetryCount},
|
||||
LastError = {siteCall.LastError},
|
||||
@@ -162,8 +153,40 @@ WHERE TrackedOperationId = {idText}
|
||||
ELSE -1
|
||||
END)
|
||||
AND {incomingRank} < {TerminalRank}
|
||||
AND UpdatedAtUtc < {siteCall.UpdatedAtUtc} ) );",
|
||||
ct);
|
||||
AND UpdatedAtUtc < {siteCall.UpdatedAtUtc} ) );
|
||||
|
||||
-- Captured IMMEDIATELY after the UPDATE: @@ROWCOUNT is reset by the next
|
||||
-- statement, and reading it inline inside a compound IF condition alongside a
|
||||
-- subquery is not safe (the subquery's own execution can clobber it).
|
||||
SET @updated = @@ROWCOUNT;
|
||||
|
||||
IF @updated = 0 AND NOT EXISTS (SELECT 1 FROM dbo.SiteCalls WHERE TrackedOperationId = {idText})
|
||||
INSERT INTO dbo.SiteCalls
|
||||
(TrackedOperationId, Channel, Target, SourceSite, SourceNode, Status, RetryCount,
|
||||
LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc)
|
||||
VALUES
|
||||
({idText}, {siteCall.Channel}, {siteCall.Target}, {siteCall.SourceSite}, {siteCall.SourceNode}, {siteCall.Status}, {siteCall.RetryCount},
|
||||
{siteCall.LastError}, {siteCall.HttpStatus}, {siteCall.CreatedAtUtc}, {siteCall.UpdatedAtUtc}, {siteCall.TerminalAtUtc}, {siteCall.IngestedAtUtc});",
|
||||
ct);
|
||||
}
|
||||
catch (SqlException ex) when (
|
||||
ex.Number == SqlErrorUniqueIndexViolation
|
||||
|| ex.Number == SqlErrorPrimaryKeyViolation)
|
||||
{
|
||||
// Two concurrent sessions both found the row absent and both raced to
|
||||
// INSERT; the loser raises 2601/2627 against the TrackedOperationId
|
||||
// primary key. The winner's row IS the first-write, and this packet's
|
||||
// content is by construction the same lifecycle state, so the race
|
||||
// outcome is semantically a no-op. Swallow at Debug — the same
|
||||
// check-then-act contract the sibling AuditLog/Notification repos
|
||||
// document. Note the loser's UPDATE leg already ran (against no row),
|
||||
// so nothing is left half-applied.
|
||||
_logger.LogDebug(
|
||||
ex,
|
||||
"SiteCallAuditRepository.UpsertAsync swallowed duplicate-key violation (error {SqlErrorNumber}) for TrackedOperationId {TrackedOperationId}; treating as no-op.",
|
||||
ex.Number,
|
||||
idText);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -26,6 +26,22 @@ public static class ServiceCollectionExtensions
|
||||
// registers IDataProtectionProvider as a singleton; resolving it here does not recurse
|
||||
// because key-ring loading is lazy (first Protect/Unprotect), not triggered by
|
||||
// CreateProtector during model building.
|
||||
//
|
||||
// POOLING IS DELIBERATELY NOT USED (WP2.2 — verified, not overlooked).
|
||||
// AddDbContextPool requires a context with a SINGLE public constructor taking
|
||||
// only DbContextOptions<TContext>; EF Core constructs pooled instances through
|
||||
// its own activator and cannot supply anything else. ScadaBridgeDbContext has
|
||||
// two public constructors and the runtime one takes IDataProtectionProvider,
|
||||
// because the encrypting value converter for secret-bearing columns is built
|
||||
// during OnModelCreating from that provider. Worse, the model itself DIFFERS
|
||||
// between the two constructors (no provider ⇒ no encrypting converter), so a
|
||||
// pooled activator would silently produce a context that reads secret columns
|
||||
// as ciphertext. Making this poolable means moving the protector out of the
|
||||
// constructor and into a DbContextOptions extension — a change to the
|
||||
// secrets-at-rest path, which is not a performance refactor. The registration
|
||||
// below (a scoped factory overriding AddDbContext's activator) is what makes
|
||||
// the provider reach the context at all, and it also bypasses pooling by
|
||||
// construction. Revisit only alongside a deliberate secrets-plumbing change.
|
||||
services.AddDbContext<ScadaBridgeDbContext>((serviceProvider, options) =>
|
||||
{
|
||||
options.UseSqlServer(
|
||||
|
||||
@@ -158,6 +158,26 @@ public class SiteCallAuditActor : ReceiveActor
|
||||
/// </summary>
|
||||
private readonly Dictionary<string, bool> _reconciliationPinned = new();
|
||||
|
||||
/// <summary>
|
||||
/// Actor-system EventStream captured on the actor thread at handler-registration
|
||||
/// time. The reconciliation pass runs off-mailbox and publishes the pinned-state
|
||||
/// transition from there, so it must not reach through <c>Context</c>.
|
||||
/// </summary>
|
||||
private Akka.Event.EventStream _eventStream = null!;
|
||||
|
||||
/// <summary>
|
||||
/// Single-flight guard for the off-mailbox reconciliation pass. Raised on the
|
||||
/// actor thread when a tick launches a pass and lowered on the actor thread
|
||||
/// when the piped <see cref="ReconciliationComplete"/> arrives, so the
|
||||
/// per-site cursor/pinned dictionaries the pass mutates are only ever touched
|
||||
/// by one task at a time and the mailbox supplies the memory barrier between
|
||||
/// consecutive passes.
|
||||
/// </summary>
|
||||
private bool _reconciling;
|
||||
|
||||
/// <summary>Single-flight guard for the off-mailbox terminal-row purge pass.</summary>
|
||||
private bool _purging;
|
||||
|
||||
private ICancelable? _reconciliationTimer;
|
||||
private ICancelable? _purgeTimer;
|
||||
|
||||
@@ -338,8 +358,79 @@ public class SiteCallAuditActor : ReceiveActor
|
||||
// the daily terminal-row purge. Handlers stay alive across faults via
|
||||
// their own per-site / per-tick try/catch (mirroring the ingest path);
|
||||
// the timers are only started when their collaborators are available.
|
||||
ReceiveAsync<ReconciliationTick>(_ => OnReconciliationTickAsync());
|
||||
ReceiveAsync<PurgeTick>(_ => OnPurgeTickAsync());
|
||||
//
|
||||
// OFF-MAILBOX (WP2.2). Both passes run as PipeTo-completed background
|
||||
// tasks behind a single-flight guard rather than as ReceiveAsync bodies.
|
||||
// A ReceiveAsync handler occupies the actor for its whole duration, and a
|
||||
// reconciliation pass is unbounded work — every site, up to
|
||||
// MaxReconciliationPagesPerTick network pulls each, one upsert per row.
|
||||
// Post-outage catch-up therefore blocked telemetry ingest, UI queries and
|
||||
// KPI Asks behind it until the drain finished, and those callers timed out
|
||||
// rather than queued. NotificationOutboxActor's dispatch sweep is the
|
||||
// in-repo reference for this shape.
|
||||
Receive<ReconciliationTick>(_ => HandleReconciliationTick());
|
||||
Receive<ReconciliationComplete>(_ => _reconciling = false);
|
||||
Receive<PurgeTick>(_ => HandlePurgeTick());
|
||||
Receive<PurgeComplete>(_ => _purging = false);
|
||||
|
||||
// Captured on the actor thread so the background passes never touch
|
||||
// Context off-thread. EventStream itself is thread-safe.
|
||||
_eventStream = Context.System.EventStream;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Launches a reconciliation pass unless one is already in flight, dropping
|
||||
/// the tick if so. Overlapping passes are not merely wasteful — they would
|
||||
/// race on the per-site cursor and pinned-latch dictionaries, which the
|
||||
/// single-flight guard keeps confined to one task at a time (the guard itself
|
||||
/// is only ever mutated on the actor thread: raised here, lowered by the
|
||||
/// piped completion message).
|
||||
/// </summary>
|
||||
private void HandleReconciliationTick()
|
||||
{
|
||||
if (_reconciling)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_reconciling = true;
|
||||
|
||||
// OnReconciliationTickAsync swallows its own per-site errors, but the
|
||||
// failure projection is kept as a belt-and-braces guard so even a faulted
|
||||
// task still lowers the guard — otherwise reconciliation would wedge
|
||||
// permanently after a single unexpected throw.
|
||||
OnReconciliationTickAsync().PipeTo(
|
||||
Self,
|
||||
success: () => ReconciliationComplete.Instance,
|
||||
failure: ex =>
|
||||
{
|
||||
_logger.LogError(ex, "SiteCallAudit reconciliation pass faulted unexpectedly.");
|
||||
return ReconciliationComplete.Instance;
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Launches a purge pass unless one is already in flight. Same single-flight
|
||||
/// discipline as <see cref="HandleReconciliationTick"/>: a purge that outlives
|
||||
/// its interval (a large catch-up after an outage) must not stack.
|
||||
/// </summary>
|
||||
private void HandlePurgeTick()
|
||||
{
|
||||
if (_purging)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_purging = true;
|
||||
|
||||
OnPurgeTickAsync().PipeTo(
|
||||
Self,
|
||||
success: () => PurgeComplete.Instance,
|
||||
failure: ex =>
|
||||
{
|
||||
_logger.LogError(ex, "SiteCallAudit purge pass faulted unexpectedly.");
|
||||
return PurgeComplete.Instance;
|
||||
});
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -743,7 +834,10 @@ public class SiteCallAuditActor : ReceiveActor
|
||||
}
|
||||
|
||||
_reconciliationPinned[siteId] = pinned;
|
||||
Context.System.EventStream.Publish(new SiteCallReconciliationPinnedChanged(siteId, pinned));
|
||||
|
||||
// _eventStream, not Context.System.EventStream: this runs on the
|
||||
// off-mailbox reconciliation pass, where Context must not be touched.
|
||||
_eventStream.Publish(new SiteCallReconciliationPinnedChanged(siteId, pinned));
|
||||
}
|
||||
|
||||
// ── Piece B: daily terminal-row purge scheduler ──
|
||||
@@ -1452,6 +1546,24 @@ public class SiteCallAuditActor : ReceiveActor
|
||||
public static readonly PurgeTick Instance = new();
|
||||
private PurgeTick() { }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Piped back to <c>Self</c> when an off-mailbox reconciliation pass ends
|
||||
/// (successfully or not) so the single-flight guard is lowered on the actor
|
||||
/// thread rather than from the background task.
|
||||
/// </summary>
|
||||
internal sealed class ReconciliationComplete
|
||||
{
|
||||
public static readonly ReconciliationComplete Instance = new();
|
||||
private ReconciliationComplete() { }
|
||||
}
|
||||
|
||||
/// <summary>Purge counterpart of <see cref="ReconciliationComplete"/>.</summary>
|
||||
internal sealed class PurgeComplete
|
||||
{
|
||||
public static readonly PurgeComplete Instance = new();
|
||||
private PurgeComplete() { }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
Reference in New Issue
Block a user