fix(#4964): hold the Normal high-water mark before a leading sequence gap

src/DaemonTests/Internals/GapDetectorTest.cs

@@ -75,6 +75,48 @@ public async Task get_max_seq_id_if_start_is_max_seq_id()
         current.ShouldBe(NumberOfEvents);
     }
 
+    [Fact]
+    public async Task normal_path_holds_at_start_before_a_leading_gap_when_start_is_a_hole()
+    {
+        // #4964: the silent-skip blind spot. Delete a contiguous block so Start itself is a hole and the
+        // first committed sequence above it sits beyond Start + 1. The interior-gap query cannot see this
+        // (there is no visible row AT Start to pair with the first row above the gap), so without the hold
+        // it falls through to max(seq_id) and the Normal high-water mark crosses the invisible events with
+        // no trace. With HoldBeforeLeadingGap on (the Normal path), it must hold at Start instead.
+        NumberOfStreams = 10;
+        await PublishSingleThreaded();
+
+        var hole = NumberOfEvents / 2;
+        await deleteEvents(hole, hole + 1, hole + 2, hole + 3, hole + 4, hole + 5);
+
+        theGapDetector.Start = hole; // Start lands ON a hole
+        theGapDetector.HoldBeforeLeadingGap = true; // Normal detection path
+
+        var current = await _runner.Query(theGapDetector, CancellationToken.None);
+
+        current.ShouldBe(hole);
+    }
+
+    [Fact]
+    public async Task safe_zone_path_still_skips_a_leading_gap_forward_to_max()
+    {
+        // The SafeZone counterpart: with HoldBeforeLeadingGap off, the same hole state must keep the
+        // original skip-forward behavior so the SafeZone path can advance past a permanent hole (and
+        // record the skip). This guards against the #4964 hold leaking into the SafeZone path.
+        NumberOfStreams = 10;
+        await PublishSingleThreaded();
+
+        var hole = NumberOfEvents / 2;
+        await deleteEvents(hole, hole + 1, hole + 2, hole + 3, hole + 4, hole + 5);
+
+        theGapDetector.Start = hole;
+        theGapDetector.HoldBeforeLeadingGap = false; // SafeZone path
+
+        var current = await _runner.Query(theGapDetector, CancellationToken.None);
+
+        current.ShouldBe(NumberOfEvents);
+    }
+
     protected async Task deleteEvents(params long[] ids)
     {
         await using var conn = theStore.CreateConnection();

src/Marten/Events/Daemon/HighWater/GapDetector.cs

@@ -18,9 +18,24 @@ public GapDetector(EventGraph graph)
 
     public long Start { get; set; }
 
+    /// <summary>
+    /// #4964: when true (the Normal high-water detection path), hold at <see cref="Start"/> instead of
+    /// advancing over a LEADING gap — a hole in the sequence immediately above <see cref="Start"/> whose
+    /// first committed row sits beyond <c>Start + 1</c>. The interior-gap query below only compares
+    /// consecutive VISIBLE rows, so it cannot see a gap when <see cref="Start"/> itself is a hole (there is
+    /// no visible row at <see cref="Start"/> to pair with the first row above the gap); it would then fall
+    /// through to <c>max(seq_id)</c> and advance the Normal high-water mark over a committed-but-not-yet-
+    /// visible event with no trace. Holding here keeps the Normal mark from ever crossing an unseen event:
+    /// a still-in-flight append fills the gap and the next poll advances normally, while a genuinely
+    /// permanent hole (a rolled-back append) is skipped — and recorded — by the SafeZone path once the mark
+    /// has been stale past the threshold. The SafeZone path leaves this false so it keeps skipping forward.
+    /// </summary>
+    public bool HoldBeforeLeadingGap { get; set; }
+
     public NpgsqlCommand BuildCommand()
     {
         var sql = $@"
+select min(seq_id) from {_graph.DatabaseSchemaName}.mt_events where seq_id > :start;
 select seq_id
 from   (select
                seq_id,
@@ -41,13 +56,29 @@ and    no - seq_id > 1
 
     public async Task<long?> HandleAsync(DbDataReader reader, CancellationToken token)
     {
-        // If there is a row, this tells us the first sequence gap
+        // (0) Leading-gap probe: the lowest committed sequence strictly above Start. In the Normal path,
+        // if it sits beyond Start + 1 there is a gap immediately above Start that the interior-gap query
+        // cannot see, so hold at Start rather than fall through to max and silently cross the gap.
+        long? firstAfterStart = null;
+        if (await reader.ReadAsync(token).ConfigureAwait(false)
+            && !await reader.IsDBNullAsync(0, token).ConfigureAwait(false))
+        {
+            firstAfterStart = await reader.GetFieldValueAsync<long>(0, token).ConfigureAwait(false);
+        }
+
+        if (HoldBeforeLeadingGap && firstAfterStart.HasValue && firstAfterStart.Value > Start + 1)
+        {
+            return Start;
+        }
+
+        // (1) If there is a row, this tells us the first interior sequence gap
+        await reader.NextResultAsync(token).ConfigureAwait(false);
         if (await reader.ReadAsync(token).ConfigureAwait(false))
         {
             return await reader.GetFieldValueAsync<long>(0, token).ConfigureAwait(false);
         }
 
-        // use the latest sequence in the event table because there is NO gap
+        // (2) use the latest sequence in the event table because there is NO gap
         await reader.NextResultAsync(token).ConfigureAwait(false);
         if (!await reader.ReadAsync(token).ConfigureAwait(false))
         {

src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs

@@ -90,8 +90,10 @@ public async Task<HighWaterStatistics> DetectInSafeZone(CancellationToken token)
     {
         var statistics = await loadCurrentStatistics(token).ConfigureAwait(false);
 
-        // Skip gap and find next safe sequence
+        // Skip gap and find next safe sequence. #4964: the SafeZone path must NOT hold before a leading
+        // gap — its whole purpose is to skip forward past a stuck (permanent) hole and record the skip.
         _gapDetector.Start = statistics.SafeStartMark + 1;
+        _gapDetector.HoldBeforeLeadingGap = false;
 
         var safeSequence = await _runner.Query(_gapDetector, token).ConfigureAwait(false);
         if (safeSequence.HasValue)
@@ -419,7 +421,7 @@ private async Task calculateHighWaterMark(HighWaterStatistics statistics, Detect
         }
         else
         {
-            statistics.CurrentMark = await findCurrentMark(statistics, token).ConfigureAwait(false);
+            statistics.CurrentMark = await findCurrentMark(statistics, detectionType, token).ConfigureAwait(false);
         }
 
         if (statistics.HasChanged)
@@ -588,10 +590,16 @@ private async Task<HighWaterStatistics> loadCurrentStatistics(CancellationToken
         return await _runner.Query(_highWaterStatisticsDetector, token).ConfigureAwait(false);
     }
 
-    private async Task<long> findCurrentMark(HighWaterStatistics statistics, CancellationToken token)
+    private async Task<long> findCurrentMark(HighWaterStatistics statistics, DetectionType detectionType,
+        CancellationToken token)
     {
         // look for the current mark
         _gapDetector.Start = statistics.SafeStartMark;
+
+        // #4964: only the Normal path holds before a leading gap. The SafeZone path deliberately skips
+        // forward past a permanent hole (and records the skip), so it must keep the old skip-to-max behavior.
+        _gapDetector.HoldBeforeLeadingGap = detectionType == DetectionType.Normal;
+
         long? current;
         try
         {