From 69140e47bb36e53b27a6353e73d190aad1f720e6 Mon Sep 17 00:00:00 2001 From: Warren Date: Sat, 24 Oct 2026 10:50:50 +0200 Subject: [PATCH 1/2] Add KurrentDB gRPC persistence adapter for EventFlow 0.x --- .github/workflows/pull-requests-v0.yml | 15 + .github/workflows/release-v0.yml | 15 + EventFlow.sln | 16 + README.md | 1 + RELEASE_NOTES.md | 2 + ...entFlow.EventStores.KurrentDB.Tests.csproj | 27 ++ .../KurrentDBEventPersistenceTests.cs | 419 ++++++++++++++++++ .../KurrentDBEventStoreTests.cs | 51 +++ .../IntegrationTests/TcpMigrationTests.cs | 238 ++++++++++ .../README.md | 126 ++++++ .../TestConnectionStrings.cs | 51 +++ .../EventFlowOptionsExtensionsTests.cs | 110 +++++ .../UnitTests/RuntimeTargetTests.cs | 60 +++ .../EventFlow.EventStores.KurrentDB.csproj | 35 ++ .../Extensions/EventFlowOptionsExtensions.cs | 57 +++ .../KurrentDBEventPersistence.cs | 232 ++++++++++ .../EventFlow.EventStores.KurrentDB/README.md | 51 +++ docker-compose.kurrentdb.yml | 75 ++++ 18 files changed, 1581 insertions(+) create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/EventFlow.EventStores.KurrentDB.Tests.csproj create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventStoreTests.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/TcpMigrationTests.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/README.md create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/TestConnectionStrings.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/EventFlowOptionsExtensionsTests.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/RuntimeTargetTests.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj create mode 100644 Source/EventFlow.EventStores.KurrentDB/Extensions/EventFlowOptionsExtensions.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs create mode 100644 Source/EventFlow.EventStores.KurrentDB/README.md create mode 100644 docker-compose.kurrentdb.yml 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..6393441bc --- /dev/null +++ b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs @@ -0,0 +1,419 @@ +// 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); + } + } + } + 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..c78d3f7f3 --- /dev/null +++ b/Source/EventFlow.EventStores.KurrentDB.Tests/README.md @@ -0,0 +1,126 @@ +# 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. + +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. 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/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..e1824285d --- /dev/null +++ b/Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj @@ -0,0 +1,35 @@ + + + + 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..933716351 --- /dev/null +++ b/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs @@ -0,0 +1,232 @@ +// 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 KurrentDB.Client; + +namespace EventFlow.EventStores.KurrentDB +{ + public class KurrentDBEventPersistence : IEventPersistence + { + 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) + { + _log = log; + _client = client; + } + + 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 _client.AppendToStreamAsync( + id.Value, + expectedState, + events, + configureOperationOptions: options => options.ThrowOnAppendFailure = true, + cancellationToken: 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(); + } + + 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..b649fe71c --- /dev/null +++ b/Source/EventFlow.EventStores.KurrentDB/README.md @@ -0,0 +1,51 @@ +# 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`. + +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 70 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" From 35c193047f60e02deb98f2828e5456410fc6dff8 Mon Sep 17 00:00:00 2001 From: Warren Date: Mon, 26 Oct 2026 07:12:17 +0200 Subject: [PATCH 2/2] Recover ambiguous KurrentDB append timeouts without duplicate events --- .../KurrentDBEventPersistenceTests.cs | 62 +++++++ .../README.md | 12 ++ .../UnitTests/AppendRetryTests.cs | 163 ++++++++++++++++++ .../EventFlow.EventStores.KurrentDB.csproj | 1 + .../KurrentDBEventPersistence.cs | 49 +++++- .../EventFlow.EventStores.KurrentDB/README.md | 11 +- 6 files changed, 291 insertions(+), 7 deletions(-) create mode 100644 Source/EventFlow.EventStores.KurrentDB.Tests/UnitTests/AppendRetryTests.cs diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs index 6393441bc..3e3baba81 100644 --- a/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs +++ b/Source/EventFlow.EventStores.KurrentDB.Tests/IntegrationTests/KurrentDBEventPersistenceTests.cs @@ -383,6 +383,68 @@ await writeScope.Resolve().CommitEventsAsync( } } } + [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 diff --git a/Source/EventFlow.EventStores.KurrentDB.Tests/README.md b/Source/EventFlow.EventStores.KurrentDB.Tests/README.md index c78d3f7f3..bf97cd64c 100644 --- a/Source/EventFlow.EventStores.KurrentDB.Tests/README.md +++ b/Source/EventFlow.EventStores.KurrentDB.Tests/README.md @@ -25,6 +25,12 @@ 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, @@ -124,3 +130,9 @@ 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/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/EventFlow.EventStores.KurrentDB.csproj b/Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj index e1824285d..da143bbdc 100644 --- a/Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj +++ b/Source/EventFlow.EventStores.KurrentDB/EventFlow.EventStores.KurrentDB.csproj @@ -24,6 +24,7 @@ + diff --git a/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs b/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs index 933716351..25988e9e3 100644 --- a/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs +++ b/Source/EventFlow.EventStores.KurrentDB/KurrentDBEventPersistence.cs @@ -30,12 +30,16 @@ 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; @@ -48,9 +52,22 @@ private class KurrentDBEvent : ICommittedDomainEvent } 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( @@ -79,12 +96,7 @@ public async Task> CommitEventsAsync( try { - await _client.AppendToStreamAsync( - id.Value, - expectedState, - events, - configureOperationOptions: options => options.ThrowOnAppendFailure = true, - cancellationToken: cancellationToken).ConfigureAwait(false); + await AppendWithRetryAsync(id.Value, expectedState, events, cancellationToken).ConfigureAwait(false); } catch (WrongExpectedVersionException e) { @@ -103,6 +115,31 @@ await _client.AppendToStreamAsync( }).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, diff --git a/Source/EventFlow.EventStores.KurrentDB/README.md b/Source/EventFlow.EventStores.KurrentDB/README.md index b649fe71c..ac750c62b 100644 --- a/Source/EventFlow.EventStores.KurrentDB/README.md +++ b/Source/EventFlow.EventStores.KurrentDB/README.md @@ -29,6 +29,15 @@ 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 @@ -37,7 +46,7 @@ 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 70 tests on .NET Framework 4.8.1 +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