fix: Resolve unobserved task exceptions in AsyncEnumerable processors

Fixed ObjectDisposedException occurring when semaphore was disposed while
background tasks were still trying to release it.

Changes:
- Wrapped task creation and execution in try-finally blocks
- Ensured all tasks complete before disposing semaphore using finally blocks
- Added exception handling for cancellation scenarios
- Removed unnecessary Task.Yield calls in AsyncEnumerable processors

This prevents the "Cannot access a disposed object" exceptions that were
appearing during test runs when tasks were finalized by the GC.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: thomhurst

EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs

@@ -25,7 +25,7 @@ internal AsyncEnumerableParallelProcessor(
     public async Task ExecuteAsync()
     {
         var cancellationToken = _cancellationTokenSource.Token;
-        var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
+        using var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
         var tasks = new List<Task>();
 
         try
@@ -39,8 +39,7 @@ public async Task ExecuteAsync()
                 {
                     try
                     {
-                        // Yield to ensure we don't block the thread if _taskSelector is synchronous
-                        await Task.Yield();
+                        // Removed Task.Yield - parallelism is now handled at the processor level
                         await _taskSelector(capturedItem).ConfigureAwait(false);
                     }
                     finally
@@ -51,12 +50,15 @@ public async Task ExecuteAsync()
 
                 tasks.Add(task);
             }
-
-            await Task.WhenAll(tasks).ConfigureAwait(false);
         }
         finally
         {
-            semaphore.Dispose();
+            // Always wait for all tasks to complete before the using block disposes the semaphore
+            // This ensures the semaphore is not disposed while tasks are still running
+            if (tasks.Count > 0)
+            {
+                await Task.WhenAll(tasks).ConfigureAwait(false);
+            }
         }
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs

@@ -26,7 +26,7 @@ internal ResultAsyncEnumerableParallelProcessor(
     public async IAsyncEnumerable<TOutput> ExecuteAsync()
     {
         var cancellationToken = _cancellationTokenSource.Token;
-        var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
+        using var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
         var tasks = new List<Task<TOutput>>();
 
         try
@@ -36,7 +36,18 @@ public async IAsyncEnumerable<TOutput> ExecuteAsync()
                 await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
 
                 var capturedItem = item;
-                var task = ProcessItemAsync(capturedItem, semaphore, cancellationToken);
+                // Use Task.Run to ensure parallelism and prevent blocking
+                var task = Task.Run(async () => 
+                {
+                    try
+                    {
+                        return await _taskSelector(capturedItem).ConfigureAwait(false);
+                    }
+                    finally
+                    {
+                        semaphore.Release();
+                    }
+                }, cancellationToken);
                 tasks.Add(task);
 
                 // Yield completed results
@@ -56,24 +67,19 @@ public async IAsyncEnumerable<TOutput> ExecuteAsync()
         }
         finally
         {
-            semaphore.Dispose();
-        }
-    }
-
-    private async Task<TOutput> ProcessItemAsync(
-        TInput item, 
-        SemaphoreSlim semaphore, 
-        CancellationToken cancellationToken)
-    {
-        try
-        {
-            // Yield to ensure we don't block the thread if _taskSelector is synchronous
-            await Task.Yield();
-            return await _taskSelector(item).ConfigureAwait(false);
-        }
-        finally
-        {
-            semaphore.Release();
+            // Ensure all tasks complete before the using block disposes the semaphore
+            // This handles cancellation or exception scenarios
+            if (tasks.Count > 0)
+            {
+                try
+                {
+                    await Task.WhenAll(tasks).ConfigureAwait(false);
+                }
+                catch
+                {
+                    // Ignore exceptions here as they've already been handled
+                }
+            }
         }
     }
 }