feat: implement native RavenDB-backed control queue for Wolverine

docs/guide/durability/ravendb.md

@@ -177,9 +177,25 @@ public static class RecordTeamHandler
 
 ## System Control Queues
 
-The RavenDb integration to Wolverine does not yet come with a built in database control queue
-mechanism, so you will need to add that from external messaging brokers as in this example
-using Azure Service Bus:
+Wolverine uses a "control queue" for low latency communication between running nodes to coordinate
+[agent assignments](/guide/durability/leadership-and-troubleshooting) such as leader election and exclusive
+listeners when running in the `Balanced` durability mode. The RavenDb integration
+ships with a **native, database-backed control queue** that requires no additional configuration —
+simply calling `opts.UseRavenDbPersistence()` (as shown at the top of this page) is enough.
+
+When running in `Balanced` mode (the default), Wolverine automatically registers a RavenDb-backed control
+endpoint that stores inter-node control messages in a `ControlMessages` collection. Each node polls for
+messages targeted at its own node id, dispatches them, and deletes them once handled. Control messages are
+also given a short expiration as a safety net for undelivered messages.
+
+::: tip
+The native control queue is only activated when the durability mode is `Balanced` and you have not already
+supplied a different node control endpoint (see below). It is not used in `Solo`, `Serverless`, or
+`MediatorOnly` modes, which do not require inter-node coordination.
+:::
+
+If you would rather use an external messaging broker for the control queue, you can still opt into that.
+For example, with Azure Service Bus:
 
 <!-- snippet: sample_enabling_azure_service_bus_control_queues -->
 <a id='snippet-sample_enabling_azure_service_bus_control_queues'></a>
