diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapAgeReleaseWhileRunningTest.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapAgeReleaseWhileRunningTest.cs new file mode 100644 index 00000000..31e65c53 --- /dev/null +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapAgeReleaseWhileRunningTest.cs @@ -0,0 +1,47 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using Eventuous.Postgresql.Subscriptions; +using Eventuous.Tests.Persistence.Base.Fixtures; +using Eventuous.Tests.Subscriptions.Base; + +namespace Eventuous.Tests.Postgres.Subscriptions; + +[NotInParallel] +public class GapAgeReleaseWhileRunningTest() : SubscriptionTestBase(Fixture) { + static readonly TombstonesFixture Fixture = new(ConfigureOptions); + + const int AgeThresholdMs = 2000; + + /// + /// The age threshold has to release a gap that was first detected while the event following it was still + /// young, not only one that was already old when the subscription started. With no skip timeout and no + /// remediation configured — the defaults — it is the only thing that ever lets the subscription past a + /// position no transaction will fill. + /// + [Test] + public async Task ShouldReleaseGapThatAgesOutWhileSubscribed(CancellationToken cancellationToken) { + await Fixture.ArrangePermanentGap(new("test-stream-gap-ages-out")); + + await Fixture.StartSubscription(); + + // The event after the gap is younger than the threshold, so the subscription holds at the gap + await Task.Delay(AgeThresholdMs / 4, cancellationToken); + await Assert.That(Fixture.Handler.Handled.Count).IsEqualTo(2); + + // Once that event is older than GapAgeThresholdMs the gap is abandoned and the rest is handled + var handled = await Fixture.WaitForHandled(5, TimeSpan.FromSeconds(10), cancellationToken); + await Assert.That(handled).IsEqualTo(5); + + await Fixture.StopSubscription(); + + var tombstonesCount = await Fixture.CountTombstones(); + await Assert.That(tombstonesCount).IsEqualTo(0); + } + + static void ConfigureOptions(PostgresAllStreamSubscriptionOptions options) { + options.GapAgeThresholdMs = AgeThresholdMs; + options.GapSkipTimeoutMs = null; // the default: never abandon a position on elapsed time alone + options.GapHandlingTimeoutMs = null; // the default: no remediation + } +} diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapRemediationBeforeSkipTest.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapRemediationBeforeSkipTest.cs new file mode 100644 index 00000000..45b69ca2 --- /dev/null +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapRemediationBeforeSkipTest.cs @@ -0,0 +1,46 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using Eventuous.Postgresql.Subscriptions; +using Eventuous.Tests.Persistence.Base.Fixtures; +using Eventuous.Tests.Subscriptions.Base; + +namespace Eventuous.Tests.Postgres.Subscriptions; + +[NotInParallel] +public class GapRemediationBeforeSkipTest() : SubscriptionTestBase(Fixture) { + static readonly TombstonesFixture Fixture = new(ConfigureOptions); + + const int TimeoutMs = 2000; + + /// + /// Both timeouts expire on the same poll, so remediation only runs if it takes precedence over the skip. + /// The tombstone resolves the position safely — it conflicts with a committed row and blocks on an + /// in-flight one — where skipping abandons it on elapsed time alone. + /// + [Test] + public async Task ShouldCreateTombstoneBeforeSkippingGap(CancellationToken cancellationToken) { + await Fixture.ArrangePermanentGap(new("test-stream-gap-remediation")); + + await Fixture.StartSubscription(); + + // Well inside both timeouts: nothing has been remediated or skipped yet + await Task.Delay(TimeoutMs / 4, cancellationToken); + await Assert.That(Fixture.Handler.Handled.Count).IsEqualTo(2); + await Assert.That(await Fixture.CountTombstones()).IsEqualTo(0); + + var handled = await Fixture.WaitForHandled(5, TimeSpan.FromSeconds(10), cancellationToken); + await Assert.That(handled).IsEqualTo(5); + + await Fixture.StopSubscription(); + + var tombstonesCount = await Fixture.CountTombstones(); + await Assert.That(tombstonesCount).IsEqualTo(1); + } + + static void ConfigureOptions(PostgresAllStreamSubscriptionOptions options) { + options.GapHandlingTimeoutMs = TimeoutMs; + options.GapSkipTimeoutMs = TimeoutMs; + options.GapAgeThresholdMs = null; // the gap must not be released by age + } +} diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapSkipTimeoutTest.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapSkipTimeoutTest.cs new file mode 100644 index 00000000..2cc308a8 --- /dev/null +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapSkipTimeoutTest.cs @@ -0,0 +1,45 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using Eventuous.Postgresql.Subscriptions; +using Eventuous.Tests.Persistence.Base.Fixtures; +using Eventuous.Tests.Subscriptions.Base; + +namespace Eventuous.Tests.Postgres.Subscriptions; + +[NotInParallel] +public class GapSkipTimeoutTest() : SubscriptionTestBase(Fixture) { + static readonly TombstonesFixture Fixture = new(ConfigureOptions); + + const int SkipTimeoutMs = 2000; + + /// + /// The skip timeout has to both hold the subscription for its configured duration and then let it past, + /// so this asserts the hold before asserting the release: releasing immediately would satisfy the second + /// assertion on its own. + /// + [Test] + public async Task ShouldSkipGapOnlyAfterSkipTimeout(CancellationToken cancellationToken) { + await Fixture.ArrangePermanentGap(new("test-stream-gap-skip")); + + await Fixture.StartSubscription(); + + // Well inside the timeout, so the subscription is still holding at the gap + await Task.Delay(SkipTimeoutMs / 4, cancellationToken); + await Assert.That(Fixture.Handler.Handled.Count).IsEqualTo(2); + + var handled = await Fixture.WaitForHandled(5, TimeSpan.FromSeconds(10), cancellationToken); + await Assert.That(handled).IsEqualTo(5); + + await Fixture.StopSubscription(); + + var tombstonesCount = await Fixture.CountTombstones(); + await Assert.That(tombstonesCount).IsEqualTo(0); + } + + static void ConfigureOptions(PostgresAllStreamSubscriptionOptions options) { + options.GapSkipTimeoutMs = SkipTimeoutMs; + options.GapAgeThresholdMs = null; // the gap must not be released by age + options.GapHandlingTimeoutMs = null; // no tombstones + } +} diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesFixture.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesFixture.cs index 2112c7be..cb676cd3 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesFixture.cs +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesFixture.cs @@ -5,6 +5,7 @@ using Eventuous.Postgresql.Extensions; using Eventuous.Postgresql.Subscriptions; using Eventuous.Sql.Base; +using Eventuous.Tests.Persistence.Base.Fixtures; using Eventuous.Tests.Subscriptions.Base; namespace Eventuous.Tests.Postgres.Subscriptions; @@ -17,6 +18,30 @@ Action configureOptions protected internal new ValueTask StartSubscription() => base.StartSubscription(); protected internal new ValueTask StopSubscription() => base.StopSubscription(); + /// + /// Two events, then a global position burnt by a rolled back append, then three more events. The burnt + /// position is never filled, so only gap handling decides whether the last three are ever seen. + /// + public async Task ArrangePermanentGap(StreamName streamName) { + await this.AppendEvents(streamName, [.. this.CreateEvents(2)], ExpectedStreamVersion.NoStream); + await InsertGap(streamName, 1); + await this.AppendEvents(streamName, [.. this.CreateEvents(3)], ExpectedStreamVersion.Any); + } + + /// + /// Polls the handled messages until arrive or the timeout expires, and returns + /// how many actually arrived so the caller can assert on it. + /// + public async Task WaitForHandled(int count, TimeSpan timeout, CancellationToken cancellationToken) { + var deadline = DateTime.UtcNow + timeout; + + while (Handler.Handled.Count < count && DateTime.UtcNow < deadline) { + await Task.Delay(50, cancellationToken); + } + + return Handler.Handled.Count; + } + public async Task InsertGap(StreamName streamName, int expectedVersion) { await using var conn = await DataSource.OpenConnectionAsync(); await using var tx = await conn.BeginTransactionAsync(); diff --git a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs index d4655bef..a1d5f39b 100644 --- a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs +++ b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs @@ -74,7 +74,7 @@ public abstract class SqlSubscriptionBase( // ReSharper disable once CognitiveComplexity - private record DetectedGap(long Position, DateTime FirstSeen); + private record DetectedGap(long Position, DateTime FirstSeen, bool RemediationAttempted = false); /// /// The polling loop. Its only clean exit is a stop request; every other exit is a fault the pump in @@ -130,6 +130,7 @@ async Task PollOnce() { if (gapAge.TotalMilliseconds >= Options.GapHandlingTimeoutMs.Value) { await HandleGapTimeout(gap.Position, start, cancellationToken).NoContext(); + gap = gap with { RemediationAttempted = true }; } } @@ -180,21 +181,30 @@ async Task ExecutePollCycle() { DetectedGap? DetectGap(long start, PersistedEvent persistedEvent, DetectedGap? previousGap) { var expectedNext = start < 0 ? 1 : start + 1; // global position identity starts at 1 - if (persistedEvent.GlobalPosition > expectedNext) { - if (previousGap != null) { - if (Options.GapSkipTimeoutMs == null || (DateTime.UtcNow - previousGap.FirstSeen) < TimeSpan.FromMilliseconds(Options.GapSkipTimeoutMs.Value)) { - return previousGap; - } - } + if (persistedEvent.GlobalPosition <= expectedNext) return null; + + // Evaluated on every poll rather than only when the gap is first seen, and ahead of the held-gap + // branch below, which returns without reaching it. A gap whose following event has aged past the + // threshold is abandoned however long it has been held; with neither timeout configured this is the + // only thing that releases a position no transaction will ever fill. + if (Options.GapAgeThresholdMs != null + && (DateTime.UtcNow - persistedEvent.Created).TotalMilliseconds >= Options.GapAgeThresholdMs.Value) { + return null; + } - var newGapAge = DateTime.UtcNow - persistedEvent.Created; + // The gap we are already holding. Keep the original FirstSeen, so the skip timeout can actually expire: + // reporting it as a new gap would restart the timer on every poll and hold the subscription forever + // on a position no transaction will ever fill (a rolled back append still consumes the sequence value). + if (previousGap?.Position == expectedNext) { + if (Options.GapSkipTimeoutMs == null) return previousGap; - if (Options.GapAgeThresholdMs == null || newGapAge.TotalMilliseconds < Options.GapAgeThresholdMs.Value) { - return new(expectedNext, DateTime.UtcNow); - } + if (DateTime.UtcNow - previousGap.FirstSeen < TimeSpan.FromMilliseconds(Options.GapSkipTimeoutMs.Value)) return previousGap; + + // Remediation resolves the position safely, unlike skipping it, so it gets its chance first. + return Options.GapHandlingTimeoutMs != null && !previousGap.RemediationAttempted ? previousGap : null; } - return null; + return new DetectedGap(expectedNext, DateTime.UtcNow); } /// diff --git a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionOptionsBase.cs b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionOptionsBase.cs index 4148e5d2..655aae81 100644 --- a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionOptionsBase.cs +++ b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionOptionsBase.cs @@ -31,22 +31,28 @@ public abstract record SqlSubscriptionOptionsBase : SubscriptionWithCheckpointOp public RetryOptions Retry { get; set; } = new(); /// - /// Gap age threshold in milliseconds. If != null, gaps older than this threshold will be ignored entirely, - /// allowing the subscription to skip past old missing positions. This is useful for scenarios where - /// old tombstones may have been deleted and should not be recreated during replay. Default is 1 hour. + /// Gap age threshold in milliseconds. If != null, a gap is ignored entirely when the event that follows it is + /// older than this threshold, so replaying history doesn't wait for transactions that finished long ago. + /// It's also what lets a subscription eventually move past a position no transaction will ever fill. + /// Default is 1 hour. /// public int? GapAgeThresholdMs { get; set; } = 60 * 60 * 1000; /// - /// Gap skip timeout in milliseconds. If != null, a detected gap will only hold the subscription from - /// advancing for this duration. Default value is 5 sec. + /// Gap skip timeout in milliseconds. If != null, a detected gap stops holding the subscription after this + /// duration, and the subscription advances past the missing position. The position is abandoned on elapsed + /// time alone, so an append taking longer than this to commit will have its event skipped. Prefer + /// , which resolves a gap without that risk and takes precedence over this + /// timeout; set this only when advancing matters more than never missing an event. + /// Default is null (never abandon a position on time alone). /// - public int? GapSkipTimeoutMs { get; set; } = 5000; + public int? GapSkipTimeoutMs { get; set; } /// /// Gap handling timeout in milliseconds. If != null, when a gap in the global position sequence is detected /// and it persists for at least this duration, the subscription will attempt to handle it in a provider-specific - /// way (e.g. creating tombstones). Default is null (don't create tombstones). + /// way (e.g. creating tombstones). Unlike , this resolves the position rather than + /// abandoning it, and so runs before any skip. Default is null (don't create tombstones). /// public int? GapHandlingTimeoutMs { get; set; }