RavenDb control transport: TransportCompliance + leadership compliance, make ravendb:// first-class

src/Persistence/LeaderElection/RavenDbTests.LeaderElection/control_queue_leadership_election_compliance.cs

@@ -0,0 +1,45 @@
+using Microsoft.Extensions.DependencyInjection;
+using Raven.Client.Documents;
+using Wolverine;
+using Wolverine.ComplianceTests;
+using Wolverine.RavenDb;
+using Xunit.Abstractions;
+using RavenDbTests;
+
+namespace RavenDbTests.LeaderElection;
+
+/// <summary>
+/// Runs the full leadership-election compliance battery while relying on the
+/// native RavenDB control queue (ravendb://) for inter-node control
+/// messaging, rather than <c>UseTcpForControlEndpoint()</c>. This exercises the
+/// RavenDbControlTransport added in #3285 under real multi-node fan-out,
+/// send-to-node, and leader-failover scenarios.
+/// </summary>
+[Collection("raven")]
+public class control_queue_leadership_election_compliance : LeadershipElectionCompliance
+{
+    private readonly DatabaseFixture _fixture;
+    private IDocumentStore _store = null!;
+
+    public control_queue_leadership_election_compliance(ITestOutputHelper output, DatabaseFixture fixture) : base(output)
+    {
+        _fixture = fixture;
+    }
+
+    protected override Task beforeBuildingHost()
+    {
+        _store = _fixture.StartRavenStore();
+        return Task.CompletedTask;
+    }
+
+    protected override void configureNode(WolverineOptions options)
+    {
+        // Deliberately NOT calling UseTcpForControlEndpoint() — the RavenDb
+        // persistence registers its native control queue transport as the
+        // NodeControlEndpoint when none is otherwise supplied.
+        options.ServiceName = "raven";
+
+        options.Services.AddSingleton(_store);
+        options.UseRavenDbPersistence();
+    }
+}

src/Persistence/RavenDbTests/control_queue_tests.cs

@@ -50,7 +50,7 @@ public async Task InitializeAsync()
             }).StartAsync();
 
         var nodeId = _receiver.GetRuntime().Options.UniqueNodeId;
-        _receiverUri = new Uri($"ravencontrol://{nodeId}");
+        _receiverUri = new Uri($"ravendb://{nodeId}");
     }
 
     public async Task DisposeAsync()
@@ -68,7 +68,7 @@ public void control_endpoint_is_wired_up_in_balanced_mode()
         // 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");
+        endpoint.Uri.Scheme.ShouldBe("ravendb");
     }
 
     [Fact]

src/Persistence/RavenDbTests/control_transport_compliance.cs

@@ -0,0 +1,79 @@
+using JasperFx.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Raven.Client.Documents;
+using Wolverine;
+using Wolverine.ComplianceTests.Compliance;
+using Wolverine.RavenDb;
+using Wolverine.Runtime;
+using Xunit;
+
+namespace RavenDbTests;
+
+public class RavenDbControlTransportFixture : TransportComplianceFixture, IAsyncLifetime
+{
+    private DatabaseFixture _databases = null!;
+    private IDocumentStore _store = null!;
+
+    public RavenDbControlTransportFixture() : base(new Uri("ravendb://placeholder"), 30)
+    {
+        // The RavenDb control queue only registers its NodeControlEndpoint under
+        // Balanced durability, so the compliance hosts must run Balanced.
+        Mode = DurabilityMode.Balanced;
+
+        // Control messages are transient and each test tracks its own envelopes;
+        // resetting the shared store between every test would wipe the running
+        // nodes' identity/leadership records out from under them.
+        MustReset = false;
+    }
+
+    public async Task InitializeAsync()
+    {
+        _databases = new DatabaseFixture();
+        _store = _databases.StartRavenStore();
+
+        await ReceiverIs(opts =>
+        {
+            opts.Services.AddSingleton(_store);
+            opts.UseRavenDbPersistence();
+            tightenClusterCadence(opts);
+        });
+
+        var receiverNodeId = Receiver.Services.GetRequiredService<IWolverineRuntime>().Options.UniqueNodeId;
+        OutboundAddress = new Uri($"ravendb://{receiverNodeId}");
+
+        await SenderIs(opts =>
+        {
+            opts.Services.AddSingleton(_store);
+            opts.UseRavenDbPersistence();
+            tightenClusterCadence(opts);
+        });
+    }
+
+    // The compliance battery runs against a freshly-started two-node Balanced
+    // cluster. With the production defaults (CheckAssignmentPeriod 30s,
+    // ScheduledJobPollingTime 5s) the leader-assigned scheduled-job agent may not
+    // even be assigned inside a single test's timeout — which is what makes
+    // schedule_send flap. Tighten the coordination cadence so agent assignment and
+    // scheduled-message release happen promptly, mirroring the leadership suite.
+    private static void tightenClusterCadence(WolverineOptions opts)
+    {
+        opts.Durability.CheckAssignmentPeriod = 1.Seconds();
+        opts.Durability.HealthCheckPollingTime = 1.Seconds();
+        opts.Durability.ScheduledJobPollingTime = 1.Seconds();
+        opts.Durability.ScheduledJobFirstExecution = 0.Seconds();
+    }
+
+    protected override Task AfterDisposeAsync()
+    {
+        _store?.Dispose();
+        _databases?.Dispose();
+        return Task.CompletedTask;
+    }
+
+    // Satisfy IAsyncLifetime; real teardown (stopping hosts + store cleanup) runs
+    // through the base IAsyncDisposable.DisposeAsync/AfterDisposeAsync path.
+    public new Task DisposeAsync() => Task.CompletedTask;
+}
+
+[Collection("raven")]
+public class control_transport_compliance : TransportCompliance<RavenDbControlTransportFixture>;

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

