ExtendedProgressionWriter: don't lose the first status transition, and say when a duplicate writer attaches

src/EventTests/Daemon/ExtendedProgressionWriterTests.cs

@@ -1,6 +1,8 @@
+using JasperFx.Core.Reflection;
 using JasperFx.Events;
 using JasperFx.Events.Daemon;
 using JasperFx.Events.Projections;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Time.Testing;
 using NSubstitute;
@@ -376,6 +378,98 @@ await agent.StartAsync(new SubscriptionExecutionRequest(0, ShardExecutionMode.Co
         writes[1].AgentStatus.ShouldBe("Stopped");
     }
 
+    // marten#5167 — a second writer on one database used to announce itself as lock contention: two
+    // writers issuing multi-row UPDATEs over the same rows in plan-dependent order is a deadlock hazard.
+    // Per-row ordered writes made a duplicate writer harmless to correctness, and therefore SILENT: it
+    // just quietly does the same work twice on a second connection. The tracker is shared per database
+    // and building a daemon does not go through a cache, so the lifecycle bug that produces one is real.
+    [Fact]
+    public void a_second_writer_attaching_to_one_tracker_is_reported()
+    {
+        var logger = new CapturingLogger();
+        var tracker = new ShardStateTracker(logger) { DatabaseIdentifier = "tenant_db_7" };
+
+        tracker.Subscribe(theWriter);
+        logger.Warnings.ShouldBeEmpty();
+
+        tracker.Subscribe(new ExtendedProgressionWriter(theStore, theDatabase, theTime, NullLogger.Instance));
+
+        tracker.CountExclusiveObservers(ExtendedProgressionWriter.ExclusiveRole).ShouldBe(2);
+
+        var warning = logger.Warnings.ShouldHaveSingleItem();
+        warning.ShouldContain("tenant_db_7");
+        warning.ShouldContain(ExtendedProgressionWriter.ExclusiveRole);
+
+        // Reported, never refused -- the duplicate is the symptom, and swallowing the subscription
+        // would hide the lifecycle bug rather than surface it
+        tracker.As<IDisposable>().Dispose();
+    }
+
+    // The daemon-restart path: stop unsubscribes, so the next start must not look like a duplicate.
+    // Without this the warning would fire on every ordinary restart and be worth nothing.
+    [Fact]
+    public void a_writer_that_has_unsubscribed_is_not_counted_against_its_replacement()
+    {
+        var logger = new CapturingLogger();
+        var tracker = new ShardStateTracker(logger) { DatabaseIdentifier = "tenant_db_7" };
+
+        var subscription = tracker.Subscribe(theWriter);
+        subscription.Dispose();
+
+        tracker.Subscribe(new ExtendedProgressionWriter(theStore, theDatabase, theTime, NullLogger.Instance));
+
+        tracker.CountExclusiveObservers(ExtendedProgressionWriter.ExclusiveRole).ShouldBe(1);
+        logger.Warnings.ShouldBeEmpty();
+
+        tracker.As<IDisposable>().Dispose();
+    }
+
+    // Ordinary observers are none of this mechanism's business
+    [Fact]
+    public void plain_observers_are_never_reported_as_duplicates()
+    {
+        var logger = new CapturingLogger();
+        var tracker = new ShardStateTracker(logger);
+
+        tracker.Subscribe(new PlainObserver());
+        tracker.Subscribe(new PlainObserver());
+
+        tracker.CountExclusiveObservers(ExtendedProgressionWriter.ExclusiveRole).ShouldBe(0);
+        logger.Warnings.ShouldBeEmpty();
+
+        tracker.As<IDisposable>().Dispose();
+    }
+
+    private class PlainObserver : IObserver<ShardState>
+    {
+        public void OnCompleted() { }
+        public void OnError(Exception error) { }
+        public void OnNext(ShardState value) { }
+    }
+
+    private class CapturingLogger : ILogger
+    {
+        public List<string> Warnings { get; } = new();
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+            Func<TState, Exception?, string> formatter)
+        {
+            if (logLevel == LogLevel.Warning)
+            {
+                Warnings.Add(formatter(state, exception));
+            }
+        }
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
+
+        private class NullScope : IDisposable
+        {
+            public static readonly NullScope Instance = new();
+            public void Dispose() { }
+        }
+    }
+
     private class SingleWriteOnlyDatabase : IEventDatabase
     {
         public List<ShardState> Writes { get; } = new();
@@ -411,6 +505,49 @@ public Task<IReadOnlyList<ShardState>> AllProjectionProgress(CancellationToken t
             => Task.FromResult<IReadOnlyList<ShardState>>([]);
     }
 
+    // jasperfx#631 -- a Started published at sequence 0 has no progression row to decorate (every
+    // store's extended-progression write is update-only), so it lands nowhere. That is the normal
+    // case: SubscriptionAgent.StartAsync publishes Started at floor 0 and the row is not created
+    // until the first batch commits. Before this, with jasperfx#622's periodic beat off, that lost
+    // write was the ONLY one and the telemetry columns stayed NULL for the life of a healthy agent.
+    [Fact]
+    public async Task replays_a_transition_that_had_no_progression_row_to_land_on()
+    {
+        theWriter.OnNext(new ShardState("Counters:All", 0)
+        {
+            Action = ShardAction.Started,
+            AgentStatus = "Running",
+            LastHeartbeat = DateTimeOffset.UtcNow
+        });
+
+        // The first publication carrying a committed sequence proves the row exists now.
+        theWriter.OnNext(heartbeat(sequence: 17));
+
+        var writes = await theDatabase.WaitForWrites(2);
+        writes[0].Sequence.ShouldBe(0);
+        writes[1].AgentStatus.ShouldBe("Running");
+        writes[1].Sequence.ShouldBe(17);
+
+        // ...and only once. Later heartbeats are dropped again, exactly as jasperfx#622 intends.
+        theWriter.OnNext(heartbeat(sequence: 18));
+        theWriter.OnNext(heartbeat(sequence: 19));
+        await Task.Delay(100);
+        (await theDatabase.Batches()).Count.ShouldBe(2);
+    }
+
+    // A shard resuming from a committed floor already HAS a row, so there is nothing to replay and the
+    // periodic-beat-off contract is untouched.
+    [Fact]
+    public async Task does_not_replay_when_the_transition_already_had_a_row()
+    {
+        theWriter.OnNext(transition(ShardAction.Started, "Running"));   // sequence 42
+        theWriter.OnNext(heartbeat(sequence: 43));
+
+        await theDatabase.WaitForWrites(1);
+        await Task.Delay(100);
+        (await theDatabase.Batches()).Count.ShouldBe(1);
+    }
+
     private class RecordingEventDatabase : IEventDatabase
     {
         private readonly List<ShardState[]> _batches = new();

src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs

@@ -43,14 +43,27 @@ namespace JasperFx.Events.Daemon;
 /// persisted status matters most, and these writes are rare, so they keep the "durable across a
 /// crash" story for the data where it means something. The pending heartbeat batch, if periodic
 /// beats are enabled, rides along in the same write.</item>
+/// <item>jasperfx#631: a transition published at sequence 0 — which is every fresh shard's
+/// <see cref="ShardAction.Started"/>, since the agent starts before its first batch commits — has no
+/// progression row to decorate and every store's write is update-only, so it lands nowhere. Such a
+/// shard is remembered and written again on the first publication carrying a committed sequence,
+/// which is proof the row now exists. Without this the telemetry columns stay NULL for the entire
+/// life of a healthy agent once periodic beats are off.</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
+public sealed class ExtendedProgressionWriter : IObserver<ShardState>, IExclusiveTrackerObserver, IAsyncDisposable
 {
+    /// <summary>
+    /// One writer per database is the design. See <see cref="IExclusiveTrackerObserver"/>.
+    /// </summary>
+    public const string ExclusiveRole = "extended progression writer";
+
+    string IExclusiveTrackerObserver.Role => ExclusiveRole;
+
     private readonly IEventStore _store;
     private readonly IEventDatabase _database;
     private readonly TimeProvider _timeProvider;
@@ -59,6 +72,10 @@ public sealed class ExtendedProgressionWriter : IObserver<ShardState>, IAsyncDis
 
     // Only ever touched from the tracker's single publication consumer, so no synchronization needed
     private readonly Dictionary<string, ShardState> _pending = new();
+
+    // jasperfx#631 — shards whose status transition was written while they had no progression row to
+    // decorate, and so must be written again as soon as one exists. See replayUnlandedTransition.
+    private readonly HashSet<string> _unlanded = new();
     private DateTimeOffset _lastFlush = DateTimeOffset.MinValue;
 
     public ExtendedProgressionWriter(IEventStore store, IEventDatabase database, TimeProvider timeProvider,
@@ -111,10 +128,38 @@ public void OnNext(ShardState value)
 
         var isTransition = value.Action is ShardAction.Started or ShardAction.Paused or ShardAction.Stopped;
 
+        // jasperfx#631 -- a transition published before the shard has committed anything has no
+        // progression row to decorate, and every store's write is update-only, so it lands nowhere.
+        // That is the normal case for a fresh shard: SubscriptionAgent.StartAsync publishes Started at
+        // floor 0, and the row is not created until the first batch commits. Until jasperfx#622 the 5s
+        // periodic beat wrote again a moment later and covered for it; with the beat off, Started was
+        // the ONLY write, so agent_status / heartbeat / running_on_node stayed NULL for the whole life
+        // of a healthy agent -- which is precisely when a consumer polling the database (the case those
+        // columns exist for: the publishing node is down, so there is no in-memory state to read)
+        // needs them. Remember the shard and write it again the moment a publication proves the row
+        // exists.
+        //
+        // Lock cost (marten#5167 is the reason this file is careful): the replay is ONE single-row
+        // UPDATE per shard per agent start, one-shot -- not periodic, and nothing like the 5s beat
+        // #622 removed. It rides the same one-row-per-transaction, shard-name-ordered write path, so
+        // it takes one row lock briefly and cannot convoy. The sequence-0 write it compensates for
+        // takes NO lock at all when the row is absent, because it matches no rows. And the replay is
+        // ordered safely by construction: the store commits the batch (which creates the progression
+        // row) BEFORE the agent calls MarkSuccessAsync, so the publication that triggers the replay
+        // always follows the row write rather than contending with it.
+        if (isTransition && value.Sequence <= 0)
+        {
+            _unlanded.Add(value.ShardName);
+        }
+
+        // A publication carrying a committed sequence proves the progression row is there now.
+        var replaysUnlandedTransition = !isTransition && value.Sequence > 0 && _unlanded.Remove(value.ShardName);
+
         // jasperfx#622: with the periodic beat off, a non-transition publication is dropped outright
         // rather than queued -- there is no later flush to carry it, and letting it ride along on the
-        // next transition would write a stale heartbeat nobody reads.
-        if (!isTransition && !PeriodicHeartbeatsEnabled) return;
+        // next transition would write a stale heartbeat nobody reads. The one exception is the replay
+        // above, which is a status write that has not landed yet, not a heartbeat.
+        if (!isTransition && !PeriodicHeartbeatsEnabled && !replaysUnlandedTransition) return;
 
         // Carry the assigned node through to the persisted running_on_node column when a
         // distribution layer (e.g. Wolverine-managed subscription distribution) stamped it
@@ -129,7 +174,7 @@ public void OnNext(ShardState value)
 
         var now = _timeProvider.GetUtcNow();
 
-        if (isTransition || now - _lastFlush >= HeartbeatWriteInterval)
+        if (isTransition || replaysUnlandedTransition || now - _lastFlush >= HeartbeatWriteInterval)
         {
             flush(now);
         }

src/JasperFx.Events/Daemon/IExclusiveTrackerObserver.cs

@@ -0,0 +1,32 @@
+namespace JasperFx.Events.Daemon;
+
+/// <summary>
+/// Marks a <see cref="ShardStateTracker"/> observer that is supposed to be the ONLY one of its
+/// <see cref="Role"/> attached to a given tracker, because it has an external side effect —
+/// typically a database write — that a second instance would simply duplicate.
+///
+/// <para>
+/// The tracker does not enforce this; it logs a warning when a duplicate attaches. Enforcement
+/// would be wrong: the second observer is not itself the bug, it is the symptom of a lifecycle bug
+/// somewhere upstream (a daemon started twice for one database), and refusing the subscription
+/// would hide that rather than surface it.
+/// </para>
+///
+/// <para>
+/// This exists because the failure it detects became SILENT. Duplicate
+/// <see cref="ExtendedProgressionWriter"/>s used to announce themselves as lock contention — two
+/// writers issuing multi-row UPDATEs over the same rows in plan-dependent order is a deadlock
+/// hazard. Since marten#5167 those writes are one row per transaction in shard-name order, which
+/// makes a duplicate writer harmless to correctness and therefore invisible: it just quietly does
+/// the same work twice on a second connection. Making it say so is cheaper than re-deriving it
+/// from a wait graph later.
+/// </para>
+/// </summary>
+internal interface IExclusiveTrackerObserver
+{
+    /// <summary>
+    /// Identifies the kind of observer, so two instances of the same role can be recognized as
+    /// duplicates. Used in the warning text, so it should read as a noun phrase.
+    /// </summary>
+    string Role { get; }
+}

src/JasperFx.Events/Daemon/ShardStateTracker.cs

@@ -114,9 +114,44 @@ internal IDisposable SubscribeAndCaptureCurrentStates(IObserver<ShardState> obse
 
     private void addListener(IObserver<ShardState> observer)
     {
-        if (!_listeners.Contains(observer))
+        if (_listeners.Contains(observer)) return;
+
+        if (observer is IExclusiveTrackerObserver exclusive)
+        {
+            warnIfDuplicate(exclusive);
+        }
+
+        _listeners = _listeners.Add(observer);
+    }
+
+    /// <summary>
+    /// marten#5167: this tracker is shared by every daemon on its database, and building a daemon does
+    /// not go through a cache, so a lifecycle bug can leave two STARTED daemons on one database — each
+    /// arming its own <see cref="ExtendedProgressionWriter"/> against these same rows. That used to
+    /// announce itself as lock contention; per-row ordered writes made it harmless and therefore
+    /// silent, so it is announced deliberately instead. Not an error and not refused — the duplicate
+    /// observer is the symptom, not the bug.
+    /// </summary>
+    private void warnIfDuplicate(IExclusiveTrackerObserver arriving)
+    {
+        // Called under _lock, so the listener list cannot move underneath this
+        var existing = _listeners.OfType<IExclusiveTrackerObserver>().Count(x => x.Role == arriving.Role);
+        if (existing == 0) return;
+
+        _logger.LogWarning(
+            "{Count} {Role}s are now attached to the shard state tracker for database {Database}. The tracker is shared per database, so they will each persist the same rows on their own connection — duplicated work, and a sign that more than one projection daemon has been started for this database. A daemon built only to READ state does not arm one.",
+            existing + 1, arriving.Role, DatabaseIdentifier ?? "(unidentified)");
+    }
+
+    /// <summary>
+    /// How many <see cref="IExclusiveTrackerObserver"/>s of <paramref name="role"/> are attached. More
+    /// than one is the condition <see cref="warnIfDuplicate"/> reports on.
+    /// </summary>
+    internal int CountExclusiveObservers(string role)
+    {
+        lock (_lock)
         {
-            _listeners = _listeners.Add(observer);
+            return _listeners.OfType<IExclusiveTrackerObserver>().Count(x => x.Role == role);
         }
     }
 

src/JasperFx.Events/ProjectionProgressRow.cs

@@ -20,13 +20,28 @@ namespace JasperFx.Events;
 /// enum's Running/Stopped/Paused.
 /// <para>
 /// Nullable because agent state is only persisted where a store both models the column and writes
-/// it. Neither Marten nor Polecat writes it today — both create an <c>agent_status</c> column on the
-/// progression table and read it back, but no daemon path populates it, so it reads NULL. A store
-/// with nothing to report must be able to say so rather than invent a value. See jasperfx#435.
+/// it — a store with nothing to report must be able to say so rather than invent a value
+/// (jasperfx#435). Marten and Polecat do populate it, via
+/// <see cref="JasperFx.Events.Daemon.ExtendedProgressionWriter" />, but only when
+/// <c>EnableExtendedProgressionTracking</c> is on; with it off the column is not even selected and
+/// this reads NULL.
 /// </para>
 /// </param>
 /// <param name="LastHeartbeat">
-/// Timestamp the cell last reported progress; null when the store does not track a heartbeat for it.
+/// Timestamp the agent driving this cell last persisted telemetry; null when the store does not
+/// track a heartbeat for it.
+/// <para>
+/// ⚠️ <b>Not a liveness signal.</b> Since jasperfx#622 the periodic per-shard beat is OFF by default
+/// (<see cref="JasperFx.Events.Daemon.IReadOnlyDaemonSettings.ExtendedProgressionHeartbeatInterval" />),
+/// so <see cref="JasperFx.Events.Daemon.ExtendedProgressionWriter" /> persists this only on a
+/// Started / Paused / Stopped transition. On a healthy long-running agent it therefore freezes at
+/// the timestamp of the last transition and ages without bound — a monitor that thresholds
+/// <c>now - LastHeartbeat</c> will report every shard as dead shortly after startup. Take liveness
+/// from the in-memory <see cref="JasperFx.Events.Projections.ShardState" /> stream instead (an
+/// <c>IObserver&lt;ShardState&gt;</c> on the running daemon, which still beats every 10s), or set a
+/// positive <c>ExtendedProgressionHeartbeatInterval</c> and accept the write cost jasperfx#622 and
+/// marten#5167 removed.
+/// </para>
 /// </param>
 public record ProjectionProgressRow(
     string ProjectionName,

src/JasperFx.Events/Projections/ShardState.cs

@@ -76,7 +76,15 @@ public ShardState(ShardName shardName, long sequence): this(shardName.Identity,
     public DateTimeOffset? LastAdvanced { get; set; }
 
     /// <summary>
-    /// Last heartbeat received from the shard's subscription agent
+    /// Last heartbeat received from the shard's subscription agent.
+    /// <para>
+    /// On a state published to the live tracker this beats every 10 seconds while the agent runs
+    /// (<c>SubscriptionAgent</c>'s heartbeat timer), so it is a usable liveness signal for an
+    /// <c>IObserver&lt;ShardState&gt;</c> subscribed to a running daemon. On a state HYDRATED FROM
+    /// THE PROGRESSION TABLE it is not: since jasperfx#622 the periodic beat is not persisted by
+    /// default, so the stored value freezes at the last Started / Paused / Stopped transition. See
+    /// <see cref="JasperFx.Events.ProjectionProgressRow.LastHeartbeat" />.
+    /// </para>
     /// </summary>
     public DateTimeOffset? LastHeartbeat { get; set; }