diff --git a/CHANGELOG.md b/CHANGELOG.md index 86e35911..6f54f8af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +## [0.21.0] - 2026-08-03 + +Minor release: convenience base classes, an assignable per-item `ErrorPolicy` on the three base stages, +and a new companion package `Wolfgang.Etl.ErrorPolicies` of ready-made policies. Purely additive — no +breaking change (validates against the 0.20.0 baseline). + +### Added + +- **Convenience base classes (#344):** `ExtractorBase`, `LoaderBase`, and + `TransformerBase` fix the progress type to the built-in `Report` and supply a + default `CreateProgressReport()`, so a component that doesn't need a custom progress-report type + implements only its worker method — no progress record and no `CreateProgressReport` override. + Override it to enrich the report (for example a known total). The existing two/three-type-parameter + bases are unchanged. Additive. +- **`ErrorPolicy` on the base stages (#344 follow-up):** `ExtractorBase`, `LoaderBase`, and + `TransformerBase` gained an `ErrorPolicy` (`Func`, non-null, + default fail-fast, **init-only** — set at construction) that the base `OnItemError` consults, so a + stage gets a configurable error policy with no per-type property or override — assign one (e.g. from + `Wolfgang.Etl.ErrorPolicies`) or override `OnItemError` for stage-internal logic. Additive; unset + behaviour is unchanged fail-fast. +- **`Wolfgang.Etl.ErrorPolicies` package (new, lockstep-versioned with `Wolfgang.Etl.Abstractions`):** + a static `ItemErrorPolicy` factory of ready-made policies assignable to a stage's `ErrorPolicy` + property — `Skip`, `Abort`, `SkipAndLog(ILogger)`, and dead-letter families `SkipAndDeadLetter` / + `SkipDeadLetterAndLog`, each overloaded for a caller-owned `ICollection` or a + `System.Threading.Channels.ChannelWriter`. Ships from this repo so the shared + policy set is defined once for the whole ETL family; the core `Wolfgang.Etl.Abstractions` assembly + keeps its minimal dependency set — only this package takes `Microsoft.Extensions.Logging.Abstractions` + and `System.Threading.Channels`. The channel dead-letter overloads use the non-blocking `TryWrite` + (the hook is synchronous); `SkipDeadLetterAndLog(ChannelWriter, ILogger)` logs a distinct warning when + a full bounded channel drops the failure record, so the loss is never silent. + +### Changed + +### Deprecated + +### Removed + +### Fixed + +### Security + ## [0.20.0] - 2026-07-30 Minor release: a dependency-free retry seam, composable per-item middleware, and pipeline-wide 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/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs index 1876d3b7..5dbccf69 100644 --- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs +++ b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs @@ -476,30 +476,49 @@ protected void IncrementCurrentSkippedItemCount() + private static readonly Func AbortPolicy = + static _ => ItemErrorAction.Abort; + + private readonly Func _errorPolicy = AbortPolicy; + + + + /// + /// Gets the policy invoked when an item fails to process. Return + /// to discard the item and continue, or 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 Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy. + /// Assigned once, at construction (init-only), so the policy cannot change during a run. + /// + /// The assigned value is . + public Func ErrorPolicy + { + get => _errorPolicy; + init => _errorPolicy = value ?? throw new ArgumentNullException(nameof(value)); + } + + + /// - /// Decides what to do when an item fails to process. Override in a derived stage to record the - /// failure and return to discard the item and continue, or - /// to re-throw and stop the run. The base implementation - /// always returns , 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 + /// (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. /// /// /// Describes the failed item — its ordinal, the exception, and optional raw content. /// /// Whether to skip the item or abort the run. /// - /// This is the policy hook a derived stage overrides; a worker does not call it directly. A worker - /// calls , 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 try/catch, and only a format that - /// can genuinely resume after a bad record overrides this and surfaces its own public knob. + /// A worker calls , which invokes this method and performs the skip + /// bookkeeping. The worker still owns the try/catch — a C# async iterator cannot + /// resume after it throws — and calls from it. On a format that + /// cannot genuinely resume after a bad record, means "swallow + /// the failure and stop at that point" rather than "skip and continue"; such a stage documents + /// that on its own type. /// 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/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; + +/// +/// A convenience that reports progress with the +/// built-in type and supplies a default , so a +/// derived extractor only has to implement ExtractWorkerAsync. 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 CreateProgressReport boilerplate. Override +/// if you want to enrich the report (for example set a known total). +/// +/// The type of the object being extracted. +public abstract class ExtractorBase : ExtractorBase + where TSource : notnull +{ + /// + /// Builds a snapshot from the current item count and timing. Override to add + /// more detail (for example a known ). + /// + /// A for the current run. + protected override Report CreateProgressReport() => new(CurrentItemCount, StartedAt, Elapsed); +} diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs index 348fedf2..16cd0360 100644 --- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs +++ b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs @@ -474,30 +474,49 @@ protected void IncrementCurrentSkippedItemCount() + private static readonly Func AbortPolicy = + static _ => ItemErrorAction.Abort; + + private readonly Func _errorPolicy = AbortPolicy; + + + + /// + /// Gets the policy invoked when an item fails to process. Return + /// to discard the item and continue, or 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 Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy. + /// Assigned once, at construction (init-only), so the policy cannot change during a run. + /// + /// The assigned value is . + public Func ErrorPolicy + { + get => _errorPolicy; + init => _errorPolicy = value ?? throw new ArgumentNullException(nameof(value)); + } + + + /// - /// Decides what to do when an item fails to process. Override in a derived stage to record the - /// failure and return to discard the item and continue, or - /// to re-throw and stop the run. The base implementation - /// always returns , 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 + /// (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. /// /// /// Describes the failed item — its ordinal, the exception, and optional raw content. /// /// Whether to skip the item or abort the run. /// - /// This is the policy hook a derived stage overrides; a worker does not call it directly. A worker - /// calls , 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 try/catch, and only a format that - /// can genuinely resume after a bad record overrides this and surfaces its own public knob. + /// A worker calls , which invokes this method and performs the skip + /// bookkeeping. The worker still owns the try/catch — a C# async iterator cannot + /// resume after it throws — and calls from it. On a format that + /// cannot genuinely resume after a bad record, means "swallow + /// the failure and stop at that point" rather than "skip and continue"; such a stage documents + /// that on its own type. /// 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{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; + +/// +/// A convenience that reports progress with the +/// built-in type and supplies a default , so a +/// derived loader only has to implement LoadWorkerAsync. 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 CreateProgressReport boilerplate. Override +/// if you want to enrich the report (for example set a known total). +/// +/// The type of the object being loaded. +public abstract class LoaderBase : LoaderBase + where TDestination : notnull +{ + /// + /// Builds a snapshot from the current item count and timing. Override to add + /// more detail (for example a known ). + /// + /// A for the current run. + protected override Report CreateProgressReport() => new(CurrentItemCount, StartedAt, Elapsed); +} 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.CurrentItemCount.get Wolfgang.Etl.Abstractions.ExtractorBase.CurrentSkippedItemCount.get -> int Wolfgang.Etl.Abstractions.ExtractorBase.Dispose() -> void Wolfgang.Etl.Abstractions.ExtractorBase.Elapsed.get -> System.TimeSpan +Wolfgang.Etl.Abstractions.ExtractorBase.ErrorPolicy.get -> System.Func! +Wolfgang.Etl.Abstractions.ExtractorBase.ErrorPolicy.init -> void Wolfgang.Etl.Abstractions.ExtractorBase.ExtractorBase() -> void Wolfgang.Etl.Abstractions.ExtractorBase.HandleItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction Wolfgang.Etl.Abstractions.ExtractorBase.IncrementCurrentItemCount() -> void @@ -29,6 +31,8 @@ Wolfgang.Etl.Abstractions.ExtractorBase.ReportingInterval.se Wolfgang.Etl.Abstractions.ExtractorBase.SkipItemCount.get -> int Wolfgang.Etl.Abstractions.ExtractorBase.SkipItemCount.set -> void Wolfgang.Etl.Abstractions.ExtractorBase.StartedAt.get -> System.DateTimeOffset? +Wolfgang.Etl.Abstractions.ExtractorBase +Wolfgang.Etl.Abstractions.ExtractorBase.ExtractorBase() -> void Wolfgang.Etl.Abstractions.IEtlPipeline Wolfgang.Etl.Abstractions.IEtlPipeline.AsAsyncEnumerable(System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! Wolfgang.Etl.Abstractions.IEtlPipeline.Through(System.Func!, System.Collections.Generic.IAsyncEnumerable!>! stage) -> Wolfgang.Etl.Abstractions.IEtlPipeline! @@ -117,6 +121,8 @@ Wolfgang.Etl.Abstractions.LoaderBase.CurrentItemCount.g Wolfgang.Etl.Abstractions.LoaderBase.CurrentSkippedItemCount.get -> int Wolfgang.Etl.Abstractions.LoaderBase.Dispose() -> void Wolfgang.Etl.Abstractions.LoaderBase.Elapsed.get -> System.TimeSpan +Wolfgang.Etl.Abstractions.LoaderBase.ErrorPolicy.get -> System.Func! +Wolfgang.Etl.Abstractions.LoaderBase.ErrorPolicy.init -> void Wolfgang.Etl.Abstractions.LoaderBase.HandleItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction Wolfgang.Etl.Abstractions.LoaderBase.IncrementCurrentItemCount() -> void Wolfgang.Etl.Abstractions.LoaderBase.IncrementCurrentSkippedItemCount() -> void @@ -128,6 +134,8 @@ Wolfgang.Etl.Abstractions.LoaderBase.ReportingInterval. Wolfgang.Etl.Abstractions.LoaderBase.SkipItemCount.get -> int Wolfgang.Etl.Abstractions.LoaderBase.SkipItemCount.set -> void Wolfgang.Etl.Abstractions.LoaderBase.StartedAt.get -> System.DateTimeOffset? +Wolfgang.Etl.Abstractions.LoaderBase +Wolfgang.Etl.Abstractions.LoaderBase.LoaderBase() -> void Wolfgang.Etl.Abstractions.MiddlewareExtensions Wolfgang.Etl.Abstractions.MiddlewareResult Wolfgang.Etl.Abstractions.MiddlewareResult @@ -155,6 +163,8 @@ Wolfgang.Etl.Abstractions.TransformerBase.Curr Wolfgang.Etl.Abstractions.TransformerBase.CurrentSkippedItemCount.get -> int Wolfgang.Etl.Abstractions.TransformerBase.Dispose() -> void Wolfgang.Etl.Abstractions.TransformerBase.Elapsed.get -> System.TimeSpan +Wolfgang.Etl.Abstractions.TransformerBase.ErrorPolicy.get -> System.Func! +Wolfgang.Etl.Abstractions.TransformerBase.ErrorPolicy.init -> void Wolfgang.Etl.Abstractions.TransformerBase.HandleItemError(Wolfgang.Etl.Abstractions.ItemErrorContext! context) -> Wolfgang.Etl.Abstractions.ItemErrorAction Wolfgang.Etl.Abstractions.TransformerBase.IncrementCurrentItemCount() -> void Wolfgang.Etl.Abstractions.TransformerBase.IncrementCurrentSkippedItemCount() -> void @@ -166,14 +176,19 @@ Wolfgang.Etl.Abstractions.TransformerBase.Skip Wolfgang.Etl.Abstractions.TransformerBase.SkipItemCount.set -> void Wolfgang.Etl.Abstractions.TransformerBase.StartedAt.get -> System.DateTimeOffset? Wolfgang.Etl.Abstractions.TransformerBase.TransformerBase() -> void +Wolfgang.Etl.Abstractions.TransformerBase +Wolfgang.Etl.Abstractions.TransformerBase.TransformerBase() -> void abstract Wolfgang.Etl.Abstractions.ExtractorBase.CreateProgressReport() -> TProgress abstract Wolfgang.Etl.Abstractions.ExtractorBase.ExtractWorkerAsync(System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! abstract Wolfgang.Etl.Abstractions.LoaderBase.CreateProgressReport() -> TProgress abstract Wolfgang.Etl.Abstractions.LoaderBase.LoadWorkerAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Threading.Tasks.Task! abstract Wolfgang.Etl.Abstractions.TransformerBase.CreateProgressReport() -> TProgress abstract Wolfgang.Etl.Abstractions.TransformerBase.TransformWorkerAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable! +override Wolfgang.Etl.Abstractions.ExtractorBase.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report! +override Wolfgang.Etl.Abstractions.LoaderBase.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report! override Wolfgang.Etl.Abstractions.MiddlewareResult.Equals(object? obj) -> bool override Wolfgang.Etl.Abstractions.MiddlewareResult.GetHashCode() -> int +override Wolfgang.Etl.Abstractions.TransformerBase.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(this Wolfgang.Etl.Abstractions.EtlPipeline! pipeline, Wolfgang.Etl.Abstractions.ExtractorBase! extractor) -> Wolfgang.Etl.Abstractions.IEtlPipeline! diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs index e6c86668..95e4a1ef 100644 --- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs +++ b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs @@ -485,30 +485,49 @@ protected void IncrementCurrentSkippedItemCount() + private static readonly Func AbortPolicy = + static _ => ItemErrorAction.Abort; + + private readonly Func _errorPolicy = AbortPolicy; + + + + /// + /// Gets the policy invoked when an item fails to process. Return + /// to discard the item and continue, or 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 Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy. + /// Assigned once, at construction (init-only), so the policy cannot change during a run. + /// + /// The assigned value is . + public Func ErrorPolicy + { + get => _errorPolicy; + init => _errorPolicy = value ?? throw new ArgumentNullException(nameof(value)); + } + + + /// - /// Decides what to do when an item fails to process. Override in a derived stage to record the - /// failure and return to discard the item and continue, or - /// to re-throw and stop the run. The base implementation - /// always returns , 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 + /// (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. /// /// /// Describes the failed item — its ordinal, the exception, and optional raw content. /// /// Whether to skip the item or abort the run. /// - /// This is the policy hook a derived stage overrides; a worker does not call it directly. A worker - /// calls , 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 try/catch, and only a format that - /// can genuinely resume after a bad record overrides this and surfaces its own public knob. + /// A worker calls , which invokes this method and performs the skip + /// bookkeeping. The worker still owns the try/catch — a C# async iterator cannot + /// resume after it throws — and calls from it. On a format that + /// cannot genuinely resume after a bad record, means "swallow + /// the failure and stop at that point" rather than "skip and continue"; such a stage documents + /// that on its own type. /// 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/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; + +/// +/// A convenience that reports progress +/// with the built-in type and supplies a default , +/// so a derived transformer only has to implement TransformWorkerAsync. 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 CreateProgressReport boilerplate. Override +/// if you want to enrich the report (for example set a known total). +/// +/// The type of the source object. +/// The type of the destination object. +public abstract class TransformerBase : TransformerBase + where TSource : notnull + where TDestination : notnull +{ + /// + /// Builds a snapshot from the current item count and timing. Override to add + /// more detail (for example a known ). + /// + /// A for the current run. + protected override Report CreateProgressReport() => new(CurrentItemCount, StartedAt, Elapsed); +} 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 @@ net462;net472;net48;net481;netstandard2.0;net5.0;net6.0;net7.0;net8.0;net9.0;net10.0 latest - 0.20.0 + 0.21.0 + 0.21.0 + 1.0.0.0 + $([System.Text.RegularExpressions.Regex]::Replace("$(Version)", "[-+].*$", "")).0 + + False + $(AssemblyName) + 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. + https://github.com/Chris-Wolfgang/ETL-Abstractions + README.md + https://github.com/Chris-Wolfgang/ETL-Abstractions + MIT + True + ETL-Abstractions.png + True + False + + true + ETL;Extract-Transform-Load;error-handling;dead-letter + + + + + + + + + True + \ + + + True + \ + + + + + + + + + + + + + + + + + + + + + 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; + +/// +/// Covers the #344 convenience base classes — , +/// , and — +/// which fix the progress type to the built-in and supply a default +/// CreateProgressReport(), so a derived component only implements its worker method. +/// +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 AsyncSource(params int[] items) + { + foreach (var item in items) + { + await Task.Yield(); + yield return item; + } + } + + + private static async Task> Drain(IAsyncEnumerable source) + { + var result = new List(); + 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 + { + public Report Report() => CreateProgressReport(); + + protected override async IAsyncEnumerable ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + foreach (var i in new[] { 1, 2, 3 }) + { + await Task.Yield(); + IncrementCurrentItemCount(); + yield return i; + } + } + } + + + private sealed class ConvenienceLoader : LoaderBase + { + public List Loaded { get; } = new(); + + public Report Report() => CreateProgressReport(); + + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + await foreach (var item in items.WithCancellation(token).ConfigureAwait(false)) + { + Loaded.Add(item); + IncrementCurrentItemCount(); + } + } + } + + + private sealed class ConvenienceTransformer : TransformerBase + { + public Report Report() => CreateProgressReport(); + + protected override async IAsyncEnumerable TransformWorkerAsync( + IAsyncEnumerable 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 + { + 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 ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + await Task.Yield(); + IncrementCurrentItemCount(); + yield return 1; + } + } +} 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(() => 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(() => 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(() => new DefaultPolicyTransformer { ErrorPolicy = null! }); + Assert.Equal("value", ex.ParamName); + } + + // ---- helpers / doubles ---- private static async IAsyncEnumerable AsyncSource(params int[] items) 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..2a7647df --- /dev/null +++ b/tests/Wolfgang.Etl.ErrorPolicies.Tests.Unit/ItemErrorPolicyTests.cs @@ -0,0 +1,229 @@ +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(() => 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(() => ItemErrorPolicy.SkipAndDeadLetter((ICollection)null!)); + } + + + + [Fact] + public void SkipAndDeadLetter_collection_records_the_failure_and_returns_Skip() + { + var deadLetters = new List(); + 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(() => ItemErrorPolicy.SkipAndDeadLetter((ChannelWriter)null!)); + } + + + + [Fact] + public void SkipAndDeadLetter_channel_writes_the_failure_and_returns_Skip() + { + var channel = Channel.CreateUnbounded(); + 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(() => ItemErrorPolicy.SkipDeadLetterAndLog((ICollection)null!, new RecordingLogger())); + } + + + + [Fact] + public void SkipDeadLetterAndLog_collection_when_logger_is_null_throws_ArgumentNullException() + { + Assert.Throws(() => ItemErrorPolicy.SkipDeadLetterAndLog(new List(), null!)); + } + + + + [Fact] + public void SkipDeadLetterAndLog_collection_records_and_logs_and_returns_Skip() + { + var deadLetters = new List(); + 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(() => ItemErrorPolicy.SkipDeadLetterAndLog((ChannelWriter)null!, new RecordingLogger())); + } + + + + [Fact] + public void SkipDeadLetterAndLog_channel_when_logger_is_null_throws_ArgumentNullException() + { + var channel = Channel.CreateUnbounded(); + Assert.Throws(() => ItemErrorPolicy.SkipDeadLetterAndLog(channel.Writer, null!)); + } + + + + [Fact] + public void SkipDeadLetterAndLog_channel_writes_and_logs_and_returns_Skip() + { + var channel = Channel.CreateUnbounded(); + 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); + 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 + ( + 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 + } + + + + private sealed class RecordingLogger : ILogger + { + public int WarningCount { get; private set; } + + public EventId LastEventId { get; private set; } + + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log + ( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter + ) + { + if (logLevel == LogLevel.Warning) + { + WarningCount++; + LastEventId = eventId; + } + } + + + + 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 @@ + + + net462;net472;net48;net481;netcoreapp3.1;net5.0;net6.0;net7.0;net8.0;net9.0;net10.0 + latest + enable + false + true + true + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + + +