From b1c2dc2e6fc8d1c366abb035ad339dbc0fc6e22b Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sat, 18 Jul 2026 10:20:50 -0500 Subject: [PATCH] fix(marten#4953): CheckNowAsync targets the committed ceiling, bounded (2.29.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CheckNowAsync — driven with no staleness gating by projection rebuilds, PrepareForRebuildsAsync, and daemon catch-up — looped DetectInSafeZone until the high water mark reached HighWaterStatistics.HighestSequence. That value can be a database sequence's last_value, which includes numbers merely RESERVED by in-flight or rolled-back transactions, so the loop either pressured safe-zone detection into skipping over in-flight appends (a rebuild or catch-up during an import silently lost their events once they committed) or could spin forever on a rolled-back tail. - New IHighWaterDetector.FetchCommittedHighWaterCeilingAsync DIM: the highest COMMITTED sequence. Default preserves prior behavior (Detect().HighestSequence) so existing detectors are unchanged; Marten overrides with max(seq_id). - CheckNowAsync loops to that committed ceiling — reachable by definition — so a detector that holds before in-flight gaps (Marten's #4953 transaction- evidence gating) simply makes the loop WAIT for the appends to land. - The loop is time-bounded (CheckNowTimeout, default 5 minutes) and logs a warning when it proceeds without reaching the ceiling. Three new HighWaterAgent tests; existing daemon suite green on net9/net10. Co-Authored-By: Claude Fable 5 --- Directory.Build.props | 2 +- ...Agent_check_now_committed_ceiling_Tests.cs | 122 ++++++++++++++++++ .../Daemon/HighWater/HighWaterAgent.cs | 31 ++++- .../Daemon/HighWater/IHighWaterDetector.cs | 16 +++ 4 files changed, 167 insertions(+), 4 deletions(-) create mode 100644 src/EventTests/Daemon/HighWater/HighWaterAgent_check_now_committed_ceiling_Tests.cs diff --git a/Directory.Build.props b/Directory.Build.props index aa52b9e9..9f3df9af 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,7 +1,7 @@ - 2.29.0 + 2.29.1 13 1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618 Jeremy D. Miller;Jaedyn Tonee diff --git a/src/EventTests/Daemon/HighWater/HighWaterAgent_check_now_committed_ceiling_Tests.cs b/src/EventTests/Daemon/HighWater/HighWaterAgent_check_now_committed_ceiling_Tests.cs new file mode 100644 index 00000000..9b154449 --- /dev/null +++ b/src/EventTests/Daemon/HighWater/HighWaterAgent_check_now_committed_ceiling_Tests.cs @@ -0,0 +1,122 @@ +using System.Diagnostics.Metrics; +using JasperFx.Core; +using JasperFx.Events.Daemon; +using JasperFx.Events.Daemon.HighWater; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; + +namespace EventTests.Daemon.HighWater; + +// marten#4953: CheckNowAsync (driven by rebuilds and catch-up) used HighWaterStatistics.HighestSequence +// as its loop target. That value can be a database sequence's last_value, which includes numbers merely +// RESERVED by in-flight or rolled-back transactions — so the loop either pressured the safe-zone +// detection into skipping over in-flight appends (silently losing their events) or spun forever on a +// rolled-back tail. The loop now targets the detector's COMMITTED ceiling and is time-bounded. +public class HighWaterAgent_check_now_committed_ceiling_Tests +{ + private static HighWaterAgent buildAgent(IHighWaterDetector detector, ShardStateTracker tracker, + CancellationToken token, string meter) + { + var settings = new DaemonSettings { SlowPollingTime = 25.Milliseconds() }; + return new HighWaterAgent(new Meter(meter), detector, tracker, NullLogger.Instance, settings, token); + } + + // Reserved sequence numbers far above the committed height (an import in flight, or a rolled-back + // tail) must not keep CheckNowAsync looping — the committed ceiling is the target. + [Fact] + public async Task check_now_targets_the_committed_ceiling_not_the_reserved_highest_sequence() + { + using var cts = new CancellationTokenSource(); + var detector = new StubDetector(committedCeiling: 12, marks: [12]) { ReservedHighest = 100 }; + var tracker = new ShardStateTracker(NullLogger.Instance); + var agent = buildAgent(detector, tracker, cts.Token, "jasperfx.tests.checknow.committed"); + + var completed = agent.CheckNowAsync(); + var winner = await Task.WhenAny(completed, Task.Delay(5.Seconds())); + + winner.ShouldBe(completed); + tracker.HighWaterMark.ShouldBe(12); + await cts.CancelAsync(); + } + + // A detector that holds before an in-flight gap (marten#4953 transaction evidence) makes the loop + // WAIT; once the append lands and the mark reaches the committed ceiling, CheckNowAsync completes. + [Fact] + public async Task check_now_waits_for_a_holding_detector_until_the_ceiling_is_reached() + { + using var cts = new CancellationTokenSource(); + var detector = new StubDetector(committedCeiling: 12, marks: [8, 8, 8, 12]) { ReservedHighest = 12 }; + var tracker = new ShardStateTracker(NullLogger.Instance); + var agent = buildAgent(detector, tracker, cts.Token, "jasperfx.tests.checknow.waits"); + + var completed = agent.CheckNowAsync(); + var winner = await Task.WhenAny(completed, Task.Delay(5.Seconds())); + + winner.ShouldBe(completed); + tracker.HighWaterMark.ShouldBe(12); + detector.SafeZoneCalls.ShouldBeGreaterThanOrEqualTo(4); + await cts.CancelAsync(); + } + + // The timeout backstop: a detector pinned below the ceiling (e.g. a leaked idle-in-transaction + // session upstream) cannot hang CheckNowAsync forever. + [Fact] + public async Task check_now_gives_up_at_the_timeout_instead_of_hanging() + { + using var cts = new CancellationTokenSource(); + var detector = new StubDetector(committedCeiling: 12, marks: [8]) { ReservedHighest = 12 }; + var tracker = new ShardStateTracker(NullLogger.Instance); + var agent = buildAgent(detector, tracker, cts.Token, "jasperfx.tests.checknow.timeout"); + agent.CheckNowTimeout = 300.Milliseconds(); + + var completed = agent.CheckNowAsync(); + var winner = await Task.WhenAny(completed, Task.Delay(5.Seconds())); + + winner.ShouldBe(completed); + tracker.HighWaterMark.ShouldBe(8); + await cts.CancelAsync(); + } + + // Returns marks[i] for the i-th DetectInSafeZone call, clamping at the last entry. + private sealed class StubDetector: IHighWaterDetector + { + private readonly long _committedCeiling; + private readonly long[] _marks; + private int _safeZoneCalls; + + public StubDetector(long committedCeiling, long[] marks) + { + _committedCeiling = committedCeiling; + _marks = marks; + } + + public long ReservedHighest { get; set; } + + public int SafeZoneCalls => Volatile.Read(ref _safeZoneCalls); + + public Uri DatabaseUri { get; } = new("fake://checknow-db"); + + public Task FetchCommittedHighWaterCeilingAsync(CancellationToken token) + { + return Task.FromResult(_committedCeiling); + } + + public Task Detect(CancellationToken token) + { + return Task.FromResult(new HighWaterStatistics + { + CurrentMark = _marks[^1], HighestSequence = ReservedHighest + }); + } + + public Task DetectInSafeZone(CancellationToken token) + { + var call = Interlocked.Increment(ref _safeZoneCalls); + var mark = _marks[Math.Min(call - 1, _marks.Length - 1)]; + return Task.FromResult(new HighWaterStatistics + { + CurrentMark = mark, HighestSequence = ReservedHighest + }); + } + } +} diff --git a/src/JasperFx.Events/Daemon/HighWater/HighWaterAgent.cs b/src/JasperFx.Events/Daemon/HighWater/HighWaterAgent.cs index 579fa05e..a571f87a 100644 --- a/src/JasperFx.Events/Daemon/HighWater/HighWaterAgent.cs +++ b/src/JasperFx.Events/Daemon/HighWater/HighWaterAgent.cs @@ -325,24 +325,49 @@ public Task DetermineCurrentMarkAsync(CancellationToken can return _detector.Detect(cancellation); } + /// + /// Backstop for the CheckNowAsync catch-up loop. The loop's target is the committed ceiling, which + /// is reachable by definition, but a detector legitimately holding before an in-flight append (or a + /// database outage) could stall the loop — after this long CheckNowAsync proceeds with whatever + /// high water mark it has. + /// + public TimeSpan CheckNowTimeout { get; set; } = TimeSpan.FromMinutes(5); + public async Task CheckNowAsync() { + // marten#4953: catch up to the highest COMMITTED sequence, never HighWaterStatistics.HighestSequence. + // That value can be a database sequence's last_value, which includes numbers reserved by + // transactions still in flight — looping DetectInSafeZone until the mark reached the reserved + // ceiling pressured the detection into skipping every in-flight gap during concurrent load + // (a rebuild or catch-up during an import silently lost events) and could spin forever on a + // rolled-back tail. A detector that holds before in-flight gaps now simply makes this loop + // wait for those appends to land; the committed ceiling is reachable by definition. + var ceiling = await _detector.FetchCommittedHighWaterCeilingAsync(_token).ConfigureAwait(false); + var statistics = await _detector.DetectInSafeZone(_token).ConfigureAwait(false); - var initialHighMark = statistics.HighestSequence; // Get out of here if you're at the initial, empty state - if (initialHighMark == 1 && statistics.CurrentMark == 0) + if (statistics.HighestSequence == 1 && statistics.CurrentMark == 0) { await _tracker.MarkHighWaterAsync(statistics.CurrentMark, lastAdvancedFrom(statistics)); return; } - while (statistics.CurrentMark < initialHighMark) + var stopwatch = Stopwatch.StartNew(); + while (statistics.CurrentMark < ceiling && stopwatch.Elapsed < CheckNowTimeout && + !_token.IsCancellationRequested) { await Task.Delay(_settings.SlowPollingTime, _token).ConfigureAwait(false); statistics = await _detector.DetectInSafeZone(_token).ConfigureAwait(false); } + if (statistics.CurrentMark < ceiling) + { + _logger.LogWarning( + "CheckNowAsync for database {Name} stopped at high water mark {CurrentMark} before reaching the committed ceiling {Ceiling} (timeout {Timeout} or cancellation). Continuing with the current mark", + _detector.DatabaseUri, statistics.CurrentMark, ceiling, CheckNowTimeout); + } + await _tracker.MarkHighWaterAsync(statistics.CurrentMark, lastAdvancedFrom(statistics)); } diff --git a/src/JasperFx.Events/Daemon/HighWater/IHighWaterDetector.cs b/src/JasperFx.Events/Daemon/HighWater/IHighWaterDetector.cs index a77b5987..98f68eef 100644 --- a/src/JasperFx.Events/Daemon/HighWater/IHighWaterDetector.cs +++ b/src/JasperFx.Events/Daemon/HighWater/IHighWaterDetector.cs @@ -6,6 +6,22 @@ public interface IHighWaterDetector Task Detect(CancellationToken token); Uri DatabaseUri { get; } + /// + /// The highest COMMITTED event sequence right now — the catch-up ceiling used by + /// (marten#4953). + /// can come from a database sequence's last_value, which includes numbers merely RESERVED by + /// in-flight or rolled-back transactions; treating it as a catch-up target either waits on events + /// that will never exist or pressures the safe-zone detection into skipping events that will. The + /// default implementation preserves the prior behavior (a Detect reading's HighestSequence) so + /// existing detectors keep compiling and behaving unchanged; stores that can read a committed + /// height directly (e.g. max(seq_id) in Marten) should override. + /// + async Task FetchCommittedHighWaterCeilingAsync(CancellationToken token) + { + var statistics = await Detect(token).ConfigureAwait(false); + return statistics.HighestSequence; + } + /// /// Does the backing event store partition events per tenant, such that this detector can emit a /// meaningful per-tenant high-water vector via ? When false (the