diff --git a/src/Testing/CoreTests/Acceptance/global_partitioned_exclusions.cs b/src/Testing/CoreTests/Acceptance/global_partitioned_exclusions.cs new file mode 100644 index 000000000..1a321947f --- /dev/null +++ b/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(); + t.Except(); + + // 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(); + before.MessagesImplementing(); + before.Matches(typeof(ThingBroadcast)).ShouldBeFalse(); + + var after = topology(); + after.MessagesImplementing(); + after.Except(); + 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(); + t.Except(); + + 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(); + t.Except(); + + 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(); + t.Message(); + + 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(); + t.Except(); + + t.Matches(typeof(Unrelated)).ShouldBeFalse("never matched in the first place"); + + t.Message(); + t.Matches(typeof(Unrelated)).ShouldBeTrue(); + } + + [Fact] + public void null_type_is_rejected() + { + Should.Throw(() => topology().Except(null!)); + } +} diff --git a/src/Wolverine/Runtime/Partitioning/GlobalPartitionedMessageTopology.cs b/src/Wolverine/Runtime/Partitioning/GlobalPartitionedMessageTopology.cs index 7909fa80a..7c70ce520 100644 --- a/src/Wolverine/Runtime/Partitioning/GlobalPartitionedMessageTopology.cs +++ b/src/Wolverine/Runtime/Partitioning/GlobalPartitionedMessageTopology.cs @@ -9,6 +9,7 @@ public class GlobalPartitionedMessageTopology { private readonly WolverineOptions _options; private readonly List _subscriptions = new(); + private readonly List _exclusions = new(); private readonly HashSet _messageTypeNames = new(StringComparer.OrdinalIgnoreCase); private PartitionedMessageTopology? _externalTopology; private LocalPartitionedMessageTopology? _localTopology; @@ -81,7 +82,14 @@ public void Message() 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()); + } } /// @@ -163,8 +171,50 @@ public void AssertValidity() } } + /// + /// Exclude a message type — or everything assignable to , so an + /// interface or base class excludes its whole family — from this topology, even when a + /// broader rule such as would otherwise match it. + /// Exclusions always win. + /// + /// + /// 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 receiving 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. + /// + /// Excluding a type does not stop this application listening 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. + /// + public void Except() + { + Except(typeof(T)); + } + + /// + /// Exclude a message type — or everything assignable to — from this + /// topology. See . + /// + 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() 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)); }