diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs
index 3003a428..23a89ca6 100644
--- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs
+++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs
@@ -53,6 +53,11 @@ [EnumeratorCancellation] CancellationToken token
{
await foreach (var item in stream.ConfigureAwait(false))
{
+ // Stryker disable once Statement: equivalent in every reachable path — the pipeline head
+ // (EtlPipelineImpl.CountExtracted) always runs the same ThrowIfCancellationRequested one
+ // layer up (it is the sole source of this sink's stream), so removing this defence-in-depth
+ // check cannot change observable cancellation behaviour. The head guard itself IS tested
+ // directly — see EtlPipelineTests.AsAsyncEnumerable_honours_cancellation_at_the_head.
token.ThrowIfCancellationRequested();
state.RecordsLoaded++;
progress?.Report(state.Snapshot());
diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
index 7dc009a7..652112c0 100644
--- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
+++ b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
@@ -409,10 +409,14 @@ protected void IncrementCurrentSkippedItemCount()
// win the CompareExchange records the start; later calls are a cheap volatile read.
private void EnsureStarted()
{
+ // Stryker disable all: equivalent mutant — the CompareExchange below is the real guard, so
+ // whether this fast-path early-out (or its whole block) executes, a re-entrant caller that
+ // has already started changes nothing: the assignment only happens on the winning exchange.
if (Volatile.Read(ref _startTimestamp) != 0)
{
return;
}
+ // Stryker restore all
var now = DateTimeOffset.UtcNow;
var timestamp = Stopwatch.GetTimestamp();
@@ -461,6 +465,9 @@ public void Dispose()
/// when called from or
/// (dispose managed resources); when called from a finalizer.
///
+ // Stryker disable all: equivalent mutant — Dispose(bool) has an inert base body: _disposed has
+ // no other reader (nothing throws ObjectDisposedException). Removing the whole body, negating the
+ // guard, or dropping the assignment is all unobservable; derived overrides supply real behaviour.
protected virtual void Dispose(bool disposing)
{
if (_disposed)
@@ -470,4 +477,5 @@ protected virtual void Dispose(bool disposing)
_disposed = true;
}
+ // Stryker restore all
}
diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
index 2159a41b..0655245d 100644
--- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
+++ b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
@@ -407,10 +407,14 @@ protected void IncrementCurrentSkippedItemCount()
// win the CompareExchange records the start; later calls are a cheap volatile read.
private void EnsureStarted()
{
+ // Stryker disable all: equivalent mutant — the CompareExchange below is the real guard, so
+ // whether this fast-path early-out (or its whole block) executes, a re-entrant caller that
+ // has already started changes nothing: the assignment only happens on the winning exchange.
if (Volatile.Read(ref _startTimestamp) != 0)
{
return;
}
+ // Stryker restore all
var now = DateTimeOffset.UtcNow;
var timestamp = Stopwatch.GetTimestamp();
@@ -459,6 +463,9 @@ public void Dispose()
/// when called from or
/// (dispose managed resources); when called from a finalizer.
///
+ // Stryker disable all: equivalent mutant — Dispose(bool) has an inert base body: _disposed has
+ // no other reader (nothing throws ObjectDisposedException). Removing the whole body, negating the
+ // guard, or dropping the assignment is all unobservable; derived overrides supply real behaviour.
protected virtual void Dispose(bool disposing)
{
if (_disposed)
@@ -468,4 +475,5 @@ protected virtual void Dispose(bool disposing)
_disposed = true;
}
+ // Stryker restore all
}
diff --git a/src/Wolfgang.Etl.Abstractions/Report.cs b/src/Wolfgang.Etl.Abstractions/Report.cs
index 730d2afb..eec61425 100644
--- a/src/Wolfgang.Etl.Abstractions/Report.cs
+++ b/src/Wolfgang.Etl.Abstractions/Report.cs
@@ -150,6 +150,10 @@ public TimeSpan? EstimatedRemaining
// Guard against TimeSpan.FromSeconds overflowing for a pathologically low
// rate (e.g. a single item after a very long elapsed time); clamp to TimeSpan.MaxValue.
var seconds = remaining / rate;
+ // Stryker disable once Equality: equivalent mutant — >= versus > differs only when
+ // seconds exactly equals TimeSpan.MaxValue.TotalSeconds. That is a computed double from
+ // (int remaining / positive rate); no clean test input lands on that exact boundary, and
+ // both branches clamp identically for every reachable value.
return seconds >= TimeSpan.MaxValue.TotalSeconds
? TimeSpan.MaxValue
: TimeSpan.FromSeconds(seconds);
diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
index d7f00f83..d78cc6b7 100644
--- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
+++ b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
@@ -417,10 +417,14 @@ protected void IncrementCurrentSkippedItemCount()
// win the CompareExchange records the start; later calls are a cheap volatile read.
private void EnsureStarted()
{
+ // Stryker disable all: equivalent mutant — the CompareExchange below is the real guard, so
+ // whether this fast-path early-out (or its whole block) executes, a re-entrant caller that
+ // has already started changes nothing: the assignment only happens on the winning exchange.
if (Volatile.Read(ref _startTimestamp) != 0)
{
return;
}
+ // Stryker restore all
var now = DateTimeOffset.UtcNow;
var timestamp = Stopwatch.GetTimestamp();
@@ -468,6 +472,9 @@ public void Dispose()
/// when called from or
/// (dispose managed resources); when called from a finalizer.
///
+ // Stryker disable all: equivalent mutant — Dispose(bool) has an inert base body: _disposed has
+ // no other reader (nothing throws ObjectDisposedException). Removing the whole body, negating the
+ // guard, or dropping the assignment is all unobservable; derived overrides supply real behaviour.
protected virtual void Dispose(bool disposing)
{
if (_disposed)
@@ -477,4 +484,5 @@ protected virtual void Dispose(bool disposing)
_disposed = true;
}
+ // Stryker restore all
}
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ConfigureAwaitContextTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ConfigureAwaitContextTests.cs
new file mode 100644
index 00000000..a107caff
--- /dev/null
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ConfigureAwaitContextTests.cs
@@ -0,0 +1,237 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+using System.Threading;
+using System.Threading.Tasks;
+using Wolfgang.Etl.Abstractions.Tests.Unit.Models;
+
+namespace Wolfgang.Etl.Abstractions.Tests.Unit;
+
+///
+/// Verifies that the library never marshals a continuation back onto the caller's
+/// — i.e. every internal await uses
+/// ConfigureAwait(false). This is the behaviour that keeps the library safe from the classic
+/// sync-over-async deadlock and from needless context-hopping for WPF / WinForms / legacy-ASP.NET
+/// consumers, and it is invisible to a normal headless test (which has no context), so it needs an
+/// explicit context installed to exercise it.
+///
+/// Each test installs a counting context, starts the operation while it is current (so the
+/// awaits that suspend on that synchronous stack capture it under a ConfigureAwait(true)
+/// mutation), then awaits the result off-context and asserts nothing was posted back. The test
+/// doubles are all context-agnostic (their own awaits use ConfigureAwait(false)), so the only
+/// thing that can post to the context is the library code under test.
+///
+public sealed class ConfigureAwaitContextTests
+{
+ [Fact]
+ public Task EtlPipeline_streaming_run_does_not_capture_the_context()
+ => AssertDoesNotCaptureContext(() => EtlPipeline
+ .Create()
+ .From(SuspendingSource(3))
+ .To(new ContextAgnosticLoader())
+ .RunAsync());
+
+
+ [Fact]
+ public Task DisposingOwned_streaming_run_does_not_capture_the_context()
+ => AssertDoesNotCaptureContext(() => EtlPipeline
+ .Create()
+ .From(SuspendingSource(3))
+ .To(new ContextAgnosticLoader())
+ .DisposingOwned(new SuspendingAsyncDisposable())
+ .RunAsync());
+
+
+ [Fact]
+ public Task DisposingOwned_resource_cleanup_does_not_capture_the_context()
+ // Synchronous run keeps execution on the context thread so the suspending owned-resource
+ // disposal is the await that would capture it.
+ => AssertDoesNotCaptureContext(() => EtlPipeline
+ .Create()
+ .From(SynchronousSource(2))
+ .To(new ContextAgnosticLoader())
+ .DisposingOwned(new SuspendingAsyncDisposable())
+ .RunAsync());
+
+
+ [Fact]
+ public Task LoaderBase_progress_run_does_not_capture_the_context()
+ => AssertDoesNotCaptureContext(() => new ContextAgnosticLoader()
+ .LoadAsync(SuspendingSource(3), new NoOpProgress()));
+
+
+ [Fact]
+ public Task Fluent_pipeline_streaming_run_does_not_capture_the_context()
+ => AssertDoesNotCaptureContext(() => Pipeline
+ .Extract(new SuspendingExtractor(3))
+ .Load(new ContextAgnosticListLoader())
+ .DisposeStagesOnCompletion()
+ .RunAsync());
+
+
+ [Fact]
+ public Task Fluent_pipeline_stage_disposal_does_not_capture_the_context()
+ // Synchronous run + a suspending IAsyncDisposable stage → the disposal await is the one that
+ // would capture the context.
+ => AssertDoesNotCaptureContext(() => Pipeline
+ .Extract(new SuspendingAsyncDisposableExtractor(2))
+ .Load(new ContextAgnosticListLoader())
+ .DisposeStagesOnCompletion()
+ .RunAsync());
+
+
+ // Starts the operation while a counting context is current on this thread, restores the previous
+ // context, then awaits off-context and asserts the library posted nothing back. A ConfigureAwait
+ // flip to (true) on any await that suspended while the context was current posts a continuation
+ // to it and trips the assertion.
+ private static async Task AssertDoesNotCaptureContext(Func start)
+ {
+ var context = new CountingSynchronizationContext();
+ var previous = SynchronizationContext.Current;
+ SynchronizationContext.SetSynchronizationContext(context);
+
+ Task run;
+ try
+ {
+ run = start();
+ }
+ finally
+ {
+ SynchronizationContext.SetSynchronizationContext(previous);
+ }
+
+ await run.ConfigureAwait(false);
+
+ Assert.Equal(0, context.Posts);
+ }
+
+
+ // Suspends between items (so the library's awaits configure real continuations) but never
+ // captures the ambient context itself.
+ private static async IAsyncEnumerable SuspendingSource(int count)
+ {
+ for (var i = 0; i < count; i++)
+ {
+ await Task.Delay(1).ConfigureAwait(false);
+ yield return i;
+ }
+ }
+
+
+ // Completes synchronously — keeps the run on the context thread so a suspending disposal await
+ // is the one under test.
+ private static async IAsyncEnumerable SynchronousSource(int count)
+ {
+ for (var i = 0; i < count; i++)
+ {
+ yield return i;
+ }
+
+ await Task.CompletedTask;
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class CountingSynchronizationContext : SynchronizationContext
+ {
+ private int _posts;
+
+ public int Posts => Volatile.Read(ref _posts);
+
+ public override void Post(SendOrPostCallback d, object? state)
+ {
+ Interlocked.Increment(ref _posts);
+ ThreadPool.QueueUserWorkItem(_ => d(state));
+ }
+
+ public override void Send(SendOrPostCallback d, object? state)
+ {
+ Interlocked.Increment(ref _posts);
+ d(state);
+ }
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class NoOpProgress : IProgress
+ {
+ public void Report(T value)
+ {
+ }
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class ContextAgnosticLoader : LoaderBase
+ {
+ protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token)
+ {
+ await foreach (var item in items.WithCancellation(token).ConfigureAwait(false))
+ {
+ _ = item;
+ IncrementCurrentItemCount();
+ }
+ }
+
+ protected override EtlProgress CreateProgressReport() => new(CurrentItemCount);
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class ContextAgnosticListLoader : ILoadAsync
+ {
+ public async Task LoadAsync(IAsyncEnumerable items)
+ {
+ await foreach (var item in items.ConfigureAwait(false))
+ {
+ _ = item;
+ }
+ }
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class SuspendingExtractor : IExtractAsync
+ {
+ private readonly int _count;
+
+ public SuspendingExtractor(int count) => _count = count;
+
+ public async IAsyncEnumerable ExtractAsync()
+ {
+ for (var i = 0; i < _count; i++)
+ {
+ await Task.Delay(1).ConfigureAwait(false);
+ yield return i;
+ }
+ }
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class SuspendingAsyncDisposableExtractor : IExtractAsync, IAsyncDisposable
+ {
+ private readonly int _count;
+
+ public SuspendingAsyncDisposableExtractor(int count) => _count = count;
+
+ public async IAsyncEnumerable ExtractAsync()
+ {
+ for (var i = 0; i < _count; i++)
+ {
+ yield return i;
+ }
+
+ await Task.CompletedTask;
+ }
+
+ public async ValueTask DisposeAsync() => await Task.Delay(1).ConfigureAwait(false);
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class SuspendingAsyncDisposable : IAsyncDisposable
+ {
+ public async ValueTask DisposeAsync() => await Task.Delay(1).ConfigureAwait(false);
+ }
+}
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs
index 85ac759f..58cfebd4 100644
--- a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineTests.cs
@@ -239,6 +239,48 @@ public async Task RunAsync_observes_cancellation_mid_stream()
}
+ [Fact]
+ public async Task AsAsyncEnumerable_honours_cancellation_at_the_head()
+ {
+ // AsAsyncEnumerable exposes the pipeline head (CountExtracted) with no sink downstream, so
+ // its explicit ThrowIfCancellationRequested is the ONLY thing that can enforce cancellation
+ // when the raw source ignores the token — isolating that guard from the sink's paired one.
+ using var cts = new CancellationTokenSource();
+ cts.Cancel();
+
+ var stream = EtlPipeline
+ .Create()
+ .From(AsyncSource(1, 2, 3))
+ .AsAsyncEnumerable(cts.Token);
+
+ await Assert.ThrowsAnyAsync(async () =>
+ {
+ await foreach (var _ in stream)
+ {
+ }
+ });
+ }
+
+
+ [Fact]
+ public async Task RunAsync_honours_a_pre_cancelled_token_even_when_the_source_ignores_it()
+ {
+ // AsyncSource never observes the token, so the pipeline head's explicit
+ // ThrowIfCancellationRequested is the only thing that can enforce cancellation.
+ using var cts = new CancellationTokenSource();
+ cts.Cancel();
+
+ var loader = new CollectingLoader();
+ var sink = EtlPipeline
+ .Create()
+ .From(AsyncSource(1, 2, 3, 4, 5))
+ .To(loader);
+
+ await Assert.ThrowsAnyAsync(() => sink.RunAsync(null, cts.Token));
+ Assert.Empty(loader.Loaded);
+ }
+
+
[Fact]
public async Task RunAsync_reports_extracted_and_loaded_counters()
{
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/FinalizationSuppressionTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/FinalizationSuppressionTests.cs
new file mode 100644
index 00000000..76dca74f
--- /dev/null
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/FinalizationSuppressionTests.cs
@@ -0,0 +1,207 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Wolfgang.Etl.Abstractions.Tests.Unit.Models;
+
+namespace Wolfgang.Etl.Abstractions.Tests.Unit;
+
+///
+/// Verifies that disposing a base ETL component calls so a derived
+/// type that owns unmanaged resources via a finalizer is not finalized after an explicit dispose.
+/// The base types ship without a finalizer, so this is invisible unless a derived type declares one —
+/// which is exactly what these doubles do. A mutation that drops the SuppressFinalize call lets
+/// the finalizer run and trips the assertion.
+///
+public sealed class FinalizationSuppressionTests
+{
+ [Fact]
+ public void LoaderBase_Dispose_suppresses_finalization()
+ {
+ var finalizerRan = new StrongBox(false);
+ CreateAndDisposeLoader(finalizerRan);
+
+ ForceFinalization();
+
+ Assert.False(finalizerRan.Value, "Dispose() must call GC.SuppressFinalize so the finalizer never runs");
+ }
+
+
+ [Fact]
+ public async Task LoaderBase_DisposeAsync_suppresses_finalization()
+ {
+ var finalizerRan = new StrongBox(false);
+ await CreateAndDisposeAsyncLoader(finalizerRan);
+
+ ForceFinalization();
+
+ Assert.False(finalizerRan.Value, "DisposeAsync() must call GC.SuppressFinalize so the finalizer never runs");
+ }
+
+
+ [Fact]
+ public void ExtractorBase_Dispose_suppresses_finalization()
+ {
+ var finalizerRan = new StrongBox(false);
+ CreateAndDisposeExtractor(finalizerRan);
+
+ ForceFinalization();
+
+ Assert.False(finalizerRan.Value, "Dispose() must call GC.SuppressFinalize so the finalizer never runs");
+ }
+
+
+ [Fact]
+ public async Task ExtractorBase_DisposeAsync_suppresses_finalization()
+ {
+ var finalizerRan = new StrongBox(false);
+ await CreateAndDisposeAsyncExtractor(finalizerRan);
+
+ ForceFinalization();
+
+ Assert.False(finalizerRan.Value, "DisposeAsync() must call GC.SuppressFinalize so the finalizer never runs");
+ }
+
+
+ [Fact]
+ public void TransformerBase_Dispose_suppresses_finalization()
+ {
+ var finalizerRan = new StrongBox(false);
+ CreateAndDisposeTransformer(finalizerRan);
+
+ ForceFinalization();
+
+ Assert.False(finalizerRan.Value, "Dispose() must call GC.SuppressFinalize so the finalizer never runs");
+ }
+
+
+ [Fact]
+ public async Task TransformerBase_DisposeAsync_suppresses_finalization()
+ {
+ var finalizerRan = new StrongBox(false);
+ await CreateAndDisposeAsyncTransformer(finalizerRan);
+
+ ForceFinalization();
+
+ Assert.False(finalizerRan.Value, "DisposeAsync() must call GC.SuppressFinalize so the finalizer never runs");
+ }
+
+
+ // Separate non-inlined methods so the instance is unreachable (and therefore collectable) once
+ // they return — the disposed instance must not be rooted by the calling test frame.
+ [MethodImpl(MethodImplOptions.NoInlining)]
+ private static void CreateAndDisposeLoader(StrongBox finalizerRan)
+ {
+ var loader = new FinalizableLoader(finalizerRan);
+ loader.Dispose();
+ }
+
+
+ [MethodImpl(MethodImplOptions.NoInlining)]
+ private static async Task CreateAndDisposeAsyncLoader(StrongBox finalizerRan)
+ {
+ var loader = new FinalizableLoader(finalizerRan);
+ await loader.DisposeAsync();
+ }
+
+
+ [MethodImpl(MethodImplOptions.NoInlining)]
+ private static void CreateAndDisposeExtractor(StrongBox finalizerRan)
+ {
+ var extractor = new FinalizableExtractor(finalizerRan);
+ extractor.Dispose();
+ }
+
+
+ [MethodImpl(MethodImplOptions.NoInlining)]
+ private static async Task CreateAndDisposeAsyncExtractor(StrongBox finalizerRan)
+ {
+ var extractor = new FinalizableExtractor(finalizerRan);
+ await extractor.DisposeAsync();
+ }
+
+
+ [MethodImpl(MethodImplOptions.NoInlining)]
+ private static void CreateAndDisposeTransformer(StrongBox finalizerRan)
+ {
+ var transformer = new FinalizableTransformer(finalizerRan);
+ transformer.Dispose();
+ }
+
+
+ [MethodImpl(MethodImplOptions.NoInlining)]
+ private static async Task CreateAndDisposeAsyncTransformer(StrongBox finalizerRan)
+ {
+ var transformer = new FinalizableTransformer(finalizerRan);
+ await transformer.DisposeAsync();
+ }
+
+
+ private static void ForceFinalization()
+ {
+ GC.Collect();
+ GC.WaitForPendingFinalizers();
+ GC.Collect();
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class FinalizableLoader : LoaderBase
+ {
+ private readonly StrongBox _finalizerRan;
+
+ public FinalizableLoader(StrongBox finalizerRan) => _finalizerRan = finalizerRan;
+
+ ~FinalizableLoader() => _finalizerRan.Value = true;
+
+ protected override Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token)
+ => Task.CompletedTask;
+
+ protected override EtlProgress CreateProgressReport() => new(CurrentItemCount);
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class FinalizableExtractor : ExtractorBase
+ {
+ private readonly StrongBox _finalizerRan;
+
+ public FinalizableExtractor(StrongBox finalizerRan) => _finalizerRan = finalizerRan;
+
+ ~FinalizableExtractor() => _finalizerRan.Value = true;
+
+#pragma warning disable CS1998 // async iterator with no yielded items is intentional
+ protected override async IAsyncEnumerable ExtractWorkerAsync(
+ [EnumeratorCancellation] CancellationToken token)
+ {
+ yield break;
+ }
+#pragma warning restore CS1998
+
+ protected override EtlProgress CreateProgressReport() => new(CurrentItemCount);
+ }
+
+
+ [ExcludeFromCodeCoverage]
+ private sealed class FinalizableTransformer : TransformerBase
+ {
+ private readonly StrongBox _finalizerRan;
+
+ public FinalizableTransformer(StrongBox finalizerRan) => _finalizerRan = finalizerRan;
+
+ ~FinalizableTransformer() => _finalizerRan.Value = true;
+
+#pragma warning disable CS1998 // async iterator with no yielded items is intentional
+ protected override async IAsyncEnumerable TransformWorkerAsync(
+ IAsyncEnumerable items,
+ [EnumeratorCancellation] CancellationToken token)
+ {
+ yield break;
+ }
+#pragma warning restore CS1998
+
+ protected override EtlProgress CreateProgressReport() => new(CurrentItemCount);
+ }
+}
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/PipelineTests/DisposeStagesTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/PipelineTests/DisposeStagesTests.cs
index 045db873..c80cc363 100644
--- a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/PipelineTests/DisposeStagesTests.cs
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/PipelineTests/DisposeStagesTests.cs
@@ -94,6 +94,109 @@ public async Task LoadAsync(IAsyncEnumerable items)
}
+ private sealed class ThrowingDisposableExtractor : IExtractAsync, IDisposable
+ {
+ private readonly int _count;
+
+ public ThrowingDisposableExtractor(int count) => _count = count;
+
+ public async IAsyncEnumerable ExtractAsync()
+ {
+ for (var i = 0; i < _count; i++)
+ {
+ yield return i;
+ }
+
+ await Task.CompletedTask;
+ }
+
+ public void Dispose() => throw new InvalidOperationException("extractor dispose failed");
+ }
+
+
+ private sealed class TrackingCancellableExtractor : IExtractWithCancellationAsync, IDisposable
+ {
+ private readonly int _count;
+
+ public TrackingCancellableExtractor(int count) => _count = count;
+
+ public bool Disposed { get; private set; }
+
+ public IAsyncEnumerable ExtractAsync() => Produce();
+
+ public IAsyncEnumerable ExtractAsync(CancellationToken token) => Produce();
+
+ private async IAsyncEnumerable Produce()
+ {
+ for (var i = 0; i < _count; i++)
+ {
+ yield return i;
+ }
+
+ await Task.CompletedTask;
+ }
+
+ public void Dispose() => Disposed = true;
+ }
+
+
+ private sealed class TrackingProgressExtractor : IExtractWithProgressAsync, IDisposable
+ {
+ private readonly int _count;
+
+ public TrackingProgressExtractor(int count) => _count = count;
+
+ public bool Disposed { get; private set; }
+
+ public IAsyncEnumerable ExtractAsync() => Produce();
+
+ public IAsyncEnumerable ExtractAsync(IProgress progress) => Produce();
+
+ private async IAsyncEnumerable Produce()
+ {
+ for (var i = 0; i < _count; i++)
+ {
+ yield return i;
+ }
+
+ await Task.CompletedTask;
+ }
+
+ public void Dispose() => Disposed = true;
+ }
+
+
+ private sealed class TrackingProgressCancellableExtractor
+ : IExtractWithProgressAndCancellationAsync, IDisposable
+ {
+ private readonly int _count;
+
+ public TrackingProgressCancellableExtractor(int count) => _count = count;
+
+ public bool Disposed { get; private set; }
+
+ public IAsyncEnumerable ExtractAsync() => Produce();
+
+ public IAsyncEnumerable ExtractAsync(CancellationToken token) => Produce();
+
+ public IAsyncEnumerable ExtractAsync(IProgress progress) => Produce();
+
+ public IAsyncEnumerable ExtractAsync(IProgress progress, CancellationToken token) => Produce();
+
+ private async IAsyncEnumerable Produce()
+ {
+ for (var i = 0; i < _count; i++)
+ {
+ yield return i;
+ }
+
+ await Task.CompletedTask;
+ }
+
+ public void Dispose() => Disposed = true;
+ }
+
+
// Not disposable — must be skipped without error.
private sealed class PlainTransformer : ITransformAsync
{
@@ -238,6 +341,72 @@ public async Task DisposeStagesOnCompletion_aggregates_disposal_exceptions()
}
+ [Fact]
+ public async Task DisposeStagesOnCompletion_disposes_a_cancellation_extractor()
+ {
+ // Covers the Extract(IExtractWithCancellationAsync) factory — a distinct stages array
+ // from the plain IExtractAsync overload.
+ var extractor = new TrackingCancellableExtractor(2);
+
+ await Pipeline
+ .Extract(extractor)
+ .Load(new TrackingLoader())
+ .DisposeStagesOnCompletion()
+ .RunAsync();
+
+ Assert.True(extractor.Disposed);
+ }
+
+
+ [Fact]
+ public async Task DisposeStagesOnCompletion_disposes_a_progress_extractor()
+ {
+ // Covers the Extract(IExtractWithProgressAsync) factory's stages array.
+ var extractor = new TrackingProgressExtractor(2);
+
+ await Pipeline
+ .Extract(extractor)
+ .Load(new TrackingLoader())
+ .DisposeStagesOnCompletion()
+ .RunAsync();
+
+ Assert.True(extractor.Disposed);
+ }
+
+
+ [Fact]
+ public async Task DisposeStagesOnCompletion_disposes_a_progress_and_cancellation_extractor()
+ {
+ // Covers the Extract(IExtractWithProgressAndCancellationAsync) factory's stages array.
+ var extractor = new TrackingProgressCancellableExtractor(2);
+
+ await Pipeline
+ .Extract(extractor)
+ .Load(new TrackingLoader())
+ .DisposeStagesOnCompletion()
+ .RunAsync();
+
+ Assert.True(extractor.Disposed);
+ }
+
+
+ [Fact]
+ public async Task DisposeStagesOnCompletion_aggregates_every_stage_disposal_failure()
+ {
+ // Two stages throw on disposal — both errors must accumulate into the aggregate,
+ // proving the error list is appended to (errors ??= ...) rather than reset per stage.
+ var ex = await Assert.ThrowsAsync(() => Pipeline
+ .Extract(new ThrowingDisposableExtractor(2))
+ .Load(new ThrowingDisposableLoader())
+ .DisposeStagesOnCompletion()
+ .RunAsync());
+
+ Assert.Equal(2, ex.InnerExceptions.Count);
+ Assert.Contains(ex.InnerExceptions, e => string.Equals(e.Message, "extractor dispose failed", StringComparison.Ordinal));
+ Assert.Contains(ex.InnerExceptions, e => string.Equals(e.Message, "dispose failed", StringComparison.Ordinal));
+ }
+
+
// Records the order in which stages were disposed by appending to a shared list.
private sealed class OrderRecordingExtractor : IExtractAsync, IDisposable
{