diff --git a/src/DaemonTests/Internals/GapDetectorTest.cs b/src/DaemonTests/Internals/GapDetectorTest.cs index a617102a1e..f96923cc9e 100644 --- a/src/DaemonTests/Internals/GapDetectorTest.cs +++ b/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(); diff --git a/src/Marten/Events/Daemon/HighWater/GapDetector.cs b/src/Marten/Events/Daemon/HighWater/GapDetector.cs index 6f86ad706f..0262b57b49 100644 --- a/src/Marten/Events/Daemon/HighWater/GapDetector.cs +++ b/src/Marten/Events/Daemon/HighWater/GapDetector.cs @@ -18,9 +18,24 @@ public GapDetector(EventGraph graph) public long Start { get; set; } + /// + /// #4964: when true (the Normal high-water detection path), hold at instead of + /// advancing over a LEADING gap — a hole in the sequence immediately above whose + /// first committed row sits beyond Start + 1. The interior-gap query below only compares + /// consecutive VISIBLE rows, so it cannot see a gap when itself is a hole (there is + /// no visible row at to pair with the first row above the gap); it would then fall + /// through to max(seq_id) 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. + /// + 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 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(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(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)) { diff --git a/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs b/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs index 4fe2938381..b0c2056883 100644 --- a/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs +++ b/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs @@ -90,8 +90,10 @@ public async Task 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 loadCurrentStatistics(CancellationToken return await _runner.Query(_highWaterStatisticsDetector, token).ConfigureAwait(false); } - private async Task findCurrentMark(HighWaterStatistics statistics, CancellationToken token) + private async Task 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 {