From 46534ac87f26789deb401d5b893d332b9aaa04ca Mon Sep 17 00:00:00 2001
From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com>
Date: Wed, 29 Jul 2026 22:37:17 -0400
Subject: [PATCH 1/6] feat: convenience base classes fixing TProgress=Report
 (#344, Part A)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add ExtractorBase<TSource>, LoaderBase<TDestination>, and TransformerBase<TSource, TDestination>
that inherit the two/three-type-parameter bases with TProgress fixed to the built-in Report and a
default CreateProgressReport() => new(CurrentItemCount, StartedAt, Elapsed). A component that
doesn't need a custom progress type now implements ONLY its worker method — no progress record,
no CreateProgressReport override. The default isn't sealed, so a component can still enrich the
report. The existing bases are unchanged; purely additive.

This is the lighter-weight alternative to the #96 source generator for the common boilerplate.
Part B (lifting the timer-injection guard into the base) tracked separately in #344.

- 4 ConvenienceBaseTests (extractor/loader/transformer smoke + report content + overridable default).
  456 tests; 100% mutation; all 11 TFMs clean.
- PublicAPI.Unshipped updated (RS0016 completeness + RS0017 correctness validated).
- CHANGELOG [Unreleased] Added entry.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |   7 +
 .../ExtractorBase{TSource}.cs                 |  21 +++
 .../LoaderBase{TDestination}.cs               |  21 +++
 .../PublicAPI.Unshipped.txt                   |   9 +
 .../TransformerBase{TSource,TDestination}.cs  |  23 +++
 .../BaseClassTests/ConvenienceBaseTests.cs    | 166 ++++++++++++++++++
 6 files changed, 247 insertions(+)
 create mode 100644 src/Wolfgang.Etl.Abstractions/ExtractorBase{TSource}.cs
 create mode 100644 src/Wolfgang.Etl.Abstractions/LoaderBase{TDestination}.cs
 create mode 100644 src/Wolfgang.Etl.Abstractions/TransformerBase{TSource,TDestination}.cs
 create mode 100644 tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/ConvenienceBaseTests.cs

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 86e35911..820acc3f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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.
+
 ### Changed
 
 ### Deprecated
diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase{TSource}.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase{TSource}.cs
new file mode 100644
index 00000000..97db3772
--- /dev/null
+++ b/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);
+}
diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase{TDestination}.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase{TDestination}.cs
new file mode 100644
index 00000000..2cbe964a
--- /dev/null
+++ b/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);
+}
diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
index 7dc5c581..514883cd 100644
--- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
+++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
@@ -1 +1,10 @@
 #nullable enable
+Wolfgang.Etl.Abstractions.ExtractorBase<TSource>
+Wolfgang.Etl.Abstractions.ExtractorBase<TSource>.ExtractorBase() -> void
+Wolfgang.Etl.Abstractions.LoaderBase<TDestination>
+Wolfgang.Etl.Abstractions.LoaderBase<TDestination>.LoaderBase() -> void
+Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>
+Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>.TransformerBase() -> 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!
diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase{TSource,TDestination}.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase{TSource,TDestination}.cs
new file mode 100644
index 00000000..7df13874
--- /dev/null
+++ b/src/Wolfgang.Etl.Abstractions/TransformerBase{TSource,TDestination}.cs
@@ -0,0 +1,23 @@
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// A convenience <see cref="TransformerBase{TSource, TDestination, TProgress}"/> that reports progress
+/// with the built-in <see cref="Report"/> type and supplies a default <see cref="CreateProgressReport"/>,
+/// so a derived transformer only has to implement <c>TransformWorkerAsync</c>. Use this instead of the
+/// three-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 source object.</typeparam>
+/// <typeparam name="TDestination">The type of the destination object.</typeparam>
+public abstract class TransformerBase<TSource, TDestination> : TransformerBase<TSource, TDestination, Report>
+    where TSource : notnull
+    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);
+}
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/ConvenienceBaseTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/ConvenienceBaseTests.cs
new file mode 100644
index 00000000..629b29f4
--- /dev/null
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/ConvenienceBaseTests.cs
@@ -0,0 +1,166 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Wolfgang.Etl.Abstractions;
+using Xunit;
+
+namespace Wolfgang.Etl.Abstractions.Tests.Unit.BaseClassTests;
+
+/// <summary>
+/// Covers the #344 convenience base classes — <see cref="ExtractorBase{TSource}"/>,
+/// <see cref="LoaderBase{TDestination}"/>, and <see cref="TransformerBase{TSource, TDestination}"/> —
+/// which fix the progress type to the built-in <see cref="Report"/> and supply a default
+/// <c>CreateProgressReport()</c>, so a derived component only implements its worker method.
+/// </summary>
+public class ConvenienceBaseTests
+{
+    [Fact]
+    public async Task Extractor_convenience_base_yields_items_and_reports_the_built_in_Report()
+    {
+        var sut = new ConvenienceExtractor();
+
+        var items = await Drain(sut.ExtractAsync(CancellationToken.None));
+        var report = sut.Report();
+
+        Assert.Equal(new[] { 1, 2, 3 }, items);
+        Assert.Equal(3, report.CurrentItemCount);
+        Assert.NotNull(report.StartedAt);              // captured once the first item was processed
+        Assert.True(report.Elapsed >= TimeSpan.Zero);
+    }
+
+
+    [Fact]
+    public async Task Loader_convenience_base_loads_items_and_reports_the_built_in_Report()
+    {
+        var sut = new ConvenienceLoader();
+
+        await sut.LoadAsync(AsyncSource(1, 2, 3), CancellationToken.None);
+        var report = sut.Report();
+
+        Assert.Equal(new[] { 1, 2, 3 }, sut.Loaded);
+        Assert.Equal(3, report.CurrentItemCount);
+    }
+
+
+    [Fact]
+    public async Task Transformer_convenience_base_transforms_items_and_reports_the_built_in_Report()
+    {
+        var sut = new ConvenienceTransformer();
+
+        var items = await Drain(sut.TransformAsync(AsyncSource(1, 2, 3), CancellationToken.None));
+        var report = sut.Report();
+
+        Assert.Equal(new[] { 10, 20, 30 }, items);
+        Assert.Equal(3, report.CurrentItemCount);
+    }
+
+
+    [Fact]
+    public async Task Convenience_base_CreateProgressReport_can_still_be_overridden()
+    {
+        // The default CreateProgressReport is not sealed — a component can enrich it (here: a known total).
+        var sut = new TotalAwareExtractor(total: 10);
+
+        await Drain(sut.ExtractAsync(CancellationToken.None));
+
+        Assert.Equal(10, sut.Report().TotalItemCount);
+    }
+
+
+    // ---------- helpers ----------
+
+    private static async IAsyncEnumerable<int> AsyncSource(params int[] items)
+    {
+        foreach (var item in items)
+        {
+            await Task.Yield();
+            yield return item;
+        }
+    }
+
+
+    private static async Task<List<int>> Drain(IAsyncEnumerable<int> source)
+    {
+        var result = new List<int>();
+        await foreach (var item in source.ConfigureAwait(false))
+        {
+            result.Add(item);
+        }
+
+        return result;
+    }
+
+
+    // ---------- doubles (each implements ONLY its worker — no progress record, no CreateProgressReport) ----------
+
+    private sealed class ConvenienceExtractor : ExtractorBase<int>
+    {
+        public Report Report() => CreateProgressReport();
+
+        protected override async IAsyncEnumerable<int> ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token)
+        {
+            foreach (var i in new[] { 1, 2, 3 })
+            {
+                await Task.Yield();
+                IncrementCurrentItemCount();
+                yield return i;
+            }
+        }
+    }
+
+
+    private sealed class ConvenienceLoader : LoaderBase<int>
+    {
+        public List<int> Loaded { get; } = new();
+
+        public Report Report() => CreateProgressReport();
+
+        protected override async Task LoadWorkerAsync(IAsyncEnumerable<int> items, CancellationToken token)
+        {
+            await foreach (var item in items.WithCancellation(token).ConfigureAwait(false))
+            {
+                Loaded.Add(item);
+                IncrementCurrentItemCount();
+            }
+        }
+    }
+
+
+    private sealed class ConvenienceTransformer : TransformerBase<int, int>
+    {
+        public Report Report() => CreateProgressReport();
+
+        protected override async IAsyncEnumerable<int> TransformWorkerAsync(
+            IAsyncEnumerable<int> items, [EnumeratorCancellation] CancellationToken token)
+        {
+            await foreach (var item in items.WithCancellation(token).ConfigureAwait(false))
+            {
+                IncrementCurrentItemCount();
+                yield return item * 10;
+            }
+        }
+    }
+
+
+    // Shows a component enriching the default report (proves CreateProgressReport isn't sealed).
+    private sealed class TotalAwareExtractor : ExtractorBase<int>
+    {
+        private readonly int _total;
+
+        public TotalAwareExtractor(int total) => _total = total;
+
+        public Report Report() => CreateProgressReport();
+
+        protected override Report CreateProgressReport() => new(CurrentItemCount, StartedAt, Elapsed, _total);
+
+        protected override async IAsyncEnumerable<int> ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token)
+        {
+            await Task.Yield();
+            IncrementCurrentItemCount();
+            yield return 1;
+        }
+    }
+}

