diff --git a/src/Persistence/LeaderElection/RavenDbTests.LeaderElection/control_queue_leadership_election_compliance.cs b/src/Persistence/LeaderElection/RavenDbTests.LeaderElection/control_queue_leadership_election_compliance.cs new file mode 100644 index 000000000..cef7c1526 --- /dev/null +++ b/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; + +/// +/// Runs the full leadership-election compliance battery while relying on the +/// native RavenDB control queue (ravendb://) for inter-node control +/// messaging, rather than UseTcpForControlEndpoint(). This exercises the +/// RavenDbControlTransport added in #3285 under real multi-node fan-out, +/// send-to-node, and leader-failover scenarios. +/// +[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(); + } +} diff --git a/src/Persistence/RavenDbTests/control_queue_tests.cs b/src/Persistence/RavenDbTests/control_queue_tests.cs index def5c30f2..a70b11f0b 100644 --- a/src/Persistence/RavenDbTests/control_queue_tests.cs +++ b/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] diff --git a/src/Persistence/RavenDbTests/control_transport_compliance.cs b/src/Persistence/RavenDbTests/control_transport_compliance.cs new file mode 100644 index 000000000..690de61ba --- /dev/null +++ b/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().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; diff --git a/src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.cs b/src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.cs index 273e6168a..05bb1f995 100644 --- a/src/Persistence/Wolverine.RavenDb/Internals/RavenDbMessageStore.cs +++ b/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().FirstOrDefault() + ?? new Transport.RavenDbControlTransport(_store, runtime.Options); runtime.Options.Transports.Add(transport); runtime.Options.Transports.NodeControlEndpoint = transport.ControlEndpoint; } diff --git a/src/Persistence/Wolverine.RavenDb/Internals/Transport/RavenDbControlTransport.cs b/src/Persistence/Wolverine.RavenDb/Internals/Transport/RavenDbControlTransport.cs index 69e636ae8..42bb74e8a 100644 --- a/src/Persistence/Wolverine.RavenDb/Internals/Transport/RavenDbControlTransport.cs +++ b/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 _endpoints; private readonly WolverineOptions _options; private RetryBlock>? _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 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>(deleteEnvelopesAsync, diff --git a/src/Persistence/Wolverine.RavenDb/WolverineRavenDbExtensions.cs b/src/Persistence/Wolverine.RavenDb/WolverineRavenDbExtensions.cs index 91edb3e5a..0009709fa 100644 --- a/src/Persistence/Wolverine.RavenDb/WolverineRavenDbExtensions.cs +++ b/src/Persistence/Wolverine.RavenDb/WolverineRavenDbExtensions.cs @@ -20,6 +20,18 @@ public static class WolverineRavenDbExtensions public static WolverineOptions UseRavenDbPersistence(this WolverineOptions options) { options.Services.AddSingleton(); + + // 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().Any()) + { + options.Transports.Add(new Internals.Transport.RavenDbControlTransport(options)); + } + options.CodeGeneration.InsertFirstPersistenceStrategy(); options.CodeGeneration.Sources.Add(new AsyncDocumentSessionSource()); options.Services.AddHostedService(); diff --git a/src/Testing/Wolverine.ComplianceTests/Compliance/TransportCompliance.cs b/src/Testing/Wolverine.ComplianceTests/Compliance/TransportCompliance.cs index 6bd9bb4c3..96a371730 100644 --- a/src/Testing/Wolverine.ComplianceTests/Compliance/TransportCompliance.cs +++ b/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; } + /// + /// 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. + /// + public DurabilityMode Mode { get; set; } = DurabilityMode.Solo; + public async ValueTask DisposeAsync() { if (Sender == null) @@ -86,7 +95,7 @@ protected async Task TheOnlyAppIs(Action 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 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(); //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 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]