From c1081d80de995a376b3ffb70cf343be7825a2ea2 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 4 Aug 2026 11:28:02 -0500 Subject: [PATCH 1/2] feat: a duplicate ExtendedProgressionWriter says so A second ExtendedProgressionWriter 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. jasperfx#630 made those writes one row per transaction in shard-name order, which makes a duplicate writer harmless to correctness and therefore SILENT: it just quietly does the same work twice on a second connection. That is worth knowing about, because the condition it indicates is real. The tracker is shared per 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 writer. jasperfx#621 gated arming so a daemon built only to READ state never subscribes one, which removed the common cause -- it did not remove the possibility. ShardStateTracker now logs a warning when an IExclusiveTrackerObserver attaches to a tracker that already has one of its role. Deliberately reported and not refused: the duplicate observer is the symptom, not the bug, and swallowing the subscription would hide the lifecycle bug rather than surface it. Unsubscribing removes the listener, so an ordinary daemon restart does not trip it -- pinned by its own test, since a warning that fires on every restart would be worth nothing. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017CTtw2kVRSZKp1p5RTxgAy --- .../Daemon/ExtendedProgressionWriterTests.cs | 94 +++++++++++++++++++ .../Daemon/ExtendedProgressionWriter.cs | 9 +- .../Daemon/IExclusiveTrackerObserver.cs | 32 +++++++ .../Daemon/ShardStateTracker.cs | 39 +++++++- 4 files changed, 171 insertions(+), 3 deletions(-) create mode 100644 src/JasperFx.Events/Daemon/IExclusiveTrackerObserver.cs diff --git a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs b/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs index 76a8978..8b68fc3 100644 --- a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs +++ b/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().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().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().Dispose(); + } + + private class PlainObserver : IObserver + { + public void OnCompleted() { } + public void OnError(Exception error) { } + public void OnNext(ShardState value) { } + } + + private class CapturingLogger : ILogger + { + public List Warnings { get; } = new(); + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + if (logLevel == LogLevel.Warning) + { + Warnings.Add(formatter(state, exception)); + } + } + + public bool IsEnabled(LogLevel logLevel) => true; + public IDisposable BeginScope(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 Writes { get; } = new(); diff --git a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs index 6087367..54277ed 100644 --- a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs +++ b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs @@ -49,8 +49,15 @@ namespace JasperFx.Events.Daemon; /// /// /// -public sealed class ExtendedProgressionWriter : IObserver, IAsyncDisposable +public sealed class ExtendedProgressionWriter : IObserver, IExclusiveTrackerObserver, IAsyncDisposable { + /// + /// One writer per database is the design. See . + /// + public const string ExclusiveRole = "extended progression writer"; + + string IExclusiveTrackerObserver.Role => ExclusiveRole; + private readonly IEventStore _store; private readonly IEventDatabase _database; private readonly TimeProvider _timeProvider; diff --git a/src/JasperFx.Events/Daemon/IExclusiveTrackerObserver.cs b/src/JasperFx.Events/Daemon/IExclusiveTrackerObserver.cs new file mode 100644 index 0000000..a49f5c8 --- /dev/null +++ b/src/JasperFx.Events/Daemon/IExclusiveTrackerObserver.cs @@ -0,0 +1,32 @@ +namespace JasperFx.Events.Daemon; + +/// +/// Marks a observer that is supposed to be the ONLY one of its +/// attached to a given tracker, because it has an external side effect — +/// typically a database write — that a second instance would simply duplicate. +/// +/// +/// 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. +/// +/// +/// +/// This exists because the failure it detects became SILENT. Duplicate +/// 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. +/// +/// +internal interface IExclusiveTrackerObserver +{ + /// + /// 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. + /// + string Role { get; } +} diff --git a/src/JasperFx.Events/Daemon/ShardStateTracker.cs b/src/JasperFx.Events/Daemon/ShardStateTracker.cs index 00a84de..93fd867 100644 --- a/src/JasperFx.Events/Daemon/ShardStateTracker.cs +++ b/src/JasperFx.Events/Daemon/ShardStateTracker.cs @@ -114,9 +114,44 @@ internal IDisposable SubscribeAndCaptureCurrentStates(IObserver obse private void addListener(IObserver observer) { - if (!_listeners.Contains(observer)) + if (_listeners.Contains(observer)) return; + + if (observer is IExclusiveTrackerObserver exclusive) + { + warnIfDuplicate(exclusive); + } + + _listeners = _listeners.Add(observer); + } + + /// + /// 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 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. + /// + private void warnIfDuplicate(IExclusiveTrackerObserver arriving) + { + // Called under _lock, so the listener list cannot move underneath this + var existing = _listeners.OfType().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)"); + } + + /// + /// How many s of are attached. More + /// than one is the condition reports on. + /// + internal int CountExclusiveObservers(string role) + { + lock (_lock) { - _listeners = _listeners.Add(observer); + return _listeners.OfType().Count(x => x.Role == role); } } From 138299d35cb407383aa1bc492fece5c8e9171d29 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 4 Aug 2026 12:26:52 -0500 Subject: [PATCH 2/2] fix: a status transition with no progression row to land on is written again (#631) Every store's extended-progression write is update-only -- "shards without a progression row yet are skipped silently" is the documented contract, and the row is not created until a shard's first batch commits. SubscriptionAgent.StartAsync publishes its Started state at floor 0, before that, so on a fresh shard the write matches zero rows and lands nowhere. That was harmless while the 5s periodic beat existed: it wrote again a moment later, once the row was there. #622 turned the beat off by default, which made Started the ONLY telemetry write -- so agent_status, heartbeat and running_on_node stayed NULL for the entire life of a healthy agent. Precisely the case those columns exist for: a consumer polling the database because the publishing node is down and there is no in-memory ShardState to read. Reproduced end to end against PostgreSQL through Marten: a Balanced two-node cluster ran its projection to sequence 30 and reported running_on_node=NULL, agent_status=NULL, heartbeat=NULL. Bouncing the agent so the same Started transition lands on a row that now exists produced agent_status=Running with a heartbeat, which isolates the cause to row absence rather than the transition write itself. A transition published at sequence 0 now marks the shard, and the first later publication carrying a committed sequence -- proof the row exists -- writes it once and clears the mark. Later heartbeats are dropped again exactly as #622 intends. Lock cost, since marten#5167 is why this file is careful: one extra single-row UPDATE per shard per agent start, one-shot, on the same one-row-per-transaction shard-name- ordered path -- one brief row lock, no convoy. The sequence-0 write it compensates for takes no lock at all, because it matches no rows. The ordering is safe by construction: the store commits the batch that creates the row BEFORE the agent calls MarkSuccessAsync, so the publication that triggers the replay always follows the row write. Also corrects two now-stale doc comments: ProjectionProgressRow.AgentStatus still claimed no store writes it (untrue since #537), and neither it nor ShardState.LastHeartbeat warned that the persisted heartbeat freezes at the last transition once #622's beat is off -- so a monitor thresholding now - LastHeartbeat off the persisted column reports every shard as dead. #5180 restored reading that column in the same unreleased batch, which makes saying so load-bearing. Both new tests fail against the neutralized guard. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016v2Aijyo8MX2AdPUZL5VtG --- .../Daemon/ExtendedProgressionWriterTests.cs | 43 ++++++++++++++++++ .../Daemon/ExtendedProgressionWriter.cs | 44 +++++++++++++++++-- src/JasperFx.Events/ProjectionProgressRow.cs | 23 ++++++++-- src/JasperFx.Events/Projections/ShardState.cs | 10 ++++- 4 files changed, 112 insertions(+), 8 deletions(-) diff --git a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs b/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs index 8b68fc3..5dd5ed8 100644 --- a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs +++ b/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs @@ -505,6 +505,49 @@ public Task> AllProjectionProgress(CancellationToken t => Task.FromResult>([]); } + // 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 _batches = new(); diff --git a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs index 54277ed..f3ce1df 100644 --- a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs +++ b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs @@ -43,6 +43,12 @@ 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. +/// jasperfx#631: a transition published at sequence 0 — which is every fresh shard's +/// , 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. /// 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. @@ -66,6 +72,10 @@ public sealed class ExtendedProgressionWriter : IObserver, IExclusiv // Only ever touched from the tracker's single publication consumer, so no synchronization needed private readonly Dictionary _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 _unlanded = new(); private DateTimeOffset _lastFlush = DateTimeOffset.MinValue; public ExtendedProgressionWriter(IEventStore store, IEventDatabase database, TimeProvider timeProvider, @@ -118,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 @@ -136,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); } diff --git a/src/JasperFx.Events/ProjectionProgressRow.cs b/src/JasperFx.Events/ProjectionProgressRow.cs index ee6e3b2..6132aec 100644 --- a/src/JasperFx.Events/ProjectionProgressRow.cs +++ b/src/JasperFx.Events/ProjectionProgressRow.cs @@ -20,13 +20,28 @@ namespace JasperFx.Events; /// enum's Running/Stopped/Paused. /// /// 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 agent_status 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 +/// , but only when +/// EnableExtendedProgressionTracking is on; with it off the column is not even selected and +/// this reads NULL. /// /// /// -/// 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. +/// +/// ⚠️ Not a liveness signal. Since jasperfx#622 the periodic per-shard beat is OFF by default +/// (), +/// so 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 +/// now - LastHeartbeat will report every shard as dead shortly after startup. Take liveness +/// from the in-memory stream instead (an +/// IObserver<ShardState> on the running daemon, which still beats every 10s), or set a +/// positive ExtendedProgressionHeartbeatInterval and accept the write cost jasperfx#622 and +/// marten#5167 removed. +/// /// public record ProjectionProgressRow( string ProjectionName, diff --git a/src/JasperFx.Events/Projections/ShardState.cs b/src/JasperFx.Events/Projections/ShardState.cs index 05d0150..83b97bb 100644 --- a/src/JasperFx.Events/Projections/ShardState.cs +++ b/src/JasperFx.Events/Projections/ShardState.cs @@ -76,7 +76,15 @@ public ShardState(ShardName shardName, long sequence): this(shardName.Identity, public DateTimeOffset? LastAdvanced { get; set; } /// - /// Last heartbeat received from the shard's subscription agent + /// Last heartbeat received from the shard's subscription agent. + /// + /// On a state published to the live tracker this beats every 10 seconds while the agent runs + /// (SubscriptionAgent's heartbeat timer), so it is a usable liveness signal for an + /// IObserver<ShardState> 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 + /// . + /// /// public DateTimeOffset? LastHeartbeat { get; set; }