fix(host): rate-limit dead-letter warnings (10/min + suppression summary); metric counting unchanged

This commit is contained in:
Joseph Doherty
2026-07-08 16:52:25 -04:00
parent 3ce21734b8
commit dea69842d5
3 changed files with 89 additions and 6 deletions
@@ -8,12 +8,31 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Actors;
/// <summary> /// <summary>
/// Subscribes to Akka.NET dead letter events, logs them, and tracks count /// Subscribes to Akka.NET dead letter events, logs them, and tracks count
/// for health monitoring integration. /// for health monitoring integration.
///
/// Warning logging is rate-limited to <see cref="MaxWarningsPerWindow"/> per
/// <see cref="WarningWindow"/>: a failover/undeploy storm can produce thousands
/// of dead letters, and one Warning per letter is an unbounded log flood
/// (review 01 [Low]). Once the per-window cap is hit, per-letter warnings are
/// suppressed and a single summary of the suppressed count is logged when the
/// window rolls over. The health-metric count is <b>never</b> throttled — every
/// dead letter is counted.
/// </summary> /// </summary>
public class DeadLetterMonitorActor : ReceiveActor public class DeadLetterMonitorActor : ReceiveActor, IWithTimers
{ {
/// <summary>Maximum per-letter Warning logs emitted within one <see cref="WarningWindow"/>.</summary>
public const int MaxWarningsPerWindow = 10;
/// <summary>Rolling window over which <see cref="MaxWarningsPerWindow"/> applies.</summary>
internal static readonly TimeSpan WarningWindow = TimeSpan.FromMinutes(1);
private long _deadLetterCount; private long _deadLetterCount;
private int _warningsThisWindow;
private long _suppressedThisWindow;
private readonly ISiteHealthCollector? _healthCollector; private readonly ISiteHealthCollector? _healthCollector;
/// <summary>Akka timer scheduler (assigned by the runtime via <see cref="IWithTimers"/>).</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary> /// <summary>
/// Initializes the actor and registers dead-letter message handlers. /// Initializes the actor and registers dead-letter message handlers.
/// </summary> /// </summary>
@@ -25,13 +44,39 @@ public class DeadLetterMonitorActor : ReceiveActor
Receive<DeadLetter>(dl => Receive<DeadLetter>(dl =>
{ {
// Metric counting is never throttled.
_deadLetterCount++; _deadLetterCount++;
_healthCollector?.IncrementDeadLetter(); _healthCollector?.IncrementDeadLetter();
logger.LogWarning(
"Dead letter: {MessageType} from {Sender} to {Recipient}", if (_warningsThisWindow < MaxWarningsPerWindow)
dl.Message.GetType().Name, {
dl.Sender, _warningsThisWindow++;
dl.Recipient); logger.LogWarning(
"Dead letter: {MessageType} from {Sender} to {Recipient}",
dl.Message.GetType().Name,
dl.Sender,
dl.Recipient);
if (_warningsThisWindow == MaxWarningsPerWindow)
logger.LogWarning(
"Dead-letter warning limit ({Limit}) reached — suppressing per-letter warnings for {Window}",
MaxWarningsPerWindow, WarningWindow);
}
else
{
_suppressedThisWindow++;
}
});
Receive<WindowTick>(_ =>
{
if (_suppressedThisWindow > 0)
logger.LogWarning(
"Suppressed {Count} dead-letter warnings in the last {Window}",
_suppressedThisWindow, WarningWindow);
_warningsThisWindow = 0;
_suppressedThisWindow = 0;
}); });
Receive<GetDeadLetterCount>(_ => Sender.Tell(new DeadLetterCountResponse(_deadLetterCount))); Receive<GetDeadLetterCount>(_ => Sender.Tell(new DeadLetterCountResponse(_deadLetterCount)));
@@ -41,6 +86,7 @@ public class DeadLetterMonitorActor : ReceiveActor
protected override void PreStart() protected override void PreStart()
{ {
Context.System.EventStream.Subscribe(Self, typeof(DeadLetter)); Context.System.EventStream.Subscribe(Self, typeof(DeadLetter));
Timers.StartPeriodicTimer("dl-window", WindowTick.Instance, WarningWindow);
} }
/// <inheritdoc /> /// <inheritdoc />
@@ -48,6 +94,13 @@ public class DeadLetterMonitorActor : ReceiveActor
{ {
Context.System.EventStream.Unsubscribe(Self, typeof(DeadLetter)); Context.System.EventStream.Unsubscribe(Self, typeof(DeadLetter));
} }
/// <summary>Periodic self-message that rolls the warning-rate-limit window.</summary>
private sealed class WindowTick
{
public static readonly WindowTick Instance = new();
private WindowTick() { }
}
} }
/// <summary> /// <summary>
@@ -1,8 +1,10 @@
using System.Linq;
using Akka.Actor; using Akka.Actor;
using Akka.Event; using Akka.Event;
using Akka.TestKit.Xunit2; using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Host.Actors; using ZB.MOM.WW.ScadaBridge.Host.Actors;
namespace ZB.MOM.WW.ScadaBridge.Host.Tests; namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
@@ -48,6 +50,33 @@ public class DeadLetterMonitorTests : TestKit
}); });
} }
[Fact]
public async Task DeadLetterStorm_LogsAtMostWindowLimit_ButCountsAll()
{
// Review 01 [Low]: a failover/undeploy storm produced one Warning per
// dead letter — unbounded log flood. Health metric counting is untouched.
var logger = Substitute.For<ILogger<DeadLetterMonitorActor>>();
var monitor = Sys.ActorOf(Props.Create(() => new DeadLetterMonitorActor(logger, null)));
// Ensure the actor has started before the storm.
monitor.Tell(GetDeadLetterCount.Instance);
ExpectMsg<DeadLetterCountResponse>();
for (var i = 0; i < 50; i++)
monitor.Tell(new DeadLetter($"m{i}", TestActor, TestActor));
var count = await monitor.Ask<DeadLetterCountResponse>(GetDeadLetterCount.Instance);
Assert.Equal(50, count.Count); // every dead letter still counted
var warnings = logger.ReceivedCalls().Count(c =>
c.GetMethodInfo().Name == "Log"
&& (Microsoft.Extensions.Logging.LogLevel)c.GetArguments()[0]!
== Microsoft.Extensions.Logging.LogLevel.Warning);
// The per-letter cap plus the one "limit reached" notice.
Assert.True(warnings <= DeadLetterMonitorActor.MaxWarningsPerWindow + 1,
$"expected <= {DeadLetterMonitorActor.MaxWarningsPerWindow + 1} warnings, got {warnings}");
}
[Fact] [Fact]
public void DeadLetterMonitor_CountAccumulates() public void DeadLetterMonitor_CountAccumulates()
{ {
@@ -18,6 +18,7 @@
<PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" /> <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" />
<PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" /> <PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" />
<PackageReference Include="Microsoft.NET.Test.Sdk" /> <PackageReference Include="Microsoft.NET.Test.Sdk" />
<PackageReference Include="NSubstitute" />
<PackageReference Include="Serilog" /> <PackageReference Include="Serilog" />
<PackageReference Include="xunit" /> <PackageReference Include="xunit" />
<PackageReference Include="xunit.runner.visualstudio" /> <PackageReference Include="xunit.runner.visualstudio" />