GH-3709: ProcessInParallelWithNativeAcks() on the global partitioned topology

src/Testing/CoreTests/Runtime/Partitioning/native_ack_global_partitioning_3709.cs

@@ -0,0 +1,215 @@
+using NSubstitute;
+using Shouldly;
+using Wolverine;
+using Wolverine.Configuration;
+using Wolverine.Runtime.Partitioning;
+using Wolverine.Runtime;
+using Wolverine.Runtime.Routing;
+using Wolverine.Transports.Sending;
+using Wolverine.Transports.Stub;
+using Xunit;
+
+namespace CoreTests.Runtime.Partitioning;
+
+/// <summary>
+/// GH-3709. <c>ProcessInParallelWithNativeAcks()</c> is what makes <see cref="EndpointMode.NativeAck"/>
+/// reachable from a global partitioned topology: the default topology bridges each slot into a companion
+/// local queue and executes there, which is precisely why a local queue -- with no broker delivery to
+/// settle -- cannot host the mode. Opting in removes the companion topology and the bridge so each slot
+/// listener settles its own deliveries.
+/// </summary>
+public class native_ack_global_partitioning_3709
+{
+    private readonly WolverineOptions _options = new();
+
+    // buildEndpoint() runs from the PartitionedMessageTopology constructor, ahead of any derived
+    // constructor body, so this cannot be an instance field.
+    private static readonly StubTransport _transport = new();
+
+    private class NativeAckCapableEndpoint(string queueName, StubTransport transport)
+        : StubEndpoint(queueName, transport)
+    {
+        protected override bool supportsNativeAck => true;
+    }
+
+    /// <summary>Stands in for a sharded Rabbit topology until RabbitMqQueue opts in (GH-3708).</summary>
+    private class NativeAckCapableTopology(WolverineOptions options, string baseName, int count)
+        : PartitionedMessageTopology(options, PartitionSlots.Five, baseName, count)
+    {
+        protected override Endpoint buildEndpoint(WolverineOptions options, string name)
+        {
+            return new NativeAckCapableEndpoint(name, _transport);
+        }
+    }
+
+    private GlobalPartitionedMessageTopology topologyWithNativeAcks(string baseName, int slots,
+        out NativeAckCapableTopology external)
+    {
+        var topology = new GlobalPartitionedMessageTopology(_options);
+        topology.ProcessInParallelWithNativeAcks();
+
+        external = new NativeAckCapableTopology(_options, baseName, slots);
+        topology.SetExternalTopology(external, baseName);
+
+        return topology;
+    }
+
+    [Fact]
+    public void sets_native_ack_on_every_external_slot()
+    {
+        topologyWithNativeAcks("na-mode", 3, out var external);
+
+        foreach (var slot in external.Slots)
+        {
+            slot.Mode.ShouldBe(EndpointMode.NativeAck);
+        }
+    }
+
+    [Fact]
+    public void creates_no_companion_local_topology()
+    {
+        // The default path builds one automatically in SetExternalTopology -- see
+        // set_external_topology_creates_companion_local_topology_with_matching_slot_count.
+        var topology = topologyWithNativeAcks("na-nolocal", 3, out _);
+
+        topology.LocalTopology.ShouldBeNull();
+        topology.UsesNativeAcks.ShouldBeTrue();
+    }
+
+    [Fact]
+    public void leaves_the_bridge_unwired()
+    {
+        // ListeningAgent wires GlobalPartitionedReceiverBridge off exactly this property, so leaving it
+        // null is what keeps the bridge out of the picture. There is no separate opt-out.
+        topologyWithNativeAcks("na-bridge", 3, out var external);
+
+        foreach (var slot in external.Slots)
+        {
+            slot.GlobalPartitionLocalQueueUri.ShouldBeNull();
+        }
+    }
+
+    [Fact]
+    public void slots_keep_their_exclusive_scope_and_group_sharding()
+    {
+        // The cluster-wide half of the guarantee: one consumer per slot, sharded into sequential lanes
+        // by group id inside it. Neither is affected by dropping the bridge.
+        topologyWithNativeAcks("na-scope", 3, out var external);
+
+        foreach (var slot in external.Slots)
+        {
+            slot.ListenerScope.ShouldBe(ListenerScope.Exclusive);
+            slot.GroupShardingSlotNumber.ShouldBe(PartitionSlots.Five);
+        }
+    }
+
+    [Fact]
+    public void assert_validity_does_not_demand_a_local_topology()
+    {
+        var topology = topologyWithNativeAcks("na-valid", 3, out _);
+        topology.Message<GlobalTestMessage>();
+
+        Should.NotThrow(() => topology.AssertValidity());
+    }
+
+    [Fact]
+    public void assert_validity_still_demands_a_subscription_and_an_external_topology()
+    {
+        // Relaxing the local-topology rules must not relax the others.
+        var noSubscription = topologyWithNativeAcks("na-nosub", 3, out _);
+        Should.Throw<InvalidOperationException>(() => noSubscription.AssertValidity())
+            .Message.ShouldContain("message type matching policy");
+
+        var noExternal = new GlobalPartitionedMessageTopology(_options);
+        noExternal.ProcessInParallelWithNativeAcks();
+        noExternal.Message<GlobalTestMessage>();
+        Should.Throw<InvalidOperationException>(() => noExternal.AssertValidity())
+            .Message.ShouldContain("external transport topology");
+    }
+
+    [Fact]
+    public void mode_native_ack_still_throws_and_names_the_supported_call()
+    {
+        // Mode(NativeAck) would set the mode WITHOUT removing the bridge, so it stays rejected -- the
+        // guard from GH-3708 is replaced, not deleted.
+        var topology = new GlobalPartitionedMessageTopology(_options);
+
+        var nativeAck = Should.Throw<ArgumentOutOfRangeException>(() => topology.Mode(EndpointMode.NativeAck));
+        var inline = Should.Throw<ArgumentOutOfRangeException>(() => topology.Mode(EndpointMode.Inline));
+
+        nativeAck.ParamName.ShouldBe("mode");
+        nativeAck.Message.ShouldContain(nameof(EndpointMode.NativeAck));
+        nativeAck.Message.ShouldContain(nameof(GlobalPartitionedMessageTopology.ProcessInParallelWithNativeAcks));
+
+        // Deliberately NOT asserting on shared prose such as "companion local queue" -- that phrase is in
+        // the Inline rejection too, so an assertion on it would still pass if NativeAck fell through to the
+        // Inline guard and the mode-specific message were lost. Comparing the two is the check that bites.
+        // (Shouldly's ShouldContain is case-insensitive, so ShouldNotContain("Inline") would trip on any
+        // lowercase "inline" in the prose -- inequality is the robust form.)
+        nativeAck.Message.ShouldNotBe(inline.Message);
+    }
+
+    [Fact]
+    public void local_queues_and_native_acks_together_is_a_configuration_error()
+    {
+        var topology = new GlobalPartitionedMessageTopology(_options);
+        topology.ProcessInParallelWithNativeAcks();
+
+        Should.Throw<InvalidOperationException>(() => topology.LocalQueues("na-explicit", 3))
+            .Message.ShouldContain("no companion local queues");
+    }
+
+    [Fact]
+    public void native_acks_after_local_queues_drops_the_local_topology()
+    {
+        // Order independence, matching how Mode() behaves: the last word wins rather than the config
+        // silently keeping queues that nothing will ever route to.
+        var topology = new GlobalPartitionedMessageTopology(_options);
+        topology.LocalQueues("na-first", 3);
+        topology.LocalTopology.ShouldNotBeNull();
+
+        topology.ProcessInParallelWithNativeAcks();
+
+        topology.LocalTopology.ShouldBeNull();
+    }
+
+    [Fact]
+    public void a_transport_that_has_not_opted_in_fails_fast()
+    {
+        // LocalPartitionedMessageTopology's slots are local queues, which never accept NativeAck.
+        var topology = new GlobalPartitionedMessageTopology(_options);
+        topology.ProcessInParallelWithNativeAcks();
+
+        var external = new LocalPartitionedMessageTopology(_options, "na-unsupported", 3);
+
+        var ex = Should.Throw<InvalidOperationException>(() => topology.SetExternalTopology(external, "na-unsupported"));
+        ex.Message.ShouldContain("does not support EndpointMode.NativeAck");
+    }
+
+    [Fact]
+    public void the_local_shortcut_is_disabled_so_sends_always_go_through_the_broker()
+    {
+        // The shortcut hands the message straight to the companion local queue when this node already
+        // owns the slot. In native-ack mode the broker delivery IS the durability story, so bypassing it
+        // would drop the message on a crash between send and handling.
+        //
+        // Passing a null runtime is the assertion: the non-native path dereferences it to look up the
+        // listening agent, so this only survives if the shortcut is skipped outright.
+        var externalSlots = new[] { Substitute.For<IMessageRoute>(), Substitute.For<IMessageRoute>() };
+        var localSlots = Array.Empty<IMessageRoute>();
+        var expected = new Envelope(new GlobalTestMessage("a"));
+
+        foreach (var slot in externalSlots)
+        {
+            slot.CreateForSending(Arg.Any<object>(), Arg.Any<DeliveryOptions?>(), Arg.Any<ISendingAgent>(),
+                Arg.Any<WolverineRuntime>(), Arg.Any<string?>()).Returns(expected);
+        }
+
+        var route = new GlobalPartitionedRoute(new Uri("shard://stub/na"), _options.MessagePartitioning,
+            externalSlots, localSlots, [], nativeAcks: true);
+
+        var envelope = route.CreateForSending(new GlobalTestMessage("a"), null, null!, null!, null);
+
+        envelope.ShouldBeSameAs(expected);
+    }
+}

