Release 0.21.0

CHANGELOG.md

@@ -19,6 +19,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+## [0.21.0] - 2026-08-03
+
+Minor release: convenience base classes, an assignable per-item `ErrorPolicy` on the three base stages,
+and a new companion package `Wolfgang.Etl.ErrorPolicies` of ready-made policies. Purely additive — no
+breaking change (validates against the 0.20.0 baseline).
+
+### Added
+
+- **Convenience base classes (#344):** `ExtractorBase<TSource>`, `LoaderBase<TDestination>`, and
+  `TransformerBase<TSource, TDestination>` fix the progress type to the built-in `Report` and supply a
+  default `CreateProgressReport()`, so a component that doesn't need a custom progress-report type
+  implements only its worker method — no progress record and no `CreateProgressReport` override.
+  Override it to enrich the report (for example a known total). The existing two/three-type-parameter
+  bases are unchanged. Additive.
+- **`ErrorPolicy` on the base stages (#344 follow-up):** `ExtractorBase`, `LoaderBase`, and
+  `TransformerBase` gained an `ErrorPolicy` (`Func<ItemErrorContext, ItemErrorAction>`, non-null,
+  default fail-fast, **init-only** — set at construction) that the base `OnItemError` consults, so a
+  stage gets a configurable error policy with no per-type property or override — assign one (e.g. from
+  `Wolfgang.Etl.ErrorPolicies`) or override `OnItemError` for stage-internal logic. Additive; unset
+  behaviour is unchanged fail-fast.
+- **`Wolfgang.Etl.ErrorPolicies` package (new, lockstep-versioned with `Wolfgang.Etl.Abstractions`):**
+  a static `ItemErrorPolicy` factory of ready-made policies assignable to a stage's `ErrorPolicy`
+  property — `Skip`, `Abort`, `SkipAndLog(ILogger)`, and dead-letter families `SkipAndDeadLetter` /
+  `SkipDeadLetterAndLog`, each overloaded for a caller-owned `ICollection<ItemErrorContext>` or a
+  `System.Threading.Channels.ChannelWriter<ItemErrorContext>`. Ships from this repo so the shared
+  policy set is defined once for the whole ETL family; the core `Wolfgang.Etl.Abstractions` assembly
+  keeps its minimal dependency set — only this package takes `Microsoft.Extensions.Logging.Abstractions`
+  and `System.Threading.Channels`. The channel dead-letter overloads use the non-blocking `TryWrite`
+  (the hook is synchronous); `SkipDeadLetterAndLog(ChannelWriter, ILogger)` logs a distinct warning when
+  a full bounded channel drops the failure record, so the loss is never silent.
+
+### Changed
+
+### Deprecated
+
+### Removed
+
+### Fixed
+
+### Security
+
 ## [0.20.0] - 2026-07-30
 
 Minor release: a dependency-free retry seam, composable per-item middleware, and pipeline-wide

ETL-Abstractions.sln

@@ -95,6 +95,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Wolfgang.Etl.Abstractions.T
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Wolfgang.Etl.Abstractions.Tests.DocExamples", "tests\Wolfgang.Etl.Abstractions.Tests.DocExamples\Wolfgang.Etl.Abstractions.Tests.DocExamples.csproj", "{76B42021-1D8C-4608-B8D4-EEDC25081248}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Wolfgang.Etl.ErrorPolicies", "src\Wolfgang.Etl.ErrorPolicies\Wolfgang.Etl.ErrorPolicies.csproj", "{A59EF662-7051-4C8F-9515-640326502B15}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Wolfgang.Etl.ErrorPolicies.Tests.Unit", "tests\Wolfgang.Etl.ErrorPolicies.Tests.Unit\Wolfgang.Etl.ErrorPolicies.Tests.Unit.csproj", "{DD88B915-99AF-4545-AD35-6F296693B92B}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -405,6 +409,30 @@ Global
 		{76B42021-1D8C-4608-B8D4-EEDC25081248}.Release|x64.Build.0 = Release|Any CPU
 		{76B42021-1D8C-4608-B8D4-EEDC25081248}.Release|x86.ActiveCfg = Release|Any CPU
 		{76B42021-1D8C-4608-B8D4-EEDC25081248}.Release|x86.Build.0 = Release|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Debug|x64.Build.0 = Debug|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Debug|x86.Build.0 = Debug|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Release|Any CPU.Build.0 = Release|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Release|x64.ActiveCfg = Release|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Release|x64.Build.0 = Release|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Release|x86.ActiveCfg = Release|Any CPU
+		{A59EF662-7051-4C8F-9515-640326502B15}.Release|x86.Build.0 = Release|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Debug|x64.Build.0 = Debug|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Debug|x86.Build.0 = Debug|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Release|Any CPU.Build.0 = Release|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Release|x64.ActiveCfg = Release|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Release|x64.Build.0 = Release|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Release|x86.ActiveCfg = Release|Any CPU
+		{DD88B915-99AF-4545-AD35-6F296693B92B}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -439,6 +467,8 @@ Global
 		{B59B4926-9CDD-41C2-A367-42F030DFAF54} = {336D72A1-8E5E-49DE-83D9-DF6BE458BA24}
 		{8F775698-9B29-40AB-A02E-A72A9B073040} = {8220BC33-6632-4D4C-9A50-B7978141A4E3}
 		{76B42021-1D8C-4608-B8D4-EEDC25081248} = {8220BC33-6632-4D4C-9A50-B7978141A4E3}
+		{A59EF662-7051-4C8F-9515-640326502B15} = {02EA681E-C7D8-13C7-8484-4AC65E1B71E8}
+		{DD88B915-99AF-4545-AD35-6F296693B92B} = {8220BC33-6632-4D4C-9A50-B7978141A4E3}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {F673635D-58CE-48A5-9AE4-31F4484BED9E}

src/Wolfgang.Etl.Abstractions/ExtractorBase.cs

@@ -476,30 +476,49 @@ protected void IncrementCurrentSkippedItemCount()
 
 
 
+    private static readonly Func<ItemErrorContext, ItemErrorAction> AbortPolicy =
+        static _ => ItemErrorAction.Abort;
+
+    private readonly Func<ItemErrorContext, ItemErrorAction> _errorPolicy = AbortPolicy;
+
+
+
+    /// <summary>
+    /// Gets the policy invoked when an item fails to process. Return <see cref="ItemErrorAction.Skip"/>
+    /// to discard the item and continue, or <see cref="ItemErrorAction.Abort"/> to re-throw and stop
+    /// the run. Defaults to fail-fast: every failed item aborts the run until a policy is assigned.
+    /// Ready-made policies are provided by <c>Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy</c>.
+    /// Assigned once, at construction (init-only), so the policy cannot change during a run.
+    /// </summary>
+    /// <exception cref="ArgumentNullException">The assigned value is <see langword="null"/>.</exception>
+    public Func<ItemErrorContext, ItemErrorAction> ErrorPolicy
+    {
+        get => _errorPolicy;
+        init => _errorPolicy = value ?? throw new ArgumentNullException(nameof(value));
+    }
+
+
+
     /// <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.
+    /// Decides what to do when an item fails to process. The base implementation delegates to
+    /// <see cref="ErrorPolicy"/> (fail-fast by default). Override in a derived stage instead only when
+    /// the decision needs stage-internal state; a worker does not call this directly.
     /// </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.
+    /// A worker calls <see cref="HandleItemError"/>, which invokes this method and performs the skip
+    /// bookkeeping. The worker still owns the <c>try</c>/<c>catch</c> — a C# async iterator cannot
+    /// resume after it throws — and calls <see cref="HandleItemError"/> from it. On a format that
+    /// cannot genuinely resume after a bad record, <see cref="ItemErrorAction.Skip"/> means "swallow
+    /// the failure and stop at that point" rather than "skip and continue"; such a stage documents
+    /// that on its own type.
     /// </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;
+        return ErrorPolicy(context);
     }
 
 

src/Wolfgang.Etl.Abstractions/ExtractorBase{TSource}.cs

@@ -0,0 +1,21 @@
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// A convenience <see cref="ExtractorBase{TSource, TProgress}"/> that reports progress with the
+/// built-in <see cref="Report"/> type and supplies a default <see cref="CreateProgressReport"/>, so a
+/// derived extractor only has to implement <c>ExtractWorkerAsync</c>. Use this instead of the
+/// two-type-parameter base when you don't need a custom progress-report type — it removes the
+/// progress-record and <c>CreateProgressReport</c> boilerplate. Override
+/// <see cref="CreateProgressReport"/> if you want to enrich the report (for example set a known total).
+/// </summary>
+/// <typeparam name="TSource">The type of the object being extracted.</typeparam>
+public abstract class ExtractorBase<TSource> : ExtractorBase<TSource, Report>
+    where TSource : notnull
+{
+    /// <summary>
+    /// Builds a <see cref="Report"/> snapshot from the current item count and timing. Override to add
+    /// more detail (for example a known <see cref="Report.TotalItemCount"/>).
+    /// </summary>
+    /// <returns>A <see cref="Report"/> for the current run.</returns>
+    protected override Report CreateProgressReport() => new(CurrentItemCount, StartedAt, Elapsed);
+}

src/Wolfgang.Etl.Abstractions/LoaderBase.cs

@@ -474,30 +474,49 @@ protected void IncrementCurrentSkippedItemCount()
 
 
 
+    private static readonly Func<ItemErrorContext, ItemErrorAction> AbortPolicy =
+        static _ => ItemErrorAction.Abort;
+
+    private readonly Func<ItemErrorContext, ItemErrorAction> _errorPolicy = AbortPolicy;
+
+
+
+    /// <summary>
+    /// Gets the policy invoked when an item fails to process. Return <see cref="ItemErrorAction.Skip"/>
+    /// to discard the item and continue, or <see cref="ItemErrorAction.Abort"/> to re-throw and stop
+    /// the run. Defaults to fail-fast: every failed item aborts the run until a policy is assigned.
+    /// Ready-made policies are provided by <c>Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy</c>.
+    /// Assigned once, at construction (init-only), so the policy cannot change during a run.
+    /// </summary>
+    /// <exception cref="ArgumentNullException">The assigned value is <see langword="null"/>.</exception>
+    public Func<ItemErrorContext, ItemErrorAction> ErrorPolicy
+    {
+        get => _errorPolicy;
+        init => _errorPolicy = value ?? throw new ArgumentNullException(nameof(value));
+    }
+
+
+
     /// <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.
+    /// Decides what to do when an item fails to process. The base implementation delegates to
+    /// <see cref="ErrorPolicy"/> (fail-fast by default). Override in a derived stage instead only when
+    /// the decision needs stage-internal state; a worker does not call this directly.
     /// </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.
+    /// A worker calls <see cref="HandleItemError"/>, which invokes this method and performs the skip
+    /// bookkeeping. The worker still owns the <c>try</c>/<c>catch</c> — a C# async iterator cannot
+    /// resume after it throws — and calls <see cref="HandleItemError"/> from it. On a format that
+    /// cannot genuinely resume after a bad record, <see cref="ItemErrorAction.Skip"/> means "swallow
+    /// the failure and stop at that point" rather than "skip and continue"; such a stage documents
+    /// that on its own type.
     /// </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;
+        return ErrorPolicy(context);
     }
 
 

