release: Wolfgang.Etl.Abstractions 0.18.0 — per-item error handling (#84) [HOLD until 2026-07-25]

CHANGELOG.md

@@ -19,6 +19,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+## [0.18.0] - 2026-07-25
+
+Minor release: adds an **opt-in per-item error-handling mechanism** (#84) to the
+three base stages, so a worker can skip a bad item and keep going instead of
+aborting the whole run. Purely additive — no existing signature changed, so
+Package Validation passes against 0.17.0 and `AssemblyVersion` remains `1.0.0.0`.
+
+### Added
+
+- `ItemErrorAction` (`Abort`, `Skip`) and `ItemErrorContext` (record number,
+  exception, and an optional lazy raw-content accessor) describing a failed item.
+- `ExtractorBase`, `LoaderBase`, and `TransformerBase` each gain a protected
+  `HandleItemError(ItemErrorContext)` helper — call it from a worker's `catch`
+  block and re-throw when it returns `Abort` — plus a `virtual OnItemError`
+  policy hook (default `Abort`) that a derived stage overrides to surface its own
+  error-handling knob.
+- `CurrentErrorItemCount` on each base stage, counting items discarded by an
+  error-`Skip`. It is kept distinct from `CurrentSkippedItemCount` (intentional
+  skip-budget skips) so a failure is never silently absorbed into the skip total.
+- `EtlPipelineProgress.RecordsErrored`, surfacing an extractor's error-item count
+  in the pipeline progress snapshot.
+
 ## [0.17.0] - 2026-07-24
 
 Minor release: a new **use-after-dispose contract** plus a substantial testing-depth

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

@@ -24,4 +24,16 @@ public sealed record EtlPipelineProgress
     int RecordsExtracted,
     int RecordsLoaded,
     TimeSpan Elapsed
-);
+)
+{
+    /// <summary>
+    /// The number of records a source discarded via its error policy so far — an
+    /// <see cref="ExtractorBase{TSource, TProgress}"/> whose <c>OnItemError</c> returned
+    /// <see cref="ItemErrorAction.Skip"/> for a bad record — or <c>0</c> for a source that does not
+    /// report errors (for example a raw <see cref="System.Collections.Generic.IAsyncEnumerable{T}"/>).
+    /// Reported separately from <see cref="RecordsExtracted"/>, which counts only the records that
+    /// successfully flowed into the pipeline, so a failed record is never silent. Distinct from
+    /// intentional skips (an extractor's <c>SkipItemCount</c> budget), which are not surfaced here.
+    /// </summary>
+    public int RecordsErrored { get; init; }
+}

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

@@ -52,6 +52,12 @@ public static IEtlPipeline<T> From<T, TProgress>(this EtlPipeline pipeline, Extr
             throw new ArgumentNullException(nameof(extractor));
         }
 
-        return EtlPipelineImpl<T>.FromStream((_, token) => extractor.ExtractAsync(token));
+        return EtlPipelineImpl<T>.FromStream((state, token) =>
+        {
+            // 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;
+            return extractor.ExtractAsync(token);
+        });
     }
 }

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

@@ -1,3 +1,4 @@
+using System;
 using System.Diagnostics;
 
 
@@ -17,9 +18,16 @@ internal sealed class EtlRunState
 
     public int RecordsLoaded;
 
+    // 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;
+
 
     public EtlPipelineProgress Snapshot()
     {
-        return new EtlPipelineProgress(RecordsExtracted, RecordsLoaded, _stopwatch.Elapsed);
+        return new EtlPipelineProgress(RecordsExtracted, RecordsLoaded, _stopwatch.Elapsed)
+        {
+            RecordsErrored = ErrorCountReader?.Invoke() ?? 0,
+        };
     }
 }

src/Wolfgang.Etl.Abstractions/ExtractorBase.cs

