GlobalPartitionedMessageTopology.Except<T>() (#3867 follow-up)

src/Testing/CoreTests/Acceptance/global_partitioned_exclusions.cs

@@ -0,0 +1,104 @@
+using Wolverine;
+using Wolverine.Runtime.Partitioning;
+using Wolverine.Util;
+using Xunit;
+
+namespace CoreTests.Acceptance;
+
+public class GlobalPartitionedExclusionTests
+{
+    public interface IThing;
+
+    public record ThingHappened(string Id) : IThing;
+
+    public record ThingBroadcast(string Id) : IThing;
+
+    public record Unrelated(string Id);
+
+    private static GlobalPartitionedMessageTopology topology()
+    {
+        return new GlobalPartitionedMessageTopology(new WolverineOptions());
+    }
+
+    [Fact]
+    public void excluded_type_does_not_match_a_broader_rule()
+    {
+        var t = topology();
+        t.MessagesImplementing<IThing>();
+        t.Except<ThingBroadcast>();
+
+        // The whole point: a broad MessagesImplementing rule stays in place so nothing silently
+        // drops out of the topology, and only the named type is carved out.
+        t.Matches(typeof(ThingHappened)).ShouldBeTrue();
+        t.Matches(typeof(ThingBroadcast)).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void exclusions_win_regardless_of_declaration_order()
+    {
+        var before = topology();
+        before.Except<ThingBroadcast>();
+        before.MessagesImplementing<IThing>();
+        before.Matches(typeof(ThingBroadcast)).ShouldBeFalse();
+
+        var after = topology();
+        after.MessagesImplementing<IThing>();
+        after.Except<ThingBroadcast>();
+        after.Matches(typeof(ThingBroadcast)).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void an_exclusion_can_be_an_interface_and_carves_out_the_whole_family()
+    {
+        var t = topology();
+        t.MessagesImplementing<IThing>();
+        t.Except<IThing>();
+
+        t.Matches(typeof(ThingHappened)).ShouldBeFalse();
+        t.Matches(typeof(ThingBroadcast)).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void an_explicitly_named_type_can_still_be_excluded()
+    {
+        var t = topology();
+        t.Message<ThingBroadcast>();
+        t.Except<ThingBroadcast>();
+
+        t.Matches(typeof(ThingBroadcast)).ShouldBeFalse();
+
+        // MatchesByMessageTypeName is the pre-deserialization path (Kafka), and it reads a separate
+        // name cache — an exclusion that only updated Matches() would leak through it.
+        t.MatchesByMessageTypeName(typeof(ThingBroadcast).ToMessageTypeName()).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void excluding_before_naming_also_keeps_the_name_cache_clean()
+    {
+        var t = topology();
+        t.Except<ThingBroadcast>();
+        t.Message<ThingBroadcast>();
+
+        t.Matches(typeof(ThingBroadcast)).ShouldBeFalse();
+        t.MatchesByMessageTypeName(typeof(ThingBroadcast).ToMessageTypeName()).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void exclusions_do_not_affect_unrelated_types()
+    {
+        var t = topology();
+        t.MessagesImplementing<IThing>();
+        t.Except<ThingBroadcast>();
+
+        t.Matches(typeof(Unrelated)).ShouldBeFalse("never matched in the first place");
+
+        t.Message<Unrelated>();
+        t.Matches(typeof(Unrelated)).ShouldBeTrue();
+    }
+
+    [Fact]
+    public void null_type_is_rejected()
+    {
+        Should.Throw<ArgumentNullException>(() => topology().Except(null!));
+    }
+}

src/Wolverine/Runtime/Partitioning/GlobalPartitionedMessageTopology.cs

@@ -9,6 +9,7 @@ public class GlobalPartitionedMessageTopology
 {
     private readonly WolverineOptions _options;
     private readonly List<Subscription> _subscriptions = new();
+    private readonly List<Type> _exclusions = new();
     private readonly HashSet<string> _messageTypeNames = new(StringComparer.OrdinalIgnoreCase);
     private PartitionedMessageTopology? _externalTopology;
     private LocalPartitionedMessageTopology? _localTopology;
@@ -81,7 +82,14 @@ public void Message<T>()
     public void Message(Type type)
     {
         _subscriptions.Add(Subscription.ForType(type));
-        _messageTypeNames.Add(type.ToMessageTypeName());
+
+        // Exclusions win regardless of declaration order, so don't seed the name cache that
+        // MatchesByMessageTypeName reads (the pre-deserialization path) for an excluded type.
+        // Except() performs the mirror-image removal for the other ordering.
+        if (!_exclusions.Any(x => x.IsAssignableFrom(type)))
+        {
+            _messageTypeNames.Add(type.ToMessageTypeName());
+        }
     }
 
     /// <summary>
@@ -163,8 +171,50 @@ public void AssertValidity()
         }
     }
 
+    /// <summary>
+    /// Exclude a message type — or everything assignable to <typeparamref name="T"/>, so an
+    /// interface or base class excludes its whole family — from this topology, even when a
+    /// broader rule such as <see cref="MessagesImplementing{T}"/> would otherwise match it.
+    /// Exclusions always win.
+    /// </summary>
+    /// <remarks>
+    /// <para>The case this exists for: a message type that legitimately belongs to the topology on
+    /// the way IN, but that the receiving application also re-publishes on its way somewhere else.
+    /// Because both sides configure their own topology, excluding it on the <em>receiving</em> side
+    /// keeps inbound partitioning intact while stopping that application's own re-publish from
+    /// re-entering the topology and coming straight back to the handler that published it — an
+    /// infinite loop that is invisible in configuration and shows up only as amplified load.</para>
+    ///
+    /// <para>Excluding a type does not stop this application <em>listening</em> for it on the
+    /// topology's slots: the companion-queue bridge is wired per endpoint, not per message type. It
+    /// only removes the type from this topology's publishing rules.</para>
+    /// </remarks>
+    public void Except<T>()
+    {
+        Except(typeof(T));
+    }
+
+    /// <summary>
+    /// Exclude a message type — or everything assignable to <paramref name="type"/> — from this
+    /// topology. See <see cref="Except{T}"/>.
+    /// </summary>
+    public void Except(Type type)
+    {
+        if (type == null) throw new ArgumentNullException(nameof(type));
+
+        _exclusions.Add(type);
+        _messageTypeNames.Remove(type.ToMessageTypeName());
+    }
+
     internal bool Matches(Type messageType)
     {
+        // Exclusions are checked first and win outright, so an Except<T>() cannot be defeated by
+        // the order in which rules were declared.
+        if (_exclusions.Any(x => x.IsAssignableFrom(messageType)))
+        {
+            return false;
+        }
+
         return _subscriptions.Any(x => x.Matches(messageType));
     }