feat(Core): add TaskCoalescer

Coalesces concurrent runs of the same task factory: at most one run in
flight, at most one queued behind it; requests accepted while the queued
run awaits its turn share its task, so a burst is served by at most two
runs. Extracted from NpgsqlDbLogWatcher NOTIFY coalescing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDuxW7dLs7KPpbTzDtyDUV

Author: alexyakunin

src/ActualLab.Core/Async/TaskCoalescer.cs

@@ -0,0 +1,61 @@
+namespace ActualLab.Async;
+
+/// <summary>
+/// Coalesces concurrent runs of the same task factory: at most one run is in flight and at most
+/// one more is queued behind it. Requests accepted while the queued run awaits its turn share
+/// its task, so any burst of requests is served by at most two runs.
+/// </summary>
+public sealed class TaskCoalescer(Func<Task> taskFactory)
+{
+    private readonly Lock _lock = new();
+    private Task? _activeTask;
+    private Task? _queuedTask;
+
+    public ILogger? Log { get; init; }
+
+    // The task covering every accepted request - e.g. to await on graceful shutdown
+    public Task? LastTask {
+        get {
+            lock (_lock)
+                return _queuedTask ?? _activeTask;
+        }
+    }
+
+    public Task Run()
+    {
+        lock (_lock) {
+            // The queued task must be checked first: the active one may be already completed
+            // while the queued one hasn't promoted itself yet, and starting a new run
+            // in this state would bypass the queued one
+            if (_queuedTask is { } queuedTask)
+                return queuedTask;
+            if (_activeTask is { IsCompleted: false } activeTask)
+                return _queuedTask = QueuedRun(activeTask);
+
+            return _activeTask = taskFactory.Invoke();
+        }
+    }
+
+    // Private methods
+
+    private async Task QueuedRun(Task activeTask)
+    {
+        await activeTask.SilentAwait(false);
+        // The yield forces an asynchronous continuation, which can't enter the promotion block
+        // below before Run (still holding _lock) assigns _queuedTask = this task
+        await Task.Yield();
+        lock (_lock) {
+            if (!ReferenceEquals(_activeTask, activeTask)) {
+                // Must be unreachable; skipping the run is the recovery here
+                Log?.LogCritical(
+                    $"{nameof(TaskCoalescer)}.{nameof(QueuedRun)}: the task it chained itself behind isn't the active one");
+                _queuedTask = null;
+                return;
+            }
+
+            _activeTask = _queuedTask;
+            _queuedTask = null;
+        }
+        await taskFactory.Invoke().ConfigureAwait(false);
+    }
+}

tests/ActualLab.Tests/Async/TaskCoalescerTest.cs

@@ -0,0 +1,107 @@
+namespace ActualLab.Tests.Async;
+
+public class TaskCoalescerTest(ITestOutputHelper @out) : TestBase(@out)
+{
+    [Fact]
+    public async Task SequentialRunsTest()
+    {
+        var runCount = 0;
+        var coalescer = new TaskCoalescer(async () => {
+            Interlocked.Increment(ref runCount);
+            await Task.Yield();
+        });
+
+        await coalescer.Run();
+        await coalescer.Run();
+        await coalescer.Run();
+        runCount.Should().Be(3); // No concurrency = no coalescing
+    }
+
+    [Fact]
+    public async Task CoalescingTest()
+    {
+        var runCount = 0;
+        var gate = TaskCompletionSourceExt.New<Unit>();
+        var coalescer = new TaskCoalescer(async () => {
+            Interlocked.Increment(ref runCount);
+            await gate.Task.ConfigureAwait(false);
+        });
+
+        var t1 = coalescer.Run();
+        var t2 = coalescer.Run();
+        var t3 = coalescer.Run();
+        runCount.Should().Be(1);
+        t2.Should().BeSameAs(t3); // Requests behind an in-flight run share the queued one
+        t2.Should().NotBeSameAs(t1);
+        coalescer.LastTask.Should().BeSameAs(t2);
+
+        gate.SetResult(default);
+        await t3;
+        await t1;
+        runCount.Should().Be(2); // The whole burst is served by two runs
+    }
+
+    [Fact]
+    public async Task ConcurrentBurstTest()
+    {
+        var runCount = 0;
+        var gate = TaskCompletionSourceExt.New<Unit>();
+        var coalescer = new TaskCoalescer(async () => {
+            Interlocked.Increment(ref runCount);
+            await gate.Task.ConfigureAwait(false);
+        });
+
+        var t1 = coalescer.Run();
+        var tasks = Enumerable.Range(0, 100)
+            .AsParallel()
+            .Select(_ => coalescer.Run())
+            .ToArray();
+        runCount.Should().Be(1);
+        tasks.Should().AllSatisfy(t => t.Should().BeSameAs(tasks[0]));
+
+        gate.SetResult(default);
+        await Task.WhenAll(tasks);
+        await t1;
+        runCount.Should().Be(2);
+    }
+
+    [Fact]
+    public async Task ErrorPropagationTest()
+    {
+        var runCount = 0;
+        var mustFail = true;
+        var coalescer = new TaskCoalescer(async () => {
+            Interlocked.Increment(ref runCount);
+            await Task.Yield();
+            if (mustFail)
+                throw new InvalidOperationException("Simulated");
+        });
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => coalescer.Run());
+        mustFail = false;
+        await coalescer.Run(); // A faulted active task must not block new runs
+        runCount.Should().Be(2);
+    }
+
+    [Fact]
+    public async Task QueuedRunErrorPropagationTest()
+    {
+        var runCount = 0;
+        var gate = TaskCompletionSourceExt.New<Unit>();
+        var coalescer = new TaskCoalescer(async () => {
+            var runIndex = Interlocked.Increment(ref runCount);
+            await gate.Task.ConfigureAwait(false);
+            if (runIndex == 2)
+                throw new InvalidOperationException("Simulated");
+        });
+
+        var t1 = coalescer.Run();
+        var t2 = coalescer.Run();
+        gate.SetResult(default);
+        await t1;
+        await Assert.ThrowsAsync<InvalidOperationException>(() => t2);
+
+        await coalescer.Run(); // And a faulted queued task must not block new runs either
+        runCount.Should().Be(3);
+    }
+}