fix(RavenDB): Node sequence startup race

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<IMessageStore> buildCleanMessageStore()
         var store = _fixture.StartRavenStore();
         return new RavenDbMessageStore(store, new WolverineOptions());
     }
-}
+
+    [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));
+    }
+}

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<int> 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>(NodeSequence.SequenceId, cancellationToken);
-        sequence ??= new NodeSequence();
+                var sequence = await session.LoadAsync<NodeSequence>(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)