@@ -80,7 +80,12 @@ public void Initialize(IWolverineRuntime runtime)
             && runtime.Options.Transports.NodeControlEndpoint == null
             && runtime.Options.Durability.Mode == DurabilityMode.Balanced)
         {
-            var transport = new Transport.RavenDbControlTransport(_store, runtime.Options);
+            // The transport itself is registered eagerly in UseRavenDbPersistence so
+            // its "ravendb://" scheme resolves for publishing rules. Here we promote
+            // its control endpoint to the NodeControlEndpoint, which marks it as a
+            // live listener — only under Balanced, so Solo hosts never poll.
+            var transport = runtime.Options.Transports.OfType<Transport.RavenDbControlTransport>().FirstOrDefault()
+                            ?? new Transport.RavenDbControlTransport(_store, runtime.Options);
             runtime.Options.Transports.Add(transport);
             runtime.Options.Transports.NodeControlEndpoint = transport.ControlEndpoint;
         }

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

@@ -12,12 +12,21 @@ namespace Wolverine.RavenDb.Internals.Transport;
 
 internal class RavenDbControlTransport : ITransport, IAsyncDisposable
 {
-    public const string ProtocolName = "ravencontrol";
+    public const string ProtocolName = "ravendb";
 
     private readonly Cache<Guid, RavenDbControlEndpoint> _endpoints;
     private readonly WolverineOptions _options;
     private RetryBlock<List<Envelope>>? _deleteBlock;
 
+    // Registered eagerly at configuration time (UseRavenDbPersistence) so the
+    // "ravendb://" scheme resolves for publishing rules. The IDocumentStore isn't
+    // resolvable until the container is built, so it's supplied lazily in
+    // InitializeAsync. The ControlEndpoint is only promoted to a live listener when
+    // the message store wires it as the NodeControlEndpoint (Balanced mode).
+    public RavenDbControlTransport(WolverineOptions options) : this(null!, options)
+    {
+    }
+
     public RavenDbControlTransport(IDocumentStore store, WolverineOptions options)
     {
         Store = store;
@@ -29,7 +38,6 @@ public RavenDbControlTransport(IDocumentStore store, WolverineOptions options)
         });
 
         ControlEndpoint = _endpoints[_options.UniqueNodeId];
-        ControlEndpoint.IsListener = true;
     }
 
     public bool TryBuildBrokerUsage(out BrokerDescription description)
@@ -40,7 +48,7 @@ public bool TryBuildBrokerUsage(out BrokerDescription description)
 
     public RavenDbControlEndpoint ControlEndpoint { get; }
 
-    public IDocumentStore Store { get; }
+    public IDocumentStore Store { get; private set; }
 
     public WolverineOptions Options => _options;
 
