Merge pull request #332 from thomhurst/fix/async-processor-correctness

Fix processor correctness: lazy re-enumeration, broken cancellation on dispose, hung awaiters

Author: thomhurst

.github/workflows/dotnet.yml

@@ -28,7 +28,11 @@ jobs:
       - name: Setup .NET
         uses: actions/setup-dotnet@v4
         with:
-          dotnet-version: 9.0.x
+          # 10.0.x runs the pipeline; 8.0.x and 9.0.x provide runtimes for the multi-targeted test project
+          dotnet-version: |
+            8.0.x
+            9.0.x
+            10.0.x
       - name: Run Pipeline
         run: dotnet run -c Release
         working-directory: "EnumerableAsyncProcessor.Pipeline"

.gitignore

@@ -351,3 +351,5 @@ MigrationBackup/
 
 # Claude local settings
 .claude/settings.local.json
+
+.playwright-mcp/

EnumerableAsyncProcessor.Pipeline/EnumerableAsyncProcessor.Pipeline.csproj

@@ -2,15 +2,20 @@
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <TargetFramework>net9.0</TargetFramework>
+        <TargetFramework>net10.0</TargetFramework>
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>enable</Nullable>
         <SignAssembly>false</SignAssembly>
     </PropertyGroup>
 
     <ItemGroup>
-      <PackageReference Include="ModularPipelines.DotNet" Version="3.1.90" />
-      <PackageReference Include="ModularPipelines.Git" Version="3.1.90" />
+      <PackageReference Include="ModularPipelines.DotNet" Version="3.2.8" />
+      <PackageReference Include="ModularPipelines.Git" Version="3.2.8" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <!-- Pipeline.CreateBuilder resolves configuration relative to the output directory -->
+      <None Include="appsettings.json" CopyToOutputDirectory="PreserveNewest" />
     </ItemGroup>
 
 </Project>

EnumerableAsyncProcessor.Pipeline/Modules/LocalMachine/AddLocalNugetSourceModule.cs

@@ -1,4 +1,5 @@
-using ModularPipelines.Attributes;
+using ModularPipelines.Attributes;
+using ModularPipelines.Configuration;
 using ModularPipelines.Context;
 using ModularPipelines.DotNet.Extensions;
 using ModularPipelines.DotNet.Options;
@@ -11,21 +12,24 @@ namespace EnumerableAsyncProcessor.Pipeline.Modules.LocalMachine;
 [DependsOn<CreateLocalNugetFolderModule>]
 public class AddLocalNugetSourceModule : Module<CommandResult>
 {
-    protected override async Task<bool> ShouldIgnoreFailures(IPipelineContext context, Exception exception)
+    protected override ModuleConfiguration Configure()
     {
-        await Task.Yield();
-        return exception is CommandException commandException &&
-                               commandException.StandardOutput.Contains("The name specified has already been added to the list of available package sources");
+        return ModuleConfiguration.Create()
+            .WithIgnoreFailuresWhen((_, exception) =>
+                exception is CommandException commandException &&
+                commandException.StandardOutput.Contains("The name specified has already been added to the list of available package sources"))
+            .Build();
     }
 
-    protected override async Task<CommandResult?> ExecuteAsync(IPipelineContext context, CancellationToken cancellationToken)
+    protected override async Task<CommandResult?> ExecuteAsync(IModuleContext context, CancellationToken cancellationToken)
     {
-        var localNugetPathResult = await GetModule<CreateLocalNugetFolderModule>();
+        var localNugetPathResult = await context.GetModule<CreateLocalNugetFolderModule>();
 
         return await context.DotNet().Nuget.Add
-            .Source(new DotNetNugetAddSourceOptions(localNugetPathResult.Value!)
+            .Source(new DotNetNugetAddSourceOptions
             {
+                Packagesourcepath = localNugetPathResult.ValueOrDefault!.Path,
                 Name = "ModularPipelinesLocalNuGet"
-            }, cancellationToken);
+            }, cancellationToken: cancellationToken);
     }
 }

EnumerableAsyncProcessor.Pipeline/Modules/LocalMachine/CreateLocalNugetFolderModule.cs

@@ -1,4 +1,4 @@
-using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging;
 using ModularPipelines.Attributes;
 using ModularPipelines.Context;
 using ModularPipelines.FileSystem;
