diff --git a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs b/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs index 76a8978..5dd5ed8 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(); @@ -411,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 6087367..f3ce1df 100644 --- a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs +++ b/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. +/// 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. /// /// /// -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; @@ -59,6 +72,10 @@ public sealed class ExtendedProgressionWriter : IObserver, IAsyncDis // 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, @@ -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); } 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); } } 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; }