diff --git a/src/Persistence/RavenDbTests/node_persistence_compliance.cs b/src/Persistence/RavenDbTests/node_persistence_compliance.cs index 640818ef9..3b09e56a8 100644 --- a/src/Persistence/RavenDbTests/node_persistence_compliance.cs +++ b/src/Persistence/RavenDbTests/node_persistence_compliance.cs @@ -1,7 +1,9 @@ using Raven.Embedded; +using Shouldly; using Wolverine; using Wolverine.ComplianceTests; using Wolverine.Persistence.Durability; +using Wolverine.Runtime.Agents; using Wolverine.RavenDb.Internals; namespace RavenDbTests; @@ -21,4 +23,38 @@ protected override async Task buildCleanMessageStore() var store = _fixture.StartRavenStore(); return new RavenDbMessageStore(store, new WolverineOptions()); } -} \ No newline at end of file + + [Fact] + public async Task concurrently_persisting_nodes_assigns_unique_node_numbers() + { + await using var messageStore = await buildCleanMessageStore(); + + var start = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var nodes = Enumerable.Range(0, 20).Select(_ => + { + var id = Guid.NewGuid(); + return new WolverineNode + { + NodeId = id, + ControlUri = new Uri($"dbcontrol://{id}"), + Description = Environment.MachineName, + Version = new Version(1, 2, 3, 0) + }; + }).ToArray(); + + var tasks = nodes.Select(async node => + { + await start.Task; + return await messageStore.Nodes.PersistAsync(node, CancellationToken.None); + }).ToArray(); + + start.SetResult(); + + var assignedNodeNumbers = await Task.WhenAll(tasks); + + assignedNodeNumbers.OrderBy(x => x).ShouldBe(Enumerable.Range(1, nodes.Length).ToArray()); + + var persisted = await messageStore.Nodes.LoadAllNodesAsync(CancellationToken.None); + persisted.Select(x => x.AssignedNodeNumber).OrderBy(x => x).ShouldBe(assignedNodeNumbers.OrderBy(x => x)); + } +} diff --git a/src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.NodeAgents.cs b/src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.NodeAgents.cs index 1fc2dc56e..e5cf7dc60 100644 --- a/src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.NodeAgents.cs +++ b/src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.NodeAgents.cs @@ -3,12 +3,16 @@ using Raven.Client.Documents.Operations; using Raven.Client.Documents.Queries; using Raven.Client.Documents.Session; +using Raven.Client.Exceptions; +using Raven.Client.Exceptions.Documents.Session; using Wolverine.Runtime.Agents; namespace Wolverine.RavenDb.Internals; public partial class RavenDbMessageStore : INodeAgentPersistence { + private const int NodePersistenceMaxAttempts = 25; + public async Task ClearAllAsync(CancellationToken cancellationToken) { // Shouldn't really get called at runtime, so we're doing it crudely @@ -29,24 +33,55 @@ public async Task ClearAllAsync(CancellationToken cancellationToken) public async Task PersistAsync(WolverineNode node, CancellationToken cancellationToken) { - using var session = _store.OpenAsyncSession(new SessionOptions + Exception? lastConflict = null; + + for (var attempt = 1; attempt <= NodePersistenceMaxAttempts; attempt++) { - TransactionMode = TransactionMode.ClusterWide - }); - // RavenDB rejects cluster-wide transactions combined with optimistic concurrency. - // Disable it explicitly so a consumer-enabled convention doesn't break this session. - session.Advanced.UseOptimisticConcurrency = false; + try + { + using var session = _store.OpenAsyncSession(new SessionOptions + { + TransactionMode = TransactionMode.ClusterWide + }); + // RavenDB rejects cluster-wide transactions combined with optimistic concurrency. + // Disable it explicitly so a consumer-enabled convention doesn't break this session. + session.Advanced.UseOptimisticConcurrency = false; - var sequence = await session.LoadAsync(NodeSequence.SequenceId, cancellationToken); - sequence ??= new NodeSequence(); + var sequence = await session.LoadAsync(NodeSequence.SequenceId, cancellationToken); + sequence ??= new NodeSequence(); - node.AssignedNodeNumber = ++sequence.Count; + node.AssignedNodeNumber = ++sequence.Count; - await session.StoreAsync(sequence, cancellationToken); - await session.StoreAsync(node, cancellationToken); - await session.SaveChangesAsync(cancellationToken); + await session.StoreAsync(sequence, cancellationToken); + await session.StoreAsync(node, cancellationToken); + await session.SaveChangesAsync(cancellationToken); + + return node.AssignedNodeNumber; + } + catch (Exception e) when (isNodeSequenceConcurrencyConflict(e)) + { + lastConflict = e; + + if (attempt == NodePersistenceMaxAttempts) + { + break; + } - return node.AssignedNodeNumber; + // Linear backoff with jitter to avoid synchronized retries (thundering herd) + // when many nodes contend for the sequence at startup. + var jitter = Random.Shared.Next(0, 20); + await Task.Delay(TimeSpan.FromMilliseconds(20 * attempt + jitter), cancellationToken); + } + } + + throw new InvalidOperationException( + $"Unable to persist Wolverine node {node.NodeId} after {NodePersistenceMaxAttempts} attempts because RavenDB node sequence allocation kept conflicting.", + lastConflict); + } + + private static bool isNodeSequenceConcurrencyConflict(Exception exception) + { + return exception is ClusterTransactionConcurrencyException or ConcurrencyException; } public async Task DeleteAsync(Guid nodeId, int assignedNodeNumber)