src/Wolverine/Runtime/Partitioning/GlobalPartitionedMessageTopology.cs

@@ -14,6 +14,7 @@ public class GlobalPartitionedMessageTopology
     private PartitionedMessageTopology? _externalTopology;
     private LocalPartitionedMessageTopology? _localTopology;
     private EndpointMode _mode = EndpointMode.Durable;
+    private bool _nativeAcks;
 
     public GlobalPartitionedMessageTopology(WolverineOptions options)
     {
@@ -23,6 +24,12 @@ public GlobalPartitionedMessageTopology(WolverineOptions options)
     internal PartitionedMessageTopology? ExternalTopology => _externalTopology;
     internal LocalPartitionedMessageTopology? LocalTopology => _localTopology;
 
+    /// <summary>
+    /// GH-3709. True when <see cref="ProcessInParallelWithNativeAcks"/> has been called: the slots settle
+    /// their own broker deliveries and there is no companion local topology or bridge at all.
+    /// </summary>
+    internal bool UsesNativeAcks => _nativeAcks;
+
     /// <summary>
     /// Opt the partitioned slots — the external endpoints AND their companion local queues — out of
     /// the default <see cref="EndpointMode.Durable"/>. Use <see cref="EndpointMode.BufferedInMemory"/>
@@ -44,9 +51,11 @@ public GlobalPartitionedMessageTopology Mode(EndpointMode mode)
             // native ack would have nothing to ack against. Endpoint-level PartitionProcessingByGroupId() on a
             // NativeAck listener is the supported shape for partitioned native-ack processing.
             throw new ArgumentOutOfRangeException(nameof(mode),
-                $"{nameof(EndpointMode)}.{nameof(EndpointMode.NativeAck)} is not supported for global partitioned topologies. "
-                + "Partitioned slots bridge into a companion local queue, which has no broker delivery to settle natively. "
-                + "Use PartitionProcessingByGroupId() directly on the native-ack listener instead.");
+                $"{nameof(EndpointMode)}.{nameof(EndpointMode.NativeAck)} cannot be set through {nameof(Mode)}() on a global partitioned topology. "
+                + "The default topology bridges each slot into a companion local queue, which has no broker delivery to settle natively. "
+                + $"Call {nameof(ProcessInParallelWithNativeAcks)}() instead -- it removes the companion local topology and the bridge so the "
+                + "slot listeners settle their own deliveries. PartitionProcessingByGroupId() directly on a native-ack listener is the "
+                + "endpoint-level equivalent.");
         }
 
         if (mode == EndpointMode.Inline)