@@ -214,6 +230,7 @@ as a control endpoint with this configuration option:
 WolverineOptions.UseTcpForControlEndpoint();
 ```
 
+
 In the option above, Wolverine is just looking for an unused port, and assigning that found port
 as the listener for the node being bootstrapped. 
 

src/Persistence/RavenDbTests/control_queue_tests.cs

@@ -0,0 +1,134 @@
+using System.Diagnostics;
+using JasperFx.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Raven.Client.Documents;
+using Shouldly;
+using Wolverine;
+using Wolverine.RavenDb;
+using Wolverine.Runtime;
+using Wolverine.Tracking;
+
+namespace RavenDbTests;
+
+[Collection("raven")]
+public class control_queue_tests : IAsyncLifetime
+{
+    private readonly DatabaseFixture _fixture;
+    private IDocumentStore _store = null!;
+    private IHost _sender = null!;
+    private IHost _receiver = null!;
+    private Uri _receiverUri = null!;
+
+    public control_queue_tests(DatabaseFixture fixture)
+    {
+        _fixture = fixture;
+    }
+
+    public async Task InitializeAsync()
+    {
+        // Both nodes must share the same RavenDB database so that control messages
+        // written by one node are visible to the other node's control listener.
+        _store = _fixture.StartRavenStore();
+
+        _sender = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Services.AddSingleton<IDocumentStore>(_store);
+                opts.UseRavenDbPersistence();
+                opts.Durability.Mode = DurabilityMode.Balanced;
+                opts.ServiceName = "Sender";
+            }).StartAsync();
+
+        _receiver = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Services.AddSingleton<IDocumentStore>(_store);
+                opts.UseRavenDbPersistence();
+                opts.Durability.Mode = DurabilityMode.Balanced;
+                opts.ServiceName = "Receiver";
+            }).StartAsync();
+
+        var nodeId = _receiver.GetRuntime().Options.UniqueNodeId;
+        _receiverUri = new Uri($"ravencontrol://{nodeId}");
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _sender.StopAsync();
+        _sender.Dispose();
+        await _receiver.StopAsync();
+        _receiver.Dispose();
+    }
+
+    [Fact]
+    public void control_endpoint_is_wired_up_in_balanced_mode()
+    {
+        // Regression: in Balanced mode a null NodeControlEndpoint causes
+        // WolverineNode.For to throw "ControlEndpoint cannot be null for this usage".
+        var endpoint = _sender.GetRuntime().Options.Transports.NodeControlEndpoint;
+        endpoint.ShouldNotBeNull();
+        endpoint.Uri.Scheme.ShouldBe("ravencontrol");
+    }
+
+    [Fact]
+    public async Task send_message_from_one_to_another()
+    {
+        var tracked = await _sender.TrackActivity()
+            .AlsoTrack(_receiver)
+            .Timeout(30.Seconds())
+            .ExecuteAndWaitAsync(m => m.EndpointFor(_receiverUri).SendAsync(new Command(10)));
+
+        tracked.Sent.RecordsInOrder().Single(x => x.Envelope!.Message?.GetType() == typeof(Command)).ServiceName!
+            .ShouldBe("Sender");
+        tracked.Received.RecordsInOrder().Single(x => x.Envelope!.Message?.GetType() == typeof(Command))
+            .ServiceName!
+            .ShouldBe("Receiver");
+    }
+
+    [Fact]
+    public async Task request_reply_message_from_one_to_another()
+    {
+        var (tracked, result) = await _sender.TrackActivity()
+            .AlsoTrack(_receiver)
+            .Timeout(60.Seconds())
+            .InvokeAndWaitAsync<Result>(new Query(13), _receiverUri);
+
+        result!.Number.ShouldBe(13);
+
+        tracked.Sent.RecordsInOrder().Single(x => x.Envelope!.Message!.GetType() == typeof(Query)).ServiceName!
+            .ShouldBe("Sender");
+        tracked.Received.RecordsInOrder().Single(x => x.Envelope!.Message!.GetType() == typeof(Query)).ServiceName!
+            .ShouldBe("Receiver");
+
+        tracked.Sent.RecordsInOrder().Single(x => x.Envelope!.Message!.GetType() == typeof(Result)).ServiceName!
+            .ShouldBe("Receiver");
+        tracked.Received.RecordsInOrder().Single(x => x.Envelope!.Message!.GetType() == typeof(Result))
+            .ServiceName!
+            .ShouldBe("Sender");
+    }
+}
+
+public record Query(int Number);
+
+public record Result(int Number);
+
+public record Command(int Number);
+
+public static class QueryMessageHandler
+{
+    public static Result Handle(Query query)
+    {
+        return new Result(query.Number);
+    }
+
+    public static void Handle(Command command)
+    {
+        Debug.WriteLine($"Got command {command.Number}");
+    }
+
+    public static void Handle(Result result)
+    {
+        Debug.WriteLine($"Got result {result.Number}");
+    }
+}

src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.cs

@@ -71,7 +71,19 @@ public ValueTask DisposeAsync()
     public IDeadLetters DeadLetters => this;
     public void Initialize(IWolverineRuntime runtime)
     {
-        // NOTHING YET
+        // In Balanced mode Wolverine requires a node control endpoint so nodes can
+        // exchange agent-coordination commands. Without this the runtime throws
+        // "ControlEndpoint cannot be null for this usage" from WolverineNode.For.
+        // Register a native RavenDB-backed control queue unless the user already
+        // supplied one (e.g. an external broker or UseTcpForControlEndpoint()).
+        if (Role == MessageStoreRole.Main
+            && runtime.Options.Transports.NodeControlEndpoint == null
+            && runtime.Options.Durability.Mode == DurabilityMode.Balanced)
+        {
+            var transport = new Transport.RavenDbControlTransport(_store, runtime.Options);
+            runtime.Options.Transports.Add(transport);
+            runtime.Options.Transports.NodeControlEndpoint = transport.ControlEndpoint;
+        }
     }
 
     [UnconditionalSuppressMessage("Trimming", "IL2026",

src/Persistence/Wolverine.RavenDb/Internals/Transport/ControlMessage.cs

@@ -0,0 +1,32 @@
+namespace Wolverine.RavenDb.Internals.Transport;
+
+/// <summary>
+/// RavenDB document representing a single Wolverine control message targeted at a
+/// specific node. Stored in the ControlMessages collection and polled by the
+/// receiving node's <see cref="RavenDbControlListener"/>.
+/// </summary>
+internal class ControlMessage
+{
+    public ControlMessage()
+    {
+    }
+
+    public ControlMessage(Envelope envelope, Guid nodeId, byte[] body, DateTimeOffset expires)
+    {
+        Id = IdFor(envelope.Id);
+        NodeId = nodeId;
+        MessageType = envelope.MessageType!;
+        Body = body;
+        Expires = expires;
+        Posted = DateTimeOffset.UtcNow;
+    }
+
+    public string Id { get; set; } = string.Empty;
+    public Guid NodeId { get; set; }
+    public string MessageType { get; set; } = string.Empty;
+    public byte[] Body { get; set; } = [];
+    public DateTimeOffset Expires { get; set; }
+    public DateTimeOffset Posted { get; set; }
+
+    public static string IdFor(Guid envelopeId) => $"ControlMessages/{envelopeId}";
+}

src/Persistence/Wolverine.RavenDb/Internals/Transport/RavenDbControlEndpoint.cs

@@ -0,0 +1,47 @@
+using Microsoft.Extensions.Logging;
+using Wolverine.Configuration;
+using Wolverine.Runtime;
+using Wolverine.Transports;
+using Wolverine.Transports.Sending;
+
+namespace Wolverine.RavenDb.Internals.Transport;
+
+internal class RavenDbControlEndpoint : Endpoint
+{
+    private readonly RavenDbControlTransport _parent;
+
+    public RavenDbControlEndpoint(RavenDbControlTransport parent, Guid nodeId) : base(
+        new Uri($"{RavenDbControlTransport.ProtocolName}://{nodeId}"), EndpointRole.System)
+    {
+        NodeId = nodeId;
+        _parent = parent;
+        Mode = EndpointMode.BufferedInMemory;
+        MaxDegreeOfParallelism = 1;
+        BrokerRole = "queue";
+
+        // No otel for this one!
+        TelemetryEnabled = false;
+    }
+
+    public Guid NodeId { get; }
+
+    // The control transport polls the RavenDB ControlMessages collection on a 1s
+    // tick and dispatches inter-node agent commands. Forcing it to Durable would
+    // route every control envelope through the same store-backed inbox/outbox the
+    // durability agent itself owns (deadlock); forcing Inline would skip the
+    // batched-poll semantics. Lock to BufferedInMemory regardless of any global
+    // endpoint policy.
+    protected override bool supportsMode(EndpointMode mode) => mode == EndpointMode.BufferedInMemory;
+
+    public override ValueTask<IListener> BuildListenerAsync(IWolverineRuntime runtime, IReceiver receiver)
+    {
+        return new ValueTask<IListener>(new RavenDbControlListener(_parent, this, receiver,
+            runtime.LoggerFactory.CreateLogger<RavenDbControlListener>(), runtime.Options.Durability.Cancellation));
+    }
+
+    protected override ISender CreateSender(IWolverineRuntime runtime)
+    {
+        return new RavenDbControlSender(this, _parent, runtime.LoggerFactory.CreateLogger<RavenDbControlSender>(),
+            runtime.Options.Durability.Cancellation);
+    }
+}

src/Persistence/Wolverine.RavenDb/Internals/Transport/RavenDbControlListener.cs

@@ -0,0 +1,116 @@
+using JasperFx.Blocks;
+using JasperFx.Core;
+using Microsoft.Extensions.Logging;
+using Raven.Client.Documents;
+using Raven.Client.Documents.Session;
+using Wolverine.Runtime;
+using Wolverine.Runtime.Serialization;
+using Wolverine.Transports;
+
+namespace Wolverine.RavenDb.Internals.Transport;
+
+internal class RavenDbControlListener : IListener
+{
+    private readonly CancellationTokenSource _cancellation;
+    private readonly IReceiver _receiver;
+    private readonly Task _receivingLoop;
+    private readonly RetryBlock<Envelope> _retryBlock;
+    private readonly RavenDbControlTransport _transport;
+
+    public RavenDbControlListener(RavenDbControlTransport transport, RavenDbControlEndpoint endpoint,
+        IReceiver receiver, ILogger<RavenDbControlListener> logger, CancellationToken cancellationToken)
+    {
+        _transport = transport;
+        _receiver = receiver;
+
+        _cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+
+        Address = endpoint.Uri;
+
+        _retryBlock = new RetryBlock<Envelope>(deleteEnvelopeAsync, logger, cancellationToken);
+
+        _receivingLoop = Task.Run(async () =>
+        {
+            await Task.Delay(Random.Shared.Next(100, 1000).Milliseconds(), _cancellation.Token);
+
+            while (!_cancellation.IsCancellationRequested)
+            {
+                try
+                {
+                    await pollForMessagesAsync();
+                }
+                catch (OperationCanceledException)
+                {
+                    // Shutting down
+                }
+                catch (Exception e)
+                {
+                    logger.LogError(e, "Error trying to poll for messages from the RavenDb control queue");
+                }
+
+                await Task.Delay(1.Seconds(), _cancellation.Token);
+            }
+        }, _cancellation.Token);
+    }
+
+    public IHandlerPipeline? Pipeline => _receiver.Pipeline;
+
+    public async ValueTask CompleteAsync(Envelope envelope)
+    {
+        await _retryBlock.PostAsync(envelope);
+    }
+
+    public ValueTask DeferAsync(Envelope envelope)
+    {
+        return ValueTask.CompletedTask;
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _cancellation.CancelAsync();
+        _receivingLoop.SafeDispose();
+        _retryBlock.SafeDispose();
+    }
+
+    public Uri Address { get; }
+
+    public async ValueTask StopAsync()
+    {
+        await _cancellation.CancelAsync();
+    }
+
+    private async Task pollForMessagesAsync()
+    {
+        var nodeId = _transport.Options.UniqueNodeId;
+
+        using var session = _transport.Store.OpenAsyncSession();
+
+        var messages = await session
+            .Query<ControlMessage>()
+            .Customize(x => x.WaitForNonStaleResults())
+            .Where(x => x.NodeId == nodeId)
+            .ToListAsync(_cancellation.Token);
+
+        if (messages.Count == 0)
+        {
+            return;
+        }
+
+        var envelopes = new List<Envelope>();
+        foreach (var message in messages)
+        {
+            var envelope = EnvelopeSerializer.Deserialize(message.Body);
+            envelopes.Add(envelope);
+        }
+
+        await _receiver.ReceivedAsync(this, envelopes.ToArray());
+        await _transport.DeleteEnvelopesAsync(envelopes, _cancellation.Token);
+    }
+
+    private async Task deleteEnvelopeAsync(Envelope envelope, CancellationToken cancellationToken)
+    {
+        using var session = _transport.Store.OpenAsyncSession();
+        session.Delete(ControlMessage.IdFor(envelope.Id));
+        await session.SaveChangesAsync(cancellationToken);
+    }
+}