diff --git a/CHANGELOG.md b/CHANGELOG.md index 820acc3..0b7e57c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 implements only its worker method — no progress record and no `CreateProgressReport` override. Override it to enrich the report (for example a known total). The existing two/three-type-parameter bases are unchanged. Additive. +- **`ErrorPolicy` on the base stages (#344 follow-up):** `ExtractorBase`, `LoaderBase`, and + `TransformerBase` gained a settable `ErrorPolicy` (`Func`, + non-null, default fail-fast) that the base `OnItemError` consults, so a stage gets a configurable + error policy with no per-type property or override — assign one (e.g. from `Wolfgang.Etl.ErrorPolicies`) + or override `OnItemError` for stage-internal logic. Additive; unset behaviour is unchanged fail-fast. +- **`Wolfgang.Etl.ErrorPolicies` package (new, lockstep-versioned with `Wolfgang.Etl.Abstractions`):** + a static `ItemErrorPolicy` factory of ready-made policies assignable to a stage's `ErrorPolicy` + property — `Skip`, `Abort`, `SkipAndLog(ILogger)`, and dead-letter families `SkipAndDeadLetter` / + `SkipDeadLetterAndLog`, each overloaded for a caller-owned `ICollection` 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 diff --git a/ETL-Abstractions.sln b/ETL-Abstractions.sln index 8a3a782..c4f8d13 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 1876d3b..9833185 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 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. + /// + /// 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.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs index 348fedf..c56d26f 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 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. + /// + /// 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/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt index 514883c..bda37a1 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 Wolfgang.Etl.Abstractions.ExtractorBase.ExtractorBase() -> void +Wolfgang.Etl.Abstractions.ExtractorBase.ErrorPolicy.get -> System.Func! +Wolfgang.Etl.Abstractions.ExtractorBase.ErrorPolicy.init -> void Wolfgang.Etl.Abstractions.LoaderBase Wolfgang.Etl.Abstractions.LoaderBase.LoaderBase() -> void +Wolfgang.Etl.Abstractions.LoaderBase.ErrorPolicy.get -> System.Func! +Wolfgang.Etl.Abstractions.LoaderBase.ErrorPolicy.init -> void Wolfgang.Etl.Abstractions.TransformerBase Wolfgang.Etl.Abstractions.TransformerBase.TransformerBase() -> void +Wolfgang.Etl.Abstractions.TransformerBase.ErrorPolicy.get -> System.Func! +Wolfgang.Etl.Abstractions.TransformerBase.ErrorPolicy.init -> void override Wolfgang.Etl.Abstractions.ExtractorBase.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report! override Wolfgang.Etl.Abstractions.LoaderBase.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report! override Wolfgang.Etl.Abstractions.TransformerBase.CreateProgressReport() -> Wolfgang.Etl.Abstractions.Report! diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs index e6c8666..78579ba 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 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. + /// + /// 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/Wolfgang.Etl.Abstractions.csproj b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj index ccd92fa..bd8b673 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/ItemErrorHandlingTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/ItemErrorHandlingTests.cs index 22ca8ac..98b015c 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 0000000..2a7647d --- /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 0000000..49e1e79 --- /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 + + + + + + + + + + + +