Fix #420: cap concurrent projection rebuilds per database

src/EventTests/CommandLine/ProjectionInputTests.cs

@@ -0,0 +1,37 @@
+using JasperFx.Events.CommandLine;
+using Shouldly;
+
+namespace EventTests.CommandLine;
+
+public class ProjectionInputTests
+{
+    [Fact]
+    public void unset_max_concurrent_is_unbounded()
+    {
+        // jasperfx#420: no flag preserves the historical unbounded fan-out (-1 == unbounded for
+        // ParallelOptions.MaxDegreeOfParallelism).
+        new ProjectionInput().ResolveMaxDegreeOfParallelism().ShouldBe(-1);
+    }
+
+    [Theory]
+    [InlineData(1)]
+    [InlineData(4)]
+    [InlineData(64)]
+    public void positive_max_concurrent_is_honored(int cap)
+    {
+        new ProjectionInput { MaxConcurrentFlag = cap }
+            .ResolveMaxDegreeOfParallelism().ShouldBe(cap);
+    }
+
+    [Theory]
+    [InlineData(0)]
+    [InlineData(-1)]
+    [InlineData(-5)]
+    public void non_positive_max_concurrent_falls_back_to_unbounded(int cap)
+    {
+        // A nonsensical override must not deadlock the rebuild (MaxDegreeOfParallelism = 0 throws);
+        // treat it as "unbounded".
+        new ProjectionInput { MaxConcurrentFlag = cap }
+            .ResolveMaxDegreeOfParallelism().ShouldBe(-1);
+    }
+}

src/JasperFx.Events/CommandLine/ProjectionHost.cs

@@ -95,7 +95,22 @@ public async Task<RebuildStatus> TryRebuildShardsAsync(EventStoreDatabaseIdentif
 
         var list = new List<Exception>();
 
-        await Parallel.ForEachAsync(projectionNames, _cancellation.Token,
+        // jasperfx#420: cap the per-database rebuild fan-out so a wide store
+        // (many projections, especially under per-tenant partitioning) cannot
+        // blow the connection pool / thrash the buffer cache. A non-positive or
+        // unset flag preserves the previous unbounded behavior.
+        var maxConcurrent = input.ResolveMaxDegreeOfParallelism();
+        var parallelOptions = new ParallelOptions
+        {
+            CancellationToken = _cancellation.Token,
+            MaxDegreeOfParallelism = maxConcurrent
+        };
+
+        AnsiConsole.MarkupLine(maxConcurrent > 0
+            ? $"[grey]Rebuilding with a maximum of {maxConcurrent} concurrent projection(s) per database[/]"
+            : "[grey]Rebuilding with unbounded concurrency per database[/]");
+
+        await Parallel.ForEachAsync(projectionNames, parallelOptions,
                 async (projectionName, token) =>
                 {
                     shardTimeout ??= 5.Minutes();

src/JasperFx.Events/CommandLine/ProjectionInput.cs

@@ -35,4 +35,15 @@ public class ProjectionInput: NetCoreInput
 
     [Description("If specified, advances the projection high water mark to the latest event sequence")]
     public bool AdvanceFlag { get; set; }
+
+    [Description("Maximum number of projections to rebuild concurrently within a single database. Caps the per-database rebuild fan-out for one-off operational rebuilds. Default is unbounded.")]
+    [FlagAlias("max-concurrent", longAliasOnly: true)]
+    public int? MaxConcurrentFlag { get; set; }
+
+    /// <summary>
+    /// jasperfx#420: resolve the effective <see cref="ParallelOptions.MaxDegreeOfParallelism"/> for the
+    /// per-database rebuild fan-out. A null or non-positive <see cref="MaxConcurrentFlag"/> yields -1
+    /// (unbounded), preserving the historical behavior.
+    /// </summary>
+    public int ResolveMaxDegreeOfParallelism() => MaxConcurrentFlag is > 0 ? MaxConcurrentFlag.Value : -1;
 }