fix(notification-outbox): clear dispatch in-flight flag on a faulted pass
This commit is contained in:
@@ -139,17 +139,30 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
|||||||
_dispatching = true;
|
_dispatching = true;
|
||||||
var now = DateTimeOffset.UtcNow;
|
var now = DateTimeOffset.UtcNow;
|
||||||
|
|
||||||
// RunDispatchPass swallows its own errors; the completion message is sent for both
|
// RunDispatchPass swallows its own errors, but the failure projection is kept as a
|
||||||
// success and failure so the guard is always cleared.
|
// belt-and-braces guard so even a faulted task still lowers the in-flight guard —
|
||||||
RunDispatchPass(now).PipeTo(Self, success: () => InternalMessages.DispatchComplete.Instance);
|
// otherwise the dispatcher would wedge permanently.
|
||||||
|
RunDispatchPass(now).PipeTo(
|
||||||
|
Self,
|
||||||
|
success: () => InternalMessages.DispatchComplete.Instance,
|
||||||
|
failure: ex =>
|
||||||
|
{
|
||||||
|
_logger.LogError(ex, "Dispatch sweep faulted unexpectedly.");
|
||||||
|
return InternalMessages.DispatchComplete.Instance;
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Runs a single dispatch sweep: claims the due batch, resolves the retry policy, and
|
/// Runs a single dispatch sweep: claims the due batch, resolves the retry policy, and
|
||||||
/// delivers each notification sequentially. Per-notification failures are caught and
|
/// delivers each notification sequentially. Per-notification failures are caught and
|
||||||
/// logged so one bad row never aborts the rest of the batch.
|
/// logged so one bad row never aborts the rest of the batch. The whole body is wrapped
|
||||||
|
/// in a try/catch so the returned task never faults — scope creation, service resolution,
|
||||||
|
/// 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.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private async Task RunDispatchPass(DateTimeOffset now)
|
private async Task RunDispatchPass(DateTimeOffset now)
|
||||||
|
{
|
||||||
|
try
|
||||||
{
|
{
|
||||||
using var scope = _serviceProvider.CreateScope();
|
using var scope = _serviceProvider.CreateScope();
|
||||||
var outboxRepository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
|
var outboxRepository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
|
||||||
@@ -187,6 +200,13 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
// Scope/service resolution or retry-policy resolution faulted; swallow and log so
|
||||||
|
// the returned task completes normally and the in-flight guard is always cleared.
|
||||||
|
_logger.LogError(ex, "Dispatch sweep failed unexpectedly.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Resolves the retry policy from the first SMTP configuration row. When no SMTP
|
/// Resolves the retry policy from the first SMTP configuration row. When no SMTP
|
||||||
|
|||||||
@@ -254,6 +254,28 @@ public class NotificationOutboxActorDispatchTests : TestKit
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void FaultedDispatchPass_ClearsInFlightGuard_SoNextTickStillRuns()
|
||||||
|
{
|
||||||
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
||||||
|
// GetDueAsync throws on every call: the dispatch pass's task could fault if the
|
||||||
|
// failure were not handled, which would leave _dispatching stuck true forever.
|
||||||
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
||||||
|
.Returns<IReadOnlyList<Notification>>(_ => throw new InvalidOperationException("db down"));
|
||||||
|
var actor = CreateActor(new Dictionary<NotificationType, INotificationDeliveryAdapter>());
|
||||||
|
|
||||||
|
// First tick: the pass faults internally but must still clear the in-flight guard.
|
||||||
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
||||||
|
AwaitAssert(() => _outboxRepository.Received(1).GetDueAsync(
|
||||||
|
Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>()));
|
||||||
|
|
||||||
|
// Second tick after the first completes: if the guard had wedged, this would be
|
||||||
|
// dropped and GetDueAsync would still show only one call.
|
||||||
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
||||||
|
AwaitAssert(() => _outboxRepository.Received(2).GetDueAsync(
|
||||||
|
Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>()));
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void OverlappingTicks_WhileDispatchInFlight_DoNotClaimConcurrently()
|
public void OverlappingTicks_WhileDispatchInFlight_DoNotClaimConcurrently()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user