Aggregate per-item error counts across all pipeline stages (#335)

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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
@@ -24,6 +27,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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.
 - **Breaking (#285):** `EtlPipelineProgress`'s counters — `ExtractedItemCount`, `LoadedItemCount`, and
   `ErrorItemCount` — are now `long` instead of `int`, so a long-running pipeline can report more than
   `int.MaxValue` (~2.1 billion) records without overflow. This changes the record's getters, positional

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;
 using System.Diagnostics;
 
 
@@ -18,16 +19,27 @@ 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 EtlPipelineProgress Snapshot()
     {
+        long errorItemCount = 0;
+        foreach (var reader in _errorCountReaders)
+        {
+            errorItemCount += reader();
+        }
+
         return new EtlPipelineProgress(ExtractedItemCount, LoadedItemCount, _stopwatch.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

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

src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt

@@ -20,3 +20,5 @@ static Wolfgang.Etl.Abstractions.MiddlewareResult<T>.operator !=(Wolfgang.Etl.Ab
 Wolfgang.Etl.Abstractions.MiddlewareExtensions
 static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware<T>(this System.Collections.Generic.IAsyncEnumerable<T>! source, Wolfgang.Etl.Abstractions.IItemMiddleware<T>! middleware, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<T>!
 static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware<T>(this System.Collections.Generic.IAsyncEnumerable<T>! source, System.Collections.Generic.IEnumerable<Wolfgang.Etl.Abstractions.IItemMiddleware<T>!>! middlewares, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<T>!
+Wolfgang.Etl.Abstractions.IReportsItemErrors
+Wolfgang.Etl.Abstractions.IReportsItemErrors.CurrentErrorItemCount.get -> int

src/Wolfgang.Etl.Abstractions/TransformerBase.cs

@@ -20,6 +20,7 @@ namespace Wolfgang.Etl.Abstractions;
 /// <typeparam name="TProgress">The type of the progress object</typeparam>
 public abstract class TransformerBase<TSource, TDestination, TProgress>
     : ITransformWithProgressAndCancellationAsync<TSource, TDestination, TProgress>,
+      IReportsItemErrors,
       IAsyncDisposable,
       IDisposable
     where TSource : notnull