[DB-1912] Persistent subscriptions: Fix wrong checkpoint when using pinned strategy

[shaan1337]

Persistent subscription consumer strategies that can skip events can produce wrong checkpoints when events are skipped. The pinned consumer strategy is currently the only strategy that can skip events.

This PR fixes the following issues:

• The logic for assigning sequence numbers / previous event positions to OutstandingMessage did not take skipped events into consideration - this could result into wrongly assigned sequence numbers / previous event positions and subsequently to wrong checkpoints too.
• The checkpointing mechanism only considers messages in the retry list but skipped events were left in the buffer - this could also result into wrong checkpoints.

[linear]

DB-1912 Investigate Pinned checkpoint behaviour

[timothycoleman]

/review

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Retry Semantics When the consumer strategy returns Skipped, the code now calls MarkSent() and re-enqueues the message via AddRetry() without incrementing retry count. This changes delivery semantics: verify it cannot lead to duplicate in-flight tracking, reordering, starvation (e.g., repeatedly skipped messages cycling), or an ever-growing retry queue under sustained skipping. private void TryPushingMessagesToClients() { lock (_lock) { if (!TryGetStreamBuffer(out var streamBuffer)) return; foreach (StreamBuffer.OutstandingMessagePointer messagePointer in streamBuffer.Scan()) { (OutstandingMessage message, bool newSequenceNumberAssigned) = OutstandingMessage.ForPushedEvent(messagePointer.Message, _nextSequenceNumber, _lastKnownMessage); if (newSequenceNumberAssigned) { _lastKnownSequenceNumber = _nextSequenceNumber++; _lastKnownMessage = message.EventPosition; } ConsumerPushResult result = _pushClients.PushMessageToClient(message); if (result == ConsumerPushResult.Sent) { messagePointer.MarkSent(); MarkBeginProcessing(message); } else if (result == ConsumerPushResult.Skipped) { // The consumer strategy skipped the message - add it back to the stream buffer as a retry. // we don't increment the retry count as it's an internal retry. messagePointer.MarkSent(); streamBuffer.AddRetry(message); } else if (result == ConsumerPushResult.NoMoreCapacity) { return; } Position Tracking The update of last-known sequence/position is now based on message.EventPosition when a new sequence is assigned, whereas previous logic derived the position via EventSource.GetStreamPositionFor(message.ResolvedEvent). Confirm these are equivalent for all event sources (e.g., link events vs resolved events, category streams, multi-stream sources) and won’t regress checkpoint correctness. lock (_lock) { if (!TryGetStreamBuffer(out var streamBuffer)) yield break; foreach (var messagePointer in streamBuffer.Scan().Take(count)) { messagePointer.MarkSent(); (OutstandingMessage message, bool newSequenceNumberAssigned) = OutstandingMessage.ForPushedEvent(messagePointer.Message, _nextSequenceNumber, _lastKnownMessage); if (newSequenceNumberAssigned) { _lastKnownSequenceNumber = _nextSequenceNumber++; _lastKnownMessage = message.EventPosition; } MarkBeginProcessing(message); yield return (messagePointer.Message.ResolvedEvent, messagePointer.Message.RetryCount); } } } private void MarkBeginProcessing(OutstandingMessage message) { _statistics.IncrementProcessed(); StartMessage(message, _settings.MessageTimeout == TimeSpan.MaxValue ? DateTime.MaxValue : DateTime.UtcNow + _settings.MessageTimeout); Test Robustness The new test relies on hashing/partitioning to route events to specific clients and on precise buffering behavior. Consider validating/locking down the routing assumption (e.g., assert which client each event is intended for) to avoid flakiness if the hashing implementation/seed or pinned strategy behavior changes. public class CheckpointingWithSkippedEvents { [Test] public void checkpointing_works_when_events_are_skipped_by_the_consumer_strategy() { var categoryVersion = 0; var stream1Version = 0; var stream2Version = 0; IPersistentSubscriptionStreamPosition checkpoint = null; var client1Envelope = new FakeEnvelope(); var client2Envelope = new FakeEnvelope(); var reader = new FakeCheckpointReader(); const string subscriptionStream = "$ce-streamName"; var sub = new KurrentDB.Core.Services.PersistentSubscription.PersistentSubscription( PersistentSubscriptionToStreamParamsBuilder.CreateFor(subscriptionStream, "groupName") .WithEventLoader(new FakeStreamReader()) .WithCheckpointReader(reader) .WithCheckpointWriter(new FakeCheckpointWriter(x => checkpoint = x)) .WithMessageParker(new FakeMessageParker()) .MinimumToCheckPoint(1) .MaximumToCheckPoint(1) .CustomConsumerStrategy(new PinnedPersistentSubscriptionConsumerStrategy(new XXHashUnsafe())) .StartFromCurrent()); reader.Load(null); var correlationId = Guid.NewGuid(); sub.AddClient(Guid.NewGuid(), Guid.NewGuid(), "connection-1", client1Envelope, maxInFlight: 1, "foo", "bar"); sub.AddClient(Guid.NewGuid(), Guid.NewGuid(), "connection-2", client2Envelope, maxInFlight: 1000, "foo", "bar"); // write three messages intended for client 1 var message_1_1 = WriteEventForStream1(); // pushed immediately var message_1_2 = WriteEventForStream1(); // buffered - it's skipped by the consumer strategy as `maxInFlight` = 1 var message_1_3 = WriteEventForStream1(); // buffered - it's skipped by the consumer strategy as `maxInFlight` = 1 // write three messages intended for client 2 var message_2_1 = WriteEventForStream2(); // pushed immediately var message_2_2 = WriteEventForStream2(); // pushed immediately var message_2_3 = WriteEventForStream2(); // pushed immediately Assert.That(client1Envelope.Replies.Count, Is.EqualTo(1)); Assert.That(client2Envelope.Replies.Count, Is.EqualTo(3)); Assert.True(sub.TryGetStreamBuffer(out var streamBuffer)); Assert.That(streamBuffer.BufferCount, Is.EqualTo(2)); Assert.AreEqual(null, checkpoint); // checkpoint should be null - no messages have been acknowledged yet Assert.IsNull(checkpoint); // acknowledge message_1_1 for client 1 sub.AcknowledgeMessagesProcessed(correlationId, [ message_1_1, ]); // checkpoint should now be at #0 Assert.AreEqual(new PersistentSubscriptionSingleStreamPosition(0), checkpoint); // acknowledge all the messages for client 2 sub.AcknowledgeMessagesProcessed(correlationId, [ message_2_1, message_2_2, message_2_3, ]); // checkpoint should still be at #0, as message_1_2 has not been acknowledged yet Assert.AreEqual(new PersistentSubscriptionSingleStreamPosition(0), checkpoint); // acknowledge message_1_2. sub.AcknowledgeMessagesProcessed(correlationId, [ message_1_2, ]); // checkpoint should now be at #1, as message_1_3 has not been acknowledged yet Assert.AreEqual(new PersistentSubscriptionSingleStreamPosition(1), checkpoint); // acknowledge message_1_3. sub.AcknowledgeMessagesProcessed(correlationId, [ message_1_3, ]); // checkpoint should now be at #5, as all messages have been acknowledged Assert.AreEqual(new PersistentSubscriptionSingleStreamPosition(5), checkpoint); Guid WriteEventForStream1() { var id = Guid.NewGuid(); sub.NotifyLiveSubscriptionMessage(Helper.BuildLinkEvent(id, subscriptionStream, categoryVersion++, Helper.BuildFakeEvent(Guid.NewGuid(), "type", "streamName-1", stream1Version++), false)); return id; } Guid WriteEventForStream2() { var id = Guid.NewGuid(); sub.NotifyLiveSubscriptionMessage(Helper.BuildLinkEvent(id, subscriptionStream, categoryVersion++, Helper.BuildFakeEvent(Guid.NewGuid(), "type", "streamName-2", stream2Version++), false)); return id; } }

📄 References No matching references available

[timothycoleman]

Originally (before pinned strategies)

• Events were leaving the _buffer in order, and assigned sequence numbers as they did so.
• Checkpointing worked by looking at the lowest sequence number that is in flight or waiting to be retried. i.e. the lowest one that we've started but not finished.

With the addition of pinned strategies

• Introduced 'Skipped' consumer push result, where we leave an event in the _buffer (with no sequence number) if its consumer is too busy.
• Events can therefore leave the buffer out of order and receive sequence numbers out of order.
• Checkpointing unchanged, does not consider events in _buffer that were skipped, only in-flight and retries.

Problems (these only affect pinned strategies):

1. skipped events, when they are pushed to client later, will be assigned a sequence number out of order, which is not suitable for checkpointing.
2. skipped events are not visible to the checkpointing mechanism

This PR

To solve this, we

1. ensure that the sequence numbers are assigned in order even when an event is skipped
2. ensure that skipped events are visible to the checkpoint mechanism

The easiest way to achieve both these things is, when we skip an event we give it a sequence number and move it to the _retry list immediately. This is natural because conceptually the StreamBuffer of events ready to push is _retry ++ _buffer.

Alternatively, when skipping an event we could give it a sequence number and leave it in the _buffer, and give the checkpointer a way to check the buffer (it would only check the initial section that have sequence numbers, not the whole buffer), but this seems unnecessarily complicated.

Note that:

1. Backpressure is maintained because StreamBuffer.CanAccept considers the count of _buffer and _retry
2. Stats will show a higher number of events in retry. this is reasonable enough because although we didn't send them to the client, but we did try to.

[github-actions]

🚨 @shaan1337 Failed to create cherry Pick PR due to error:

RequestError [HttpError]: Merge conflict
   at /home/runner/work/_actions/kurrent-io/Automations/master/cherry-pick-pr-for-label/node_modules/@octokit/request/dist-node/index.js:66:23
   at process.processTicksAndRejections (node:internal/process/task_queues:95:5) {
 status: '409',
 headers: {
   'access-control-allow-origin': '*',
   'access-control-expose-headers': 'ETag, Link, Location, Retry-After, X-GitHub-OTP, X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Used, X-RateLimit-Resource, X-RateLimit-Reset, X-OAuth-Scopes, X-Accepted-OAuth-Scopes, X-Poll-Interval, X-GitHub-Media-Type, X-GitHub-SSO, X-GitHub-Request-Id, Deprecation, Sunset',
   'content-length': '127',
   'content-security-policy': "default-src 'none'",
   'content-type': 'application/json; charset=utf-8',
   date: 'Tue, 14 Apr 2026 06:37:10 GMT',
   'referrer-policy': 'origin-when-cross-origin, strict-origin-when-cross-origin',
   server: 'github.com',
   'strict-transport-security': 'max-age=31536000; includeSubdomains; preload',
   vary: 'Accept-Encoding, Accept, X-Requested-With',
   'x-accepted-github-permissions': 'contents=write',
   'x-content-type-options': 'nosniff',
   'x-frame-options': 'deny',
   'x-github-api-version-selected': '2022-11-28',
   'x-github-media-type': 'github.v3; format=json',
   'x-github-request-id': 'F401:230BC3:2136B40:88AA66D:69DDE095',
   'x-ratelimit-limit': '5000',
   'x-ratelimit-remaining': '4990',
   'x-ratelimit-reset': '1776152226',
   'x-ratelimit-resource': 'core',
   'x-ratelimit-used': '10',
   'x-xss-protection': '0'
 },
 request: {
   method: 'POST',
   url: 'https://api.github.com/repos/kurrent-io/KurrentDB/merges',
   headers: {
     accept: 'application/vnd.github.v3+json',
     'user-agent': 'octokit-core.js/3.3.2 Node.js/20.20.1 (linux; x64)',
     authorization: 'token [REDACTED]',
     'content-type': 'application/json; charset=utf-8'
   },
   body: '{"base":"cherry-pick-cherry-pick/5497/shaan1337/fix-wrong-ps-checkpoint-release/v24.10-01e7b37d-b44d-41d0-89ef-e650209d5b0a","commit_message":"Merge c07492e966e97aff3cf4fff0f27e51c960d21509 into cherry-pick-cherry-pick/5497/shaan1337/fix-wrong-ps-checkpoint-release/v24.10-01e7b37d-b44d-41d0-89ef-e650209d5b0a [skip ci]\\n\\n\\nskip-checks: true\\n","head":"c07492e966e97aff3cf4fff0f27e51c960d21509"}',
   request: { agent: [Agent], hook: [Function: bound bound register] }
 },
 documentation_url: 'https://docs.github.com/rest/branches/branches#merge-a-branch'
}

🚨👉 Check https://github.com/kurrent-io/KurrentDB/actions/runs/24384744335

[github-actions]

@shaan1337 👉 Created pull request targeting release/v26.0: #5578