@@ -10,17 +10,16 @@ namespace EnumerableAsyncProcessor.Pipeline.Modules.LocalMachine;
 [DependsOn<PackagePathsParserModule>]
 public class CreateLocalNugetFolderModule : Module<Folder>
 {
-    protected override async Task<Folder?> ExecuteAsync(IPipelineContext context, CancellationToken cancellationToken)
+    protected override Task<Folder?> ExecuteAsync(IModuleContext context, CancellationToken cancellationToken)
     {
-        var localNugetRepositoryFolder = context.FileSystem.GetFolder(Environment.SpecialFolder.ApplicationData)
+        var localNugetRepositoryFolder = context.Files
+            .GetFolder(Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData))
             .GetFolder("ModularPipelines")
             .GetFolder("LocalNuget")
             .Create();
-        
-        await Task.Yield();
 
         context.Logger.LogInformation("Local NuGet Repository Path: {Path}", localNugetRepositoryFolder.Path);
 
-        return localNugetRepositoryFolder;
+        return Task.FromResult<Folder?>(localNugetRepositoryFolder);
     }
 }

EnumerableAsyncProcessor.Pipeline/Modules/LocalMachine/UploadPackagesToLocalNuGetModule.cs

@@ -14,26 +14,31 @@ namespace EnumerableAsyncProcessor.Pipeline.Modules.LocalMachine;
 [DependsOn<CreateLocalNugetFolderModule>]
 public class UploadPackagesToLocalNuGetModule : Module<CommandResult[]>
 {
-    protected override async Task OnBeforeExecute(IPipelineContext context)
+    protected override async Task OnBeforeExecuteAsync(IModuleContext context, CancellationToken cancellationToken)
     {
-        var packagePaths = await GetModule<PackagePathsParserModule>();
-        foreach (var packagePath in packagePaths.Value!)
+        var packagePaths = await context.GetModule<PackagePathsParserModule>();
+
+        foreach (var packagePath in packagePaths.ValueOrDefault ?? [])
         {
             context.Logger.LogInformation("[Local Directory] Uploading {File}", packagePath);
         }
 
-        await base.OnBeforeExecute(context);
+        await base.OnBeforeExecuteAsync(context, cancellationToken);
     }
 
-    protected override async Task<CommandResult[]?> ExecuteAsync(IPipelineContext context, CancellationToken cancellationToken)
+    protected override async Task<CommandResult[]?> ExecuteAsync(IModuleContext context, CancellationToken cancellationToken)
     {
-        var localRepoLocation = await GetModule<CreateLocalNugetFolderModule>();
-        var packagePaths = await GetModule<PackagePathsParserModule>();
-        return await packagePaths.Value!.SelectAsync(async file => await context.DotNet()
-            .Nuget
-            .Push(new DotNetNugetPushOptions(file)
-            {
-                Source = localRepoLocation.Value!,
-            }, cancellationToken), cancellationToken: cancellationToken).ProcessOneAtATime();
+        var localRepoLocation = await context.GetModule<CreateLocalNugetFolderModule>();
+        var packagePaths = await context.GetModule<PackagePathsParserModule>();
+
+        return await packagePaths.ValueOrDefault!
+            .SelectAsync(file => context.DotNet()
+                .Nuget
+                .Push(new DotNetNugetPushOptions
+                {
+                    Path = file.Path,
+                    Source = localRepoLocation.ValueOrDefault!.Path,
+                }, cancellationToken: cancellationToken), cancellationToken: cancellationToken)
+            .ProcessOneAtATime();
     }
-}
+}

EnumerableAsyncProcessor.Pipeline/Modules/NugetVersionGeneratorModule.cs

@@ -1,30 +1,28 @@
-using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging;
 using ModularPipelines.Context;
 using ModularPipelines.Git.Extensions;
+using ModularPipelines.Models;
 using ModularPipelines.Modules;
 
-// ReSharper disable HeuristicUnreachableCode
-#pragma warning disable CS0162
-
 namespace EnumerableAsyncProcessor.Pipeline.Modules;
 
 public class NugetVersionGeneratorModule : Module<string>
 {
-    protected override async Task<string?> ExecuteAsync(IPipelineContext context, CancellationToken cancellationToken)
+    protected override async Task<string?> ExecuteAsync(IModuleContext context, CancellationToken cancellationToken)
     {
         var gitVersionInformation = await context.Git().Versioning.GetGitVersioningInformation();
-        
+
         if (gitVersionInformation.BranchName == "main")
         {
             return gitVersionInformation.SemVer!;
         }
-        
+
         return $"{gitVersionInformation.Major}.{gitVersionInformation.Minor}.{gitVersionInformation.Patch}-{gitVersionInformation.PreReleaseLabel}-{gitVersionInformation.CommitsSinceVersionSource}";
     }
 
-    protected override async Task OnAfterExecute(IPipelineContext context)
+    protected override Task<ModuleResult<string>?> OnAfterExecuteAsync(IModuleContext context, ModuleResult<string> result, CancellationToken cancellationToken)
     {
-        var moduleResult = await this;
-        context.Logger.LogInformation("NuGet Version to Package: {Version}", moduleResult.Value);
+        context.Logger.LogInformation("NuGet Version to Package: {Version}", result.ValueOrDefault);
+        return base.OnAfterExecuteAsync(context, result, cancellationToken);
     }
 }

