Release v0.20.0 — retry seam, middleware, error aggregation

CHANGELOG.md

@@ -19,6 +19,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+## [0.20.0] - 2026-07-30
+
+Minor release: a dependency-free retry seam, composable per-item middleware, and pipeline-wide
+error aggregation on the base stages. Purely additive — no breaking change (validates against the
+0.19.0 baseline).
+
+### Added
+
+- **`IReportsItemErrors` (#335):** a small interface (`int CurrentErrorItemCount { get; }`) implemented
+  by `ExtractorBase`, `LoaderBase`, and `TransformerBase`, letting a pipeline read any stage's
+  error-item count uniformly regardless of concrete type.
+- **Middleware / interceptor (#93):** a composable per-item hook — `IItemMiddleware<T>` returning
+  `MiddlewareResult<T>` (`Continue` to keep/replace an item, `Drop` to remove it) — attached to any
+  stream with the `WithMiddleware(...)` extensions (single or ordered chain). Lets cross-cutting
+  concerns (logging, validation, metrics, throttling, dedup) decorate an extractor/transformer output
+  or loader input without changing the component, and composes inside an `EtlPipeline` via
+  `Through(s => s.WithMiddleware(...))`. Dependency-free.
+- **Retry seam (#94):** `ExtractorBase`, `LoaderBase`, and `TransformerBase` gained a
+  `protected virtual WrapWorkerExecution(...)` hook wrapped around every worker invocation. The
+  default implementation is a no-op, so behaviour is unchanged; override it to run the worker through
+  a retry / resilience strategy. The override receives a re-invocable worker factory (call it again to
+  retry) and stream-level semantics are documented on the method. Kept dependency-free — a ready-made
+  Polly integration will ship as a separate opt-in `Wolfgang.Etl.Polly` package (#332).
+
+### Changed
+
+- **`EtlPipelineProgress.ErrorItemCount` now aggregates every stage (#335).** Previously it reported
+  only the extractor's error-item count; it now sums the error-item counts of the source, every
+  transformer, and the loader (each stage that implements `IReportsItemErrors`), so an item any
+  stage's error policy discarded is reflected in the total. Pre-1.0 behaviour change.
+- _Internal (no public API change):_ the base classes' `StartedAt` / `Elapsed` timing now reads
+  through an injectable time source (#338), so downstream test kits can drive the `Report`
+  throughput / ETA metrics from a fake clock (unblocks ETL-Test-Kit#262).
+
 ## [0.19.0] - 2026-07-29
 
 Minor release: makes the `EtlPipelineProgress` record counters overflow-safe.

README.md

@@ -156,6 +156,9 @@ await Pipeline
 | Throughput & ETA | `Report` exposes `StartedAt`, `Elapsed`, `ItemsPerSecond`, `PercentComplete`, and `EstimatedRemaining` derived from the item count and elapsed time |
 | Resource Disposal | Base classes implement `IDisposable` / `IAsyncDisposable`; override `Dispose(bool)` or `DisposeAsync()` to release resources deterministically |
 | Per-run State | Item counts and timing reset at the start of each enumeration, so a reused component reports the current run rather than cumulative totals |
+| Per-item Error Handling | Opt-in `OnItemError` policy (`Abort` / `Skip`) on each base stage with a distinct `CurrentErrorItemCount`; the pipeline sums error counts across every stage (source, transformers, sink) into `EtlPipelineProgress.ErrorItemCount` |
+| Middleware / Interceptors | Attach composable per-item behaviour (logging, validation, metrics, throttling, dedup) to any stream with `WithMiddleware(...)` — an `IItemMiddleware<T>` returning `MiddlewareResult<T>` (`Continue` / `Drop`) |
+| Retry Seam | `WrapWorkerExecution` hook on each base stage to run the worker through a retry / resilience strategy (for example Polly); default no-op, dependency-free |
 | Cancellation | Full `CancellationToken` support across all operations |
 | Multi-TFM | Targets .NET Framework 4.6.2–4.8.1, .NET Standard 2.0, and .NET 5.0–10.0 |
 | Skip & Limit | `SkipItemCount` and `MaximumItemCount` for partial extraction/loading |

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs

@@ -46,7 +46,11 @@ public IEtlPipeline<TOut> Through<TOut>(ITransformAsync<T, TOut> transformer)
             throw new ArgumentNullException(nameof(transformer));
         }
 
-        return new EtlPipelineImpl<TOut>((state, token) => transformer.TransformAsync(_factory(state, token)));
+        return new EtlPipelineImpl<TOut>((state, token) =>
+        {
+            RegisterErrorReader(transformer, state);
+            return transformer.TransformAsync(_factory(state, token));
+        });
     }
 
 
@@ -59,7 +63,11 @@ public IEtlPipeline<TOut> Through<TOut>(ITransformWithCancellationAsync<T, TOut>
             throw new ArgumentNullException(nameof(transformer));
         }
 
-        return new EtlPipelineImpl<TOut>((state, token) => transformer.TransformAsync(_factory(state, token), token));
+        return new EtlPipelineImpl<TOut>((state, token) =>
+        {
+            RegisterErrorReader(transformer, state);
+            return transformer.TransformAsync(_factory(state, token), token);
+        });
     }
 
 
@@ -109,6 +117,18 @@ public IAsyncEnumerable<T> AsAsyncEnumerable(CancellationToken token = default)
     }
 
 
+    // Registers a transformer stage's error-item count with the run so
+    // EtlPipelineProgress.ErrorItemCount sums it alongside the source and sink. A transformer that
+    // does not report errors (for example a plain delegate transform) is skipped.
+    private static void RegisterErrorReader(object transformer, EtlRunState state)
+    {
+        if (transformer is IReportsItemErrors reporter)
+        {
+            state.AddErrorCountReader(() => reporter.CurrentErrorItemCount);
+        }
+    }
+
+
     // The head of every pipeline: pulls from the raw source, honours cancellation, and counts each
     // record as extracted. WithCancellation covers sources that observe the token via
     // [EnumeratorCancellation]; the explicit throw covers sources that ignore it.

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs

@@ -35,6 +35,11 @@ LoaderBase<T, TProgress> loader
     public async Task RunAsync(IProgress<EtlPipelineProgress>? progress = null, CancellationToken token = default)
     {
         var state = new EtlRunState();
+
+        // Surface the loader's error-item count into the snapshot too, summed with the source and any
+        // transformer error counts so the pipeline total reflects every stage, not just the source.
+        state.AddErrorCountReader(() => _loader.CurrentErrorItemCount);
+
         var stream = CountLoaded(_factory(state, token), state, progress, token);
 
         await _loader.LoadAsync(stream, token).ConfigureAwait(false);

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs

@@ -56,7 +56,8 @@ public static IEtlPipeline<T> From<T, TProgress>(this EtlPipeline pipeline, Extr
         {
             // Surface the extractor's error-item count into the pipeline snapshot so a bad record
             // the extractor's error policy discarded is reported, not silently absent from the totals.
-            state.ErrorCountReader = () => extractor.CurrentErrorItemCount;
+            // Summed with any transformer/loader error counts (see EtlRunState.AddErrorCountReader).
+            state.AddErrorCountReader(() => extractor.CurrentErrorItemCount);
             return extractor.ExtractAsync(token);
         });
     }

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 
 
 namespace Wolfgang.Etl.Abstractions;
@@ -18,9 +19,14 @@ internal sealed class EtlRunState
 
     public long LoadedItemCount;
 
-    // Optional reader that surfaces an error-reporting source's error-item count into the snapshot.
-    // Left null for sources that don't report errors (e.g. a raw IAsyncEnumerable), which reads as 0.
-    public Func<int>? ErrorCountReader;
+    // Error-item-count readers, one per stage that reports errors (source, transformers, sink). Their
+    // values are summed into the snapshot so an item any stage's error policy discarded is reported,
+    // not just the source's. Empty for a pipeline of stages that don't report errors, which reads as 0.
+    private readonly List<Func<int>> _errorCountReaders = new();
+
+
+    // Registers a stage's error-item-count reader. Called once per stage as the factory chain runs.
+    public void AddErrorCountReader(Func<int> reader) => _errorCountReaders.Add(reader);
 
 
     public EtlRunState()
@@ -49,9 +55,15 @@ private TimeSpan Elapsed
 
     public EtlPipelineProgress Snapshot()
     {
+        long errorItemCount = 0;
+        foreach (var reader in _errorCountReaders)
+        {
+            errorItemCount += reader();
+        }
+
         return new EtlPipelineProgress(ExtractedItemCount, LoadedItemCount, Elapsed)
         {
-            ErrorItemCount = ErrorCountReader?.Invoke() ?? 0,
+            ErrorItemCount = errorItemCount,
         };
     }
 }

src/Wolfgang.Etl.Abstractions/ExtractorBase.cs

@@ -19,6 +19,7 @@ namespace Wolfgang.Etl.Abstractions;
 /// <typeparam name="TProgress">The type of the progress object</typeparam>
 public abstract class ExtractorBase<TSource, TProgress>
     : IExtractWithProgressAndCancellationAsync<TSource, TProgress>,
+      IReportsItemErrors,
       IAsyncDisposable,
       IDisposable
     where TSource : notnull
@@ -322,7 +323,7 @@ [EnumeratorCancellation] CancellationToken token
     {
         ResetRunState();
 
-        await foreach (var item in ExtractWorkerAsync(token))
+        await foreach (var item in WrapWorkerExecution(ExtractWorkerAsync, token))
         {
             yield return item;
         }
@@ -340,7 +341,7 @@ private async IAsyncEnumerable<TSource> ExtractWithProgressAsync(
 
         try
         {
-            await foreach (var item in ExtractWorkerAsync(token))
+            await foreach (var item in WrapWorkerExecution(ExtractWorkerAsync, token))
             {
                 yield return item;
             }
@@ -382,6 +383,43 @@ private void ResetRunState()
 
 
 
+    /// <summary>
+    /// A resilience seam wrapped around every invocation of <see cref="ExtractWorkerAsync"/>. The
+    /// default implementation simply invokes <paramref name="workerFactory"/> once, so extraction
+    /// behaves exactly as if the seam were absent. Override it to run the worker through a retry /
+    /// resilience strategy (for example a Polly <c>ResiliencePipeline</c>): the strategy can invoke
+    /// <paramref name="workerFactory"/> more than once, each call producing a fresh stream, to retry
+    /// a transient failure.
+    /// </summary>
+    /// <remarks>
+    /// This is <b>stream-level</b> resilience: a retry re-runs the whole worker from the start, so a
+    /// failure part-way through re-yields items already seen. The per-run counters
+    /// (<see cref="CurrentItemCount"/>, <see cref="CurrentSkippedItemCount"/>,
+    /// <see cref="CurrentErrorItemCount"/>) are reset once at the start of the run, <b>not</b> on each
+    /// retry, so they accumulate across attempts unless the override resets them. Any delay the
+    /// override introduces must observe <paramref name="token"/>. Kept dependency-free by design — a
+    /// concrete Polly integration lives in a separate opt-in package rather than in this library.
+    /// </remarks>
+    /// <param name="workerFactory">A factory that produces a fresh worker stream for the supplied token. Re-invocable — call it again to retry.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe, including during any retry delay.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="workerFactory"/> is <see langword="null"/>.</exception>
+    /// <returns>The (possibly resilience-wrapped) stream of extracted items.</returns>
+    protected virtual IAsyncEnumerable<TSource> WrapWorkerExecution
+    (
+        Func<CancellationToken, IAsyncEnumerable<TSource>> workerFactory,
+        CancellationToken token
+    )
+    {
+        if (workerFactory is null)
+        {
+            throw new ArgumentNullException(nameof(workerFactory));
+        }
+
+        return workerFactory(token);
+    }
+
+
+
     /// <summary>
     /// Creates a progress report of type TProgress. This gives the derived class the opportunity to
     /// implement a custom progress report that is specific to the extraction process.

src/Wolfgang.Etl.Abstractions/IItemMiddleware.cs

@@ -0,0 +1,25 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// A composable, reusable hook for cross-cutting per-item behaviour (logging, validation, metrics,
+/// throttling, deduplication) that can be attached to any stream without modifying the extractor,
+/// transformer, or loader that produced it. Attach one or more with
+/// <see cref="MiddlewareExtensions.WithMiddleware{T}(System.Collections.Generic.IAsyncEnumerable{T}, IItemMiddleware{T}, CancellationToken)"/>;
+/// they run in the order attached, each seeing the item the previous one passed on.
+/// </summary>
+/// <typeparam name="T">The item type flowing through the pipeline.</typeparam>
+public interface IItemMiddleware<T>
+{
+    /// <summary>
+    /// Invoked once per item. Return <see cref="MiddlewareResult.Continue{T}(T)"/> to keep the item
+    /// flowing (optionally replacing it), or <see cref="MiddlewareResult.Drop{T}"/> to remove it from
+    /// the stream.
+    /// </summary>
+    /// <param name="item">The item to process.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The outcome describing whether to keep or drop the item.</returns>
+    ValueTask<MiddlewareResult<T>> OnItemAsync(T item, CancellationToken token);
+}

src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs

@@ -0,0 +1,17 @@
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Implemented by an ETL stage that counts items its error policy discarded, so a pipeline can read
+/// that count uniformly regardless of the stage's concrete type. <see cref="ExtractorBase{TSource, TProgress}"/>,
+/// <see cref="LoaderBase{TDestination, TProgress}"/>, and <see cref="TransformerBase{TSource, TDestination, TProgress}"/>
+/// all implement it, and <c>EtlPipeline</c> sums <see cref="CurrentErrorItemCount"/> across every stage
+/// that reports it into <see cref="EtlPipelineProgress.ErrorItemCount"/>.
+/// </summary>
+public interface IReportsItemErrors
+{
+    /// <summary>
+    /// The number of items this stage's error policy has discarded (<c>OnItemError</c> returned
+    /// <see cref="ItemErrorAction.Skip"/>) so far in the current run.
+    /// </summary>
+    int CurrentErrorItemCount { get; }
+}

src/Wolfgang.Etl.Abstractions/LoaderBase.cs

@@ -18,6 +18,7 @@ namespace Wolfgang.Etl.Abstractions;
 /// <typeparam name="TProgress">The type of the progress object</typeparam>
 public abstract class LoaderBase<TDestination, TProgress>
     : ILoadWithProgressAndCancellationAsync<TDestination, TProgress>,
+      IReportsItemErrors,
       IAsyncDisposable,
       IDisposable
     where TDestination : notnull
@@ -323,7 +324,7 @@ CancellationToken token
     )
     {
         ResetRunState();
-        return LoadWorkerAsync(items, token);
+        return WrapWorkerExecution(ct => LoadWorkerAsync(items, ct), token);
     }
 
 
@@ -339,7 +340,7 @@ private async Task LoadWithProgressAsync(
 
         try
         {
-            await LoadWorkerAsync(items, token).ConfigureAwait(false);
+            await WrapWorkerExecution(ct => LoadWorkerAsync(items, ct), token).ConfigureAwait(false);
         }
         finally
         {
@@ -380,6 +381,43 @@ private void ResetRunState()
 
 
 
+    /// <summary>
+    /// A resilience seam wrapped around every invocation of <see cref="LoadWorkerAsync"/>. The
+    /// default implementation simply invokes <paramref name="workerFactory"/> once, so loading
+    /// behaves exactly as if the seam were absent. Override it to run the worker through a retry /
+    /// resilience strategy (for example a Polly <c>ResiliencePipeline</c>): the strategy can invoke
+    /// <paramref name="workerFactory"/> more than once to retry a transient failure.
+    /// </summary>
+    /// <remarks>
+    /// This is <b>stream-level</b> resilience: a retry re-runs the whole worker, which re-enumerates
+    /// the source <c>items</c> from the start — so retry is only safe when that source can be
+    /// enumerated more than once. The per-run counters (<see cref="CurrentItemCount"/>,
+    /// <see cref="CurrentSkippedItemCount"/>, <see cref="CurrentErrorItemCount"/>) are reset once at
+    /// the start of the run, <b>not</b> on each retry, so they accumulate across attempts unless the
+    /// override resets them. Any delay the override introduces must observe <paramref name="token"/>.
+    /// Kept dependency-free by design — a concrete Polly integration lives in a separate opt-in
+    /// package rather than in this library.
+    /// </remarks>
+    /// <param name="workerFactory">A factory that runs the worker for the supplied token. Re-invocable — call it again to retry.</param>
+    /// <param name="token">A <see cref="CancellationToken"/> to observe, including during any retry delay.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="workerFactory"/> is <see langword="null"/>.</exception>
+    /// <returns>A task representing the (possibly resilience-wrapped) load operation.</returns>
+    protected virtual Task WrapWorkerExecution
+    (
+        Func<CancellationToken, Task> workerFactory,
+        CancellationToken token
+    )
+    {
+        if (workerFactory is null)
+        {
+            throw new ArgumentNullException(nameof(workerFactory));
+        }
+
+        return workerFactory(token);
+    }
+
+
+
     /// <summary>
     /// Creates a progress report of type TProgress. This gives the derived class the opportunity to
     /// implement a custom progress report that is specific to the loading process.