diff --git a/src/EventTests/Daemon/SideEffectGateTimeoutTests.cs b/src/EventTests/Daemon/SideEffectGateTimeoutTests.cs new file mode 100644 index 0000000..506c35e --- /dev/null +++ b/src/EventTests/Daemon/SideEffectGateTimeoutTests.cs @@ -0,0 +1,297 @@ +using System.Diagnostics.CodeAnalysis; +using System.Diagnostics.Metrics; +using EventTests.Projections; +using JasperFx; +using JasperFx.Core; +using JasperFx.Events; +using JasperFx.Events.Aggregation; +using JasperFx.Events.Daemon; +using JasperFx.Events.Daemon.HighWater; +using JasperFx.Events.Projections; +using JasperFx.Events.Subscriptions; +using Microsoft.Extensions.Logging; +using NSubstitute; +using Shouldly; + +namespace EventTests.Daemon; + +// jasperfx#594: the blue/green side-effect gate (jasperfx#480) runs a bounded, side-effect-suppressed +// warm-up replay INSIDE the agent start path, and that bound used to be a HARDCODED 5 minutes. Two +// things went wrong at scale (512 tenant databases, warm-ups measured at 27s p50 / 82s p95 / 288.5s max +// against the 300s ceiling): +// +// 1. The ceiling could not be raised, so start failures were the next sample rather than a tail risk. +// 2. A timeout was treated as a FAILURE even when the replay had actually finished — a shard was seen +// failing its start three times on this timeout while its progression sat at exactly the prior +// version's mark, each retry re-running a warm-up that had nothing left to do. +// +// These tests drive the REAL JasperFxAsyncDaemon over a substituted store/database, with an execution +// that never completes its rebuild so the gate's timeout fires deterministically. +public class SideEffectGateTimeoutTests +{ + private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(30); + + private const long PriorMark = 100; + + [Fact] + public void the_default_is_still_five_minutes() + { + new DaemonSettings().SideEffectGateTimeout.ShouldBe(5.Minutes()); + } + + [Fact] + public async Task a_timeout_that_reached_the_prior_mark_is_treated_as_warmed_up() + { + // THE issue's scenario. The warm-up replay wins the race against nothing — it never signals + // completion, so the gate times out — but the PERSISTED progression has reached the prior + // version's mark. That is a warm-up that succeeded and a wait that expired, and the shard must + // go on to start continuous execution rather than be failed and retried forever. + await using var harness = new DaemonHarness( + settings => settings.SideEffectGateTimeout = 250.Milliseconds(), + progressReads: [0, PriorMark]); + + await harness.Daemon.StartAgentAsync("Trip:V2:All", CancellationToken.None); + + // The gate returned true, so the continuous agent was built and started. + harness.Daemon.CurrentAgents().ShouldContain(x => x.Name.Identity == "Trip:V2:All"); + } + + [Fact] + public async Task a_timeout_that_did_not_reach_the_prior_mark_pauses_the_shard() + { + // The genuine failure. Side effects must NOT be enabled over history the prior version already + // covered, so no continuous agent starts — but the shard is published as Paused rather than left + // silently stopped, so it is visible to a supervisor and resumes from its persisted floor. + var paused = new List(); + + await using var harness = new DaemonHarness( + settings => settings.SideEffectGateTimeout = 250.Milliseconds(), + progressReads: [0, PriorMark - 40], + onShardState: state => + { + if (state.Action == ShardAction.Paused) paused.Add(state); + }); + + await harness.Daemon.StartAgentAsync("Trip:V2:All", CancellationToken.None); + + harness.Daemon.CurrentAgents().ShouldNotContain(x => x.Name.Identity == "Trip:V2:All"); + + var state = paused.ShouldHaveSingleItem(); + state.ShardName.ShouldBe("Trip:V2:All"); + state.Sequence.ShouldBe(PriorMark - 40); + state.PauseReason.ShouldContain("side-effect gate warm-up timed out"); + } + + [Theory] + [InlineData("zero")] // TimeSpan.Zero would cancel the replay IMMEDIATELY if it reached the timeout + [InlineData("negative")] // and a negative value is not a legal CancellationTokenSource delay + [InlineData("infinite")] // Timeout.InfiniteTimeSpan is the explicit "no separate bound" spelling + public async Task a_non_positive_or_infinite_timeout_opts_out_of_the_separate_bound(string mode) + { + var timeout = mode switch + { + "zero" => TimeSpan.Zero, + "negative" => TimeSpan.FromSeconds(-5), + _ => Timeout.InfiniteTimeSpan + }; + + await using var harness = new DaemonHarness( + settings => settings.SideEffectGateTimeout = timeout, + progressReads: [0, PriorMark]); + + var start = harness.Daemon.StartAgentAsync("Trip:V2:All", CancellationToken.None); + + // Nothing will ever complete this warm-up, so with the bound opted out the start must still be + // parked. That is the whole assertion: a zero or negative value reaching the replay's + // CancellationTokenSource/TimeoutAfterAsync would have fired essentially immediately, and the + // real 5-minute default would too if the setting were not being read at all. + await Task.Delay(500, TestContext.Current.CancellationToken); + start.IsCompleted.ShouldBeFalse(); + + // The start stays parked for the life of the test; teardown stops the daemon out from under it. + // Observe it so a post-teardown fault never surfaces as an unobserved task exception. + _ = start.ContinueWith(static t => _ = t.Exception, TaskScheduler.Default); + } + + // Real daemon + real SubscriptionAgent over a substituted store/database. The shard opts into the + // gate and is at V2, and the database reports a V1 row at PriorMark, so the gate always triggers. + private sealed class DaemonHarness : IAsyncDisposable + { + public DaemonHarness( + Action configureSettings, + long[] progressReads, + Action? onShardState = null) + { + var shardName = new ShardName("Trip", ShardName.All, 2, null); + Execution = new NeverCompletingRebuildExecution(shardName); + + var factory = Substitute.For>(); + factory.BuildExecution(Arg.Any>(), Arg.Any(), + Arg.Any(), Arg.Any()).Returns(Execution); + factory.BuildExecution(Arg.Any>(), Arg.Any(), + Arg.Any(), Arg.Any()).Returns(Execution); + + var options = new AsyncOptions { GateSideEffectsBehindPriorVersion = true }; + var shard = new AsyncShard(options, ShardRole.Projection, + shardName, factory, new EventFilterable()); + + Store = Substitute.For>(); + Store.Meter.Returns(new Meter("tests")); + Store.TimeProvider.Returns(TimeProvider.System); + Store.AutoCreateSchemaObjects.Returns(AutoCreate.None); + Store.ContinuousErrors.Returns(new ErrorHandlingOptions()); + Store.RebuildErrors.Returns(new ErrorHandlingOptions()); + Store.AllShards().Returns([shard]); + + var loader = Substitute.For(); + loader.LoadAsync(Arg.Any(), Arg.Any()) + .Returns(callInfo => + { + var request = callInfo.Arg(); + var page = new EventPage(request.Floor); + page.CalculateCeiling(request.BatchSize, request.HighWater); + return Task.FromResult(page); + }); + + Store.BuildEventLoader(Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any()).Returns(loader); + Store.BuildEventLoader(Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any()).Returns(loader); + + Database = Substitute.For(); + Database.Identifier.Returns("db1"); + Database.DatabaseUri.Returns(new Uri("fake://db1")); + + var tracker = new ShardStateTracker(new NulloLogger()); + if (onShardState != null) + { + tracker.Subscribe(new ShardStateObserver(onShardState)); + } + + Database.Tracker.Returns(tracker); + + // The gate reads persisted progression twice on the timeout path: once up front to decide + // whether it triggers, then again to judge whether the replay actually finished. + var reads = new Queue(progressReads); + Database.ProjectionProgressFor(Arg.Any(), Arg.Any()) + .Returns(_ => Task.FromResult(reads.Count > 0 ? reads.Dequeue() : 0L)); + + // The prior version's row, which is what makes the gate trigger at all. + Database.AllProjectionProgress(Arg.Any()) + .Returns(Task.FromResult>( + [new ShardState(new ShardName("Trip", ShardName.All, 1, null), PriorMark)])); + + Projections = new FakeProjectionGraph(); + configureSettings(Projections); + + Daemon = new JasperFxAsyncDaemon>( + Store, Database, new NulloLogger(), new GlobalOnlyStubDetector(), Projections); + } + + public IEventStore Store { get; } + public IEventDatabase Database { get; } + public NeverCompletingRebuildExecution Execution { get; } + public ProjectionGraph, FakeOperations, FakeSession> Projections { get; } + public JasperFxAsyncDaemon> Daemon { get; } + + public async ValueTask DisposeAsync() + { + Execution.CompleteRebuild(); + Projections.StopAndDrainTimeout = 100.Milliseconds(); + await Daemon.StopAllAsync(); + Daemon.Dispose(); + } + } + + // The warm-up replay parks forever unless the test releases it, so the gate's timeout — and only the + // gate's timeout — decides when the start path moves on. + private sealed class NeverCompletingRebuildExecution : ISubscriptionExecution + { + private readonly TaskCompletionSource _rebuild = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public NeverCompletingRebuildExecution(ShardName shardName) + { + ShardName = shardName; + } + + public ShardName ShardName { get; } + + public ShardExecutionMode Mode { get; set; } = ShardExecutionMode.Continuous; + + public void CompleteRebuild() => _rebuild.TrySetResult(); + + + public ValueTask EnqueueAsync(EventPage page, ISubscriptionAgent subscriptionAgent) + { + // In Rebuild mode swallow the page so the agent never reports its way to completion; in + // Continuous mode behave normally so a started agent looks healthy. + return Mode == ShardExecutionMode.Rebuild + ? ValueTask.CompletedTask + : subscriptionAgent.MarkSuccessAsync(page.Ceiling); + } + + public Task StopAndDrainAsync(CancellationToken token) => Task.CompletedTask; + + public Task HardStopAsync() => Task.CompletedTask; + + public bool TryBuildReplayExecutor([NotNullWhen(true)] out IReplayExecutor? executor) + { + // Force the non-optimized path, which is the one bounded by TimeoutAfterAsync. + executor = null; + return false; + } + + public Task ProcessImmediatelyAsync(SubscriptionAgent subscriptionAgent, EventPage events, + CancellationToken cancellation) => Task.CompletedTask; + + public Task ProcessRangeAsync(EventRange range) => Task.CompletedTask; + + public bool TryGetAggregateCache([NotNullWhen(true)] out IAggregateCaching? caching) + { + caching = null; + return false; + } + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } + + private sealed class ShardStateObserver : IObserver + { + private readonly Action _onNext; + + public ShardStateObserver(Action onNext) => _onNext = onNext; + + public void OnCompleted() + { + } + + public void OnError(Exception error) + { + } + + public void OnNext(ShardState value) => _onNext(value); + } + + private sealed class GlobalOnlyStubDetector : IHighWaterDetector + { + public Uri DatabaseUri { get; } = new("fake://db1"); + + public Task Detect(CancellationToken token) + => Task.FromResult(new HighWaterStatistics { CurrentMark = PriorMark, HighestSequence = PriorMark }); + + public Task DetectInSafeZone(CancellationToken token) => Detect(token); + } + + private sealed class FakeProjectionGraph : + ProjectionGraph, FakeOperations, FakeSession> + { + public FakeProjectionGraph() : base(Substitute.For(), "tests") + { + } + + protected override void onAddProjection(object projection) + { + // Nothing + } + } +} diff --git a/src/JasperFx.Events/Daemon/DaemonSettings.cs b/src/JasperFx.Events/Daemon/DaemonSettings.cs index a573015..9366d60 100644 --- a/src/JasperFx.Events/Daemon/DaemonSettings.cs +++ b/src/JasperFx.Events/Daemon/DaemonSettings.cs @@ -49,6 +49,16 @@ public interface IReadOnlyDaemonSettings /// TimeSpan StopAndDrainTimeout { get; } + /// + /// How long the blue/green side-effect gate (see + /// AsyncOptions.GateSideEffectsBehindPriorVersion) may spend on the bounded, side-effect- + /// suppressed warm-up replay that runs before a new projection version starts continuous + /// execution. The default is 5 minutes. or any + /// non-positive value means "no separate bound" — the warm-up is then only limited by the + /// daemon's own cancellation. + /// + TimeSpan SideEffectGateTimeout { get; } + /// /// Projection Daemon mode. The default is Disabled /// @@ -133,6 +143,22 @@ public class DaemonSettings: IReadOnlyDaemonSettings /// public TimeSpan StopAndDrainTimeout { get; set; } = 5.Seconds(); + /// + /// jasperfx#594: how long the blue/green side-effect gate (jasperfx#480, + /// AsyncOptions.GateSideEffectsBehindPriorVersion) may spend on the bounded, + /// side-effect-suppressed warm-up replay that runs before a new projection version starts + /// continuous execution. Five minutes was hardcoded originally, which cannot suit both a small + /// store and a database-per-tenant deployment: at 512 tenant databases a warm-up was measured at + /// 27s p50, 82s p95 and 288.5s max — twelve seconds inside the old ceiling, so start failures + /// were the next sample rather than a tail risk. Raise it when a version bump has to replay a + /// large backlog per shard. or any non-positive value + /// means "no separate bound" — the warm-up is then only limited by the daemon's own + /// cancellation. Note that exceeding the bound is no longer automatically a failure: the gate + /// re-reads persisted progression first and treats a shard that reached the prior version's + /// mark as warmed up regardless of the clock. + /// + public TimeSpan SideEffectGateTimeout { get; set; } = 5.Minutes(); + /// /// Projection Daemon mode. The default is Disabled. /// diff --git a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs index b076891..c799698 100644 --- a/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs +++ b/src/JasperFx.Events/Daemon/JasperFxAsyncDaemon.cs @@ -432,9 +432,19 @@ private async Task rebuildAgent(ISubscriptionAgent agent, long highWaterMark, Ti } // jasperfx#480: the bounded warm-up replay is capped like any other rebuild; a timeout leaves the - // shard stopped with its partial progress persisted, and the next start resumes the warm-up from + // shard paused with its partial progress persisted, and the next start resumes the warm-up from // that floor (the gate triggers on progress < prior mark, not only on zero progress). - private static readonly TimeSpan SideEffectGateTimeout = 5.Minutes(); + // + // jasperfx#594: the cap was a hardcoded 5 minutes, which cannot suit both a small store and a + // database-per-tenant deployment — at 512 tenant databases a warm-up measured 288.5s against the + // 300s ceiling. It is now DaemonSettings.SideEffectGateTimeout. A non-positive value or + // Timeout.InfiniteTimeSpan opts out of the separate bound entirely (the daemon's own _cancellation + // still applies), mirroring StopAndDrainTimeout. + private TimeSpan sideEffectGateTimeout() + { + var timeout = _projections.SideEffectGateTimeout; + return timeout > TimeSpan.Zero ? timeout : Timeout.InfiniteTimeSpan; + } // jasperfx#480: single entry point for starting a shard in Continuous mode so every start path // (StartAllAsync, StartAgentAsync by name, the per-tenant fan-outs) runs the opt-in blue/green @@ -501,19 +511,59 @@ private async Task tryApplySideEffectVersionGateAsync(AsyncShard= prior) + { + Logger.LogInformation( + "The blue/green side-effect gate warm-up for projection shard {Name} exceeded the {Timeout} gate timeout but had already reached {Reached} of {Prior}; treating it as warmed up and enabling side effects. Consider raising DaemonSettings.SideEffectGateTimeout", + name.Identity, gateTimeout, reached, prior); + return true; + } + + // A genuine timeout. Publish a Paused state so the shard is observably paused rather + // than silently stopped, and recovers on the next start without operator action — the + // gate resumes from the persisted floor, so the work already done is not repeated. The + // disposed warm-up agent itself is deliberately NOT registered: it would report Running. + Logger.LogError( + "The blue/green side-effect gate warm-up for projection shard {Name} timed out after {Timeout} at {Reached} of {Prior}. The shard is left paused; restarting it will resume the suppressed warm-up from its persisted progress. Consider raising DaemonSettings.SideEffectGateTimeout", + name.Identity, gateTimeout, reached, prior); + + await Tracker.PublishAsync(new ShardState(name, reached) + { + Action = ShardAction.Paused, + AgentStatus = "Paused", + PauseReason = + $"The blue/green side-effect gate warm-up timed out after {gateTimeout} at {reached} of {prior}", + LastHeartbeat = DateTimeOffset.UtcNow + }).ConfigureAwait(false); + + return false; + } + catch (Exception e) { // A failed replay faults ReplayAsync: the agent pauses itself via // ReportCriticalFailureAsync and faults the rebuild completion, which rebuildAgent // propagates — skipping its registration step. Register the paused agent here so the // shard is observably Paused and carries the failure for observers, and do not start // continuous execution over history the prior version already covered. A TimeoutException - // is NOT an agent failure and falls through to the outer catch: the wedged agent is + // never reaches here — it is handled above (jasperfx#594), because the wedged agent is // disposed but never paused, so registering it would misreport the shard as Running. Logger.LogError(e, "The blue/green side-effect gate warm-up for projection shard {Name} failed at {Position}. The shard is left paused; restarting it will resume the suppressed warm-up from its persisted progress",