fix(outbox): resolve the delivery adapter per delivery, not per actor lifetime
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
@@ -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