@@ -87,6 +95,10 @@ public IEnumerable<Endpoint> Endpoints()
 
     public ValueTask InitializeAsync(IWolverineRuntime runtime)
     {
+        // The store isn't available when the transport is registered at config time;
+        // resolve it now from the built container.
+        Store ??= (IDocumentStore)runtime.Services.GetService(typeof(IDocumentStore))!;
+
         foreach (var endpoint in Endpoints()) endpoint.Compile(runtime);
 
         _deleteBlock = new RetryBlock<List<Envelope>>(deleteEnvelopesAsync,

src/Persistence/Wolverine.RavenDb/WolverineRavenDbExtensions.cs

@@ -20,6 +20,18 @@ public static class WolverineRavenDbExtensions
     public static WolverineOptions UseRavenDbPersistence(this WolverineOptions options)
     {
         options.Services.AddSingleton<IMessageStore, RavenDbMessageStore>();
+
+        // Register the native RavenDB control-queue transport eagerly so the
+        // "ravendb://" scheme resolves for publishing rules configured at bootstrap.
+        // The endpoint only becomes a live listener when the message store promotes
+        // it to the NodeControlEndpoint under Balanced durability (see
+        // RavenDbMessageStore.Initialize). The store is resolved later, in the
+        // transport's InitializeAsync.
+        if (!options.Transports.OfType<Internals.Transport.RavenDbControlTransport>().Any())
+        {
+            options.Transports.Add(new Internals.Transport.RavenDbControlTransport(options));
+        }
+
         options.CodeGeneration.InsertFirstPersistenceStrategy<RavenDbPersistenceFrameProvider>();
         options.CodeGeneration.Sources.Add(new AsyncDocumentSessionSource());
         options.Services.AddHostedService<DeadLetterQueueReplayer>();

src/Testing/Wolverine.ComplianceTests/Compliance/TransportCompliance.cs

@@ -34,9 +34,18 @@ protected TransportComplianceFixture(Uri destination, int defaultTimeInSeconds =
     public bool AllLocally { get; set; }
 
     public bool MustReset { get; set; } = true;
-    
+
     public bool IsSenderOnlyTransport { get; set; }
 
+    /// <summary>
+    /// Durability mode applied to the sender and receiver hosts. Defaults to Solo,
+    /// which is correct for the vast majority of broker transports. Control-plane
+    /// transports that only register in a clustered mode (e.g. the RavenDB control
+    /// queue, which wires its NodeControlEndpoint only under Balanced) can override
+    /// this to DurabilityMode.Balanced.
+    /// </summary>
+    public DurabilityMode Mode { get; set; } = DurabilityMode.Solo;
+
     public async ValueTask DisposeAsync()
     {
         if (Sender == null)
@@ -86,7 +95,7 @@ protected async Task TheOnlyAppIs(Action<WolverineOptions> configure)
             .UseWolverine(options =>
             {
                 configure(options);
-                options.Durability.Mode = DurabilityMode.Solo;
+                options.Durability.Mode = Mode;
                 configureReceiver(options);
                 configureSender(options);
             }).StartAsync();
@@ -99,7 +108,7 @@ protected async Task SenderIs(Action<WolverineOptions> configure)
             {
                 configure(opts);
                 configureSender(opts);
-                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.Durability.Mode = Mode;
             }).StartAsync();
 
     }
@@ -117,7 +126,7 @@ private void configureSender(WolverineOptions options)
         options.Services.AddSingleton<IMessageSerializer, GreenTextWriter>();
         //options.Services.AddResourceSetupOnStartup(StartupAction.ResetState);
 
-        options.Durability.Mode = DurabilityMode.Solo;
+        options.Durability.Mode = Mode;
 
         options.UseNewtonsoftForSerialization();
     }
@@ -127,7 +136,7 @@ public async Task ReceiverIs(Action<WolverineOptions> configure)
         Receiver = await Host.CreateDefaultBuilder()
             .UseWolverine(opts =>
             {
-                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.Durability.Mode = Mode;
                 configure(opts);
                 configureReceiver(opts);
             }).StartAsync();
@@ -179,11 +188,11 @@ public virtual void BeforeEach()
     private readonly bool _ownFixture;
 
     protected TransportCompliance()
-    {
+    {
         Fixture = new T();
         _ownFixture = true;
-    }
-
+    }
+
     protected TransportCompliance(T fixture)
     {
         Fixture = fixture;
@@ -203,13 +212,13 @@ public async Task InitializeAsync()
         theReceiver = Fixture.Receiver;
         theOutboundAddress = Fixture.OutboundAddress;
 
-        if (Fixture.MustReset)
-        {
-            await Fixture.Sender.ResetResourceState();
-
-            if (Fixture.Receiver != null && !ReferenceEquals(Fixture.Sender, Fixture.Receiver))
-            {
-                await Fixture.Receiver.ResetResourceState();
+        if (Fixture.MustReset)
+        {
+            await Fixture.Sender.ResetResourceState();
+
+            if (Fixture.Receiver != null && !ReferenceEquals(Fixture.Sender, Fixture.Receiver))
+            {
+                await Fixture.Receiver.ResetResourceState();
             }
         }
 
@@ -218,12 +227,12 @@ public async Task InitializeAsync()
 
     public async Task DisposeAsync()
     {
-        if (!_ownFixture)
-            return;
-        if (Fixture is IAsyncDisposable)
-            await Fixture.DisposeAsync();
-        else
-            Fixture?.SafeDispose();
+        if (!_ownFixture)
+            return;
+        if (Fixture is IAsyncDisposable)
+            await Fixture.DisposeAsync();
+        else
+            Fixture?.SafeDispose();
     }
 
     [Fact]