From d770fa7ac0a5b85a4e6c42e94cba891ebb2a09ef Mon Sep 17 00:00:00 2001
From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com>
Date: Sun, 2 Aug 2026 19:57:06 -0400
Subject: [PATCH 2/6] Add Wolfgang.Etl.ErrorPolicies package (lockstep 0.21.0)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

New second package built from this repo, versioned in lockstep with
Wolfgang.Etl.Abstractions (same <Version>, one tag — like TestKit +
TestKit.Xunit). Hosts the shared generic ItemErrorPolicy factory for the
family's OnError hook so the policies are defined once instead of per format:

- Skip / Abort (Func<ItemErrorContext, ItemErrorAction> properties)
- SkipAndLog(ILogger)
- SkipAndDeadLetter / SkipDeadLetterAndLog, each overloaded for a caller-owned
  ICollection<ItemErrorContext> or a ChannelWriter<ItemErrorContext>

The core Abstractions assembly keeps its minimal deps; only this package takes
Microsoft.Extensions.Logging.Abstractions (all TFMs) and System.Threading.Channels
(netFx / netstandard2.0 only — in-box on net5.0+). Bumps Abstractions to 0.21.0
for the lockstep release; PackageValidation baseline stays at last-published 0.20.0.
No release.yaml change needed — it already packs every src/ project, validates the
tag against any src <Version>, and pushes all nupkgs.

14 unit tests pass net462 -> net10.0; full-matrix build clean; pack emits the
correct per-TFM dependency groups (Abstractions 0.21.0 lockstep dep on every TFM).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |   9 +
 ETL-Abstractions.sln                          |  30 +++
 .../Wolfgang.Etl.Abstractions.csproj          |   2 +-
 .../ItemErrorPolicy.cs                        | 195 +++++++++++++++++
 .../ItemErrorPolicyLog.cs                     |  19 ++
 .../PublicAPI.Shipped.txt                     |   1 +
 .../PublicAPI.Unshipped.txt                   |   9 +
 .../Wolfgang.Etl.ErrorPolicies.csproj         |  61 ++++++
 .../ItemErrorPolicyTests.cs                   | 199 ++++++++++++++++++
 ...lfgang.Etl.ErrorPolicies.Tests.Unit.csproj |  83 ++++++++
 10 files changed, 607 insertions(+), 1 deletion(-)
 create mode 100644 src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
 create mode 100644 src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs
 create mode 100644 src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt
 create mode 100644 src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt
 create mode 100644 src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj
 create mode 100644 tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs
 create mode 100644 tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/Wolfgang.Etl.ErrorPolicies.Tests.Unit.csproj

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 86e35911..efff659a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **`Wolfgang.Etl.ErrorPolicies` package (new, lockstep-versioned with `Wolfgang.Etl.Abstractions`):**
+  a static `ItemErrorPolicy` factory of ready-made policies for a stage's `OnError` hook —
+  `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`.
+
 ### Changed
 
 ### Deprecated
diff --git a/ETL-Abstractions.sln b/ETL-Abstractions.sln
index 8a3a782b..c4f8d136 100644
--- a/ETL-Abstractions.sln
+++ b/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}
diff --git a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj
index ccd92fae..bd8b6730 100644
--- a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj
+++ b/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 +
diff --git a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
new file mode 100644
index 00000000..4ba8e15b
--- /dev/null
+++ b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
@@ -0,0 +1,195 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Channels;
+using Microsoft.Extensions.Logging;
+using Wolfgang.Etl.Abstractions;
+
+namespace Wolfgang.Etl.ErrorPolicies;
+
+/// <summary>
+/// Ready-made policies for an ETL stage's <c>OnError</c> hook. Each is a
+/// <see cref="Func{T, TResult}"/> from an <see cref="ItemErrorContext"/> to an
+/// <see cref="ItemErrorAction"/>, so it can be assigned directly to a stage's <c>OnError</c>:
+/// <example><code>
+/// var deadLetters = new List&lt;ItemErrorContext&gt;();
+/// var extractor = new SomeExtractor&lt;Record&gt;(source)
+/// {
+///     OnError = ItemErrorPolicy.SkipDeadLetterAndLog(deadLetters, logger)
+/// };
+/// </code></example>
+/// The dead-letter overloads write to a caller-owned sink (a collection or a channel), so its size —
+/// and therefore the memory a bad feed can consume — stays under the caller's control.
+/// </summary>
+public static class ItemErrorPolicy
+{
+    /// <summary>
+    /// A policy that discards the failed item and continues with the next one. The stage increments
+    /// its error-item count (<c>CurrentErrorItemCount</c>) so the skip is never silent.
+    /// </summary>
+    public static Func<ItemErrorContext, ItemErrorAction> Skip { get; } = _ => ItemErrorAction.Skip;
+
+
+
+    /// <summary>
+    /// A policy that re-throws the failure and stops the run. Equivalent to leaving <c>OnError</c>
+    /// unset, provided for symmetry and explicitness.
+    /// </summary>
+    public static Func<ItemErrorContext, ItemErrorAction> Abort { get; } = _ => ItemErrorAction.Abort;
+
+
+
+    /// <summary>
+    /// Logs the failure as a warning through <paramref name="logger"/>, then discards the item and
+    /// continues.
+    /// </summary>
+    /// <param name="logger">The logger the returned policy writes each failure to.</param>
+    /// <returns>A policy that logs and returns <see cref="ItemErrorAction.Skip"/>.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="logger"/> is <see langword="null"/>.</exception>
+    public static Func<ItemErrorContext, ItemErrorAction> SkipAndLog(ILogger logger)
+    {
+        if (logger is null)
+        {
+            throw new ArgumentNullException(nameof(logger));
+        }
+
+        return context =>
+        {
+            ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception);
+            return ItemErrorAction.Skip;
+        };
+    }
+
+
+
+    /// <summary>
+    /// Records the failure in <paramref name="deadLetters"/> (a "dead-letter" queue the caller owns),
+    /// then discards the item and continues.
+    /// </summary>
+    /// <param name="deadLetters">The caller-owned collection each failed item is added to.</param>
+    /// <returns>A policy that dead-letters and returns <see cref="ItemErrorAction.Skip"/>.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="deadLetters"/> is <see langword="null"/>.</exception>
+    /// <remarks>
+    /// A single stage invokes this policy serially, so a plain <see cref="List{T}"/> is safe. If you
+    /// share one collection across stages running concurrently, either supply a thread-safe collection
+    /// or use the <see cref="SkipAndDeadLetter(ChannelWriter{ItemErrorContext})"/> overload — the policy
+    /// adds to the collection without locking.
+    /// </remarks>
+    public static Func<ItemErrorContext, ItemErrorAction> SkipAndDeadLetter(ICollection<ItemErrorContext> deadLetters)
+    {
+        if (deadLetters is null)
+        {
+            throw new ArgumentNullException(nameof(deadLetters));
+        }
+
+        return context =>
+        {
+            deadLetters.Add(context);
+            return ItemErrorAction.Skip;
+        };
+    }
+
+
+
+    /// <summary>
+    /// Writes the failure to <paramref name="deadLetters"/> (a caller-owned channel) with
+    /// <see cref="ChannelWriter{T}.TryWrite(T)"/>, then discards the item and continues. Because the
+    /// hook is synchronous the non-blocking <c>TryWrite</c> is used, so a bounded channel that is full
+    /// drops the failure — size the channel, or use <see cref="BoundedChannelFullMode"/>, accordingly.
+    /// </summary>
+    /// <param name="deadLetters">The caller-owned channel each failed item is written to.</param>
+    /// <returns>A policy that dead-letters and returns <see cref="ItemErrorAction.Skip"/>.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="deadLetters"/> is <see langword="null"/>.</exception>
+    public static Func<ItemErrorContext, ItemErrorAction> SkipAndDeadLetter(ChannelWriter<ItemErrorContext> deadLetters)
+    {
+        if (deadLetters is null)
+        {
+            throw new ArgumentNullException(nameof(deadLetters));
+        }
+
+        return context =>
+        {
+            deadLetters.TryWrite(context);
+            return ItemErrorAction.Skip;
+        };
+    }
+
+
+
+    /// <summary>
+    /// Records the failure in <paramref name="deadLetters"/> and logs it as a warning through
+    /// <paramref name="logger"/>, then discards the item and continues.
+    /// </summary>
+    /// <param name="deadLetters">The caller-owned collection each failed item is added to.</param>
+    /// <param name="logger">The logger the returned policy writes each failure to.</param>
+    /// <returns>A policy that dead-letters, logs, and returns <see cref="ItemErrorAction.Skip"/>.</returns>
+    /// <exception cref="ArgumentNullException">
+    /// <paramref name="deadLetters"/> or <paramref name="logger"/> is <see langword="null"/>.
+    /// </exception>
+    /// <remarks>
+    /// A single stage invokes this policy serially, so a plain <see cref="List{T}"/> is safe. If you
+    /// share one collection across stages running concurrently, either supply a thread-safe collection
+    /// or use the <see cref="SkipDeadLetterAndLog(ChannelWriter{ItemErrorContext}, ILogger)"/> overload —
+    /// the policy adds to the collection without locking.
+    /// </remarks>
+    public static Func<ItemErrorContext, ItemErrorAction> SkipDeadLetterAndLog
+    (
+        ICollection<ItemErrorContext> deadLetters,
+        ILogger logger
+    )
+    {
+        if (deadLetters is null)
+        {
+            throw new ArgumentNullException(nameof(deadLetters));
+        }
+
+        if (logger is null)
+        {
+            throw new ArgumentNullException(nameof(logger));
+        }
+
+        return context =>
+        {
+            deadLetters.Add(context);
+            ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception);
+            return ItemErrorAction.Skip;
+        };
+    }
+
+
+
+    /// <summary>
+    /// Writes the failure to <paramref name="deadLetters"/> with
+    /// <see cref="ChannelWriter{T}.TryWrite(T)"/> and logs it as a warning through
+    /// <paramref name="logger"/>, then discards the item and continues. See
+    /// <see cref="SkipAndDeadLetter(ChannelWriter{ItemErrorContext})"/> for the <c>TryWrite</c> caveat.
+    /// </summary>
+    /// <param name="deadLetters">The caller-owned channel each failed item is written to.</param>
+    /// <param name="logger">The logger the returned policy writes each failure to.</param>
+    /// <returns>A policy that dead-letters, logs, and returns <see cref="ItemErrorAction.Skip"/>.</returns>
+    /// <exception cref="ArgumentNullException">
+    /// <paramref name="deadLetters"/> or <paramref name="logger"/> is <see langword="null"/>.
+    /// </exception>
+    public static Func<ItemErrorContext, ItemErrorAction> SkipDeadLetterAndLog
+    (
+        ChannelWriter<ItemErrorContext> deadLetters,
+        ILogger logger
+    )
+    {
+        if (deadLetters is null)
+        {
+            throw new ArgumentNullException(nameof(deadLetters));
+        }
+
+        if (logger is null)
+        {
+            throw new ArgumentNullException(nameof(logger));
+        }
+
+        return context =>
+        {
+            deadLetters.TryWrite(context);
+            ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception);
+            return ItemErrorAction.Skip;
+        };
+    }
+}
diff --git a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs
new file mode 100644
index 00000000..2a881977
--- /dev/null
+++ b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs
@@ -0,0 +1,19 @@
+using System;
+using Microsoft.Extensions.Logging;
+
+namespace Wolfgang.Etl.ErrorPolicies;
+
+/// <summary>
+/// Cached <see cref="LoggerMessage"/> delegates for the logging error policies, so a per-item log
+/// call allocates nothing on the hot path.
+/// </summary>
+internal static class ItemErrorPolicyLog
+{
+    internal static readonly Action<ILogger, long, Exception?> ItemFailedAndSkipped =
+        LoggerMessage.Define<long>
+        (
+            LogLevel.Warning,
+            new EventId(1, nameof(ItemFailedAndSkipped)),
+            "Item {ItemNumber} failed to process and was skipped."
+        );
+}
diff --git a/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt
new file mode 100644
index 00000000..7dc5c581
--- /dev/null
+++ b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt
@@ -0,0 +1 @@
+#nullable enable
diff --git a/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt
new file mode 100644
index 00000000..8af9f268
--- /dev/null
+++ b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt
@@ -0,0 +1,9 @@
+#nullable enable
+Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Abort.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Skip.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndLog(Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
diff --git a/src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj b/src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj
new file mode 100644
index 00000000..75d0bbeb
--- /dev/null
+++ b/src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj
@@ -0,0 +1,61 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+	<TargetFrameworks>net462;net472;net48;net481;netstandard2.0;net5.0;net6.0;net7.0;net8.0;net9.0;net10.0</TargetFrameworks>
+	<LangVersion>latest</LangVersion>
+	<!-- Lockstep with Wolfgang.Etl.Abstractions: this package is built and released from the same
+	     repo under the same version and tag (like Wolfgang.Etl.TestKit + Wolfgang.Etl.TestKit.Xunit). -->
+	<Version>0.21.0</Version>
+	<AssemblyVersion>1.0.0.0</AssemblyVersion>
+	<FileVersion>$([System.Text.RegularExpressions.Regex]::Replace("$(Version)", "[-+].*$", "")).0</FileVersion>
+    <!-- No EnablePackageValidation on the FIRST release: there is no previously-published baseline to
+         compare against. Enable it next cycle with PackageValidationBaselineVersion set to 0.21.0. -->
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <Title>$(AssemblyName)</Title>
+    <Description>Ready-made item-error policies — skip, log, and dead-letter (to a collection or a channel) — for the OnError hook of Wolfgang.Etl extractors, loaders, and transformers. Built on Wolfgang.Etl.Abstractions.</Description>
+    <PackageProjectUrl>https://github.com/Chris-Wolfgang/ETL-Abstractions</PackageProjectUrl>
+    <PackageReadmeFile>README.md</PackageReadmeFile>
+    <RepositoryUrl>https://github.com/Chris-Wolfgang/ETL-Abstractions</RepositoryUrl>
+    <PackageLicenseExpression>MIT</PackageLicenseExpression>
+    <PackageRequireLicenseAcceptance>True</PackageRequireLicenseAcceptance>
+    <PackageIcon>ETL-Abstractions.png</PackageIcon>
+    <GenerateDocumentationFile>True</GenerateDocumentationFile>
+    <SignAssembly>False</SignAssembly>
+    <!-- The 10.0.10 logging packages drop net5.0/net7.0 from their tested TFM list; they resolve the
+         netstandard2.0 asset there and work. Silence the NETSDK TFM-support warnings on those TFMs. -->
+    <SuppressTfmSupportBuildWarnings>true</SuppressTfmSupportBuildWarnings>
+    <PackageTags>ETL;Extract-Transform-Load;error-handling;dead-letter</PackageTags>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <InternalsVisibleTo Include="Wolfgang.Etl.ErrorPolicies.Tests.Unit" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Include="..\..\ETL-Abstractions.png">
+      <Pack>True</Pack>
+      <PackagePath>\</PackagePath>
+    </None>
+    <None Include="..\..\README.md">
+      <Pack>True</Pack>
+      <PackagePath>\</PackagePath>
+    </None>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Wolfgang.Etl.Abstractions\Wolfgang.Etl.Abstractions.csproj" />
+  </ItemGroup>
+
+  <!-- Logging.Abstractions is always a package reference (never in-box). -->
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.10" />
+  </ItemGroup>
+
+  <!-- System.Threading.Channels is in-box on net5.0+; only the netFx / netstandard2.0 TFMs need the package. -->
+  <ItemGroup Condition="'$(TargetFramework)' == 'net462' OR '$(TargetFramework)' == 'net472' OR '$(TargetFramework)' == 'net48' OR '$(TargetFramework)' == 'net481' OR '$(TargetFramework)' == 'netstandard2.0'">
+    <PackageReference Include="System.Threading.Channels" Version="10.0.10" />
+  </ItemGroup>
+
+  <!-- Analyzer PackageReferences are centralized in Directory.Build.props -->
+
+
+</Project>
diff --git a/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs b/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs
new file mode 100644
index 00000000..a7c2df04
--- /dev/null
+++ b/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs
@@ -0,0 +1,199 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Channels;
+using Microsoft.Extensions.Logging;
+using Wolfgang.Etl.Abstractions;
+using Wolfgang.Etl.ErrorPolicies;
+
+namespace Wolfgang.Etl.ErrorPolicies.Tests.Unit;
+
+public sealed class ItemErrorPolicyTests
+{
+    private static ItemErrorContext Context() =>
+        new(42, new InvalidOperationException("boom"), () => "raw");
+
+
+
+    [Fact]
+    public void Skip_returns_Skip()
+    {
+        Assert.Equal(ItemErrorAction.Skip, ItemErrorPolicy.Skip(Context()));
+    }
+
+
+
+    [Fact]
+    public void Abort_returns_Abort()
+    {
+        Assert.Equal(ItemErrorAction.Abort, ItemErrorPolicy.Abort(Context()));
+    }
+
+
+
+    [Fact]
+    public void SkipAndLog_when_logger_is_null_throws_ArgumentNullException()
+    {
+        Assert.Throws<ArgumentNullException>(() => ItemErrorPolicy.SkipAndLog(null!));
+    }
+
+
+
+    [Fact]
+    public void SkipAndLog_logs_the_failure_and_returns_Skip()
+    {
+        var logger = new RecordingLogger();
+
+        var action = ItemErrorPolicy.SkipAndLog(logger)(Context());
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.Equal(1, logger.WarningCount);
+    }
+
+
+
+    [Fact]
+    public void SkipAndDeadLetter_collection_when_deadLetters_is_null_throws_ArgumentNullException()
+    {
+        Assert.Throws<ArgumentNullException>(() => ItemErrorPolicy.SkipAndDeadLetter((ICollection<ItemErrorContext>)null!));
+    }
+
+
+
+    [Fact]
+    public void SkipAndDeadLetter_collection_records_the_failure_and_returns_Skip()
+    {
+        var deadLetters = new List<ItemErrorContext>();
+        var context = Context();
+
+        var action = ItemErrorPolicy.SkipAndDeadLetter(deadLetters)(context);
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.Same(context, Assert.Single(deadLetters));
+    }
+
+
+
+    [Fact]
+    public void SkipAndDeadLetter_channel_when_deadLetters_is_null_throws_ArgumentNullException()
+    {
+        Assert.Throws<ArgumentNullException>(() => ItemErrorPolicy.SkipAndDeadLetter((ChannelWriter<ItemErrorContext>)null!));
+    }
+
+
+
+    [Fact]
+    public void SkipAndDeadLetter_channel_writes_the_failure_and_returns_Skip()
+    {
+        var channel = Channel.CreateUnbounded<ItemErrorContext>();
+        var context = Context();
+
+        var action = ItemErrorPolicy.SkipAndDeadLetter(channel.Writer)(context);
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.True(channel.Reader.TryRead(out var written));
+        Assert.Same(context, written);
+    }
+
+
+
+    [Fact]
+    public void SkipDeadLetterAndLog_collection_when_deadLetters_is_null_throws_ArgumentNullException()
+    {
+        Assert.Throws<ArgumentNullException>(() => ItemErrorPolicy.SkipDeadLetterAndLog((ICollection<ItemErrorContext>)null!, new RecordingLogger()));
+    }
+
+
+
+    [Fact]
+    public void SkipDeadLetterAndLog_collection_when_logger_is_null_throws_ArgumentNullException()
+    {
+        Assert.Throws<ArgumentNullException>(() => ItemErrorPolicy.SkipDeadLetterAndLog(new List<ItemErrorContext>(), null!));
+    }
+
+
+
+    [Fact]
+    public void SkipDeadLetterAndLog_collection_records_and_logs_and_returns_Skip()
+    {
+        var deadLetters = new List<ItemErrorContext>();
+        var logger = new RecordingLogger();
+        var context = Context();
+
+        var action = ItemErrorPolicy.SkipDeadLetterAndLog(deadLetters, logger)(context);
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.Same(context, Assert.Single(deadLetters));
+        Assert.Equal(1, logger.WarningCount);
+    }
+
+
+
+    [Fact]
+    public void SkipDeadLetterAndLog_channel_when_deadLetters_is_null_throws_ArgumentNullException()
+    {
+        Assert.Throws<ArgumentNullException>(() => ItemErrorPolicy.SkipDeadLetterAndLog((ChannelWriter<ItemErrorContext>)null!, new RecordingLogger()));
+    }
+
+
+
+    [Fact]
+    public void SkipDeadLetterAndLog_channel_when_logger_is_null_throws_ArgumentNullException()
+    {
+        var channel = Channel.CreateUnbounded<ItemErrorContext>();
+        Assert.Throws<ArgumentNullException>(() => ItemErrorPolicy.SkipDeadLetterAndLog(channel.Writer, null!));
+    }
+
+
+
+    [Fact]
+    public void SkipDeadLetterAndLog_channel_writes_and_logs_and_returns_Skip()
+    {
+        var channel = Channel.CreateUnbounded<ItemErrorContext>();
+        var logger = new RecordingLogger();
+        var context = Context();
+
+        var action = ItemErrorPolicy.SkipDeadLetterAndLog(channel.Writer, logger)(context);
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.True(channel.Reader.TryRead(out var written));
+        Assert.Same(context, written);
+        Assert.Equal(1, logger.WarningCount);
+    }
+
+
+
+    private sealed class RecordingLogger : ILogger
+    {
+        public int WarningCount { get; private set; }
+
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>
+        (
+            LogLevel logLevel,
+            EventId eventId,
+            TState state,
+            Exception? exception,
+            Func<TState, Exception?, string> formatter
+        )
+        {
+            if (logLevel == LogLevel.Warning)
+            {
+                WarningCount++;
+            }
+        }
+
+
+
+        private sealed class NullScope : IDisposable
+        {
+            public static readonly NullScope Instance = new();
+
+            public void Dispose()
+            {
+            }
+        }
+    }
+}
diff --git a/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/Wolfgang.Etl.ErrorPolicies.Tests.Unit.csproj b/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/Wolfgang.Etl.ErrorPolicies.Tests.Unit.csproj
new file mode 100644
index 00000000..49e1e79e
--- /dev/null
+++ b/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/Wolfgang.Etl.ErrorPolicies.Tests.Unit.csproj
@@ -0,0 +1,83 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFrameworks>net462;net472;net48;net481;netcoreapp3.1;net5.0;net6.0;net7.0;net8.0;net9.0;net10.0</TargetFrameworks>
+    <LangVersion>latest</LangVersion>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <IsPackable>false</IsPackable>
+    <IsTestProject>true</IsTestProject>
+    <SuppressTfmSupportBuildWarnings>true</SuppressTfmSupportBuildWarnings>
+  </PropertyGroup>
+
+  <!-- net462; net472; net48; net481 -->
+  <ItemGroup Condition="'$(TargetFramework)' == 'net462' OR '$(TargetFramework)' == 'net472' OR '$(TargetFramework)' == 'net48' OR '$(TargetFramework)' == 'net481'">
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.13.0" />
+    <PackageReference Include="xunit" Version="2.9.3" />
+    <PackageReference Include="xunit.runner.console" Version="2.9.3">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2" allowedVersions="(,3.0.0)">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <!-- netcoreapp3.1; net5.0 -->
+  <ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp3.1' OR '$(TargetFramework)' == 'net5.0'">
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.13.0" />
+    <PackageReference Include="xunit" Version="2.9.3" />
+    <PackageReference Include="xunit.runner.console" Version="2.9.3">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.5" allowedVersions="(,2.5.0)">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <!-- net6.0; net7.0 -->
+  <ItemGroup Condition="'$(TargetFramework)' == 'net6.0' OR '$(TargetFramework)' == 'net7.0'">
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.13.0" />
+    <PackageReference Include="xunit" Version="2.9.3" />
+    <PackageReference Include="xunit.runner.console" Version="2.9.3">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2" allowedVersions="(,3.0.0)">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <!-- net8.0; net9.0; net10.0 -->
+  <ItemGroup Condition="'$(TargetFramework)' == 'net8.0' OR '$(TargetFramework)' == 'net9.0' OR '$(TargetFramework)' == 'net10.0'">
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.3.0" />
+    <PackageReference Include="xunit" Version="2.9.3" />
+    <PackageReference Include="xunit.runner.console" Version="2.9.3">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+    <PackageReference Include="xunit.runner.visualstudio" Version="3.0.1">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <!-- coverlet: net5.0+ -->
+  <ItemGroup Condition="'$(TargetFramework)' == 'net5.0' OR '$(TargetFramework)' == 'net6.0' OR '$(TargetFramework)' == 'net7.0' OR '$(TargetFramework)' == 'net8.0' OR '$(TargetFramework)' == 'net9.0' OR '$(TargetFramework)' == 'net10.0'">
+    <PackageReference Include="coverlet.collector" Version="8.0.1">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Wolfgang.Etl.ErrorPolicies\Wolfgang.Etl.ErrorPolicies.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <Using Include="Xunit" />
+  </ItemGroup>
+
+</Project>

From 924059b443ebd07e2f0193baaa5a21cf2a84d73d Mon Sep 17 00:00:00 2001
From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com>
Date: Sun, 2 Aug 2026 21:49:13 -0400
Subject: [PATCH 3/6] Add ErrorPolicy delegate property to the base stages
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Makes ErrorPolicies' assign-a-policy design real. ExtractorBase / LoaderBase /
TransformerBase gain:

    public Func<ItemErrorContext, ItemErrorAction> ErrorPolicy { get; init; }

non-null, defaulting to a fail-fast AbortPolicy, throw-on-null. The base
OnItemError now delegates to it, so `ErrorPolicy = ItemErrorPolicy.SkipAndLog(...)`
works on any stage with no per-type property or override — while overriding
OnItemError is still available for stage-internal logic. Supersedes the earlier
"no public base property" decision; the rationale comment is rewritten and the
resume caveat (Skip = swallow-and-stop on formats that can't resume) documented.

- ErrorPolicy on the two/three-type-parameter bases (convenience bases inherit it)
- PublicAPI.Unshipped entries; +6 base tests (default fail-fast, assigned policy is
  used by the base OnItemError, null throws) across all three stages
- ItemErrorPolicy package docs + csproj description + CHANGELOG updated from the
  override-and-invoke wording back to the assignable `ErrorPolicy = ...` form

Full-matrix build clean (0/0); Abstractions PackageValidation passes vs 0.20.0
(additive); Abstractions unit suite 463 green, ErrorPolicies 14 green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  9 ++-
 .../ExtractorBase.cs                          | 46 ++++++++----
 src/Wolfgang.Etl.Abstractions/LoaderBase.cs   | 46 ++++++++----
 .../PublicAPI.Unshipped.txt                   |  6 ++
 .../TransformerBase.cs                        | 46 ++++++++----
 .../ItemErrorPolicy.cs                        | 22 +++---
 .../Wolfgang.Etl.ErrorPolicies.csproj         |  2 +-
 .../ItemErrorHandlingTests.cs                 | 71 +++++++++++++++++++
 8 files changed, 191 insertions(+), 57 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 958c1542..d0de9feb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,9 +15,14 @@ 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 to invoke from a stage's `OnItemError`
-  hook — `Skip`, `Abort`, `SkipAndLog(ILogger)`, and dead-letter families `SkipAndDeadLetter` /
+  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
diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
index 1876d3b7..98331855 100644
--- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
+++ b/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);
     }
 
 
diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
index 348fedf2..c56d26ff 100644
--- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
+++ b/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);
     }
 
 
diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
index 514883cd..bda37a15 100644
--- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
+++ b/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!
diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
index e6c86668..78579bae 100644
--- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
+++ b/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);
     }
 
 
diff --git a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
index 9c30de55..2f0b977a 100644
--- a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
+++ b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
@@ -7,17 +7,16 @@
 namespace Wolfgang.Etl.ErrorPolicies;
 
 /// <summary>
-/// Ready-made policies for an ETL stage's per-item error hook. Each is a
+/// Ready-made policies for an ETL stage's error handling. Each is a
 /// <see cref="Func{T, TResult}"/> from an <see cref="ItemErrorContext"/> to an
-/// <see cref="ItemErrorAction"/>, so it can be stored and invoked from a stage's overridden
-/// <c>OnItemError(ItemErrorContext)</c> (the <c>protected virtual</c> policy hook on
-/// <c>ExtractorBase</c> / <c>LoaderBase</c> / <c>TransformerBase</c>):
+/// <see cref="ItemErrorAction"/>, so it can be assigned directly to a stage's <c>ErrorPolicy</c>
+/// property (on <c>ExtractorBase</c> / <c>LoaderBase</c> / <c>TransformerBase</c>):
 /// <example><code>