EnumerableAsyncProcessor.Pipeline/Modules/PackProjectsModule.cs

@@ -15,29 +15,31 @@ namespace EnumerableAsyncProcessor.Pipeline.Modules;
 [DependsOn<RunUnitTestsModule>]
 public class PackProjectsModule : Module<List<CommandResult>>
 {
-    protected override async Task<List<CommandResult>?> ExecuteAsync(IPipelineContext context, CancellationToken cancellationToken)
+    protected override async Task<List<CommandResult>?> ExecuteAsync(IModuleContext context, CancellationToken cancellationToken)
     {
         var results = new List<CommandResult>();
-        var packageVersion = await GetModule<NugetVersionGeneratorModule>();
-        var projectFiles = context.Git().RootDirectory!.GetFiles(f => GetProjectsPredicate(f, context));
+        var packageVersion = await context.GetModule<NugetVersionGeneratorModule>();
+        var projectFiles = context.Git().RootDirectory.GetFiles(f => GetProjectsPredicate(f, context));
+
         foreach (var projectFile in projectFiles)
         {
-            results.Add(await context.DotNet().Pack(new DotNetPackOptions { 
-                ProjectSolution = projectFile.Path, 
-                Configuration = Configuration.Release, 
+            results.Add(await context.DotNet().Pack(new DotNetPackOptions
+            {
+                ProjectSolution = projectFile.Path,
+                Configuration = "Release",
                 Properties =
                 [
-                    ("PackageVersion", packageVersion.Value)!, 
-                    ("Version", packageVersion.Value)!
+                    ("PackageVersion", packageVersion.ValueOrDefault)!,
+                    ("Version", packageVersion.ValueOrDefault)!
                 ],
                 IncludeSource = true
-            }, cancellationToken));
+            }, cancellationToken: cancellationToken));
         }
 
         return results;
     }
 
-    private bool GetProjectsPredicate(File file, IPipelineContext context)
+    private bool GetProjectsPredicate(File file, IModuleContext context)
     {
         var path = file.Path;
         if (!path.EndsWith(".csproj", StringComparison.OrdinalIgnoreCase))
@@ -55,4 +57,4 @@ private bool GetProjectsPredicate(File file, IPipelineContext context)
         context.Logger.LogInformation("Found File: {File}", path);
         return true;
     }
-}
+}

EnumerableAsyncProcessor.Pipeline/Modules/PackageFilesRemovalModule.cs

@@ -6,15 +6,15 @@ namespace EnumerableAsyncProcessor.Pipeline.Modules;
 
 public class PackageFilesRemovalModule : Module
 {
-    protected override async Task<IDictionary<string, object>?> ExecuteAsync(IPipelineContext context, CancellationToken cancellationToken)
+    protected override Task ExecuteModuleAsync(IModuleContext context, CancellationToken cancellationToken)
     {
-        var packageFiles = context.Git().RootDirectory.GetFiles(path => path.Extension is ".nupkg");
+        var packageFiles = context.Git().RootDirectory.GetFiles(file => file.Extension is ".nupkg");
 
         foreach (var packageFile in packageFiles)
         {
             packageFile.Delete();
         }
 
-        return await NothingAsync();
+        return Task.CompletedTask;
     }
 }

EnumerableAsyncProcessor.Pipeline/Modules/PackagePathsParserModule.cs

@@ -8,11 +8,11 @@ namespace EnumerableAsyncProcessor.Pipeline.Modules;
 [DependsOn<PackProjectsModule>]
 public class PackagePathsParserModule : Module<List<File>>
 {
-    protected override async Task<List<File>?> ExecuteAsync(IPipelineContext context, CancellationToken cancellationToken)
+    protected override async Task<List<File>?> ExecuteAsync(IModuleContext context, CancellationToken cancellationToken)
     {
-        var packPackagesModuleResult = await GetModule<PackProjectsModule>();
+        var packPackagesModuleResult = await context.GetModule<PackProjectsModule>();
 
-        return packPackagesModuleResult.Value!
+        return packPackagesModuleResult.ValueOrDefault!
             .Select(x => x.StandardOutput)
             .Select(x => x.Split("Successfully created package '")[1])
             .Select(x => x.Split("'.")[0])