dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(outbox): resolve the delivery adapter per delivery, not per actor lifetime

Browse Source 
Register row 45. NotificationOutboxActor.ResolveAdapters() cached adapter
instances in an actor-lifetime IServiceScope (_adaptersScope), but both
EmailNotificationDeliveryAdapter and SmsNotificationDeliveryAdapter are
AddScoped and hold a scoped INotificationRepository over one EF DbContext.
MaxParallelDeliveries (default 4) then drove up to four concurrent deliveries
through that single adapter and single context, which central-b logged
continuously on the rig as "InvalidOperationException: A second operation was
started on this context instance before a previous operation completed". The
collided delivery's row is never written and the wasted attempt counts against
its retry budget, so sustained contention could park a legitimate notification.
The per-delivery scope DeliverGatedAsync already opened for the outbox
repository — added for exactly this hazard — did not cover the adapter.

_adaptersCache, _adaptersScope and ResolveAdapters() are deleted. Delivery
adapters are now resolved inside the per-delivery scope via a static
ResolveAdapter(IServiceProvider, NotificationType), preserving DI's last-wins
resolution semantics (the enumeration runs to completion rather than
short-circuiting) and the null-adapter park path. Nothing shared is lost: the
Email adapter builds a fresh SMTP client per attempt from an injected factory
and the SMS adapter draws pooled connections from the singleton
IHttpClientFactory, so no adapter carried state worth preserving across
deliveries. The Central UI's NotificationChannelCatalog is an independent
consumer of the same scoped registrations and is untouched, so the
adapter-gated Type selector still enumerates exactly the deliverable channels.

Tests: Dispatch_ResolvesAdaptersOnce_AcrossMultipleSweeps (which pinned the
superseded caching behaviour) is replaced by
ParallelDeliveries_EachGetTheirOwnAdapterAndDbContext — two parallel deliveries
through a scoped repository whose simulated DbContext throws EF's exact
second-operation exception on overlap, with a rendezvous that guarantees the
overlap window — and Dispatch_ResolvesAdapterPerDelivery_NotOncePerActorLifetime.
Both fail against the pre-fix actor (stash-verified: 1 of 2 rows persisted; 1
distinct adapter instead of 4) and pass after. Parallel-delivery
ordering/throughput semantics are unchanged.

Docs: Component-NotificationOutbox.md gains the scoped-adapter contract under
Delivery Adapters (resolve per delivery; adapters carry no cross-delivery
state; genuinely shared resources belong in a singleton/factory), and the
NotificationOutbox-006 finding is marked SUPERSEDED with the reasoning.