@@ -26,6 +26,7 @@ public abstract class ExtractorBase<TSource, TProgress>
 {
     private int _currentItemCount;
     private int _currentSkippedItemCount;
+    private int _currentErrorItemCount;
     private long _startTimestamp;
     private DateTimeOffset _startedAtUtc;
     private bool _disposed;
@@ -114,6 +115,16 @@ public int ReportingInterval
 
 
 
+    /// <summary>
+    /// The number of items that raised an error and were discarded by the stage's error policy
+    /// (<see cref="OnItemError"/> returned <see cref="ItemErrorAction.Skip"/>) so far. This is distinct
+    /// from <see cref="CurrentSkippedItemCount"/>, which counts items skipped intentionally (for
+    /// example by <c>SkipItemCount</c>): a failed record is counted here, never silently dropped.
+    /// </summary>
+    public int CurrentErrorItemCount => Volatile.Read(ref _currentErrorItemCount);
+
+
+
     /// <summary>
     /// The maximum number of items to extract. Once the extractor has reached this limit,
     /// it should stop extracting and signal the end of the sequence.
@@ -334,6 +345,7 @@ private void ResetRunState()
     {
         Volatile.Write(ref _currentItemCount, 0);
         Volatile.Write(ref _currentSkippedItemCount, 0);
+        Volatile.Write(ref _currentErrorItemCount, 0);
         Volatile.Write(ref _startTimestamp, 0L);
     }
 
@@ -408,6 +420,65 @@ protected void IncrementCurrentSkippedItemCount()
 
 
 
+    /// <summary>
+    /// Decides what to do when an item fails to process. Override in a derived stage to record the
+    /// failure and return <see cref="ItemErrorAction.Skip"/> to discard the item and continue, or
+    /// <see cref="ItemErrorAction.Abort"/> to re-throw and stop the run. The base implementation
+    /// always returns <see cref="ItemErrorAction.Abort"/>, so a stage that does not opt in keeps its
+    /// fail-fast behaviour.
+    /// </summary>
+    /// <param name="context">
+    /// Describes the failed item — its ordinal, the exception, and optional raw content.
+    /// </param>
+    /// <returns>Whether to skip the item or abort the run.</returns>
+    /// <remarks>
+    /// This is the policy hook a derived stage overrides; a worker does not call it directly. A worker
+    /// calls <see cref="HandleItemError"/>, which invokes this method and performs the skip
+    /// bookkeeping. The base classes deliberately expose no public error-handling property: a base
+    /// class cannot catch a per-item failure on the worker's behalf — a C# async iterator cannot
+    /// resume after it throws — so the worker owns the <c>try</c>/<c>catch</c>, and only a format that
+    /// can genuinely resume after a bad record overrides this and surfaces its own public knob.
+    /// </remarks>
+    protected virtual ItemErrorAction OnItemError(ItemErrorContext context)
+    // Stryker disable once all: equivalent — Abort is the enum's default (0), so removing the body
+    // (which makes it return default) yields the identical value; no test can distinguish them.
+    {
+        return ItemErrorAction.Abort;
+    }
+
+
+
+    /// <summary>
+    /// Applies the stage's error policy to a failed item: invokes <see cref="OnItemError"/> and, when
+    /// it returns <see cref="ItemErrorAction.Skip"/>, increments the error-item count so the failure is
+    /// never silent. Call this from a worker's <c>catch</c> block and re-throw when it returns
+    /// <see cref="ItemErrorAction.Abort"/>.
+    /// </summary>
+    /// <param name="context">Describes the failed item.</param>
+    /// <returns>
+    /// <see cref="ItemErrorAction.Skip"/> to discard the item and continue, or
+    /// <see cref="ItemErrorAction.Abort"/> to re-throw.
+    /// </returns>
+    /// <exception cref="ArgumentNullException"><paramref name="context"/> is <see langword="null"/>.</exception>
+    protected ItemErrorAction HandleItemError(ItemErrorContext context)
+    {
+        if (context is null)
+        {
+            throw new ArgumentNullException(nameof(context));
+        }
+
+        var action = OnItemError(context);
+        if (action == ItemErrorAction.Skip)
+        {
+            EnsureStarted();
+            _ = Interlocked.Increment(ref _currentErrorItemCount);
+        }
+
+        return action;
+    }
+
+
+
     // Captures the start timestamp (monotonic) and wall-clock StartedAt the first
     // time any item is processed. Idempotent and thread-safe: the first caller to
     // win the CompareExchange records the start; later calls are a cheap volatile read.

src/Wolfgang.Etl.Abstractions/ItemErrorAction.cs

@@ -0,0 +1,20 @@
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// The action a stage takes for an item that failed to process — returned by a stage's
+/// <c>OnItemError</c> policy and applied by the worker that called <c>HandleItemError</c>.
+/// </summary>
+public enum ItemErrorAction
+{
+    /// <summary>
+    /// Re-throw the failure and stop the run. This is the default when no error policy is
+    /// configured, preserving the fail-fast behaviour of a stage that does not opt in.
+    /// </summary>
+    Abort,
+
+    /// <summary>
+    /// Discard the failed item and continue with the next one. The stage's skipped-item count is
+    /// incremented so the skip is never silent.
+    /// </summary>
+    Skip,
+}

src/Wolfgang.Etl.Abstractions/ItemErrorContext.cs

@@ -0,0 +1,45 @@
+using System;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Describes a single item that failed to process. It is passed to a stage's <c>OnItemError</c>
+/// policy so the policy can record the failure (a "dead letter") and decide whether to
+/// <see cref="ItemErrorAction.Skip"/> the item or <see cref="ItemErrorAction.Abort"/> the run.
+/// </summary>
+public sealed class ItemErrorContext
+{
+    /// <summary>
+    /// Initialises a new <see cref="ItemErrorContext"/>.
+    /// </summary>
+    /// <param name="recordNumber">The 1-based ordinal of the failed item within the current run.</param>
+    /// <param name="exception">The exception the item raised.</param>
+    /// <param name="rawContent">
+    /// An optional, lazily-evaluated accessor for the item's raw source text (a line, a fixed-width
+    /// record, an XML element). It is a delegate rather than a value so a stage that would have to
+    /// buffer or reconstruct the raw content pays that cost only if the policy actually reads it.
+    /// </param>
+    /// <exception cref="ArgumentNullException"><paramref name="exception"/> is <see langword="null"/>.</exception>
+    public ItemErrorContext(long recordNumber, Exception exception, Func<string?>? rawContent = null)
+    {
+        RecordNumber = recordNumber;
+        Exception = exception ?? throw new ArgumentNullException(nameof(exception));
+        RawContent = rawContent;
+    }
+
+    /// <summary>
+    /// The 1-based ordinal of the failed item within the current run.
+    /// </summary>
+    public long RecordNumber { get; }
+
+    /// <summary>
+    /// The exception the item raised.
+    /// </summary>
+    public Exception Exception { get; }
+
+    /// <summary>
+    /// An optional, lazily-evaluated accessor for the item's raw source text, or
+    /// <see langword="null"/> if the stage does not supply one. Invoke it to obtain the text.
+    /// </summary>
+    public Func<string?>? RawContent { get; }
+}

src/Wolfgang.Etl.Abstractions/LoaderBase.cs

@@ -25,6 +25,7 @@ public abstract class LoaderBase<TDestination, TProgress>
 {
     private int _currentItemCount;
     private int _currentSkippedItemCount;
+    private int _currentErrorItemCount;
     private long _startTimestamp;
     private DateTimeOffset _startedAtUtc;
     private bool _disposed;
@@ -112,6 +113,16 @@ public int ReportingInterval
 
 
 
+    /// <summary>
+    /// The number of items that raised an error and were discarded by the stage's error policy
+    /// (<see cref="OnItemError"/> returned <see cref="ItemErrorAction.Skip"/>) so far. This is distinct
+    /// from <see cref="CurrentSkippedItemCount"/>, which counts items skipped intentionally (for
+    /// example by <c>SkipItemCount</c>): a failed record is counted here, never silently dropped.
+    /// </summary>
+    public int CurrentErrorItemCount => Volatile.Read(ref _currentErrorItemCount);
+
+
+
     /// <summary>
     /// The maximum number of items to load. Once the loader has reached this limit,
     /// it should stop loading items as if it had reached the end of the sequence.
@@ -330,6 +341,7 @@ private void ResetRunState()
     {
         Volatile.Write(ref _currentItemCount, 0);
         Volatile.Write(ref _currentSkippedItemCount, 0);
+        Volatile.Write(ref _currentErrorItemCount, 0);
         Volatile.Write(ref _startTimestamp, 0L);
     }
 
@@ -406,6 +418,65 @@ protected void IncrementCurrentSkippedItemCount()
 
 
 
+    /// <summary>
+    /// Decides what to do when an item fails to process. Override in a derived stage to record the
+    /// failure and return <see cref="ItemErrorAction.Skip"/> to discard the item and continue, or
+    /// <see cref="ItemErrorAction.Abort"/> to re-throw and stop the run. The base implementation
+    /// always returns <see cref="ItemErrorAction.Abort"/>, so a stage that does not opt in keeps its
+    /// fail-fast behaviour.
+    /// </summary>
+    /// <param name="context">
+    /// Describes the failed item — its ordinal, the exception, and optional raw content.
+    /// </param>
+    /// <returns>Whether to skip the item or abort the run.</returns>
+    /// <remarks>
+    /// This is the policy hook a derived stage overrides; a worker does not call it directly. A worker
+    /// calls <see cref="HandleItemError"/>, which invokes this method and performs the skip
+    /// bookkeeping. The base classes deliberately expose no public error-handling property: a base
+    /// class cannot catch a per-item failure on the worker's behalf — a C# async iterator cannot
+    /// resume after it throws — so the worker owns the <c>try</c>/<c>catch</c>, and only a format that
+    /// can genuinely resume after a bad record overrides this and surfaces its own public knob.
+    /// </remarks>
+    protected virtual ItemErrorAction OnItemError(ItemErrorContext context)
+    // Stryker disable once all: equivalent — Abort is the enum's default (0), so removing the body
+    // (which makes it return default) yields the identical value; no test can distinguish them.
+    {
+        return ItemErrorAction.Abort;
+    }
+
+
+
+    /// <summary>
+    /// Applies the stage's error policy to a failed item: invokes <see cref="OnItemError"/> and, when
+    /// it returns <see cref="ItemErrorAction.Skip"/>, increments the error-item count so the failure is
+    /// never silent. Call this from a worker's <c>catch</c> block and re-throw when it returns
+    /// <see cref="ItemErrorAction.Abort"/>.
+    /// </summary>
+    /// <param name="context">Describes the failed item.</param>
+    /// <returns>
+    /// <see cref="ItemErrorAction.Skip"/> to discard the item and continue, or
+    /// <see cref="ItemErrorAction.Abort"/> to re-throw.
+    /// </returns>
+    /// <exception cref="ArgumentNullException"><paramref name="context"/> is <see langword="null"/>.</exception>
+    protected ItemErrorAction HandleItemError(ItemErrorContext context)
+    {
+        if (context is null)
+        {
+            throw new ArgumentNullException(nameof(context));
+        }
+
+        var action = OnItemError(context);
+        if (action == ItemErrorAction.Skip)
+        {
+            EnsureStarted();
+            _ = Interlocked.Increment(ref _currentErrorItemCount);
+        }
+
+        return action;
+    }
+
+
+
     // Captures the start timestamp (monotonic) and wall-clock StartedAt the first
     // time any item is processed. Idempotent and thread-safe: the first caller to
     // win the CompareExchange records the start; later calls are a cheap volatile read.