diff --git a/CHANGELOG.md b/CHANGELOG.md index 4920f5a2..e94c45fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`IReportsItemErrors` (#335):** a small interface (`int CurrentErrorItemCount { get; }`) implemented + by `ExtractorBase`, `LoaderBase`, and `TransformerBase`, letting a pipeline read any stage's + error-item count uniformly regardless of concrete type. - **Middleware / interceptor (#93):** a composable per-item hook — `IItemMiddleware` returning `MiddlewareResult` (`Continue` to keep/replace an item, `Drop` to remove it) — attached to any stream with the `WithMiddleware(...)` extensions (single or ordered chain). Lets cross-cutting @@ -24,6 +27,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **`EtlPipelineProgress.ErrorItemCount` now aggregates every stage (#335).** Previously it reported + only the extractor's error-item count; it now sums the error-item counts of the source, every + transformer, and the loader (each stage that implements `IReportsItemErrors`), so an item any + stage's error policy discarded is reflected in the total. Pre-1.0 behaviour change. - **Breaking (#285):** `EtlPipelineProgress`'s counters — `ExtractedItemCount`, `LoadedItemCount`, and `ErrorItemCount` — are now `long` instead of `int`, so a long-running pipeline can report more than `int.MaxValue` (~2.1 billion) records without overflow. This changes the record's getters, positional diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs index aceca4a7..53dc8dc8 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineImpl.cs @@ -46,7 +46,11 @@ public IEtlPipeline Through(ITransformAsync transformer) throw new ArgumentNullException(nameof(transformer)); } - return new EtlPipelineImpl((state, token) => transformer.TransformAsync(_factory(state, token))); + return new EtlPipelineImpl((state, token) => + { + RegisterErrorReader(transformer, state); + return transformer.TransformAsync(_factory(state, token)); + }); } @@ -59,7 +63,11 @@ public IEtlPipeline Through(ITransformWithCancellationAsync throw new ArgumentNullException(nameof(transformer)); } - return new EtlPipelineImpl((state, token) => transformer.TransformAsync(_factory(state, token), token)); + return new EtlPipelineImpl((state, token) => + { + RegisterErrorReader(transformer, state); + return transformer.TransformAsync(_factory(state, token), token); + }); } @@ -109,6 +117,18 @@ public IAsyncEnumerable AsAsyncEnumerable(CancellationToken token = default) } + // Registers a transformer stage's error-item count with the run so + // EtlPipelineProgress.ErrorItemCount sums it alongside the source and sink. A transformer that + // does not report errors (for example a plain delegate transform) is skipped. + private static void RegisterErrorReader(object transformer, EtlRunState state) + { + if (transformer is IReportsItemErrors reporter) + { + state.AddErrorCountReader(() => reporter.CurrentErrorItemCount); + } + } + + // The head of every pipeline: pulls from the raw source, honours cancellation, and counts each // record as extracted. WithCancellation covers sources that observe the token via // [EnumeratorCancellation]; the explicit throw covers sources that ignore it. diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs index 25675a23..c806c9d7 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSink.cs @@ -35,6 +35,11 @@ LoaderBase loader public async Task RunAsync(IProgress? progress = null, CancellationToken token = default) { var state = new EtlRunState(); + + // Surface the loader's error-item count into the snapshot too, summed with the source and any + // transformer error counts so the pipeline total reflects every stage, not just the source. + state.AddErrorCountReader(() => _loader.CurrentErrorItemCount); + var stream = CountLoaded(_factory(state, token), state, progress, token); await _loader.LoadAsync(stream, token).ConfigureAwait(false); diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs index 7b36f8a3..b4ea611d 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSourceExtensions.cs @@ -56,7 +56,8 @@ public static IEtlPipeline From(this EtlPipeline pipeline, Extr { // Surface the extractor's error-item count into the pipeline snapshot so a bad record // the extractor's error policy discarded is reported, not silently absent from the totals. - state.ErrorCountReader = () => extractor.CurrentErrorItemCount; + // Summed with any transformer/loader error counts (see EtlRunState.AddErrorCountReader). + state.AddErrorCountReader(() => extractor.CurrentErrorItemCount); return extractor.ExtractAsync(token); }); } diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs index 9c8c7a42..c07c4bde 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Diagnostics; @@ -18,16 +19,27 @@ internal sealed class EtlRunState public long LoadedItemCount; - // Optional reader that surfaces an error-reporting source's error-item count into the snapshot. - // Left null for sources that don't report errors (e.g. a raw IAsyncEnumerable), which reads as 0. - public Func? ErrorCountReader; + // Error-item-count readers, one per stage that reports errors (source, transformers, sink). Their + // values are summed into the snapshot so an item any stage's error policy discarded is reported, + // not just the source's. Empty for a pipeline of stages that don't report errors, which reads as 0. + private readonly List> _errorCountReaders = new(); + + + // Registers a stage's error-item-count reader. Called once per stage as the factory chain runs. + public void AddErrorCountReader(Func reader) => _errorCountReaders.Add(reader); public EtlPipelineProgress Snapshot() { + long errorItemCount = 0; + foreach (var reader in _errorCountReaders) + { + errorItemCount += reader(); + } + return new EtlPipelineProgress(ExtractedItemCount, LoadedItemCount, _stopwatch.Elapsed) { - ErrorItemCount = ErrorCountReader?.Invoke() ?? 0, + ErrorItemCount = errorItemCount, }; } } diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs index 66f08364..7fb1b76e 100644 --- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs +++ b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs @@ -19,6 +19,7 @@ namespace Wolfgang.Etl.Abstractions; /// The type of the progress object public abstract class ExtractorBase : IExtractWithProgressAndCancellationAsync, + IReportsItemErrors, IAsyncDisposable, IDisposable where TSource : notnull diff --git a/src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs b/src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs new file mode 100644 index 00000000..3b160c2d --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/IReportsItemErrors.cs @@ -0,0 +1,17 @@ +namespace Wolfgang.Etl.Abstractions; + +/// +/// Implemented by an ETL stage that counts items its error policy discarded, so a pipeline can read +/// that count uniformly regardless of the stage's concrete type. , +/// , and +/// all implement it, and EtlPipeline sums across every stage +/// that reports it into . +/// +public interface IReportsItemErrors +{ + /// + /// The number of items this stage's error policy has discarded (OnItemError returned + /// ) so far in the current run. + /// + int CurrentErrorItemCount { get; } +} diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs index ca2c40e5..3b1e8ad6 100644 --- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs +++ b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs @@ -18,6 +18,7 @@ namespace Wolfgang.Etl.Abstractions; /// The type of the progress object public abstract class LoaderBase : ILoadWithProgressAndCancellationAsync, + IReportsItemErrors, IAsyncDisposable, IDisposable where TDestination : notnull diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt index e53b6f35..77924326 100644 --- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt +++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt @@ -20,3 +20,5 @@ static Wolfgang.Etl.Abstractions.MiddlewareResult.operator !=(Wolfgang.Etl.Ab Wolfgang.Etl.Abstractions.MiddlewareExtensions static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware(this System.Collections.Generic.IAsyncEnumerable! source, Wolfgang.Etl.Abstractions.IItemMiddleware! middleware, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! static Wolfgang.Etl.Abstractions.MiddlewareExtensions.WithMiddleware(this System.Collections.Generic.IAsyncEnumerable! source, System.Collections.Generic.IEnumerable!>! middlewares, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! +Wolfgang.Etl.Abstractions.IReportsItemErrors +Wolfgang.Etl.Abstractions.IReportsItemErrors.CurrentErrorItemCount.get -> int diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs index 1928435b..9c92e7e4 100644 --- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs +++ b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs @@ -20,6 +20,7 @@ namespace Wolfgang.Etl.Abstractions; /// The type of the progress object public abstract class TransformerBase : ITransformWithProgressAndCancellationAsync, + IReportsItemErrors, IAsyncDisposable, IDisposable where TSource : notnull diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/AggregateErrorsTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/AggregateErrorsTests.cs new file mode 100644 index 00000000..cf58f441 --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/AggregateErrorsTests.cs @@ -0,0 +1,251 @@ +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 Wolfgang.Etl.Abstractions.Tests.Unit.Models; +using Xunit; + +namespace Wolfgang.Etl.Abstractions.Tests.Unit.EtlPipelineTests; + +/// +/// Covers #335: aggregates the per-item error counts +/// of every stage that reports them (source, transformers, sink) via , +/// not just the extractor. +/// +public class AggregateErrorsTests +{ + [Fact] + public void All_three_base_classes_implement_IReportsItemErrors() + { + Assert.IsAssignableFrom(new ErroringExtractor(good: 0, errors: 0)); + Assert.IsAssignableFrom(new ErroringTransformer(errors: 0)); + Assert.IsAssignableFrom(new ErroringLoader(errors: 0)); + } + + + [Fact] + public async Task ErrorItemCount_sums_extractor_transformer_and_loader_errors() + { + var reports = new List(); + var extractor = new ErroringExtractor(good: 3, errors: 2); + var transformer = new ErroringTransformer(errors: 4); + var loader = new ErroringLoader(errors: 1); + + await EtlPipeline + .Create() + .From(extractor) + .Through(transformer) + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + var final = reports[^1]; + Assert.Equal(3, final.ExtractedItemCount); + Assert.Equal(3, final.LoadedItemCount); + Assert.Equal(7, final.ErrorItemCount); // 2 (extract) + 4 (transform) + 1 (load) + } + + + [Fact] + public async Task ErrorItemCount_counts_only_stages_that_report_errors() + { + // A delegate transform stage does not implement IReportsItemErrors, so it contributes nothing. + var reports = new List(); + var extractor = new ErroringExtractor(good: 2, errors: 5); + var loader = new ErroringLoader(errors: 3); + + await EtlPipeline + .Create() + .From(extractor) + .Through(s => s) + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + Assert.Equal(8, reports[^1].ErrorItemCount); // 5 (extract) + 3 (load); delegate adds 0 + } + + + [Fact] + public async Task ErrorItemCount_includes_a_plain_ITransformAsync_stage_that_reports_errors() + { + // Binds the non-cancellation Through overload (a bare ITransformAsync, not a TransformerBase). + var reports = new List(); + var extractor = new ErroringExtractor(good: 2, errors: 1); + var transform = new ReportingPassThroughTransform(errors: 5); + var loader = new ErroringLoader(errors: 0); + + await EtlPipeline + .Create() + .From(extractor) + .Through(transform) + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + Assert.Equal(6, reports[^1].ErrorItemCount); // 1 (extract) + 5 (plain transform) + } + + + [Fact] + public async Task ErrorItemCount_is_zero_when_no_stage_reports_errors() + { + var reports = new List(); + var loader = new ErroringLoader(errors: 0); + + await EtlPipeline + .Create() + .From(AsyncSource(1, 2, 3)) // raw IAsyncEnumerable — not an IReportsItemErrors stage + .To(loader) + .RunAsync(new SyncProgress(reports.Add)); + + Assert.Equal(0, reports[^1].ErrorItemCount); + } + + + // ---------- helpers ---------- + + private static async IAsyncEnumerable AsyncSource(params int[] items) + { + foreach (var item in items) + { + await Task.Yield(); + yield return item; + } + } + + + // A bare ITransformAsync (not a TransformerBase) that still reports an error count — exercises the + // non-cancellation Through overload's error-reader registration. + private sealed class ReportingPassThroughTransform : ITransformAsync, IReportsItemErrors + { + public ReportingPassThroughTransform(int errors) => CurrentErrorItemCount = errors; + + public int CurrentErrorItemCount { get; } + + public IAsyncEnumerable TransformAsync(IAsyncEnumerable items) => items; + } + + + private sealed class SyncProgress : IProgress + { + private readonly Action _report; + + public SyncProgress(Action report) => _report = report; + + public void Report(EtlPipelineProgress value) => _report(value); + } + + + // ---------- error-reporting doubles (route real #84 errors through HandleItemError) ---------- + + private sealed class ErroringExtractor : ExtractorBase + { + private readonly int _good; + private readonly int _errors; + + public ErroringExtractor(int good, int errors) + { + _good = good; + _errors = errors; + } + + protected override ItemErrorAction OnItemError(ItemErrorContext context) => ItemErrorAction.Skip; + + protected override async IAsyncEnumerable ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + for (var i = 0; i < _good; i++) + { + await Task.Yield(); + IncrementCurrentItemCount(); + yield return i; + } + + for (var e = 0; e < _errors; e++) + { + // Route a synthetic failure through the base #84 hook: with OnItemError => Skip, + // HandleItemError increments CurrentErrorItemCount and returns without rethrowing. + try + { + throw new InvalidOperationException("bad item"); + } + catch (InvalidOperationException ex) + { + HandleItemError(new ItemErrorContext(e, ex)); + } + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + private sealed class ErroringTransformer : TransformerBase + { + private readonly int _errors; + + public ErroringTransformer(int errors) => _errors = errors; + + protected override ItemErrorAction OnItemError(ItemErrorContext context) => ItemErrorAction.Skip; + + protected override async IAsyncEnumerable TransformWorkerAsync( + IAsyncEnumerable items, [EnumeratorCancellation] CancellationToken token) + { + await foreach (var item in items.WithCancellation(token)) + { + IncrementCurrentItemCount(); + yield return item; + } + + for (var e = 0; e < _errors; e++) + { + // Route a synthetic failure through the base #84 hook: with OnItemError => Skip, + // HandleItemError increments CurrentErrorItemCount and returns without rethrowing. + try + { + throw new InvalidOperationException("bad item"); + } + catch (InvalidOperationException ex) + { + HandleItemError(new ItemErrorContext(e, ex)); + } + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + private sealed class ErroringLoader : LoaderBase + { + private readonly int _errors; + + public ErroringLoader(int errors) => _errors = errors; + + protected override ItemErrorAction OnItemError(ItemErrorContext context) => ItemErrorAction.Skip; + + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + await foreach (var item in items.WithCancellation(token)) + { + IncrementCurrentItemCount(); + } + + for (var e = 0; e < _errors; e++) + { + // Route a synthetic failure through the base #84 hook: with OnItemError => Skip, + // HandleItemError increments CurrentErrorItemCount and returns without rethrowing. + try + { + throw new InvalidOperationException("bad item"); + } + catch (InvalidOperationException ex) + { + HandleItemError(new ItemErrorContext(e, ex)); + } + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } +}