feat(#537): wire the extended progression write path into the async daemon

src/EventTests/Daemon/ExtendedProgressionWriterTests.cs

@@ -0,0 +1,258 @@
+using JasperFx.Events;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Time.Testing;
+using NSubstitute;
+using NSubstitute.ExceptionExtensions;
+using Shouldly;
+
+namespace EventTests.Daemon;
+
+public class ExtendedProgressionWriterTests
+{
+    private readonly IEventStore theStore = Substitute.For<IEventStore>();
+    private readonly RecordingEventDatabase theDatabase = new();
+    private readonly FakeTimeProvider theTime = new();
+    private readonly ExtendedProgressionWriter theWriter;
+
+    public ExtendedProgressionWriterTests()
+    {
+        theStore.ExtendedProgressionEnabled.Returns(true);
+        theWriter = new ExtendedProgressionWriter(theStore, theDatabase, theTime, NullLogger.Instance);
+    }
+
+    private static ShardState transition(ShardAction action, string status, string? pauseReason = null,
+        string shardName = "Counters:All")
+    {
+        return new ShardState(shardName, 42)
+        {
+            Action = action,
+            AgentStatus = status,
+            PauseReason = pauseReason,
+            LastHeartbeat = DateTimeOffset.UtcNow
+        };
+    }
+
+    private static ShardState heartbeat(string shardName = "Counters:All")
+    {
+        return new ShardState(shardName, 42)
+        {
+            Action = ShardAction.Updated,
+            AgentStatus = "Running",
+            LastHeartbeat = DateTimeOffset.UtcNow
+        };
+    }
+
+    [Fact]
+    public async Task writes_status_transitions_through_the_database()
+    {
+        theWriter.OnNext(transition(ShardAction.Started, "Running"));
+        theWriter.OnNext(transition(ShardAction.Paused, "Paused", "boom"));
+        theWriter.OnNext(transition(ShardAction.Stopped, "Stopped"));
+
+        var writes = await theDatabase.WaitForWrites(3);
+
+        writes[0].AgentStatus.ShouldBe("Running");
+        writes[1].AgentStatus.ShouldBe("Paused");
+        writes[1].PauseReason.ShouldBe("boom");
+        writes[2].AgentStatus.ShouldBe("Stopped");
+    }
+
+    [Fact]
+    public async Task no_writes_at_all_when_the_store_has_not_opted_in()
+    {
+        theStore.ExtendedProgressionEnabled.Returns(false);
+
+        theWriter.OnNext(transition(ShardAction.Started, "Running"));
+        theWriter.OnNext(heartbeat());
+        theWriter.OnNext(transition(ShardAction.Stopped, "Stopped"));
+
+        await theDatabase.AssertNoWrites();
+    }
+
+    [Fact]
+    public async Task skips_high_water_mark_and_all_projections_states()
+    {
+        theWriter.OnNext(new ShardState(ShardState.HighWaterMark, 100)
+        {
+            AgentStatus = "Running", LastHeartbeat = DateTimeOffset.UtcNow
+        });
+        theWriter.OnNext(new ShardState(ShardState.AllProjections, 100)
+        {
+            AgentStatus = "Running", LastHeartbeat = DateTimeOffset.UtcNow
+        });
+
+        await theDatabase.AssertNoWrites();
+    }
+
+    [Fact]
+    public async Task skips_plain_progress_publications_with_no_agent_telemetry()
+    {
+        // e.g. the RangeCompleted publication from a rebuild
+        theWriter.OnNext(new ShardState("Counters:All", 42));
+
+        await theDatabase.AssertNoWrites();
+    }
+
+    [Fact]
+    public async Task throttles_heartbeat_writes_per_shard_but_never_transitions()
+    {
+        theWriter.OnNext(heartbeat());
+        theWriter.OnNext(heartbeat()); // same instant, throttled
+        theWriter.OnNext(heartbeat("Others:All")); // different shard, its own budget
+
+        var writes = await theDatabase.WaitForWrites(2);
+        writes.Count(x => x.ShardName == "Counters:All").ShouldBe(1);
+        writes.Count(x => x.ShardName == "Others:All").ShouldBe(1);
+
+        // A transition inside the throttle window still writes immediately
+        theWriter.OnNext(transition(ShardAction.Paused, "Paused", "boom"));
+        await theDatabase.WaitForWrites(3);
+
+        // And once the interval passes, heartbeats flow again
+        theTime.Advance(theWriter.HeartbeatWriteInterval + TimeSpan.FromMilliseconds(1));
+        theWriter.OnNext(heartbeat());
+        var all = await theDatabase.WaitForWrites(4);
+        all.Count(x => x.ShardName == "Counters:All").ShouldBe(3);
+    }
+
+    [Fact]
+    public async Task a_throwing_database_write_is_swallowed_and_does_not_stop_subsequent_writes()
+    {
+        theDatabase.FailNextWrite = true;
+
+        theWriter.OnNext(transition(ShardAction.Started, "Running"));
+        theWriter.OnNext(transition(ShardAction.Stopped, "Stopped"));
+
+        // The first write threw, so only the second lands — and the writer kept going
+        var writes = await theDatabase.WaitForWrites(1);
+        writes[0].AgentStatus.ShouldBe("Stopped");
+    }
+
+    [Fact]
+    public async Task carries_the_assigned_node_number_into_running_on_node()
+    {
+        var state = transition(ShardAction.Started, "Running");
+        state.AssignedNodeNumber = 3;
+
+        theWriter.OnNext(state);
+
+        var writes = await theDatabase.WaitForWrites(1);
+        writes[0].RunningOnNode.ShouldBe(3);
+    }
+
+    [Fact]
+    public async Task does_not_clobber_an_explicit_running_on_node()
+    {
+        var state = transition(ShardAction.Started, "Running");
+        state.AssignedNodeNumber = 3;
+        state.RunningOnNode = 7;
+
+        theWriter.OnNext(state);
+
+        var writes = await theDatabase.WaitForWrites(1);
+        writes[0].RunningOnNode.ShouldBe(7);
+    }
+
+    [Fact]
+    public async Task subscription_agent_lifecycle_flows_through_a_tracker_into_the_database()
+    {
+        // End-to-end through the real tracker + a real SubscriptionAgent: start (Running heartbeat)
+        // then stop (Stopped), proving the daemon-published states reach the store write
+        var tracker = new ShardStateTracker(NullLogger.Instance);
+        tracker.Subscribe(theWriter);
+
+        var agent = new SubscriptionAgent(new ShardName("Counters"), new AsyncOptions(), theTime,
+            Substitute.For<IEventLoader>(), Substitute.For<ISubscriptionExecution>(), tracker,
+            Substitute.For<ISubscriptionMetrics>(), NullLogger.Instance);
+
+        await agent.StartAsync(new SubscriptionExecutionRequest(0, ShardExecutionMode.Continuous,
+            new ErrorHandlingOptions(), new NulloDaemonRuntime()));
+
+        var writes = await theDatabase.WaitForWrites(1);
+        writes[0].ShardName.ShouldBe("Counters:All");
+        writes[0].AgentStatus.ShouldBe("Running");
+        writes[0].LastHeartbeat.ShouldNotBeNull();
+
+        await agent.StopAndDrainAsync(CancellationToken.None);
+
+        writes = await theDatabase.WaitForWrites(2);
+        writes[1].AgentStatus.ShouldBe("Stopped");
+    }
+
+    private class RecordingEventDatabase : IEventDatabase
+    {
+        private readonly List<ShardState> _writes = new();
+        private readonly SemaphoreSlim _signal = new(0);
+
+        public bool FailNextWrite { get; set; }
+
+        public Task WriteExtendedProgressionAsync(ShardState state, CancellationToken token = default)
+        {
+            if (FailNextWrite)
+            {
+                FailNextWrite = false;
+                _signal.Release();
+                throw new InvalidOperationException("The database is grumpy");
+            }
+
+            lock (_writes)
+            {
+                _writes.Add(state);
+            }
+
+            _signal.Release();
+            return Task.CompletedTask;
+        }
+
+        public async Task<IReadOnlyList<ShardState>> WaitForWrites(int count)
+        {
+            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+            while (true)
+            {
+                lock (_writes)
+                {
+                    if (_writes.Count >= count) return _writes.ToList();
+                }
+
+                await _signal.WaitAsync(timeout.Token);
+            }
+        }
+
+        public async Task AssertNoWrites()
+        {
+            // Give the background block a beat to (not) do its thing
+            await Task.Delay(100);
+            lock (_writes)
+            {
+                _writes.ShouldBeEmpty();
+            }
+        }
+
+        // Unused IEventDatabase surface
+        public string Identifier => "recording";
+        public Uri DatabaseUri => new("db://recording");
+        public ShardStateTracker Tracker => null!;
+
+        public Task StoreDeadLetterEventAsync(object storage, DeadLetterEvent deadLetterEvent,
+            CancellationToken token) => Task.CompletedTask;
+
+        public Task EnsureStorageExistsAsync(Type storageType, CancellationToken token) => Task.CompletedTask;
+
+        public Task WaitForNonStaleProjectionDataAsync(TimeSpan timeout) => Task.CompletedTask;
+
+        public Task<long> ProjectionProgressFor(ShardName name, CancellationToken token = default)
+            => Task.FromResult(0L);
+
+        public Task<long?> FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token)
+            => Task.FromResult<long?>(null);
+
+        public string StorageIdentifier => "recording";
+
+        public Task<long> FetchHighestEventSequenceNumber(CancellationToken token) => Task.FromResult(0L);
+
+        public Task<IReadOnlyList<ShardState>> AllProjectionProgress(CancellationToken token = default)
+            => Task.FromResult<IReadOnlyList<ShardState>>([]);
+    }
+}

