fix: TimedOutboxSweeper starvation and publish-confirm floods (#4560)

bugfixes/0052-timed-outbox-sweeper-starvation/.issue-number

@@ -0,0 +1 @@
+4560

bugfixes/0052-timed-outbox-sweeper-starvation/diagnostic/README.md

@@ -0,0 +1,49 @@
+# Diagnostics for #4560
+
+These scratch console apps produced the evidence in `../bugfix.md`. Use them as the starting point for the
+`samples/` harness. Both target net9.0 and reference the repo's projects by relative path.
+
+## Builds against a fixed commit
+
+Both apps were written against the code at `2890f8610`, before the fix.
+
+- **Both still build against the fixed code.** The fix left the `TimedOutboxSweeper` constructor
+  source compatible.
+- **The results change.** Against the fixed code, `blocked` mode with `DOTNET_PROCESSOR_COUNT=1`
+  gives the real sweeper 10 ticks with a 1.01 s maximum gap. Before the fix it gave 7 ticks and a
+  3.03 s gap. The `PeriodicTimer` async loop still slips (2.01 s). That makes the apps useful as
+  before/after demonstrations.
+- **`TimerInterval` must be at least 1.** Values below 1 now throw `ConfigurationException`.
+
+## `starve/`: when does the sweeper's tick slip?
+
+`starve/` runs the real `TimedOutboxSweeper` with `TimerInterval = 1`. A recording `IDistributedLock`
+times each tick. Three loads are compared:
+- none;
+- 200 blocked pool items;
+- a flood of 20k `Task.Run` per second from a non-pool thread.
+
+Each load also runs a dedicated-thread loop and a `PeriodicTimer` async loop for comparison.
+
+```bash
+DOTNET_PROCESSOR_COUNT=1 dotnet run -c Release -- blocked 10   # modes: none | blocked | flood
+```
+
+## `pileup/`: the full pipeline under load
+
+`pileup/` runs the real `CommandProcessor`, `OutboxProducerMediator`, `InMemoryOutbox` and
+`TimedOutboxSweeper`. A fake Kafka-like producer delivers on a non-pool thread and calls `Task.Run` for
+each confirm.
+
+```bash
+# args: sync|async  workers  seconds
+DOTNET_PROCESSOR_COUNT=2 dotnet run -c Release -- async 64 12
+```
+
+| Variable | Effect |
+| --- | --- |
+| `DI_DEFAULT=1` | Outstanding-check interval of zero, as DI sets it |
+| `NO_CHECK=1` | Effectively turns the outstanding check off (1 h interval) |
+
+- **Posters yield.** The posting loops `await Task.Yield()`, which gives fair load. Remove the yield
+  to saturate the pool.

bugfixes/0052-timed-outbox-sweeper-starvation/diagnostic/pileup/Program.cs

@@ -0,0 +1,165 @@
+using Paramore.Brighter.Extensions;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Net.Mime;
+using System.Text.Json;
+using System.Transactions;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Paramore.Brighter;
+using Paramore.Brighter.JsonConverters;
+using Paramore.Brighter.Logging;
+using Paramore.Brighter.Observability;
+using Paramore.Brighter.Outbox.Hosting;
+using Polly.Registry;
+
+// args: sync|async workers seconds
+var mode = args.Length > 0 ? args[0] : "async";
+var workers = args.Length > 1 ? int.Parse(args[1]) : 64;
+var seconds = args.Length > 2 ? int.Parse(args[2]) : 15;
+
+var sw = Stopwatch.StartNew();
+var counter = new Counter(sw);
+ApplicationLogging.LoggerFactory = LoggerFactory.Create(b => b.SetMinimumLevel(LogLevel.Debug).AddProvider(counter));
+
+var routingKey = new RoutingKey("MyCommand");
+var producer = new FakeKafkaProducer(new Publication { Topic = routingKey, RequestType = typeof(MyCommand) });
+var mapperRegistry = new MessageMapperRegistry(new SimpleMessageMapperFactory(_ => new MyCommandMessageMapper()), new SimpleMessageMapperFactoryAsync(_ => new MyCommandMessageMapper()));
+mapperRegistry.Register<MyCommand, MyCommandMessageMapper>(); mapperRegistry.RegisterAsync<MyCommand, MyCommandMessageMapper>();
+var producerRegistry = new ProducerRegistry(new Dictionary<RoutingKey, IAmAMessageProducer> { { routingKey, producer } });
+var resilience = new ResiliencePipelineRegistry<string>().AddBrighterDefault();
+var outbox = new InMemoryOutbox(TimeProvider.System) { EntryLimit = 1_000_000 };
+var mediator = new OutboxProducerMediator<Message, CommittableTransaction>(
+    producerRegistry, resilience, mapperRegistry,
+    new EmptyMessageTransformerFactory(), new EmptyMessageTransformerFactoryAsync(),
+    new BrighterTracer(TimeProvider.System), new FindPublicationByPublicationTopicOrRequestType(), outbox,
+    maxOutStandingCheckInterval: Environment.GetEnvironmentVariable("NO_CHECK") == "1" ? TimeSpan.FromHours(1) : Environment.GetEnvironmentVariable("DI_DEFAULT") == "1" ? TimeSpan.Zero : null);
+Console.WriteLine($"checkInterval={(Environment.GetEnvironmentVariable("DI_DEFAULT") == "1" ? "Zero (DI default)" : "1s (ctor default)")}");
+var cp = new CommandProcessor(new InMemoryRequestContextFactory(), new DefaultPolicy(), resilience, mediator,
+    requestSchedulerFactory: new InMemorySchedulerFactory());
+
+var services = new ServiceCollection();
+services.AddSingleton<IAmAnOutboxProducerMediator>(mediator);
+var sp = services.BuildServiceProvider();
+var ticks = new ConcurrentQueue<double>();
+var sweeper = new TimedOutboxSweeper(sp.GetRequiredService<IServiceScopeFactory>(), new RecordingLock(ticks, sw, new InMemoryLock()),
+    new TimedOutboxSweeperOptions { TimerInterval = 1, MinimumMessageAge = TimeSpan.FromSeconds(1) });
+
+ThreadPool.GetMinThreads(out var minW, out _);
+Console.WriteLine($"mode={mode} workers={workers} procs={Environment.ProcessorCount} minWorkers={minW}");
+
+await sweeper.StartAsync(CancellationToken.None);
+
+var stop = new CancellationTokenSource();
+long posts = 0; int reported = 0;
+var loops = Enumerable.Range(0, workers).Select(_ => Task.Run(async () =>
+{
+    while (!stop.IsCancellationRequested)
+    {
+        try {
+        if (mode == "sync") cp.Post(new MyCommand());
+        else await cp.PostAsync(new MyCommand());
+        } catch (Exception e) { if (Interlocked.Exchange(ref reported, 1) == 0) Console.WriteLine("POST FAILED: " + e); throw; }
+        Interlocked.Increment(ref posts);
+        await Task.Yield();
+    }
+})).ToArray();
+
+// sampler on dedicated thread
+var samples = new List<string>();
+var sampler = new Thread(() =>
+{
+    for (int s = 1; s <= seconds; s++)
+    {
+        Thread.Sleep(1000);
+        var outstanding = outbox.Requests_Outstanding();
+        samples.Add($"t={s,2}s posts={Interlocked.Read(ref posts),8} undispatched={outstanding,7} entries={outbox.EntryCount,8} pool={ThreadPool.ThreadCount,3} queue={ThreadPool.PendingWorkItemCount,8} checksQueued={counter.Queued,6} checksStarted={counter.Started,6} backlog={counter.Queued - counter.Started,6} sweepTicks={ticks.Count,3} sweeps={counter.Sweeps,3} resends={producer.Resends,6}");
+    }
+}) { IsBackground = true };
+sampler.Start();
+sampler.Join();
+stop.Cancel();
+foreach (var l in samples) Console.WriteLine(l);
+var a = ticks.ToArray();
+var gaps = a.Zip(a.Skip(1), (x, y) => y - x).DefaultIfEmpty(double.NaN).Max();
+Console.WriteLine($"sweeper ticks={a.Length} maxGap={gaps:F2}s  sweepsCompleted={counter.Sweeps} maxSweep={counter.MaxSweep:F2}s maxCheckBacklog={counter.MaxBacklog}");
+Environment.Exit(0);
+
+static class OutboxExt
+{
+    public static int Requests_Outstanding(this InMemoryOutbox o) =>
+        o.GetOutstandingMessageCount(TimeSpan.Zero, null);
+}
+
+class Counter(Stopwatch sw) : ILoggerProvider, ILogger
+{
+    public long Queued, Started, Sweeps; public long MaxBacklog; public double MaxSweep;
+    double _sweepStart;
+    public ILogger CreateLogger(string categoryName) => this;
+    public void Dispose() { }
+    public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+    public bool IsEnabled(LogLevel logLevel) => true;
+    public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+    {
+        var fmt = (state as IReadOnlyList<KeyValuePair<string, object?>>)?.LastOrDefault(kv => kv.Key == "{OriginalFormat}").Value as string ?? "";
+        if (fmt.StartsWith("Running outstanding message check")) { var q = Interlocked.Increment(ref Queued); var b = q - Interlocked.Read(ref Started); if (b > MaxBacklog) MaxBacklog = b; }
+        else if (fmt.StartsWith("Begin count of outstanding")) Interlocked.Increment(ref Started);
+        else if (fmt.StartsWith("Outbox Sweeper looking")) _sweepStart = sw.Elapsed.TotalSeconds;
+        else if (fmt.StartsWith("Outbox Sweeper sleeping") && _sweepStart > 0) { Interlocked.Increment(ref Sweeps); var d = sw.Elapsed.TotalSeconds - _sweepStart; if (d > MaxSweep) MaxSweep = d; _sweepStart = 0; }
+        else if (logLevel >= LogLevel.Warning && !fmt.StartsWith("Outbox Sweeper is still running")) Console.WriteLine($"[{logLevel}] {formatter(state, exception)} {exception?.GetType().Name}");
+    }
+}
+
+class RecordingLock(ConcurrentQueue<double> q, Stopwatch sw, IDistributedLock inner) : IDistributedLock
+{
+    public Task<string?> ObtainLockAsync(string r, CancellationToken ct) { q.Enqueue(sw.Elapsed.TotalSeconds); return inner.ObtainLockAsync(r, ct); }
+    public Task ReleaseLockAsync(string r, string id, CancellationToken ct) => inner.ReleaseLockAsync(r, id, ct);
+}
+
+// Mimics KafkaMessageProducer: Produce enqueues to librdkafka; delivery reports arrive on a non-pool
+// thread; Brighter raises each confirmation with Task.Run (KafkaMessageProducer.cs:456).
+class FakeKafkaProducer : IAmAMessageProducerSync, IAmAMessageProducerAsync, ISupportPublishConfirmation, ISupportPublishConfirmationAsync
+{
+    private readonly BlockingCollection<Id> _inFlight = new();
+    private readonly ConcurrentDictionary<string, byte> _seen = new();
+    public long Resends;
+    private Func<PublishConfirmationResult, Task>? _async;
+    public FakeKafkaProducer(Publication p)
+    {
+        Publication = p;
+        new Thread(() =>
+        {
+            foreach (var id in _inFlight.GetConsumingEnumerable())
+            {
+                var r = new PublishConfirmationResult(true, id, Publication.Topic!, null);
+                Task.Run(async () => { OnMessagePublished?.Invoke(r); if (_async != null) await _async(r); });
+            }
+        }) { IsBackground = true, Name = "librdkafka-delivery" }.Start();
+    }
+    public bool UseAsyncPublishConfirmation => true;
+    public Publication Publication { get; }
+    public Activity? Span { get; set; }
+    public IAmAMessageScheduler? Scheduler { get; set; }
+    public event Action<PublishConfirmationResult>? OnMessagePublished;
+    public event Func<PublishConfirmationResult, Task> OnMessagePublishedAsync { add => _async += value; remove => _async -= value; }
+    public void Send(Message m) { if (!_seen.TryAdd(m.Id.Value, 0)) Interlocked.Increment(ref Resends); _inFlight.Add(m.Id); }
+    public void SendWithDelay(Message m, TimeSpan? d) => Send(m);
+    public Task SendAsync(Message m, CancellationToken ct = default) { Send(m); return Task.CompletedTask; }
+    public Task SendWithDelayAsync(Message m, TimeSpan? d, CancellationToken ct = default) => SendAsync(m, ct);
+    public void Dispose() { }
+    public ValueTask DisposeAsync() => ValueTask.CompletedTask;
+}
+
+public class MyCommand() : Command(Id.Random()) { public string? Value { get; set; } = "hello"; }
+
+class MyCommandMessageMapper : IAmAMessageMapper<MyCommand>, IAmAMessageMapperAsync<MyCommand>
+{
+    public IRequestContext? Context { get; set; }
+    public Message MapToMessage(MyCommand request, Publication publication) =>
+        new(new MessageHeader(request.Id, publication.Topic!, MessageType.MT_COMMAND, timeStamp: DateTimeOffset.UtcNow,
+                contentType: new ContentType(MediaTypeNames.Application.Json)),
+            new MessageBody(JsonSerializer.Serialize(request, JsonSerialisationOptions.Options)));
+    public MyCommand MapToRequest(Message message) => throw new NotImplementedException();
+    public Task<Message> MapToMessageAsync(MyCommand r, Publication p, CancellationToken ct = default) => Task.FromResult(MapToMessage(r, p));
+    public Task<MyCommand> MapToRequestAsync(Message m, CancellationToken ct = default) => throw new NotImplementedException();
+}

bugfixes/0052-timed-outbox-sweeper-starvation/diagnostic/pileup/pileup.csproj

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net9.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="../../../../src/Paramore.Brighter.Outbox.Hosting/Paramore.Brighter.Outbox.Hosting.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Logging" />
+  </ItemGroup>
+</Project>

bugfixes/0052-timed-outbox-sweeper-starvation/diagnostic/starve/Program.cs

@@ -0,0 +1,80 @@
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using Paramore.Brighter;
+using Paramore.Brighter.Outbox.Hosting;
+
+// args: mode(none|blocked|flood) seconds
+var mode = args.Length > 0 ? args[0] : "none";
+var seconds = args.Length > 1 ? int.Parse(args[1]) : 10;
+var sw = Stopwatch.StartNew();
+ThreadPool.GetMinThreads(out var minW, out _);
+Console.WriteLine($"mode={mode} procs={Environment.ProcessorCount} minWorkers={minW}");
+
+var timerTicks = new ConcurrentQueue<double>();
+var threadTicks = new ConcurrentQueue<double>();
+var asyncLoopTicks = new ConcurrentQueue<double>();
+
+var stop = new CancellationTokenSource();
+var gate = new ManualResetEventSlim(false);
+
+// load
+Thread? flooder = null;
+if (mode == "blocked")
+{
+    for (int i = 0; i < 200; i++) Task.Run(() => gate.Wait());
+}
+else if (mode == "flood")
+{
+    // simulate Confluent's delivery thread: a non-pool thread doing Task.Run per delivery report
+    flooder = new Thread(() =>
+    {
+        var perMs = 20; // 20k reports/sec
+        while (!stop.IsCancellationRequested)
+        {
+            for (int i = 0; i < perMs; i++)
+                Task.Run(() => { var s = Stopwatch.StartNew(); while (s.Elapsed.TotalMilliseconds < 0.2) { } });
+            Thread.Sleep(1);
+        }
+    }) { IsBackground = true };
+    flooder.Start();
+}
+
+// 1) the real TimedOutboxSweeper, ticks recorded by the lock
+var sweeper = new TimedOutboxSweeper(null!, new RecordingLock(timerTicks, sw), new TimedOutboxSweeperOptions { TimerInterval = 1 });
+await sweeper.StartAsync(CancellationToken.None);
+
+// 2) control: dedicated thread
+var t = new Thread(() => { while (!stop.IsCancellationRequested) { threadTicks.Enqueue(sw.Elapsed.TotalSeconds); Thread.Sleep(1000); } }) { IsBackground = true };
+t.Start();
+
+// 3) BackgroundService-style async loop with PeriodicTimer on a LongRunning task
+_ = Task.Factory.StartNew(async () =>
+{
+    using var pt = new PeriodicTimer(TimeSpan.FromSeconds(1));
+    asyncLoopTicks.Enqueue(sw.Elapsed.TotalSeconds);
+    while (await pt.WaitForNextTickAsync()) asyncLoopTicks.Enqueue(sw.Elapsed.TotalSeconds);
+}, TaskCreationOptions.LongRunning);
+
+Thread.Sleep(seconds * 1000);
+stop.Cancel();
+gate.Set();
+ThreadPool.GetAvailableThreads(out var avail, out _);
+Console.WriteLine($"threadCount={ThreadPool.ThreadCount} pending={ThreadPool.PendingWorkItemCount}");
+Report("TimedOutboxSweeper", timerTicks, seconds);
+Report("dedicated thread", threadTicks, seconds);
+Report("PeriodicTimer async loop", asyncLoopTicks, seconds);
+Environment.Exit(0);
+
+static void Report(string name, ConcurrentQueue<double> q, int seconds)
+{
+    var a = q.Where(x => x <= seconds).ToArray();
+    var first = a.Length > 0 ? a[0] : double.NaN;
+    var gaps = a.Zip(a.Skip(1), (x, y) => y - x).DefaultIfEmpty(double.NaN).Max();
+    Console.WriteLine($"  {name,-38} ticks={a.Length,3} first={first,6:F2}s maxGap={gaps,6:F2}s");
+}
+
+class RecordingLock(ConcurrentQueue<double> q, Stopwatch sw) : IDistributedLock
+{
+    public Task<string?> ObtainLockAsync(string resource, CancellationToken ct) { q.Enqueue(sw.Elapsed.TotalSeconds); return Task.FromResult<string?>(null); }
+    public Task ReleaseLockAsync(string resource, string lockId, CancellationToken ct) => Task.CompletedTask;
+}

bugfixes/0052-timed-outbox-sweeper-starvation/diagnostic/starve/starve.csproj

@@ -0,0 +1,11 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net9.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="../../../../src/Paramore.Brighter.Outbox.Hosting/Paramore.Brighter.Outbox.Hosting.csproj" />
+  </ItemGroup>
+</Project>