Wires Bundle E of the M2 site-sync pipeline: - AddAuditLog extended to register the site writer chain (SqliteAuditWriter singleton + ISiteAuditQueue forward + RingBufferFallback + FallbackAuditWriter composing them) and the telemetry collaborators (SiteAuditTelemetryOptions, SqliteAuditWriterOptions, IAuditWriteFailureCounter NoOp default, ISiteStreamAuditClient NoOp default). - AkkaHostedService central role: AuditLogIngestActor as ClusterSingletonManager (singleton name 'audit-log-ingest') + ClusterSingletonProxy, mirroring the Notification Outbox pattern. Proxy is offered to SiteStreamGrpcServer if it resolves (Site path only today; M6 reconciliation will host gRPC on central). - AkkaHostedService site role: SiteAuditTelemetryActor (per-site, NOT a singleton because each site is its own cluster), bound to a dedicated audit-telemetry-dispatcher (ForkJoinDispatcher, 2 dedicated threads). - Program.cs + SiteServiceRegistration.Configure call AddAuditLog on both roles. - AuditLogIngestActor gains a second constructor that takes IServiceProvider so the cluster singleton can create a fresh scope per message — IAuditLogRepository is a scoped EF Core service and cannot be pre-resolved from the root. The IAuditLogRepository constructor remains for Bundle D's MSSQL-fixture tests. NoOp ISiteStreamAuditClient is deliberate: no site→central gRPC channel exists in M2 (sites talk to central via Akka ClusterClient; gRPC SiteStreamService is hosted on sites for central→site streaming). M6 reconciliation introduces the real gRPC site→central client + central-hosted gRPC server. Bundle H's integration test substitutes a stub client directly via the actor's Props. Tests: - tests/ScadaLink.AuditLog.Tests/AddAuditLogTests.cs — 11 tests (was 3): writer singleton, IAuditWriter as FallbackAuditWriter, ISiteAuditQueue same-instance as SqliteAuditWriter, options bind round-trip, NoOp default assertions. - tests/ScadaLink.Host.Tests/AkkaHostedServiceAuditWiringTests.cs (new) — 13 tests: BuildHocon emits audit-telemetry-dispatcher block with the expected type/throughput/thread-count; Central composition root resolves the writer chain + options; Site composition root resolves the writer chain + options + NoOp client. Verified: dotnet build clean, 23 test suites green (Host 194 + AuditLog 54).
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
using Akka.Actor;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ScadaLink.Commons.Entities.Audit;
|
||||
using ScadaLink.Commons.Interfaces.Repositories;
|
||||
@@ -28,12 +29,27 @@ namespace ScadaLink.AuditLog.Central;
|
||||
/// inside <c>ReceiveAsync</c> does not restart the actor (which would also
|
||||
/// reset any in-flight state).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Two constructors exist for a deliberate reason: Bundle D's tests inject a
|
||||
/// concrete <see cref="IAuditLogRepository"/> against a per-test MSSQL fixture
|
||||
/// (the only way to verify the IngestedAtUtc stamp + duplicate-key idempotency
|
||||
/// end to end), while Bundle E's host wiring registers the actor as a cluster
|
||||
/// singleton and must therefore resolve the repository — which is a scoped EF
|
||||
/// Core service — from a fresh DI scope per message. Mirroring the Notification
|
||||
/// Outbox actor's pattern.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public class AuditLogIngestActor : ReceiveActor
|
||||
{
|
||||
private readonly IAuditLogRepository _repository;
|
||||
private readonly IServiceProvider? _serviceProvider;
|
||||
private readonly IAuditLogRepository? _injectedRepository;
|
||||
private readonly ILogger<AuditLogIngestActor> _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Test-mode constructor — injects a concrete repository instance whose
|
||||
/// lifetime exceeds the test, so the actor reuses the same instance across
|
||||
/// every message. Used by Bundle D's MSSQL-backed TestKit fixture.
|
||||
/// </summary>
|
||||
public AuditLogIngestActor(
|
||||
IAuditLogRepository repository,
|
||||
ILogger<AuditLogIngestActor> logger)
|
||||
@@ -41,7 +57,27 @@ public class AuditLogIngestActor : ReceiveActor
|
||||
ArgumentNullException.ThrowIfNull(repository);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
|
||||
_repository = repository;
|
||||
_injectedRepository = repository;
|
||||
_logger = logger;
|
||||
|
||||
ReceiveAsync<IngestAuditEventsCommand>(OnIngestAsync);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Production constructor — resolves <see cref="IAuditLogRepository"/> from
|
||||
/// a fresh DI scope per message because the repository is a scoped EF Core
|
||||
/// service registered by <c>AddConfigurationDatabase</c>. The actor itself
|
||||
/// is a long-lived cluster singleton, so it cannot hold a scope across
|
||||
/// messages.
|
||||
/// </summary>
|
||||
public AuditLogIngestActor(
|
||||
IServiceProvider serviceProvider,
|
||||
ILogger<AuditLogIngestActor> logger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(serviceProvider);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
|
||||
_serviceProvider = serviceProvider;
|
||||
_logger = logger;
|
||||
|
||||
ReceiveAsync<IngestAuditEventsCommand>(OnIngestAsync);
|
||||
@@ -68,27 +104,49 @@ public class AuditLogIngestActor : ReceiveActor
|
||||
var nowUtc = DateTime.UtcNow;
|
||||
var accepted = new List<Guid>(cmd.Events.Count);
|
||||
|
||||
foreach (var evt in cmd.Events)
|
||||
// Resolve the repository for the whole batch — one DbContext per
|
||||
// message, mirroring NotificationOutboxActor. The injected-repository
|
||||
// mode (Bundle D tests) skips the scope entirely.
|
||||
IServiceScope? scope = null;
|
||||
IAuditLogRepository repository;
|
||||
if (_injectedRepository is not null)
|
||||
{
|
||||
try
|
||||
repository = _injectedRepository;
|
||||
}
|
||||
else
|
||||
{
|
||||
scope = _serviceProvider!.CreateScope();
|
||||
repository = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
foreach (var evt in cmd.Events)
|
||||
{
|
||||
// Stamp IngestedAtUtc here, not at the site. Bundle A's
|
||||
// repository hardening already swallows duplicate-key races,
|
||||
// so the same id arriving twice (site retry, reconciliation)
|
||||
// is a silent no-op.
|
||||
var ingested = evt with { IngestedAtUtc = nowUtc };
|
||||
await _repository.InsertIfNotExistsAsync(ingested).ConfigureAwait(false);
|
||||
accepted.Add(evt.EventId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Per-row catch — one bad row never sinks the whole batch.
|
||||
// The row stays Pending at the site; the next drain retries.
|
||||
_logger.LogError(ex,
|
||||
"Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.",
|
||||
evt.EventId);
|
||||
try
|
||||
{
|
||||
// Stamp IngestedAtUtc here, not at the site. Bundle A's
|
||||
// repository hardening already swallows duplicate-key races,
|
||||
// so the same id arriving twice (site retry, reconciliation)
|
||||
// is a silent no-op.
|
||||
var ingested = evt with { IngestedAtUtc = nowUtc };
|
||||
await repository.InsertIfNotExistsAsync(ingested).ConfigureAwait(false);
|
||||
accepted.Add(evt.EventId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Per-row catch — one bad row never sinks the whole batch.
|
||||
// The row stays Pending at the site; the next drain retries.
|
||||
_logger.LogError(ex,
|
||||
"Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.",
|
||||
evt.EventId);
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
scope?.Dispose();
|
||||
}
|
||||
|
||||
replyTo.Tell(new IngestAuditEventsReply(accepted));
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user