+semver:minor - Feature/refactor waiting dependencies (#486)

* Remove WaitHandler.cs

* Simplify start dependency

* ExceptionContainer.cs

* Message

* Update dotnet.yml

* Formatting Markdown

* Expose StartInternal

* Refactor StartModule

* Fix test

---------

Co-authored-by: Tom Longhurst <[email protected]>

Author: thomhurst

.github/workflows/dotnet.yml

@@ -74,8 +74,8 @@ jobs:
           GitHub__PullRequest__Branch: ${{ github.event.pull_request.head.ref }}
           GitHub__PullRequest__Sha: ${{ github.event.pull_request.head.sha }}
           GitHub__PullRequest__Author: ${{ github.event.pull_request.user.login }}
-          Publish__ShouldPublish: ${{ github.event.inputs.publish-packages }}
-          Publish__IsAlpha: ${{ github.event.inputs.is-alpha }}
+          Publish__ShouldPublish: ${{ github.event.inputs.publish-packages || false}}
+          Publish__IsAlpha: ${{ github.event.inputs.is-alpha || true}}
           Codacy__ApiKey: ${{ secrets.CODACY_APIKEY }}
           CodeCov__Token: ${{ secrets.CODECOV_TOKEN }}
           EMAIL_PASSWORD: ${{ secrets.EMAIL_PASSWORD }}

README.md

@@ -63,7 +63,6 @@ Define your pipeline in .NET! Strong types, intellisense, parallelisation, and t
 | ModularPipelines.WinGet | Helpers for interacting with the Windows Package Manager. | [![nuget](https://img.shields.io/nuget/v/ModularPipelines.WinGet.svg)](https://www.nuget.org/packages/ModularPipelines.WinGet/) |
 | ModularPipelines.Yarn | Helpers for interacting with Yarn CLI. | [![nuget](https://img.shields.io/nuget/v/ModularPipelines.Yarn.svg)](https://www.nuget.org/packages/ModularPipelines.Yarn/) |
 
-
 ## Getting Started
 
 If you want to see how to get started, or want to know more about ModularPipelines, [read the Documentation here](https://thomhurst.github.io/ModularPipelines)

src/ModularPipelines.Build/ReleaseNotes.md

@@ -0,0 +1 @@
+*   Refactor how modules wait for their dependencies

src/ModularPipelines/DependencyInjection/DependencyInjectionSetup.cs

@@ -67,6 +67,7 @@ public static void Initialize(IServiceCollection services)
         // Singletons
         services
             .AddSingleton<IConsolePrinter, ConsolePrinter>()
+            .AddSingleton<IExceptionContainer, ExceptionContainer>()
             .AddSingleton<IPipelineContextProvider, ModuleContextProvider>()
             .AddSingleton<IDependencyChainProvider, DependencyChainProvider>()
             .AddSingleton<IDependencyDetector, DependencyDetector>()

src/ModularPipelines/Engine/ExceptionContainer.cs

@@ -0,0 +1,24 @@
+namespace ModularPipelines.Engine;
+
+internal class ExceptionContainer : IExceptionContainer
+{
+    private readonly List<Exception> _exceptions = [];
+    
+    public void RegisterException(Exception exception)
+    {
+        _exceptions.Add(exception);
+    }
+
+    public void ThrowExceptions()
+    {
+        if (_exceptions.Count == 1)
+        {
+            throw _exceptions.First();
+        }
+
+        if (_exceptions.Any())
+        {
+            throw new AggregateException(_exceptions);
+        }
+    }
+}

src/ModularPipelines/Engine/Executors/ModuleHandlers/IWaitHandler.cs

@@ -11,17 +11,20 @@ internal class PipelineExecutor : IPipelineExecutor
     private readonly IModuleExecutor _moduleExecutor;
     private readonly EngineCancellationToken _engineCancellationToken;
     private readonly ILogger<PipelineExecutor> _logger;
+    private readonly IExceptionContainer _exceptionContainer;
 
     public PipelineExecutor(
         IPipelineSetupExecutor pipelineSetupExecutor,
         IModuleExecutor moduleExecutor,
         EngineCancellationToken engineCancellationToken,
-        ILogger<PipelineExecutor> logger)
+        ILogger<PipelineExecutor> logger,
+        IExceptionContainer exceptionContainer)
     {
         _pipelineSetupExecutor = pipelineSetupExecutor;
         _moduleExecutor = moduleExecutor;
         _engineCancellationToken = engineCancellationToken;
         _logger = logger;
+        _exceptionContainer = exceptionContainer;
     }
 
     public async Task<PipelineSummary> ExecuteAsync(List<ModuleBase> runnableModules,
@@ -57,6 +60,8 @@ public async Task<PipelineSummary> ExecuteAsync(List<ModuleBase> runnableModules
         {
             throw exception;
         }
+        
+        _exceptionContainer.ThrowExceptions();
 
         return pipelineSummary;
     }

src/ModularPipelines/Engine/Executors/ModuleHandlers/WaitHandler.cs

@@ -0,0 +1,8 @@
+namespace ModularPipelines.Engine;
+
+internal interface IExceptionContainer
+{
+    void RegisterException(Exception exception);
+    
+    void ThrowExceptions();
+}

src/ModularPipelines/Engine/Executors/PipelineExecutor.cs

@@ -1,9 +1,12 @@
 using System.Collections.Concurrent;
 using System.Reflection;
 using EnumerableAsyncProcessor.Extensions;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using ModularPipelines.Attributes;
+using ModularPipelines.Exceptions;
 using ModularPipelines.Extensions;
+using ModularPipelines.Models;
 using ModularPipelines.Modules;
 using ModularPipelines.Options;
 
@@ -15,18 +18,24 @@ internal class ModuleExecutor : IModuleExecutor
     private readonly IOptions<PipelineOptions> _pipelineOptions;
     private readonly ISafeModuleEstimatedTimeProvider _moduleEstimatedTimeProvider;
     private readonly IModuleDisposer _moduleDisposer;
+    private readonly IEnumerable<ModuleBase> _allModules;
+    private readonly IExceptionContainer _exceptionContainer;
 
     private readonly ConcurrentDictionary<ModuleBase, Task<ModuleBase>> _moduleExecutionTasks = new();
 
     public ModuleExecutor(IPipelineSetupExecutor pipelineSetupExecutor,
         IOptions<PipelineOptions> pipelineOptions,
         ISafeModuleEstimatedTimeProvider moduleEstimatedTimeProvider,
-        IModuleDisposer moduleDisposer)
+        IModuleDisposer moduleDisposer,
+        IEnumerable<ModuleBase> allModules,
+        IExceptionContainer exceptionContainer)
     {
         _pipelineSetupExecutor = pipelineSetupExecutor;
         _pipelineOptions = pipelineOptions;
         _moduleEstimatedTimeProvider = moduleEstimatedTimeProvider;
         _moduleDisposer = moduleDisposer;
+        _allModules = allModules;
+        _exceptionContainer = exceptionContainer;
     }
 
     public async Task<IEnumerable<ModuleBase>> ExecuteAsync(IReadOnlyList<ModuleBase> modules)
@@ -75,7 +84,7 @@ public async Task<ModuleBase> ExecuteAsync(ModuleBase module)
     {
         try
         {
-            return await ExecuteWithLockAsync(module);
+            return await StartModule(module);
         }
         catch (TaskCanceledException)
         {
@@ -86,14 +95,6 @@ public async Task<ModuleBase> ExecuteAsync(ModuleBase module)
         }
     }
 
-    private Task<ModuleBase> ExecuteWithLockAsync(ModuleBase module)
-    {
-        lock (module)
-        {
-            return _moduleExecutionTasks.GetOrAdd(module, @base => StartModule(module));
-        }
-    }
-
     private async Task<ModuleBase[]> ProcessKeyedNonParallelModules(List<ModuleBase> keyedNonParallelModules,
         List<ModuleBase> moduleResults)
     {
@@ -150,44 +151,70 @@ private async Task<ModuleBase[]> ProcessKeyedNonParallelModules(List<ModuleBase>
 
     private async Task<ModuleBase> StartModule(ModuleBase module)
     {
-        if (module.IsStarted || module.ExecutionTask.IsCompleted)
+        return await _moduleExecutionTasks.GetOrAdd(module, async @base =>
         {
-            await module.ExecutionTask;
-            return module;
-        }
+            var dependencies = module.GetModuleDependencies();
 
-        try
-        {
-            await _pipelineSetupExecutor.OnBeforeModuleStartAsync(module);
+            foreach (var dependency in dependencies)
+            {
+                await StartDependency(module, dependency.DependencyType, dependency.IgnoreIfNotRegistered);
+            }
 
-            await module.ExecutionTask;
+            try
+            {
+                await _pipelineSetupExecutor.OnBeforeModuleStartAsync(module);
 
-            await _moduleEstimatedTimeProvider.SaveModuleTimeAsync(module.GetType(), module.Duration);
+                await module.StartInternal();
 
-            await _pipelineSetupExecutor.OnAfterModuleEndAsync(module);
+                await _moduleEstimatedTimeProvider.SaveModuleTimeAsync(module.GetType(), module.Duration);
 
-            return module;
-        }
-        finally
-        {
-            if (!_pipelineOptions.Value.ShowProgressInConsole)
+                await _pipelineSetupExecutor.OnAfterModuleEndAsync(module);
+
+                return module;
+            }
+            finally
             {
-                await _moduleDisposer.DisposeAsync(module);
+                if (!_pipelineOptions.Value.ShowProgressInConsole)
+                {
+                    await _moduleDisposer.DisposeAsync(module);
+                }
             }
-        }
+        });
     }
 
-    private async Task<IEnumerable<ModuleBase>> ProcessGroup(
-        IGrouping<string?, ModuleBase> moduleBases,
-        ICollection<ModuleBase> moduleResults)
+    private async Task StartDependency(ModuleBase requestingModule, Type dependencyType, bool ignoreIfNotRegistered)
     {
-        var executionProcessor = moduleBases.SelectAsync(ExecuteAsync);
+        var module = _allModules.FirstOrDefault(x => x.GetType() == dependencyType);
+
+        if (module is null && ignoreIfNotRegistered)
+        {
+            requestingModule.Context.Logger.LogDebug("{Module} was not registered so not waiting", dependencyType.Name);
+            return;
+        }
 
-        if (string.IsNullOrEmpty(moduleBases.Key))
+        if (module is null)
         {
-            return await executionProcessor.ProcessInParallel().GetEnumerableTasks().ToArray().WhenAllFailFast();
+            throw new ModuleNotRegisteredException($"The module {dependencyType.Name} has not been registered", null);
         }
+        
+        requestingModule.Context.Logger.LogDebug("{RequestingModule} is waiting for {Module}", requestingModule.GetType().Name, dependencyType.Name);
 
-        return await executionProcessor.ProcessOneAtATime();
+        try
+        {
+            await StartModule(module);
+        }
+        catch (Exception e) when (requestingModule.ModuleRunType == ModuleRunType.AlwaysRun)
+        {
+            _exceptionContainer.RegisterException(new AlwaysRunPostponedException($"{dependencyType.Name} threw an exception when {requestingModule.GetType().Name} was waiting for it as a dependency", e));
+            requestingModule.Context.Logger.LogError(e, "Ignoring Exception due to 'AlwaysRun' set");
+        }
+        catch (DependencyFailedException)
+        {
+            throw;
+        }
+        catch (Exception e)
+        {
+            throw new DependencyFailedException(e, module);
+        }
     }
 }