Skip to content

fix(subscriptions): release relational gaps once the skip timeout expires #588

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexeyzimarev merged 3 commits into from
Aug 27, 2026
Merged

fix(subscriptions): release relational gaps once the skip timeout expires #588

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b74e9e7
fix(subscriptions): release relational gaps once the skip timeout exp…
alexeyzimarev Aug 27, 2026
81d7a80
fix(subscriptions): resolve relational gaps before abandoning them
alexeyzimarev Aug 27, 2026
457919f
fix(subscriptions): apply the gap age threshold to a gap already held
alexeyzimarev Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapSkipTimeoutTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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);

[Test]
public async Task ShouldSkipGapAfterSkipTimeout(CancellationToken cancellationToken) {
var streamName = new StreamName("test-stream-gap-skip");

await Fixture.AppendEvents(streamName, [.. Fixture.CreateEvents(2)], ExpectedStreamVersion.NoStream);

// The rolled back append burnt a global position that no transaction will ever fill,
// so the gap can only be released by the skip timeout.
await Fixture.InsertGap(streamName, 1);

await Fixture.AppendEvents(streamName, [.. Fixture.CreateEvents(3)], ExpectedStreamVersion.Any);

await Fixture.StartSubscription();

await Fixture.Handler.AssertThat()
.Timebox(TimeSpan.FromSeconds(10))
.Exactly(5)
.Match(_ => true)
.Validate(cancellationToken);

await Fixture.StopSubscription();

var tombstonesCount = await Fixture.CountTombstones();
await Assert.That(tombstonesCount).IsEqualTo(0);
}

static void ConfigureOptions(PostgresAllStreamSubscriptionOptions options) {
options.GapSkipTimeoutMs = 500; // the gap must hold the subscription for this long, then be skipped
options.GapAgeThresholdMs = null; // never release a gap by age, so the skip timeout is the only way out
options.GapHandlingTimeoutMs = null; // no tombstones
}
}
23 changes: 12 additions & 11 deletions src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,21 +180,22 @@ 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;

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);
}
return DateTime.UtcNow - previousGap.FirstSeen < TimeSpan.FromMilliseconds(Options.GapSkipTimeoutMs.Value) ? previousGap : null;
Comment thread
qodo-free-for-open-source-projects[bot] marked this conversation as resolved.
Outdated

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Run gap remediation before clearing an expired gap

When PostgreSQL sets GapHandlingTimeoutMs equal to the default GapSkipTimeoutMs (or sufficiently close that one poll crosses both thresholds), this returns null as soon as the skip timeout expires. PollOnce subsequently calls HandleGapTimeout only when gap != null, so the configured tombstone is never created even though the gap persisted for the handling duration; later subscriptions must each wait for and skip the same permanent gap. Perform the timeout remediation before clearing the expired gap, or otherwise preserve enough state to invoke it.

Useful? React with 👍 / 👎.

@alexeyzimarev alexeyzimarev Aug 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed, and reproduced: with GapHandlingTimeoutMs = GapSkipTimeoutMs = 500, the release cleared gap on the same poll that crossed both thresholds, so the remediation block — gated on gap != null — never ran. No tombstone was created, and start then advanced past the position so it was never retried. TombstonesCreationTest uses 500 against the 5000 default, which is why it stayed green.

Fixed in 81d7a80 by tracking remediation on the gap and giving it precedence over the skip:

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;

GapRemediationBeforeSkipTest covers it — it failed with Expected to be 1 but found 0 tombstones before the change.

}

return null;
var newGapAge = DateTime.UtcNow - persistedEvent.Created;

return Options.GapAgeThresholdMs == null || newGapAge.TotalMilliseconds < Options.GapAgeThresholdMs.Value
? new DetectedGap(expectedNext, DateTime.UtcNow)
: null;
}

/// <summary>
Expand Down
Loading