Solution build clean; NotificationOutbox (153), AuditLog (383), integration
NotificationOutbox flow (2) and CentralUI notification (14) suites green.
This commit is contained in:
Joseph Doherty
2026-08-15 05:42:24 -04:00
parent 9cf24a4496
commit 9e7697359e
5 changed files with 320 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs
+246 -75
View File
@@ -110,29 +110,6 @@ public class NotificationOutboxActorDispatchTests : TestKit
.Returns(new[] { config });
}
/// <summary>
/// Drives a sweep to completion by re-telling <see cref="InternalMessages.DispatchTick"/>
/// on every poll iteration until <paramref name="assertion"/> holds. This is the durable
/// barrier for chained-sweep tests: the in-flight guard (<c>_dispatching</c>) is only
/// lowered when a sweep's <see cref="InternalMessages.DispatchComplete"/> round-trips back
/// to the actor — AFTER the side-effect the assertion observes — so a single one-shot tick
/// can race the still-raised guard and be silently dropped (the dispatcher's intended
/// overlap protection), permanently stalling the count. A dropped tick is never retried, so
/// merely widening the wait cannot recover it. Re-telling each poll is idempotent (a tick
/// landing while the guard is up is harmlessly dropped) and the assertion's exact-count
/// strength is preserved unchanged.
/// </summary>
private void DrivePollingTick(IActorRef actor, Action assertion)
{
AwaitAssert(
() =>
{
actor.Tell(InternalMessages.DispatchTick.Instance);
assertion();
},
duration: TimeSpan.FromSeconds(10));
}
[Fact]
public void DispatchTick_ClaimsDueNotifications_AndInvokesAdapter()
{
@@ -477,72 +454,266 @@ public class NotificationOutboxActorDispatchTests : TestKit
"PostStop did not cancel the in-flight delivery promptly.");
}
// ── NotificationOutbox-006: adapter dictionary cached for the actor's lifetime ──
// ── Register row 45: the delivery adapter is resolved PER DELIVERY ────────
//
// Supersedes NotificationOutbox-006 ("resolve the adapter dictionary once for the
// actor's lifetime"). That optimisation cached scoped adapter instances — which hold
// a scoped INotificationRepository over an EF DbContext — in an actor-lifetime DI
// scope, so MaxParallelDeliveries concurrent deliveries all ran on ONE DbContext and
// the rig logged `InvalidOperationException: A second operation was started on this
// context instance before a previous operation completed` continuously. The adapter
// now resolves inside the per-delivery scope that DeliverGatedAsync already opens.
/// <summary>
/// Aggregates what the parallel deliveries observed: how many times two deliveries
/// were simultaneously inside ONE simulated DbContext, and a rendezvous that only
/// completes once every delivery is concurrently in flight.
/// </summary>
private sealed class DeliveryConcurrencyLedger
{
private readonly int _expectedArrivals;
private readonly TaskCompletionSource _allArrived =
new(TaskCreationOptions.RunContinuationsAsynchronously);
private int _arrived;
private int _overlaps;
public DeliveryConcurrencyLedger(int expectedArrivals) => _expectedArrivals = expectedArrivals;
/// <summary>Number of second-operation-on-one-context collisions observed.</summary>
public int Overlaps => Volatile.Read(ref _overlaps);
public void RecordOverlap() => Interlocked.Increment(ref _overlaps);
/// <summary>
/// Holds the caller's simulated DbContext open until every delivery has reached
/// this point — guaranteeing real overlap — or a safety timeout elapses. The
/// timeout matters for the shared-context (pre-fix) run, where the colliding
/// delivery throws instead of arriving and the rendezvous can never complete.
/// </summary>
public async Task ArriveAndWaitAsync(CancellationToken cancellationToken)
{
if (Interlocked.Increment(ref _arrived) >= _expectedArrivals)
{
_allArrived.TrySetResult();
}
await Task.WhenAny(
_allArrived.Task,
Task.Delay(TimeSpan.FromSeconds(2), cancellationToken))
.ConfigureAwait(false);
}
}
/// <summary>
/// Stands in for the EF <c>DbContext</c> behind a scoped
/// <see cref="INotificationRepository"/> — one probe per resolved repository. A second
/// concurrent operation throws the exact <see cref="InvalidOperationException"/> EF
/// raises, so a test that shares one repository across parallel deliveries fails the
/// same way the rig did.
/// </summary>
private sealed class SimulatedDbContextProbe
{
private readonly DeliveryConcurrencyLedger _ledger;
private int _inFlight;
public SimulatedDbContextProbe(DeliveryConcurrencyLedger ledger) => _ledger = ledger;
public async Task EnterAsync(CancellationToken cancellationToken)
{
if (Interlocked.Increment(ref _inFlight) > 1)
{
Interlocked.Decrement(ref _inFlight);
_ledger.RecordOverlap();
throw new InvalidOperationException(
"A second operation was started on this context instance before a "
+ "previous operation completed.");
}
try
{
await _ledger.ArriveAndWaitAsync(cancellationToken);
}
finally
{
Interlocked.Decrement(ref _inFlight);
}
}
}
/// <summary>
/// Adapter shaped like the real Email/SMS adapters: it does its work THROUGH the
/// scoped <see cref="INotificationRepository"/> it was constructed with, so sharing
/// the adapter across deliveries also shares that repository's DbContext.
/// </summary>
private sealed class RepositoryBackedAdapter : INotificationDeliveryAdapter
{
private readonly INotificationRepository _repository;
public RepositoryBackedAdapter(INotificationRepository repository) => _repository = repository;
public NotificationType Type => NotificationType.Email;
public async Task<DeliveryOutcome> DeliverAsync(
Notification notification, CancellationToken cancellationToken = default)
{
// Mirrors the real adapters' first act: resolve the target list off the
// scoped repository (and therefore the scoped DbContext).
await _repository.GetListByNameAsync(notification.ListName, cancellationToken);
return DeliveryOutcome.Success("ops@example.com");
}
}
private static Task<NotificationList?> EnterProbeAsync(
SimulatedDbContextProbe probe, CancellationToken cancellationToken)
{
return EnterProbeCoreAsync(probe, cancellationToken);
static async Task<NotificationList?> EnterProbeCoreAsync(
SimulatedDbContextProbe probe, CancellationToken cancellationToken)
{
await probe.EnterAsync(cancellationToken);
// A null list is enough: the stub adapter ignores the result, and the
// dispatcher never sees it.
return null;
}
}
[Fact]
public void Dispatch_ResolvesAdaptersOnce_AcrossMultipleSweeps()
public void ParallelDeliveries_EachGetTheirOwnAdapterAndDbContext()
{
// NotificationOutbox-006: adapter registration is static per process lifetime,
// so the NotificationType -> adapter lookup must be built ONCE for the actor's
// lifetime, not per dispatch sweep. The cache is paired with an actor-lifetime
// DI scope (see _adaptersScope) so scoped adapter instances are reused safely.
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
// Isolated substitutes for this test — we replace the dispatcher's per-sweep
// INotificationOutboxRepository registration with a private counting factory,
// so we don't mutate the shared _outboxRepository field that other tests in
// this class configure differently.
var outboxRepository = OutboxRepositorySubstitute.Healthy();
// De-race (S11): hand out a fresh due notification for the FIRST THREE claims, then
// an empty batch forever. This caps the deliverable work — and therefore the
// UpdateAsync count — at exactly three, no matter how many dispatch ticks the
// driver below fires. Without this ceiling the exact-count barrier is doubly
// brittle: a single one-shot tick that races the in-flight guard is silently
// dropped (stalling the count below three), while a tick-per-poll driver lets
// queued ticks pile up under CPU starvation and overshoot three. Capping the
// claimable batch lets the driver re-tell ticks idempotently — surplus sweeps
// claim nothing and never call UpdateAsync — so Received(3) is a stable target
// that, once reached, cannot be exceeded.
var dueClaims = 0;
outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(_ => Interlocked.Increment(ref dueClaims) <= 3
? new[] { MakeNotification() }
: Array.Empty<Notification>());
// Register row 45 (fail-before-fix). Two due notifications dispatched with
// MaxParallelDeliveries = 4. Each delivery's adapter reaches its scoped
// INotificationRepository, whose simulated DbContext throws EF's
// second-operation exception if a second delivery enters it concurrently.
//
// Pre-fix: the actor-lifetime _adaptersScope hands BOTH deliveries the same
// adapter -> the same repository -> one probe. The rendezvous holds the first
// delivery inside the context while the second collides, so Overlaps == 1 and
// the collided row is never persisted (DeliverGatedAsync logs and moves on).
// Post-fix: two per-delivery scopes -> two adapters -> two probes, both arrive
// at the rendezvous concurrently, nothing collides, both rows persist.
const int dueCount = 2;
var ledger = new DeliveryConcurrencyLedger(expectedArrivals: dueCount);
var outboxRepository = OutboxRepositorySubstitute.Healthy();
var batch = Enumerable.Range(0, dueCount).Select(_ => MakeNotification()).ToArray();
var claimed = 0;
outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(_ => Interlocked.Increment(ref claimed) == 1 ? batch : Array.Empty<Notification>());
var smtpConfiguration = new SmtpConfiguration("smtp.example.com", "Basic", "noreply@example.com")
{
MaxRetries = 5,
RetryDelay = TimeSpan.FromMinutes(1),
};
// Counting factory: increments each time the DI container resolves an
// INotificationDeliveryAdapter. Pre-fix this would have ticked once per
// sweep; post-fix it ticks exactly once for the actor's lifetime.
var resolutionCount = 0;
var services = new ServiceCollection();
services.AddScoped(_ => outboxRepository);
services.AddScoped(_ => _notificationRepository);
services.AddScoped<INotificationDeliveryAdapter>(_ =>
// Scoped, with its OWN simulated DbContext per resolution — exactly the
// production lifetime of INotificationRepository.
services.AddScoped<INotificationRepository>(_ =>
{
Interlocked.Increment(ref resolutionCount);
return new StubAdapter(() => DeliveryOutcome.Success("ops@example.com"));
var probe = new SimulatedDbContextProbe(ledger);
var repository = Substitute.For<INotificationRepository>();
repository.GetAllSmtpConfigurationsAsync(Arg.Any<CancellationToken>())
.Returns(new[] { smtpConfiguration });
repository.GetListByNameAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
.Returns(call => EnterProbeAsync(probe, call.Arg<CancellationToken>()));
return repository;
});
var sp = services.BuildServiceProvider();
// Scoped adapter holding the scoped repository — the production shape of both
// EmailNotificationDeliveryAdapter and SmsNotificationDeliveryAdapter.
services.AddScoped<INotificationDeliveryAdapter>(sp =>
new RepositoryBackedAdapter(sp.GetRequiredService<INotificationRepository>()));
var serviceProvider = services.BuildServiceProvider();
var actor = Sys.ActorOf(Props.Create(() => new NotificationOutboxActor(
sp,
new NotificationOutboxOptions { DispatchInterval = TimeSpan.FromHours(1) },
serviceProvider,
new NotificationOutboxOptions
{
DispatchInterval = TimeSpan.FromHours(1),
MaxParallelDeliveries = 4,
},
new NoOpCentralAuditWriter(),
NullLogger<NotificationOutboxActor>.Instance)));
// Drive dispatch sweeps until all three claimable notifications have been
// delivered (Received(3).UpdateAsync). DrivePollingTick re-tells the tick on
// every poll iteration; the GetDueAsync cap above makes that idempotent (a
// surplus sweep claims an empty batch and calls UpdateAsync zero times), so
// the count climbs to exactly three and then holds there — Received(3) is a
// stable ceiling rather than a fragile one-shot target. This proves multiple
// dispatch sweeps ran across the actor's lifetime, which is the precondition
// for the resolve-adapters-once assertion that follows.
DrivePollingTick(actor, () => outboxRepository.Received(3).UpdateAsync(
Arg.Any<Notification>(), Arg.Any<CancellationToken>()));
actor.Tell(InternalMessages.DispatchTick.Instance);
// The adapter resolution must have happened EXACTLY ONCE despite the multiple
// dispatch sweeps just driven. Pre-fix this would have been once per sweep.
// This is the real invariant under test (NotificationOutbox-006).
Assert.Equal(1, resolutionCount);
AwaitAssert(
() => outboxRepository.Received(dueCount).UpdateAsync(
Arg.Is<Notification>(n => n.Status == NotificationStatus.Delivered),
Arg.Any<CancellationToken>()),
duration: TimeSpan.FromSeconds(10));
Assert.Equal(0, ledger.Overlaps);
}
[Fact]
public void Dispatch_ResolvesAdapterPerDelivery_NotOncePerActorLifetime()
{
// Register row 45 (isolation, direct form): every delivery must resolve its own
// adapter instance from its own DI scope. Four due rows dispatched in one sweep
// must therefore yield four DISTINCT adapter instances. Pre-fix the actor cached
// a single instance for its whole lifetime and this collapses to 1.
const int dueCount = 4;
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
var outboxRepository = OutboxRepositorySubstitute.Healthy();
var batch = Enumerable.Range(0, dueCount).Select(_ => MakeNotification()).ToArray();
var claimed = 0;
outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(_ => Interlocked.Increment(ref claimed) == 1 ? batch : Array.Empty<Notification>());
var deliveringInstances = new System.Collections.Concurrent.ConcurrentBag<INotificationDeliveryAdapter>();
var services = new ServiceCollection();
services.AddScoped(_ => outboxRepository);
services.AddScoped(_ => _notificationRepository);
services.AddScoped<INotificationDeliveryAdapter>(_ => new RecordingStubAdapter(deliveringInstances));
var serviceProvider = services.BuildServiceProvider();
var actor = Sys.ActorOf(Props.Create(() => new NotificationOutboxActor(
serviceProvider,
new NotificationOutboxOptions
{
DispatchInterval = TimeSpan.FromHours(1),
MaxParallelDeliveries = 4,
},
new NoOpCentralAuditWriter(),
NullLogger<NotificationOutboxActor>.Instance)));
actor.Tell(InternalMessages.DispatchTick.Instance);
AwaitAssert(
() => outboxRepository.Received(dueCount).UpdateAsync(
Arg.Is<Notification>(n => n.Status == NotificationStatus.Delivered),
Arg.Any<CancellationToken>()),
duration: TimeSpan.FromSeconds(10));
// RecordingStubAdapter does not override Equals, so Distinct() is reference
// identity — exactly the "one instance per delivery" question being asked.
Assert.Equal(dueCount, deliveringInstances.Distinct().Count());
}
/// <summary>
/// Stub adapter that records which instance actually served each delivery, so a
/// test can prove one adapter instance per delivery.
/// </summary>
private sealed class RecordingStubAdapter : INotificationDeliveryAdapter
{
private readonly System.Collections.Concurrent.ConcurrentBag<INotificationDeliveryAdapter> _seen;
public RecordingStubAdapter(
System.Collections.Concurrent.ConcurrentBag<INotificationDeliveryAdapter> seen) => _seen = seen;
public NotificationType Type => NotificationType.Email;
public Task<DeliveryOutcome> DeliverAsync(
Notification notification, CancellationToken cancellationToken = default)
{
_seen.Add(this);
return Task.FromResult(DeliveryOutcome.Success("ops@example.com"));
}
}
[Fact]