diff --git a/src/EventTests/Daemon/ExtendedProgressionHeartbeatIntervalTests.cs b/src/EventTests/Daemon/ExtendedProgressionHeartbeatIntervalTests.cs new file mode 100644 index 0000000..7084ea9 --- /dev/null +++ b/src/EventTests/Daemon/ExtendedProgressionHeartbeatIntervalTests.cs @@ -0,0 +1,184 @@ +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#622 — the periodic per-shard extended progression heartbeat had no reader anywhere and +/// cost one pooled connection + one transaction per database per node every 5 seconds, with no +/// configuration path at all (the interval was private on the daemon and no DaemonSettings knob +/// reached it). It is off by default now, and DaemonSettings.ExtendedProgressionHeartbeatInterval is +/// the compatibility hatch. +/// +public class ExtendedProgressionHeartbeatIntervalTests : IDisposable +{ + private readonly ShardStateTracker theTracker = new(new NulloLogger()); + private readonly RecordingDatabase theDatabase; + + private readonly List>> + theDaemons = []; + + public ExtendedProgressionHeartbeatIntervalTests() + { + theDatabase = new RecordingDatabase(theTracker); + } + + public void Dispose() + { + foreach (var daemon in theDaemons) daemon.Dispose(); + } + + private async Task>> + startedDaemon(TimeSpan? heartbeatInterval) + { + 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 graph = new FakeGraph { ExtendedProgressionHeartbeatInterval = heartbeatInterval }; + + var daemon = new JasperFxAsyncDaemon>( + store, theDatabase, new NulloLogger(), new StubDetector(), graph); + + theDaemons.Add(daemon); + await daemon.StartHighWaterDetectionAsync(); + return daemon; + } + + private static ShardState heartbeat() => new("Trip:All", 5) + { + Action = ShardAction.Updated, AgentStatus = "Running", LastHeartbeat = DateTimeOffset.UtcNow + }; + + private static ShardState transition() => new("Trip:All", 5) + { + Action = ShardAction.Paused, AgentStatus = "Paused", PauseReason = "boom" + }; + + 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 no_periodic_heartbeat_write_by_default() + { + await startedDaemon(null); + + (await writeCountAfterPublishing(heartbeat(), 0)).ShouldBe(0); + } + + [Fact] + public async Task transitions_are_still_persisted_by_default() + { + await startedDaemon(null); + + (await writeCountAfterPublishing(transition(), 1)).ShouldBe(1); + } + + [Fact] + public async Task a_configured_interval_restores_the_periodic_beat() + { + await startedDaemon(TimeSpan.FromSeconds(5)); + + (await writeCountAfterPublishing(heartbeat(), 1)).ShouldBe(1); + } + + [Fact] + public async Task a_non_positive_configured_interval_is_off() + { + await startedDaemon(TimeSpan.Zero); + + (await writeCountAfterPublishing(heartbeat(), 0)).ShouldBe(0); + } + + 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/EventTests/Daemon/ExtendedProgressionSubscriptionLifecycleTests.cs b/src/EventTests/Daemon/ExtendedProgressionSubscriptionLifecycleTests.cs index 0759b02..d6e709a 100644 --- a/src/EventTests/Daemon/ExtendedProgressionSubscriptionLifecycleTests.cs +++ b/src/EventTests/Daemon/ExtendedProgressionSubscriptionLifecycleTests.cs @@ -55,8 +55,13 @@ private JasperFxAsyncDaemon - new(shardName, 5) { AgentStatus = "Running", LastHeartbeat = DateTimeOffset.UtcNow }; + // jasperfx#622: only status transitions are persisted by default, so the probe publication that + // proves "this daemon is writing" has to be one + private static ShardState startedState(string shardName) => + new(shardName, 5) + { + Action = ShardAction.Started, 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 @@ -84,7 +89,7 @@ public async Task an_unstarted_daemon_does_not_write_extended_progression() // every 15 seconds purely to read CurrentAgents() var daemon = buildDaemon(); - (await writeCountAfterPublishing(heartbeat("Trip:All"), 0)).ShouldBe(0); + (await writeCountAfterPublishing(startedState("Trip:All"), 0)).ShouldBe(0); } [Fact] @@ -95,7 +100,7 @@ public async Task many_unstarted_daemons_on_one_shared_tracker_still_write_nothi buildDaemon(); } - (await writeCountAfterPublishing(heartbeat("Trip:All"), 0)).ShouldBe(0); + (await writeCountAfterPublishing(startedState("Trip:All"), 0)).ShouldBe(0); } [Fact] @@ -104,7 +109,7 @@ public async Task a_started_daemon_does_write_extended_progression() var daemon = buildDaemon(); await daemon.StartHighWaterDetectionAsync(); - (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + (await writeCountAfterPublishing(startedState("Trip:All"), 1)).ShouldBe(1); } [Fact] @@ -119,7 +124,7 @@ public async Task only_the_started_daemon_writes_when_readers_share_the_tracker( await owner.StartHighWaterDetectionAsync(); - (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + (await writeCountAfterPublishing(startedState("Trip:All"), 1)).ShouldBe(1); } [Fact] @@ -129,7 +134,7 @@ public async Task starting_twice_does_not_stack_a_second_subscription() await daemon.StartHighWaterDetectionAsync(); await daemon.StartHighWaterDetectionAsync(); - (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + (await writeCountAfterPublishing(startedState("Trip:All"), 1)).ShouldBe(1); } [Fact] @@ -137,11 +142,11 @@ public async Task a_stopped_daemon_stops_writing() { var daemon = buildDaemon(); await daemon.StartHighWaterDetectionAsync(); - (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + (await writeCountAfterPublishing(startedState("Trip:All"), 1)).ShouldBe(1); await daemon.StopAllAsync(); - (await writeCountAfterPublishing(heartbeat("Trip:All"), 2)).ShouldBe(1); + (await writeCountAfterPublishing(startedState("Trip:All"), 2)).ShouldBe(1); } [Fact] @@ -154,7 +159,7 @@ public async Task a_restarted_daemon_resumes_writing() await daemon.StopAllAsync(); await daemon.StartHighWaterDetectionAsync(); - (await writeCountAfterPublishing(heartbeat("Trip:All"), 1)).ShouldBe(1); + (await writeCountAfterPublishing(startedState("Trip:All"), 1)).ShouldBe(1); } [Fact] diff --git a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs b/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs index 0b8858f..08886a4 100644 --- a/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs +++ b/src/EventTests/Daemon/ExtendedProgressionWriterTests.cs @@ -21,6 +21,14 @@ public ExtendedProgressionWriterTests() theWriter = new ExtendedProgressionWriter(theStore, theDatabase, theTime, NullLogger.Instance); } + // jasperfx#622: the periodic per-shard beat is OFF by default now. The machinery is unchanged and + // still shipped behind DaemonSettings.ExtendedProgressionHeartbeatInterval, so the tests that pin + // its coalescing/flush behavior opt into it explicitly. + private void enablePeriodicHeartbeats() + { + theWriter.HeartbeatWriteInterval = TimeSpan.FromSeconds(5); + } + private static ShardState transition(ShardAction action, string status, string? pauseReason = null, string shardName = "Counters:All") { @@ -100,6 +108,8 @@ public async Task skips_plain_progress_publications_with_no_agent_telemetry() [Fact] public async Task coalesces_heartbeats_into_one_batched_write_per_flush_interval() { + enablePeriodicHeartbeats(); + // The very first heartbeat flushes immediately (nothing to wait for) theWriter.OnNext(heartbeat(sequence: 1)); var writes = await theDatabase.WaitForWrites(1); @@ -127,6 +137,8 @@ public async Task coalesces_heartbeats_into_one_batched_write_per_flush_interval [Fact] public async Task a_transition_flushes_immediately_and_carries_the_pending_heartbeats_along() { + enablePeriodicHeartbeats(); + // Seed a flush so the interval throttle is active theWriter.OnNext(heartbeat(sequence: 1)); await theDatabase.WaitForWrites(1); @@ -148,6 +160,8 @@ public async Task a_transition_flushes_immediately_and_carries_the_pending_heart [Fact] public async Task a_transition_replaces_a_pending_heartbeat_for_the_same_shard() { + enablePeriodicHeartbeats(); + theWriter.OnNext(heartbeat(sequence: 1)); await theDatabase.WaitForWrites(1); @@ -205,6 +219,8 @@ public async Task does_not_clobber_an_explicit_running_on_node() [Fact] public async Task disposing_flushes_the_pending_batch() { + enablePeriodicHeartbeats(); + theWriter.OnNext(heartbeat(sequence: 1)); await theDatabase.WaitForWrites(1); @@ -242,6 +258,54 @@ public async Task dispose_awaits_the_in_flight_write_rather_than_draining_in_the writes[0].AgentStatus.ShouldBe("Stopped"); } + // jasperfx#622 + [Fact] + public async Task no_periodic_heartbeat_write_by_default() + { + // The whole point: a heartbeat nobody reads costs a connection + a transaction per database + // per node per interval (marten#5167). Off unless asked for. + theWriter.PeriodicHeartbeatsEnabled.ShouldBeFalse(); + + theWriter.OnNext(heartbeat(sequence: 1)); + theWriter.OnNext(heartbeat(sequence: 2)); + theWriter.OnNext(heartbeat("Others:All", sequence: 7)); + theTime.Advance(TimeSpan.FromMinutes(5)); + theWriter.OnNext(heartbeat(sequence: 3)); + + await theDatabase.AssertNoWrites(); + } + + // jasperfx#622: the transitional columns are exactly what survives -- rare writes, and the data + // the "durable across a crash" story was actually about + [Fact] + public async Task transitions_are_still_written_with_the_periodic_beat_off() + { + theWriter.OnNext(heartbeat(sequence: 1)); + theWriter.OnNext(transition(ShardAction.Paused, "Paused", "boom")); + + var writes = await theDatabase.WaitForWrites(1); + var batches = await theDatabase.Batches(); + + // ...and the dropped heartbeat does NOT ride along on the transition's write + batches.Count.ShouldBe(1); + batches[0].Length.ShouldBe(1); + writes[0].AgentStatus.ShouldBe("Paused"); + writes[0].PauseReason.ShouldBe("boom"); + } + + // jasperfx#622: the compatibility hatch the interval never had + [Fact] + public async Task a_positive_interval_restores_the_periodic_beat() + { + enablePeriodicHeartbeats(); + theWriter.PeriodicHeartbeatsEnabled.ShouldBeTrue(); + + theWriter.OnNext(heartbeat(sequence: 1)); + + var writes = await theDatabase.WaitForWrites(1); + writes[0].Sequence.ShouldBe(1); + } + [Fact] public async Task the_default_batch_implementation_degrades_to_single_state_writes() { diff --git a/src/JasperFx.Events/Daemon/DaemonSettings.cs b/src/JasperFx.Events/Daemon/DaemonSettings.cs index 3332b6b..03c0807 100644 --- a/src/JasperFx.Events/Daemon/DaemonSettings.cs +++ b/src/JasperFx.Events/Daemon/DaemonSettings.cs @@ -40,6 +40,18 @@ public interface IReadOnlyDaemonSettings /// TimeSpan HighWaterStalenessThreshold { get; } + /// + /// jasperfx#622: cadence of the PERIODIC per-shard extended progression heartbeat write. + /// Null (the default) or any non-positive value means no periodic write at all — only agent + /// status transitions (Started/Paused/Stopped) are persisted, which is what the extended + /// columns are actually read for. A positive value restores the pre-#622 behavior at that + /// cadence and is a compatibility hatch, not the recommended shape: the periodic beat costs + /// one pooled connection and one transaction per database per node per interval, and nothing + /// in JasperFx, Marten or CritterWatch reads the persisted heartbeat (marten#5167). + /// Only consulted when IEventStore.ExtendedProgressionEnabled is on. + /// + TimeSpan? ExtendedProgressionHeartbeatInterval { get; } + /// /// How long the daemon will wait for a single subscription or projection shard to gracefully /// finish its in-flight page and flush its progression when that agent is stopped. Exceeding @@ -111,6 +123,15 @@ public class DaemonSettings: IReadOnlyDaemonSettings /// public TimeSpan SlowPollingTime { get; set; } = 1.Seconds(); + /// + /// jasperfx#622: cadence of the PERIODIC per-shard extended progression heartbeat write. Null + /// (the default) or any non-positive value means no periodic write at all — only agent status + /// transitions are persisted. A positive value restores the pre-#622 5 second beat at that + /// cadence; see + /// for why that is a compatibility hatch rather than a recommendation. + /// + public TimeSpan? ExtendedProgressionHeartbeatInterval { get; set; } + /// /// Polling time between looking for a new high water sequence mark /// if the daemon detects high activity. The default is 250ms diff --git a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs index d2582d2..7f97b51 100644 --- a/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs +++ b/src/JasperFx.Events/Daemon/ExtendedProgressionWriter.cs @@ -17,16 +17,28 @@ namespace JasperFx.Events.Daemon; /// /// Gated on , read live per publication — /// nothing is written (or even queued) for stores that have not opted in. -/// Heartbeat/telemetry publications ( — the ~10s heartbeat -/// timer ticks and the per-batch commit publications) are coalesced per shard (latest state wins) -/// and flushed as ONE batched database write per for the whole +/// jasperfx#622: heartbeat/telemetry publications () are +/// DROPPED by default. The periodic per-shard beat had no reader anywhere — not in JasperFx, not +/// in Marten, and not in CritterWatch, which reads agent status and heartbeats off in-memory +/// objects and drops the persisted columns on the floor — while costing one pooled connection and +/// one transaction per database per node every 5 seconds. On a 512-shard-database deployment that +/// was ~37 connection acquisitions/sec/node to keep 6-12 rows current, and it made a production web +/// app unresponsive (marten#5167). Liveness is a node property and is tracked as one per node +/// upstream; last_updated plus the agent assignment grid reconstructs what any consumer +/// actually renders. Set (or +/// DaemonSettings.ExtendedProgressionHeartbeatInterval) to a positive value to restore the +/// old behavior — that is the compatibility hatch, not the recommended shape. +/// When periodic beats ARE enabled, they are coalesced per shard (latest state wins) and +/// flushed as ONE batched database write per for the whole /// database. The write rate is therefore constant per database instead of O(shards): under /// per-tenant agent fan-out (agents = projections × tenants) the previous /// one-connection-rent-per-shard-per-interval write path drove a sharded multi-tenant deployment /// to its database server's connection ceiling (jasperfx#553). /// Agent status transitions (, , /// ) flush immediately — a paused/stopped shard is exactly when the -/// persisted status matters most. The pending heartbeat batch rides along in the same write. +/// 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. /// 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. @@ -64,11 +76,24 @@ public ExtendedProgressionWriter(IEventStore store, IEventDatabase database, Tim /// /// Spacing between two batched heartbeat/telemetry flushes for the database. All shard states /// that arrive within the interval are coalesced (latest state per shard) into the next flush, - /// so no heartbeat is ever more than one interval stale. Status transitions flush immediately, - /// carrying any pending batch with them. Defaults to 5 seconds so every tick of the agents' - /// 10 second heartbeat timer lands. + /// so no heartbeat is ever more than one interval stale. Status transitions always flush + /// immediately, carrying any pending batch with them. + /// + /// + /// jasperfx#622: defaults to — periodic heartbeat writes are OFF, and + /// only status transitions are persisted. Zero or negative disables them; any positive value + /// restores the periodic beat at that cadence. Before #622 this was hardcoded to 5 seconds with + /// no configuration path at all (the field was private on the daemon, reachable from no + /// DaemonSettings knob), which is what made the cost impossible to opt out of. + /// /// - public TimeSpan HeartbeatWriteInterval { get; set; } = TimeSpan.FromSeconds(5); + public TimeSpan HeartbeatWriteInterval { get; set; } = TimeSpan.Zero; + + /// + /// Whether this writer persists the periodic per-shard heartbeat at all. False by default; see + /// . + /// + public bool PeriodicHeartbeatsEnabled => HeartbeatWriteInterval > TimeSpan.Zero; public void OnNext(ShardState value) { @@ -80,6 +105,13 @@ public void OnNext(ShardState value) // 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; + + // 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; + // 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) @@ -91,7 +123,6 @@ public void OnNext(ShardState value) // simply replaces it _pending[value.ShardName] = value; - var isTransition = value.Action is ShardAction.Started or ShardAction.Paused or ShardAction.Stopped; var now = _timeProvider.GetUtcNow(); if (isTransition || now - _lastFlush >= HeartbeatWriteInterval) diff --git a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs index dbd3d4c..2d83633 100644 --- a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs +++ b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs @@ -256,7 +256,14 @@ private RetryBlock buildDeadLetterBlock() // StopAllAsync drained it, exactly as it rebuilds the dead-letter block (jasperfx#557). private ExtendedProgressionWriter buildExtendedProgressionWriter() => new(_store, Database, _store.TimeProvider, - _loggerFactory?.CreateLogger() ?? Logger); + _loggerFactory?.CreateLogger() ?? Logger) + { + // jasperfx#622: off unless the application asks for it. This is the configuration path + // the interval never had -- before #622 it was a hardcoded 5 seconds that no + // DaemonSettings knob could reach. + HeartbeatWriteInterval = + _projections.ExtendedProgressionHeartbeatInterval ?? TimeSpan.Zero + }; /// /// jasperfx#621: arm the extended progression writer. Called from every path that actually starts @@ -1189,10 +1196,15 @@ private async Task pollTenantHighWaterAsync() await _tenantHighWater.PollAndRouteAsync(CurrentAgents(), _cancellation.Token).ConfigureAwait(false); // jasperfx#539: publish the per-cycle liveness heartbeat for Path B. The coordinator has already - // stamped its in-memory LastPolledAt; this surfaces the same beat on the live Tracker (and the - // ExtendedProgression columns) so remote consumers can tell "no new events" from "the tenant - // high-water poll died". Carries the store-global mark unchanged, so it never advances it and, - // by the OnNext guard above, never re-triggers a poll. + // stamped its in-memory LastPolledAt; this surfaces the same beat on the live Tracker so + // in-process consumers can tell "no new events" from "the tenant high-water poll died". + // Carries the store-global mark unchanged, so it never advances it and, by the OnNext guard + // above, never re-triggers a poll. + // + // jasperfx#622: this beat does NOT reach the ExtendedProgression columns, and never did -- + // ExtendedProgressionWriter.OnNext drops HighWaterMark and AllProjections states outright + // (pinned by skips_high_water_mark_and_all_projections_states). The live Tracker is the only + // place it shows up. await publishHighWaterStatusAsync(ShardAction.Updated, "Running").ConfigureAwait(false); } catch (Exception e)