diff --git a/Directory.Packages.props b/Directory.Packages.props
index fa1fb11344..1ee98eedd3 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -107,14 +107,19 @@
natural key that CHANGES via an update event is written to mt_natural_key_X on live append and
rebuild; previously only the create event's key was ever recorded.
JasperFx 2.29.0: jasperfx#529 (marten#4975) — exact ReadProjectionProgressAsync(ShardName) overload
- on IEventDatabase (no version/shard collapsing), adopted by MartenDatabase here. -->
-
-
-
+ on IEventDatabase (no version/shard collapsing), adopted by MartenDatabase here.
+ JasperFx 2.29.1: jasperfx#530 (marten#4953) — HighWaterAgent.CheckNowAsync catches up to the
+ COMMITTED ceiling via the new IHighWaterDetector.FetchCommittedHighWaterCeilingAsync DIM
+ (overridden here with max(seq_id)) instead of looping DetectInSafeZone to the reserved
+ last_value, and the loop is time-bounded — rebuild/catch-up during concurrent appends now
+ waits for in-flight events instead of pressuring detection to skip them. -->
+
+
+
all
runtime; build; native; contentfiles; analyzers; buildtransitive
-
+
diff --git a/src/DaemonTests/Bugs/Bug_4953_outstanding_sequence_gap_skips.cs b/src/DaemonTests/Bugs/Bug_4953_outstanding_sequence_gap_skips.cs
new file mode 100644
index 0000000000..ac049f7533
--- /dev/null
+++ b/src/DaemonTests/Bugs/Bug_4953_outstanding_sequence_gap_skips.cs
@@ -0,0 +1,482 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using DaemonTests.TestingSupport;
+using JasperFx.Core;
+using JasperFx.Events;
+using JasperFx.Events.Projections;
+using Marten;
+using Marten.Events;
+using Marten.Events.Daemon.HighWater;
+using Marten.Events.Projections;
+using Marten.Storage;
+using Marten.Testing;
+using Marten.Testing.Harness;
+using Microsoft.Extensions.Logging.Abstractions;
+using Npgsql;
+using Shouldly;
+using Weasel.Postgresql;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace DaemonTests.Bugs;
+
+///
+/// Regressions for discussion #4953: the async daemon's high water detection must never advance
+/// across an OUTSTANDING sequence number — one reserved by a transaction that is still in flight
+/// and will commit later. Four mechanisms could do exactly that (see the issue): per-statement
+/// snapshot skew inside the GapDetector command, threshold-free safe-zone skips from
+/// rebuild/catch-up, the "-32" stale fallback teleporting into reserved-but-uncommitted sequence
+/// numbers, and the wall-clock stale skip crossing transactions that are merely slow.
+///
+public class Bug_4953_outstanding_sequence_gap_skips: DaemonContext
+{
+ private readonly ITestOutputHelper _output;
+
+ public Bug_4953_outstanding_sequence_gap_skips(ITestOutputHelper output): base(output)
+ {
+ _output = output;
+ }
+
+ private string Schema => theStore.Events.DatabaseSchemaName;
+
+ #region helpers
+
+ private async Task openConnection()
+ {
+ var conn = new NpgsqlConnection(ConnectionSource.ConnectionString);
+ await conn.OpenAsync();
+ return conn;
+ }
+
+ private async Task appendEvents(int count, Guid? streamId = null)
+ {
+ await using var session = theStore.LightweightSession();
+ for (var i = 0; i < count; i++)
+ {
+ session.Events.StartStream(Guid.NewGuid(), new Bug4953GapEvent(streamId ?? Guid.NewGuid(), i + 1));
+ }
+
+ await session.SaveChangesAsync();
+ }
+
+ private async Task scalar(string sql)
+ {
+ await using var conn = await openConnection();
+ var raw = await conn.CreateCommand(sql).ExecuteScalarAsync();
+ return raw is long l ? l : Convert.ToInt64(raw ?? 0L);
+ }
+
+ private async Task execute(string sql)
+ {
+ await using var conn = await openConnection();
+ await conn.CreateCommand(sql).ExecuteNonQueryAsync();
+ }
+
+ // A faithful stand-in for a slow concurrent append: reserves the next sequence number and
+ // inserts its event row WITHOUT committing, exactly like an in-flight SaveChanges.
+ private async Task<(NpgsqlConnection conn, NpgsqlTransaction tx, long seq)> startOutstandingAppend()
+ {
+ var conn = await openConnection();
+ var tx = await conn.BeginTransactionAsync();
+ var seq = (long)(await conn.CreateCommand($"select nextval('{Schema}.mt_events_sequence')")
+ .ExecuteScalarAsync())!;
+ await conn.CreateCommand($@"
+insert into {Schema}.mt_events(seq_id, id, stream_id, version, data, type, timestamp, tenant_id, mt_dotnet_type, is_archived)
+select {seq}, gen_random_uuid(), stream_id, 100000 + {seq}, data, type, now(), tenant_id, mt_dotnet_type, false
+from {Schema}.mt_events where seq_id = 1").ExecuteNonQueryAsync();
+ return (conn, tx, seq);
+ }
+
+ private HighWaterDetector buildDetector()
+ {
+ return new HighWaterDetector((MartenDatabase)theStore.Tenancy.Default.Database, theStore.Events,
+ NullLogger.Instance);
+ }
+
+ private static int countStatements(string commandText)
+ {
+ return commandText.Split(';', StringSplitOptions.RemoveEmptyEntries)
+ .Count(s => !string.IsNullOrWhiteSpace(s));
+ }
+
+ #endregion
+
+ // ---------------------------------------------------------------------------------------------
+ // Mechanism 1: the GapDetector command batches multiple statements, and under READ COMMITTED
+ // every statement gets its OWN snapshot. Commits landing between the leading-gap probe, the
+ // interior-gap query, and the max() fallback can defeat all gap checks and let the NORMAL
+ // (silent) path cross an outstanding sequence. The only structural guarantee against snapshot
+ // skew is a single statement.
+ // ---------------------------------------------------------------------------------------------
+
+ [Fact]
+ public void gap_detector_command_is_a_single_statement_for_a_single_snapshot()
+ {
+ StoreOptions(opts => { });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ var command = new GapDetector(theStore.Events).BuildCommand();
+ countStatements(command.CommandText).ShouldBe(1);
+ }
+
+ [Fact]
+ public void statistics_detector_command_is_a_single_statement_for_a_single_snapshot()
+ {
+ StoreOptions(opts => { });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ var command = new HighWaterStatisticsDetector(theStore.Events).BuildCommand();
+ countStatements(command.CommandText).ShouldBe(1);
+ }
+
+ // Static pin (already green): the Normal path must hold before a leading gap held by an open
+ // transaction. Guards the #4964 fix while the command shape changes.
+ [Fact]
+ public async Task normal_detection_holds_before_an_open_transaction_gap()
+ {
+ StoreOptions(opts => { });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ await appendEvents(8);
+ await execute($"select {Schema}.mt_mark_event_progression('HighWaterMark', 8)");
+
+ var (conn, tx, seq) = await startOutstandingAppend();
+ try
+ {
+ seq.ShouldBe(9);
+ await appendEvents(3); // 10..12 committed
+
+ var statistics = await buildDetector().Detect(CancellationToken.None);
+ statistics.CurrentMark.ShouldBe(8);
+ }
+ finally
+ {
+ await conn.DisposeAsync();
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Mechanisms 2 + 4: DetectInSafeZone is invoked with no staleness gating by rebuilds and
+ // catch-up (CheckNowAsync), and by the poll loop after StaleSequenceThreshold. In both cases it
+ // must NOT cross a sequence whose reserving transaction is demonstrably still alive.
+ // ---------------------------------------------------------------------------------------------
+
+ [Fact]
+ public async Task safe_zone_detection_holds_while_the_reserving_transaction_is_alive()
+ {
+ StoreOptions(opts =>
+ {
+ opts.Projections.StaleSequenceThreshold = 500.Milliseconds();
+ });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ await appendEvents(8);
+ await execute($"select {Schema}.mt_mark_event_progression('HighWaterMark', 8)");
+
+ var (conn, tx, seq) = await startOutstandingAppend(); // seq 9, held open
+ try
+ {
+ seq.ShouldBe(9);
+ await appendEvents(3); // 10..12 committed
+
+ var detector = buildDetector();
+
+ // First sighting of the gap — must hold regardless (threshold not elapsed)
+ var first = await detector.DetectInSafeZone(CancellationToken.None);
+ first.CurrentMark.ShouldBe(8);
+
+ // Well past the stale threshold — must STILL hold, because the reserving transaction
+ // is provably alive (RowExclusiveLock on mt_events + open transaction evidence)
+ await Task.Delay(700);
+ var second = await detector.DetectInSafeZone(CancellationToken.None);
+ _output.WriteLine($"After threshold with live reserver: CurrentMark={second.CurrentMark}");
+ second.CurrentMark.ShouldBe(8);
+
+ var persisted = await scalar(
+ $"select coalesce(max(last_seq_id), 0) from {Schema}.mt_event_progression where name = 'HighWaterMark'");
+ persisted.ShouldBe(8);
+ }
+ finally
+ {
+ await conn.DisposeAsync();
+ }
+ }
+
+ [Fact]
+ public async Task safe_zone_detection_skips_a_dead_gap_after_the_threshold_and_records_it()
+ {
+ StoreOptions(opts =>
+ {
+ opts.Projections.StaleSequenceThreshold = 500.Milliseconds();
+ opts.Events.EnableAdvancedAsyncTracking = true;
+ });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ await appendEvents(8);
+ await execute($"select {Schema}.mt_mark_event_progression('HighWaterMark', 8)");
+
+ var (conn, tx, seq) = await startOutstandingAppend(); // seq 9
+ long persistedMark;
+ try
+ {
+ seq.ShouldBe(9);
+ await appendEvents(3); // 10..12 committed
+ await tx.RollbackAsync(); // seq 9 is now a permanently dead hole
+
+ var detector = buildDetector();
+
+ // First sighting — hold (threshold measured from first observation of THIS gap)
+ var first = await detector.DetectInSafeZone(CancellationToken.None);
+ first.CurrentMark.ShouldBe(8);
+
+ await Task.Delay(700);
+
+ // Threshold elapsed + no live reserver = provably dead: skip, and record the skip
+ var second = await detector.DetectInSafeZone(CancellationToken.None);
+ second.CurrentMark.ShouldBe(12);
+ second.IncludesSkipping.ShouldBeTrue();
+
+ persistedMark = await scalar(
+ $"select coalesce(max(last_seq_id), 0) from {Schema}.mt_event_progression where name = 'HighWaterMark'");
+ }
+ finally
+ {
+ await conn.DisposeAsync();
+ }
+
+ persistedMark.ShouldBe(12);
+ var skips = await scalar($"select count(*) from {Schema}.mt_high_water_skips");
+ skips.ShouldBeGreaterThan(0);
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Mechanism 3: idle store, then a sudden crush reserves thousands of sequence numbers before
+ // the first commits land. The stale fallback must not teleport to (reserved last_value - 32):
+ // it must hold while reservers live, and once the tail is provably dead it may advance to the
+ // reserved ceiling recorded when the gap was first observed — all of it, no magic -32.
+ // ---------------------------------------------------------------------------------------------
+
+ [Fact]
+ public async Task stale_fallback_holds_for_live_reservations_then_skips_to_the_observed_ceiling()
+ {
+ StoreOptions(opts =>
+ {
+ opts.Projections.StaleSequenceThreshold = 500.Milliseconds();
+ });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ await appendEvents(8);
+ await execute($"select {Schema}.mt_mark_event_progression('HighWaterMark', 8)");
+ // the store has been idle for an hour — last_updated must NOT satisfy the stale gate
+ await execute(
+ $"update {Schema}.mt_event_progression set last_updated = now() - interval '1 hour' where name = 'HighWaterMark'");
+
+ var (conn, tx, seq) = await startOutstandingAppend(); // seq 9, in flight
+ try
+ {
+ seq.ShouldBe(9);
+ // the crush: 5000 more reservations, none committed yet
+ await execute($"select nextval('{Schema}.mt_events_sequence') from generate_series(1, 5000)");
+ var lastValue = await scalar($"select last_value from {Schema}.mt_events_sequence");
+
+ var detector = buildDetector();
+
+ // Despite the ancient last_updated, the first sighting of this gap must hold
+ var first = await detector.DetectInSafeZone(CancellationToken.None);
+ _output.WriteLine($"First sighting: CurrentMark={first.CurrentMark}");
+ first.CurrentMark.ShouldBe(8);
+
+ // Still holding while the reserving transaction lives, even past the threshold
+ await Task.Delay(700);
+ var second = await detector.DetectInSafeZone(CancellationToken.None);
+ second.CurrentMark.ShouldBe(8);
+
+ // Kill the whole tail: every reservation is now provably dead
+ await tx.RollbackAsync();
+ await Task.Delay(700);
+
+ var third = await detector.DetectInSafeZone(CancellationToken.None);
+ _output.WriteLine($"After tail death: CurrentMark={third.CurrentMark} (ceiling {lastValue})");
+ third.CurrentMark.ShouldBe(lastValue);
+ third.IncludesSkipping.ShouldBeTrue();
+ }
+ finally
+ {
+ await conn.DisposeAsync();
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Mechanism 2 end-to-end: a projection rebuild kicked off while an append is in flight must
+ // wait for that append instead of silently skipping it.
+ // ---------------------------------------------------------------------------------------------
+
+ [Fact]
+ public async Task rebuild_during_an_inflight_append_waits_instead_of_skipping()
+ {
+ StoreOptions(opts =>
+ {
+ opts.Projections.Add(new Bug4953GapViewProjection(), ProjectionLifecycle.Async);
+ });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ var viewId = Guid.NewGuid();
+ await using (var session = theStore.LightweightSession())
+ {
+ for (var i = 0; i < 8; i++)
+ {
+ session.Events.StartStream(Guid.NewGuid(), new Bug4953GapEvent(viewId, i + 1));
+ }
+
+ await session.SaveChangesAsync();
+ }
+
+ var (conn, tx, seq) = await startOutstandingAppend(); // seq 9 in flight
+ try
+ {
+ seq.ShouldBe(9);
+
+ await using (var session = theStore.LightweightSession())
+ {
+ for (var i = 0; i < 3; i++)
+ {
+ session.Events.StartStream(Guid.NewGuid(), new Bug4953GapEvent(viewId, 10 + i));
+ }
+
+ await session.SaveChangesAsync(); // seqs 10..12 committed
+ }
+
+ using var daemon = await StartDaemon();
+
+ // release the in-flight append while the rebuild is waiting on the high water mark
+ var release = Task.Run(async () =>
+ {
+ await Task.Delay(2000);
+ await tx.CommitAsync();
+ });
+
+ await daemon.RebuildProjectionAsync(CancellationToken.None);
+ await release;
+
+ await daemon.WaitForNonStaleData(30.Seconds());
+
+ var persisted = await scalar($"select count(*) from {Schema}.mt_events");
+ await using var query = theStore.QuerySession();
+ var views = await query.Query().ToListAsync();
+ var projectedSequences = views.SelectMany(x => x.Sequences).OrderBy(x => x).ToArray();
+
+ _output.WriteLine($"persisted={persisted}, projected=[{string.Join(",", projectedSequences)}]");
+ projectedSequences.Length.ShouldBe((int)persisted);
+ projectedSequences.ShouldContain(9);
+ }
+ finally
+ {
+ await conn.DisposeAsync();
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Mechanism 4 end-to-end (the reporter's trigger technique): the running daemon's stale skip
+ // must not cross an append transaction that is merely slow — the trigger holds the FIRST event
+ // of the store in flight past the stale threshold while later events commit.
+ // ---------------------------------------------------------------------------------------------
+
+ [Fact]
+ public async Task running_daemon_holds_for_a_slow_append_instead_of_stale_skipping_it()
+ {
+ StoreOptions(opts =>
+ {
+ opts.Projections.StaleSequenceThreshold = 1.Seconds();
+ opts.Projections.Add(new Bug4953GapViewProjection(), ProjectionLifecycle.Async);
+ });
+ theStore.EnsureStorageExists(typeof(IEvent));
+
+ var viewId = Guid.NewGuid();
+
+ await execute($$"""
+ CREATE OR REPLACE FUNCTION {{Schema}}.bug4953_sleep_insert() RETURNS trigger LANGUAGE plpgsql AS $function$
+ BEGIN
+ PERFORM pg_sleep(4);
+ RETURN NEW;
+ END;
+ $function$;
+
+ DROP TRIGGER IF EXISTS bug4953_sleep_insert ON {{Schema}}.mt_events;
+
+ CREATE TRIGGER bug4953_sleep_insert
+ AFTER INSERT ON {{Schema}}.mt_events
+ FOR EACH ROW
+ WHEN (NEW.data ->> 'Number' = '0' OR NEW.data ->> 'number' = '0')
+ EXECUTE FUNCTION {{Schema}}.bug4953_sleep_insert();
+ """);
+
+ try
+ {
+ using var daemon = await StartDaemon();
+
+ // the very first event of the store is slow: its transaction holds seq 1 in flight ~4s
+ var slowAppend = Task.Run(async () =>
+ {
+ await using var session = theStore.LightweightSession();
+ session.Events.StartStream(Guid.NewGuid(), new Bug4953GapEvent(viewId, 0));
+ await session.SaveChangesAsync();
+ });
+
+ await Task.Delay(300);
+
+ // meanwhile 60 later events commit normally
+ for (var i = 0; i < 20; i++)
+ {
+ await using var session = theStore.LightweightSession();
+ session.Events.StartStream(Guid.NewGuid(), new Bug4953GapEvent(viewId, i + 1));
+ session.Events.StartStream(Guid.NewGuid(), new Bug4953GapEvent(viewId, 100 + i));
+ session.Events.StartStream(Guid.NewGuid(), new Bug4953GapEvent(viewId, 200 + i));
+ await session.SaveChangesAsync();
+ }
+
+ await slowAppend;
+
+ await daemon.WaitForNonStaleData(60.Seconds());
+
+ var persisted = await scalar($"select count(*) from {Schema}.mt_events");
+ await using var query = theStore.QuerySession();
+ var views = await query.Query().ToListAsync();
+ var projected = views.SelectMany(x => x.Sequences).Count();
+
+ _output.WriteLine($"persisted={persisted}, projected={projected}");
+ projected.ShouldBe((int)persisted);
+ }
+ finally
+ {
+ await execute($"""
+ DROP TRIGGER IF EXISTS bug4953_sleep_insert ON {Schema}.mt_events;
+ DROP FUNCTION IF EXISTS {Schema}.bug4953_sleep_insert();
+ """);
+ }
+ }
+
+}
+
+public sealed record Bug4953GapEvent(Guid AggregateId, int Number);
+
+public class Bug4953GapView
+{
+ public Guid Id { get; set; }
+ public List Sequences { get; set; } = new();
+}
+
+public partial class Bug4953GapViewProjection: MultiStreamProjection
+{
+ public Bug4953GapViewProjection()
+ {
+ Identity(x => x.AggregateId);
+ }
+
+ public void Apply(Bug4953GapView view, IEvent e)
+ {
+ view.Sequences.Add(e.Sequence);
+ }
+}
diff --git a/src/DaemonTests/Internals/detecting_the_high_water_mark.cs b/src/DaemonTests/Internals/detecting_the_high_water_mark.cs
index 3a5c515271..97edce6756 100644
--- a/src/DaemonTests/Internals/detecting_the_high_water_mark.cs
+++ b/src/DaemonTests/Internals/detecting_the_high_water_mark.cs
@@ -101,7 +101,11 @@ public async Task second_run_detect_same_gap_when_stale(bool useAdvancedTracking
[InlineData(false)]
public async Task starting_from_first_detection_some_gaps_with_nonzero_buffer(bool useAdvancedTracking)
{
- StoreOptions(opts => opts.Events.EnableAdvancedAsyncTracking = useAdvancedTracking);
+ StoreOptions(opts =>
+ {
+ opts.Events.EnableAdvancedAsyncTracking = useAdvancedTracking;
+ opts.Projections.StaleSequenceThreshold = 500.Milliseconds();
+ });
await theStore.EnsureStorageExistsAsync(typeof(IEvent));
NumberOfStreams = 10;
@@ -118,9 +122,18 @@ public async Task starting_from_first_detection_some_gaps_with_nonzero_buffer(bo
statistics.CurrentMark.ShouldBe(NumberOfEvents - 101);
statistics.HighestSequence.ShouldBe(NumberOfEvents);
+ // #4953: a stale gap is no longer skipped on first sighting — the safe zone holds until the
+ // gap has been stuck past StaleSequenceThreshold (measured from when the detector first
+ // observed it) and no live transaction could still fill it
+ var held = await theDetector.DetectInSafeZone(CancellationToken.None);
+ held.CurrentMark.ShouldBe(NumberOfEvents - 101);
+
+ await Task.Delay(700);
+
var statistics2 = await theDetector.DetectInSafeZone(CancellationToken.None);
statistics2.CurrentMark.ShouldBe(NumberOfEvents - 96);
+ statistics2.IncludesSkipping.ShouldBeTrue();
}
[Theory]
@@ -128,7 +141,11 @@ public async Task starting_from_first_detection_some_gaps_with_nonzero_buffer(bo
[InlineData(false)]
public async Task look_for_safe_harbor_time_if_there_are_gaps_between_highest_assigned_event_and_the_sequence(bool useAdvancedTracking)
{
- StoreOptions(opts => opts.Events.EnableAdvancedAsyncTracking = useAdvancedTracking);
+ StoreOptions(opts =>
+ {
+ opts.Events.EnableAdvancedAsyncTracking = useAdvancedTracking;
+ opts.Projections.StaleSequenceThreshold = 500.Milliseconds();
+ });
await theStore.EnsureStorageExistsAsync(typeof(IEvent));
NumberOfStreams = 10;
@@ -136,23 +153,33 @@ public async Task look_for_safe_harbor_time_if_there_are_gaps_between_highest_as
var statistics = await theDetector.Detect(CancellationToken.None);
- await Task.Delay(5.Seconds());
-
- await makeOldWhereSequenceIsLessThanOrEqualTo(NumberOfEvents + 10000);
-
- // Should not move at all.
+ // A tail of reserved-but-never-committed sequence numbers (rolled-back appends)
await advanceSequenceBy(20);
+ // #4953: first sighting of the dead tail — hold, never teleport on sight
var statistics2 = await theDetector.DetectInSafeZone(CancellationToken.None);
statistics2.CurrentMark.ShouldBe(statistics.CurrentMark);
- await advanceSequenceBy(20);
+ await Task.Delay(700);
+ // Stuck past the threshold with no live transaction that could fill the tail: advance to the
+ // reserved ceiling recorded when the gap was first observed — the whole dead tail, no magic -32
var statistics3 = await theDetector.DetectInSafeZone(CancellationToken.None);
- // 20 + 20 - 32 = 8
- statistics3.CurrentMark.ShouldBe(statistics.CurrentMark + 8);
+ statistics3.CurrentMark.ShouldBe(statistics.CurrentMark + 20);
+ statistics3.IncludesSkipping.ShouldBeTrue();
+
+ // A LATER batch of dead reservations needs its own observation cycle
+ await advanceSequenceBy(20);
+
+ var statistics4 = await theDetector.DetectInSafeZone(CancellationToken.None);
+ statistics4.CurrentMark.ShouldBe(statistics.CurrentMark + 20);
+
+ await Task.Delay(700);
+
+ var statistics5 = await theDetector.DetectInSafeZone(CancellationToken.None);
+ statistics5.CurrentMark.ShouldBe(statistics.CurrentMark + 40);
if (useAdvancedTracking)
{
diff --git a/src/Marten/Events/Daemon/HighWater/GapDetector.cs b/src/Marten/Events/Daemon/HighWater/GapDetector.cs
index 0262b57b49..a20dd40a1e 100644
--- a/src/Marten/Events/Daemon/HighWater/GapDetector.cs
+++ b/src/Marten/Events/Daemon/HighWater/GapDetector.cs
@@ -21,7 +21,7 @@ public GapDetector(EventGraph graph)
///
/// #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
+ /// first committed row sits beyond Start + 1. The interior-gap check 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-
@@ -34,19 +34,22 @@ public GapDetector(EventGraph graph)
public NpgsqlCommand BuildCommand()
{
+ // #4953: this MUST stay a single SQL statement. Under READ COMMITTED every statement in a
+ // command takes its own snapshot, so splitting the leading-gap probe, the interior-gap query,
+ // and the max() fallback across statements lets commits that land mid-command defeat the gap
+ // checks — the fallback would then advance the mark over a sequence number whose reserving
+ // transaction is still in flight, silently skipping that event once it commits. One statement
+ // = one snapshot = the three readings are always mutually consistent.
var sql = $@"
-select min(seq_id) from {_graph.DatabaseSchemaName}.mt_events where seq_id > :start;
-select seq_id
-from (select
- seq_id,
- lead(seq_id)
- over (order by seq_id) as no
- from
- {_graph.DatabaseSchemaName}.mt_events where seq_id >= :start) ct
-where no is not null
- and no - seq_id > 1
-LIMIT 1;
-select max(seq_id) from {_graph.DatabaseSchemaName}.mt_events where seq_id >= :start;
+select
+ (select min(seq_id) from {_graph.DatabaseSchemaName}.mt_events where seq_id > :start) as first_after,
+ (select seq_id
+ from (select seq_id, lead(seq_id) over (order by seq_id) as next_seq
+ from {_graph.DatabaseSchemaName}.mt_events where seq_id >= :start) gaps
+ where next_seq is not null and next_seq - seq_id > 1
+ order by seq_id
+ limit 1) as gap_edge,
+ (select max(seq_id) from {_graph.DatabaseSchemaName}.mt_events where seq_id >= :start) as max_seq
".Trim();
var command = new NpgsqlCommand(sql);
command.AddNamedParameter("start", Start);
@@ -56,40 +59,35 @@ and no - seq_id > 1
public async Task HandleAsync(DbDataReader reader, CancellationToken token)
{
- // (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))
+ if (!await reader.ReadAsync(token).ConfigureAwait(false))
{
- firstAfterStart = await reader.GetFieldValueAsync(0, token).ConfigureAwait(false);
+ return null;
}
+ long? firstAfterStart = await reader.IsDBNullAsync(0, token).ConfigureAwait(false)
+ ? null
+ : await reader.GetFieldValueAsync(0, token).ConfigureAwait(false);
+ long? gapEdge = await reader.IsDBNullAsync(1, token).ConfigureAwait(false)
+ ? null
+ : await reader.GetFieldValueAsync(1, token).ConfigureAwait(false);
+ long? maxSeq = await reader.IsDBNullAsync(2, token).ConfigureAwait(false)
+ ? null
+ : await reader.GetFieldValueAsync(2, token).ConfigureAwait(false);
+
+ // (0) Leading-gap hold: the lowest committed sequence strictly above Start sits beyond
+ // Start + 1, so a gap the interior-gap query cannot see lies immediately above Start.
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);
- }
-
- // (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))
- {
- return null;
- }
-
- if (!await reader.IsDBNullAsync(0, token).ConfigureAwait(false))
+ // (1) The first interior sequence gap at or after Start
+ if (gapEdge.HasValue)
{
- return await reader.GetFieldValueAsync(0, token).ConfigureAwait(false);
+ return gapEdge;
}
- return null;
+ // (2) No gap — use the latest committed sequence
+ return maxSeq;
}
}
diff --git a/src/Marten/Events/Daemon/HighWater/GapLivenessProbe.cs b/src/Marten/Events/Daemon/HighWater/GapLivenessProbe.cs
new file mode 100644
index 0000000000..204a603bbb
--- /dev/null
+++ b/src/Marten/Events/Daemon/HighWater/GapLivenessProbe.cs
@@ -0,0 +1,113 @@
+using System;
+using System.Data.Common;
+using System.Threading;
+using System.Threading.Tasks;
+using Marten.Services;
+using Npgsql;
+using Weasel.Postgresql;
+
+namespace Marten.Events.Daemon.HighWater;
+
+///
+/// #4953: evidence that a stale sequence gap might still be filled by a transaction that is alive
+/// right now. A sequence number inside a gap was reserved by SOME transaction before the gap was
+/// first observed; if that transaction is still running the gap is merely OUTSTANDING (its event
+/// will commit later) and must not be skipped. Only when no candidate reserver remains — and the
+/// rows are still absent — is the gap proven permanently dead (the reserver rolled back).
+///
+internal record GapLiveness(long OldLockHolders, long OlderTransactions, long OlderWriteXids)
+{
+ public bool IndicatesLiveReserver => OldLockHolders > 0 || OlderTransactions > 0 || OlderWriteXids > 0;
+
+ public override string ToString()
+ {
+ return
+ $"mt_events write locks held by transactions from before the gap: {OldLockHolders}, open transactions from before the gap: {OlderTransactions}, in-progress write transaction ids from before the gap: {OlderWriteXids}";
+ }
+}
+
+///
+/// Single-statement probe for , fenced to the moment a stuck gap was first
+/// observed (server-side transaction_timestamp + the pg_current_snapshot xmax recorded with it):
+///
+/// 1. pg_locks: granted RowExclusiveLock on the mt_events lineage (parent + partitions) held
+/// by a transaction that began at or before the observation. An in-flight INSERT holds this lock
+/// until commit/abort, and pg_locks is fully visible to every role.
+/// 2. pg_stat_activity: any client-backend transaction in this database that began at or
+/// before the observation. This is the only signal that covers a reserver which called nextval
+/// but has not yet issued its first INSERT (no lock, possibly no xid yet). Cross-role sessions
+/// hide xact_start from unprivileged viewers (rows show query = '<insufficient privilege>'),
+/// so in mixed-role deployments without pg_read_all_stats this clause only sees the daemon's own
+/// role — the common single-role deployment sees everything.
+/// 3. pg_snapshot_xip(pg_current_snapshot()): any in-progress write transaction id below the
+/// xmax recorded at observation. Purely MVCC data — visible regardless of role/privileges — and
+/// covers cross-role writers that the redacted pg_stat_activity clause cannot see.
+///
+/// All three are proc-array/lock-table scans (monitoring-grade cost) and only run while a stale gap
+/// actually exists, at high-water poll cadence.
+///
+internal class GapLivenessProbe: ISingleQueryHandler
+{
+ private readonly EventGraph _graph;
+ private readonly DateTimeOffset _gapFirstObserved;
+ private readonly long _xmaxAtObservation;
+
+ public GapLivenessProbe(EventGraph graph, DateTimeOffset gapFirstObserved, long xmaxAtObservation)
+ {
+ _graph = graph;
+ _gapFirstObserved = gapFirstObserved;
+ _xmaxAtObservation = xmaxAtObservation;
+ }
+
+ public NpgsqlCommand BuildCommand()
+ {
+ var sql = @"
+select
+ (select count(*)
+ from pg_locks l
+ join pg_stat_activity a on a.pid = l.pid
+ where l.locktype = 'relation'
+ and l.granted
+ and l.mode = 'RowExclusiveLock'
+ and l.pid <> pg_backend_pid()
+ and l.database = (select d.oid from pg_database d where d.datname = current_database())
+ and a.xact_start <= :first_observed
+ and l.relation in (select c.oid
+ from pg_class c
+ join pg_namespace n on n.oid = c.relnamespace
+ where n.nspname = :schema
+ and c.relname like 'mt_events%'
+ and c.relkind in ('r', 'p'))) as old_lock_holders,
+ (select count(*)
+ from pg_stat_activity a
+ where a.datname = current_database()
+ and a.pid <> pg_backend_pid()
+ and a.backend_type = 'client backend'
+ and a.xact_start is not null
+ and a.xact_start <= :first_observed) as older_transactions,
+ (select count(*)
+ from pg_snapshot_xip(pg_current_snapshot()) as xip(xid)
+ where xip.xid::text::bigint < :xmax0) as older_write_xids
+".Trim();
+
+ var command = new NpgsqlCommand(sql);
+ command.AddNamedParameter("first_observed", _gapFirstObserved);
+ command.AddNamedParameter("schema", _graph.DatabaseSchemaName);
+ command.AddNamedParameter("xmax0", _xmaxAtObservation);
+
+ return command;
+ }
+
+ public async Task HandleAsync(DbDataReader reader, CancellationToken token)
+ {
+ if (!await reader.ReadAsync(token).ConfigureAwait(false))
+ {
+ return new GapLiveness(0, 0, 0);
+ }
+
+ return new GapLiveness(
+ await reader.GetFieldValueAsync(0, token).ConfigureAwait(false),
+ await reader.GetFieldValueAsync(1, token).ConfigureAwait(false),
+ await reader.GetFieldValueAsync(2, token).ConfigureAwait(false));
+ }
+}
diff --git a/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs b/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs
index b0c2056883..83a19d5d63 100644
--- a/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs
+++ b/src/Marten/Events/Daemon/HighWater/HighWaterDetector.cs
@@ -28,7 +28,6 @@ internal enum DetectionType
internal class HighWaterDetector: IHighWaterDetector
{
- private readonly GapDetector _gapDetector;
private readonly HighWaterStatisticsDetector _highWaterStatisticsDetector;
private readonly ILogger _logger;
private readonly ISingleQueryRunner _runner;
@@ -43,7 +42,23 @@ internal class HighWaterDetector: IHighWaterDetector
// analogue of HighWaterAgent's in-memory `_current` timestamp on the store-global path: the state
// is detector-scoped (one detector instance per running daemon) and resets on restart, which only
// means a stale tenant waits one fresh threshold before skipping — never that events are lost.
- private readonly ConcurrentDictionary _tenantStaleSince = new();
+ // #4953: value carries the snapshot xmax observed when the tenant first went stale so the
+ // liveness probe can fence to transactions that could have reserved the tenant's gap.
+ private readonly ConcurrentDictionary _tenantStaleSince = new();
+
+ private sealed record TenantStaleObservation(DateTimeOffset Since, long Xmax);
+
+ // #4953: bookkeeping for the store-global sequence gap the mark is currently stuck under. Records
+ // WHEN the detector first saw the mark pinned at this position, plus the snapshot xmax and the
+ // highest reserved sequence from that same statistics reading. The stale threshold is measured
+ // from Since (never from mt_event_progression.last_updated, which is arbitrarily old on an idle
+ // store), Xmax fences the liveness probe to transactions that could have reserved the gap, and
+ // ReservedCeiling bounds how far a proven-dead skip may advance — sequence numbers reserved AFTER
+ // the observation belong to newer transactions whose fate is not proven. Detector-scoped state:
+ // resets on restart, which only means a stuck gap waits one fresh threshold before skipping.
+ private sealed record StuckGapObservation(long Mark, DateTimeOffset Since, long Xmax, long ReservedCeiling);
+
+ private StuckGapObservation? _stuckGap;
public HighWaterDetector(MartenDatabase runner, EventGraph graph, ILogger logger)
{
@@ -51,7 +66,6 @@ public HighWaterDetector(MartenDatabase runner, EventGraph graph, ILogger logger
_database = runner;
_graph = graph;
_logger = logger;
- _gapDetector = new GapDetector(graph);
_highWaterStatisticsDetector = new HighWaterStatisticsDetector(graph);
_settings = graph.Options.Projections;
@@ -80,8 +94,51 @@ public HighWaterDetector(MartenDatabase runner, EventGraph graph, ILogger logger
///
public async Task AdvanceHighWaterMarkToLatest(CancellationToken token)
{
+ // #4953: advance to the highest COMMITTED sequence, never the reserved last_value of
+ // mt_events_sequence — reserved numbers can belong to transactions still in flight, and
+ // marking past them permanently skips their events once they commit.
var statistics = await loadCurrentStatistics(token).ConfigureAwait(false);
- await MarkHighWaterMarkInDatabaseAsync(statistics.HighestSequence, token).ConfigureAwait(false);
+ var committed = await _runner.Query(new CommittedSequenceHandler(_graph), token).ConfigureAwait(false);
+ if (committed > statistics.LastMark)
+ {
+ await MarkHighWaterMarkInDatabaseAsync(committed, token).ConfigureAwait(false);
+ }
+ }
+
+ ///
+ /// #4953: the catch-up ceiling HighWaterAgent.CheckNowAsync loops toward (jasperfx#530) — the
+ /// highest COMMITTED sequence (max(seq_id)), never mt_events_sequence.last_value, which includes
+ /// numbers reserved by transactions still in flight.
+ ///
+ public Task FetchCommittedHighWaterCeilingAsync(CancellationToken token)
+ {
+ return _runner.Query(new CommittedSequenceHandler(_graph), token);
+ }
+
+ internal class CommittedSequenceHandler: ISingleQueryHandler
+ {
+ private readonly EventGraph _graph;
+
+ public CommittedSequenceHandler(EventGraph graph)
+ {
+ _graph = graph;
+ }
+
+ public NpgsqlCommand BuildCommand()
+ {
+ return new NpgsqlCommand(
+ $"select coalesce(max(seq_id), 0) from {_graph.DatabaseSchemaName}.mt_events");
+ }
+
+ public async Task HandleAsync(DbDataReader reader, CancellationToken token)
+ {
+ if (await reader.ReadAsync(token).ConfigureAwait(false))
+ {
+ return await reader.GetFieldValueAsync(0, token).ConfigureAwait(false);
+ }
+
+ return 0;
+ }
}
public string DatabaseIdentity { get; }
@@ -90,12 +147,113 @@ public async Task DetectInSafeZone(CancellationToken token)
{
var statistics = await loadCurrentStatistics(token).ConfigureAwait(false);
+ if (!_settings.UseTransactionEvidenceForGapSkipping)
+ {
+ return await detectInSafeZoneLegacy(statistics, token).ConfigureAwait(false);
+ }
+
+ // #4953: this method is reachable with NO staleness gating at all (rebuilds and catch-up call
+ // it through HighWaterAgent.CheckNowAsync), and even the threshold-gated route cannot tell a
+ // permanently dead sequence hole from one still held by an in-flight append. So: advance
+ // normally when possible, and only skip a gap that (a) has been stuck past the stale threshold
+ // measured from when THIS detector first observed it and (b) has no evidence of a live
+ // reserving transaction — or the configured escape-hatch cap has expired.
+
+ // Caught up / empty store: nothing to do. (A pristine store — last_value = 1 with nothing
+ // committed — falls through to the held walk, which finds no rows; trackStuckGap then refuses
+ // to observe it. A store with exactly ONE committed event must NOT be short-circuited here:
+ // its HighestSequence is also 1, but the held walk legitimately advances the mark to 1.)
+ if (statistics.LastMark == statistics.HighestSequence || statistics.HighestSequence == 0)
+ {
+ statistics.CurrentMark = statistics.LastMark;
+ _stuckGap = null;
+ return statistics;
+ }
+
+ // First: the same gap-holding walk the Normal path uses. Contiguous committed progress is
+ // taken as a plain advance — no skipping, no observability noise.
+ var held = await runGapDetectorAsync(statistics.SafeStartMark, true, token).ConfigureAwait(false);
+ if (held.HasValue && held.Value > statistics.LastMark)
+ {
+ statistics.CurrentMark = held.Value;
+ _stuckGap = null;
+ await persistDetectedMarkAsync(statistics, DetectionType.Normal, token).ConfigureAwait(false);
+ return statistics;
+ }
+
+ // The mark is pinned under a gap. Start (or continue) the per-gap clock.
+ statistics.CurrentMark = statistics.LastMark;
+ trackStuckGap(statistics);
+ var observed = _stuckGap;
+ if (observed == null)
+ {
+ return statistics;
+ }
+
+ var age = statistics.Timestamp.Subtract(observed.Since);
+ if (age < _settings.StaleSequenceThreshold)
+ {
+ // Give the gap a chance to fill in before even considering a skip
+ return statistics;
+ }
+
+ var liveness = await _runner
+ .Query(new GapLivenessProbe(_graph, observed.Since, observed.Xmax), token).ConfigureAwait(false);
+ if (liveness.IndicatesLiveReserver)
+ {
+ var cap = _settings.SkipStaleGapsDespiteLiveTransactionsAfter;
+ if (cap == null || age < cap.Value)
+ {
+ if (age > _settings.StaleSequenceThreshold * 5)
+ {
+ _logger.LogWarning(
+ "Daemon high water detection has held before the sequence gap above {Mark} for {Age} because a transaction that may still fill it appears to be alive ({Liveness}). Projections will not advance until it commits, aborts, or the SkipStaleGapsDespiteLiveTransactionsAfter cap expires",
+ observed.Mark, age, liveness);
+ }
+ else
+ {
+ _logger.LogInformation(
+ "Daemon high water detection is holding before the sequence gap above {Mark}: {Liveness}",
+ observed.Mark, liveness);
+ }
+
+ return statistics;
+ }
+
+ _logger.LogWarning(
+ "Daemon high water detection is skipping the sequence gap above {Mark} DESPITE evidence of a live transaction ({Liveness}) because the gap has been stuck for {Age}, past the configured SkipStaleGapsDespiteLiveTransactionsAfter cap of {Cap}. Events committed later inside the skipped range will NOT be projected",
+ observed.Mark, liveness, age, cap);
+ }
+
+ // Proven dead (or cap expired): walk past the gap, but never beyond the reserved ceiling
+ // recorded when the gap was first observed — sequence numbers reserved after that belong to
+ // newer transactions whose fate is not proven.
+ var walk = await runGapDetectorAsync(statistics.SafeStartMark + 1, false, token).ConfigureAwait(false);
+ var target = walk.HasValue ? Math.Min(walk.Value, observed.ReservedCeiling) : observed.ReservedCeiling;
+ if (target <= statistics.LastMark)
+ {
+ return statistics;
+ }
+
+ _logger.LogWarning(
+ "Daemon high water detection is skipping the event sequence range ({Mark}, {Target}] after the gap above {Mark} was stuck for {Age} with no evidence of a live transaction that could still fill it ({Liveness}). Any sequence numbers in that range that never committed were lost to rolled-back appends",
+ observed.Mark, target, observed.Mark, age, liveness);
+
+ statistics.SafeStartMark = target;
+ statistics.CurrentMark = target;
+ _stuckGap = null;
+ await persistDetectedMarkAsync(statistics, DetectionType.SafeZoneSkipping, token).ConfigureAwait(false);
+
+ return statistics;
+ }
+
+ // Pre-#4953 behavior, kept verbatim behind ProjectionOptions.UseTransactionEvidenceForGapSkipping = false
+ private async Task detectInSafeZoneLegacy(HighWaterStatistics statistics,
+ CancellationToken token)
+ {
// 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);
+ var safeSequence = await runGapDetectorAsync(statistics.SafeStartMark + 1, false, token).ConfigureAwait(false);
if (safeSequence.HasValue)
{
_logger.LogInformation(
@@ -136,6 +294,57 @@ public async Task DetectInSafeZone(CancellationToken token)
return statistics;
}
+ // #4953: one GapDetector instance per call — Detect (the poll loop) and DetectInSafeZone
+ // (rebuild/catch-up) can run concurrently, and shared mutable Start/Hold state between them was
+ // a race. Retains the long-standing open-data-reader retry.
+ private async Task runGapDetectorAsync(long start, bool holdBeforeLeadingGap, CancellationToken token)
+ {
+ var gapDetector = new GapDetector(_graph) { Start = start, HoldBeforeLeadingGap = holdBeforeLeadingGap };
+
+ try
+ {
+ return await _runner.Query(gapDetector, token).ConfigureAwait(false);
+ }
+ catch (InvalidOperationException e)
+ {
+ if (e.Message.Contains("An open data reader exists for this command"))
+ {
+ await Task.Delay(250.Milliseconds(), token).ConfigureAwait(false);
+ return await _runner.Query(gapDetector, token).ConfigureAwait(false);
+ }
+
+ throw;
+ }
+ }
+
+ // #4953: per-gap observation bookkeeping — see StuckGapObservation
+ private void trackStuckGap(HighWaterStatistics statistics)
+ {
+ if (statistics.CurrentMark >= statistics.HighestSequence || statistics.CurrentMark > statistics.LastMark)
+ {
+ // caught up, or the mark advanced — whatever gap existed before is resolved
+ _stuckGap = null;
+ return;
+ }
+
+ if (statistics.CurrentMark == 0 && statistics.HighestSequence <= 1)
+ {
+ // pristine store (Postgres sequences report last_value = 1 before first use)
+ _stuckGap = null;
+ return;
+ }
+
+ var current = _stuckGap;
+ if (current == null || current.Mark != statistics.CurrentMark)
+ {
+ _stuckGap = new StuckGapObservation(
+ statistics.CurrentMark,
+ statistics.Timestamp,
+ (statistics as MartenHighWaterStatistics)?.CurrentXmax ?? 0,
+ statistics.HighestSequence);
+ }
+ }
+
public async Task Detect(CancellationToken token)
{
@@ -143,6 +352,10 @@ public async Task Detect(CancellationToken token)
await calculateHighWaterMark(statistics, DetectionType.Normal, token).ConfigureAwait(false);
+ // #4953: the poll loop is where a stuck gap is usually seen first — record it here so the
+ // per-gap stale clock starts as early as possible
+ trackStuckGap(statistics);
+
return statistics;
}
@@ -241,7 +454,8 @@ with inputs(tenant_id) as (select unnest(:tenants))
coalesce(prog.last_seq_id, 0) as last_seq_id,
prog.last_updated as last_updated,
transaction_timestamp() as ""timestamp"",
- walk.current_bound as current_bound
+ walk.current_bound as current_bound,
+ pg_snapshot_xmax(pg_current_snapshot())::text::bigint as current_xmax
from inputs i
left join lateral (
select max(e.seq_id) as max_seq_id
@@ -294,14 +508,16 @@ and coalesce(ev.max_seq_id, 0) > prog.last_seq_id
long? currentBound = await reader.IsDBNullAsync(5, token).ConfigureAwait(false)
? null
: await reader.GetFieldValueAsync(5, token).ConfigureAwait(false);
+ var currentXmax = await reader.GetFieldValueAsync(6, token).ConfigureAwait(false);
- var statistics = new HighWaterStatistics
+ var statistics = new MartenHighWaterStatistics
{
TenantId = tenantId,
HighestSequence = lastValue,
LastMark = lastSeqId,
LastUpdated = lastUpdated,
- Timestamp = timestamp
+ Timestamp = timestamp,
+ CurrentXmax = currentXmax
};
// CurrentMark = the latest sequence the daemon may treat as "caught up".
@@ -337,8 +553,9 @@ and coalesce(ev.max_seq_id, 0) > prog.last_seq_id
// once the tenant has been stuck past StaleSequenceThreshold, skip the gap
// below — the per-tenant mirror of the store-global DetectInSafeZone wait.
statistics.CurrentMark = statistics.SafeStartMark = lastSeqId;
- var staleSince = _tenantStaleSince.GetOrAdd(tenantId, timestamp);
- if (timestamp.Subtract(staleSince) > _settings.StaleSequenceThreshold)
+ var observation = _tenantStaleSince.GetOrAdd(tenantId,
+ new TenantStaleObservation(timestamp, currentXmax));
+ if (timestamp.Subtract(observation.Since) > _settings.StaleSequenceThreshold)
{
staleTenants.Add(statistics);
}
@@ -386,6 +603,34 @@ order by g.seq_id
where e.tenant_id = :tenant
and e.seq_id > :mark));";
+ // #4953: same evidence rule as the store-global DetectInSafeZone — a tenant gap whose
+ // reserving transaction may still be alive is outstanding, not dead, and must not be skipped.
+ // The lock/transaction/xip fencing is store-wide rather than per-tenant (a lock on any
+ // mt_events partition blocks every tenant's skip), which errs on the conservative side.
+ if (_settings.UseTransactionEvidenceForGapSkipping
+ && _tenantStaleSince.TryGetValue(statistics.TenantId!, out var observation))
+ {
+ var age = statistics.Timestamp.Subtract(observation.Since);
+ var liveness = await _runner
+ .Query(new GapLivenessProbe(_graph, observation.Since, observation.Xmax), token)
+ .ConfigureAwait(false);
+ if (liveness.IndicatesLiveReserver)
+ {
+ var cap = _settings.SkipStaleGapsDespiteLiveTransactionsAfter;
+ if (cap == null || age < cap.Value)
+ {
+ _logger.LogInformation(
+ "Daemon high water detection for tenant {TenantId} is holding before the sequence gap above {Mark}: {Liveness}",
+ statistics.TenantId, statistics.LastMark, liveness);
+ return;
+ }
+
+ _logger.LogWarning(
+ "Daemon high water detection for tenant {TenantId} is skipping the sequence gap above {Mark} DESPITE evidence of a live transaction ({Liveness}) because the gap has been stuck for {Age}, past the configured SkipStaleGapsDespiteLiveTransactionsAfter cap of {Cap}",
+ statistics.TenantId, statistics.LastMark, liveness, age, cap);
+ }
+ }
+
await using var cmd = conn.CreateCommand(sql)
.With("tenant", statistics.TenantId!)
.With("mark", statistics.LastMark);
@@ -396,9 +641,9 @@ order by g.seq_id
return;
}
- _logger.LogInformation(
- "Daemon projection high water detection for tenant {TenantId} skipping a gap in the event sequence after being stale past the {Threshold} threshold, determined that the 'safe harbor' sequence is at {SafeHarborSequence}",
- statistics.TenantId, _settings.StaleSequenceThreshold, safeSequence);
+ _logger.LogWarning(
+ "Daemon projection high water detection for tenant {TenantId} is skipping the event sequence range ({Mark}, {SafeHarborSequence}] after being stale past the {Threshold} threshold. Any sequence numbers in that range that never committed were lost to rolled-back appends",
+ statistics.TenantId, statistics.LastMark, safeSequence, _settings.StaleSequenceThreshold);
statistics.CurrentMark = statistics.SafeStartMark = safeSequence;
statistics.IncludesSkipping = true;
@@ -424,41 +669,49 @@ private async Task calculateHighWaterMark(HighWaterStatistics statistics, Detect
statistics.CurrentMark = await findCurrentMark(statistics, detectionType, token).ConfigureAwait(false);
}
- if (statistics.HasChanged)
+ await persistDetectedMarkAsync(statistics, detectionType, token).ConfigureAwait(false);
+ }
+
+ private async Task persistDetectedMarkAsync(HighWaterStatistics statistics, DetectionType detectionType,
+ CancellationToken token)
+ {
+ if (!statistics.HasChanged)
{
- var currentMark = statistics.CurrentMark;
+ return;
+ }
- if (detectionType == DetectionType.SafeZoneSkipping)
+ var currentMark = statistics.CurrentMark;
+
+ if (detectionType == DetectionType.SafeZoneSkipping)
+ {
+ if (_graph.EnableAdvancedAsyncTracking)
{
- if (_graph.EnableAdvancedAsyncTracking)
+ var actual = await TryMarkHighWaterSkippingAsync(currentMark, statistics.LastMark, token).ConfigureAwait(false);
+ if (actual == currentMark)
{
- var actual = await TryMarkHighWaterSkippingAsync(currentMark, statistics.LastMark, token).ConfigureAwait(false);
- if (actual == currentMark)
- {
- statistics.IncludesSkipping = true;
- }
- else
- {
- statistics.CurrentMark = actual;
- statistics.IncludesSkipping = false;
- }
+ statistics.IncludesSkipping = true;
}
else
{
- statistics.IncludesSkipping = true;
- await MarkHighWaterMarkInDatabaseAsync(currentMark, token).ConfigureAwait(false);
+ statistics.CurrentMark = actual;
+ statistics.IncludesSkipping = false;
}
}
else
{
+ statistics.IncludesSkipping = true;
await MarkHighWaterMarkInDatabaseAsync(currentMark, token).ConfigureAwait(false);
}
+ }
+ else
+ {
+ await MarkHighWaterMarkInDatabaseAsync(currentMark, token).ConfigureAwait(false);
+ }
- if (!statistics.LastUpdated.HasValue)
- {
- var current = await loadCurrentStatistics(token).ConfigureAwait(false);
- statistics.LastUpdated = current.LastUpdated;
- }
+ if (!statistics.LastUpdated.HasValue)
+ {
+ var current = await loadCurrentStatistics(token).ConfigureAwait(false);
+ statistics.LastUpdated = current.LastUpdated;
}
}
@@ -593,30 +846,10 @@ private async Task loadCurrentStatistics(CancellationToken
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
- {
- current = await _runner.Query(_gapDetector, token).ConfigureAwait(false);
- }
- catch (InvalidOperationException e)
- {
- if (e.Message.Contains("An open data reader exists for this command"))
- {
- await Task.Delay(250.Milliseconds(), token).ConfigureAwait(false);
- current = await _runner.Query(_gapDetector, token).ConfigureAwait(false);
- }
- else
- {
- throw;
- }
- }
+ var current = await runGapDetectorAsync(statistics.SafeStartMark,
+ detectionType == DetectionType.Normal, token).ConfigureAwait(false);
if (current.HasValue)
{
diff --git a/src/Marten/Events/Daemon/HighWater/HighWaterStatisticsDetector.cs b/src/Marten/Events/Daemon/HighWater/HighWaterStatisticsDetector.cs
index beb16e57fc..b6dabcb008 100644
--- a/src/Marten/Events/Daemon/HighWater/HighWaterStatisticsDetector.cs
+++ b/src/Marten/Events/Daemon/HighWater/HighWaterStatisticsDetector.cs
@@ -8,6 +8,18 @@
namespace Marten.Events.Daemon.HighWater;
+///
+/// #4953: Marten-side extension of the JasperFx statistics reading that also carries the xmax of the
+/// database snapshot the statistics were read under. The high-water detector records this when it
+/// first observes a stuck sequence gap: any in-progress write transaction with an xid below this
+/// value existed when the gap was observed and is therefore a candidate reserver of the gap — the
+/// gap cannot be treated as permanently dead while any such transaction is still running.
+///
+internal class MartenHighWaterStatistics: HighWaterStatistics
+{
+ public long CurrentXmax { get; set; }
+}
+
internal class HighWaterStatisticsDetector: ISingleQueryHandler
{
private readonly string _commandText;
@@ -20,19 +32,27 @@ public HighWaterStatisticsDetector(EventGraph graph)
// far higher. That made the store-global high-water agent treat the store as perpetually
// Stale. Read the real height from mt_events instead (mirrors FetchHighestEventSequenceNumber).
var highestSequenceSql = graph.UseTenantPartitionedEvents
- ? $"select coalesce(max(seq_id), 0), transaction_timestamp() from {graph.DatabaseSchemaName}.mt_events"
- : $"select last_value, transaction_timestamp() from {graph.DatabaseSchemaName}.mt_events_sequence";
+ ? $"(select coalesce(max(seq_id), 0) from {graph.DatabaseSchemaName}.mt_events)"
+ : $"(select last_value from {graph.DatabaseSchemaName}.mt_events_sequence)";
- // #4712: stamp Timestamp from THIS first result, which always returns exactly one row. Reading
- // the timestamp off the second (mt_event_progression) result left Timestamp at
- // default(DateTimeOffset) = 0001-01-01 whenever the store-global 'HighWaterMark' progression
- // row was absent — and that default is exactly what produced the bogus SafeHarborTime
- // (0001-01-01 + 3s threshold) that turned the gap-skip into a no-op and hung composite rebuilds.
+ // #4953: a single statement so every reading comes from ONE snapshot (see GapDetector for the
+ // multi-statement snapshot-skew hazard this rules out). The LEFT JOIN from a one-row VALUES
+ // clause preserves the #4712 guarantee that exactly one row always comes back with a real
+ // Timestamp even when the store-global progression row does not exist yet — the missing-row
+ // default(DateTimeOffset) Timestamp was what produced the bogus 0001-01-01 SafeHarborTime.
+ // pg_snapshot_xmax feeds the #4953 outstanding-transaction fencing (see MartenHighWaterStatistics).
// #4681: the literal 'HighWaterMark' name is produced by HighWaterShardIdentity so any future
// change to the grammar lands in one place rather than scattered SQL string literals.
_commandText = $@"
-{highestSequenceSql};
-select last_seq_id, last_updated from {graph.DatabaseSchemaName}.mt_event_progression where name = '{HighWaterShardIdentity.StoreGlobal}';
+select
+ {highestSequenceSql} as highest_sequence,
+ transaction_timestamp() as ""timestamp"",
+ pg_snapshot_xmax(pg_current_snapshot())::text::bigint as current_xmax,
+ p.last_seq_id,
+ p.last_updated
+from (values (1)) as one(x)
+left join {graph.DatabaseSchemaName}.mt_event_progression p
+ on p.name = '{HighWaterShardIdentity.StoreGlobal}'
".Trim();
}
@@ -43,25 +63,24 @@ public NpgsqlCommand BuildCommand()
public async Task HandleAsync(DbDataReader reader, CancellationToken token)
{
- var statistics = new HighWaterStatistics();
+ var statistics = new MartenHighWaterStatistics();
- if (await reader.ReadAsync(token).ConfigureAwait(false))
+ if (!await reader.ReadAsync(token).ConfigureAwait(false))
{
- statistics.HighestSequence = await reader.GetFieldValueAsync(0, token).ConfigureAwait(false);
- statistics.Timestamp = await reader.GetFieldValueAsync(1, token).ConfigureAwait(false);
+ return statistics;
}
- await reader.NextResultAsync(token).ConfigureAwait(false);
+ statistics.HighestSequence = await reader.GetFieldValueAsync(0, token).ConfigureAwait(false);
+ statistics.Timestamp = await reader.GetFieldValueAsync(1, token).ConfigureAwait(false);
+ statistics.CurrentXmax = await reader.GetFieldValueAsync(2, token).ConfigureAwait(false);
- if (!await reader.ReadAsync(token).ConfigureAwait(false))
+ if (!await reader.IsDBNullAsync(3, token).ConfigureAwait(false))
{
- return statistics;
+ statistics.LastMark = statistics.SafeStartMark =
+ await reader.GetFieldValueAsync(3, token).ConfigureAwait(false);
+ statistics.LastUpdated = await reader.GetFieldValueAsync(4, token).ConfigureAwait(false);
}
- statistics.LastMark = statistics.SafeStartMark =
- await reader.GetFieldValueAsync(0, token).ConfigureAwait(false);
- statistics.LastUpdated = await reader.GetFieldValueAsync(1, token).ConfigureAwait(false);
-
return statistics;
}
}
diff --git a/src/Marten/Events/Projections/ProjectionOptions.cs b/src/Marten/Events/Projections/ProjectionOptions.cs
index e18274672f..44bf95def2 100644
--- a/src/Marten/Events/Projections/ProjectionOptions.cs
+++ b/src/Marten/Events/Projections/ProjectionOptions.cs
@@ -51,6 +51,25 @@ internal ProjectionOptions(StoreOptions options): base(options.EventGraph, "mart
///
internal List FetchPlanners { get; } = new();
+ ///
+ /// #4953: when true (the default), the async daemon's high water detection consults PostgreSQL
+ /// transaction evidence (pg_locks on the mt_events tables, open transactions in pg_stat_activity,
+ /// and pg_current_snapshot's in-progress list) before skipping a stale sequence gap, and holds
+ /// while a transaction that could still fill the gap appears to be alive — a slow concurrent
+ /// append is then waited out instead of having its events silently skipped. Set false to restore
+ /// the pre-#4953 wall-clock-only skipping.
+ ///
+ public bool UseTransactionEvidenceForGapSkipping { get; set; } = true;
+
+ ///
+ /// #4953: escape hatch for the transaction-evidence hold above. When set, a stale sequence gap is
+ /// skipped once it has been stuck this long even if evidence says a transaction that could fill it
+ /// is still alive — trading potential event loss for guaranteed projection progress (e.g. against
+ /// leaked idle-in-transaction sessions). Null (the default) never knowingly skips past a live
+ /// appender; PostgreSQL's own idle_in_transaction_session_timeout is the recommended backstop.
+ ///
+ public TimeSpan? SkipStaleGapsDespiteLiveTransactionsAfter { get; set; }
+
///
/// Opt into a performance optimization that directs Marten to always use the identity map for an
/// Inline single stream projection's aggregate type when FetchForWriting() is called. Default is false.