-/// // in your extractor / loader / transformer:
-/// private readonly Func&lt;ItemErrorContext, ItemErrorAction&gt; _onError =
-///     ItemErrorPolicy.SkipDeadLetterAndLog(deadLetters, logger);
-///
-/// protected override ItemErrorAction OnItemError(ItemErrorContext context) =&gt; _onError(context);
+/// var deadLetters = new List&lt;ItemErrorContext&gt;();
+/// var extractor = new SomeExtractor&lt;Record&gt;(source)
+/// {
+///     ErrorPolicy = ItemErrorPolicy.SkipDeadLetterAndLog(deadLetters, logger)
+/// };
 /// </code></example>
 /// The dead-letter overloads write to a caller-owned sink (a collection or a channel), so its size —
 /// and therefore the memory a bad feed can consume — stays under the caller's control.
@@ -33,9 +32,8 @@ public static class ItemErrorPolicy
 
 
     /// <summary>
-    /// A policy that re-throws the failure and stops the run. Equivalent to not overriding
-    /// <c>OnItemError</c> (whose default returns <see cref="ItemErrorAction.Abort"/>); provided for
-    /// symmetry and explicitness.
+    /// A policy that re-throws the failure and stops the run. Equivalent to leaving a stage's
+    /// <c>ErrorPolicy</c> unset (which defaults to fail-fast); provided for symmetry and explicitness.
     /// </summary>
     public static Func<ItemErrorContext, ItemErrorAction> Abort { get; } = _ => ItemErrorAction.Abort;
 