src/Wolfgang.Etl.Abstractions/LoaderBase{TDestination}.cs

@@ -0,0 +1,21 @@
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// A convenience <see cref="LoaderBase{TDestination, TProgress}"/> that reports progress with the
+/// built-in <see cref="Report"/> type and supplies a default <see cref="CreateProgressReport"/>, so a
+/// derived loader only has to implement <c>LoadWorkerAsync</c>. Use this instead of the
+/// two-type-parameter base when you don't need a custom progress-report type — it removes the
+/// progress-record and <c>CreateProgressReport</c> boilerplate. Override
+/// <see cref="CreateProgressReport"/> if you want to enrich the report (for example set a known total).
+/// </summary>
+/// <typeparam name="TDestination">The type of the object being loaded.</typeparam>
+public abstract class LoaderBase<TDestination> : LoaderBase<TDestination, Report>
+    where TDestination : notnull
+{
+    /// <summary>
+    /// Builds a <see cref="Report"/> snapshot from the current item count and timing. Override to add
+    /// more detail (for example a known <see cref="Report.TotalItemCount"/>).
+    /// </summary>
+    /// <returns>A <see cref="Report"/> for the current run.</returns>
+    protected override Report CreateProgressReport() => new(CurrentItemCount, StartedAt, Elapsed);
+}