src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs

@@ -0,0 +1,131 @@
+using ImTools;
+using JasperFx.Blocks;
+using JasperFx.Events.Projections;
+using Microsoft.Extensions.Logging;
+
+namespace JasperFx.Events.Daemon;
+
+/// <summary>
+/// Subscribes to a daemon's <see cref="ShardStateTracker"/> and persists the extended progression
+/// telemetry (heartbeat, agent status, pause reason, running node) that the subscription agents
+/// already compute in process, by driving <see cref="IEventDatabase.WriteExtendedProgressionAsync"/>.
+/// This is the missing write half of extended progression tracking — the schema columns and the read
+/// surface existed, but no daemon path ever persisted them (jasperfx#537, "built and never connected"
+/// per #519).
+///
+/// <para>
+/// Behavior:
+/// <list type="bullet">
+/// <item>Gated on <see cref="IEventStore.ExtendedProgressionEnabled"/>, read live per publication —
+/// nothing is written (or even queued) for stores that have not opted in.</item>
+/// <item>Agent status transitions (<see cref="ShardAction.Started"/>, <see cref="ShardAction.Paused"/>,
+/// <see cref="ShardAction.Stopped"/>) are always written, immediately — a paused/stopped shard is
+/// exactly when the persisted status matters most.</item>
+/// <item><see cref="ShardAction.Updated"/> publications carrying agent telemetry (the ~10s heartbeat
+/// timer ticks and the per-batch commit publications) are throttled to at most one write per
+/// <see cref="HeartbeatWriteInterval"/> per shard, so a fast-committing shard does not turn every
+/// batch into an extra progression-table write.</item>
+/// <item>Writes are best-effort and serialized on a background block: a failed write is logged at
+/// debug and can never fail or stall the shard, and a slow database can never back up the
+/// tracker's publication loop.</item>
+/// </list>
+/// </para>
+/// </summary>
+public sealed class ExtendedProgressionWriter : IObserver<ShardState>, IAsyncDisposable
+{
+    private readonly IEventStore _store;
+    private readonly IEventDatabase _database;
+    private readonly TimeProvider _timeProvider;
+    private readonly ILogger _logger;
+    private readonly Block<ShardState> _block;
+
+    // Only ever touched from the tracker's single publication consumer, so no synchronization needed
+    private ImHashMap<string, DateTimeOffset> _lastWrites = ImHashMap<string, DateTimeOffset>.Empty;
+
+    public ExtendedProgressionWriter(IEventStore store, IEventDatabase database, TimeProvider timeProvider,
+        ILogger logger)
+    {
+        _store = store;
+        _database = database;
+        _timeProvider = timeProvider;
+        _logger = logger;
+
+        _block = new Block<ShardState>(writeAsync);
+
+        // Belt and braces: writeAsync already swallows its own failures, but a failure escaping the
+        // block must still never take anything else down
+        _block.OnError = (state, ex) =>
+            _logger.LogDebug(ex, "Failed to persist extended progression for shard {ShardName}", state.ShardName);
+    }
+
+    /// <summary>
+    /// Minimum spacing between two persisted heartbeat/telemetry writes for the same shard on the
+    /// non-transition (<see cref="ShardAction.Updated"/>) path. Status transitions are never throttled.
+    /// Defaults to 5 seconds so every tick of the agents' 10 second heartbeat timer lands.
+    /// </summary>
+    public TimeSpan HeartbeatWriteInterval { get; set; } = TimeSpan.FromSeconds(5);
+
+    public void OnNext(ShardState value)
+    {
+        if (!_store.ExtendedProgressionEnabled) return;
+
+        // Only real projection/subscription shards have a progression row to decorate
+        if (value.ShardName == ShardState.HighWaterMark || value.ShardName == ShardState.AllProjections) return;
+
+        // Plain progress publications (e.g. rebuild range completions) carry no agent telemetry
+        if (value.AgentStatus == null && value.LastHeartbeat == null) return;
+
+        var isTransition = value.Action is ShardAction.Started or ShardAction.Paused or ShardAction.Stopped;
+        var now = _timeProvider.GetUtcNow();
+
+        if (!isTransition)
+        {
+            if (_lastWrites.TryFind(value.ShardName, out var last) && now - last < HeartbeatWriteInterval)
+            {
+                return;
+            }
+        }
+
+        _lastWrites = _lastWrites.AddOrUpdate(value.ShardName, now);
+
+        // Carry the assigned node through to the persisted running_on_node column when a
+        // distribution layer (e.g. Wolverine-managed subscription distribution) stamped it
+        if (value.RunningOnNode == null && value.AssignedNodeNumber != 0)
+        {
+            value.RunningOnNode = value.AssignedNodeNumber;
+        }
+
+        _block.Post(value);
+    }
+
+    private async Task writeAsync(ShardState state, CancellationToken token)
+    {
+        try
+        {
+            await _database.WriteExtendedProgressionAsync(state, token).ConfigureAwait(false);
+        }
+        catch (Exception e)
+        {
+            // Best-effort telemetry: a failed extended-progression write must NEVER fail or
+            // stall the shard
+            _logger.LogDebug(e, "Failed to persist extended progression for shard {ShardName} on database {Database}",
+                state.ShardName, _database.Identifier);
+        }
+    }
+
+    public void OnCompleted()
+    {
+    }
+
+    public void OnError(Exception error)
+    {
+    }
+
+    public ValueTask DisposeAsync()
+    {
+        // Lets any queued final writes (e.g. the Stopped state published during shutdown) drain
+        // in the background rather than dropping them on the floor
+        _block.Complete();
+        return ValueTask.CompletedTask;
+    }
+}