diff --git a/src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj b/src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj
index 040fbb21..925eb3a0 100644
--- a/src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj
+++ b/src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj
@@ -11,7 +11,7 @@
          compare against. Enable it next cycle with PackageValidationBaselineVersion set to 0.21.0. -->
     <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
     <Title>$(AssemblyName)</Title>
-    <Description>Ready-made item-error policies — skip, log, and dead-letter (to a collection or a channel) — to invoke from the OnItemError hook of Wolfgang.Etl extractors, loaders, and transformers. Built on Wolfgang.Etl.Abstractions.</Description>
+    <Description>Ready-made item-error policies — skip, log, and dead-letter (to a collection or a channel) — assignable to the ErrorPolicy property of Wolfgang.Etl extractors, loaders, and transformers. Built on Wolfgang.Etl.Abstractions.</Description>
     <PackageProjectUrl>https://github.com/Chris-Wolfgang/ETL-Abstractions</PackageProjectUrl>
     <PackageReadmeFile>README.md</PackageReadmeFile>
     <RepositoryUrl>https://github.com/Chris-Wolfgang/ETL-Abstractions</RepositoryUrl>
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ItemErrorHandlingTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ItemErrorHandlingTests.cs
index 22ca8ac8..98b015c6 100644
--- a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ItemErrorHandlingTests.cs
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ItemErrorHandlingTests.cs
@@ -376,6 +376,77 @@ public async Task Transformer_CurrentErrorItemCount_resets_between_runs()
     }
 
 
+    // ---- ErrorPolicy property: the base OnItemError delegates to it (no override needed) ----
+
+    [Fact]
+    public void ErrorPolicy_defaults_to_fail_fast()
+    {
+        var sut = new DefaultPolicyExtractor();
+
+        Assert.Equal(ItemErrorAction.Abort, sut.ErrorPolicy(new ItemErrorContext(1, new Exception())));
+    }
+
+
+    [Fact]
+    public void ErrorPolicy_when_assigned_is_used_by_the_base_OnItemError()
+    {
+        var sut = new DefaultPolicyExtractor { ErrorPolicy = _ => ItemErrorAction.Skip };
+
+        var action = sut.Handle(new ItemErrorContext(1, new Exception()));
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.Equal(1, sut.CurrentErrorItemCount);
+    }
+
+
+    [Fact]
+    public void ErrorPolicy_when_assigned_null_throws()
+    {
+        var ex = Assert.Throws<ArgumentNullException>(() => new DefaultPolicyExtractor { ErrorPolicy = null! });
+        Assert.Equal("value", ex.ParamName);
+    }
+
+
+    [Fact]
+    public void Loader_ErrorPolicy_when_assigned_is_used_by_the_base_OnItemError()
+    {
+        var sut = new DefaultPolicyLoader { ErrorPolicy = _ => ItemErrorAction.Skip };
+
+        var action = sut.Handle(new ItemErrorContext(1, new Exception()));
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.Equal(1, sut.CurrentErrorItemCount);
+    }
+
+
+    [Fact]
+    public void Loader_ErrorPolicy_when_assigned_null_throws()
+    {
+        var ex = Assert.Throws<ArgumentNullException>(() => new DefaultPolicyLoader { ErrorPolicy = null! });
+        Assert.Equal("value", ex.ParamName);
+    }
+
+
+    [Fact]
+    public void Transformer_ErrorPolicy_when_assigned_is_used_by_the_base_OnItemError()
+    {
+        var sut = new DefaultPolicyTransformer { ErrorPolicy = _ => ItemErrorAction.Skip };
+
+        var action = sut.Handle(new ItemErrorContext(1, new Exception()));
+
+        Assert.Equal(ItemErrorAction.Skip, action);
+        Assert.Equal(1, sut.CurrentErrorItemCount);
+    }
+
+
+    [Fact]
+    public void Transformer_ErrorPolicy_when_assigned_null_throws()
+    {
+        var ex = Assert.Throws<ArgumentNullException>(() => new DefaultPolicyTransformer { ErrorPolicy = null! });
+        Assert.Equal("value", ex.ParamName);
+    }
+
+
     // ---- helpers / doubles ----
 
     private static async IAsyncEnumerable<int> AsyncSource(params int[] items)

