diff --git a/src/EventTests/CommandLine/ProjectionInputTests.cs b/src/EventTests/CommandLine/ProjectionInputTests.cs new file mode 100644 index 00000000..91606083 --- /dev/null +++ b/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); + } +} diff --git a/src/JasperFx.Events/CommandLine/ProjectionHost.cs b/src/JasperFx.Events/CommandLine/ProjectionHost.cs index 850b04e7..f2c43028 100644 --- a/src/JasperFx.Events/CommandLine/ProjectionHost.cs +++ b/src/JasperFx.Events/CommandLine/ProjectionHost.cs @@ -95,7 +95,22 @@ public async Task TryRebuildShardsAsync(EventStoreDatabaseIdentif var list = new List(); - 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(); diff --git a/src/JasperFx.Events/CommandLine/ProjectionInput.cs b/src/JasperFx.Events/CommandLine/ProjectionInput.cs index 21f0a52d..87b5b81a 100644 --- a/src/JasperFx.Events/CommandLine/ProjectionInput.cs +++ b/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; } + + /// + /// jasperfx#420: resolve the effective for the + /// per-database rebuild fan-out. A null or non-positive yields -1 + /// (unbounded), preserving the historical behavior. + /// + public int ResolveMaxDegreeOfParallelism() => MaxConcurrentFlag is > 0 ? MaxConcurrentFlag.Value : -1; }