fix(marten#4953): CheckNowAsync targets the committed ceiling, time-bounded

Directory.Build.props

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project>
     <PropertyGroup>
-        <JasperFxVersion>2.29.0</JasperFxVersion>
+        <JasperFxVersion>2.29.1</JasperFxVersion>
         <LangVersion>13</LangVersion>
         <NoWarn>1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618</NoWarn>
         <Authors>Jeremy D. Miller;Jaedyn Tonee</Authors>

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<long> FetchCommittedHighWaterCeilingAsync(CancellationToken token)
+        {
+            return Task.FromResult(_committedCeiling);
+        }
+
+        public Task<HighWaterStatistics> Detect(CancellationToken token)
+        {
+            return Task.FromResult(new HighWaterStatistics
+            {
+                CurrentMark = _marks[^1], HighestSequence = ReservedHighest
+            });
+        }
+
+        public Task<HighWaterStatistics> 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
+            });
+        }
+    }
+}

src/JasperFx.Events/Daemon/HighWater/HighWaterAgent.cs

@@ -325,24 +325,49 @@ public Task<HighWaterStatistics> DetermineCurrentMarkAsync(CancellationToken can
         return _detector.Detect(cancellation);
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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));
     }
 

src/JasperFx.Events/Daemon/HighWater/IHighWaterDetector.cs

@@ -6,6 +6,22 @@ public interface IHighWaterDetector
     Task<HighWaterStatistics> Detect(CancellationToken token);
     Uri DatabaseUri { get; }
 
+    /// <summary>
+    /// The highest COMMITTED event sequence right now — the catch-up ceiling used by
+    /// <see cref="HighWaterAgent.CheckNowAsync" /> (marten#4953). <see cref="HighWaterStatistics.HighestSequence" />
+    /// 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.
+    /// </summary>
+    async Task<long> FetchCommittedHighWaterCeilingAsync(CancellationToken token)
+    {
+        var statistics = await Detect(token).ConfigureAwait(false);
+        return statistics.HighestSequence;
+    }
+
     /// <summary>
     /// Does the backing event store partition events per tenant, such that this detector can emit a
     /// meaningful per-tenant high-water vector via <see cref="DetectForTenantsAsync" />? When false (the