From 3d506c03509211dccb5a555c13e28c666f3a3c00 Mon Sep 17 00:00:00 2001
From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com>
Date: Mon, 3 Aug 2026 08:04:11 -0400
Subject: [PATCH 4/6] Make full-channel dead-letter drops observable (#347)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The channel dead-letter policies use non-blocking TryWrite because the
OnItemError hook is synchronous. On a full bounded channel that drops the
failure record. SkipDeadLetterAndLog(ChannelWriter, ILogger) now logs a
distinct warning (EventId 2, ItemDeadLetterDropped) when the write is
dropped, so the loss is never silent — throwing would convert an error-sink
overflow into a full pipeline abort, which is worse. The logger-less
SkipAndDeadLetter(ChannelWriter) overload keeps the documented silent drop
(no logger to escalate to; caller controls overflow via BoundedChannelFullMode).

Adds a bounded-channel-full test and CHANGELOG note.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  4 ++-
 .../ItemErrorPolicy.cs                        | 15 ++++++++--
 .../ItemErrorPolicyLog.cs                     | 10 +++++++
 .../ItemErrorPolicyTests.cs                   | 30 +++++++++++++++++++
 4 files changed, 55 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d0de9feb..0b7e57c9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,7 +27,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `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`.
+  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
 
diff --git a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
index 2f0b977a..c497962f 100644
--- a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
+++ b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicy.cs
@@ -162,7 +162,9 @@ ILogger logger
     /// Writes the failure to <paramref name="deadLetters"/> with
     /// <see cref="ChannelWriter{T}.TryWrite(T)"/> and logs it as a warning through
     /// <paramref name="logger"/>, then discards the item and continues. See
-    /// <see cref="SkipAndDeadLetter(ChannelWriter{ItemErrorContext})"/> for the <c>TryWrite</c> caveat.
+    /// <see cref="SkipAndDeadLetter(ChannelWriter{ItemErrorContext})"/> for the <c>TryWrite</c> caveat —
+    /// but unlike that logger-less overload, when the write is dropped (a bounded channel that is full)
+    /// this policy logs a distinct warning so the lost failure record is never silent.
     /// </summary>
     /// <param name="deadLetters">The caller-owned channel each failed item is written to.</param>
     /// <param name="logger">The logger the returned policy writes each failure to.</param>
@@ -188,8 +190,15 @@ ILogger logger
 
         return context =>
         {
-            deadLetters.TryWrite(context);
-            ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception);
+            if (deadLetters.TryWrite(context))
+            {
+                ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception);
+            }
+            else
+            {
+                ItemErrorPolicyLog.ItemDeadLetterDropped(logger, context.ItemNumber, context.Exception);
+            }
+
             return ItemErrorAction.Skip;
         };
     }
diff --git a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs
index 2a881977..532a6e0f 100644
--- a/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs
+++ b/src/Wolfgang.Etl.ErrorPolicies/ItemErrorPolicyLog.cs
@@ -16,4 +16,14 @@ internal static class ItemErrorPolicyLog
             new EventId(1, nameof(ItemFailedAndSkipped)),
             "Item {ItemNumber} failed to process and was skipped."
         );
+
+
+
+    internal static readonly Action<ILogger, long, Exception?> ItemDeadLetterDropped =
+        LoggerMessage.Define<long>
+        (
+            LogLevel.Warning,
+            new EventId(2, nameof(ItemDeadLetterDropped)),
+            "Item {ItemNumber} failed and its dead-letter could not be recorded because the channel was full; the failure record was dropped."
+        );
 }
diff --git a/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs b/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs
index a7c2df04..2a7647df 100644
--- a/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs
+++ b/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs
@@ -158,6 +158,33 @@ public void SkipDeadLetterAndLog_channel_writes_and_logs_and_returns_Skip()
         Assert.True(channel.Reader.TryRead(out var written));
         Assert.Same(context, written);
         Assert.Equal(1, logger.WarningCount);
+        Assert.Equal(1, logger.LastEventId.Id);            // ItemFailedAndSkipped
+    }
+
+
+
+    [Fact]
+    public void SkipDeadLetterAndLog_channel_when_full_logs_the_dropped_write_and_returns_Skip()
+    {
+        // A bounded channel at capacity: TryWrite returns false, so the failure record is dropped.
+        var channel = Channel.CreateBounded<ItemErrorContext>
+        (
+            new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.Wait }
+        );
+        var prefilled = Context();
+        Assert.True(channel.Writer.TryWrite(prefilled));   // fill it to capacity
+
+        var logger = new RecordingLogger();
+        var context = Context();
+
+        var action = ItemErrorPolicy.SkipDeadLetterAndLog(channel.Writer, logger)(context);
+
+        Assert.Equal(ItemErrorAction.Skip, action);        // still skips — a full sink never aborts the run
+        Assert.True(channel.Reader.TryRead(out var only));
+        Assert.Same(prefilled, only);                      // the new failure was dropped, not enqueued
+        Assert.False(channel.Reader.TryRead(out _));       // nothing else in the channel
+        Assert.Equal(1, logger.WarningCount);              // the drop is logged, not silent
+        Assert.Equal(2, logger.LastEventId.Id);            // ItemDeadLetterDropped
     }
 
 
@@ -166,6 +193,8 @@ private sealed class RecordingLogger : ILogger
     {
         public int WarningCount { get; private set; }
 
+        public EventId LastEventId { get; private set; }
+
         public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
 
         public bool IsEnabled(LogLevel logLevel) => true;
@@ -182,6 +211,7 @@ public void Log<TState>
             if (logLevel == LogLevel.Warning)
             {
                 WarningCount++;
+                LastEventId = eventId;
             }
         }
 

From 6ae134d1b4b0187ab4dda0f7659c6f55855a97f5 Mon Sep 17 00:00:00 2001
From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com>
Date: Mon, 3 Aug 2026 08:23:49 -0400
Subject: [PATCH 5/6] Release prep 0.21.0: promote CHANGELOG + PublicAPI

- CHANGELOG: [Unreleased] -> [0.21.0] - 2026-08-03 (convenience bases #344,
  assignable ErrorPolicy on base stages, new Wolfgang.Etl.ErrorPolicies package);
  fresh [Unreleased] scaffold.
- PublicAPI: promote Unshipped -> Shipped for Wolfgang.Etl.Abstractions
  (convenience bases + ErrorPolicy) and Wolfgang.Etl.ErrorPolicies (8 entries).

Version stays 0.21.0; baseline stays 0.20.0 for the release build (bumped to
0.21.0 post-release). Full Release build clean (0/0); both packages pack with
PackageValidation passing vs 0.20.0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 CHANGELOG.md                                   | 18 ++++++++++++++++++
 .../PublicAPI.Shipped.txt                      | 15 +++++++++++++++
 .../PublicAPI.Unshipped.txt                    | 15 ---------------
 .../PublicAPI.Shipped.txt                      |  8 ++++++++
 .../PublicAPI.Unshipped.txt                    |  8 --------
 5 files changed, 41 insertions(+), 23 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0b7e57c9..60e2e3aa 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+### Changed
