fix(config-db): EnableRetryOnFailure fleet-wide + execution-strategy wrap of the combined-telemetry dual-write
Installs SqlServerRetryingExecutionStrategy (5 retries, 30s cap) so transient SQL faults retry transparently on every read-side path. Manual transactions aren't auto-retried, so AuditLogIngestActor's idempotent dual-write is wrapped in an explicit CreateExecutionStrategy().ExecuteAsync to become retriable. Verified empirically that existing manual transactions (BundleImporter) are unaffected — EF skips the strategy while a transaction is already active.
This commit is contained in:
@@ -1,4 +1,5 @@
|
|||||||
using Akka.Actor;
|
using Akka.Actor;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
using Microsoft.Extensions.DependencyInjection;
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
using ZB.MOM.WW.Audit;
|
using ZB.MOM.WW.Audit;
|
||||||
@@ -259,37 +260,51 @@ public class AuditLogIngestActor : ReceiveActor
|
|||||||
// skip the registration.
|
// skip the registration.
|
||||||
var failureCounter = scope.ServiceProvider.GetService<ICentralAuditWriteFailureCounter>();
|
var failureCounter = scope.ServiceProvider.GetService<ICentralAuditWriteFailureCounter>();
|
||||||
|
|
||||||
|
// Task 12 (arch-review 04 S5): retry the dual-write under the DbContext's
|
||||||
|
// configured resiliency strategy. A manual (user-initiated) transaction is
|
||||||
|
// NOT auto-retried by EnableRetryOnFailure — EF executes directly while a
|
||||||
|
// transaction is active — so the whole { BEGIN; insert; upsert; COMMIT }
|
||||||
|
// unit is wrapped here to become a single retriable operation. The unit is
|
||||||
|
// idempotent (InsertIfNotExists on EventId + monotonic upsert on
|
||||||
|
// TrackedOperationId), so replay after a rolled-back transient failure is
|
||||||
|
// safe. NoOpExecutionStrategy on non-configured/test contexts runs it once.
|
||||||
|
var strategy = dbContext.Database.CreateExecutionStrategy();
|
||||||
|
|
||||||
foreach (var entry in cmd.Entries)
|
foreach (var entry in cmd.Entries)
|
||||||
{
|
{
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
await using var tx = await dbContext.Database
|
await strategy.ExecuteAsync(async () =>
|
||||||
.BeginTransactionAsync()
|
{
|
||||||
.ConfigureAwait(false);
|
await using var tx = await dbContext.Database
|
||||||
|
.BeginTransactionAsync()
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
// Stamp IngestedAtUtc on both rows from a single
|
// Stamp IngestedAtUtc on both rows from a single
|
||||||
// central-side instant so a join on the two tables sees
|
// central-side instant so a join on the two tables sees
|
||||||
// matching timestamps (debugging convenience, not a
|
// matching timestamps (debugging convenience, not a
|
||||||
// correctness invariant).
|
// correctness invariant).
|
||||||
var ingestedAt = DateTime.UtcNow;
|
var ingestedAt = DateTime.UtcNow;
|
||||||
// Redact the audit half BEFORE the dual-write — only the
|
// Redact the audit half BEFORE the dual-write — only the
|
||||||
// AuditLog row's payload columns are redactable; SiteCalls
|
// AuditLog row's payload columns are redactable; SiteCalls
|
||||||
// carries operational state only (status, retry count) and
|
// carries operational state only (status, retry count) and
|
||||||
// is left untouched. Null redactor falls back
|
// is left untouched. Null redactor falls back
|
||||||
// to SafeDefault so header redaction always runs.
|
// to SafeDefault so header redaction always runs.
|
||||||
// IngestedAtUtc is a DetailsJson field
|
// IngestedAtUtc is a DetailsJson field
|
||||||
// on the canonical record, so stamp it via the projection helper.
|
// on the canonical record, so stamp it via the projection helper.
|
||||||
var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
|
var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
|
||||||
var filteredAudit = safeRedactor.Apply(entry.Audit);
|
var filteredAudit = safeRedactor.Apply(entry.Audit);
|
||||||
var auditStamped = AuditRowProjection.WithIngestedAtUtc(filteredAudit, ingestedAt);
|
var auditStamped = AuditRowProjection.WithIngestedAtUtc(filteredAudit, ingestedAt);
|
||||||
var siteCallStamped = entry.SiteCall with { IngestedAtUtc = ingestedAt };
|
var siteCallStamped = entry.SiteCall with { IngestedAtUtc = ingestedAt };
|
||||||
|
|
||||||
await auditRepo.InsertIfNotExistsAsync(auditStamped)
|
await auditRepo.InsertIfNotExistsAsync(auditStamped)
|
||||||
.ConfigureAwait(false);
|
.ConfigureAwait(false);
|
||||||
await siteCallRepo.UpsertAsync(siteCallStamped)
|
await siteCallRepo.UpsertAsync(siteCallStamped)
|
||||||
.ConfigureAwait(false);
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
await tx.CommitAsync().ConfigureAwait(false);
|
||||||
|
}).ConfigureAwait(false);
|
||||||
|
|
||||||
await tx.CommitAsync().ConfigureAwait(false);
|
|
||||||
accepted.Add(entry.Audit.EventId);
|
accepted.Add(entry.Audit.EventId);
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
|
|||||||
@@ -229,6 +229,13 @@ VALUES
|
|||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public async Task<long> SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default)
|
public async Task<long> SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default)
|
||||||
{
|
{
|
||||||
|
// Task 12 (arch-review 04 S5) note: the drop-and-rebuild 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.
|
||||||
|
//
|
||||||
// Maintenance timeout in whole seconds (ADO.NET CommandTimeout unit). Null leaves the
|
// Maintenance timeout in whole seconds (ADO.NET CommandTimeout unit). Null leaves the
|
||||||
// provider default in place. See AuditLogPurgeOptions.MaintenanceCommandTimeoutMinutes /
|
// provider default in place. See AuditLogPurgeOptions.MaintenanceCommandTimeoutMinutes /
|
||||||
// arch-review 04 S2 for why the ~30s default is unsafe for the switch-out dance.
|
// arch-review 04 S2 for why the ~30s default is unsafe for the switch-out dance.
|
||||||
|
|||||||
@@ -28,7 +28,22 @@ public static class ServiceCollectionExtensions
|
|||||||
// CreateProtector during model building.
|
// CreateProtector during model building.
|
||||||
services.AddDbContext<ScadaBridgeDbContext>((serviceProvider, options) =>
|
services.AddDbContext<ScadaBridgeDbContext>((serviceProvider, options) =>
|
||||||
{
|
{
|
||||||
options.UseSqlServer(connectionString)
|
options.UseSqlServer(
|
||||||
|
connectionString,
|
||||||
|
// Task 12 (arch-review 04 S5): connection resiliency. A transient
|
||||||
|
// SQL fault (failover, throttling, dropped connection) previously
|
||||||
|
// surfaced raw to every read-side caller — KPI asks, outbox
|
||||||
|
// queries, Tracking.Status, execution-tree walks. EnableRetryOnFailure
|
||||||
|
// installs the SqlServerRetryingExecutionStrategy so those retry
|
||||||
|
// transparently. Note: manual (user-initiated) transactions are NOT
|
||||||
|
// auto-retried — EF executes them directly while a transaction is
|
||||||
|
// active — so the combined-telemetry dual-write is wrapped in an
|
||||||
|
// explicit CreateExecutionStrategy() to become retriable (see
|
||||||
|
// AuditLogIngestActor.OnCachedTelemetryAsync).
|
||||||
|
sql => sql.EnableRetryOnFailure(
|
||||||
|
maxRetryCount: 5,
|
||||||
|
maxRetryDelay: TimeSpan.FromSeconds(30),
|
||||||
|
errorNumbersToAdd: null))
|
||||||
.ConfigureWarnings(w => w.Ignore(
|
.ConfigureWarnings(w => w.Ignore(
|
||||||
Microsoft.EntityFrameworkCore.Diagnostics.RelationalEventId.PendingModelChangesWarning));
|
Microsoft.EntityFrameworkCore.Diagnostics.RelationalEventId.PendingModelChangesWarning));
|
||||||
});
|
});
|
||||||
|
|||||||
+28
-1
@@ -1,4 +1,5 @@
|
|||||||
using System.Reflection;
|
using System.Reflection;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
using Microsoft.Extensions.DependencyInjection;
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||||
@@ -13,7 +14,7 @@ public class ServiceCollectionExtensionsTests
|
|||||||
{
|
{
|
||||||
var services = new ServiceCollection();
|
var services = new ServiceCollection();
|
||||||
|
|
||||||
services.AddConfigurationDatabase("DataSource=:memory:");
|
services.AddConfigurationDatabase("Server=unused;Database=unused;");
|
||||||
|
|
||||||
Assert.Contains(services, d => d.ServiceType == typeof(ITemplateEngineRepository));
|
Assert.Contains(services, d => d.ServiceType == typeof(ITemplateEngineRepository));
|
||||||
Assert.Contains(services, d => d.ServiceType == typeof(IAuditService));
|
Assert.Contains(services, d => d.ServiceType == typeof(IAuditService));
|
||||||
@@ -21,6 +22,32 @@ public class ServiceCollectionExtensionsTests
|
|||||||
Assert.Contains(services, d => d.ServiceType == typeof(IAuditLogRepository));
|
Assert.Contains(services, d => d.ServiceType == typeof(IAuditLogRepository));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void AddConfigurationDatabase_Configures_RetryingExecutionStrategy()
|
||||||
|
{
|
||||||
|
// Task 12 (arch-review 04 S5): the central DbContext must survive transient
|
||||||
|
// SQL faults. EnableRetryOnFailure installs a non-null ExecutionStrategyFactory
|
||||||
|
// on the SqlServer options extension — covering every read-side path (KPI asks,
|
||||||
|
// outbox queries, Tracking.Status, execution trees) that previously surfaced
|
||||||
|
// transient faults raw.
|
||||||
|
var services = new ServiceCollection();
|
||||||
|
services.AddConfigurationDatabase("Server=unused;Database=unused;");
|
||||||
|
using var sp = services.BuildServiceProvider();
|
||||||
|
|
||||||
|
var options = sp.GetRequiredService<DbContextOptions<ScadaBridgeDbContext>>();
|
||||||
|
|
||||||
|
// EF1001: SqlServerOptionsExtension is an internal EF surface, but it is the
|
||||||
|
// only place the retrying-strategy factory is observable — asserting it is the
|
||||||
|
// point of this test. Scoped suppression, not a project-wide one.
|
||||||
|
#pragma warning disable EF1001
|
||||||
|
var sqlExt = options.Extensions
|
||||||
|
.OfType<Microsoft.EntityFrameworkCore.SqlServer.Infrastructure.Internal.SqlServerOptionsExtension>()
|
||||||
|
.Single();
|
||||||
|
|
||||||
|
Assert.NotNull(sqlExt.ExecutionStrategyFactory);
|
||||||
|
#pragma warning restore EF1001
|
||||||
|
}
|
||||||
|
|
||||||
// The no-arg overload is [Obsolete(error: true)], so it cannot be referenced directly
|
// The no-arg overload is [Obsolete(error: true)], so it cannot be referenced directly
|
||||||
// from source — that is the compile-time guard. Invoke it via reflection to verify the
|
// from source — that is the compile-time guard. Invoke it via reflection to verify the
|
||||||
// runtime defence-in-depth behaviour.
|
// runtime defence-in-depth behaviour.
|
||||||
|
|||||||
Reference in New Issue
Block a user