merge: test-infra hardening (register rows 32, 40, 61) — queue-depth gauge collection isolation, wall-clock sleeps replaced with observable edges + negative controls, _injectedRepository test ctors deleted with empty guard allowlist
This commit is contained in:
@@ -42,13 +42,14 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
|||||||
/// placeholder.
|
/// placeholder.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// Two constructors exist for a deliberate reason: the test ctor injects a
|
/// There is exactly ONE constructor, taking an <see cref="IServiceProvider"/>:
|
||||||
/// concrete <see cref="IAuditLogRepository"/> against a per-test MSSQL fixture
|
/// the actor is a cluster singleton and the repository is a scoped EF Core
|
||||||
/// (the only way to verify the ingest-timestamp stamp + duplicate-key
|
/// service, so it must be resolved from a fresh DI scope per message and never
|
||||||
/// idempotency end to end), while the production host wiring registers the
|
/// held on a field. Mirroring the Notification Outbox actor's pattern. Tests
|
||||||
/// actor as a cluster singleton and must therefore resolve the repository —
|
/// that need a concrete repository (e.g. the MSSQL-backed fixture verifying the
|
||||||
/// which is a scoped EF Core service — from a fresh DI scope per message.
|
/// ingest-timestamp stamp and duplicate-key idempotency) supply a stub provider
|
||||||
/// Mirroring the Notification Outbox actor's pattern.
|
/// that hands one out, rather than a second constructor that would cache it —
|
||||||
|
/// see register row 61.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
public class AuditLogIngestActor : ReceiveActor
|
public class AuditLogIngestActor : ReceiveActor
|
||||||
@@ -100,41 +101,25 @@ public class AuditLogIngestActor : ReceiveActor
|
|||||||
/// </remarks>
|
/// </remarks>
|
||||||
internal static readonly TimeSpan IngestFallbackBudget = TimeSpan.FromSeconds(5);
|
internal static readonly TimeSpan IngestFallbackBudget = TimeSpan.FromSeconds(5);
|
||||||
|
|
||||||
private readonly IServiceProvider? _serviceProvider;
|
private readonly IServiceProvider _serviceProvider;
|
||||||
private readonly IAuditLogRepository? _injectedRepository;
|
|
||||||
private readonly ILogger<AuditLogIngestActor> _logger;
|
private readonly ILogger<AuditLogIngestActor> _logger;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Test-mode constructor — injects a concrete repository instance whose
|
/// The only constructor — resolves <see cref="IAuditLogRepository"/> from
|
||||||
/// lifetime exceeds the test, so the actor reuses the same instance across
|
|
||||||
/// every message. Used by the MSSQL-backed TestKit fixture.
|
|
||||||
/// </summary>
|
|
||||||
/// <param name="repository">Audit log repository instance shared across all messages.</param>
|
|
||||||
/// <param name="logger">Logger for ingest diagnostics.</param>
|
|
||||||
public AuditLogIngestActor(
|
|
||||||
IAuditLogRepository repository,
|
|
||||||
ILogger<AuditLogIngestActor> logger)
|
|
||||||
{
|
|
||||||
ArgumentNullException.ThrowIfNull(repository);
|
|
||||||
ArgumentNullException.ThrowIfNull(logger);
|
|
||||||
|
|
||||||
_injectedRepository = repository;
|
|
||||||
_logger = logger;
|
|
||||||
|
|
||||||
ReceiveAsync<IngestAuditEventsCommand>(OnIngestAsync);
|
|
||||||
// The single-repository test ctor cannot service the dual-write —
|
|
||||||
// it has no SiteCalls repo and no DbContext. The handler still
|
|
||||||
// registers (so callers don't dead-letter) but replies empty so the
|
|
||||||
// test surface stays explicit about what this ctor supports.
|
|
||||||
ReceiveAsync<IngestCachedTelemetryCommand>(OnCachedTelemetryWithoutDualWriteAsync);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Production constructor — resolves <see cref="IAuditLogRepository"/> from
|
|
||||||
/// a fresh DI scope per message because the repository is a scoped EF Core
|
/// a fresh DI scope per message because the repository is a scoped EF Core
|
||||||
/// service registered by <c>AddConfigurationDatabase</c>. The actor itself
|
/// service registered by <c>AddConfigurationDatabase</c>. The actor itself
|
||||||
/// is a long-lived cluster singleton, so it cannot hold a scope across
|
/// is a long-lived cluster singleton, so it cannot hold a scope across
|
||||||
/// messages.
|
/// messages.
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// Register row 61: a second, repository-taking "test-mode" constructor used
|
||||||
|
/// to cache a scoped repository on a field for the actor's whole lifetime.
|
||||||
|
/// Nothing in production selected it, but it left the captive-dependency bug
|
||||||
|
/// one call site away and had to be allowlisted in the Host tests'
|
||||||
|
/// <c>CaptiveDependencyGuard</c>. It is gone; test fixtures build a stub
|
||||||
|
/// <see cref="IServiceProvider"/> instead, which keeps the actor's own
|
||||||
|
/// contract "resolve per scope, hold nothing".
|
||||||
|
/// </para>
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="serviceProvider">Root service provider used to open a fresh scope per message.</param>
|
/// <param name="serviceProvider">Root service provider used to open a fresh scope per message.</param>
|
||||||
/// <param name="logger">Logger for ingest diagnostics.</param>
|
/// <param name="logger">Logger for ingest diagnostics.</param>
|
||||||
@@ -169,59 +154,51 @@ public class AuditLogIngestActor : ReceiveActor
|
|||||||
var accepted = new List<Guid>(cmd.Events.Count);
|
var accepted = new List<Guid>(cmd.Events.Count);
|
||||||
|
|
||||||
// Resolve the repository for the whole batch — one DbContext per
|
// Resolve the repository for the whole batch — one DbContext per
|
||||||
// message, mirroring NotificationOutboxActor. The injected-repository
|
// message, mirroring NotificationOutboxActor.
|
||||||
// mode (test ctor) skips the scope entirely.
|
|
||||||
// The IAuditRedactor is also resolved from the per-message scope when
|
// The IAuditRedactor is also resolved from the per-message scope when
|
||||||
// one is available so the row is truncated + redacted before
|
// one is available so the row is truncated + redacted before
|
||||||
// InsertIfNotExistsAsync. The single-repository test ctor has no
|
// InsertIfNotExistsAsync. A composition root that registers no redactor
|
||||||
// service provider — it falls through with no redactor, which preserves
|
// (test fixtures) falls back to the SafeDefault inside
|
||||||
// the small-payload assumptions baked into the existing fixtures.
|
// IngestWithRepositoryAsync rather than to pass-through.
|
||||||
// Use CreateAsyncScope + await using so scoped EF Core
|
// Use CreateAsyncScope + await using so scoped EF Core
|
||||||
// services (IAsyncDisposable DbContexts) dispose asynchronously
|
// services (IAsyncDisposable DbContexts) dispose asynchronously
|
||||||
// without blocking on sync Dispose() of pending connection cleanup.
|
// without blocking on sync Dispose() of pending connection cleanup.
|
||||||
if (_injectedRepository is not null)
|
//
|
||||||
|
// Guard scope-creation + repository resolution in a
|
||||||
|
// try/catch, mirroring OnCachedTelemetryAsync. A transient DI /
|
||||||
|
// DbContext-factory fault (pooled-context init, SQL-connection
|
||||||
|
// exhaustion, a resolution race during host churn) would otherwise
|
||||||
|
// propagate out of the ReceiveAsync handler, trip the parent's
|
||||||
|
// supervision, and RESTART this central singleton over a transient
|
||||||
|
// fault — dropping the captured reply so the site's Ask times out.
|
||||||
|
// Best-effort audit must never wedge the singleton: log, optionally
|
||||||
|
// bump the failure counter, and still reply with whatever was
|
||||||
|
// accepted (empty on an up-front scope-resolution throw) so the
|
||||||
|
// site keeps its rows Pending and retries on the next drain.
|
||||||
|
try
|
||||||
{
|
{
|
||||||
await IngestWithRepositoryAsync(_injectedRepository, redactor: null, failureCounter: null, cmd, nowUtc, accepted)
|
await using var scope = _serviceProvider.CreateAsyncScope();
|
||||||
|
var repository = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
|
||||||
|
var redactor = scope.ServiceProvider.GetService<IAuditRedactor>();
|
||||||
|
// Central health counter is best-effort —
|
||||||
|
// unregistered (test composition roots) means the per-row catch
|
||||||
|
// simply logs without surfacing on the health dashboard.
|
||||||
|
var failureCounter = scope.ServiceProvider.GetService<ICentralAuditWriteFailureCounter>();
|
||||||
|
await IngestWithRepositoryAsync(repository, redactor, failureCounter, cmd, nowUtc, accepted)
|
||||||
.ConfigureAwait(false);
|
.ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
else
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
// Guard scope-creation + repository resolution in a
|
// Scope creation or a required-service resolution threw before
|
||||||
// try/catch, mirroring OnCachedTelemetryAsync. A transient DI /
|
// (or while) processing the batch. Surface a sustained fault on
|
||||||
// DbContext-factory fault (pooled-context init, SQL-connection
|
// the dashboard if the counter is registered, but never let the
|
||||||
// exhaustion, a resolution race during host churn) would otherwise
|
// throw escape the handler and restart the singleton.
|
||||||
// propagate out of the ReceiveAsync handler, trip the parent's
|
try { _serviceProvider.GetService<ICentralAuditWriteFailureCounter>()?.Increment(); }
|
||||||
// supervision, and RESTART this central singleton over a transient
|
catch { /* counter must never throw — defence in depth */ }
|
||||||
// fault — dropping the captured reply so the site's Ask times out.
|
_logger.LogError(
|
||||||
// Best-effort audit must never wedge the singleton: log, optionally
|
ex,
|
||||||
// bump the failure counter, and still reply with whatever was
|
"Audit event batch ingest failed before/while resolving the repository scope; replying with {Accepted} accepted row(s). The site keeps unaccepted rows Pending and retries on the next drain.",
|
||||||
// accepted (empty on an up-front scope-resolution throw) so the
|
accepted.Count);
|
||||||
// site keeps its rows Pending and retries on the next drain.
|
|
||||||
try
|
|
||||||
{
|
|
||||||
await using var scope = _serviceProvider!.CreateAsyncScope();
|
|
||||||
var repository = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
|
|
||||||
var redactor = scope.ServiceProvider.GetService<IAuditRedactor>();
|
|
||||||
// Central health counter is best-effort —
|
|
||||||
// unregistered (test composition roots) means the per-row catch
|
|
||||||
// simply logs without surfacing on the health dashboard.
|
|
||||||
var failureCounter = scope.ServiceProvider.GetService<ICentralAuditWriteFailureCounter>();
|
|
||||||
await IngestWithRepositoryAsync(repository, redactor, failureCounter, cmd, nowUtc, accepted)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
catch (Exception ex)
|
|
||||||
{
|
|
||||||
// Scope creation or a required-service resolution threw before
|
|
||||||
// (or while) processing the batch. Surface a sustained fault on
|
|
||||||
// the dashboard if the counter is registered, but never let the
|
|
||||||
// throw escape the handler and restart the singleton.
|
|
||||||
try { _serviceProvider!.GetService<ICentralAuditWriteFailureCounter>()?.Increment(); }
|
|
||||||
catch { /* counter must never throw — defence in depth */ }
|
|
||||||
_logger.LogError(
|
|
||||||
ex,
|
|
||||||
"Audit event batch ingest failed before/while resolving the repository scope; replying with {Accepted} accepted row(s). The site keeps unaccepted rows Pending and retries on the next drain.",
|
|
||||||
accepted.Count);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
replyTo.Tell(new IngestAuditEventsReply(accepted));
|
replyTo.Tell(new IngestAuditEventsReply(accepted));
|
||||||
@@ -357,7 +334,7 @@ public class AuditLogIngestActor : ReceiveActor
|
|||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
await using var scope = _serviceProvider!.CreateAsyncScope();
|
await using var scope = _serviceProvider.CreateAsyncScope();
|
||||||
var auditRepo = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
|
var auditRepo = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
|
||||||
var siteCallRepo = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
var siteCallRepo = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
||||||
var dbContext = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
|
var dbContext = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
|
||||||
@@ -544,18 +521,4 @@ public class AuditLogIngestActor : ReceiveActor
|
|||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Fallback handler installed on the single-repository test ctor — that
|
|
||||||
/// ctor has no DbContext and no <see cref="ISiteCallAuditRepository"/>, so
|
|
||||||
/// it cannot service the dual-write. Logs a warning and replies with an
|
|
||||||
/// empty ack so callers fall through to their retry path.
|
|
||||||
/// </summary>
|
|
||||||
private Task OnCachedTelemetryWithoutDualWriteAsync(IngestCachedTelemetryCommand cmd)
|
|
||||||
{
|
|
||||||
_logger.LogWarning(
|
|
||||||
"AuditLogIngestActor received IngestCachedTelemetryCommand on the single-repository ctor; dual-write requires the IServiceProvider ctor. Replying with empty ack ({Count} entries).",
|
|
||||||
cmd.Entries.Count);
|
|
||||||
Sender.Tell(new IngestCachedTelemetryReply(Array.Empty<Guid>()));
|
|
||||||
return Task.CompletedTask;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -87,8 +87,7 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private const int MaxReconciliationPagesPerTick = 50;
|
private const int MaxReconciliationPagesPerTick = 50;
|
||||||
|
|
||||||
private readonly IServiceProvider? _serviceProvider;
|
private readonly IServiceProvider _serviceProvider;
|
||||||
private readonly ISiteCallAuditRepository? _injectedRepository;
|
|
||||||
private readonly SiteCallAuditOptions _options;
|
private readonly SiteCallAuditOptions _options;
|
||||||
private readonly ILogger<SiteCallAuditActor> _logger;
|
private readonly ILogger<SiteCallAuditActor> _logger;
|
||||||
|
|
||||||
@@ -108,28 +107,17 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
/// (<see cref="IPullSiteCallsClient"/>) and the site list
|
/// (<see cref="IPullSiteCallsClient"/>) and the site list
|
||||||
/// (<see cref="ISiteEnumerator"/>). On the production path these are
|
/// (<see cref="ISiteEnumerator"/>). On the production path these are
|
||||||
/// resolved once from the root <see cref="IServiceProvider"/> (central
|
/// resolved once from the root <see cref="IServiceProvider"/> (central
|
||||||
/// singletons registered by <c>AddAuditLogCentralReconciliationClient</c>);
|
/// singletons registered by <c>AddAuditLogCentralReconciliationClient</c>).
|
||||||
/// in the test path they are injected directly. They are <c>null</c> when
|
/// They are <c>null</c> when the composition root did not register them (a
|
||||||
/// the actor was built via the repo-only test ctor — in that case the
|
/// minimal host, or a test fixture that deliberately omits them) — in that
|
||||||
/// reconciliation tick is NOT started (see <see cref="StartReconciliationTimer"/>).
|
/// case the reconciliation tick is NOT started (see
|
||||||
/// The purge tick, by contrast, does NOT depend on these collaborators:
|
/// <see cref="StartReconciliationTimer"/>). The purge tick, by contrast, does
|
||||||
/// it needs only the repository, so it is armed by
|
/// NOT depend on these collaborators: it needs only the repository, so it is
|
||||||
/// <see cref="_backgroundTimersEnabled"/> alone (see <see cref="StartPurgeTimer"/>).
|
/// always armed (see <see cref="StartPurgeTimer"/>).
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private readonly IPullSiteCallsClient? _pullClient;
|
private readonly IPullSiteCallsClient? _pullClient;
|
||||||
private readonly ISiteEnumerator? _siteEnumerator;
|
private readonly ISiteEnumerator? _siteEnumerator;
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Master switch for the two background schedulers (reconciliation + purge),
|
|
||||||
/// set <c>true</c> by the production and reconciliation ctors and <c>false</c>
|
|
||||||
/// by the repo-only MSSQL test ctor. The purge timer is
|
|
||||||
/// gated on THIS flag rather than on the reconciliation collaborators, so a
|
|
||||||
/// host that omits the reconciliation client still purges (no unbounded
|
|
||||||
/// central <c>SiteCalls</c> growth) while the MSSQL read/upsert tests stay
|
|
||||||
/// free of any scheduled side effects.
|
|
||||||
/// </summary>
|
|
||||||
private readonly bool _backgroundTimersEnabled;
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Per-site reconciliation watermark — the composite
|
/// Per-site reconciliation watermark — the composite
|
||||||
/// <c>(UpdatedAtUtc, TrackedOperationId)</c> keyset of the highest row seen
|
/// <c>(UpdatedAtUtc, TrackedOperationId)</c> keyset of the highest row seen
|
||||||
@@ -195,98 +183,23 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
private IActorRef? _centralCommunication;
|
private IActorRef? _centralCommunication;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Test-mode constructor — injects a concrete repository instance whose
|
/// The only constructor — resolves <see cref="ISiteCallAuditRepository"/>
|
||||||
/// lifetime exceeds the test, so the actor reuses the same instance
|
|
||||||
/// across every message. Used by the MSSQL-backed TestKit fixture.
|
|
||||||
/// An optional <paramref name="options"/> lets a test pin the stuck/KPI
|
|
||||||
/// windows; when omitted the production defaults apply.
|
|
||||||
/// <para>
|
|
||||||
/// This ctor disables BOTH background timers (sets
|
|
||||||
/// <see cref="_backgroundTimersEnabled"/> to <c>false</c>) and injects no
|
|
||||||
/// reconciliation client/enumerator, so neither the reconciliation tick nor
|
|
||||||
/// the purge tick fires — the MSSQL-backed read/upsert tests must see no
|
|
||||||
/// scheduled side effects (no phantom pulls, no background purge).
|
|
||||||
/// </para>
|
|
||||||
/// </summary>
|
|
||||||
/// <param name="repository">Concrete repository instance to use for all messages.</param>
|
|
||||||
/// <param name="logger">Logger for diagnostics and error reporting.</param>
|
|
||||||
/// <param name="options">Optional configuration overrides; production defaults apply when null.</param>
|
|
||||||
/// <param name="auditWriter">Optional central direct-write audit sink for operator relay actions; null disables the relay audit row.</param>
|
|
||||||
public SiteCallAuditActor(
|
|
||||||
ISiteCallAuditRepository repository,
|
|
||||||
ILogger<SiteCallAuditActor> logger,
|
|
||||||
SiteCallAuditOptions? options = null,
|
|
||||||
ICentralAuditWriter? auditWriter = null)
|
|
||||||
{
|
|
||||||
ArgumentNullException.ThrowIfNull(repository);
|
|
||||||
ArgumentNullException.ThrowIfNull(logger);
|
|
||||||
|
|
||||||
_injectedRepository = new SerializedRepository(repository);
|
|
||||||
_logger = logger;
|
|
||||||
_options = options ?? new SiteCallAuditOptions();
|
|
||||||
_auditWriter = auditWriter;
|
|
||||||
|
|
||||||
// Repo-only MSSQL test ctor: keep BOTH background timers off so the
|
|
||||||
// read/upsert tests see no scheduled side effects.
|
|
||||||
_backgroundTimersEnabled = false;
|
|
||||||
|
|
||||||
RegisterHandlers();
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Test-mode constructor for the reconciliation tick (Piece A) — injects a
|
|
||||||
/// concrete repository PLUS the two reconciliation collaborators directly,
|
|
||||||
/// so the per-site self-heal pull is unit-testable in-memory without a DI
|
|
||||||
/// container or a live gRPC channel. Background timers are enabled, so the
|
|
||||||
/// purge tick starts (it needs only the repository) and the reconciliation
|
|
||||||
/// tick starts too because the client + enumerator are present — see
|
|
||||||
/// <see cref="StartReconciliationTimer"/> / <see cref="StartPurgeTimer"/>.
|
|
||||||
/// </summary>
|
|
||||||
/// <param name="repository">Concrete repository instance used for upserts and purges.</param>
|
|
||||||
/// <param name="siteEnumerator">Enumerates the sites to reconcile each tick.</param>
|
|
||||||
/// <param name="pullClient">Pull client used to fetch changed rows from each site.</param>
|
|
||||||
/// <param name="logger">Logger for diagnostics and error reporting.</param>
|
|
||||||
/// <param name="options">Optional configuration overrides; production defaults apply when null.</param>
|
|
||||||
/// <remarks>
|
|
||||||
/// Public (not internal) because Akka's default <c>ActivatorProducer</c>
|
|
||||||
/// instantiates the actor via reflection with public-only binding flags —
|
|
||||||
/// an internal ctor yields a <c>MissingMethodException</c> at actor
|
|
||||||
/// creation. Distinguished from the production <see cref="IServiceProvider"/>
|
|
||||||
/// ctor by its concrete-collaborator parameter list; only the test project
|
|
||||||
/// (or a host that hand-resolves the collaborators) constructs it this way.
|
|
||||||
/// </remarks>
|
|
||||||
public SiteCallAuditActor(
|
|
||||||
ISiteCallAuditRepository repository,
|
|
||||||
ISiteEnumerator siteEnumerator,
|
|
||||||
IPullSiteCallsClient pullClient,
|
|
||||||
ILogger<SiteCallAuditActor> logger,
|
|
||||||
SiteCallAuditOptions? options = null)
|
|
||||||
{
|
|
||||||
ArgumentNullException.ThrowIfNull(repository);
|
|
||||||
ArgumentNullException.ThrowIfNull(siteEnumerator);
|
|
||||||
ArgumentNullException.ThrowIfNull(pullClient);
|
|
||||||
ArgumentNullException.ThrowIfNull(logger);
|
|
||||||
|
|
||||||
_injectedRepository = new SerializedRepository(repository);
|
|
||||||
_siteEnumerator = siteEnumerator;
|
|
||||||
_pullClient = pullClient;
|
|
||||||
_logger = logger;
|
|
||||||
_options = options ?? new SiteCallAuditOptions();
|
|
||||||
|
|
||||||
// Reconciliation test ctor: collaborators present, so both timers arm
|
|
||||||
// (the reconciliation tick uses the collaborators; the purge tick needs
|
|
||||||
// only the repository).
|
|
||||||
_backgroundTimersEnabled = true;
|
|
||||||
|
|
||||||
RegisterHandlers();
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Production constructor — resolves <see cref="ISiteCallAuditRepository"/>
|
|
||||||
/// from a fresh DI scope per message because the repository is a scoped EF
|
/// from a fresh DI scope per message because the repository is a scoped EF
|
||||||
/// Core service registered by <c>AddConfigurationDatabase</c>. The actor
|
/// Core service registered by <c>AddConfigurationDatabase</c>. The actor
|
||||||
/// itself is a long-lived cluster singleton, so it cannot hold a scope
|
/// itself is a long-lived cluster singleton, so it cannot hold a scope
|
||||||
/// across messages.
|
/// across messages.
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// Register row 61: two further "test-mode" constructors used to take a
|
||||||
|
/// concrete repository and cache it (wrapped in a serializing decorator) on
|
||||||
|
/// a field for the actor's whole lifetime. Nothing in production selected
|
||||||
|
/// them, but they left the captive-dependency bug one call site away and had
|
||||||
|
/// to be allowlisted in the Host tests' <c>CaptiveDependencyGuard</c>. They
|
||||||
|
/// are gone; test fixtures build a stub <see cref="IServiceProvider"/>
|
||||||
|
/// instead — and own the serializing decorator themselves, since it exists
|
||||||
|
/// only to compensate for a fixture handing the SAME repository instance to
|
||||||
|
/// every scope. Production never needed it: each scope gets its own.
|
||||||
|
/// </para>
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="serviceProvider">DI service provider used to create a scoped repository per message.</param>
|
/// <param name="serviceProvider">DI service provider used to create a scoped repository per message.</param>
|
||||||
/// <param name="options">Actor configuration (stuck threshold, KPI interval, relay timeout).</param>
|
/// <param name="options">Actor configuration (stuck threshold, KPI interval, relay timeout).</param>
|
||||||
@@ -321,12 +234,6 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
_pullClient = serviceProvider.GetService<IPullSiteCallsClient>();
|
_pullClient = serviceProvider.GetService<IPullSiteCallsClient>();
|
||||||
_siteEnumerator = serviceProvider.GetService<ISiteEnumerator>();
|
_siteEnumerator = serviceProvider.GetService<ISiteEnumerator>();
|
||||||
|
|
||||||
// Production path: background timers run. The purge tick is armed
|
|
||||||
// unconditionally here (it needs only the repository); the reconciliation
|
|
||||||
// tick additionally requires its collaborators and logs a Warning if
|
|
||||||
// they were not registered.
|
|
||||||
_backgroundTimersEnabled = true;
|
|
||||||
|
|
||||||
RegisterHandlers();
|
RegisterHandlers();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -462,11 +369,6 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private void StartReconciliationTimer()
|
private void StartReconciliationTimer()
|
||||||
{
|
{
|
||||||
if (!_backgroundTimersEnabled)
|
|
||||||
{
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (_pullClient is null || _siteEnumerator is null)
|
if (_pullClient is null || _siteEnumerator is null)
|
||||||
{
|
{
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
@@ -488,23 +390,16 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Starts the daily purge tick. The purge needs ONLY the
|
/// Starts the daily purge tick. The purge needs ONLY the
|
||||||
/// repository — never the reconciliation collaborators — so it is gated on
|
/// repository — never the reconciliation collaborators — so it is armed
|
||||||
/// <see cref="_backgroundTimersEnabled"/> alone, NOT on
|
/// unconditionally, NOT gated on
|
||||||
/// <see cref="_pullClient"/> / <see cref="_siteEnumerator"/>. This decouples
|
/// <see cref="_pullClient"/> / <see cref="_siteEnumerator"/>. This decouples
|
||||||
/// the daily terminal-row purge from the reconciliation client: a host that
|
/// the daily terminal-row purge from the reconciliation client: a host that
|
||||||
/// registers Site Call Audit without the reconciliation client still purges,
|
/// registers Site Call Audit without the reconciliation client still purges,
|
||||||
/// so the central <c>SiteCalls</c> table can never grow unbounded just
|
/// so the central <c>SiteCalls</c> table can never grow unbounded just
|
||||||
/// because the self-heal puller is absent. Only the repo-only MSSQL test ctor
|
/// because the self-heal puller is absent.
|
||||||
/// (background timers off) skips it, keeping the read/upsert tests free of
|
|
||||||
/// scheduled side effects.
|
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private void StartPurgeTimer()
|
private void StartPurgeTimer()
|
||||||
{
|
{
|
||||||
if (!_backgroundTimersEnabled)
|
|
||||||
{
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
var interval = _options.ResolvedPurgeInterval;
|
var interval = _options.ResolvedPurgeInterval;
|
||||||
_purgeTimer = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
|
_purgeTimer = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
|
||||||
// Short first tick (min(interval, 5min)): purges are idempotent, and a full-interval
|
// Short first tick (min(interval, 5min)): purges are idempotent, and a full-interval
|
||||||
@@ -532,20 +427,11 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
var replyTo = Sender;
|
var replyTo = Sender;
|
||||||
var id = cmd.SiteCall.TrackedOperationId;
|
var id = cmd.SiteCall.TrackedOperationId;
|
||||||
|
|
||||||
// Scope-per-message mirrors AuditLogIngestActor — production EF
|
// Scope-per-message mirrors AuditLogIngestActor — the EF repository is
|
||||||
// repository is scoped; the injected-repository mode (tests) skips
|
// a scoped service, so it is resolved fresh here and never held.
|
||||||
// the scope entirely.
|
IServiceScope? scope = _serviceProvider.CreateScope();
|
||||||
IServiceScope? scope = null;
|
ISiteCallAuditRepository repository =
|
||||||
ISiteCallAuditRepository repository;
|
scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
||||||
if (_injectedRepository is not null)
|
|
||||||
{
|
|
||||||
repository = _injectedRepository;
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
scope = _serviceProvider!.CreateScope();
|
|
||||||
repository = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
|
||||||
}
|
|
||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
@@ -624,15 +510,8 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
// thread on a synchronous Dispose() of pending connection cleanup — the tick
|
// thread on a synchronous Dispose() of pending connection cleanup — the tick
|
||||||
// holds the scope across many awaited UpsertAsync calls. Mirrors the sibling
|
// holds the scope across many awaited UpsertAsync calls. Mirrors the sibling
|
||||||
// SiteAuditReconciliationActor.OnTickAsync. ResolveRepository() (sync Dispose)
|
// SiteAuditReconciliationActor.OnTickAsync. ResolveRepository() (sync Dispose)
|
||||||
// is retained for the synchronous message-handler paths. In the injected-
|
// is retained for the synchronous message-handler paths.
|
||||||
// repository test path there is no scope to open and the test repo is reused.
|
await using var scope = _serviceProvider.CreateAsyncScope();
|
||||||
if (_injectedRepository is not null)
|
|
||||||
{
|
|
||||||
await ReconcileSitesAsync(sites, client, _injectedRepository).ConfigureAwait(false);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
await using var scope = _serviceProvider!.CreateAsyncScope();
|
|
||||||
var repository = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
var repository = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
||||||
await ReconcileSitesAsync(sites, client, repository).ConfigureAwait(false);
|
await ReconcileSitesAsync(sites, client, repository).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
@@ -860,15 +739,8 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
// Open the scope INLINE with CreateAsyncScope + await using
|
// Open the scope INLINE with CreateAsyncScope + await using
|
||||||
// so the scoped EF Core repository (an IAsyncDisposable DbContext) disposes
|
// so the scoped EF Core repository (an IAsyncDisposable DbContext) disposes
|
||||||
// asynchronously rather than blocking the Akka dispatcher thread on a
|
// asynchronously rather than blocking the Akka dispatcher thread on a
|
||||||
// synchronous Dispose(). Mirrors SiteAuditReconciliationActor; the
|
// synchronous Dispose(). Mirrors SiteAuditReconciliationActor.
|
||||||
// injected-repository test path reuses the test repo with no scope.
|
await using var scope = _serviceProvider.CreateAsyncScope();
|
||||||
if (_injectedRepository is not null)
|
|
||||||
{
|
|
||||||
await PurgeWithRepositoryAsync(_injectedRepository, threshold).ConfigureAwait(false);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
await using var scope = _serviceProvider!.CreateAsyncScope();
|
|
||||||
var repository = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
var repository = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
|
||||||
await PurgeWithRepositoryAsync(repository, threshold).ConfigureAwait(false);
|
await PurgeWithRepositoryAsync(repository, threshold).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
@@ -1454,19 +1326,13 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Resolves an <see cref="ISiteCallAuditRepository"/> for one read message.
|
/// Resolves an <see cref="ISiteCallAuditRepository"/> for one read message.
|
||||||
/// In test mode the injected instance is returned with a null scope; in
|
/// A fresh DI scope is created and returned so the caller can dispose it
|
||||||
/// production a fresh DI scope is created and returned so the caller can
|
/// once the read completes — the same scope-per-message pattern as
|
||||||
/// dispose it once the read completes — the same scope-per-message pattern
|
/// <see cref="OnUpsertAsync"/>.
|
||||||
/// as <see cref="OnUpsertAsync"/>.
|
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private (IServiceScope? Scope, ISiteCallAuditRepository Repository) ResolveRepository()
|
private (IServiceScope? Scope, ISiteCallAuditRepository Repository) ResolveRepository()
|
||||||
{
|
{
|
||||||
if (_injectedRepository is not null)
|
var scope = _serviceProvider.CreateScope();
|
||||||
{
|
|
||||||
return (null, _injectedRepository);
|
|
||||||
}
|
|
||||||
|
|
||||||
var scope = _serviceProvider!.CreateScope();
|
|
||||||
return (scope, scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>());
|
return (scope, scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1565,91 +1431,6 @@ public class SiteCallAuditActor : ReceiveActor
|
|||||||
private PurgeComplete() { }
|
private PurgeComplete() { }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Serializing wrapper applied to a repository handed in through the
|
|
||||||
/// test constructors. One call at a time, in arrival order.
|
|
||||||
/// </summary>
|
|
||||||
/// <remarks>
|
|
||||||
/// <para>
|
|
||||||
/// In production every consumer of this actor's repository gets its OWN DI
|
|
||||||
/// scope — one per message, one per background pass — so the off-mailbox
|
|
||||||
/// reconciliation and purge passes can safely run alongside mailbox handlers:
|
|
||||||
/// separate scopes mean separate <c>DbContext</c>s. The injected-repository
|
|
||||||
/// constructors break that assumption: ONE instance (typically wrapping one
|
|
||||||
/// <c>DbContext</c>) is shared by the mailbox handlers AND the background
|
|
||||||
/// passes, and <c>DbContext</c> forbids concurrent operations — "A second
|
|
||||||
/// operation was started on this context instance" is a hard fault, and the
|
|
||||||
/// interleaving is nondeterministic, so it surfaces as a flaky test rather
|
|
||||||
/// than a reliable one.
|
|
||||||
/// </para>
|
|
||||||
/// <para>
|
|
||||||
/// Serializing at the CALL, not around the whole pass, is deliberate: it
|
|
||||||
/// removes the concurrency hazard while preserving the property the
|
|
||||||
/// off-mailbox passes exist for — a long drain (blocked in a network pull,
|
|
||||||
/// holding no repository call) still lets ingest, query and KPI messages be
|
|
||||||
/// answered. Suspending the mailbox for the duration of a pass would have
|
|
||||||
/// been simpler and would have invalidated exactly those regression tests.
|
|
||||||
/// Production is untouched: it never constructs this type.
|
|
||||||
/// </para>
|
|
||||||
/// </remarks>
|
|
||||||
private sealed class SerializedRepository : ISiteCallAuditRepository
|
|
||||||
{
|
|
||||||
private readonly ISiteCallAuditRepository _inner;
|
|
||||||
private readonly SemaphoreSlim _gate = new(1, 1);
|
|
||||||
|
|
||||||
public SerializedRepository(ISiteCallAuditRepository inner) => _inner = inner;
|
|
||||||
|
|
||||||
public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) =>
|
|
||||||
RunAsync(() => _inner.UpsertAsync(siteCall, ct));
|
|
||||||
|
|
||||||
public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
|
|
||||||
RunAsync(() => _inner.GetAsync(id, ct));
|
|
||||||
|
|
||||||
public Task<IReadOnlyList<SiteCall>> QueryAsync(
|
|
||||||
SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
|
|
||||||
RunAsync(() => _inner.QueryAsync(filter, paging, ct));
|
|
||||||
|
|
||||||
public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
|
|
||||||
RunAsync(() => _inner.PurgeTerminalAsync(olderThanUtc, ct));
|
|
||||||
|
|
||||||
public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
|
|
||||||
DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) =>
|
|
||||||
RunAsync(() => _inner.ComputeKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct));
|
|
||||||
|
|
||||||
public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
|
|
||||||
DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) =>
|
|
||||||
RunAsync(() => _inner.ComputePerSiteKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct));
|
|
||||||
|
|
||||||
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
|
|
||||||
DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) =>
|
|
||||||
RunAsync(() => _inner.ComputePerNodeKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct));
|
|
||||||
|
|
||||||
private async Task RunAsync(Func<Task> call)
|
|
||||||
{
|
|
||||||
await _gate.WaitAsync().ConfigureAwait(false);
|
|
||||||
try
|
|
||||||
{
|
|
||||||
await call().ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
finally
|
|
||||||
{
|
|
||||||
_gate.Release();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task<T> RunAsync<T>(Func<Task<T>> call)
|
|
||||||
{
|
|
||||||
await _gate.WaitAsync().ConfigureAwait(false);
|
|
||||||
try
|
|
||||||
{
|
|
||||||
return await call().ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
finally
|
|
||||||
{
|
|
||||||
_gate.Release();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
|||||||
@@ -51,10 +51,15 @@ public class AuditLogIngestActorTests : TestKit, IClassFixture<MsSqlMigrationFix
|
|||||||
status: AuditStatus.Delivered,
|
status: AuditStatus.Delivered,
|
||||||
sourceSiteId: siteId);
|
sourceSiteId: siteId);
|
||||||
|
|
||||||
private IActorRef CreateActor(IAuditLogRepository repository) =>
|
private IActorRef CreateActor(IAuditLogRepository repository)
|
||||||
Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
{
|
||||||
repository,
|
// Register row 61: the actor's repository-taking test ctor is gone; the fixture
|
||||||
|
// supplies a stub provider that hands this repository out of every scope.
|
||||||
|
var provider = TestSupport.StubAuditRepositoryProvider.Build(repository);
|
||||||
|
return Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
||||||
|
provider,
|
||||||
NullLogger<AuditLogIngestActor>.Instance)));
|
NullLogger<AuditLogIngestActor>.Instance)));
|
||||||
|
}
|
||||||
|
|
||||||
[SkippableFact]
|
[SkippableFact]
|
||||||
public async Task Receive_BatchOf5_Calls_Repo_5Times_Acks_All_5()
|
public async Task Receive_BatchOf5_Calls_Repo_5Times_Acks_All_5()
|
||||||
|
|||||||
+8
-3
@@ -130,10 +130,15 @@ public class DatabaseSyncEmissionEndToEndTests : TestKit, IClassFixture<MsSqlMig
|
|||||||
IdleIntervalSeconds = 1,
|
IdleIntervalSeconds = 1,
|
||||||
});
|
});
|
||||||
|
|
||||||
private IActorRef CreateIngestActor(IAuditLogRepository repo) =>
|
private IActorRef CreateIngestActor(IAuditLogRepository repo)
|
||||||
Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
{
|
||||||
repo,
|
// Register row 61: the actor's repository-taking test ctor is gone; the fixture
|
||||||
|
// supplies a stub provider that hands this repository out of every scope.
|
||||||
|
var provider = TestSupport.StubAuditRepositoryProvider.Build(repo);
|
||||||
|
return Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
||||||
|
provider,
|
||||||
NullLogger<AuditLogIngestActor>.Instance)));
|
NullLogger<AuditLogIngestActor>.Instance)));
|
||||||
|
}
|
||||||
|
|
||||||
private IActorRef CreateTelemetryActor(
|
private IActorRef CreateTelemetryActor(
|
||||||
ISiteAuditQueue queue,
|
ISiteAuditQueue queue,
|
||||||
|
|||||||
+8
-3
@@ -127,10 +127,15 @@ public class ExecutionIdCorrelationTests : TestKit, IClassFixture<MsSqlMigration
|
|||||||
IdleIntervalSeconds = 1,
|
IdleIntervalSeconds = 1,
|
||||||
});
|
});
|
||||||
|
|
||||||
private IActorRef CreateIngestActor(IAuditLogRepository repo) =>
|
private IActorRef CreateIngestActor(IAuditLogRepository repo)
|
||||||
Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
{
|
||||||
repo,
|
// Register row 61: the actor's repository-taking test ctor is gone; the fixture
|
||||||
|
// supplies a stub provider that hands this repository out of every scope.
|
||||||
|
var provider = TestSupport.StubAuditRepositoryProvider.Build(repo);
|
||||||
|
return Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
||||||
|
provider,
|
||||||
NullLogger<AuditLogIngestActor>.Instance)));
|
NullLogger<AuditLogIngestActor>.Instance)));
|
||||||
|
}
|
||||||
|
|
||||||
private IActorRef CreateTelemetryActor(
|
private IActorRef CreateTelemetryActor(
|
||||||
ISiteAuditQueue queue,
|
ISiteAuditQueue queue,
|
||||||
|
|||||||
+8
-3
@@ -110,10 +110,15 @@ public class SyncCallEmissionEndToEndTests : TestKit, IClassFixture<MsSqlMigrati
|
|||||||
IdleIntervalSeconds = 1,
|
IdleIntervalSeconds = 1,
|
||||||
});
|
});
|
||||||
|
|
||||||
private IActorRef CreateIngestActor(IAuditLogRepository repo) =>
|
private IActorRef CreateIngestActor(IAuditLogRepository repo)
|
||||||
Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
{
|
||||||
repo,
|
// Register row 61: the actor's repository-taking test ctor is gone; the fixture
|
||||||
|
// supplies a stub provider that hands this repository out of every scope.
|
||||||
|
var provider = TestSupport.StubAuditRepositoryProvider.Build(repo);
|
||||||
|
return Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
||||||
|
provider,
|
||||||
NullLogger<AuditLogIngestActor>.Instance)));
|
NullLogger<AuditLogIngestActor>.Instance)));
|
||||||
|
}
|
||||||
|
|
||||||
private IActorRef CreateTelemetryActor(
|
private IActorRef CreateTelemetryActor(
|
||||||
ISiteAuditQueue queue,
|
ISiteAuditQueue queue,
|
||||||
|
|||||||
@@ -0,0 +1,41 @@
|
|||||||
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Deferred-work register row 61 — the replacement for <c>AuditLogIngestActor</c>'s deleted
|
||||||
|
/// repository-taking "test-mode" constructor.
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// That constructor cached a scoped repository on an actor field for the actor's whole
|
||||||
|
/// lifetime. Nothing in production selected it, but the shape was one call site away from
|
||||||
|
/// re-introducing a captive-dependency bug, so it had to be allowlisted in the Host tests'
|
||||||
|
/// <c>CaptiveDependencyGuard</c>. The actor now has a single <see cref="IServiceProvider"/>
|
||||||
|
/// constructor; a fixture that wants a concrete repository builds one of these instead.
|
||||||
|
/// </para>
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// Behaviour is unchanged for the ingest path: the old ctor passed <c>redactor: null</c> and
|
||||||
|
/// <c>failureCounter: null</c>, and a provider with neither registered resolves both to null
|
||||||
|
/// too — so the actor still falls back to <c>SafeDefaultAuditRedactor</c> and still logs
|
||||||
|
/// without a health counter. The one deliberate difference is that
|
||||||
|
/// <c>IngestCachedTelemetryCommand</c> now takes the real dual-write handler rather than the
|
||||||
|
/// deleted "no dual write" stub; no fixture built on this helper sends that message.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
internal static class StubAuditRepositoryProvider
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Builds a provider whose every scope yields the same audit-log repository instance.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="repository">Repository instance shared by every scope.</param>
|
||||||
|
public static ServiceProvider Build(IAuditLogRepository repository)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(repository);
|
||||||
|
|
||||||
|
return new ServiceCollection()
|
||||||
|
.AddScoped(_ => repository)
|
||||||
|
.BuildServiceProvider();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -41,6 +41,17 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
|
|||||||
/// <summary>Graceful-completion callbacks captured from every subscribe.</summary>
|
/// <summary>Graceful-completion callbacks captured from every subscribe.</summary>
|
||||||
public List<Action> Completions { get { lock (_lock) { return _completions.ToList(); } } }
|
public List<Action> Completions { get { lock (_lock) { return _completions.ToList(); } } }
|
||||||
|
|
||||||
|
private int _streamEnds;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Register row 40: how many site-wide subscriptions have been torn down (their
|
||||||
|
/// cancellation token fired). This is the aggregator-stop observable — the linger
|
||||||
|
/// stop cancels the aggregator's token, which ends the hanging subscribe below.
|
||||||
|
/// Tests that must show an aggregator was NOT stopped assert on this rather than
|
||||||
|
/// sampling <c>IsLive</c> once after a fixed sleep.
|
||||||
|
/// </summary>
|
||||||
|
public int StreamEndCount => Volatile.Read(ref _streamEnds);
|
||||||
|
|
||||||
public override Task SubscribeSiteAsync(
|
public override Task SubscribeSiteAsync(
|
||||||
string correlationId, Action<Commons.Messages.Streaming.AlarmStateChanged> onAlarmEvent,
|
string correlationId, Action<Commons.Messages.Streaming.AlarmStateChanged> onAlarmEvent,
|
||||||
Action<Exception> onError, Action onCompleted, CancellationToken ct,
|
Action<Exception> onError, Action onCompleted, CancellationToken ct,
|
||||||
@@ -55,7 +66,11 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
|
|||||||
// subscribe), which is what makes the aggregator report the stream live.
|
// subscribe), which is what makes the aggregator report the stream live.
|
||||||
if (AutoConnect) onConnected?.Invoke();
|
if (AutoConnect) onConnected?.Invoke();
|
||||||
var tcs = new TaskCompletionSource();
|
var tcs = new TaskCompletionSource();
|
||||||
ct.Register(() => tcs.TrySetResult());
|
ct.Register(() =>
|
||||||
|
{
|
||||||
|
Interlocked.Increment(ref _streamEnds);
|
||||||
|
tcs.TrySetResult();
|
||||||
|
});
|
||||||
return tcs.Task;
|
return tcs.Task;
|
||||||
}
|
}
|
||||||
public override void Unsubscribe(string correlationId) { }
|
public override void Unsubscribe(string correlationId) { }
|
||||||
@@ -184,27 +199,47 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
|
|||||||
[Fact]
|
[Fact]
|
||||||
public void Last_Viewer_Leaving_Stops_Aggregator_After_Linger()
|
public void Last_Viewer_Leaving_Stops_Aggregator_After_Linger()
|
||||||
{
|
{
|
||||||
var service = CreateService(TimeSpan.FromMilliseconds(200), out _);
|
var linger = TimeSpan.FromMilliseconds(200);
|
||||||
|
var service = CreateService(linger, out var factory);
|
||||||
|
|
||||||
var sub1 = service.Subscribe(SiteId, () => { });
|
var sub1 = service.Subscribe(SiteId, () => { });
|
||||||
var sub2 = service.Subscribe(SiteId, () => { });
|
var sub2 = service.Subscribe(SiteId, () => { });
|
||||||
AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
|
AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
|
||||||
|
|
||||||
sub1.Dispose();
|
sub1.Dispose();
|
||||||
// One viewer remains — still live, not stopped.
|
|
||||||
Thread.Sleep(300);
|
// Register row 40: one viewer remains, so the aggregator must NOT be stopped. The old
|
||||||
Assert.True(service.IsLive(SiteId));
|
// test slept 300 ms and sampled IsLive once — a teardown that happened at any other
|
||||||
|
// instant (or one that stopped and restarted the stream) was invisible to it. Poll
|
||||||
|
// CONTINUOUSLY across three linger windows instead, bailing out the moment either
|
||||||
|
// signal says the aggregator went away.
|
||||||
|
var held = System.Diagnostics.Stopwatch.StartNew();
|
||||||
|
AwaitCondition(
|
||||||
|
() => held.Elapsed > linger * 3
|
||||||
|
|| !service.IsLive(SiteId)
|
||||||
|
|| factory.Client.StreamEndCount > 0,
|
||||||
|
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||||
|
Assert.True(service.IsLive(SiteId),
|
||||||
|
"the aggregator was reported not-live while a viewer was still subscribed");
|
||||||
|
Assert.Equal(0, factory.Client.StreamEndCount);
|
||||||
|
Assert.Equal(1, factory.GetOrCreateCount);
|
||||||
|
|
||||||
sub2.Dispose();
|
sub2.Dispose();
|
||||||
// Last viewer left → after the linger the aggregator is torn down and the entry cleared.
|
// Last viewer left → after the linger the aggregator is torn down and the entry cleared.
|
||||||
AwaitCondition(() => !service.IsLive(SiteId), TimeSpan.FromSeconds(3));
|
AwaitCondition(() => !service.IsLive(SiteId), TimeSpan.FromSeconds(3));
|
||||||
|
|
||||||
|
// Positive control on the SAME observable the negative above rests on: the site-wide
|
||||||
|
// stream really is torn down now, so the zero above was a live signal rather than a
|
||||||
|
// counter that never moves.
|
||||||
|
AwaitCondition(() => factory.Client.StreamEndCount > 0, TimeSpan.FromSeconds(3));
|
||||||
Assert.Empty(service.GetCurrentAlarms(SiteId));
|
Assert.Empty(service.GetCurrentAlarms(SiteId));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void Resubscribe_Within_Linger_Cancels_The_Pending_Stop()
|
public void Resubscribe_Within_Linger_Cancels_The_Pending_Stop()
|
||||||
{
|
{
|
||||||
var service = CreateService(TimeSpan.FromMilliseconds(400), out var factory);
|
var linger = TimeSpan.FromMilliseconds(400);
|
||||||
|
var service = CreateService(linger, out var factory);
|
||||||
|
|
||||||
var sub1 = service.Subscribe(SiteId, () => { });
|
var sub1 = service.Subscribe(SiteId, () => { });
|
||||||
AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
|
AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
|
||||||
@@ -212,13 +247,28 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
|
|||||||
sub1.Dispose(); // schedules linger stop
|
sub1.Dispose(); // schedules linger stop
|
||||||
var sub2 = service.Subscribe(SiteId, () => { }); // returns within the linger window → cancels it
|
var sub2 = service.Subscribe(SiteId, () => { }); // returns within the linger window → cancels it
|
||||||
|
|
||||||
// Wait comfortably past the original linger; the aggregator must still be alive.
|
// Register row 40: the pending stop must be CANCELLED, not merely outrun. The old test
|
||||||
Thread.Sleep(700);
|
// slept 700 ms and sampled once — it could not see a stream that was torn down and
|
||||||
Assert.True(service.IsLive(SiteId));
|
// silently replaced inside that window, because GetOrCreateCount is checked only at the
|
||||||
|
// end and the mock hands back the same client object either way. Poll continuously past
|
||||||
|
// the original linger on the teardown observable itself.
|
||||||
|
var kept = System.Diagnostics.Stopwatch.StartNew();
|
||||||
|
AwaitCondition(
|
||||||
|
() => kept.Elapsed > linger * 3
|
||||||
|
|| !service.IsLive(SiteId)
|
||||||
|
|| factory.Client.StreamEndCount > 0,
|
||||||
|
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
|
||||||
|
Assert.True(service.IsLive(SiteId),
|
||||||
|
"the re-subscribe did not cancel the pending linger stop");
|
||||||
|
Assert.Equal(0, factory.Client.StreamEndCount);
|
||||||
// No second aggregator/stream was opened — the same one was kept warm.
|
// No second aggregator/stream was opened — the same one was kept warm.
|
||||||
Assert.Equal(1, factory.GetOrCreateCount);
|
Assert.Equal(1, factory.GetOrCreateCount);
|
||||||
|
|
||||||
sub2.Dispose();
|
sub2.Dispose();
|
||||||
|
|
||||||
|
// Positive control: with the last viewer gone the linger stop is allowed to run, so the
|
||||||
|
// teardown counter this test's zero rests on does in fact move.
|
||||||
|
AwaitCondition(() => factory.Client.StreamEndCount > 0, TimeSpan.FromSeconds(5));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
@@ -76,6 +76,23 @@ public class DataConnectionActorTests : TestKit
|
|||||||
adapter.Disconnected += Raise.Event<Action>();
|
adapter.Disconnected += Raise.Event<Action>();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Register row 40: true once the actor has actually ENTERED the Connected state for
|
||||||
|
/// <paramref name="connectionName"/>. <c>BecomeConnected</c> publishes
|
||||||
|
/// <see cref="ConnectionHealth.Connected"/> to the health collector immediately before
|
||||||
|
/// its <c>Become</c>, so this is the state transition's own observable edge.
|
||||||
|
///
|
||||||
|
/// <para>An adapter-side connect counter is NOT that edge: it only proves
|
||||||
|
/// <c>ConnectAsync</c> was called, while the actor is still in Connecting until it
|
||||||
|
/// processes the piped <c>ConnectResult</c>. Tests used to bridge that gap with a
|
||||||
|
/// fixed <c>Task.Delay</c>.</para>
|
||||||
|
/// </summary>
|
||||||
|
private bool HasEnteredConnectedState(string connectionName) =>
|
||||||
|
_mockHealthCollector.ReceivedCalls().Any(c =>
|
||||||
|
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
|
||||||
|
c.GetArguments() is [string name, ConnectionHealth.Connected, ..] &&
|
||||||
|
name == connectionName);
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void WP6_StartsInConnectingState_AttemptsConnect()
|
public void WP6_StartsInConnectingState_AttemptsConnect()
|
||||||
{
|
{
|
||||||
@@ -430,9 +447,14 @@ public class DataConnectionActorTests : TestKit
|
|||||||
|
|
||||||
var actor = CreateFailoverActor(primaryAdapter, "resub-test", primaryConfig, backupConfig, failoverRetryCount: 1);
|
var actor = CreateFailoverActor(primaryAdapter, "resub-test", primaryConfig, backupConfig, failoverRetryCount: 1);
|
||||||
|
|
||||||
// Wait for initial connect
|
// Wait for the actor to reach Connected — not merely for ConnectAsync to have been
|
||||||
|
// called, and not for a fixed 200 ms to elapse (register row 40). The subscribe below
|
||||||
|
// is only meaningful against the Connected state: issued during Connecting it is
|
||||||
|
// STASHED, so a too-short settle would silently test the stash path instead, and a
|
||||||
|
// too-long one just slows the suite.
|
||||||
AwaitCondition(() => primaryConnectCount >= 1, TimeSpan.FromSeconds(2));
|
AwaitCondition(() => primaryConnectCount >= 1, TimeSpan.FromSeconds(2));
|
||||||
await Task.Delay(200);
|
AwaitCondition(() => HasEnteredConnectedState("resub-test"), TimeSpan.FromSeconds(5),
|
||||||
|
TimeSpan.FromMilliseconds(10));
|
||||||
|
|
||||||
// Subscribe to tags while connected on primary
|
// Subscribe to tags while connected on primary
|
||||||
actor.Tell(new SubscribeTagsRequest("corr1", "inst1", "resub-test", ["sensor/temp"], DateTimeOffset.UtcNow));
|
actor.Tell(new SubscribeTagsRequest("corr1", "inst1", "resub-test", ["sensor/temp"], DateTimeOffset.UtcNow));
|
||||||
|
|||||||
@@ -72,28 +72,25 @@ internal static class CaptiveDependencyGuard
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// Both entries below are the same shape: a nullable repository field assigned ONLY by an
|
/// EMPTY, and pinned empty by <c>CaptiveDependencyGuardTests.Allowlist_IsEmpty</c>.
|
||||||
/// explicitly test-mode constructor, so the production wiring
|
|
||||||
/// (<c>AkkaHostedService</c> constructs both actors through their <see cref="IServiceProvider"/>
|
|
||||||
/// overload) leaves the field null and every handler takes the create-a-scope branch. The
|
|
||||||
/// test fixtures that do use those constructors are MSSQL-backed TestKit fixtures that
|
|
||||||
/// drive the actor one message at a time, and <c>SiteCallAuditActor</c> additionally wraps
|
|
||||||
/// the injected instance in its own <c>SerializedRepository</c> — i.e. the test path
|
|
||||||
/// already carries a serialization guard for precisely the concurrency hazard row 45 hit.
|
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// They are allowlisted rather than removed because deleting a test-mode constructor is a
|
/// It previously held two entries — <c>AuditLogIngestActor._injectedRepository</c> and
|
||||||
/// larger change than this row justifies, and rather than left unflagged because the shape
|
/// <c>SiteCallAuditActor._injectedRepository</c> — nullable repository fields assigned only
|
||||||
/// is genuinely one constructor call away from being row 45 again: if a production call
|
/// by explicitly test-mode constructors. They were safe solely because no production call
|
||||||
/// site ever switched to the repository overload, the actor would hold a scoped EF-backed
|
/// site picked those overloads, which left the row-45 captive-dependency bug one
|
||||||
/// repository for its whole lifetime. That residual risk is recorded, not dismissed.
|
/// constructor call away. Register row 61 deleted the constructors instead: both actors now
|
||||||
|
/// have a single <see cref="IServiceProvider"/> constructor, the fields are gone, and the
|
||||||
|
/// test fixtures build stub providers that hand a repository out of a scope. With the shape
|
||||||
|
/// removed there is nothing left to exempt.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// Keep it empty. A new entry means someone reintroduced a scoped service cached on an
|
||||||
|
/// actor field; prefer fixing the actor.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
internal static readonly IReadOnlySet<string> Allowlist = new HashSet<string>(StringComparer.Ordinal)
|
internal static readonly IReadOnlySet<string> Allowlist =
|
||||||
{
|
new HashSet<string>(StringComparer.Ordinal);
|
||||||
"ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor._injectedRepository",
|
|
||||||
"ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditActor._injectedRepository",
|
|
||||||
};
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Scans the given actor types for instance fields holding captive scope state.
|
/// Scans the given actor types for instance fields holding captive scope state.
|
||||||
|
|||||||
@@ -166,38 +166,65 @@ public class CaptiveDependencyGuardSelfTests
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The allowlist is an escape hatch, so it is itself pinned: every entry must name a
|
/// Register row 61: the allowlist is now EMPTY and must stay that way.
|
||||||
/// field that still exists and would still be flagged. An entry that has stopped matching
|
///
|
||||||
/// is either a stale exemption or — worse — a typo silently covering nothing.
|
/// <para>
|
||||||
|
/// This replaces the former staleness self-test, which checked that each entry still named
|
||||||
|
/// a field the scanner would flag. Its two entries — <c>AuditLogIngestActor</c> and
|
||||||
|
/// <c>SiteCallAuditActor</c>'s <c>_injectedRepository</c> — covered repository fields
|
||||||
|
/// assigned only by test-mode constructors; those constructors have been deleted and the
|
||||||
|
/// fixtures fold onto the <c>IServiceProvider</c> overload, so there is no longer a field
|
||||||
|
/// to exempt. An empty allowlist is a strictly stronger statement than a correct one:
|
||||||
|
/// nothing is exempt, so nothing can rot.
|
||||||
|
/// </para>
|
||||||
/// </summary>
|
/// </summary>
|
||||||
[Fact]
|
[Fact]
|
||||||
public void Allowlist_EntriesAllStillMatchARealFlaggedField()
|
public void Allowlist_IsEmpty_NoActorExemptFromTheCaptiveDependencyScan()
|
||||||
{
|
{
|
||||||
var actorTypes = CaptiveDependencyGuard.ProductionActorTypes();
|
Assert.True(
|
||||||
|
CaptiveDependencyGuard.Allowlist.Count == 0,
|
||||||
// Scan with the allowlist disabled against a scoped set wide enough to include the
|
"The captive-dependency allowlist must stay empty (register row 61). Entries present: " +
|
||||||
// repositories the allowlisted fields hold.
|
string.Join(", ", CaptiveDependencyGuard.Allowlist) +
|
||||||
var everyScopedType = CaptiveDependencyGuard.ScopedTypes(BuildCentralishScopedProbe());
|
". An entry means an actor caches a scoped service on a field — fix the actor " +
|
||||||
var unfiltered = CaptiveDependencyGuard.Scan(actorTypes, everyScopedType, applyAllowlist: false);
|
"(fold test fixtures onto the IServiceProvider overload) rather than exempting it.");
|
||||||
|
|
||||||
var flagged = unfiltered
|
|
||||||
.Select(v => $"{v.ActorType.FullName}.{v.FieldName}")
|
|
||||||
.ToHashSet(StringComparer.Ordinal);
|
|
||||||
|
|
||||||
foreach (var entry in CaptiveDependencyGuard.Allowlist)
|
|
||||||
{
|
|
||||||
Assert.True(
|
|
||||||
flagged.Contains(entry),
|
|
||||||
$"Allowlist entry '{entry}' no longer matches a flagged field. Remove it — a stale " +
|
|
||||||
"exemption is indistinguishable from an exemption that never worked.");
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// A minimal descriptor set naming the two repository types the allowlisted fields hold,
|
/// Register row 61 (the other half of the pin): with the allowlist empty, the two actors
|
||||||
/// so <see cref="Allowlist_EntriesAllStillMatchARealFlaggedField"/> does not need to boot a
|
/// that used to need exemptions must now scan CLEAN against the scoped repository types
|
||||||
/// host. Both are registered <c>Scoped</c> by <c>AddConfigurationDatabase</c>; the
|
/// they resolve. This is what makes the empty allowlist meaningful — it proves the entries
|
||||||
/// role-scan tests below verify that against the real container.
|
/// were removed because the fields are gone, not because the scan stopped looking.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void FormerlyAllowlistedActors_NowScanClean()
|
||||||
|
{
|
||||||
|
var actorTypes = CaptiveDependencyGuard.ProductionActorTypes();
|
||||||
|
var everyScopedType = CaptiveDependencyGuard.ScopedTypes(BuildCentralishScopedProbe());
|
||||||
|
|
||||||
|
// Allowlist disabled, so a surviving _injectedRepository-shaped field would still show.
|
||||||
|
var unfiltered = CaptiveDependencyGuard.Scan(actorTypes, everyScopedType, applyAllowlist: false);
|
||||||
|
|
||||||
|
var offenders = unfiltered
|
||||||
|
.Where(v => v.ActorType.Name is "AuditLogIngestActor" or "SiteCallAuditActor")
|
||||||
|
.Select(v => v.ToString())
|
||||||
|
.ToList();
|
||||||
|
|
||||||
|
Assert.True(offenders.Count == 0,
|
||||||
|
"AuditLogIngestActor / SiteCallAuditActor must hold no scoped repository field " +
|
||||||
|
"(register row 61). Found: " + string.Join(" | ", offenders));
|
||||||
|
|
||||||
|
// Sanity: the scan really did examine those two actors, so the clean result above is
|
||||||
|
// evidence rather than an empty walk.
|
||||||
|
Assert.Contains(actorTypes, t => t.Name == "AuditLogIngestActor");
|
||||||
|
Assert.Contains(actorTypes, t => t.Name == "SiteCallAuditActor");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A minimal descriptor set naming the two repository types the deleted test-mode
|
||||||
|
/// constructors used to cache, so <see cref="FormerlyAllowlistedActors_NowScanClean"/> does
|
||||||
|
/// not need to boot a host. Both are registered <c>Scoped</c> by
|
||||||
|
/// <c>AddConfigurationDatabase</c>; the role-scan tests below verify that against the real
|
||||||
|
/// container.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>The probe descriptors.</returns>
|
/// <returns>The probe descriptors.</returns>
|
||||||
private static IServiceCollection BuildCentralishScopedProbe()
|
private static IServiceCollection BuildCentralishScopedProbe()
|
||||||
|
|||||||
@@ -149,11 +149,16 @@ public class SiteAuditPushFlowTests : TestKit
|
|||||||
public async Task SiteAuditEvent_DrainsToCentral_AndFlipsSiteRowToForwarded()
|
public async Task SiteAuditEvent_DrainsToCentral_AndFlipsSiteRowToForwarded()
|
||||||
{
|
{
|
||||||
// ── Central side ──────────────────────────────────────────────────
|
// ── Central side ──────────────────────────────────────────────────
|
||||||
// Real AuditLogIngestActor over an in-memory repository (test-mode ctor).
|
// Real AuditLogIngestActor over an in-memory repository. Register row 61: the actor's
|
||||||
|
// repository-taking test ctor is gone, so the repository is handed out of a stub
|
||||||
|
// provider's scope instead of being cached on an actor field.
|
||||||
var centralRepo = new InMemoryAuditLogRepository();
|
var centralRepo = new InMemoryAuditLogRepository();
|
||||||
|
var ingestProvider = new ServiceCollection()
|
||||||
|
.AddScoped<IAuditLogRepository>(_ => centralRepo)
|
||||||
|
.BuildServiceProvider();
|
||||||
var ingestActor = Sys.ActorOf(Props.Create(() =>
|
var ingestActor = Sys.ActorOf(Props.Create(() =>
|
||||||
new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor(
|
new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor(
|
||||||
centralRepo,
|
(IServiceProvider)ingestProvider,
|
||||||
NullLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor>.Instance)));
|
NullLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor>.Instance)));
|
||||||
|
|
||||||
// Real CentralCommunicationActor. It is NOT on the audit path any more (the
|
// Real CentralCommunicationActor. It is NOT on the audit path any more (the
|
||||||
|
|||||||
@@ -73,11 +73,19 @@ public class SiteCallAuditActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
|
|||||||
}
|
}
|
||||||
|
|
||||||
private IActorRef CreateActor(
|
private IActorRef CreateActor(
|
||||||
ISiteCallAuditRepository repository, SiteCallAuditOptions? options = null) =>
|
ISiteCallAuditRepository repository, SiteCallAuditOptions? options = null)
|
||||||
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
{
|
||||||
repository,
|
// Register row 61: the actor's repository-taking test ctor is gone. This MSSQL-backed
|
||||||
NullLogger<SiteCallAuditActor>.Instance,
|
// fixture supplies a stub provider that yields the same repository from every scope,
|
||||||
options)));
|
// serialized (one DbContext cannot service concurrent operations) — see
|
||||||
|
// StubRepositoryProvider. No reconciliation collaborators are registered, so the
|
||||||
|
// reconciliation tick stays disarmed exactly as it did under the old ctor.
|
||||||
|
var provider = TestSupport.StubRepositoryProvider.Build(repository);
|
||||||
|
return Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||||
|
provider,
|
||||||
|
options ?? new SiteCallAuditOptions(),
|
||||||
|
NullLogger<SiteCallAuditActor>.Instance)));
|
||||||
|
}
|
||||||
|
|
||||||
[SkippableFact]
|
[SkippableFact]
|
||||||
public async Task Receive_UpsertSiteCallCommand_Persists_Replies_Accepted()
|
public async Task Receive_UpsertSiteCallCommand_Persists_Replies_Accepted()
|
||||||
|
|||||||
@@ -102,13 +102,17 @@ public class SiteCallAuditPurgeTests : TestKit
|
|||||||
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(DateTime a, DateTime b, CancellationToken ct = default) => Task.FromResult<IReadOnlyList<SiteCallNodeKpiSnapshot>>(Array.Empty<SiteCallNodeKpiSnapshot>());
|
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(DateTime a, DateTime b, CancellationToken ct = default) => Task.FromResult<IReadOnlyList<SiteCallNodeKpiSnapshot>>(Array.Empty<SiteCallNodeKpiSnapshot>());
|
||||||
}
|
}
|
||||||
|
|
||||||
private IActorRef CreateActor(ISiteCallAuditRepository repo, SiteCallAuditOptions options) =>
|
private IActorRef CreateActor(ISiteCallAuditRepository repo, SiteCallAuditOptions options)
|
||||||
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
{
|
||||||
repo,
|
// Register row 61: the actor's repository-taking test ctor is gone. The fixture now
|
||||||
new EmptyEnumerator(),
|
// supplies a stub provider that hands the recording repo out of every scope.
|
||||||
new NoOpPullClient(),
|
var provider = TestSupport.StubRepositoryProvider.Build(
|
||||||
NullLogger<SiteCallAuditActor>.Instance,
|
repo, new EmptyEnumerator(), new NoOpPullClient());
|
||||||
options)));
|
return Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||||
|
provider,
|
||||||
|
options,
|
||||||
|
NullLogger<SiteCallAuditActor>.Instance)));
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------
|
// ---------------------------------------------------------------------
|
||||||
// 1. PurgeTick_CallsPurgeTerminal_WithRetentionThreshold
|
// 1. PurgeTick_CallsPurgeTerminal_WithRetentionThreshold
|
||||||
|
|||||||
+30
-23
@@ -238,13 +238,18 @@ public class SiteCallAuditReconciliationTests : TestKit
|
|||||||
ISiteEnumerator sites,
|
ISiteEnumerator sites,
|
||||||
IPullSiteCallsClient client,
|
IPullSiteCallsClient client,
|
||||||
ISiteCallAuditRepository repo,
|
ISiteCallAuditRepository repo,
|
||||||
SiteCallAuditOptions options) =>
|
SiteCallAuditOptions options)
|
||||||
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
{
|
||||||
repo,
|
// Register row 61: the actor's repository+collaborators test ctor is gone. The
|
||||||
sites,
|
// collaborators now reach it the way production supplies them — resolved from the
|
||||||
client,
|
// provider — and the repository comes out of a scope, serialized because this fixture
|
||||||
NullLogger<SiteCallAuditActor>.Instance,
|
// deliberately shares one instance across every scope.
|
||||||
options)));
|
var provider = TestSupport.StubRepositoryProvider.Build(repo, sites, client);
|
||||||
|
return Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||||
|
provider,
|
||||||
|
options,
|
||||||
|
NullLogger<SiteCallAuditActor>.Instance)));
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------
|
// ---------------------------------------------------------------------
|
||||||
// 1. AbsentRow_PulledFromSite_IsUpserted
|
// 1. AbsentRow_PulledFromSite_IsUpserted
|
||||||
@@ -348,28 +353,30 @@ public class SiteCallAuditReconciliationTests : TestKit
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------
|
// ---------------------------------------------------------------------
|
||||||
// 4. RepoOnly test ctor does NOT start the reconciliation tick
|
// 4. A composition root without the reconciliation collaborators does NOT
|
||||||
|
// start the reconciliation tick
|
||||||
// ---------------------------------------------------------------------
|
// ---------------------------------------------------------------------
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void TestCtor_RepositoryOnly_DoesNotStartReconciliationTick()
|
public void NoReconciliationCollaboratorsRegistered_DoesNotStartReconciliationTick()
|
||||||
{
|
{
|
||||||
// The repo-only test ctor (used by the MSSQL-backed actor tests) injects
|
// Register row 61: this used to be phrased as "the repo-only test ctor gates the tick
|
||||||
// no client/enumerator, so the tick must be gated OFF — otherwise those
|
// off". With that ctor deleted, the same guarantee is expressed the way production
|
||||||
// tests would fire phantom pulls. Build the actor via that ctor and
|
// actually reaches it — a composition root that registers NEITHER IPullSiteCallsClient
|
||||||
// confirm no pull ever happens. We can't observe a non-event directly,
|
// NOR ISiteEnumerator leaves the tick disarmed (GetService returns null for both), so
|
||||||
// so we share a ScriptedPullClient with an isolated actor that DOES run
|
// the MSSQL-backed actor tests fire no phantom pulls. We can't observe a non-event
|
||||||
// the tick to bound the wait, then assert the repo-only actor's client
|
// directly, so we run an isolated actor that DOES tick to bound the wait, then assert
|
||||||
// (a separate instance) recorded nothing.
|
// this actor upserted nothing on its own.
|
||||||
var repo = new RecordingRepo();
|
var repo = new RecordingRepo();
|
||||||
|
var bareProvider = TestSupport.StubRepositoryProvider.Build(repo);
|
||||||
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||||
repo,
|
bareProvider,
|
||||||
NullLogger<SiteCallAuditActor>.Instance,
|
FastTickOptions(),
|
||||||
FastTickOptions())));
|
NullLogger<SiteCallAuditActor>.Instance)));
|
||||||
|
|
||||||
// Run a parallel actor with the full reconciliation ctor and a fast
|
// Run a parallel actor whose provider DOES carry the collaborators, with a fast
|
||||||
// tick; once IT has pulled we know enough wall-clock elapsed that the
|
// tick; once IT has pulled we know enough wall-clock elapsed that the
|
||||||
// repo-only actor would have ticked too, had it been wired.
|
// collaborator-less actor would have ticked too, had it been wired.
|
||||||
var liveClient = new ScriptedPullClient();
|
var liveClient = new ScriptedPullClient();
|
||||||
var liveRepo = new RecordingRepo();
|
var liveRepo = new RecordingRepo();
|
||||||
CreateActor(
|
CreateActor(
|
||||||
@@ -383,8 +390,8 @@ public class SiteCallAuditReconciliationTests : TestKit
|
|||||||
duration: TimeSpan.FromSeconds(3),
|
duration: TimeSpan.FromSeconds(3),
|
||||||
interval: TimeSpan.FromMilliseconds(50));
|
interval: TimeSpan.FromMilliseconds(50));
|
||||||
|
|
||||||
// The repo-only actor never reconciles: it has no client to pull with,
|
// The collaborator-less actor never reconciles: it has no client to pull
|
||||||
// so it upserts nothing on its own.
|
// with, so it upserts nothing on its own.
|
||||||
Assert.Equal(0, repo.UpsertCallCount);
|
Assert.Equal(0, repo.UpsertCallCount);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -64,10 +64,12 @@ public class SiteCallRelayTests : TestKit
|
|||||||
private IActorRef CreateActor(IActorRef centralComm)
|
private IActorRef CreateActor(IActorRef centralComm)
|
||||||
{
|
{
|
||||||
var options = new SiteCallAuditOptions { RelayTimeout = TimeSpan.FromMilliseconds(500) };
|
var options = new SiteCallAuditOptions { RelayTimeout = TimeSpan.FromMilliseconds(500) };
|
||||||
|
// Register row 61: stub provider in place of the deleted repository ctor.
|
||||||
|
var provider = TestSupport.StubRepositoryProvider.Build(new ThrowingRepository());
|
||||||
var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||||
new ThrowingRepository(),
|
provider,
|
||||||
NullLogger<SiteCallAuditActor>.Instance,
|
options,
|
||||||
options)));
|
NullLogger<SiteCallAuditActor>.Instance)));
|
||||||
actor.Tell(new RegisterCentralCommunication(centralComm));
|
actor.Tell(new RegisterCentralCommunication(centralComm));
|
||||||
return actor;
|
return actor;
|
||||||
}
|
}
|
||||||
@@ -202,10 +204,11 @@ public class SiteCallRelayTests : TestKit
|
|||||||
// No RegisterCentralCommunication — the actor has no transport to reach
|
// No RegisterCentralCommunication — the actor has no transport to reach
|
||||||
// any site, so the only honest answer is "unreachable".
|
// any site, so the only honest answer is "unreachable".
|
||||||
var options = new SiteCallAuditOptions { RelayTimeout = TimeSpan.FromMilliseconds(500) };
|
var options = new SiteCallAuditOptions { RelayTimeout = TimeSpan.FromMilliseconds(500) };
|
||||||
|
var provider = TestSupport.StubRepositoryProvider.Build(new ThrowingRepository());
|
||||||
var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||||
new ThrowingRepository(),
|
provider,
|
||||||
NullLogger<SiteCallAuditActor>.Instance,
|
options,
|
||||||
options)));
|
NullLogger<SiteCallAuditActor>.Instance)));
|
||||||
|
|
||||||
actor.Tell(new RetrySiteCallRequest("corr-7", Guid.NewGuid(), "site-north"));
|
actor.Tell(new RetrySiteCallRequest("corr-7", Guid.NewGuid(), "site-north"));
|
||||||
|
|
||||||
@@ -286,11 +289,15 @@ public class SiteCallRelayTests : TestKit
|
|||||||
{
|
{
|
||||||
var writer = new RecordingCentralAuditWriter();
|
var writer = new RecordingCentralAuditWriter();
|
||||||
var options = new SiteCallAuditOptions { RelayTimeout = TimeSpan.FromMilliseconds(500) };
|
var options = new SiteCallAuditOptions { RelayTimeout = TimeSpan.FromMilliseconds(500) };
|
||||||
var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
// The audit writer now reaches the actor through the provider (production resolves it
|
||||||
|
// the same way, from the root provider) rather than through a ctor parameter.
|
||||||
|
var provider = TestSupport.StubRepositoryProvider.Build(
|
||||||
new RowReturningRepository(row),
|
new RowReturningRepository(row),
|
||||||
NullLogger<SiteCallAuditActor>.Instance,
|
auditWriter: (ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ICentralAuditWriter)writer);
|
||||||
|
var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||||
|
provider,
|
||||||
options,
|
options,
|
||||||
(ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ICentralAuditWriter)writer)));
|
NullLogger<SiteCallAuditActor>.Instance)));
|
||||||
actor.Tell(new RegisterCentralCommunication(centralComm));
|
actor.Tell(new RegisterCentralCommunication(centralComm));
|
||||||
return (actor, writer);
|
return (actor, writer);
|
||||||
}
|
}
|
||||||
|
|||||||
+162
@@ -0,0 +1,162 @@
|
|||||||
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests.TestSupport;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Deferred-work register row 61 — the replacement for <c>SiteCallAuditActor</c>'s deleted
|
||||||
|
/// repository-taking "test-mode" constructors.
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// Those constructors cached a scoped repository on an actor field for the actor's whole
|
||||||
|
/// lifetime. Nothing in production selected them, but the shape was one call site away from
|
||||||
|
/// re-introducing a captive-dependency bug, so it had to be allowlisted in the Host tests'
|
||||||
|
/// <c>CaptiveDependencyGuard</c>. The actor now has a single <see cref="IServiceProvider"/>
|
||||||
|
/// constructor; a fixture that wants a concrete repository builds one of these instead.
|
||||||
|
/// </para>
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// The actor's contract is unchanged and unweakened — it still resolves per scope and holds
|
||||||
|
/// nothing. Handing the SAME instance out of every scope is now the FIXTURE's explicit
|
||||||
|
/// choice, which is also why the serializing decorator below lives here rather than in
|
||||||
|
/// product code: it exists solely to compensate for that choice.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
internal static class StubRepositoryProvider
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Builds a provider whose every scope yields the same repository instance, wrapped in
|
||||||
|
/// <see cref="SerializingSiteCallAuditRepository"/>, plus any collaborators supplied.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="repository">Repository instance shared by every scope.</param>
|
||||||
|
/// <param name="siteEnumerator">Optional site enumerator; omit to leave the reconciliation tick disarmed.</param>
|
||||||
|
/// <param name="pullClient">Optional pull client; omit to leave the reconciliation tick disarmed.</param>
|
||||||
|
/// <param name="auditWriter">Optional central direct-write audit sink for relay actions.</param>
|
||||||
|
/// <param name="serialize">
|
||||||
|
/// False registers the repository unwrapped — for fixtures asserting on call ordering or
|
||||||
|
/// identity where the decorator would get in the way.
|
||||||
|
/// </param>
|
||||||
|
public static ServiceProvider Build(
|
||||||
|
ISiteCallAuditRepository repository,
|
||||||
|
ISiteEnumerator? siteEnumerator = null,
|
||||||
|
IPullSiteCallsClient? pullClient = null,
|
||||||
|
ICentralAuditWriter? auditWriter = null,
|
||||||
|
bool serialize = true)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(repository);
|
||||||
|
|
||||||
|
var shared = serialize ? new SerializingSiteCallAuditRepository(repository) : repository;
|
||||||
|
|
||||||
|
var services = new ServiceCollection();
|
||||||
|
services.AddScoped(_ => shared);
|
||||||
|
|
||||||
|
if (siteEnumerator is not null)
|
||||||
|
{
|
||||||
|
services.AddSingleton(siteEnumerator);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (pullClient is not null)
|
||||||
|
{
|
||||||
|
services.AddSingleton(pullClient);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (auditWriter is not null)
|
||||||
|
{
|
||||||
|
services.AddSingleton(auditWriter);
|
||||||
|
}
|
||||||
|
|
||||||
|
return services.BuildServiceProvider();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Serializing wrapper for a repository instance shared across every DI scope in a test.
|
||||||
|
/// One call at a time, in arrival order.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// In production every consumer of <c>SiteCallAuditActor</c>'s repository gets its OWN DI
|
||||||
|
/// scope — one per message, one per background pass — so the off-mailbox reconciliation and
|
||||||
|
/// purge passes can safely run alongside mailbox handlers: separate scopes mean separate
|
||||||
|
/// <c>DbContext</c>s. A fixture that hands ONE instance (typically wrapping one
|
||||||
|
/// <c>DbContext</c>) to every scope breaks that assumption, and <c>DbContext</c> forbids
|
||||||
|
/// concurrent operations — "A second operation was started on this context instance" is a
|
||||||
|
/// hard fault, and the interleaving is nondeterministic, so it surfaces as a flaky test
|
||||||
|
/// rather than a reliable one.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// Serializing at the CALL, not around the whole pass, is deliberate: it removes the
|
||||||
|
/// concurrency hazard while preserving the property the off-mailbox passes exist for — a long
|
||||||
|
/// drain (blocked in a network pull, holding no repository call) still lets ingest, query and
|
||||||
|
/// KPI messages be answered. Suspending the mailbox for the duration of a pass would have
|
||||||
|
/// been simpler and would have invalidated exactly those regression tests.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// This type used to live inside the product actor, reachable only from its test
|
||||||
|
/// constructors. It moved here with register row 61: production never constructed it, and the
|
||||||
|
/// hazard it guards is created by the fixture, not by the actor.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
internal sealed class SerializingSiteCallAuditRepository : ISiteCallAuditRepository
|
||||||
|
{
|
||||||
|
private readonly ISiteCallAuditRepository _inner;
|
||||||
|
private readonly SemaphoreSlim _gate = new(1, 1);
|
||||||
|
|
||||||
|
public SerializingSiteCallAuditRepository(ISiteCallAuditRepository inner) => _inner = inner;
|
||||||
|
|
||||||
|
public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) =>
|
||||||
|
RunAsync(() => _inner.UpsertAsync(siteCall, ct));
|
||||||
|
|
||||||
|
public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
|
||||||
|
RunAsync(() => _inner.GetAsync(id, ct));
|
||||||
|
|
||||||
|
public Task<IReadOnlyList<SiteCall>> QueryAsync(
|
||||||
|
SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
|
||||||
|
RunAsync(() => _inner.QueryAsync(filter, paging, ct));
|
||||||
|
|
||||||
|
public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
|
||||||
|
RunAsync(() => _inner.PurgeTerminalAsync(olderThanUtc, ct));
|
||||||
|
|
||||||
|
public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
|
||||||
|
DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) =>
|
||||||
|
RunAsync(() => _inner.ComputeKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct));
|
||||||
|
|
||||||
|
public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
|
||||||
|
DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) =>
|
||||||
|
RunAsync(() => _inner.ComputePerSiteKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct));
|
||||||
|
|
||||||
|
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
|
||||||
|
DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) =>
|
||||||
|
RunAsync(() => _inner.ComputePerNodeKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct));
|
||||||
|
|
||||||
|
private async Task RunAsync(Func<Task> call)
|
||||||
|
{
|
||||||
|
await _gate.WaitAsync().ConfigureAwait(false);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await call().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_gate.Release();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<T> RunAsync<T>(Func<Task<T>> call)
|
||||||
|
{
|
||||||
|
await _gate.WaitAsync().ConfigureAwait(false);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return await call().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_gate.Release();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+40
-8
@@ -65,6 +65,21 @@ public class DeploymentWarmThenGateTests : TestKit, IDisposable
|
|||||||
new SiteRuntimeOptions(), NullLogger<DeploymentManagerActor>.Instance, null,
|
new SiteRuntimeOptions(), NullLogger<DeploymentManagerActor>.Instance, null,
|
||||||
healthCollector)));
|
healthCollector)));
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Register row 40: starts a Deployment Manager and blocks until it has PROVABLY finished
|
||||||
|
/// its startup config load, by waiting for its first instance-count report. Replaces a bare
|
||||||
|
/// <c>await Task.Delay(500)</c>, which merely hoped 500 ms was enough and — for the empty-site
|
||||||
|
/// assertion — could not tell "startup ran and found nothing" from "startup had not run yet".
|
||||||
|
/// </summary>
|
||||||
|
private (IActorRef Dm, DeployedCountCollector Health) CreateStartedDeploymentManager()
|
||||||
|
{
|
||||||
|
var health = new DeployedCountCollector();
|
||||||
|
var dm = CreateDeploymentManager(health);
|
||||||
|
AwaitCondition(() => health.ReportCount > 0, TimeSpan.FromSeconds(20),
|
||||||
|
TimeSpan.FromMilliseconds(25));
|
||||||
|
return (dm, health);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Captures the deployed-instance count the Deployment Manager reports. The count is
|
/// Captures the deployed-instance count the Deployment Manager reports. The count is
|
||||||
/// mutated only on the actor thread — <c>HandleDeploy</c> adds the instance name,
|
/// mutated only on the actor thread — <c>HandleDeploy</c> adds the instance name,
|
||||||
@@ -73,7 +88,20 @@ public class DeploymentWarmThenGateTests : TestKit, IDisposable
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private sealed class DeployedCountCollector : ISiteHealthCollector
|
private sealed class DeployedCountCollector : ISiteHealthCollector
|
||||||
{
|
{
|
||||||
|
private int _reports;
|
||||||
|
|
||||||
public int LastDeployedCount { get; private set; }
|
public int LastDeployedCount { get; private set; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Register row 40: how many times the Deployment Manager has reported instance
|
||||||
|
/// counts. The manager calls <see cref="SetInstanceCounts"/> unconditionally the
|
||||||
|
/// moment its startup config load completes (<c>HandleStartupConfigsLoaded</c>) —
|
||||||
|
/// including for an empty site — so a non-zero value here is the explicit
|
||||||
|
/// observable edge for "startup has finished", replacing a wall-clock sleep.
|
||||||
|
/// Without it, <c>LastDeployedCount == 0</c> is satisfied by the field's initial
|
||||||
|
/// value and proves nothing about the manager having started at all.
|
||||||
|
/// </summary>
|
||||||
|
public int ReportCount => Volatile.Read(ref _reports);
|
||||||
public void IncrementScriptError() { }
|
public void IncrementScriptError() { }
|
||||||
public void IncrementAlarmError() { }
|
public void IncrementAlarmError() { }
|
||||||
public void IncrementDeadLetter() { }
|
public void IncrementDeadLetter() { }
|
||||||
@@ -86,7 +114,12 @@ public class DeploymentWarmThenGateTests : TestKit, IDisposable
|
|||||||
public void UpdateConnectionEndpoint(string connectionName, string endpoint) { }
|
public void UpdateConnectionEndpoint(string connectionName, string endpoint) { }
|
||||||
public void UpdateTagQuality(string connectionName, int good, int bad, int uncertain) { }
|
public void UpdateTagQuality(string connectionName, int good, int bad, int uncertain) { }
|
||||||
public void SetStoreAndForwardDepths(IReadOnlyDictionary<string, int> depths) { }
|
public void SetStoreAndForwardDepths(IReadOnlyDictionary<string, int> depths) { }
|
||||||
public void SetInstanceCounts(int deployed, int enabled, int disabled) => LastDeployedCount = deployed;
|
public void SetInstanceCounts(int deployed, int enabled, int disabled)
|
||||||
|
{
|
||||||
|
LastDeployedCount = deployed;
|
||||||
|
Interlocked.Increment(ref _reports);
|
||||||
|
}
|
||||||
|
|
||||||
public void SetParkedMessageCount(int count) { }
|
public void SetParkedMessageCount(int count) { }
|
||||||
public void SetNodeHostname(string hostname) { }
|
public void SetNodeHostname(string hostname) { }
|
||||||
public void SetClusterNodes(IReadOnlyList<Commons.Messages.Health.NodeStatus> nodes) { }
|
public void SetClusterNodes(IReadOnlyList<Commons.Messages.Health.NodeStatus> nodes) { }
|
||||||
@@ -112,9 +145,10 @@ public class DeploymentWarmThenGateTests : TestKit, IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task DeleteArrivingDuringTheCompileWarm_IsQueuedAndAppliedAfterTheDeploy()
|
public async Task DeleteArrivingDuringTheCompileWarm_IsQueuedAndAppliedAfterTheDeploy()
|
||||||
{
|
{
|
||||||
var health = new DeployedCountCollector();
|
// The startup load has demonstrably completed (it reported counts at least once), so
|
||||||
var dm = CreateDeploymentManager(health);
|
// the zero below is the manager's own observation of an empty site rather than an
|
||||||
await Task.Delay(500); // empty startup
|
// untouched field.
|
||||||
|
var (dm, health) = CreateStartedDeploymentManager();
|
||||||
Assert.Equal(0, health.LastDeployedCount);
|
Assert.Equal(0, health.LastDeployedCount);
|
||||||
|
|
||||||
var deployProbe = CreateTestProbe();
|
var deployProbe = CreateTestProbe();
|
||||||
@@ -145,8 +179,7 @@ public class DeploymentWarmThenGateTests : TestKit, IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task SecondDeployDuringTheWarm_SupersedesTheFirst_AndAnswersItsDeployer()
|
public async Task SecondDeployDuringTheWarm_SupersedesTheFirst_AndAnswersItsDeployer()
|
||||||
{
|
{
|
||||||
var dm = CreateDeploymentManager();
|
var (dm, _) = CreateStartedDeploymentManager();
|
||||||
await Task.Delay(500);
|
|
||||||
|
|
||||||
var first = CreateTestProbe();
|
var first = CreateTestProbe();
|
||||||
var second = CreateTestProbe();
|
var second = CreateTestProbe();
|
||||||
@@ -177,8 +210,7 @@ public class DeploymentWarmThenGateTests : TestKit, IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task DeploysForDifferentInstances_DoNotBlockEachOther()
|
public async Task DeploysForDifferentInstances_DoNotBlockEachOther()
|
||||||
{
|
{
|
||||||
var dm = CreateDeploymentManager();
|
var (dm, _) = CreateStartedDeploymentManager();
|
||||||
await Task.Delay(500);
|
|
||||||
|
|
||||||
var probeX = CreateTestProbe();
|
var probeX = CreateTestProbe();
|
||||||
var probeY = CreateTestProbe();
|
var probeY = CreateTestProbe();
|
||||||
|
|||||||
+49
-15
@@ -107,16 +107,54 @@ public class InstanceActorChildAttributeRaceTests : TestKit, IDisposable
|
|||||||
]
|
]
|
||||||
};
|
};
|
||||||
|
|
||||||
/// <summary>Resolves the live actor instance behind a local <see cref="IActorRef"/>.</summary>
|
/// <summary>
|
||||||
private static object GetActorInstance(IActorRef actorRef)
|
/// Resolves a named child and the live actor instance behind it, retrying until BOTH
|
||||||
|
/// exist (register row 40 — absence/settle timing pins).
|
||||||
|
///
|
||||||
|
/// <para>Two distinct publication races sit between "the parent reported N children" and
|
||||||
|
/// "the child's actor object is reachable by reflection", and neither is ordered against
|
||||||
|
/// the test thread: <see cref="ActorSelection.ResolveOne"/> throws
|
||||||
|
/// <see cref="ActorNotFoundException"/> until the child's cell is published on the parent,
|
||||||
|
/// and the cell's internal <c>Actor</c> property stays null until the child's constructor
|
||||||
|
/// has run on the child's own mailbox thread. A single-shot resolve therefore fails
|
||||||
|
/// intermittently under load — observed both ways in full-suite runs ("subject was null"
|
||||||
|
/// from ResolveOne, and a null instance from the reflection read). Retrying to a bounded
|
||||||
|
/// deadline removes the race without weakening the assertion: what the test proves is
|
||||||
|
/// still the seed-dictionary isolation contract, checked once both objects exist.</para>
|
||||||
|
/// </summary>
|
||||||
|
private async Task<T> ResolveChildInstanceAsync<T>(IActorRef parent, string childName)
|
||||||
|
where T : class
|
||||||
{
|
{
|
||||||
var cell = ((ActorRefWithCell)actorRef).Underlying;
|
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(15);
|
||||||
// ActorCell exposes the actor instance via its internal Actor property.
|
while (true)
|
||||||
var actorProp = cell.GetType().GetProperty(
|
{
|
||||||
"Actor", BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public);
|
try
|
||||||
var instance = actorProp!.GetValue(cell);
|
{
|
||||||
Assert.NotNull(instance);
|
var child = await Sys.ActorSelection(parent.Path / childName)
|
||||||
return instance!;
|
.ResolveOne(TimeSpan.FromSeconds(1));
|
||||||
|
var cell = ((ActorRefWithCell)child).Underlying;
|
||||||
|
// ActorCell exposes the actor instance via its internal Actor property.
|
||||||
|
var actorProp = cell.GetType().GetProperty(
|
||||||
|
"Actor", BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public);
|
||||||
|
if (actorProp!.GetValue(cell) is T typed)
|
||||||
|
{
|
||||||
|
return typed;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (ActorNotFoundException)
|
||||||
|
{
|
||||||
|
// The child's cell is not published on the parent yet — retry below.
|
||||||
|
}
|
||||||
|
|
||||||
|
if (DateTime.UtcNow > deadline)
|
||||||
|
{
|
||||||
|
throw new TimeoutException(
|
||||||
|
$"Child actor '{childName}' (or its {typeof(T).Name} instance) never " +
|
||||||
|
"materialised within the resolve deadline.");
|
||||||
|
}
|
||||||
|
|
||||||
|
await Task.Delay(25);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private static Dictionary<string, object?> GetPrivateAttributes(InstanceActor instance)
|
private static Dictionary<string, object?> GetPrivateAttributes(InstanceActor instance)
|
||||||
@@ -157,9 +195,7 @@ public class InstanceActorChildAttributeRaceTests : TestKit, IDisposable
|
|||||||
// mutates on another thread (SiteRuntime-017).
|
// mutates on another thread (SiteRuntime-017).
|
||||||
foreach (var name in new[] { "WorkerA", "WorkerB" })
|
foreach (var name in new[] { "WorkerA", "WorkerB" })
|
||||||
{
|
{
|
||||||
var child = await Sys.ActorSelection(testRef.Path / $"script-{name}")
|
var scriptActor = await ResolveChildInstanceAsync<ScriptActor>(testRef, $"script-{name}");
|
||||||
.ResolveOne(TimeSpan.FromSeconds(5));
|
|
||||||
var scriptActor = (ScriptActor)GetActorInstance(child);
|
|
||||||
|
|
||||||
Assert.NotNull(scriptActor.SeedAttributesReference);
|
Assert.NotNull(scriptActor.SeedAttributesReference);
|
||||||
Assert.False(
|
Assert.False(
|
||||||
@@ -177,9 +213,7 @@ public class InstanceActorChildAttributeRaceTests : TestKit, IDisposable
|
|||||||
}
|
}
|
||||||
|
|
||||||
// The Alarm Actor must likewise be seeded from an isolated copy.
|
// The Alarm Actor must likewise be seeded from an isolated copy.
|
||||||
var alarmChild = await Sys.ActorSelection(testRef.Path / "alarm-HighTemp")
|
var alarmActor = await ResolveChildInstanceAsync<AlarmActor>(testRef, "alarm-HighTemp");
|
||||||
.ResolveOne(TimeSpan.FromSeconds(5));
|
|
||||||
var alarmActor = (AlarmActor)GetActorInstance(alarmChild);
|
|
||||||
|
|
||||||
Assert.NotNull(alarmActor.SeedAttributesReference);
|
Assert.NotNull(alarmActor.SeedAttributesReference);
|
||||||
Assert.False(
|
Assert.False(
|
||||||
|
|||||||
+69
-7
@@ -101,8 +101,26 @@ public class ScriptDeadlineAtEnqueueTests : TestKit, IDisposable
|
|||||||
var caller = CreateTestProbe();
|
var caller = CreateTestProbe();
|
||||||
late.Tell(new ScriptCallRequest("Late", null, 0, "corr-late"), caller.Ref);
|
late.Tell(new ScriptCallRequest("Late", null, 0, "corr-late"), caller.Ref);
|
||||||
|
|
||||||
// Let its whole budget elapse while it is still queued, then free the worker.
|
// Register row 40 — edge 1 (replaces a bare Thread.Sleep(2s)): the late run must
|
||||||
Thread.Sleep(TimeSpan.FromSeconds(2));
|
// actually be SITTING IN THE SCHEDULER QUEUE. The old test only slept and hoped;
|
||||||
|
// a blocker that failed to occupy the worker would have let the late run execute
|
||||||
|
// normally, and the shed-at-dequeue path under test would never have been reached.
|
||||||
|
AwaitCondition(() => scheduler.QueueDepth >= 1, TimeSpan.FromSeconds(15),
|
||||||
|
TimeSpan.FromMilliseconds(25));
|
||||||
|
|
||||||
|
// Register row 40 — edge 2: hold until the late run's 1 s budget has provably
|
||||||
|
// elapsed, re-reading the queue on every poll and bailing out early the instant the
|
||||||
|
// run leaves the queue. The post-condition below then proves the budget burned
|
||||||
|
// *while queued* — the actual design claim ("queue wait consumes the run's budget"),
|
||||||
|
// which a fixed sleep asserted nothing about.
|
||||||
|
var queuedFor = System.Diagnostics.Stopwatch.StartNew();
|
||||||
|
AwaitCondition(
|
||||||
|
() => queuedFor.Elapsed >= TimeSpan.FromSeconds(1.5) || scheduler.QueueDepth == 0,
|
||||||
|
TimeSpan.FromSeconds(15), TimeSpan.FromMilliseconds(25));
|
||||||
|
Assert.True(scheduler.QueueDepth >= 1,
|
||||||
|
"the late run left the queue before its budget elapsed, so this run never " +
|
||||||
|
"exercised the shed-at-dequeue path");
|
||||||
|
|
||||||
DeadlineHooks.Gate.Release();
|
DeadlineHooks.Gate.Release();
|
||||||
|
|
||||||
var result = caller.ExpectMsg<ScriptCallResult>(TimeSpan.FromSeconds(15));
|
var result = caller.ExpectMsg<ScriptCallResult>(TimeSpan.FromSeconds(15));
|
||||||
@@ -123,12 +141,56 @@ public class ScriptDeadlineAtEnqueueTests : TestKit, IDisposable
|
|||||||
Assert.DoesNotContain(rows, r => r.Message.Contains("started", StringComparison.OrdinalIgnoreCase));
|
Assert.DoesNotContain(rows, r => r.Message.Contains("started", StringComparison.OrdinalIgnoreCase));
|
||||||
}, TimeSpan.FromSeconds(5));
|
}, TimeSpan.FromSeconds(5));
|
||||||
|
|
||||||
// And the watchdog did NOT report it as a stuck thread: it was cancelled while queued,
|
// ── Register row 40 — the watchdog absence, proved by a positive control ──
|
||||||
// so it never held a worker.
|
//
|
||||||
Thread.Sleep(TimeSpan.FromSeconds(2)); // well past the 1 s grace
|
// The claim is that the shed run was NOT reported as a stuck thread: it was cancelled
|
||||||
|
// while queued, so it never held a worker. The old test slept 2 s and asserted the
|
||||||
|
// absence, which false-passes whenever the watchdog simply had not got around to
|
||||||
|
// evaluating anything yet — the assertion could not distinguish "the watchdog looked
|
||||||
|
// and correctly stayed silent" from "the watchdog never looked".
|
||||||
|
//
|
||||||
|
// So: enqueue a script that genuinely wedges a worker, on the SAME scheduler and with
|
||||||
|
// the SAME 1 s timeout + 1 s grace, and armed strictly LATER than the late run's
|
||||||
|
// watchdog. Its stuck report is the observable edge — once it lands, every watchdog
|
||||||
|
// armed at or before it has provably run its grace delay and made its decision.
|
||||||
|
//
|
||||||
|
// The body blocks its worker SYNCHRONOUSLY (so the watchdog can detach the thread it
|
||||||
|
// holds) but for a BOUNDED 2.5 s that comfortably clears 1 s timeout + 1 s grace and
|
||||||
|
// then self-releases. A gate held until test teardown would work too, but it leaves a
|
||||||
|
// blocked OS thread alive past the test and perturbs timing-sensitive tests elsewhere
|
||||||
|
// in this assembly.
|
||||||
|
var wedged = BuildScriptActor(
|
||||||
|
"Wedged",
|
||||||
|
CompileRaw("System.Threading.Thread.Sleep(2500); return null;"),
|
||||||
|
new SiteRuntimeOptions { ScriptExecutionTimeoutSeconds = 1, StuckScriptGraceMs = 1000 },
|
||||||
|
scheduler, new SingleServiceProvider(siteLog));
|
||||||
|
wedged.Tell(new ScriptCallRequest("Wedged", null, 0, "corr-wedged"), ActorRefs.NoSender);
|
||||||
|
|
||||||
|
// Positive event, bounded wait: the watchdog fires, names the wedged script and
|
||||||
|
// detaches its worker.
|
||||||
|
AwaitAssert(() =>
|
||||||
|
{
|
||||||
|
Assert.Contains(siteLog.OfType("script"),
|
||||||
|
r => r.Message.Contains("'Wedged'", StringComparison.Ordinal) &&
|
||||||
|
r.Message.Contains("still executing", StringComparison.OrdinalIgnoreCase));
|
||||||
|
}, TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(50));
|
||||||
|
|
||||||
|
// Negative control, evaluated only now that a later-armed watchdog has demonstrably
|
||||||
|
// completed its grace delay: the shed run was never reported stuck.
|
||||||
Assert.DoesNotContain(siteLog.OfType("script"),
|
Assert.DoesNotContain(siteLog.OfType("script"),
|
||||||
r => r.Message.Contains("still executing", StringComparison.OrdinalIgnoreCase));
|
r => r.Message.Contains("'Late'", StringComparison.Ordinal) &&
|
||||||
Assert.Equal(0, scheduler.DetachedThreadCount);
|
r.Message.Contains("still executing", StringComparison.OrdinalIgnoreCase));
|
||||||
|
|
||||||
|
// And the detach counter moved for exactly one run — the wedged one. The old
|
||||||
|
// `Assert.Equal(0, DetachedThreadCount)` was satisfied by a counter that never moves
|
||||||
|
// for any reason; this pins that it does move, and that the shed run did not move it.
|
||||||
|
Assert.Equal(1, scheduler.DetachedThreadCount);
|
||||||
|
|
||||||
|
// Wait for the positive control's body to unwind and its detached worker to exit
|
||||||
|
// BEFORE the scheduler is disposed at end of scope, so the test leaves no threads
|
||||||
|
// behind for the rest of the assembly to contend with.
|
||||||
|
AwaitCondition(() => scheduler.BusyThreadCount == 0 && scheduler.DetachedThreadCount == 0,
|
||||||
|
TimeSpan.FromSeconds(20), TimeSpan.FromMilliseconds(25));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -17,7 +17,13 @@ namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
|
|||||||
/// <c>DefaultRetryInterval=Zero</c> so the timer-driven retry sweep is
|
/// <c>DefaultRetryInterval=Zero</c> so the timer-driven retry sweep is
|
||||||
/// short-circuited by directly invoking
|
/// short-circuited by directly invoking
|
||||||
/// <see cref="StoreAndForwardService.RetryPendingMessagesAsync"/>.
|
/// <see cref="StoreAndForwardService.RetryPendingMessagesAsync"/>.
|
||||||
|
///
|
||||||
|
/// Register row 32: one fact here calls <c>StartAsync</c>, which writes the
|
||||||
|
/// process-global queue-depth provider slot, so this class joins
|
||||||
|
/// <see cref="QueueDepthGaugeCollection"/> to stay serialized against
|
||||||
|
/// <see cref="QueueDepthGaugeTests"/>.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
[Collection(QueueDepthGaugeCollection.Name)]
|
||||||
public class CachedCallAttemptEmissionTests : IAsyncLifetime, IDisposable
|
public class CachedCallAttemptEmissionTests : IAsyncLifetime, IDisposable
|
||||||
{
|
{
|
||||||
private readonly TestLocalDb _localDb;
|
private readonly TestLocalDb _localDb;
|
||||||
|
|||||||
@@ -0,0 +1,124 @@
|
|||||||
|
using System.Runtime.CompilerServices;
|
||||||
|
using System.Text.RegularExpressions;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Deferred-work register row 32 — isolation for the process-global StoreAndForward
|
||||||
|
/// queue-depth gauge.
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// <c>scadabridge.store_and_forward.queue.depth</c> is an <see cref="System.Diagnostics.Metrics.ObservableGauge{T}"/>
|
||||||
|
/// created once on <c>ScadaBridgeTelemetry</c>'s static meter, reading a single
|
||||||
|
/// process-global provider slot (<c>SetQueueDepthProvider</c> /
|
||||||
|
/// <c>ClearQueueDepthProvider</c>). Every <c>StoreAndForwardService.StartAsync</c>
|
||||||
|
/// writes that one slot — last registrant wins. That is the PRODUCT design (one S&F
|
||||||
|
/// engine per node), and three of <see cref="QueueDepthGaugeTests"/>' facts assert
|
||||||
|
/// exactly the slot-ownership semantics (compare-and-clear identity, stop/restart
|
||||||
|
/// re-seed), so they cannot be rewritten against a per-test meter without changing
|
||||||
|
/// what they prove — the global slot IS their subject.
|
||||||
|
/// </para>
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// The resource therefore needs mutual exclusion, not duplication: every test class in
|
||||||
|
/// this assembly that starts a <c>StoreAndForwardService</c> (and so writes the slot)
|
||||||
|
/// joins this collection, which xUnit runs serially with respect to itself. Classes
|
||||||
|
/// that never start a service stay free to run in parallel. Before this, a concurrent
|
||||||
|
/// <c>StartAsync</c> in a sibling class clobbered the gauge mid-assertion and
|
||||||
|
/// <c>Gauge_TracksBufferedDepth_AcrossEnqueueDrainAndPark</c> read the sibling's depth
|
||||||
|
/// (observed: expected 0, actual 6 — roughly 2 runs in 5 with
|
||||||
|
/// <c>parallelizeTestCollections: true</c>).
|
||||||
|
/// </para>
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// <see cref="QueueDepthGaugeCollectionGuardTests"/> keeps the membership honest: a new
|
||||||
|
/// test class that starts a service without joining this collection fails the guard
|
||||||
|
/// rather than silently reintroducing the flake.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
[CollectionDefinition(Name)]
|
||||||
|
public sealed class QueueDepthGaugeCollection
|
||||||
|
{
|
||||||
|
/// <summary>Collection name; referenced from <c>[Collection(...)]</c> on member classes.</summary>
|
||||||
|
public const string Name = "StoreAndForward queue-depth gauge (process-global provider slot)";
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Staleness self-test for <see cref="QueueDepthGaugeCollection"/>: asserts that every
|
||||||
|
/// source file in this test project which both constructs a <c>StoreAndForwardService</c>
|
||||||
|
/// and calls <c>StartAsync</c> — i.e. writes the process-global queue-depth provider
|
||||||
|
/// slot — declares <c>[Collection(QueueDepthGaugeCollection.Name)]</c>.
|
||||||
|
///
|
||||||
|
/// <para>
|
||||||
|
/// The project directory is resolved from <see cref="CallerFilePathAttribute"/> (baked in
|
||||||
|
/// at compile time by the compiler that built these sources), not from the test host's
|
||||||
|
/// working directory or the output folder, so the scan finds the sources that produced
|
||||||
|
/// this assembly.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
public class QueueDepthGaugeCollectionGuardTests
|
||||||
|
{
|
||||||
|
/// <summary>Absolute path of this source file, captured at compile time.</summary>
|
||||||
|
private static string ThisFilePath([CallerFilePath] string path = "") => path;
|
||||||
|
|
||||||
|
/// <summary>Matches a construction of the service under a global-slot-writing test.</summary>
|
||||||
|
private static readonly Regex ConstructsService =
|
||||||
|
new(@"new\s+StoreAndForwardService\s*\(", RegexOptions.Compiled);
|
||||||
|
|
||||||
|
/// <summary>Matches a start call, which is what actually writes the global provider slot.</summary>
|
||||||
|
private static readonly Regex StartsService =
|
||||||
|
new(@"\.StartAsync\s*\(", RegexOptions.Compiled);
|
||||||
|
|
||||||
|
/// <summary>Matches the collection attribute that puts a class in the serialized group.</summary>
|
||||||
|
private static readonly Regex DeclaresCollection =
|
||||||
|
new(@"\[\s*Collection\s*\(\s*QueueDepthGaugeCollection\.Name\s*\)\s*\]", RegexOptions.Compiled);
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void EverySourceFileThatStartsAService_JoinsTheGaugeCollection()
|
||||||
|
{
|
||||||
|
var projectDir = Path.GetDirectoryName(ThisFilePath())!;
|
||||||
|
Assert.True(Directory.Exists(projectDir),
|
||||||
|
$"Guard could not locate the test project sources at '{projectDir}'. " +
|
||||||
|
"The [CallerFilePath] anchor must point at this project's directory.");
|
||||||
|
|
||||||
|
var offenders = new List<string>();
|
||||||
|
var members = new List<string>();
|
||||||
|
|
||||||
|
foreach (var file in Directory.EnumerateFiles(projectDir, "*.cs", SearchOption.AllDirectories))
|
||||||
|
{
|
||||||
|
// Skip generated output (bin/obj) and this guard's own definition file.
|
||||||
|
var relative = Path.GetRelativePath(projectDir, file);
|
||||||
|
if (relative.StartsWith("bin", StringComparison.Ordinal) ||
|
||||||
|
relative.StartsWith("obj", StringComparison.Ordinal))
|
||||||
|
{
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
var text = File.ReadAllText(file);
|
||||||
|
if (!ConstructsService.IsMatch(text) || !StartsService.IsMatch(text))
|
||||||
|
{
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (DeclaresCollection.IsMatch(text))
|
||||||
|
{
|
||||||
|
members.Add(relative);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
offenders.Add(relative);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Sanity: the scan must actually find files, or a rename has silently defanged it.
|
||||||
|
Assert.True(members.Count + offenders.Count > 0,
|
||||||
|
"Guard found no source file constructing and starting a StoreAndForwardService — " +
|
||||||
|
"the scan patterns are stale.");
|
||||||
|
|
||||||
|
Assert.True(offenders.Count == 0,
|
||||||
|
"These test sources start a StoreAndForwardService (writing the process-global " +
|
||||||
|
$"queue-depth provider slot) without joining [Collection(QueueDepthGaugeCollection.Name)]: " +
|
||||||
|
string.Join(", ", offenders) +
|
||||||
|
". Add the attribute, or the queue-depth gauge assertions will flake under parallel collections.");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -17,7 +17,13 @@ namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
|
|||||||
/// and does no I/O, which is the whole point of caching the count). <see cref="StartAsync"/>
|
/// and does no I/O, which is the whole point of caching the count). <see cref="StartAsync"/>
|
||||||
/// seeds the counter from storage and registers the provider against this service
|
/// seeds the counter from storage and registers the provider against this service
|
||||||
/// instance, so the gauge resolves to this test's counter.
|
/// instance, so the gauge resolves to this test's counter.
|
||||||
|
///
|
||||||
|
/// Register row 32: that provider slot is process-global, so this class joins
|
||||||
|
/// <see cref="QueueDepthGaugeCollection"/> along with every other class in this assembly
|
||||||
|
/// that starts a service — see that type for why mutual exclusion (rather than a per-test
|
||||||
|
/// meter) is the correct isolation here.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
[Collection(QueueDepthGaugeCollection.Name)]
|
||||||
public class QueueDepthGaugeTests : IAsyncLifetime, IDisposable
|
public class QueueDepthGaugeTests : IAsyncLifetime, IDisposable
|
||||||
{
|
{
|
||||||
private readonly TestLocalDb _localDb;
|
private readonly TestLocalDb _localDb;
|
||||||
|
|||||||
@@ -9,7 +9,13 @@ namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
|
|||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// WP-10/12/13/14: Tests for the StoreAndForwardService retry engine and management.
|
/// WP-10/12/13/14: Tests for the StoreAndForwardService retry engine and management.
|
||||||
|
///
|
||||||
|
/// Register row 32: several facts here call <c>StartAsync</c>, which writes the
|
||||||
|
/// process-global queue-depth provider slot, so this class joins
|
||||||
|
/// <see cref="QueueDepthGaugeCollection"/> to stay serialized against
|
||||||
|
/// <see cref="QueueDepthGaugeTests"/>.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
[Collection(QueueDepthGaugeCollection.Name)]
|
||||||
public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable
|
public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable
|
||||||
{
|
{
|
||||||
private readonly TestLocalDb _localDb;
|
private readonly TestLocalDb _localDb;
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
{
|
{
|
||||||
"$schema": "https://xunit.net/schema/current/xunit.runner.schema.json",
|
"$schema": "https://xunit.net/schema/current/xunit.runner.schema.json",
|
||||||
"parallelizeTestCollections": false
|
"parallelizeTestCollections": true
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user