Merge branch 'outbox-delivery-scope' — resolve delivery adapter per delivery, own DbContext per concurrent delivery (register row 45)
This commit is contained in:
Joseph Doherty
@@ -48,7 +48,7 @@ from `Component-NotificationOutbox.md`. No Critical findings; two High, six Medi
|
||||
| 3 | Concurrency & thread safety | Yes | Actor state mutated only on actor thread. Inherited CD-015 race on `InsertIfNotExistsAsync` (NotificationOutbox-005) is the only race; the dispatcher's in-flight guard correctly serializes sweeps. |
|
||||
| 4 | Error handling & resilience | Yes | Outer try/catch on `RunDispatchPass`/`RunPurgePass` keeps the in-flight guard sane; per-notification isolation is correct. CT not threaded into delivery (NotificationOutbox-003). |
|
||||
| 5 | Security | Yes | Inherited OAuth2 empty-user (NotificationOutbox-001) reachable through the adapter. No new credential or trust-boundary issues introduced by the outbox itself. |
|
||||
| 6 | Performance & resource management | Yes | Dispatch interval & batch size are simple polling; `ResolveAdapters` rebuilds the lookup per sweep (NotificationOutbox-006). No leaks. |
|
||||
| 6 | Performance & resource management | Yes | Dispatch interval & batch size are simple polling; `ResolveAdapters` rebuilds the lookup per sweep (NotificationOutbox-006 — its 2026-05-28 caching fix was later **superseded**, see the finding). No leaks. |
|
||||
| 7 | Design-document adherence | Yes | `NotificationOutboxOptions.DispatchBatchSize`, `DeliveredKpiWindow`, `PurgeInterval` are not in the design doc (NotificationOutbox-007). |
|
||||
| 8 | Code organization & conventions | Yes | Options class lives in the component project (correct); DI extension lives in the component (correct); adapter is `scoped`, actor singleton — interaction correctly documented in `ServiceCollectionExtensions`. No issues. |
|
||||
| 9 | Testing coverage | Yes | Solid actor-behaviour coverage. Missing tests for `FallbackMaxRetries` / empty-SMTP-config dispatch path (NotificationOutbox-008). |
|
||||
@@ -312,6 +312,8 @@ stateful intent (timeouts, circuit breakers) cannot accidentally lose state.
|
||||
|
||||
**Resolution (2026-05-28):** Cached the `NotificationType → adapter` dictionary on a new actor field `_adaptersCache`, built lazily on the first dispatch sweep that needs it. The cache is paired with an actor-lifetime `IServiceScope` (`_adaptersScope`, created on first use and disposed in `PostStop`) so the resolved scoped adapter instances live as long as the cache itself — respecting `EmailNotificationDeliveryAdapter`'s scoped lifetime without leaking captive-DbContext references across the actor's full lifetime. `RunDispatchPass` now calls `ResolveAdapters()` (instance method, no args), and the existing "resolved from the per-sweep scope" comment was rewritten to cross-reference the new cache rationale. Adapter set is static per process lifetime (confirmed against the DI registration in `AddNotificationOutbox`), so no invalidation hook is needed. Regression test `Dispatch_ResolvesAdaptersOnce_AcrossMultipleSweeps` registers a counting factory and pins the resolution count at exactly 1 across three end-to-end dispatch sweeps.
|
||||
|
||||
**SUPERSEDED 2026-08-15 (deferred register row 45).** The 2026-05-28 resolution above was **wrong and is reverted** — it was written before the dispatcher gained bounded parallel delivery (`MaxParallelDeliveries`, default 4). An actor-lifetime adapter instance is a *captive scoped service*: the adapter holds a scoped `INotificationRepository` over one EF `DbContext`, so all four concurrent deliveries ran on that single context and central-b logged `InvalidOperationException: A second operation was started on this context instance before a previous operation completed` continuously on the rig. 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. `_adaptersCache`, `_adaptersScope` and `ResolveAdapters()` are deleted; `DeliverGatedAsync` now resolves the adapter for the notification's `Type` from the per-delivery DI scope it already opens (static `ResolveAdapter(IServiceProvider, NotificationType)`, last-wins preserved). The original NotificationOutbox-006 concern — "a future stateful adapter (rate limiter, circuit breaker) silently loses its state" — is answered by the design rule instead of by caching: an adapter must carry no cross-delivery state, and anything genuinely shared belongs in a singleton/factory it depends on (as the SMTP client factory and `IHttpClientFactory` already are). `Dispatch_ResolvesAdaptersOnce_AcrossMultipleSweeps` is replaced by `ParallelDeliveries_EachGetTheirOwnAdapterAndDbContext` (a simulated per-repository DbContext that throws EF's exact second-operation exception; fail-verified against the cached build) and `Dispatch_ResolvesAdapterPerDelivery_NotOncePerActorLifetime`. See [Component-NotificationOutbox.md](../../docs/requirements/Component-NotificationOutbox.md), Delivery Adapters.
|
||||
|
||||
### NotificationOutbox-007 — `NotificationOutboxOptions.DispatchBatchSize`, `DeliveredKpiWindow`, and `PurgeInterval` are not in the design document
|
||||
|
||||
| | |
|
||||
|
||||
@@ -117,7 +117,7 @@ The dispatcher loop runs on a fixed interval. On each tick the `NotificationOutb
|
||||
|
||||
1. Polls the `Notifications` table for **due rows** — `Pending` rows, and `Retrying` rows whose `NextAttemptAt` has passed.
|
||||
2. Resolves the target notification list to its recipients/targets at central, at delivery time.
|
||||
3. Hands the notification to the delivery adapter registered for its `Type`. Deliveries within the sweep run with bounded parallelism (`MaxParallelDeliveries`, default 4), each on its own DI scope/repository, off the actor thread.
|
||||
3. Hands the notification to the delivery adapter registered for its `Type`. Deliveries within the sweep run with bounded parallelism (`MaxParallelDeliveries`, default 4), each on its own DI scope, off the actor thread. **That per-delivery scope owns the channel adapter as well as the outbox repository** — see Delivery Adapters below.
|
||||
4. Applies the result:
|
||||
- **success** → `Delivered`, set `DeliveredAt`, snapshot `ResolvedTargets`.
|
||||
- **transient failure** → `Retrying`, increment `RetryCount`, set `NextAttemptAt`, record `LastError`; once retries are exhausted → `Parked`.
|
||||
@@ -136,7 +136,11 @@ A delivery adapter implementing `INotificationDeliveryAdapter` is registered per
|
||||
- **Email adapter.** The existing SMTP composition/send logic, relocated to the central cluster. Sends a single BCC email to all list recipients.
|
||||
- **SMS adapter (Twilio REST).** Added in T9/T10 (2026-06-19). Sends one Twilio REST request per recipient phone number. Per-recipient results are rolled up: all-accepted → `Success`; any-transient → retry/park; mix of accepted + permanent-bad → `Success` with bad numbers noted in `LastError`; all-permanent → `Permanent` (Park). See [Component-NotificationService.md](Component-NotificationService.md), SMS Delivery Adapter.
|
||||
|
||||
The outbox dispatches by looking up the adapter registered for the notification's `Type`. If no adapter is registered for a given `Type`, the notification is parked with a "no adapter" error — the seam is open for future delivery channels.
|
||||
The outbox dispatches by looking up the adapter registered for the notification's `Type`. If no adapter is registered for a given `Type`, the notification is parked with a "no adapter" error — the seam is open for future delivery channels. Where more than one adapter declares the same `Type`, the last registered wins, mirroring DI's own last-wins resolution.
|
||||
|
||||
**Adapters are scoped services, resolved once per delivery.** Both shipped adapters hold a scoped `INotificationRepository` over the EF `DbContext`, so an adapter instance is bound to exactly one DbContext. The dispatcher therefore resolves the matching adapter *inside the per-delivery DI scope* it already opens for the outbox repository — never once per sweep, and never cached on the singleton actor. Sharing one adapter across the `MaxParallelDeliveries` concurrent deliveries would put several deliveries on one DbContext, which fails with `InvalidOperationException: A second operation was started on this context instance before a previous operation completed`; the affected delivery's outcome is lost, its row is left unwritten, and the wasted attempt eats into the row's retry budget toward `Parked`. An adapter must accordingly carry **no state across deliveries**: anything genuinely shared (connection pools, token caches) belongs in a singleton or factory the adapter depends on — as the Email adapter's SMTP client factory and the SMS adapter's `IHttpClientFactory` already are — not in adapter fields.
|
||||
|
||||
The Central UI's channel catalog is a separate consumer of the same registrations: it enumerates the registered adapters' `Type` values to populate the notification-list Type selector, so the set of offerable channels can never drift from the set the central node can actually deliver.
|
||||
|
||||
Delivery adapters are provided by the Notification Service, which manages notification-list, SMTP, and SMS definitions and supplies the stateless per-type "deliver one notification" implementations. The SMS `SmsConfiguration` (including encrypted Auth Token) travels in Transport bundles alongside SMTP config.
|
||||
|
||||
|
||||
@@ -50,34 +50,6 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
/// </summary>
|
||||
private bool _dispatching;
|
||||
|
||||
/// <summary>
|
||||
/// Cached <see cref="NotificationType"/> → adapter lookup, built
|
||||
/// lazily on the first dispatch sweep and reused for the lifetime of the actor. The
|
||||
/// adapter registration is decided at startup by <c>AddNotificationOutbox</c> (the set is
|
||||
/// keyed by <see cref="NotificationType"/> and is static per process lifetime), so
|
||||
/// rebuilding this dictionary on every sweep was pure allocation waste.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The cache is paired with <see cref="_adaptersScope"/>, an actor-lifetime
|
||||
/// <see cref="IServiceScope"/> created on first use so the cached scoped adapter
|
||||
/// instances and their dependencies live as long as the cache itself. The scope is
|
||||
/// disposed in <see cref="PostStop"/>. The adapters are stateless wrappers that
|
||||
/// resolve their per-call collaborators (e.g. <see cref="INotificationRepository"/>'s
|
||||
/// underlying DbContext) through their own injected dependencies; holding them for
|
||||
/// the actor's lifetime is consistent with the actor's own singleton lifetime on the
|
||||
/// active central node.
|
||||
/// </remarks>
|
||||
private IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter>? _adaptersCache;
|
||||
|
||||
/// <summary>
|
||||
/// Actor-lifetime DI scope that owns the cached
|
||||
/// <see cref="_adaptersCache"/> adapter instances. Created lazily on the first
|
||||
/// dispatch sweep that needs adapters; disposed in <see cref="PostStop"/> so the
|
||||
/// scoped adapter graph (and any disposable dependencies it transitively holds) is
|
||||
/// torn down with the actor.
|
||||
/// </summary>
|
||||
private IServiceScope? _adaptersScope;
|
||||
|
||||
/// <summary>
|
||||
/// Lifecycle-scoped cancellation source, cancelled in <see cref="PostStop"/> so
|
||||
/// any in-flight dispatch sweep — including a long-running SMTP send via the channel
|
||||
@@ -156,13 +128,9 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
_shutdownCts?.Dispose();
|
||||
_shutdownCts = null;
|
||||
|
||||
// Dispose the actor-lifetime adapter scope so the cached
|
||||
// scoped adapter instances and their disposable dependencies are torn down with
|
||||
// the actor (e.g. on a CoordinatedShutdown / failover that stops the singleton).
|
||||
_adaptersScope?.Dispose();
|
||||
_adaptersScope = null;
|
||||
_adaptersCache = null;
|
||||
|
||||
// No adapter state to tear down: delivery adapters are resolved from — and
|
||||
// disposed with — the per-delivery DI scope opened in DeliverGatedAsync, so the
|
||||
// actor holds no scoped service beyond the lifetime of a single delivery.
|
||||
base.PostStop();
|
||||
}
|
||||
|
||||
@@ -278,12 +246,13 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
/// and retry-policy resolution can all throw, and a faulted task would otherwise leave
|
||||
/// the dispatcher's in-flight guard stuck and wedge the loop permanently.
|
||||
///
|
||||
/// The per-sweep DI scope still owns the repository graph
|
||||
/// (<see cref="INotificationOutboxRepository"/> + <see cref="INotificationRepository"/>),
|
||||
/// which is correct because those services back a fresh DbContext per sweep. The
|
||||
/// channel delivery adapters, however, are cached for the actor's lifetime via
|
||||
/// <see cref="ResolveAdapters"/> — see <see cref="_adaptersCache"/> for the
|
||||
/// rationale.
|
||||
/// The per-sweep DI scope owns only the sweep-level repository graph
|
||||
/// (<see cref="INotificationOutboxRepository"/> + <see cref="INotificationRepository"/>)
|
||||
/// used to claim the due batch and resolve the retry policies — both of which run
|
||||
/// before any delivery starts, on the sweep's own DbContext. The channel delivery
|
||||
/// adapters are NOT resolved here: each is resolved inside its own per-delivery scope
|
||||
/// by <see cref="DeliverGatedAsync"/>, so parallel deliveries never share an adapter
|
||||
/// or the DbContext behind it.
|
||||
/// </summary>
|
||||
private async Task RunDispatchPass(DateTimeOffset now, CancellationToken cancellationToken)
|
||||
{
|
||||
@@ -292,7 +261,6 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var outboxRepository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
|
||||
var notificationRepository = scope.ServiceProvider.GetRequiredService<INotificationRepository>();
|
||||
var adapters = ResolveAdapters();
|
||||
|
||||
IReadOnlyList<Notification> due;
|
||||
try
|
||||
@@ -322,18 +290,18 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
// (multi-second SMTP/Twilio round trips), so a strictly sequential sweep caps
|
||||
// throughput at ~1/attempt-latency — an alarm-storm backlog behind one slow
|
||||
// host would drain far too slowly (arch-review 04). A SemaphoreSlim caps the
|
||||
// number of in-flight deliveries; each delivery runs on its OWN DI scope/
|
||||
// repository (the EF DbContext is not thread-safe and rows are distinct per
|
||||
// notification, so concurrent deliveries never contend). The cached adapters
|
||||
// and the lifetime-scoped audit writer are already safe for concurrent use.
|
||||
// MaxParallelDeliveries = 1 preserves the strictly sequential behaviour.
|
||||
// number of in-flight deliveries; each delivery runs on its OWN DI scope —
|
||||
// repository AND channel adapter alike (the EF DbContext is not thread-safe
|
||||
// and rows are distinct per notification, so concurrent deliveries never
|
||||
// contend). The lifetime-scoped audit writer is already safe for concurrent
|
||||
// use. MaxParallelDeliveries = 1 preserves the strictly sequential behaviour.
|
||||
using var deliveryGate = new SemaphoreSlim(
|
||||
_options.ResolvedMaxParallelDeliveries, _options.ResolvedMaxParallelDeliveries);
|
||||
var deliveries = new List<Task>(due.Count);
|
||||
foreach (var notification in due)
|
||||
{
|
||||
deliveries.Add(DeliverGatedAsync(
|
||||
notification, now, policies, adapters, deliveryGate, cancellationToken));
|
||||
notification, now, policies, deliveryGate, cancellationToken));
|
||||
}
|
||||
|
||||
// Each DeliverGatedAsync isolates its own faults, so WhenAll never throws.
|
||||
@@ -355,11 +323,24 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
/// pre-attempt state for the next active sweep) and any other exception is logged — so
|
||||
/// one bad delivery never aborts the rest of the batch and <c>Task.WhenAll</c> never faults.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The per-delivery scope covers the channel delivery adapter as well as the outbox
|
||||
/// repository. Both <see cref="EmailNotificationDeliveryAdapter"/> and
|
||||
/// <see cref="SmsNotificationDeliveryAdapter"/> are registered <em>scoped</em> and hold
|
||||
/// a scoped <see cref="INotificationRepository"/> backed by the EF DbContext, so an
|
||||
/// adapter resolved outside this scope and reused across concurrent deliveries put
|
||||
/// several deliveries on ONE DbContext — which fails with
|
||||
/// <c>InvalidOperationException: A second operation was started on this context
|
||||
/// instance…</c> and burns the row's retries. Resolving the adapter here binds its
|
||||
/// lifetime to exactly one delivery. Neither adapter carries state worth sharing: the
|
||||
/// Email adapter builds a fresh SMTP client per attempt from an injected factory and
|
||||
/// the SMS adapter takes its pooled connections from <c>IHttpClientFactory</c>, which
|
||||
/// is a singleton and shares the handler pool across scopes regardless.
|
||||
/// </remarks>
|
||||
private async Task DeliverGatedAsync(
|
||||
Notification notification,
|
||||
DateTimeOffset now,
|
||||
RetryPolicies policies,
|
||||
IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter> adapters,
|
||||
SemaphoreSlim gate,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
@@ -381,13 +362,14 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
return;
|
||||
}
|
||||
|
||||
// Each delivery owns its DI scope + repository so parallel UpdateAsync calls
|
||||
// never share a DbContext. The adapters map + audit writer are already safe
|
||||
// for concurrent use.
|
||||
// Each delivery owns its DI scope, repository AND channel adapter so parallel
|
||||
// deliveries never share a DbContext. The audit writer is already safe for
|
||||
// concurrent use.
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var outboxRepository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
|
||||
var adapter = ResolveAdapter(scope.ServiceProvider, notification.Type);
|
||||
await DeliverOneAsync(
|
||||
notification, now, policies, outboxRepository, adapters, cancellationToken)
|
||||
notification, now, policies, outboxRepository, adapter, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
@@ -522,37 +504,36 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns the <see cref="NotificationType"/> → adapter lookup, building it lazily on
|
||||
/// the first call and caching it on <see cref="_adaptersCache"/> for the actor's
|
||||
/// lifetime. The last adapter registered for a given type wins, mirroring DI's
|
||||
/// last-wins resolution semantics.
|
||||
/// Resolves the channel delivery adapter registered for <paramref name="type"/> from
|
||||
/// <paramref name="scopedServices"/> — the caller's per-delivery DI scope — or null
|
||||
/// when no adapter handles that type (the caller then parks the notification). The
|
||||
/// LAST adapter registered for a given type wins, mirroring DI's last-wins resolution
|
||||
/// semantics, which is why the enumeration runs to completion rather than short-
|
||||
/// circuiting on the first match.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The lookup used to be rebuilt on every dispatch sweep
|
||||
/// from the per-sweep DI scope. Adapter registration is static per process
|
||||
/// lifetime, so the dict is now built ONCE — on the first sweep that needs it —
|
||||
/// and reused. To respect each adapter's scoped lifetime
|
||||
/// (<see cref="EmailNotificationDeliveryAdapter"/> takes a scoped
|
||||
/// <see cref="INotificationRepository"/>), the cache is paired with
|
||||
/// <see cref="_adaptersScope"/>, an actor-lifetime <see cref="IServiceScope"/> that
|
||||
/// owns the cached adapter instances and is disposed in <see cref="PostStop"/>.
|
||||
/// This deliberately resolves per delivery rather than caching an actor-lifetime
|
||||
/// lookup: the adapters are scoped services holding a scoped
|
||||
/// <see cref="INotificationRepository"/>/DbContext, so a cached instance shared by up
|
||||
/// to <c>MaxParallelDeliveries</c> concurrent deliveries drove a second-operation-on-
|
||||
/// this-context failure. See <see cref="DeliverGatedAsync"/>. Enumerating
|
||||
/// <c>GetServices</c> constructs every registered adapter (two today), which is a
|
||||
/// negligible cost next to the SMTP/HTTPS round trip that follows and keeps the
|
||||
/// last-wins contract identical to the previous lookup.
|
||||
/// </remarks>
|
||||
private IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter> ResolveAdapters()
|
||||
private static INotificationDeliveryAdapter? ResolveAdapter(
|
||||
IServiceProvider scopedServices, NotificationType type)
|
||||
{
|
||||
if (_adaptersCache is not null)
|
||||
INotificationDeliveryAdapter? match = null;
|
||||
foreach (var adapter in scopedServices.GetServices<INotificationDeliveryAdapter>())
|
||||
{
|
||||
return _adaptersCache;
|
||||
if (adapter.Type == type)
|
||||
{
|
||||
match = adapter;
|
||||
}
|
||||
}
|
||||
|
||||
_adaptersScope = _serviceProvider.CreateScope();
|
||||
var adapters = new Dictionary<NotificationType, INotificationDeliveryAdapter>();
|
||||
foreach (var adapter in _adaptersScope.ServiceProvider.GetServices<INotificationDeliveryAdapter>())
|
||||
{
|
||||
adapters[adapter.Type] = adapter;
|
||||
}
|
||||
|
||||
_adaptersCache = adapters;
|
||||
return _adaptersCache;
|
||||
return match;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -585,7 +566,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
DateTimeOffset now,
|
||||
RetryPolicies policies,
|
||||
INotificationOutboxRepository outboxRepository,
|
||||
IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter> adapters,
|
||||
INotificationDeliveryAdapter? adapter,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
// Select the retry policy by delivery channel: SMS honors SmsConfiguration's own
|
||||
@@ -594,7 +575,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
||||
? policies.SmsPolicy
|
||||
: policies.SmtpPolicy;
|
||||
|
||||
if (!adapters.TryGetValue(notification.Type, out var adapter))
|
||||
if (adapter is null)
|
||||
{
|
||||
// Missing-adapter park: from the dispatcher's perspective this is an
|
||||
// attempt that resolved to a terminal park. Emit Attempted then the
|
||||
|
||||
@@ -37,9 +37,11 @@ public static class ServiceCollectionExtensions
|
||||
///
|
||||
/// <see cref="EmailNotificationDeliveryAdapter"/> is registered <em>scoped</em> because it
|
||||
/// takes a scoped <see cref="ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories.INotificationRepository"/>
|
||||
/// directly. The <see cref="NotificationOutboxActor"/> resolves the adapters from a fresh
|
||||
/// scope per dispatch sweep rather than holding them, so no scoped adapter is captured by
|
||||
/// the singleton actor.
|
||||
/// directly. The <see cref="NotificationOutboxActor"/> resolves the matching adapter from
|
||||
/// the fresh DI scope it opens <em>per delivery</em> — not per sweep and never held on the
|
||||
/// actor — so no scoped adapter is captured by the singleton actor and no two of the
|
||||
/// dispatcher's concurrent deliveries (<c>MaxParallelDeliveries</c>) can land on one
|
||||
/// DbContext.
|
||||
///
|
||||
/// The <see cref="SmsNotificationDeliveryAdapter"/> (Twilio-REST) is registered the same
|
||||
/// way alongside the Email adapter, so both resolve from
|
||||
|
||||
+246
-75
@@ -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]
|
||||
|
||||
Reference in New Issue
Block a user