Add support for multiple filters in consumer grous (#40)

* Add support for multiple filters in static and elastic consumer group configurations

Refactored filtering logic, added utility methods, extended validations, updated related tests, and enhanced handling of wildcard-based partitioning.

* Add test to verify partitioning behavior with full-subject hash in elastic consumer groups.

* Add compatibility notes for .NET-only multi-filter and [-1] sentinel features in consumer groups

* Add server version guard in elastic consumer group tests for [-1] sentinel support.

* Update compatibility notes for .NET-only multi-filter and [-1] sentinel features in consumer group configurations

* Refactor elastic consumer group configuration to replace wildcards with `NatsPcgPartitioningFilter`, simplify validations, and update related tests.

* Update partitioning filter in paritiy

* fix build

* tests: fix elastic PCGroups CI failures

Use unique GUID-based subject prefixes in all elastic tests to prevent
"subjects overlap with an existing stream" errors when multiple TFMs
share the same NATS server in CI.

Add SkipBelow212Async to the empty PartitioningFilters test since
full-subject partitioning requires NATS 2.12+. Fix the work-queue
stream name assertion (was incorrectly prefixed with "pcg-") and add
proper cleanup in try/finally.

* tests: add Go interop tests for elastic PCGroups

Use GoHarness to verify cross-language interop with orbit.go
pcgroups v0.2.0. Three tests cover .NET-creates/Go-consumes,
Go-creates/.NET-consumes, and empty PartitioningFilters interop.

* tests: skip Go interop consume tests on NATS < 2.11

ElasticConsume uses priority groups which require NATS 2.11+.
The Go process exits with an error on 2.10, causing the test to
fail on ReadLineAsync.

Author: mtmk

src/Synadia.Orbit.PCGroups/Elastic/NatsPcgElasticConfig.cs

@@ -17,16 +17,10 @@ public sealed record NatsPcgElasticConfig
     public required uint MaxMembers { get; init; }
 
     /// <summary>
-    /// Gets the subject filter for the consumer group.
+    /// Gets the partitioning filters, each pairing a subject filter with its wildcard positions for partitioning.
     /// </summary>
-    [JsonPropertyName("filter")]
-    public required string Filter { get; init; }
-
-    /// <summary>
-    /// Gets the wildcard positions used for partitioning (1-indexed).
-    /// </summary>
-    [JsonPropertyName("partitioning_wildcards")]
-    public required int[] PartitioningWildcards { get; init; }
+    [JsonPropertyName("partitioning_filters")]
+    public required NatsPcgPartitioningFilter[] PartitioningFilters { get; init; }
 
     /// <summary>
     /// Gets the optional maximum number of buffered messages.

src/Synadia.Orbit.PCGroups/Elastic/NatsPcgElasticConsumeContext.cs

@@ -260,11 +260,7 @@ private async Task CreateOrGetConsumerAsync(CancellationToken cancellationToken)
             config = _config;
         }
 
-        string[] filters = NatsPcgPartitionDistributor.GeneratePartitionFilters(
-            config.Members,
-            config.MaxMembers,
-            config.MemberMappings,
-            _memberName);
+        string[] filters = GenerateFiltersForMember(config, _memberName);
 
         _currentFilters = filters;
 
@@ -313,11 +309,7 @@ private async Task RecreateConsumerAsync()
         }
 
         // Recalculate filters
-        string[] filters = NatsPcgPartitionDistributor.GeneratePartitionFilters(
-            config.Members,
-            config.MaxMembers,
-            config.MemberMappings,
-            _memberName);
+        string[] filters = GenerateFiltersForMember(config, _memberName);
 
         // Only recreate if filters changed
         if (FiltersEqual(filters, _currentFilters))
@@ -347,6 +339,34 @@ private async Task RecreateConsumerAsync()
         _consumer = await _js.CreateOrUpdateConsumerAsync(workQueueStreamName, consumerConfig, _cts.Token).ConfigureAwait(false);
     }
 
