diff --git a/.github/workflows/pull-requests-v0.yml b/.github/workflows/pull-requests-v0.yml
index cb1b61ea7..9ec26dcf3 100644
--- a/.github/workflows/pull-requests-v0.yml
+++ b/.github/workflows/pull-requests-v0.yml
@@ -29,10 +29,25 @@ jobs:
run: |
powershell -NoProfile -ExecutionPolicy unrestricted -Command ".\up-integration-environment.ps1"
+ - name: Install .NET SDK for KurrentDB adapter
+ uses: actions/setup-dotnet@v4
+ with:
+ dotnet-version: '8.0.x'
+
+ - name: Start isolated KurrentDB test databases
+ run: docker compose -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml up -d --wait --force-recreate
+
+ - name: Run KurrentDB adapter tests
+ run: dotnet test Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj -c Release -f net8.0 --logger trx
+
- name: Run Bake
run: |
bake run --build-version 0.85.${{ github.run_number }}
+ - name: Stop isolated KurrentDB test databases
+ if: always()
+ run: docker compose -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml down
+
- name: Upload NuGet packages
uses: actions/upload-artifact@v4
if: success() || failure()
diff --git a/.github/workflows/release-v0.yml b/.github/workflows/release-v0.yml
index f30a37a8f..5d89a3a39 100644
--- a/.github/workflows/release-v0.yml
+++ b/.github/workflows/release-v0.yml
@@ -36,12 +36,27 @@ jobs:
run: |
powershell -NoProfile -ExecutionPolicy unrestricted -Command ".\up-integration-environment.ps1"
+ - name: Install .NET SDK for KurrentDB adapter
+ uses: actions/setup-dotnet@v4
+ with:
+ dotnet-version: '8.0.x'
+
+ - name: Start isolated KurrentDB test databases
+ run: docker compose -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml up -d --wait --force-recreate
+
+ - name: Run KurrentDB adapter tests
+ run: dotnet test Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj -c Release -f net8.0 --logger trx
+
- name: Run Bake
run: |
declare -i REVISION
REVISION=4714+$GITHUB_RUN_NUMBER
bake run --convention=Release --build-version 0.85.$REVISION --destination="nuget>github,nuget,release>github"
+ - name: Stop isolated KurrentDB test databases
+ if: always()
+ run: docker compose -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml down
+
- name: Upload NuGet packages
uses: actions/upload-artifact@v4
if: success() || failure()
diff --git a/EventFlow.sln b/EventFlow.sln
index b5d59ed96..e78dd1ddd 100644
--- a/EventFlow.sln
+++ b/EventFlow.sln
@@ -101,6 +101,12 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventFlow.MongoDB", "Source
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventFlow.MongoDB.Tests", "Source\EventFlow.MongoDB.Tests\EventFlow.MongoDB.Tests.csproj", "{2D837E47-E92B-422E-AC29-AB5607E9E6D1}"
EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "KurrentDB", "KurrentDB", "{497AB91A-2753-E755-424A-31A4410A0CB4}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventFlow.EventStores.KurrentDB", "Source\EventFlow.EventStores.KurrentDB\EventFlow.EventStores.KurrentDB.csproj", "{0B92ADF2-29EE-499B-9BBD-A31E3BFC4EF7}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventFlow.EventStores.KurrentDB.Tests", "Source\EventFlow.EventStores.KurrentDB.Tests\EventFlow.EventStores.KurrentDB.Tests.csproj", "{EF4EB55C-855B-41FB-B5CA-BBF5B2E718E9}"
+EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@@ -243,6 +249,14 @@ Global
{2D837E47-E92B-422E-AC29-AB5607E9E6D1}.Debug|Any CPU.Build.0 = Debug|Any CPU
{2D837E47-E92B-422E-AC29-AB5607E9E6D1}.Release|Any CPU.ActiveCfg = Release|Any CPU
{2D837E47-E92B-422E-AC29-AB5607E9E6D1}.Release|Any CPU.Build.0 = Release|Any CPU
+ {0B92ADF2-29EE-499B-9BBD-A31E3BFC4EF7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {0B92ADF2-29EE-499B-9BBD-A31E3BFC4EF7}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {0B92ADF2-29EE-499B-9BBD-A31E3BFC4EF7}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {0B92ADF2-29EE-499B-9BBD-A31E3BFC4EF7}.Release|Any CPU.Build.0 = Release|Any CPU
+ {EF4EB55C-855B-41FB-B5CA-BBF5B2E718E9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {EF4EB55C-855B-41FB-B5CA-BBF5B2E718E9}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {EF4EB55C-855B-41FB-B5CA-BBF5B2E718E9}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {EF4EB55C-855B-41FB-B5CA-BBF5B2E718E9}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@@ -282,6 +296,8 @@ Global
{5B2706DA-B66C-4CD9-8441-8C9BDACB8348} = {92F3C263-8C0C-4D12-B41A-452E48D2E5E8}
{C34BFD0C-E7CD-4DEE-A1F4-8FBE9DE0B239} = {7832AC5E-5CE2-4694-BA43-E53222B75EDC}
{2D837E47-E92B-422E-AC29-AB5607E9E6D1} = {7832AC5E-5CE2-4694-BA43-E53222B75EDC}
+ {0B92ADF2-29EE-499B-9BBD-A31E3BFC4EF7} = {497AB91A-2753-E755-424A-31A4410A0CB4}
+ {EF4EB55C-855B-41FB-B5CA-BBF5B2E718E9} = {497AB91A-2753-E755-424A-31A4410A0CB4}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {17607E2C-4E8E-45A2-85BD-0A5808E1C0F3}
diff --git a/README.md b/README.md
index ee03639da..77a83f6ff 100644
--- a/README.md
+++ b/README.md
@@ -128,6 +128,7 @@ to the documentation.
* SQLite
* PostgreSQL
* EventStore - [home page](https://eventstore.org/)
+ * [KurrentDB gRPC](Source/EventFlow.EventStores.KurrentDB/README.md) - .NET Framework 4.8 and .NET 8+
* [**Subscribers:**](https://docs.geteventflow.net/Subscribers.html)
Listeners that act on specific domain events. Useful if an specific action
needs to be triggered after a domain event has been committed.
diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
index 8ed449691..2ebb062dd 100644
--- a/RELEASE_NOTES.md
+++ b/RELEASE_NOTES.md
@@ -1,5 +1,7 @@
### New in 0.85 (not released yet)
+* New: Added the separate `EventFlow.EventStores.KurrentDB` gRPC adapter for .NET Framework 4.8 and .NET 8, preserving the existing EventStore TCP adapter.
+
* New: Update from `System.Data.SqlClient` to `Microsoft.Data.SqlClient` (thanks @janrybka)
### New in 0.84.4 (released 2024-11-26)
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj b/Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj
new file mode 100644
index 000000000..8a03bcad5
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj
@@ -0,0 +1,27 @@
+
+
+
+
+ net8.0
+ net48;net8.0
+ 8.0
+ true
+ true
+ True
+ False
+ true
+ False
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs
new file mode 100644
index 000000000..3e3baba81
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs
@@ -0,0 +1,481 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using EventFlow.Aggregates;
+using EventFlow.Exceptions;
+using EventFlow.EventStores.KurrentDB.Extensions;
+using EventFlow.Logs;
+using EventFlow.TestHelpers;
+using EventFlow.TestHelpers.Aggregates;
+using KurrentDB.Client;
+using NUnit.Framework;
+
+namespace EventFlow.EventStores.KurrentDB.Tests.IntegrationTests
+{
+ [TestFixture]
+ [Category(Categories.Integration)]
+ public class KurrentDBEventPersistenceTests
+ {
+ private KurrentDBClient _client;
+ private KurrentDBEventPersistence _persistence;
+ private ThingyId _id;
+ private CancellationTokenSource _timeout;
+
+ [SetUp]
+ public void SetUp()
+ {
+ var connectionString = TestConnectionStrings.Persistence;
+ var settings = KurrentDBClientSettings.Create(connectionString);
+ // The adapter must enforce optimistic concurrency even when the
+ // application's shared client does not throw on append failure.
+ settings.OperationOptions.ThrowOnAppendFailure = false;
+ _client = new KurrentDBClient(settings);
+ _persistence = new KurrentDBEventPersistence(new NullLog(), _client);
+ _id = ThingyId.New;
+ _timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+ }
+
+ [TearDown]
+ public void TearDown()
+ {
+ _client.Dispose();
+ _timeout.Dispose();
+ }
+
+ [Test]
+ public async Task CommitPreservesSerializedDataMetadataAndEventType()
+ {
+ var serialized = CreateEvent(1);
+ var committed = await _persistence.CommitEventsAsync(_id, new[] { serialized }, _timeout.Token);
+ AssertEvent(committed.Single(), serialized);
+
+ var records = await ReadRawEvents();
+ var record = records.Single().Event;
+ Assert.That(record.EventType, Is.EqualTo("Thingy.Ping.2"));
+ Assert.That(record.ContentType, Is.EqualTo("application/json"));
+ Assert.That(record.Data.ToArray(), Is.EqualTo(Encoding.UTF8.GetBytes(serialized.SerializedData)));
+ Assert.That(record.Metadata.ToArray(), Is.EqualTo(Encoding.UTF8.GetBytes(serialized.SerializedMetadata)));
+
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ AssertEvent(loaded.Single(), serialized);
+ }
+
+ [Test]
+ public async Task MultipleCommitsContinueAggregateSequenceNumbers()
+ {
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1), CreateEvent(2) }, _timeout.Token);
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(3), CreateEvent(4) }, _timeout.Token);
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(loaded.Select(e => e.AggregateSequenceNumber), Is.EqualTo(new[] { 1, 2, 3, 4 }));
+ }
+
+ [TestCase(-1, 3)]
+ [TestCase(0, 3)]
+ [TestCase(1, 3)]
+ [TestCase(2, 2)]
+ [TestCase(4, 0)]
+ public async Task ReadsFromRequestedSequenceNumber(int from, int count)
+ {
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1), CreateEvent(2), CreateEvent(3) }, _timeout.Token);
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, from, _timeout.Token);
+ Assert.That(loaded.Select(e => e.AggregateSequenceNumber),
+ Is.EqualTo(Enumerable.Range(Math.Max(1, from), count)));
+ }
+
+ [Test]
+ public async Task MissingAggregateReturnsNoEvents()
+ {
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(loaded, Is.Empty);
+ }
+
+ [Test]
+ public async Task ReadsMoreThanLegacyPageSize()
+ {
+ var serialized = Enumerable.Range(1, 405).Select(CreateEvent).ToArray();
+ await _persistence.CommitEventsAsync(_id, serialized, _timeout.Token);
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(loaded.Select(e => e.AggregateSequenceNumber), Is.EqualTo(Enumerable.Range(1, 405)));
+ AssertEvent(loaded.Last(), serialized.Last());
+ }
+
+ [Test]
+ public async Task ReadsExistingRecordsUsingLegacyEventFormat()
+ {
+ var serialized = CreateEvent(1);
+ await _client.AppendToStreamAsync(_id.Value, StreamState.NoStream,
+ new[] { new EventData(Uuid.NewUuid(), "Thingy.Ping.2",
+ Encoding.UTF8.GetBytes(serialized.SerializedData),
+ Encoding.UTF8.GetBytes(serialized.SerializedMetadata)) },
+ cancellationToken: _timeout.Token);
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ AssertEvent(loaded.Single(), serialized);
+ }
+
+ [TestCase(1)]
+ [TestCase(2)]
+ public async Task RepeatedStaleCommitThrowsOptimisticConcurrencyException(int sequenceNumber)
+ {
+ if (sequenceNumber > 1)
+ {
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token);
+ }
+
+ var serialized = CreateEvent(sequenceNumber);
+ await _persistence.CommitEventsAsync(_id, new[] { serialized }, _timeout.Token);
+ var exception = Assert.ThrowsAsync(async () =>
+ await _persistence.CommitEventsAsync(_id, new[] { serialized }, _timeout.Token));
+ Assert.That(exception.InnerException, Is.TypeOf());
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(loaded.Count, Is.EqualTo(sequenceNumber));
+
+ // A rejected write must leave the stream usable at its current revision.
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(sequenceNumber + 1) }, _timeout.Token);
+ loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(loaded.Select(e => e.AggregateSequenceNumber),
+ Is.EqualTo(Enumerable.Range(1, sequenceNumber + 1)));
+ }
+
+ [Test]
+ public void AppendingToMissingStreamWithNonInitialSequenceThrows()
+ {
+ Assert.ThrowsAsync(async () =>
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(2) }, _timeout.Token));
+ }
+
+ [Test]
+ public async Task EmptyCommitDoesNotCreateStream()
+ {
+ var committed = await _persistence.CommitEventsAsync(_id, Array.Empty(), _timeout.Token);
+ Assert.That(committed, Is.Empty);
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(loaded, Is.Empty);
+ }
+
+ [Test]
+ public void PreCancelledOperationsAreCancelled()
+ {
+ using (var cancellation = new CancellationTokenSource())
+ {
+ cancellation.Cancel();
+ Assert.ThrowsAsync(async () =>
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, cancellation.Token));
+ Assert.ThrowsAsync(async () =>
+ await _persistence.LoadCommittedEventsAsync(_id, 1, cancellation.Token));
+ Assert.ThrowsAsync(async () =>
+ await _persistence.DeleteEventsAsync(_id, cancellation.Token));
+ Assert.ThrowsAsync(async () =>
+ await _persistence.LoadAllCommittedEvents(GlobalPosition.Start, 10, cancellation.Token));
+ }
+ }
+
+ [Test]
+ public async Task DeleteRemovesOnlyTheRequestedAggregate()
+ {
+ var otherId = ThingyId.New;
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1), CreateEvent(2) }, _timeout.Token);
+ await _persistence.CommitEventsAsync(otherId, new[] { CreateEvent(1) }, _timeout.Token);
+ await _persistence.DeleteEventsAsync(_id, _timeout.Token);
+
+ var deleted = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ var other = await _persistence.LoadCommittedEventsAsync(otherId, 1, _timeout.Token);
+ Assert.That(deleted, Is.Empty);
+ Assert.That(other.Single().AggregateId, Is.EqualTo(otherId.Value));
+ }
+
+ [Test]
+ public async Task DeleteIsSoftAndDoesNotTombstoneTheStream()
+ {
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token);
+ await _persistence.DeleteEventsAsync(_id, _timeout.Token);
+ await _client.AppendToStreamAsync(_id.Value, StreamState.Any,
+ new[] { new EventData(Uuid.NewUuid(), "Thingy.Ping.2", Encoding.UTF8.GetBytes("{}")) },
+ cancellationToken: _timeout.Token);
+ var loaded = await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(loaded.Single().AggregateSequenceNumber, Is.EqualTo(2));
+ }
+
+ [Test]
+ public void DeleteMissingStreamPreservesServerFailure()
+ {
+ Assert.ThrowsAsync(async () =>
+ await _persistence.DeleteEventsAsync(_id, _timeout.Token));
+ }
+ [TestCase(1)]
+ [TestCase(2)]
+ public async Task GlobalPagesDoNotSkipOrRepeatEventsWithinBatch(int pageSize)
+ {
+ var checkpoint = await TailCheckpoint();
+ var serialized = Enumerable.Range(1, 5).Select(CreateEvent).ToArray();
+ await _persistence.CommitEventsAsync(_id, serialized, _timeout.Token);
+ var loaded = new System.Collections.Generic.List();
+ for (var pageIndex = 0; pageIndex < 6; pageIndex++)
+ {
+ var page = await _persistence.LoadAllCommittedEvents(checkpoint, pageSize, _timeout.Token);
+ Assert.That(page.CommittedDomainEvents.Count, Is.LessThanOrEqualTo(pageSize));
+ Assert.That(page.NextGlobalPosition.Value, Does.StartWith("after:"));
+ loaded.AddRange(page.CommittedDomainEvents);
+ checkpoint = page.NextGlobalPosition;
+ if (page.CommittedDomainEvents.Count == 0) break;
+ }
+
+ Assert.That(loaded.Select(e => e.AggregateSequenceNumber), Is.EqualTo(Enumerable.Range(1, 5)));
+ for (var i = 0; i < serialized.Length; i++) AssertEvent(loaded[i], serialized[i]);
+ }
+
+ [Test]
+ public async Task LegacyCheckpointIsInclusiveAndNewCheckpointIsExclusive()
+ {
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1), CreateEvent(2), CreateEvent(3) }, _timeout.Token);
+ var records = await ReadRawEvents();
+ var second = records[1].OriginalPosition.Value;
+ var checkpoint = new GlobalPosition(string.Format(System.Globalization.CultureInfo.InvariantCulture,
+ "{0}-{1}", second.CommitPosition, second.PreparePosition));
+ var page = await _persistence.LoadAllCommittedEvents(checkpoint, 1, _timeout.Token);
+ Assert.That(page.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(2));
+ var next = await _persistence.LoadAllCommittedEvents(page.NextGlobalPosition, 1, _timeout.Token);
+ Assert.That(next.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(3));
+ }
+
+ [Test]
+ public async Task TailCheckpointRemainsUsableAfterEmptyReadsAndNewAppends()
+ {
+ var checkpoint = await TailCheckpoint();
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token);
+ var first = await _persistence.LoadAllCommittedEvents(checkpoint, 10, _timeout.Token);
+ Assert.That(first.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(1));
+ var empty = await _persistence.LoadAllCommittedEvents(first.NextGlobalPosition, 10, _timeout.Token);
+ Assert.That(empty.CommittedDomainEvents, Is.Empty);
+ var emptyAgain = await _persistence.LoadAllCommittedEvents(empty.NextGlobalPosition, 10, _timeout.Token);
+ Assert.That(emptyAgain.CommittedDomainEvents, Is.Empty);
+ Assert.That(emptyAgain.NextGlobalPosition, Is.EqualTo(empty.NextGlobalPosition));
+
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(2) }, _timeout.Token);
+ var appended = await _persistence.LoadAllCommittedEvents(emptyAgain.NextGlobalPosition, 10, _timeout.Token);
+ Assert.That(appended.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(2));
+ }
+
+ [Test]
+ public async Task GlobalReadsSkipSystemStreamsAndContinueAcrossLargeGaps()
+ {
+ var checkpoint = await TailCheckpoint();
+ await WriteSystemEvents(20);
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token);
+ await WriteSystemEvents(20);
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(2) }, _timeout.Token);
+ await WriteSystemEvents(20);
+
+ var page = await _persistence.LoadAllCommittedEvents(checkpoint, 2, _timeout.Token);
+ Assert.That(page.CommittedDomainEvents.Select(e => e.AggregateSequenceNumber), Is.EqualTo(new[] { 1, 2 }));
+ Assert.That(page.CommittedDomainEvents.All(e => e.AggregateId == _id.Value), Is.True);
+ var empty = await _persistence.LoadAllCommittedEvents(page.NextGlobalPosition, 2, _timeout.Token);
+ Assert.That(empty.CommittedDomainEvents, Is.Empty);
+ Assert.That(empty.NextGlobalPosition, Is.Not.EqualTo(page.NextGlobalPosition));
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(3) }, _timeout.Token);
+ var appended = await _persistence.LoadAllCommittedEvents(empty.NextGlobalPosition, 2, _timeout.Token);
+ Assert.That(appended.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(3));
+ }
+
+ [Test]
+ public async Task GlobalReadsPreserveUnscavengedEventsAfterSoftDeletion()
+ {
+ var checkpoint = await TailCheckpoint();
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token);
+ await _persistence.DeleteEventsAsync(_id, _timeout.Token);
+ var page = await _persistence.LoadAllCommittedEvents(checkpoint, 10, _timeout.Token);
+ Assert.That(page.CommittedDomainEvents.Single().AggregateId, Is.EqualTo(_id.Value));
+ Assert.That(await _persistence.LoadCommittedEventsAsync(_id, 1, _timeout.Token), Is.Empty);
+ }
+
+ [Test]
+ public async Task GlobalReadFromStartIncludesAggregateAndExcludesSystemStreams()
+ {
+ var serialized = CreateEvent(1);
+ await _persistence.CommitEventsAsync(_id, new[] { serialized }, _timeout.Token);
+ var page = await _persistence.LoadAllCommittedEvents(GlobalPosition.Start, 10000, _timeout.Token);
+ Assert.That(page.CommittedDomainEvents.All(e => !e.AggregateId.StartsWith("$", StringComparison.Ordinal)), Is.True);
+ AssertEvent(page.CommittedDomainEvents.Single(e => e.AggregateId == _id.Value), serialized);
+ }
+
+ [TestCase("garbage")]
+ [TestCase("after:")]
+ [TestCase("after:1-2")]
+ [TestCase("1-2")]
+ [TestCase("-1-0")]
+ [TestCase("1--1")]
+ [TestCase("after:9223372036854775808-0")]
+ [TestCase("unknown:1-0")]
+ [TestCase("1-0-0")]
+ public void MalformedGlobalCheckpointsAreRejected(string checkpoint)
+ {
+ Assert.ThrowsAsync(async () =>
+ await _persistence.LoadAllCommittedEvents(new GlobalPosition(checkpoint), 10, _timeout.Token));
+ }
+
+ [TestCase(0)]
+ [TestCase(-1)]
+ public void GlobalReadsRequirePositivePageSize(int pageSize)
+ {
+ Assert.ThrowsAsync(async () =>
+ await _persistence.LoadAllCommittedEvents(GlobalPosition.Start, pageSize, _timeout.Token));
+ }
+
+ private async Task TailCheckpoint()
+ {
+ await foreach (var record in _client.ReadAllAsync(Direction.Backwards, Position.End,
+ maxCount: 1, cancellationToken: _timeout.Token))
+ {
+ var position = record.OriginalPosition.Value;
+ return new GlobalPosition(string.Format(System.Globalization.CultureInfo.InvariantCulture,
+ "after:{0}-{1}", position.CommitPosition, position.PreparePosition));
+ }
+ return GlobalPosition.Start;
+ }
+
+ private async Task WriteSystemEvents(int count)
+ {
+ await _client.AppendToStreamAsync("$eventflow-checkpoint-test-" + Guid.NewGuid().ToString("N"),
+ StreamState.NoStream,
+ Enumerable.Range(0, count).Select(_ => new EventData(Uuid.NewUuid(), "$test", Encoding.UTF8.GetBytes("{}"))),
+ cancellationToken: _timeout.Token);
+ }
+ [Test]
+ public async Task RegisteredPersistenceCanCommitAndReadAcrossScopes()
+ {
+ var connectionString = TestConnectionStrings.Persistence;
+ using (var resolver = EventFlowOptions.New.UseKurrentDBEventStore(connectionString).CreateResolver())
+ {
+ var serialized = CreateEvent(1);
+ using (var writeScope = resolver.BeginScope())
+ {
+ await writeScope.Resolve().CommitEventsAsync(
+ _id, new[] { serialized }, _timeout.Token);
+ }
+ using (var readScope = resolver.BeginScope())
+ {
+ var loaded = await readScope.Resolve().LoadCommittedEventsAsync(
+ _id, 1, _timeout.Token);
+ AssertEvent(loaded.Single(), serialized);
+ }
+ }
+ }
+ [TestCase(1)]
+ [TestCase(2)]
+ public async Task LostAppendAcknowledgementDoesNotDuplicateEvents(int sequenceNumber)
+ {
+ if (sequenceNumber > 1)
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token);
+
+ var attempts = 0;
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, async (stream, state, events, token) =>
+ {
+ await _client.AppendToStreamAsync(stream, state, events,
+ configureOperationOptions: options => options.ThrowOnAppendFailure = true, cancellationToken: token);
+ if (++attempts == 1)
+ throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.DeadlineExceeded, "acknowledgement lost after commit"));
+ });
+ var serialized = new[] { CreateEvent(sequenceNumber), CreateEvent(sequenceNumber + 1) };
+ var committed = await persistence.CommitEventsAsync(_id, serialized, _timeout.Token);
+ Assert.That(attempts, Is.EqualTo(2));
+ Assert.That(committed.Count, Is.EqualTo(2));
+ for (var i = 0; i < serialized.Length; i++) AssertEvent(committed.ElementAt(i), serialized[i]);
+ var stored = await ReadRawEvents();
+ Assert.That(stored.Length, Is.EqualTo(sequenceNumber + 1));
+ Assert.That(stored.Last().Event.Data.ToArray(), Is.EqualTo(Encoding.UTF8.GetBytes(serialized.Last().SerializedData)));
+ }
+
+ [Test]
+ public async Task TimeoutBeforeServerCommitRetriesTheWrite()
+ {
+ var attempts = 0;
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, async (stream, state, events, token) =>
+ {
+ if (++attempts == 1)
+ throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.DeadlineExceeded, "timeout before commit"));
+ await _client.AppendToStreamAsync(stream, state, events,
+ configureOperationOptions: options => options.ThrowOnAppendFailure = true, cancellationToken: token);
+ });
+ await persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token);
+ Assert.That(attempts, Is.EqualTo(2));
+ Assert.That((await ReadRawEvents()).Length, Is.EqualTo(1));
+ }
+
+ [Test]
+ public async Task RivalWriteAfterTimeoutRemainsAConcurrencyConflict()
+ {
+ var attempts = 0;
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, async (stream, state, events, token) =>
+ {
+ if (++attempts == 1)
+ {
+ // Another command wins the revision before this command retries.
+ await _persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, token);
+ throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.DeadlineExceeded, "timeout before retry"));
+ }
+ await _client.AppendToStreamAsync(stream, state, events,
+ configureOperationOptions: options => options.ThrowOnAppendFailure = true, cancellationToken: token);
+ });
+ var exception = Assert.ThrowsAsync(async () =>
+ await persistence.CommitEventsAsync(_id, new[] { CreateEvent(1) }, _timeout.Token));
+ Assert.That(exception.InnerException, Is.TypeOf());
+ Assert.That(attempts, Is.EqualTo(2));
+ Assert.That((await ReadRawEvents()).Length, Is.EqualTo(1));
+ }
+ private static SerializedEvent CreateEvent(int sequenceNumber)
+ {
+ var metadata = new Metadata
+ {
+ AggregateName = "Thingy",
+ EventName = "Ping",
+ EventVersion = 2,
+ EventId = EventId.New
+ };
+ return new SerializedEvent("{ \"custom\": \"München 東京 🚚\" }",
+ "{ \"message\": \"héllo 東京 🚚\", \"sequence\": " + sequenceNumber + " }",
+ sequenceNumber, metadata);
+ }
+
+ private void AssertEvent(ICommittedDomainEvent actual, SerializedEvent expected)
+ {
+ Assert.That(actual.AggregateId, Is.EqualTo(_id.Value));
+ Assert.That(actual.AggregateSequenceNumber, Is.EqualTo(expected.AggregateSequenceNumber));
+ Assert.That(actual.Data, Is.EqualTo(expected.SerializedData));
+ Assert.That(actual.Metadata, Is.EqualTo(expected.SerializedMetadata));
+ }
+
+ private async Task ReadRawEvents()
+ {
+ var events = new System.Collections.Generic.List();
+ await foreach (var record in _client.ReadStreamAsync(Direction.Forwards,
+ _id.Value, StreamPosition.Start, cancellationToken: _timeout.Token))
+ {
+ events.Add(record);
+ }
+ return events.ToArray();
+ }
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventStoreTests.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventStoreTests.cs
new file mode 100644
index 000000000..3c725f612
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventStoreTests.cs
@@ -0,0 +1,51 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using EventFlow.Configuration;
+using EventFlow.EventStores.KurrentDB.Extensions;
+using EventFlow.Extensions;
+using EventFlow.MetadataProviders;
+using EventFlow.TestHelpers;
+using EventFlow.TestHelpers.Suites;
+using NUnit.Framework;
+
+namespace EventFlow.EventStores.KurrentDB.Tests.IntegrationTests
+{
+ [TestFixture]
+ [Category(Categories.Integration)]
+ [NonParallelizable]
+ public class KurrentDBEventStoreTests : TestSuiteForEventStore
+ {
+ protected override IRootResolver CreateRootResolver(IEventFlowOptions eventFlowOptions)
+ {
+ // The shared suite deserializes $all, so it needs a separate database
+ // from the persistence tests that intentionally seed raw event shapes.
+ var connectionString = TestConnectionStrings.Suite;
+
+ return eventFlowOptions
+ .AddMetadataProvider()
+ .UseKurrentDBEventStore(connectionString)
+ .CreateResolver();
+ }
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/TcpMigrationTests.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/TcpMigrationTests.cs
new file mode 100644
index 000000000..1fc4ddeb2
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/TcpMigrationTests.cs
@@ -0,0 +1,238 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using EventFlow.Aggregates;
+using EventFlow.Configuration;
+using EventFlow.Core;
+using EventFlow.EventStores.EventStore;
+using EventFlow.EventStores.KurrentDB.Extensions;
+using EventFlow.Exceptions;
+using EventFlow.Extensions;
+using EventFlow.Logs;
+using EventFlow.Snapshots;
+using EventFlow.TestHelpers;
+using EventFlow.TestHelpers.Aggregates;
+using EventFlow.TestHelpers.Aggregates.Events;
+using EventFlow.TestHelpers.Aggregates.Queries;
+using EventFlow.TestHelpers.Aggregates.ValueObjects;
+using KurrentDB.Client;
+using Newtonsoft.Json;
+using NUnit.Framework;
+using Tcp = EventStore.ClientAPI;
+
+namespace EventFlow.EventStores.KurrentDB.Tests.IntegrationTests
+{
+ [TestFixture]
+ [Category(Categories.Integration)]
+ [NonParallelizable]
+ public class TcpMigrationTests
+ {
+ private Tcp.IEventStoreConnection _connection;
+ private KurrentDBClient _client;
+ private EventStoreEventPersistence _tcp;
+ private KurrentDBEventPersistence _grpc;
+ private ThingyId _id;
+ private CancellationTokenSource _timeout;
+ private string _grpcConnectionString;
+
+ [SetUp]
+ public async Task SetUp()
+ {
+ _id = ThingyId.New;
+ _timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+ var settings = Tcp.ConnectionSettings.Create()
+ .LimitReconnectionsTo(0)
+ .LimitRetriesForOperationTo(0)
+ .SetOperationTimeoutTo(TimeSpan.FromSeconds(10))
+#if NET48
+ .SetDefaultUserCredentials(new Tcp.SystemData.UserCredentials("admin", "changeit"))
+#endif
+ .Build();
+ var tcpUri = new Uri(TestConnectionStrings.MigrationTcp);
+ _connection = Tcp.EventStoreConnection.Create(settings, tcpUri);
+ await _connection.ConnectAsync();
+ _tcp = new EventStoreEventPersistence(new NullLog(), _connection);
+ _grpcConnectionString = TestConnectionStrings.Migration;
+ _client = new KurrentDBClient(KurrentDBClientSettings.Create(_grpcConnectionString));
+ _grpc = new KurrentDBEventPersistence(new NullLog(), _client);
+ }
+
+ [TearDown]
+ public void TearDown()
+ {
+ _client?.Dispose();
+ _connection?.Dispose();
+ _timeout?.Dispose();
+ }
+
+ [Test]
+ public async Task TcpWrittenAggregateCanBeReplayedAndExtendedThroughGrpc()
+ {
+ var pingIds = Enumerable.Range(0, 3).Select(_ => PingId.New).ToArray();
+ var serialized = pingIds.Select((pingId, index) => CreateEvent(index + 1, pingId)).ToArray();
+ await _tcp.CommitEventsAsync(_id, serialized, _timeout.Token);
+
+ var loaded = await _grpc.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ AssertEventsEqual(loaded, serialized);
+
+ using (var resolver = EventFlowOptions.New
+ .RegisterServices(services => services.Register(Lifetime.Scoped))
+ .AddEvents(typeof(ThingyPingEvent))
+ .UseKurrentDBEventStore(_grpcConnectionString)
+ .CreateResolver())
+ {
+ var aggregate = await resolver.Resolve()
+ .LoadAsync(_id, _timeout.Token);
+ Assert.That(aggregate.Version, Is.EqualTo(3));
+ Assert.That(aggregate.PingsReceived, Is.EqualTo(pingIds));
+ var nextPing = PingId.New;
+ aggregate.Ping(nextPing);
+ await aggregate.CommitAsync(resolver.Resolve(), resolver.Resolve(),
+ SourceId.New, _timeout.Token);
+ var reloaded = await resolver.Resolve()
+ .LoadAsync(_id, _timeout.Token);
+ Assert.That(reloaded.Version, Is.EqualTo(4));
+ Assert.That(reloaded.PingsReceived, Is.EqualTo(pingIds.Concat(new[] { nextPing })));
+ }
+
+ var readBackThroughTcp = await _tcp.LoadCommittedEventsAsync(_id, 1, _timeout.Token);
+ Assert.That(readBackThroughTcp.Select(e => e.AggregateSequenceNumber), Is.EqualTo(new[] { 1, 2, 3, 4 }));
+ AssertEventsEqual(readBackThroughTcp.Take(3).ToArray(), serialized);
+ }
+
+ [Test]
+ public async Task TcpCheckpointResumesWithoutRepeatingOrSkippingEvents()
+ {
+ var start = await TcpTailCheckpoint();
+ var serialized = Enumerable.Range(1, 3).Select(n => CreateEvent(n, PingId.New)).ToArray();
+ await _tcp.CommitEventsAsync(_id, serialized, _timeout.Token);
+ var first = await _tcp.LoadAllCommittedEvents(start, 1, _timeout.Token);
+ Assert.That(first.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(1));
+ Assert.That(first.NextGlobalPosition.Value, Does.Not.StartWith("after:"));
+
+ var second = await _grpc.LoadAllCommittedEvents(first.NextGlobalPosition, 1, _timeout.Token);
+ Assert.That(second.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(2));
+ Assert.That(second.NextGlobalPosition.Value, Does.StartWith("after:"));
+ var third = await _grpc.LoadAllCommittedEvents(second.NextGlobalPosition, 1, _timeout.Token);
+ Assert.That(third.CommittedDomainEvents.Single().AggregateSequenceNumber, Is.EqualTo(3));
+ AssertEventsEqual(first.CommittedDomainEvents.Concat(second.CommittedDomainEvents)
+ .Concat(third.CommittedDomainEvents).ToArray(), serialized);
+ }
+
+ [Test]
+ public async Task TcpCheckpointAtEndOfLogSeesLaterAppends()
+ {
+ await _tcp.CommitEventsAsync(_id, new[] { CreateEvent(1, PingId.New) }, _timeout.Token);
+ var checkpoint = await TcpTailCheckpoint();
+ var empty = await _grpc.LoadAllCommittedEvents(checkpoint, 10, _timeout.Token);
+ Assert.That(empty.CommittedDomainEvents, Is.Empty);
+ var next = CreateEvent(2, PingId.New);
+ await _tcp.CommitEventsAsync(_id, new[] { next }, _timeout.Token);
+ var appended = await _grpc.LoadAllCommittedEvents(empty.NextGlobalPosition, 10, _timeout.Token);
+ AssertEventsEqual(appended.CommittedDomainEvents, new[] { next });
+ }
+
+ [Test]
+ public async Task ExplicitTcpTransactionCanBePagedByCommitAndPreparePosition()
+ {
+ var start = await TcpTailCheckpoint();
+ var serialized = Enumerable.Range(1, 3).Select(n => CreateEvent(n, PingId.New)).ToArray();
+ using (var transaction = await _connection.StartTransactionAsync(_id.Value, Tcp.ExpectedVersion.NoStream))
+ {
+ await transaction.WriteAsync(serialized.Select(e => new Tcp.EventData(
+ Guid.NewGuid(), "Thingy.ThingyPing.1", true,
+ Encoding.UTF8.GetBytes(e.SerializedData), Encoding.UTF8.GetBytes(e.SerializedMetadata))));
+ await transaction.CommitAsync();
+ }
+
+ var first = await _tcp.LoadAllCommittedEvents(start, 1, _timeout.Token);
+ var second = await _grpc.LoadAllCommittedEvents(first.NextGlobalPosition, 1, _timeout.Token);
+ var third = await _grpc.LoadAllCommittedEvents(second.NextGlobalPosition, 1, _timeout.Token);
+ AssertEventsEqual(first.CommittedDomainEvents.Concat(second.CommittedDomainEvents)
+ .Concat(third.CommittedDomainEvents).ToArray(), serialized);
+
+ // TCP stream reads omit global positions; $all carries both offsets.
+ var parts = start.Value.Split('-');
+ var raw = await _connection.ReadAllEventsForwardAsync(new Tcp.Position(
+ long.Parse(parts[0], System.Globalization.CultureInfo.InvariantCulture),
+ long.Parse(parts[1], System.Globalization.CultureInfo.InvariantCulture)), 4096, false);
+ var positions = raw.Events.Where(e => e.OriginalStreamId == _id.Value)
+ .Select(e => e.OriginalPosition.Value).ToArray();
+ Assert.That(positions.Select(p => p.CommitPosition).Distinct().Count(), Is.EqualTo(1));
+ Assert.That(positions.Select(p => p.PreparePosition).Distinct().Count(), Is.EqualTo(3));
+ }
+
+ [Test]
+ public async Task StaleTcpWriterConflictsAfterGrpcAppend()
+ {
+ await _tcp.CommitEventsAsync(_id, new[] { CreateEvent(1, PingId.New) }, _timeout.Token);
+ await _grpc.CommitEventsAsync(_id, new[] { CreateEvent(2, PingId.New) }, _timeout.Token);
+ Assert.ThrowsAsync(async () =>
+ await _tcp.CommitEventsAsync(_id, new[] { CreateEvent(2, PingId.New) }, _timeout.Token));
+ Assert.ThrowsAsync(async () =>
+ await _grpc.CommitEventsAsync(_id, new[] { CreateEvent(1, PingId.New) }, _timeout.Token));
+ }
+
+ private async Task TcpTailCheckpoint()
+ {
+ var page = await _tcp.LoadAllCommittedEvents(GlobalPosition.Start, 4096, _timeout.Token);
+ return page.NextGlobalPosition;
+ }
+
+ private SerializedEvent CreateEvent(int sequenceNumber, PingId pingId)
+ {
+ var metadata = new Metadata
+ {
+ AggregateName = "Thingy",
+ AggregateId = _id.Value,
+ AggregateSequenceNumber = sequenceNumber,
+ EventName = "ThingyPing",
+ EventVersion = 1,
+ EventId = EventId.New,
+ SourceId = SourceId.New,
+ Timestamp = DateTimeOffset.UtcNow,
+ ["custom"] = "München 東京 🚚"
+ };
+ return new SerializedEvent(JsonConvert.SerializeObject(metadata),
+ JsonConvert.SerializeObject(new ThingyPingEvent(pingId)), sequenceNumber, metadata);
+ }
+
+ private void AssertEventsEqual(IReadOnlyCollection actual, SerializedEvent[] expected)
+ {
+ Assert.That(actual.Count, Is.EqualTo(expected.Length));
+ var records = actual.ToArray();
+ for (var i = 0; i < expected.Length; i++)
+ {
+ Assert.That(records[i].AggregateId, Is.EqualTo(_id.Value));
+ Assert.That(records[i].AggregateSequenceNumber, Is.EqualTo(expected[i].AggregateSequenceNumber));
+ Assert.That(records[i].Data, Is.EqualTo(expected[i].SerializedData));
+ Assert.That(records[i].Metadata, Is.EqualTo(expected[i].SerializedMetadata));
+ }
+ }
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/README.md b/Source/EventFlow.EventStores.KurrentDB.Tests/README.md
new file mode 100644
index 000000000..bf97cd64c
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/README.md
@@ -0,0 +1,138 @@
+# KurrentDB persistence integration tests
+
+Run these commands from the repository root for .NET 8:
+
+```powershell
+docker compose -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml up -d --wait --force-recreate
+dotnet test Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj -f net8.0
+docker compose -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml down
+```
+
+The three in-memory servers use insecure connections for local testing. Stopping
+and removing the containers discards their test data. Tests use fresh aggregate
+streams and do not reset other streams.
+
+| Test fixture | Default server | Override variable |
+| --- | --- | --- |
+| Persistence tests | `kurrentdb://localhost:22113?tls=false` | `KURRENTDB_CONNECTION_STRING` |
+| Shared event-store suite | `kurrentdb://localhost:22114?tls=false` | `KURRENTDB_SUITE_CONNECTION_STRING` |
+| TCP migration (gRPC) | `kurrentdb://localhost:22115?tls=false` | `KURRENTDB_MIGRATION_CONNECTION_STRING` |
+| TCP migration (TCP) | `tcp://localhost:21115` | `KURRENTDB_MIGRATION_TCP_URL` |
+
+Use separate dedicated databases: the persistence tests seed raw event shapes,
+while the shared suite deserializes `$all` as EventFlow domain events. Start the
+shared suite on a fresh database for each run; its first-page assertions assume
+that earlier runs have not filled the first 200 events. Run the cleanup command
+after tests, including when a test fails.
+
+Append retry tests inject deadline failures before a real server append and after
+a successful append whose acknowledgement is lost. They verify idempotent recovery,
+a rival command winning the revision, bounded attempts, unchanged batches and
+cancellation during backoff. Production transport behaviour is unchanged except
+for the explicit per-attempt deadline and retry policy.
+
+Tests cover aggregate appends/reads, optimistic concurrency, global pagination,
+checkpoint boundaries, system-stream gaps, cancellation, and soft deletion.
+Registration tests cover adapter selection, singleton sharing across scopes,
+separate clients per root resolver, and validation without connecting to a server.
+The migration fixture references the unchanged TCP adapter only from the test
+project. It writes through `EventStore.Client` 5.0.1 and reads/appends through
+`KurrentDB.Client` 1.6.0 against the same EventStoreDB 23.10.8 server.
+
+## Global checkpoints
+
+The adapter accepts existing TCP `commit-prepare` checkpoints inclusively: the
+record at that location, if present, has not been processed yet. New checkpoints
+use `after:commit-prepare` and identify the last scanned record, which is excluded
+on resume. Store `NextGlobalPosition` unchanged, including its prefix. These new
+checkpoints cannot be passed back to the TCP adapter.
+
+No arithmetic is performed on log offsets. A checkpoint remains usable at the
+end of the log and after new events are appended. System streams (names starting
+with `$`) are excluded from results but still advance the checkpoint.
+
+Deletion uses the server's soft-delete operation with any expected revision,
+matching the TCP adapter. It does not tombstone a stream. Deletion errors are
+propagated. Global replay can still return soft-deleted events until scavenging
+removes them, matching the existing `$all` behavior.
+
+The test project targets `net8.0` by default. Set `EnableNet48Tests=true` to add the `net48` target, and use the separate TLS profile below. The adapter package always targets both frameworks.
+
+## Registration tests without a server
+
+```powershell
+dotnet test Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj -f net8.0 --filter FullyQualifiedName~UnitTests
+```
+
+## Shared EventFlow event-store suite
+
+`KurrentDBEventStoreTests` inherits all tests from `TestSuiteForEventStore` and
+uses the public registration extension. No inherited tests are overridden or
+skipped, including the large-gap test. To run only this fixture:
+
+```powershell
+dotnet test Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj -f net8.0 --filter FullyQualifiedName~KurrentDBEventStoreTests
+```
+
+## TCP-to-gRPC migration
+
+The third server uses `eventstore/eventstore:23.10.8-bookworm-slim` with external
+TCP enabled. Both migration connection-string overrides must address the same
+dedicated database. The two other fixtures continue to use KurrentDB 26.1.1.
+
+```powershell
+dotnet test Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj -f net8.0 --filter FullyQualifiedName~TcpMigrationTests
+```
+
+The tests verify actual TCP-written aggregate replay, continued gRPC writes
+readable by TCP, unchanged serialized data and metadata, checkpoints returned by
+the legacy adapter (including at the end of the log), cross-client optimistic
+concurrency, and paging within a transaction whose events share one commit
+position and have different prepare positions.
+
+These tests validate switching client protocols on a compatible server. They do
+not perform a database-file upgrade from EventStoreDB 23.10 to KurrentDB 26.1.
+
+## .NET Framework runtime tests on Windows
+
+The `net48` target requires .NET Framework 4.8 or later and Windows HTTP/2
+support for client/bidirectional streaming (Windows 11 or later). The KurrentDB
+client selects `WinHttpHandler` automatically. Use TLS; the insecure .NET 8
+endpoints do not support this transport.
+
+```powershell
+docker compose --profile net48 -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml up -d --wait --force-recreate kurrentdb-net48 kurrentdb-suite-net48 eventstore-migration-net48
+dotnet test Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj -p:EnableNet48Tests=true -f net48
+docker compose --profile net48 -p eventflow-kurrentdb-tests -f docker-compose.kurrentdb.yml down
+```
+
+| Fixture | TLS port | Override variable |
+| --- | --- | --- |
+| Persistence | 22116 | `KURRENTDB_NET48_CONNECTION_STRING` |
+| Shared suite | 22117 | `KURRENTDB_NET48_SUITE_CONNECTION_STRING` |
+| Migration gRPC | 22118 | `KURRENTDB_NET48_MIGRATION_CONNECTION_STRING` |
+| Migration TCP | 21118 | `KURRENTDB_NET48_MIGRATION_TCP_URL` |
+
+TLS defaults use `kurrentdb://admin:changeit@localhost:PORT?tlsVerifyCert=false`.
+These self-signed development certificates exist only in private container tmpfs
+storage. No host certificate-store changes are required. Certificate validation
+is disabled only in these local test connection strings; production connections
+must use trusted certificates. The migration TCP connection is plaintext on
+loopback with development credentials.
+
+`RuntimeTargetTests` checks the actual runtime, target framework attributes of
+the loaded adapter/client assemblies, and the Framework HTTP handler. Running
+`net48` on a machine with Framework 4.8.1 validates the 4.8-targeted binaries on
+that installed runtime, not a separate Framework 4.8-only installation.
+
+
+Validation on 2026-10-24: all 70 tests passed for the net48 target on .NET
+Framework 4.8.1 (runtime 4.8.10008.0), Windows build 26300, over TLS. No tests
+were skipped. This includes the unchanged shared EventFlow suite and real TCP
+protocol-migration tests.
+
+Validation on 2026-10-26: all 82 tests passed on both net8.0 and net48, with
+zero skipped tests, using isolated KurrentDB 26.1.1 and EventStoreDB 23.10.8
+databases. The strengthened multi-event lost-acknowledgement cases also passed
+on both targets. The new tests simulate failures at the append boundary; they
+do not attempt to induce actual network packet loss.
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/TestConnectionStrings.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/TestConnectionStrings.cs
new file mode 100644
index 000000000..069f4e399
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/TestConnectionStrings.cs
@@ -0,0 +1,51 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+
+namespace EventFlow.EventStores.KurrentDB.Tests
+{
+ internal static class TestConnectionStrings
+ {
+#if NET48
+ // WinHTTP requires TLS for HTTP/2. Certificate validation is disabled
+ // only for the dedicated containers' self-signed development certificates.
+ public static string Persistence => Environment.GetEnvironmentVariable("KURRENTDB_NET48_CONNECTION_STRING")
+ ?? "kurrentdb://admin:changeit@localhost:22116?tlsVerifyCert=false";
+ public static string Suite => Environment.GetEnvironmentVariable("KURRENTDB_NET48_SUITE_CONNECTION_STRING")
+ ?? "kurrentdb://admin:changeit@localhost:22117?tlsVerifyCert=false";
+ public static string Migration => Environment.GetEnvironmentVariable("KURRENTDB_NET48_MIGRATION_CONNECTION_STRING")
+ ?? "kurrentdb://admin:changeit@localhost:22118?tlsVerifyCert=false";
+ public static string MigrationTcp => Environment.GetEnvironmentVariable("KURRENTDB_NET48_MIGRATION_TCP_URL")
+ ?? "tcp://localhost:21118";
+#else
+ public static string Persistence => Environment.GetEnvironmentVariable("KURRENTDB_CONNECTION_STRING")
+ ?? "kurrentdb://localhost:22113?tls=false";
+ public static string Suite => Environment.GetEnvironmentVariable("KURRENTDB_SUITE_CONNECTION_STRING")
+ ?? "kurrentdb://localhost:22114?tls=false";
+ public static string Migration => Environment.GetEnvironmentVariable("KURRENTDB_MIGRATION_CONNECTION_STRING")
+ ?? "kurrentdb://localhost:22115?tls=false";
+ public static string MigrationTcp => Environment.GetEnvironmentVariable("KURRENTDB_MIGRATION_TCP_URL")
+ ?? "tcp://localhost:21115";
+#endif
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/AppendRetryTests.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/AppendRetryTests.cs
new file mode 100644
index 000000000..40bb9f502
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/AppendRetryTests.cs
@@ -0,0 +1,163 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using EventFlow.Aggregates;
+using Metadata = EventFlow.Aggregates.Metadata;
+using EventFlow.Logs;
+using EventFlow.TestHelpers.Aggregates;
+using Grpc.Core;
+using KurrentDB.Client;
+using NUnit.Framework;
+
+namespace EventFlow.EventStores.KurrentDB.Tests.UnitTests
+{
+ [TestFixture]
+ public class AppendRetryTests
+ {
+ private KurrentDBClient _client;
+
+ [SetUp]
+ public void SetUp()
+ {
+ _client = new KurrentDBClient(KurrentDBClientSettings.Create("kurrentdb://localhost:1?tls=false"));
+ }
+
+ [TearDown]
+ public void TearDown() => _client.Dispose();
+
+ [Test]
+ public async Task TimeoutRetriesKeepTheSameBatchIdsPayloadAndRevision()
+ {
+ var attempts = 0;
+ EventData[] firstBatch = null;
+ Uuid[] firstIds = null;
+ byte[] firstData = null;
+ byte[] firstMetadata = null;
+ var id = ThingyId.New;
+ using (var cancellation = new CancellationTokenSource())
+ {
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, (stream, state, events, token) =>
+ {
+ attempts++;
+ if (firstBatch == null)
+ {
+ firstBatch = events;
+ firstIds = events.Select(e => e.EventId).ToArray();
+ firstData = events[0].Data.ToArray();
+ firstMetadata = events[0].Metadata.ToArray();
+ }
+ Assert.That(stream, Is.EqualTo(id.Value));
+ Assert.That(state, Is.EqualTo(StreamState.NoStream));
+ Assert.That(events, Is.SameAs(firstBatch));
+ Assert.That(events.Select(e => e.EventId), Is.EqualTo(firstIds));
+ Assert.That(events[0].Data.ToArray(), Is.EqualTo(firstData));
+ Assert.That(events[0].Metadata.ToArray(), Is.EqualTo(firstMetadata));
+ Assert.That(token, Is.EqualTo(cancellation.Token));
+ return attempts < 3 ? Task.FromException(Deadline()) : Task.CompletedTask;
+ });
+ var committed = await persistence.CommitEventsAsync(id, new[] { CreateEvent(1), CreateEvent(2) }, cancellation.Token);
+ Assert.That(attempts, Is.EqualTo(3));
+ Assert.That(committed.Select(e => e.AggregateSequenceNumber), Is.EqualTo(new[] { 1, 2 }));
+ }
+ }
+
+ [Test]
+ public void ExhaustedTimeoutRetriesPropagateTheLastFailure()
+ {
+ var attempts = 0;
+ RpcException last = null;
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, (stream, state, events, token) =>
+ {
+ attempts++;
+ last = Deadline();
+ return Task.FromException(last);
+ });
+ var exception = Assert.ThrowsAsync(async () =>
+ await persistence.CommitEventsAsync(ThingyId.New, new[] { CreateEvent(1) }, CancellationToken.None));
+ Assert.That(attempts, Is.EqualTo(5));
+ Assert.That(exception, Is.SameAs(last));
+ }
+
+ [TestCase(StatusCode.Unavailable)]
+ [TestCase(StatusCode.PermissionDenied)]
+ [TestCase(StatusCode.Aborted)]
+ [TestCase(StatusCode.Cancelled)]
+ public void NonDeadlineFailuresAreNotRetried(StatusCode statusCode)
+ {
+ var attempts = 0;
+ var failure = new RpcException(new Status(statusCode, "test"));
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, (stream, state, events, token) =>
+ {
+ attempts++;
+ return Task.FromException(failure);
+ });
+ var exception = Assert.ThrowsAsync(async () =>
+ await persistence.CommitEventsAsync(ThingyId.New, new[] { CreateEvent(1) }, CancellationToken.None));
+ Assert.That(attempts, Is.EqualTo(1));
+ Assert.That(exception, Is.SameAs(failure));
+ }
+
+ [Test]
+ public void CancellationDuringBackoffStopsBeforeAnotherAppend()
+ {
+ var attempts = 0;
+ using (var cancellation = new CancellationTokenSource())
+ {
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, (stream, state, events, token) =>
+ {
+ attempts++;
+ return Task.FromException(Deadline());
+ });
+ var commit = persistence.CommitEventsAsync(ThingyId.New, new[] { CreateEvent(1) }, cancellation.Token);
+ Assert.That(attempts, Is.EqualTo(1));
+ cancellation.Cancel();
+ Assert.That(async () => await commit, Throws.InstanceOf());
+ Assert.That(attempts, Is.EqualTo(1));
+ }
+ }
+
+ [Test]
+ public async Task SeparateCommitsDoNotShareStorageIds()
+ {
+ Uuid? previousId = null;
+ var persistence = new KurrentDBEventPersistence(new NullLog(), _client, (stream, state, events, token) =>
+ {
+ if (previousId.HasValue) Assert.That(events[0].EventId, Is.Not.EqualTo(previousId.Value));
+ previousId = events[0].EventId;
+ return Task.CompletedTask;
+ });
+ var id = ThingyId.New;
+ var serialized = CreateEvent(1);
+ await persistence.CommitEventsAsync(id, new[] { serialized }, CancellationToken.None);
+ await persistence.CommitEventsAsync(id, new[] { serialized }, CancellationToken.None);
+ }
+
+ private static RpcException Deadline() => new RpcException(new Status(StatusCode.DeadlineExceeded, "simulated append timeout"));
+
+ private static SerializedEvent CreateEvent(int sequence) => new SerializedEvent("{\"metadata\":true}", "{\"data\":true}", sequence,
+ new Metadata { AggregateName = "Thingy", EventName = "Ping", EventVersion = 2, EventId = EventId.New });
+ }
+}
\ No newline at end of file
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/EventFlowOptionsExtensionsTests.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/EventFlowOptionsExtensionsTests.cs
new file mode 100644
index 000000000..8c18255ae
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/EventFlowOptionsExtensionsTests.cs
@@ -0,0 +1,110 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using EventFlow.EventStores.KurrentDB.Extensions;
+using EventFlow.Extensions;
+using KurrentDB.Client;
+using EventFlow.TestHelpers;
+using NUnit.Framework;
+
+namespace EventFlow.EventStores.KurrentDB.Tests.UnitTests
+{
+ [TestFixture]
+ [Category(Categories.Unit)]
+ public class EventFlowOptionsExtensionsTests
+ {
+ // No database is required: configuring and resolving services must not connect.
+ private const string ConnectionString = "kurrentdb://localhost:1?tls=false";
+
+ [Test]
+ public void RegistrationReturnsTheSameOptionsForChaining()
+ {
+ var options = EventFlowOptions.New;
+ Assert.That(options.UseKurrentDBEventStore(ConnectionString), Is.SameAs(options));
+ }
+
+ [Test]
+ public void RegistrationSelectsKurrentDBPersistence()
+ {
+ using (var resolver = EventFlowOptions.New.UseKurrentDBEventStore(ConnectionString).CreateResolver())
+ {
+ Assert.That(resolver.Resolve(), Is.TypeOf());
+ Assert.That(resolver.Resolve(), Is.TypeOf());
+ }
+ }
+
+ [Test]
+ public void ClientIsSharedAcrossResolutionsAndScopes()
+ {
+ using (var resolver = EventFlowOptions.New.UseKurrentDBEventStore(ConnectionString).CreateResolver())
+ {
+ var client = resolver.Resolve();
+ Assert.That(resolver.Resolve(), Is.SameAs(client));
+ using (var first = resolver.BeginScope())
+ using (var second = resolver.BeginScope())
+ {
+ Assert.That(first.Resolve(), Is.SameAs(client));
+ Assert.That(second.Resolve(), Is.SameAs(client));
+ Assert.That(first.Resolve(), Is.TypeOf());
+ Assert.That(second.Resolve(), Is.TypeOf());
+ }
+ Assert.That(resolver.Resolve(), Is.SameAs(client));
+ }
+ }
+
+ [Test]
+ public void EachRootResolverOwnsADifferentClient()
+ {
+ var options = EventFlowOptions.New.UseKurrentDBEventStore(ConnectionString);
+ using (var first = options.CreateResolver())
+ using (var second = options.CreateResolver())
+ {
+ Assert.That(first.Resolve(), Is.Not.SameAs(second.Resolve()));
+ }
+ }
+
+ [Test]
+ public void NullOptionsAreRejected()
+ {
+ var exception = Assert.Throws(() =>
+ EventFlowOptionsExtensions.UseKurrentDBEventStore(null, ConnectionString));
+ Assert.That(exception.ParamName, Is.EqualTo("eventFlowOptions"));
+ }
+
+ [TestCase(null)]
+ [TestCase("")]
+ [TestCase(" ")]
+ public void EmptyConnectionStringsAreRejected(string connectionString)
+ {
+ var exception = Assert.Throws(() =>
+ EventFlowOptions.New.UseKurrentDBEventStore(connectionString));
+ Assert.That(exception.ParamName, Is.EqualTo("connectionString"));
+ }
+
+ [Test]
+ public void MalformedConnectionStringIsRejectedDuringConfiguration()
+ {
+ Assert.That(() => EventFlowOptions.New.UseKurrentDBEventStore("not a connection string"), Throws.Exception);
+ }
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/RuntimeTargetTests.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/RuntimeTargetTests.cs
new file mode 100644
index 000000000..2e1681ea2
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/RuntimeTargetTests.cs
@@ -0,0 +1,60 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using System.Reflection;
+using System.Runtime.InteropServices;
+using System.Runtime.Versioning;
+using KurrentDB.Client;
+using EventFlow.TestHelpers;
+using NUnit.Framework;
+
+namespace EventFlow.EventStores.KurrentDB.Tests.UnitTests
+{
+ [TestFixture]
+ [Category(Categories.Unit)]
+ public class RuntimeTargetTests
+ {
+ [Test]
+ public void LoadsAdapterAndClientAssetsForTheRequestedRuntime()
+ {
+#if NET48
+ const string expectedFramework = ".NETFramework,Version=v4.8";
+ Assert.That(RuntimeInformation.FrameworkDescription, Does.StartWith(".NET Framework"));
+ var settings = KurrentDBClientSettings.Create(TestConnectionStrings.Persistence);
+ using (var handler = settings.CreateHttpMessageHandler())
+ {
+ Assert.That(handler.GetType().FullName, Is.EqualTo("System.Net.Http.WinHttpHandler"));
+ }
+#else
+ const string expectedFramework = ".NETCoreApp,Version=v8.0";
+ Assert.That(Environment.Version.Major, Is.GreaterThanOrEqualTo(8));
+#endif
+ Assert.That(typeof(KurrentDBEventPersistence).Assembly
+ .GetCustomAttribute().FrameworkName, Is.EqualTo(expectedFramework));
+ Assert.That(typeof(KurrentDBClient).Assembly
+ .GetCustomAttribute().FrameworkName, Is.EqualTo(expectedFramework));
+ TestContext.WriteLine(RuntimeInformation.FrameworkDescription);
+ TestContext.WriteLine(Environment.OSVersion.VersionString);
+ }
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj b/Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj
new file mode 100644
index 000000000..da143bbdc
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj
@@ -0,0 +1,36 @@
+
+
+
+ net48;net8.0
+ 8.0
+ True
+ True
+ False
+ True
+ EventFlow.EventStores.KurrentDB
+ Rasmus Mikkelsen
+ Rasmus Mikkelsen
+ Copyright (c) Rasmus Mikkelsen 2015 - 2026
+ KurrentDB gRPC event store for EventFlow.
+ CQRS ES event sourcing KurrentDB
+ git
+ https://github.com/eventflow/EventFlow
+ https://docs.geteventflow.net/
+ MIT
+ README.md
+ en-US
+ UPDATED BY BUILD
+ true
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Source/EventFlow.EventStores.KurrentDB/Extensions/EventFlowOptionsExtensions.cs b/Source/EventFlow.EventStores.KurrentDB/Extensions/EventFlowOptionsExtensions.cs
new file mode 100644
index 000000000..da1b00a0f
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB/Extensions/EventFlowOptionsExtensions.cs
@@ -0,0 +1,57 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using EventFlow.Configuration;
+using EventFlow.Extensions;
+using KurrentDB.Client;
+
+namespace EventFlow.EventStores.KurrentDB.Extensions
+{
+ public static class EventFlowOptionsExtensions
+ {
+ ///
+ /// Uses KurrentDB for event persistence. Each root resolver owns one
+ /// client, shared by all of its scopes and disposed with the resolver.
+ ///
+ public static IEventFlowOptions UseKurrentDBEventStore(
+ this IEventFlowOptions eventFlowOptions,
+ string connectionString)
+ {
+ if (eventFlowOptions == null)
+ {
+ throw new ArgumentNullException(nameof(eventFlowOptions));
+ }
+ if (string.IsNullOrWhiteSpace(connectionString))
+ {
+ throw new ArgumentException("A KurrentDB connection string is required.", nameof(connectionString));
+ }
+
+ // Validate configuration now; the client is created and owned by DI.
+ var settings = KurrentDBClientSettings.Create(connectionString);
+ return eventFlowOptions
+ .RegisterServices(services => services.Register(
+ _ => new KurrentDBClient(settings), Lifetime.Singleton))
+ .UseEventStore();
+ }
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs b/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs
new file mode 100644
index 000000000..25988e9e3
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs
@@ -0,0 +1,269 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2024 Rasmus Mikkelsen
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using EventFlow.Core;
+using EventFlow.Exceptions;
+using EventFlow.Logs;
+using Grpc.Core;
+using KurrentDB.Client;
+
+namespace EventFlow.EventStores.KurrentDB
+{
+ public class KurrentDBEventPersistence : IEventPersistence
+ {
+ private const int MaxAppendAttempts = 5;
+ private static readonly TimeSpan AppendDeadline = TimeSpan.FromSeconds(30);
+ private readonly Func _append;
+ private readonly ILog _log;
+ private readonly KurrentDBClient _client;
+
+ private class KurrentDBEvent : ICommittedDomainEvent
+ {
+ public string AggregateId { get; set; }
+ public string Data { get; set; }
+ public string Metadata { get; set; }
+ public int AggregateSequenceNumber { get; set; }
+ }
+
+ public KurrentDBEventPersistence(ILog log, KurrentDBClient client)
+ : this(log, client, null)
+ {
+ }
+
+ // Internal transport seam allows tests to reproduce a lost acknowledgement
+ // after a real server commit without changing the public registration API.
+ internal KurrentDBEventPersistence(ILog log, KurrentDBClient client,
+ Func append)
+ {
+ _log = log;
+ _client = client;
+ _append = append ?? ((stream, expectedState, events, token) => client.AppendToStreamAsync(
+ stream, expectedState, events,
+ deadline: AppendDeadline,
+ configureOperationOptions: options => options.ThrowOnAppendFailure = true,
+ cancellationToken: token));
+ }
+
+ public async Task> CommitEventsAsync(
+ IIdentity id,
+ IReadOnlyCollection serializedEvents,
+ CancellationToken cancellationToken)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ if (serializedEvents.Count == 0)
+ {
+ return Array.Empty();
+ }
+
+ var firstSequenceNumber = serializedEvents.Min(e => e.AggregateSequenceNumber);
+ var expectedState = firstSequenceNumber <= 1
+ ? StreamState.NoStream
+ : StreamState.StreamRevision((ulong)(firstSequenceNumber - 2));
+ var events = serializedEvents.Select(e => new EventData(
+ // Use a fresh storage ID, as in the TCP adapter, so a repeated
+ // stale commit cannot be accepted as an idempotent write.
+ Uuid.NewUuid(),
+ string.Format("{0}.{1}.{2}", e.Metadata.AggregateName, e.Metadata.EventName, e.Metadata.EventVersion),
+ Encoding.UTF8.GetBytes(e.SerializedData),
+ Encoding.UTF8.GetBytes(e.SerializedMetadata),
+ "application/json")).ToArray();
+
+ try
+ {
+ await AppendWithRetryAsync(id.Value, expectedState, events, cancellationToken).ConfigureAwait(false);
+ }
+ catch (WrongExpectedVersionException e)
+ {
+ throw new OptimisticConcurrencyException(e.Message, e);
+ }
+
+ _log.Verbose("Wrote entity {0} with sequence number {1}",
+ id, serializedEvents.Max(e => e.AggregateSequenceNumber));
+
+ return serializedEvents.Select(e => (ICommittedDomainEvent)new KurrentDBEvent
+ {
+ AggregateId = id.Value,
+ AggregateSequenceNumber = e.AggregateSequenceNumber,
+ Data = e.SerializedData,
+ Metadata = e.SerializedMetadata
+ }).ToList();
+ }
+
+ private async Task AppendWithRetryAsync(string stream, StreamState expectedState,
+ EventData[] events, CancellationToken cancellationToken)
+ {
+ // A deadline does not establish whether the server committed the write.
+ // Retain the exact batch, storage IDs and expected revision on every retry.
+ // IDs are fresh again on a separate CommitEventsAsync call, so rival
+ // commands at the same aggregate sequence are still concurrency conflicts.
+ for (var attempt = 1; ; attempt++)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ try
+ {
+ await _append(stream, expectedState, events, cancellationToken).ConfigureAwait(false);
+ return;
+ }
+ catch (RpcException exception) when (exception.StatusCode == StatusCode.DeadlineExceeded
+ && attempt < MaxAppendAttempts && !cancellationToken.IsCancellationRequested)
+ {
+ _log.Warning("Append to stream '{0}' timed out on attempt {1}/{2}; retrying the identical batch because the outcome is unknown",
+ stream, attempt, MaxAppendAttempts);
+ await Task.Delay(TimeSpan.FromMilliseconds(50 * attempt), cancellationToken).ConfigureAwait(false);
+ }
+ }
+ }
+
+ public async Task> LoadCommittedEventsAsync(
+ IIdentity id,
+ int fromEventSequenceNumber,
+ CancellationToken cancellationToken)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ var start = fromEventSequenceNumber <= 1
+ ? StreamPosition.Start
+ : new StreamPosition((ulong)(fromEventSequenceNumber - 1));
+ var events = new List();
+ var read = _client.ReadStreamAsync(
+ Direction.Forwards,
+ id.Value,
+ start,
+ resolveLinkTos: false,
+ cancellationToken: cancellationToken);
+
+ try
+ {
+ await foreach (var resolvedEvent in read.ConfigureAwait(false))
+ {
+ events.Add(Map(resolvedEvent));
+ }
+ }
+ catch (StreamNotFoundException)
+ {
+ return Array.Empty();
+ }
+
+ return events;
+ }
+
+ public async Task LoadAllCommittedEvents(
+ GlobalPosition globalPosition,
+ int pageSize,
+ CancellationToken cancellationToken)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ if (pageSize <= 0)
+ {
+ throw new ArgumentOutOfRangeException(nameof(pageSize));
+ }
+
+ var position = ParsePosition(globalPosition, out var exclusive);
+ var nextPosition = globalPosition;
+ var events = new List();
+ var read = _client.ReadAllAsync(Direction.Forwards, position,
+ resolveLinkTos: false, cancellationToken: cancellationToken);
+
+ await foreach (var resolvedEvent in read.ConfigureAwait(false))
+ {
+ var currentPosition = resolvedEvent.OriginalPosition.Value;
+ // Legacy TCP checkpoints identify the next unread location and
+ // are inclusive. Our versioned checkpoints identify the last
+ // scanned record, since gRPC does not expose TCP's NextPosition.
+ if (exclusive && currentPosition <= position)
+ {
+ continue;
+ }
+
+ nextPosition = new GlobalPosition(string.Format(CultureInfo.InvariantCulture,
+ "after:{0}-{1}", currentPosition.CommitPosition, currentPosition.PreparePosition));
+ if (resolvedEvent.OriginalStreamId.StartsWith("$", StringComparison.Ordinal))
+ {
+ continue;
+ }
+
+ events.Add(Map(resolvedEvent));
+ if (events.Count == pageSize)
+ {
+ break;
+ }
+ }
+
+ return new AllCommittedEventsPage(nextPosition, events);
+ }
+
+ private static Position ParsePosition(GlobalPosition globalPosition, out bool exclusive)
+ {
+ if (globalPosition == null)
+ {
+ throw new ArgumentNullException(nameof(globalPosition));
+ }
+
+ exclusive = false;
+ if (globalPosition.IsStart)
+ {
+ return Position.Start;
+ }
+
+ const string prefix = "after:";
+ exclusive = globalPosition.Value.StartsWith(prefix, StringComparison.Ordinal);
+ var value = exclusive ? globalPosition.Value.Substring(prefix.Length) : globalPosition.Value;
+ var parts = value.Split('-');
+ if (parts.Length != 2
+ || !long.TryParse(parts[0], NumberStyles.None, CultureInfo.InvariantCulture, out var commit)
+ || !long.TryParse(parts[1], NumberStyles.None, CultureInfo.InvariantCulture, out var prepare)
+ || commit < prepare)
+ {
+ throw new ArgumentException(string.Format(CultureInfo.InvariantCulture,
+ "Invalid global position '{0}'. Expected 'commit-prepare' or 'after:commit-prepare' with non-negative 64-bit positions and commit >= prepare.",
+ globalPosition.Value), nameof(globalPosition));
+ }
+
+ return new Position((ulong)commit, (ulong)prepare);
+ }
+
+ private static ICommittedDomainEvent Map(ResolvedEvent resolvedEvent)
+ {
+ var record = resolvedEvent.Event;
+ return new KurrentDBEvent
+ {
+ AggregateId = record.EventStreamId,
+ AggregateSequenceNumber = checked((int)(record.EventNumber.ToUInt64() + 1)),
+ Data = Encoding.UTF8.GetString(record.Data.ToArray()),
+ Metadata = Encoding.UTF8.GetString(record.Metadata.ToArray())
+ };
+ }
+
+ public async Task DeleteEventsAsync(IIdentity id, CancellationToken cancellationToken)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ await _client.DeleteAsync(id.Value, StreamState.Any,
+ cancellationToken: cancellationToken).ConfigureAwait(false);
+ }
+ }
+}
diff --git a/Source/EventFlow.EventStores.KurrentDB/README.md b/Source/EventFlow.EventStores.KurrentDB/README.md
new file mode 100644
index 000000000..ac750c62b
--- /dev/null
+++ b/Source/EventFlow.EventStores.KurrentDB/README.md
@@ -0,0 +1,60 @@
+# KurrentDB event persistence for EventFlow 0.x
+
+This adapter uses `KurrentDB.Client` 1.6.0 and targets .NET Framework 4.8 and
+.NET 8.0. It is separate from the existing EventStore TCP adapter.
+
+```csharp
+using EventFlow;
+using EventFlow.EventStores.KurrentDB.Extensions;
+
+using (var resolver = EventFlowOptions.New
+ .UseKurrentDBEventStore(connectionString)
+ .CreateResolver())
+{
+ // Resolve EventFlow services here.
+}
+```
+
+Use a KurrentDB gRPC connection string rather than the TCP adapter's `tcp://`
+connection string. For the local integration-test server, use
+`kurrentdb://localhost:22113?tls=false`. Configure TLS and credentials for your
+server in the connection string.
+
+Registration validates the connection string and registers `KurrentDBClient` as
+a singleton factory. Each root resolver creates and owns its own client; child
+scopes share it. Dispose the root resolver to release the client. Registration
+and service resolution do not wait for a network connection.
+
+Existing aggregate stream names, event type names, UTF-8 data/metadata, and
+one-based aggregate sequence numbers are preserved. Writes enforce optimistic
+concurrency and translate conflicts to EventFlow's `OptimisticConcurrencyException`.
+
+Append deadline failures are retried up to five attempts with a 30-second deadline
+per attempt and cancellable delays of 50, 100, 150 and 200 milliseconds. A retry
+reuses the exact event batch, storage IDs and expected revision: if the first
+append committed but its acknowledgement was lost, the repeated append is
+idempotent. Separate CommitEventsAsync calls still create fresh storage IDs so
+competing commands cannot be mistaken for the same write. Other errors are not
+retried by the adapter; revision conflicts remain OptimisticConcurrencyException.
+The final deadline error is propagated if retries are exhausted. Cancellation
+stops the retry sequence. An exhausted timeout still has an unknown write outcome.
+Existing TCP global checkpoints (`commit-prepare`) are accepted inclusively.
+New checkpoints use `after:commit-prepare` and resume after the last scanned
+record. Store `NextGlobalPosition` unchanged. New checkpoints cannot be consumed
+by the TCP adapter. Global reads exclude streams whose names start with `$`.
+
+Deletion soft-deletes streams. Global replay can still return unscavenged events
+from deleted streams, as with the TCP adapter.
+
+The `net48` binaries passed all 82 tests on .NET Framework 4.8.1
+(runtime 4.8.10008.0), Windows build 26300, using TLS and `WinHttpHandler`.
+This validates that installed runtime; a Framework 4.8-only installation has not
+been tested. Framework applications require TLS and Windows HTTP/2 support for
+client/bidirectional streaming (Windows 11 or later). Windows Server 2019/2022
+has streaming restrictions, so do not assume append support from read support.
+See the adjacent test project's README for separate Framework and .NET 8 setups.
+
+Protocol migration tests exercise actual streams and checkpoints written by the
+unchanged legacy TCP adapter against EventStoreDB 23.10.8. They verify aggregate
+replay, continued writes, concurrency, and transaction-boundary pagination. They
+do not validate a server storage upgrade; see the test project's README.
diff --git a/docker-compose.kurrentdb.yml b/docker-compose.kurrentdb.yml
new file mode 100644
index 000000000..f79446452
--- /dev/null
+++ b/docker-compose.kurrentdb.yml
@@ -0,0 +1,75 @@
+services:
+ kurrentdb: &kurrentdb
+ image: docker.kurrent.io/kurrent-latest/kurrentdb:26.1.1
+ environment:
+ KURRENTDB_INSECURE: "true"
+ KURRENTDB_CLUSTER_SIZE: "1"
+ KURRENTDB_MEM_DB: "true"
+ healthcheck:
+ test: ["CMD", "curl", "--fail", "--silent", "http://localhost:2113/health/live"]
+ interval: 2s
+ timeout: 5s
+ retries: 30
+ ports:
+ - "127.0.0.1:22113:2113"
+
+ kurrentdb-suite:
+ <<: *kurrentdb
+ ports:
+ - "127.0.0.1:22114:2113"
+
+ eventstore-migration:
+ image: eventstore/eventstore:23.10.8-bookworm-slim
+ environment:
+ EVENTSTORE_INSECURE: "true"
+ EVENTSTORE_CLUSTER_SIZE: "1"
+ EVENTSTORE_MEM_DB: "true"
+ EVENTSTORE_ENABLE_EXTERNAL_TCP: "true"
+ EVENTSTORE_NODE_IP: "0.0.0.0"
+ healthcheck:
+ test: ["CMD", "curl", "--fail", "--silent", "http://localhost:2113/health/live"]
+ interval: 2s
+ timeout: 5s
+ retries: 30
+ ports:
+ - "127.0.0.1:22115:2113"
+ - "127.0.0.1:21115:1113"
+
+ kurrentdb-net48: &kurrentdb-net48
+ <<: *kurrentdb
+ profiles: [net48]
+ tmpfs:
+ - /home/kurrent:uid=1001,gid=1001,mode=0700
+ environment:
+ KURRENTDB_DEV: "true"
+ KURRENTDB_CLUSTER_SIZE: "1"
+ KURRENTDB_MEM_DB: "true"
+ KURRENTDB_NODE_IP: "0.0.0.0"
+ healthcheck:
+ test: ["CMD", "curl", "--insecure", "--fail", "--silent", "https://localhost:2113/health/live"]
+ interval: 2s
+ timeout: 5s
+ retries: 60
+ ports:
+ - "127.0.0.1:22116:2113"
+
+ kurrentdb-suite-net48:
+ <<: *kurrentdb-net48
+ ports:
+ - "127.0.0.1:22117:2113"
+
+ eventstore-migration-net48:
+ <<: *kurrentdb-net48
+ image: eventstore/eventstore:23.10.8-bookworm-slim
+ tmpfs:
+ - /home/eventstore:uid=1000,gid=1000,mode=0700
+ environment:
+ EVENTSTORE_DEV: "true"
+ EVENTSTORE_CLUSTER_SIZE: "1"
+ EVENTSTORE_MEM_DB: "true"
+ EVENTSTORE_ENABLE_EXTERNAL_TCP: "true"
+ EVENTSTORE_DISABLE_EXTERNAL_TCP_TLS: "true"
+ EVENTSTORE_NODE_IP: "0.0.0.0"
+ ports:
+ - "127.0.0.1:22118:2113"
+ - "127.0.0.1:21118:1113"