Skip to content

[DB-1872] Improve pinned persistent subscription performance under burst load #5576

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
timothycoleman merged 2 commits into from
Apr 16, 2026
Merged

[DB-1872] Improve pinned persistent subscription performance under burst load #5576

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e574150
Improve persistent subscription performance under burst load
timothycoleman Apr 10, 2026
b7a2f19
update package ref for cve CVE-2026-26171
timothycoleman Apr 16, 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
5 changes: 2 additions & 3 deletions src/Directory.Packages.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,9 +154,8 @@
<PackageVersion Include="System.IdentityModel.Tokens.Jwt" Version="8.14.0" />
<PackageVersion Include="System.Linq.Async" Version="6.0.3" />
<PackageVersion Include="System.Reactive" Version="6.1.0" />
<!-- CVE-2023-29331 -->
<PackageVersion Include="System.Security.Cryptography.Pkcs" Version="10.0.0" />
<PackageVersion Include="System.Security.Cryptography.Xml" Version="10.0.0" />
<PackageVersion Include="System.Security.Cryptography.Pkcs" Version="10.0.6" />
<PackageVersion Include="System.Security.Cryptography.Xml" Version="10.0.6" />
<PackageVersion Include="System.ServiceModel.Http" Version="8.1.2" />
<PackageVersion Include="TestableIO.System.IO.Abstractions" Version="21.3.1" />
<PackageVersion Include="TestableIO.System.IO.Abstractions.TestingHelpers" Version="21.3.1" />
Expand Down
218 changes: 217 additions & 1 deletion src/KurrentDB.Core.Tests/Services/PersistentSubscription/PersistentSubscriptionTests.cs
View file
Open in desktop

Large diffs are not rendered by default.

12 changes: 12 additions & 0 deletions ...DB.Core.Tests/Services/PersistentSubscription/PinnedByCorrelationConsumerStrategyTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,13 +46,15 @@ private static void
var client1Envelope = new FakeEnvelope();
var client2Envelope = new FakeEnvelope();
var reader = new FakeCheckpointReader();
var pushScheduler = new FakePushScheduler();

var sub = new KurrentDB.Core.Services.PersistentSubscription.PersistentSubscription(
PersistentSubscriptionToStreamParamsBuilder.CreateFor("$ce-streamName", "groupName")
.WithEventLoader(new FakeStreamReader())
.WithCheckpointReader(reader)
.WithCheckpointWriter(new FakeCheckpointWriter(x => { }))
.WithMessageParker(new FakeMessageParker())
.WithPushScheduler(pushScheduler)
.CustomConsumerStrategy(new PinnedByCorrelationPersistentSubscriptionConsumerStrategy(new XXHashUnsafe()))
.StartFromCurrent());

Expand All @@ -64,6 +66,7 @@ private static void
Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-1", 0, metaData));
sub.NotifyLiveSubscriptionMessage(
Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-1", 1, metaData));
pushScheduler.Push(sub);

Assert.AreEqual(2, client1Envelope.Replies.Count);
Assert.AreEqual(0, client2Envelope.Replies.Count);
Expand All @@ -72,18 +75,21 @@ private static void
Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-2", 2, metaData));
sub.NotifyLiveSubscriptionMessage(
Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-2", 3, metaData));
pushScheduler.Push(sub);

Assert.AreEqual(2, client1Envelope.Replies.Count);
Assert.AreEqual(2, client2Envelope.Replies.Count);

sub.NotifyLiveSubscriptionMessage(
Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-1", 4, metaData));
pushScheduler.Push(sub);

Assert.AreEqual(3, client1Envelope.Replies.Count);
Assert.AreEqual(2, client2Envelope.Replies.Count);

sub.NotifyLiveSubscriptionMessage(
Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-3", 5, metaData));
pushScheduler.Push(sub);

Assert.AreEqual(4, client1Envelope.Replies.Count);
Assert.AreEqual(2, client2Envelope.Replies.Count);
Expand All @@ -95,13 +101,15 @@ public void live_subscription_with_pinned_by_correlation_pushes_events_with_meta
var client1Envelope = new FakeEnvelope();
var client2Envelope = new FakeEnvelope();
var reader = new FakeCheckpointReader();
var pushScheduler = new FakePushScheduler();

var sub = new KurrentDB.Core.Services.PersistentSubscription.PersistentSubscription(
PersistentSubscriptionToStreamParamsBuilder.CreateFor("$ce-streamName", "groupName")
.WithEventLoader(new FakeStreamReader())
.WithCheckpointReader(reader)
.WithCheckpointWriter(new FakeCheckpointWriter(x => { }))
.WithMessageParker(new FakeMessageParker())
.WithPushScheduler(pushScheduler)
.CustomConsumerStrategy(new PinnedByCorrelationPersistentSubscriptionConsumerStrategy(new XXHashUnsafe()))
.StartFromCurrent());

Expand All @@ -114,23 +122,27 @@ public void live_subscription_with_pinned_by_correlation_pushes_events_with_meta
var metaData1 = Encoding.UTF8.GetBytes(@"{ ""x"": ""x"", ""$correlationId2"": ""1234567890"", ""y"": ""y"" }");
sub.NotifyLiveSubscriptionMessage(Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-1", 0, metaData1));
sub.NotifyLiveSubscriptionMessage(Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-1", 1, metaData1));
pushScheduler.Push(sub);

Assert.AreEqual(2, client1Envelope.Replies.Count);
Assert.AreEqual(0, client2Envelope.Replies.Count);

sub.NotifyLiveSubscriptionMessage(Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-2", 2, metaData1));
sub.NotifyLiveSubscriptionMessage(Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-2", 3, metaData1));
pushScheduler.Push(sub);

Assert.AreEqual(4, client1Envelope.Replies.Count);
Assert.AreEqual(0, client2Envelope.Replies.Count);

var metaData2 = Encoding.UTF8.GetBytes(@"{ ""$correlationId2"": ""1234567891"", ""y"": ""y"" }");
sub.NotifyLiveSubscriptionMessage(Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-1", 4, metaData2));
pushScheduler.Push(sub);

Assert.AreEqual(4, client1Envelope.Replies.Count);
Assert.AreEqual(1, client2Envelope.Replies.Count);

sub.NotifyLiveSubscriptionMessage(Helper.BuildFakeEventWithMetadata(Guid.NewGuid(), "type", "streamName-3", 5, metaData2));
pushScheduler.Push(sub);

Assert.AreEqual(4, client1Envelope.Replies.Count);
Assert.AreEqual(2, client2Envelope.Replies.Count);
Expand Down
Loading
Loading