Add Wolfgang.Etl.ErrorPolicies package (lockstep 0.21.0)

CHANGELOG.md

@@ -15,6 +15,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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 a settable `ErrorPolicy` (`Func<ItemErrorContext, ItemErrorAction>`,
+  non-null, default fail-fast) 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
 

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,48 @@ 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>.
+    /// </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.cs

@@ -474,30 +474,48 @@ 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>.
+    /// </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/PublicAPI.Unshipped.txt

@@ -1,10 +1,16 @@
 #nullable enable
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource>
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource>.ExtractorBase() -> void
+Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.ErrorPolicy.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.ErrorPolicy.init -> void
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination>
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination>.LoaderBase() -> void
+Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.ErrorPolicy.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.ErrorPolicy.init -> void
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>.TransformerBase() -> void
+Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.ErrorPolicy.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.ErrorPolicy.init -> void
 override Wolfgang.Etl.Abstractions.ExtractorBase<TSource>.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report!
 override Wolfgang.Etl.Abstractions.LoaderBase<TDestination>.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report!
 override Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report!

src/Wolfgang.Etl.Abstractions/TransformerBase.cs

@@ -485,30 +485,48 @@ 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>.
+    /// </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/Wolfgang.Etl.Abstractions.csproj

@@ -2,7 +2,7 @@
   <PropertyGroup>
 	<TargetFrameworks>net462;net472;net48;net481;netstandard2.0;net5.0;net6.0;net7.0;net8.0;net9.0;net10.0</TargetFrameworks>
 	<LangVersion>latest</LangVersion>
-	<Version>0.20.0</Version>
+	<Version>0.21.0</Version>
 	<!-- Pin AssemblyVersion to a fixed binding-stability baseline so consumers
 	     do not need to recompile / add binding redirects on every minor/patch
 	     bump. Bump only on a deliberate breaking API change. FileVersion +