diff --git a/src/EventTests/Daemon/ExtendedProgressionSubscriptionLifecycleTests.cs b/src/EventTests/Daemon/ExtendedProgressionSubscriptionLifecycleTests.cs new file mode 100644 index 0000000..0759b02 --- /dev/null +++ b/src/EventTests/Daemon/ExtendedProgressionSubscriptionLifecycleTests.cs @@ -0,0 +1,232 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Threading; +using System.Threading.Tasks; +using EventTests.Projections; +using JasperFx; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Daemon.HighWater; +using JasperFx.Events.Projections; +using NSubstitute; +using Shouldly; + +namespace EventTests.Daemon; + +/// +/// jasperfx#621 — the ShardStateTracker is per-database and SHARED, so subscribing an +/// ExtendedProgressionWriter in the daemon constructor meant every daemon ever built for a database +/// added another writer to the same publication stream, each renting its own connection and issuing +/// the same UPDATE against the same rows. Building a daemon is a documented way to *read* projection +/// state (BuildProjectionDaemonAsync returns a fresh instance per call, no caching), and a read must +/// not acquire a background write loop as an invisible side effect. +/// +public class ExtendedProgressionSubscriptionLifecycleTests : IDisposable +{ + private readonly ShardStateTracker theTracker = new(new NulloLogger()); + private readonly RecordingDatabase theDatabase; + + public ExtendedProgressionSubscriptionLifecycleTests() + { + theDatabase = new RecordingDatabase(theTracker); + } + + private readonly List>> + theDaemons = []; + + public void Dispose() + { + foreach (var daemon in theDaemons) daemon.Dispose(); + } + + private JasperFxAsyncDaemon> buildDaemon() + { + var store = Substitute.For>(); + store.Meter.Returns(new Meter("tests")); + store.TimeProvider.Returns(TimeProvider.System); + store.ExtendedProgressionEnabled.Returns(true); + store.AutoCreateSchemaObjects.Returns(AutoCreate.None); + + var daemon = new JasperFxAsyncDaemon>( + store, theDatabase, new NulloLogger(), new StubDetector(), new FakeGraph()); + + theDaemons.Add(daemon); + return daemon; + } + + private static ShardState heartbeat(string shardName) => + new(shardName, 5) { AgentStatus = "Running", LastHeartbeat = DateTimeOffset.UtcNow }; + + // Publications are delivered asynchronously through the tracker's block, so both the positive and + // the negative assertion have to be given the same chance to happen + private async Task writeCountAfterPublishing(ShardState state, int expected) + { + await theTracker.PublishAsync(state); + + for (var i = 0; i < 100 && theDatabase.Writes < expected; i++) + { + await Task.Delay(20, TestContext.Current.CancellationToken); + } + + if (expected == 0) + { + await Task.Delay(250, TestContext.Current.CancellationToken); + } + + return theDatabase.Writes; + } + + [Fact] + public async Task an_unstarted_daemon_does_not_write_extended_progression() + { + // The reported shape: an EventProgressionPoller building a daemon per node-owned database + // every 15 seconds purely to read CurrentAgents() + var daemon = buildDaemon(); + + (await writeCountAfterPublishing(heartbeat("Trip:All"), 0)).ShouldBe(0); + } + + [Fact] + public async Task many_unstarted_daemons_on_one_shared_tracker_still_write_nothing() + { + for (var i = 0; i < 5; i++) + { + buildDaemon(); + } + + (await writeCountAfterPublishing(heartbeat("Trip:All"), 0)).ShouldBe(0); + } + + [Fact] + public async Task a_started_daemon_does_write_extended_progression() + { + var daemon = buildDaemon(); + await daemon.StartHighWaterDetectionAsync(); + + (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + } + + [Fact] + public async Task only_the_started_daemon_writes_when_readers_share_the_tracker() + { + // One owner plus three ad-hoc readers on the same database: exactly one write per publication, + // not four. This is the multiplication behind the pg_stat_activity self-blocking in marten#5167. + var owner = buildDaemon(); + buildDaemon(); + buildDaemon(); + buildDaemon(); + + await owner.StartHighWaterDetectionAsync(); + + (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + } + + [Fact] + public async Task starting_twice_does_not_stack_a_second_subscription() + { + var daemon = buildDaemon(); + await daemon.StartHighWaterDetectionAsync(); + await daemon.StartHighWaterDetectionAsync(); + + (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + } + + [Fact] + public async Task a_stopped_daemon_stops_writing() + { + var daemon = buildDaemon(); + await daemon.StartHighWaterDetectionAsync(); + (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + + await daemon.StopAllAsync(); + + (await writeCountAfterPublishing(heartbeat("Trip:All"), 2)).ShouldBe(1); + } + + [Fact] + public async Task a_restarted_daemon_resumes_writing() + { + // jasperfx#557's guarantee, preserved: the drained writer is rebuilt on the next start rather + // than eagerly at the end of StopAllAsync + var daemon = buildDaemon(); + await daemon.StartHighWaterDetectionAsync(); + await daemon.StopAllAsync(); + await daemon.StartHighWaterDetectionAsync(); + + (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + } + + [Fact] + public void disposing_a_never_started_daemon_is_clean() + { + var daemon = buildDaemon(); + Should.NotThrow(daemon.Dispose); + Should.NotThrow(daemon.Dispose); + } + + private sealed class RecordingDatabase : IEventDatabase + { + private int _writes; + + public RecordingDatabase(ShardStateTracker tracker) => Tracker = tracker; + + public int Writes => Volatile.Read(ref _writes); + + public Task WriteExtendedProgressionAsync(IReadOnlyList states, CancellationToken token = default) + { + Interlocked.Increment(ref _writes); + return Task.CompletedTask; + } + + public Task WriteExtendedProgressionAsync(ShardState state, CancellationToken token = default) + { + Interlocked.Increment(ref _writes); + return Task.CompletedTask; + } + + public ShardStateTracker Tracker { get; } + public string Identifier => "db1"; + public Uri DatabaseUri { get; } = new("fake://db1"); + public string StorageIdentifier => "db1"; + + 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 ProjectionProgressFor(ShardName name, CancellationToken token = default) + => Task.FromResult(0L); + + public Task FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token) + => Task.FromResult(null); + + public Task FetchHighestEventSequenceNumber(CancellationToken token) => Task.FromResult(0L); + + public Task> AllProjectionProgress(CancellationToken token = default) + => Task.FromResult>([]); + } + + private sealed class StubDetector : IHighWaterDetector + { + public Uri DatabaseUri { get; } = new("fake://db1"); + + public Task Detect(CancellationToken token) + => Task.FromResult(new HighWaterStatistics()); + + public Task DetectInSafeZone(CancellationToken token) => Detect(token); + } + + private sealed class FakeGraph : ProjectionGraph, FakeOperations, FakeSession> + { + public FakeGraph() : base(Substitute.For(), "tests") + { + } + + protected override void onAddProjection(object projection) + { + // Nothing + } + } +} diff --git a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs index 2ac0f1a..dbd3d4c 100644 --- a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs +++ b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs @@ -36,8 +36,15 @@ public partial class JasperFxAsyncDaemon _deadLetterBlock; private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1); @@ -188,9 +195,7 @@ public JasperFxAsyncDaemon(IEventStore store, IEvent _breakSubscription = database.Tracker.Subscribe(this); - _extendedProgression = buildExtendedProgressionWriter(); - _extendedProgressionSubscription = Tracker.Subscribe(_extendedProgression); - + // jasperfx#621: the extended progression writer is armed on the start path, NOT here _deadLetterBlock = buildDeadLetterBlock(); MaxConcurrentEventLoadsPerDatabase = _projections.MaxConcurrentEventLoadsPerDatabase; @@ -223,9 +228,7 @@ public JasperFxAsyncDaemon(IEventStore store, IEvent _breakSubscription = database.Tracker.Subscribe(this); - _extendedProgression = buildExtendedProgressionWriter(); - _extendedProgressionSubscription = Tracker.Subscribe(_extendedProgression); - + // jasperfx#621: the extended progression writer is armed on the start path, NOT here _deadLetterBlock = buildDeadLetterBlock(); MaxConcurrentEventLoadsPerDatabase = _projections.MaxConcurrentEventLoadsPerDatabase; @@ -248,13 +251,40 @@ private RetryBlock buildDeadLetterBlock() }, Logger, _cancellation.Token); } - // jasperfx#537: subscribe unconditionally; the writer checks the store's ExtendedProgressionEnabled - // flag live per publication so runtime opt-in is honored. Built through a helper so StopAllAsync can - // rebuild it after draining, exactly as it rebuilds the dead-letter block (jasperfx#557). + // jasperfx#537: the writer checks the store's ExtendedProgressionEnabled flag live per publication + // so runtime opt-in is honored. Built through a helper so the start path can rebuild it after + // StopAllAsync drained it, exactly as it rebuilds the dead-letter block (jasperfx#557). private ExtendedProgressionWriter buildExtendedProgressionWriter() => new(_store, Database, _store.TimeProvider, _loggerFactory?.CreateLogger() ?? Logger); + /// + /// jasperfx#621: arm the extended progression writer. Called from every path that actually starts + /// this daemon's agents -- and from nowhere else, so a daemon built purely to inspect state + /// (CurrentAgents(), Tracker reads) never subscribes a telemetry writer to the database's SHARED + /// tracker, and never writes to the database at all. Idempotent: repeated starts re-use the + /// armed writer rather than stacking a second subscription on the same tracker. + /// + private void armExtendedProgressionWriter() + { + if (_disposed || _extendedProgressionSubscription != null) return; + + _extendedProgression = buildExtendedProgressionWriter(); + _extendedProgressionSubscription = Tracker.Subscribe(_extendedProgression); + } + + // jasperfx#621: unsubscribe from the shared tracker and drain. Split out because Dispose() (sync) + // and StopAllAsync (async, where the drain can be awaited) both need it, and both must leave the + // daemon disarmed so a later start re-arms a fresh writer rather than resurrecting a completed one. + private ExtendedProgressionWriter? detachExtendedProgressionWriter() + { + var writer = _extendedProgression; + _extendedProgressionSubscription?.Dispose(); + _extendedProgressionSubscription = null; + _extendedProgression = null; + return writer; + } + public IEventDatabase Database { get; } public ILogger Logger { get; } @@ -273,9 +303,13 @@ public void Dispose() _tenantHighWaterTimer?.Stop(); _tenantHighWaterTimer?.Dispose(); _breakSubscription.Dispose(); - _extendedProgressionSubscription.Dispose(); - // Completes the writer's queue so a final Stopped write can drain in the background - _ = _extendedProgression.DisposeAsync(); + // Completes the writer's queue so a final Stopped write can drain in the background. Null + // when this daemon was never started (jasperfx#621) -- nothing to unsubscribe or drain. + var writer = detachExtendedProgressionWriter(); + if (writer != null) + { + _ = writer.DisposeAsync(); + } _deadLetterBlock.Dispose(); _loadThrottle?.Dispose(); _batchWriteThrottle?.Dispose(); @@ -300,6 +334,10 @@ public void Dispose() private async Task tryStartAgentAsync(ISubscriptionAgent agent, ShardExecutionMode mode, long sideEffectGateMark = 0) { + // jasperfx#621: this daemon is about to own a running agent, so it owns that agent's telemetry. + // Every continuous start path funnels through here; a daemon built only to read state does not. + armExtendedProgressionWriter(); + // Be idempotent, don't start an agent that is already running if (_agents.TryFind(agent.Name.Identity, out var running) && running.Status == AgentStatus.Running) { @@ -429,6 +467,9 @@ private async Task tryStartAgentAsync(ISubscriptionAgent agent, ShardExecu // an effective concurrency of one, making any cap > 1 unreachable. private async Task rebuildAgent(ISubscriptionAgent agent, long highWaterMark, TimeSpan shardTimeout) { + // jasperfx#621: a rebuild agent publishes real status transitions for this daemon's shards + armExtendedProgressionWriter(); + var budget = _rebuildBudget; if (budget != null) { @@ -925,8 +966,11 @@ await Parallel.ForEachAsync(activeAgents, cancellation.Token, // deliberate write to the same progression row. try { - _extendedProgressionSubscription.Dispose(); - await _extendedProgression.DisposeAsync().ConfigureAwait(false); + var writer = detachExtendedProgressionWriter(); + if (writer != null) + { + await writer.DisposeAsync().ConfigureAwait(false); + } } catch (Exception e) { @@ -939,10 +983,9 @@ await Parallel.ForEachAsync(activeAgents, cancellation.Token, _cancellation.TryReset(); _deadLetterBlock = buildDeadLetterBlock(); - // Rebuild the drained writer + resubscribe so a subsequent StartAllAsync (e.g. resume after a - // rebuild) keeps persisting extended progression, mirroring the dead-letter block rebuild above. - _extendedProgression = buildExtendedProgressionWriter(); - _extendedProgressionSubscription = Tracker.Subscribe(_extendedProgression); + // jasperfx#621: deliberately NOT resubscribed here. A stopped daemon writes no telemetry; + // the next start path re-arms a fresh writer (armExtendedProgressionWriter), which is where + // the dead-letter block's eager rebuild above and this part company. } finally { @@ -952,6 +995,9 @@ await Parallel.ForEachAsync(activeAgents, cancellation.Token, public async Task StartHighWaterDetectionAsync() { + // jasperfx#621: this daemon is genuinely starting, so it owns its shards' telemetry + armExtendedProgressionWriter(); + if (_store.AutoCreateSchemaObjects != AutoCreate.None) { await Database.EnsureStorageExistsAsync(typeof(IEvent), _cancellation.Token).ConfigureAwait(false); @@ -1722,6 +1768,9 @@ private async Task stopRunningAgents(string subscriptionName) public async Task PrepareForRebuildsAsync() { + // jasperfx#621: a rebuild runs real agents that publish real status transitions + armExtendedProgressionWriter(); + if (_highWater.IsRunning) { await _highWater.StopAsync().ConfigureAwait(false);