Skip to content

Commit 124b0b5

Browse files
danielmarbachdanielmarbach
authored
Add support for eventual consistent reads on sagas (#855)
* Support for eventual consistent reads on saga data * UseEventuallyConsistentReads --------- Co-authored-by: Daniel Marbach <danielmarbach@users.noreply.github.com>
1 parent 043dde0 commit 124b0b5

11 files changed

Lines changed: 228 additions & 29 deletions

File tree

‎src/NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests/.editorconfig‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
[*.cs]
2+
3+
# Justification: Test project
4+
dotnet_diagnostic.CA2007.severity = none
5+
6+
# Justification: Tests don't support cancellation and don't need to forward IMessageHandlerContext.CancellationToken
7+
dotnet_diagnostic.NSB0002.severity = suggestion
8+
dotnet_diagnostic.PS0018.severity = suggestion
9+
dotnet_diagnostic.PS0013.severity = suggestion
10+
11+
# Persistence library doesn't need saga analyzers
12+
dotnet_diagnostic.NSB0004.severity = none

‎src/NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests/ConfigureEndpointDynamoDBPersistence.cs‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
using System.Threading.Tasks;
2+
using NServiceBus;
3+
using NServiceBus.AcceptanceTesting.Support;
4+
using NServiceBus.AcceptanceTests;
5+
using NServiceBus.Configuration.AdvancedExtensibility;
6+
7+
public class ConfigureEndpointDynamoDBPersistence : IConfigureEndpointTestExecution
8+
{
9+
public Task Configure(string endpointName, EndpointConfiguration configuration, RunSettings settings, PublisherMetadata publisherMetadata)
10+
{
11+
if (configuration.GetSettings().Get<bool>("Endpoint.SendOnly"))
12+
{
13+
return Task.CompletedTask;
14+
}
15+
16+
// disable installers which are enabled by default in the standard endpoint templates
17+
configuration.GetSettings().Set("Installers.Enable", false);
18+
19+
var persistence = configuration.UsePersistence<DynamoPersistence>();
20+
persistence.DynamoClient(SetupFixture.DynamoDBClient);
21+
persistence.UseSharedTable(SetupFixture.TableConfiguration);
22+
23+
var sagas = persistence.Sagas();
24+
sagas.UseEventuallyConsistentReads = true;
25+
26+
return Task.CompletedTask;
27+
}
28+
29+
public Task Cleanup() => Task.CompletedTask;
30+
}

‎src/NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests/NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests.csproj‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<TargetFrameworks>net8.0;net9.0</TargetFrameworks>
5+
<SignAssembly>true</SignAssembly>
6+
<AssemblyOriginatorKeyFile>..\NServiceBusTests.snk</AssemblyOriginatorKeyFile>
7+
</PropertyGroup>
8+
9+
<ItemGroup>
10+
<ProjectReference Include="..\NServiceBus.Persistence.DynamoDB\NServiceBus.Persistence.DynamoDB.csproj" />
11+
</ItemGroup>
12+
13+
<ItemGroup>
14+
<PackageReference Include="GitHubActionsTestLogger" Version="2.4.1" />
15+
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0" />
16+
<PackageReference Include="NServiceBus.AcceptanceTests.Sources" Version="9.2.4" GeneratePathProperty="true" />
17+
<PackageReference Include="NUnit" Version="4.3.2" />
18+
<PackageReference Include="NUnit.Analyzers" Version="4.7.0" />
19+
<PackageReference Include="NUnit3TestAdapter" Version="4.6.0" />
20+
</ItemGroup>
21+
22+
<ItemGroup>
23+
<Compile Include="..\NServiceBus.Persistence.DynamoDB.Tests\ClientFactory.cs" />
24+
</ItemGroup>
25+
26+
<ItemGroup Condition="'$(PkgNServiceBus_AcceptanceTests_Sources)' != ''">
27+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\ConfigureEndpointInMemoryPersistence.cs" />
28+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\ConfigureEndpointLearningPersistence.cs" />
29+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\ConventionEnforcementTests.cs" />
30+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\DeterministicGuid.cs" />
31+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Audit\*.*" />
32+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Correlation\*.*" />
33+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\DataBus\*.*" />
34+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\DelayedDelivery\*.*" />
35+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Forwarding\*.*" />
36+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Feature\*.*" />
37+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\MessageId\*.*" />
38+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Pipeline\*.*" />
39+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Recoverability\*.*" />
40+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Routing\**\*.*" />
41+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Satellites\*.*" />
42+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Scheduling\*.*" />
43+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\SelfVerification\*.*" />
44+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Serialization\*.*" />
45+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Timeout\*.*" />
46+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\TimeToBeReceived\*.*" />
47+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Tx\**\*.*" />
48+
<Compile Remove="$(PkgNServiceBus_AcceptanceTests_Sources)\**\Versioning\*.*" />
49+
</ItemGroup>
50+
51+
</Project>

‎src/NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests/SetupFixture.cs‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
namespace NServiceBus.AcceptanceTests;
2+
3+
using System;
4+
using System.IO;
5+
using System.Threading.Tasks;
6+
using Amazon.DynamoDBv2;
7+
using NUnit.Framework;
8+
using Persistence.DynamoDB;
9+
using Persistence.DynamoDB.Tests;
10+
11+
[SetUpFixture]
12+
public class SetupFixture
13+
{
14+
[OneTimeSetUp]
15+
public async Task OneTimeSetUp()
16+
{
17+
TableConfiguration = new TableConfiguration
18+
{
19+
TableName = $"{DateTime.UtcNow.Ticks}_{Path.GetFileNameWithoutExtension(Path.GetTempFileName())}",
20+
};
21+
22+
DynamoDBClient = ClientFactory.CreateDynamoDBClient();
23+
24+
var installer = new Installer(DynamoDBClient);
25+
26+
await installer.CreateTable(TableConfiguration);
27+
}
28+
29+
[OneTimeTearDown]
30+
public async Task OneTimeTearDown()
31+
{
32+
await DynamoDBClient.DeleteTableAsync(TableConfiguration.TableName);
33+
DynamoDBClient.Dispose();
34+
}
35+
36+
public static IAmazonDynamoDB DynamoDBClient;
37+
public static TableConfiguration TableConfiguration;
38+
}

‎src/NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests/TestSuiteConstraints.cs‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
namespace NServiceBus.AcceptanceTests;
2+
3+
using AcceptanceTesting.Support;
4+
5+
public partial class TestSuiteConstraints
6+
{
7+
public bool SupportsDtc { get; } = false;
8+
public bool SupportsCrossQueueTransactions { get; } = true;
9+
public bool SupportsNativePubSub { get; } = true;
10+
public bool SupportsOutbox { get; } = true;
11+
public bool SupportsDelayedDelivery { get; } = true;
12+
public bool SupportsPurgeOnStartup { get; } = true;
13+
14+
public IConfigureEndpointTestExecution CreateTransportConfiguration() => new ConfigureEndpointAcceptanceTestingTransport(true, true, TransportTransactionMode.ReceiveOnly);
15+
16+
public IConfigureEndpointTestExecution CreatePersistenceConfiguration() => new ConfigureEndpointDynamoDBPersistence();
17+
}

‎src/NServiceBus.Persistence.DynamoDB.PersistenceTests/PersistenceTestsConfiguration.cs‎

Lines changed: 29 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -14,26 +14,20 @@ public partial class PersistenceTestsConfiguration : IDynamoClientProvider
1414
{
1515
static PersistenceTestsConfiguration()
1616
{
17-
SagaVariants = new[]
18-
{
19-
new TestFixtureData(new TestVariant(new PersistenceConfiguration(usePessimisticLocking: false))).SetArgDisplayNames("Optimistic"),
20-
new TestFixtureData(new TestVariant(new PersistenceConfiguration(usePessimisticLocking: true))).SetArgDisplayNames("Pessimistic"),
21-
};
22-
23-
OutboxVariants = new[]
24-
{
25-
new TestFixtureData(new TestVariant(new PersistenceConfiguration(usePessimisticLocking: false))).SetArgDisplayNames("Optimistic"),
26-
};
17+
SagaVariants =
18+
[
19+
new TestFixtureData(new TestVariant(new PersistenceConfiguration())).SetArgDisplayNames("Optimistic"),
20+
new TestFixtureData(new TestVariant(new PersistenceConfiguration(UseEventuallyConsistentReads: true))).SetArgDisplayNames("Optimistic Eventual Consistent"),
21+
new TestFixtureData(new TestVariant(new PersistenceConfiguration(UsePessimisticLocking: true))).SetArgDisplayNames("Pessimistic")
22+
];
23+
24+
OutboxVariants =
25+
[
26+
new TestFixtureData(new TestVariant(new PersistenceConfiguration())).SetArgDisplayNames("Optimistic"),
27+
];
2728
}
28-
public class PersistenceConfiguration
29-
{
30-
public readonly bool UsePessimisticLocking;
3129

32-
public PersistenceConfiguration(bool usePessimisticLocking)
33-
{
34-
UsePessimisticLocking = usePessimisticLocking;
35-
}
36-
}
30+
public record PersistenceConfiguration(bool? UsePessimisticLocking = null, bool? UseEventuallyConsistentReads = null);
3731

3832
public bool SupportsDtc => false;
3933

@@ -57,14 +51,25 @@ public Task Configure(CancellationToken cancellationToken = default)
5751
{
5852
var configuration = (PersistenceConfiguration)Variant.Values[0];
5953

54+
var sagaPersistenceConfiguration = new SagaPersistenceConfiguration
55+
{
56+
Table = SetupFixture.SagaTable,
57+
LeaseAcquisitionTimeout = Variant.SessionTimeout ?? TimeSpan.FromSeconds(10)
58+
};
59+
60+
if (configuration.UsePessimisticLocking.HasValue)
61+
{
62+
sagaPersistenceConfiguration.UsePessimisticLocking = SupportsPessimisticConcurrency = configuration.UsePessimisticLocking.Value;
63+
}
64+
65+
if (configuration.UseEventuallyConsistentReads.HasValue)
66+
{
67+
sagaPersistenceConfiguration.UseEventuallyConsistentReads = configuration.UseEventuallyConsistentReads.Value;
68+
}
69+
6070
SagaStorage = new SagaPersister(
6171
Client,
62-
new SagaPersistenceConfiguration
63-
{
64-
Table = SetupFixture.SagaTable,
65-
UsePessimisticLocking = SupportsPessimisticConcurrency = configuration.UsePessimisticLocking,
66-
LeaseAcquisitionTimeout = Variant.SessionTimeout ?? TimeSpan.FromSeconds(10)
67-
},
72+
sagaPersistenceConfiguration,
6873
"PersistenceTest");
6974

7075
OutboxStorage = new OutboxPersister(

‎src/NServiceBus.Persistence.DynamoDB.Tests/ApprovalFiles/APIApprovals.Approve.approved.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
[assembly: System.Runtime.CompilerServices.InternalsVisibleTo(@"NServiceBus.Persistence.DynamoDB.AcceptanceTests, PublicKey=00240000048000009400000006020000002400005253413100040000010001007f16e21368ff041183fab592d9e8ed37e7be355e93323147a1d29983d6e591b04282e4da0c9e18bd901e112c0033925eb7d7872c2f1706655891c5c9d57297994f707d16ee9a8f40d978f064ee1ffc73c0db3f4712691b23bf596f75130f4ec978cf78757ec034625a5f27e6bb50c618931ea49f6f628fd74271c32959efb1c5")]
2+
[assembly: System.Runtime.CompilerServices.InternalsVisibleTo(@"NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests, PublicKey=00240000048000009400000006020000002400005253413100040000010001007f16e21368ff041183fab592d9e8ed37e7be355e93323147a1d29983d6e591b04282e4da0c9e18bd901e112c0033925eb7d7872c2f1706655891c5c9d57297994f707d16ee9a8f40d978f064ee1ffc73c0db3f4712691b23bf596f75130f4ec978cf78757ec034625a5f27e6bb50c618931ea49f6f628fd74271c32959efb1c5")]
23
[assembly: System.Runtime.CompilerServices.InternalsVisibleTo(@"NServiceBus.Persistence.DynamoDB.PersistenceTests, PublicKey=00240000048000009400000006020000002400005253413100040000010001007f16e21368ff041183fab592d9e8ed37e7be355e93323147a1d29983d6e591b04282e4da0c9e18bd901e112c0033925eb7d7872c2f1706655891c5c9d57297994f707d16ee9a8f40d978f064ee1ffc73c0db3f4712691b23bf596f75130f4ec978cf78757ec034625a5f27e6bb50c618931ea49f6f628fd74271c32959efb1c5")]
34
[assembly: System.Runtime.CompilerServices.InternalsVisibleTo(@"NServiceBus.Persistence.DynamoDB.PessimisticLock.AcceptanceTests, PublicKey=00240000048000009400000006020000002400005253413100040000010001007f16e21368ff041183fab592d9e8ed37e7be355e93323147a1d29983d6e591b04282e4da0c9e18bd901e112c0033925eb7d7872c2f1706655891c5c9d57297994f707d16ee9a8f40d978f064ee1ffc73c0db3f4712691b23bf596f75130f4ec978cf78757ec034625a5f27e6bb50c618931ea49f6f628fd74271c32959efb1c5")]
45
[assembly: System.Runtime.CompilerServices.InternalsVisibleTo(@"NServiceBus.Persistence.DynamoDB.Tests, PublicKey=00240000048000009400000006020000002400005253413100040000010001007f16e21368ff041183fab592d9e8ed37e7be355e93323147a1d29983d6e591b04282e4da0c9e18bd901e112c0033925eb7d7872c2f1706655891c5c9d57297994f707d16ee9a8f40d978f064ee1ffc73c0db3f4712691b23bf596f75130f4ec978cf78757ec034625a5f27e6bb50c618931ea49f6f628fd74271c32959efb1c5")]
@@ -40,6 +41,7 @@ namespace NServiceBus
4041
public System.TimeSpan LeaseDuration { get; set; }
4142
public System.Text.Json.JsonSerializerOptions MapperOptions { get; set; }
4243
public NServiceBus.TableConfiguration Table { get; set; }
44+
public bool UseEventuallyConsistentReads { get; set; }
4345
public bool UsePessimisticLocking { get; set; }
4446
}
4547
public static class SynchronizedStorageSessionExtensions

‎src/NServiceBus.Persistence.DynamoDB.sln‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NServiceBus.Persistence.Dyn
2626
EndProject
2727
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NServiceBus.Persistence.DynamoDB.TransactionalSession.Tests", "NServiceBus.Persistence.DynamoDB.TransactionalSession.Tests\NServiceBus.Persistence.DynamoDB.TransactionalSession.Tests.csproj", "{EB66E88B-5B05-4D26-8FEF-739E325A14D6}"
2828
EndProject
29+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests", "NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests\NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests.csproj", "{5C63649F-25A3-4568-95EC-AF98761992EC}"
30+
EndProject
2931
Global
3032
GlobalSection(SolutionConfigurationPlatforms) = preSolution
3133
Debug|Any CPU = Debug|Any CPU
@@ -64,6 +66,10 @@ Global
6466
{EB66E88B-5B05-4D26-8FEF-739E325A14D6}.Debug|Any CPU.Build.0 = Debug|Any CPU
6567
{EB66E88B-5B05-4D26-8FEF-739E325A14D6}.Release|Any CPU.ActiveCfg = Release|Any CPU
6668
{EB66E88B-5B05-4D26-8FEF-739E325A14D6}.Release|Any CPU.Build.0 = Release|Any CPU
69+
{5C63649F-25A3-4568-95EC-AF98761992EC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
70+
{5C63649F-25A3-4568-95EC-AF98761992EC}.Debug|Any CPU.Build.0 = Debug|Any CPU
71+
{5C63649F-25A3-4568-95EC-AF98761992EC}.Release|Any CPU.ActiveCfg = Release|Any CPU
72+
{5C63649F-25A3-4568-95EC-AF98761992EC}.Release|Any CPU.Build.0 = Release|Any CPU
6773
EndGlobalSection
6874
GlobalSection(SolutionProperties) = preSolution
6975
HideSolutionNode = FALSE

‎src/NServiceBus.Persistence.DynamoDB/NServiceBus.Persistence.DynamoDB.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
<ItemGroup>
2121
<InternalsVisibleTo Include="NServiceBus.Persistence.DynamoDB.AcceptanceTests" Key="$(NServiceBusTestsKey)" />
22+
<InternalsVisibleTo Include="NServiceBus.Persistence.DynamoDB.EventualConsistent.AcceptanceTests" Key="$(NServiceBusTestsKey)" />
2223
<InternalsVisibleTo Include="NServiceBus.Persistence.DynamoDB.PersistenceTests" Key="$(NServiceBusTestsKey)" />
2324
<InternalsVisibleTo Include="NServiceBus.Persistence.DynamoDB.PessimisticLock.AcceptanceTests" Key="$(NServiceBusTestsKey)" />
2425
<InternalsVisibleTo Include="NServiceBus.Persistence.DynamoDB.Tests" Key="$(NServiceBusTestsKey)" />

‎src/NServiceBus.Persistence.DynamoDB/Saga/SagaPersistenceConfiguration.cs‎

Lines changed: 39 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,15 +12,52 @@ public class SagaPersistenceConfiguration
1212
/// <summary>
1313
/// The configuration of the table used by the outbox persistence.
1414
/// </summary>
15-
public TableConfiguration Table { get; set; } = new TableConfiguration()
15+
public TableConfiguration Table { get; set; } = new()
1616
{
1717
TimeToLiveAttributeName = null
1818
};
1919

20+
/// <summary>
21+
/// Enables eventual consistent reads on table containing the saga data. This might reduce costs for reads since
22+
/// eventual consistent reads are cheaper than strongly consistent reads. However, it might lead to stale data being read which can lead
23+
/// too more retries in case of concurrent updates to the same saga.
24+
/// </summary>
25+
/// <remarks>This setting is mutually exclusive to <see cref="UsePessimisticLocking"/> meaning when opting into eventual
26+
/// consistent reads pessimistic locking is disabled when previously explicitly enabled and vice versa.</remarks>
27+
public bool UseEventuallyConsistentReads
28+
{
29+
get => useEventuallyConsistentReads.GetValueOrDefault(false);
30+
set
31+
{
32+
useEventuallyConsistentReads = value;
33+
if (usePessimisticLocking.HasValue)
34+
{
35+
usePessimisticLocking = !value;
36+
}
37+
}
38+
}
39+
40+
bool? useEventuallyConsistentReads;
41+
2042
/// <summary>
2143
/// Enables pessimistic locking mode to avoid concurrent modifications to the same saga. Enable this mode to reduce retries due to optimistic concurrency control violations.
2244
/// </summary>
23-
public bool UsePessimisticLocking { get; set; } = false;
45+
/// <remarks>This setting is mutually exclusive to <see cref="UseEventuallyConsistentReads"/> meaning when opting into pessimistic locking
46+
/// eventual consistent reads are disabled when previously explicitly enabled and vice versa.</remarks>
47+
public bool UsePessimisticLocking
48+
{
49+
get => usePessimisticLocking.GetValueOrDefault(false);
50+
set
51+
{
52+
usePessimisticLocking = value;
53+
if (useEventuallyConsistentReads.HasValue)
54+
{
55+
useEventuallyConsistentReads = !value;
56+
}
57+
}
58+
}
59+
60+
bool? usePessimisticLocking;
2461

2562
/// <summary>
2663
/// Determines whether the NServiceBus installer should create the Outbox table when enabled.

0 commit comments

Comments
 (0)