+
+### Deprecated
+
+### Removed
+
+### Fixed
+
+### 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
diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt
index 98ca92ef..81d85f74 100644
--- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt
+++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt
@@ -18,6 +18,8 @@ Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.CurrentItemCount.get
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.CurrentSkippedItemCount.get -> int
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.Dispose() -> void
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.Elapsed.get -> System.TimeSpan
+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.ExtractorBase<TSource, TProgress>.ExtractorBase() -> void
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.HandleItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.IncrementCurrentItemCount() -> void
@@ -29,6 +31,8 @@ Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.ReportingInterval.se
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.SkipItemCount.get -> int
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.SkipItemCount.set -> void
 Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.StartedAt.get -> System.DateTimeOffset?
+Wolfgang.Etl.Abstractions.ExtractorBase<TSource>
+Wolfgang.Etl.Abstractions.ExtractorBase<TSource>.ExtractorBase() -> void
 Wolfgang.Etl.Abstractions.IEtlPipeline<T>
 Wolfgang.Etl.Abstractions.IEtlPipeline<T>.AsAsyncEnumerable(System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<T>!
 Wolfgang.Etl.Abstractions.IEtlPipeline<T>.Through<TOut>(System.Func<System.Collections.Generic.IAsyncEnumerable<T>!, System.Collections.Generic.IAsyncEnumerable<TOut>!>! stage) -> Wolfgang.Etl.Abstractions.IEtlPipeline<TOut>!
@@ -117,6 +121,8 @@ Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.CurrentItemCount.g
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.CurrentSkippedItemCount.get -> int
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.Dispose() -> void
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.Elapsed.get -> System.TimeSpan
+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.LoaderBase<TDestination, TProgress>.HandleItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.IncrementCurrentItemCount() -> void
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.IncrementCurrentSkippedItemCount() -> void
@@ -128,6 +134,8 @@ Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.ReportingInterval.
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.SkipItemCount.get -> int
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.SkipItemCount.set -> void
 Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.StartedAt.get -> System.DateTimeOffset?
+Wolfgang.Etl.Abstractions.LoaderBase<TDestination>
+Wolfgang.Etl.Abstractions.LoaderBase<TDestination>.LoaderBase() -> void
 Wolfgang.Etl.Abstractions.MiddlewareExtensions
 Wolfgang.Etl.Abstractions.MiddlewareResult
 Wolfgang.Etl.Abstractions.MiddlewareResult<T>
@@ -155,6 +163,8 @@ Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.Curr
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.CurrentSkippedItemCount.get -> int
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.Dispose() -> void
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.Elapsed.get -> System.TimeSpan
+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
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.HandleItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.IncrementCurrentItemCount() -> void
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.IncrementCurrentSkippedItemCount() -> void
@@ -166,14 +176,19 @@ Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.Skip
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.SkipItemCount.set -> void
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.StartedAt.get -> System.DateTimeOffset?
 Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.TransformerBase() -> void
+Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>
+Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>.TransformerBase() -> void
 abstract Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.CreateProgressReport() -> TProgress
 abstract Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.ExtractWorkerAsync(System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable<TSource>!
 abstract Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.CreateProgressReport() -> TProgress
 abstract Wolfgang.Etl.Abstractions.LoaderBase<TDestination, TProgress>.LoadWorkerAsync(System.Collections.Generic.IAsyncEnumerable<TDestination>! items, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task!
 abstract Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.CreateProgressReport() -> TProgress
 abstract Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination, TProgress>.TransformWorkerAsync(System.Collections.Generic.IAsyncEnumerable<TSource>! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable<TDestination>!
+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.MiddlewareResult<T>.Equals(object? obj) -> bool
 override Wolfgang.Etl.Abstractions.MiddlewareResult<T>.GetHashCode() -> int
+override Wolfgang.Etl.Abstractions.TransformerBase<TSource, TDestination>.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report!
 static Wolfgang.Etl.Abstractions.EtlPipeline.Create() -> Wolfgang.Etl.Abstractions.EtlPipeline!
 static Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions.DisposingOwned(this Wolfgang.Etl.Abstractions.IEtlPipelineSink! sink, params object![]! ownedResources) -> Wolfgang.Etl.Abstractions.IEtlPipelineSink!
 static Wolfgang.Etl.Abstractions.EtlPipelineSourceExtensions.From<T, TProgress>(this Wolfgang.Etl.Abstractions.EtlPipeline! pipeline, Wolfgang.Etl.Abstractions.ExtractorBase<T, TProgress>! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline<T>!
diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
index bda37a15..7dc5c581 100644
--- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
+++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt
@@ -1,16 +1 @@
 #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!
diff --git a/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt
index 7dc5c581..8af9f268 100644
--- a/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt
+++ b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Shipped.txt
@@ -1 +1,9 @@
 #nullable enable
+Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Abort.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Skip.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndLog(Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
+static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
diff --git a/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt
index 8af9f268..7dc5c581 100644
--- a/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt
+++ b/src/Wolfgang.Etl.ErrorPolicies/PublicAPI.Unshipped.txt
@@ -1,9 +1 @@
 #nullable enable
-Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy
-static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Abort.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
-static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Skip.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
-static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
-static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
-static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndLog(Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
-static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!
-static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>!

From 7e34b02008f14e2cc1e1d26083880b9d8850305c Mon Sep 17 00:00:00 2001
From: Chris Wolfgang <210299580+Chris-Wolfgang@users.noreply.github.com>
Date: Mon, 3 Aug 2026 11:08:04 -0400
Subject: [PATCH 6/6] Document ErrorPolicy as intentionally init-only (0.21.0)

Per the design decision, ErrorPolicy stays init-only for 0.21.0 (settable PR
#350 closed). Add a doc note on all three base stages that the policy is
assigned once at construction and cannot change during a run, and correct the
CHANGELOG wording from "settable" to "init-only". This satisfies the Copilot
review's alternative ("document that immutability is intentional").

The broader move of the whole base config surface (MaximumItemCount /
SkipItemCount / ReportingInterval + ErrorPolicy) to init-only is a coordinated
pre-1.0 change tracked for a future release, here and downstream.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 CHANGELOG.md                                     | 9 +++++----
 src/Wolfgang.Etl.Abstractions/ExtractorBase.cs   | 1 +
 src/Wolfgang.Etl.Abstractions/LoaderBase.cs      | 1 +
 src/Wolfgang.Etl.Abstractions/TransformerBase.cs | 1 +
 4 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 60e2e3aa..6f54f8af 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -34,10 +34,11 @@ breaking change (validates against the 0.20.0 baseline).
   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.
+  `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` /
diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
index 98331855..5dbccf69 100644
--- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
+++ b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs
@@ -488,6 +488,7 @@ protected void IncrementCurrentSkippedItemCount()
     /// 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
diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
index c56d26ff..16cd0360 100644
--- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
+++ b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs
@@ -486,6 +486,7 @@ protected void IncrementCurrentSkippedItemCount()
     /// 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
diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
index 78579bae..95e4a1ef 100644
--- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
+++ b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs
@@ -497,6 +497,7 @@ protected void IncrementCurrentSkippedItemCount()
     /// 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