@@ -60,6 +69,42 @@ public GlobalPartitionedMessageTopology Mode(EndpointMode mode)
         return this;
     }
 
+    /// <summary>
+    /// Process this topology's slots in parallel with native broker acknowledgements instead of the
+    /// durable inbox: each slot listener settles its own deliveries when the handler completes, and
+    /// shards into sequential lanes by group id in memory. Partitioned clustering with no database at
+    /// all — one exclusive consumer per slot across the cluster, no two messages of a group running
+    /// concurrently, at-least-once delivery owned by the broker rather than by the inbox.
+    /// </summary>
+    /// <remarks>
+    /// <para>GH-3709. This is what makes <see cref="EndpointMode.NativeAck"/> reachable from a global
+    /// partitioned topology. The default topology bridges each external listener into a companion local
+    /// queue and does the partitioned execution there (<c>GlobalPartitionedReceiverBridge</c>), which is
+    /// exactly why <see cref="Mode"/> refuses NativeAck — a local queue has no broker delivery to settle.
+    /// Calling this removes the companion topology and the bridge, so the slot's own receiver shards
+    /// directly and the ack stays tied to the delivery's own channel.</para>
+    ///
+    /// <para>Trade-offs against the Durable default: no inbox insert or mark-handled per message and no
+    /// database on the path, but also no inbox dedup, no outbox atomicity with handler side effects, and
+    /// recovery is the broker's redelivery rather than inbox recovery. Ordering is per-slot best effort;
+    /// two groups hashing to the same slot serialize against each other.</para>
+    ///
+    /// <para>The transport must opt in to <see cref="EndpointMode.NativeAck"/>. If it has not, applying
+    /// the mode throws at bootstrap naming the endpoint type — see <c>Endpoint.supportsNativeAck</c>.</para>
+    /// </remarks>
+    public GlobalPartitionedMessageTopology ProcessInParallelWithNativeAcks()
+    {
+        _nativeAcks = true;
+        _mode = EndpointMode.NativeAck;
+
+        // A companion local topology may already exist if LocalQueues() ran first. It is meaningless
+        // here -- drop it rather than leaving queues nothing routes to.
+        _localTopology = null;
+
+        applyMode();
+        return this;
+    }
+
     private void applyMode()
     {
         if (_externalTopology != null)
@@ -81,6 +126,14 @@ private void applyMode()
 
     public void LocalQueues(string baseQueueName, int numberOfEndpoints)
     {
+        if (_nativeAcks)
+        {
+            throw new InvalidOperationException(
+                $"A native-ack global partitioned topology has no companion local queues -- {nameof(ProcessInParallelWithNativeAcks)}() "
+                + $"makes each slot settle its own broker deliveries, so there is nothing for {nameof(LocalQueues)}() to configure. "
+                + $"Remove one of the two calls.");
+        }
+
         _localTopology = new LocalPartitionedMessageTopology(_options, baseQueueName, numberOfEndpoints);
         applyMode();
     }
@@ -94,7 +147,9 @@ internal void SetExternalTopology(PartitionedMessageTopology topology, string ba
     {
         _externalTopology = topology;
 
-        if (_localTopology == null)
+        // GH-3709. A native-ack topology has no companion local topology and no bridge: the slot's own
+        // receiver shards by group id so the ack stays tied to the delivery's own channel.
+        if (_localTopology == null && !_nativeAcks)
         {
             // Create companion local topology with matching slot count
             var localBaseName = $"global-{baseName}";
@@ -108,9 +163,11 @@ internal void SetExternalTopology(PartitionedMessageTopology topology, string ba
         // overwritten here.
         applyMode();
 
-        // Tag each external slot endpoint with its companion local queue URI
+        // Tag each external slot endpoint with its companion local queue URI. ListeningAgent wires the
+        // GlobalPartitionedReceiverBridge off exactly this property, so leaving it null on a native-ack
+        // topology is what keeps the bridge out of the picture -- there is no separate opt-out there.
         // Only tag if slot counts match; mismatches will be caught by AssertValidity()
-        if (topology.Slots.Count == _localTopology.Slots.Count)
+        if (_localTopology != null && topology.Slots.Count == _localTopology.Slots.Count)
         {
             for (var i = 0; i < topology.Slots.Count; i++)
             {
@@ -211,6 +268,14 @@ public void AssertValidity()
                 "An external transport topology must be configured for global partitioning");
         }
 
+        // GH-3709. A native-ack topology deliberately has no local topology, so the two rules below --
+        // both of which exist to keep the companion queues lined up with the bridge -- do not apply.
+        // The subscription and external-topology rules above still do.
+        if (_nativeAcks)
+        {
+            return;
+        }
+
         if (_localTopology == null)
         {
             throw new InvalidOperationException(
@@ -305,7 +370,14 @@ internal bool TryMatch(Type messageType, IWolverineRuntime runtime, out IMessage
             return false;
         }
 
-        if (_externalTopology == null || _localTopology == null)
+        if (_externalTopology == null)
+        {
+            return false;
+        }
+
+        // A native-ack topology has no local slots at all; every other topology needs them for the
+        // local shortcut and is not routable until they exist.
+        if (_localTopology == null && !_nativeAcks)
         {
             return false;
         }
@@ -314,9 +386,9 @@ internal bool TryMatch(Type messageType, IWolverineRuntime runtime, out IMessage
             .Select(x => (IMessageRoute)MessageRoute.For(messageType, x, runtime))
             .ToArray();
 
-        var localRoutes = _localTopology.Slots
+        var localRoutes = _localTopology?.Slots
             .Select(x => (IMessageRoute)MessageRoute.For(messageType, x, runtime))
-            .ToArray();
+            .ToArray() ?? [];
 
         var externalEndpoints = _externalTopology.Slots.ToArray();
 
@@ -325,7 +397,8 @@ internal bool TryMatch(Type messageType, IWolverineRuntime runtime, out IMessage
             runtime.Options.MessagePartitioning,
             externalRoutes,
             localRoutes,
-            externalEndpoints);
+            externalEndpoints,
+            _nativeAcks);
 
         return true;
     }