+    private static string[] GenerateFiltersForMember(NatsPcgElasticConfig config, string memberName)
+    {
+        var allFilters = new List<string>();
+        if (config.PartitioningFilters.Length > 0)
+        {
+            foreach (var pf in config.PartitioningFilters)
+            {
+                allFilters.AddRange(NatsPcgPartitionDistributor.GeneratePartitionFilters(
+                    config.Members,
+                    config.MaxMembers,
+                    config.MemberMappings,
+                    memberName,
+                    pf.Filter));
+            }
+        }
+        else
+        {
+            allFilters.AddRange(NatsPcgPartitionDistributor.GeneratePartitionFilters(
+                config.Members,
+                config.MaxMembers,
+                config.MemberMappings,
+                memberName,
+                ">"));
+        }
+
+        return allFilters.ToArray();
+    }
+
     private async Task WatchConfigLoopAsync()
     {
         try

src/Synadia.Orbit.PCGroups/Elastic/NatsPcgElasticExtensions.cs

@@ -25,8 +25,7 @@ public static class NatsPcgElasticExtensions
     /// <param name="streamName">Name of the source stream.</param>
     /// <param name="consumerGroupName">Name of the consumer group.</param>
     /// <param name="maxNumMembers">Maximum number of members (also the number of partitions).</param>
-    /// <param name="filter">Subject filter for the consumer group.</param>
-    /// <param name="partitioningWildcards">Wildcard positions for partitioning (1-indexed).</param>
+    /// <param name="partitioningFilters">Partitioning filters, each pairing a subject filter with wildcard positions.</param>
     /// <param name="maxBufferedMessages">Optional maximum number of buffered messages.</param>
     /// <param name="maxBufferedBytes">Optional maximum bytes of buffered messages.</param>
     /// <param name="cancellationToken">Cancellation token.</param>
@@ -36,19 +35,17 @@ public static async Task<NatsPcgElasticConfig> CreatePcgElasticAsync(
         string streamName,
         string consumerGroupName,
         uint maxNumMembers,
-        string filter,
-        int[] partitioningWildcards,
+        NatsPcgPartitioningFilter[] partitioningFilters,
         long? maxBufferedMessages = null,
         long? maxBufferedBytes = null,
         CancellationToken cancellationToken = default)
     {
-        ValidateConfig(maxNumMembers, filter, partitioningWildcards);
+        ValidateConfig(maxNumMembers, partitioningFilters);
 
         var config = new NatsPcgElasticConfig
         {
             MaxMembers = maxNumMembers,
-            Filter = filter,
-            PartitioningWildcards = partitioningWildcards,
+            PartitioningFilters = partitioningFilters,
             MaxBufferedMsgs = maxBufferedMessages,
             MaxBufferedBytes = maxBufferedBytes,
         };
@@ -461,11 +458,30 @@ public static async Task DeletePcgElasticMemberMappingsAsync(
     /// <returns>Array of partition filters.</returns>
     public static string[] GetPcgElasticPartitionFilters(this NatsPcgElasticConfig config, string memberName)
     {
-        return NatsPcgPartitionDistributor.GeneratePartitionFilters(
-            config.Members,
-            config.MaxMembers,
-            config.MemberMappings,
-            memberName);
+        var filters = new List<string>();
+        if (config.PartitioningFilters.Length > 0)
+        {
+            foreach (var pf in config.PartitioningFilters)
+            {
+                filters.AddRange(NatsPcgPartitionDistributor.GeneratePartitionFilters(
+                    config.Members,
+                    config.MaxMembers,
+                    config.MemberMappings,
+                    memberName,
+                    pf.Filter));
+            }
+        }
+        else
+        {
+            filters.AddRange(NatsPcgPartitionDistributor.GeneratePartitionFilters(
+                config.Members,
+                config.MaxMembers,
+                config.MemberMappings,
+                memberName,
+                ">"));
+        }
+
+        return filters.ToArray();
     }
 
     /// <summary>
@@ -491,15 +507,15 @@ public static async Task PcgElasticMemberStepDownAsync(
     }
 
     // ReSharper disable ParameterOnlyUsedForPreconditionCheck.Local
-    private static void ValidateConfig(uint maxNumMembers, string filter, int[] partitioningWildcards)
+    private static void ValidateConfig(uint maxNumMembers, NatsPcgPartitioningFilter[] partitioningFilters)
     {
         // ReSharper restore ParameterOnlyUsedForPreconditionCheck.Local
         if (maxNumMembers == 0)
         {
             throw new NatsPcgConfigurationException("maxNumMembers must be greater than 0");
         }
 
-        NatsPcgMemberMappingValidator.ValidateFilterAndWildcards(filter, partitioningWildcards);
+        NatsPcgMemberMappingValidator.ValidatePartitioningFilters(partitioningFilters);
     }
 
     private static async Task CreateWorkQueueStreamAsync(
@@ -511,37 +527,26 @@ private static async Task CreateWorkQueueStreamAsync(
     {
         string workQueueStreamName = GetWorkQueueStreamName(streamName, consumerGroupName);
 
-        // Build subject transform: add partition prefix based on wildcards
-        // Transform syntax: {{Partition(numPartitions, wildcardIndexes...)}}
-        // Replace * wildcards with {{wildcard(N)}} in the filter
-        string wildcardStr = string.Join(",", config.PartitioningWildcards);
-        var filterTokens = config.Filter.Split('.');
-        int wildcardIndex = 1;
-        for (int i = 0; i < filterTokens.Length; i++)
+        var subjectTransforms = new List<SubjectTransform>();
+        if (config.PartitioningFilters.Length > 0)
         {
-            if (filterTokens[i] == "*")
+            foreach (var pf in config.PartitioningFilters)
             {
-                filterTokens[i] = $"{{{{wildcard({wildcardIndex})}}}}";
-                wildcardIndex++;
+                subjectTransforms.Add(BuildSubjectTransform(pf, config.MaxMembers));
             }
         }
-
-        string destFromFilter = string.Join(".", filterTokens);
-        string subjectTransform = $"{{{{Partition({config.MaxMembers},{wildcardStr})}}}}.{destFromFilter}";
+        else
+        {
+            var defaultFilter = new NatsPcgPartitioningFilter(">", Array.Empty<int>());
+            subjectTransforms.Add(BuildSubjectTransform(defaultFilter, config.MaxMembers));
+        }
 
         var sources = new List<StreamSource>
         {
             new()
             {
                 Name = streamName,
-                SubjectTransforms = new List<SubjectTransform>
-                {
-                    new()
-                    {
-                        Src = config.Filter,
-                        Dest = subjectTransform,
-                    },
-                },
+                SubjectTransforms = subjectTransforms,
             },
         };
 
@@ -565,6 +570,46 @@ private static async Task CreateWorkQueueStreamAsync(
         await js.CreateStreamAsync(streamConfig, cancellationToken).ConfigureAwait(false);
     }
 
+    private static SubjectTransform BuildSubjectTransform(NatsPcgPartitioningFilter pf, uint maxMembers)
+    {
+        bool isFullSubject = NatsPcgMemberMappingValidator.IsPartitionByFullSubject(pf.PartitioningWildcards);
+
+        // Build subject transform: add partition prefix based on wildcards
+        // Replace * wildcards with {{wildcard(N)}} in the filter
+        var filterTokens = pf.Filter.Split('.');
+        int wildcardIndex = 1;
+        for (int i = 0; i < filterTokens.Length; i++)
+        {
+            if (filterTokens[i] == "*")
+            {
+                filterTokens[i] = $"{{{{wildcard({wildcardIndex})}}}}";
+                wildcardIndex++;
+            }
+        }
+
+        string destFromFilter = string.Join(".", filterTokens);
+
+        string partitionFunction;
+        if (isFullSubject)
+        {
+            // {{Partition(N)}} without wildcard positions hashes the entire subject
+            partitionFunction = $"{{{{Partition({maxMembers})}}}}";
+        }
+        else
+        {
+            string wildcardStr = string.Join(",", pf.PartitioningWildcards);
+            partitionFunction = $"{{{{Partition({maxMembers},{wildcardStr})}}}}";
+        }
+
+        string subjectTransform = $"{partitionFunction}.{destFromFilter}";
+
+        return new SubjectTransform
+        {
+            Src = pf.Filter,
+            Dest = subjectTransform,
+        };
+    }
+
     private static async Task<string[]> UpdateMembersAsync(
         INatsJSContext js,
         string streamName,

src/Synadia.Orbit.PCGroups/NatsPcgMemberMappingValidator.cs

@@ -77,10 +77,10 @@ public static void Validate(NatsPcgMemberMapping[]? memberMappings, uint maxMemb
     }
 
     /// <summary>
-    /// Validates filter and partitioning wildcards for elastic consumer groups.
+    /// Validates a partitioning filter for elastic consumer groups.
     /// </summary>
     /// <param name="filter">The subject filter.</param>
-    /// <param name="partitioningWildcards">The partitioning wildcard positions (1-indexed).</param>
+    /// <param name="partitioningWildcards">The partitioning wildcard positions (1-indexed). Empty array means partition by full subject.</param>
     /// <exception cref="NatsPcgConfigurationException">Thrown when validation fails.</exception>
     public static void ValidateFilterAndWildcards(string filter, int[] partitioningWildcards)
     {
@@ -89,25 +89,23 @@ public static void ValidateFilterAndWildcards(string filter, int[] partitioningW
             throw new NatsPcgConfigurationException("Filter is required for elastic consumer groups");
         }
 
-        // Count wildcards in filter
-        var filterTokens = filter.Split('.');
-        int numWildcards = 0;
-        foreach (var token in filterTokens)
+        if (partitioningWildcards == null)
         {
-            if (token == "*")
-            {
-                numWildcards++;
-            }
+            throw new NatsPcgConfigurationException("PartitioningWildcards must not be null");
         }
 
-        if (numWildcards < 1)
+        // Empty array means partition by full subject - no wildcard requirements
+        if (IsPartitionByFullSubject(partitioningWildcards))
         {
-            throw new NatsPcgConfigurationException("Filter must contain at least one '*' wildcard");
+            return;
         }
 
-        if (partitioningWildcards == null || partitioningWildcards.Length == 0)
+        // Count wildcards in filter
+        int numWildcards = CountWildcards(filter);
+
+        if (numWildcards < 1)
         {
-            throw new NatsPcgConfigurationException("PartitioningWildcards must contain at least one element");
+            throw new NatsPcgConfigurationException("Filter must contain at least one '*' wildcard");
         }
 
         if (partitioningWildcards.Length > numWildcards)
@@ -133,4 +131,57 @@ public static void ValidateFilterAndWildcards(string filter, int[] partitioningW
             }
         }
     }
+
+    /// <summary>
+    /// Validates a partitioning filter record for elastic consumer groups.
+    /// </summary>
+    /// <param name="partitioningFilter">The partitioning filter to validate.</param>
+    /// <exception cref="NatsPcgConfigurationException">Thrown when validation fails.</exception>
+    public static void ValidatePartitioningFilter(NatsPcgPartitioningFilter partitioningFilter)
+    {
+        ValidateFilterAndWildcards(partitioningFilter.Filter, partitioningFilter.PartitioningWildcards);
+    }
+
+    /// <summary>
+    /// Validates multiple partitioning filters for elastic consumer groups.
+    /// </summary>
+    /// <param name="partitioningFilters">The partitioning filters to validate.</param>
+    /// <exception cref="NatsPcgConfigurationException">Thrown when validation fails.</exception>
+    public static void ValidatePartitioningFilters(NatsPcgPartitioningFilter[] partitioningFilters)
+    {
+        if (partitioningFilters == null)
+        {
+            return;
+        }
+
+        foreach (var pf in partitioningFilters)
+        {
+            ValidatePartitioningFilter(pf);
+        }
+    }
+
+    /// <summary>
+    /// Determines if the partitioning wildcards represent partition-by-full-subject mode.
+    /// </summary>
+    /// <param name="partitioningWildcards">The partitioning wildcard positions.</param>
+    /// <returns>True if the array is empty (partition by full subject).</returns>
+    public static bool IsPartitionByFullSubject(int[] partitioningWildcards)
+    {
+        return partitioningWildcards != null && partitioningWildcards.Length == 0;
+    }
+
+    private static int CountWildcards(string filter)
+    {
+        var filterTokens = filter.Split('.');
+        int numWildcards = 0;
+        foreach (var token in filterTokens)
+        {
+            if (token == "*")
+            {
+                numWildcards++;
+            }
+        }
+
+        return numWildcards;
+    }
 }

src/Synadia.Orbit.PCGroups/NatsPcgPartitionDistributor.cs

@@ -15,12 +15,14 @@ public static class NatsPcgPartitionDistributor
     /// <param name="maxMembers">Maximum number of members (equals number of partitions).</param>
     /// <param name="memberMappings">Optional explicit member-to-partition mappings.</param>
     /// <param name="memberName">The member name to generate filters for.</param>
-    /// <returns>Array of filters like ["0.>", "3.>", "6.>"].</returns>
+    /// <param name="filter">The subject filter to embed in partition filters.</param>
+    /// <returns>Array of filters like ["0.orders.*", "3.orders.*"].</returns>
     public static string[] GeneratePartitionFilters(
         string[]? members,
         uint maxMembers,
         NatsPcgMemberMapping[]? memberMappings,
-        string memberName)
+        string memberName,
+        string filter)
     {
         int[] partitions;
 
@@ -54,7 +56,7 @@ public static string[] GeneratePartitionFilters(
         string[] filters = new string[partitions.Length];
         for (int i = 0; i < partitions.Length; i++)
         {
-            filters[i] = $"{partitions[i]}.>";
+            filters[i] = $"{partitions[i]}.{filter}";
         }
 
         return filters;