Release v0.19.0 — overflow-safe progress counters

CHANGELOG.md

@@ -19,6 +19,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+## [0.19.0] - 2026-07-29
+
+Minor release: makes the `EtlPipelineProgress` record counters overflow-safe.
+
+### Changed
+
+- **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
+  constructor, and `Deconstruct` from `int` to `long`; Package Validation against the 0.18.1 baseline
+  waives the change via `CompatibilitySuppressions.xml`. `AssemblyVersion` stays pinned at `1.0.0.0`.
+  Pre-1.0.
+
 ## [0.18.1] - 2026-07-27
 
 Patch release. Purely additive — no breaking change.

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineProgress.cs

@@ -21,8 +21,8 @@ namespace Wolfgang.Etl.Abstractions;
 /// <param name="Elapsed">The wall-clock time elapsed since the run started.</param>
 public sealed record EtlPipelineProgress
 (
-    int ExtractedItemCount,
-    int LoadedItemCount,
+    long ExtractedItemCount,
+    long LoadedItemCount,
     TimeSpan Elapsed
 )
 {
@@ -35,5 +35,5 @@ TimeSpan Elapsed
     /// successfully flowed into the pipeline, so a failed record is never silent. Distinct from
     /// intentional skips (an extractor's <c>SkipItemCount</c> budget), which are not surfaced here.
     /// </summary>
-    public int ErrorItemCount { get; init; }
+    public long ErrorItemCount { get; init; }
 }

src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Diagnostics;
 
 
 namespace Wolfgang.Etl.Abstractions;
@@ -12,20 +11,45 @@ namespace Wolfgang.Etl.Abstractions;
 /// </summary>
 internal sealed class EtlRunState
 {
-    private readonly Stopwatch _stopwatch = Stopwatch.StartNew();
+    private readonly ITimeSource _timeSource;
+    private readonly long _startTimestamp;
 
-    public int ExtractedItemCount;
+    public long ExtractedItemCount;
 
-    public int LoadedItemCount;
+    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<int>? ErrorCountReader;
 
 
+    public EtlRunState()
+        : this(SystemTimeSource.Instance)
+    {
+    }
+
+
+    // Test seam (#338): inject a fake time source so the elapsed metric is deterministic.
+    internal EtlRunState(ITimeSource timeSource)
+    {
+        _timeSource = timeSource;
+        _startTimestamp = timeSource.GetTimestamp();
+    }
+
+
+    private TimeSpan Elapsed
+    {
+        get
+        {
+            var ticks = _timeSource.GetTimestamp() - _startTimestamp;
+            return TimeSpan.FromSeconds(ticks / (double)_timeSource.TimestampFrequency);
+        }
+    }
+
+
     public EtlPipelineProgress Snapshot()
     {
-        return new EtlPipelineProgress(ExtractedItemCount, LoadedItemCount, _stopwatch.Elapsed)
+        return new EtlPipelineProgress(ExtractedItemCount, LoadedItemCount, Elapsed)
         {
             ErrorItemCount = ErrorCountReader?.Invoke() ?? 0,
         };

src/Wolfgang.Etl.Abstractions/ExtractorBase.cs

@@ -33,6 +33,15 @@ public abstract class ExtractorBase<TSource, TProgress>
 
 
 
+    // Test seam (#338): when set, StartedAt/Elapsed derive from this time source instead of the
+    // system clock, so the timing-derived Report metrics can be driven deterministically. Left null
+    // in production (real clock). Internal + InternalsVisibleTo, mirroring the IProgressTimer
+    // injection pattern, so Test-Kit doubles can advance a fake clock.
+    internal ITimeSource? TimeSource;
+
+
+
+
     /// <summary>
     /// The UTC time at which the first item was processed (extracted or skipped), or
     /// <c>null</c> if extraction has not produced any items yet. Captured automatically
@@ -61,8 +70,9 @@ protected TimeSpan Elapsed
                 return TimeSpan.Zero;
             }
 
-            var ticks = Stopwatch.GetTimestamp() - start;
-            return TimeSpan.FromSeconds(ticks / (double)Stopwatch.Frequency);
+            var source = TimeSource ?? SystemTimeSource.Instance;
+            var ticks = source.GetTimestamp() - start;
+            return TimeSpan.FromSeconds(ticks / (double)source.TimestampFrequency);
         }
     }
 
@@ -290,12 +300,20 @@ public virtual IAsyncEnumerable<TSource> ExtractAsync(IProgress<TProgress> progr
     /// <returns>A started <see cref="IProgressTimer"/> instance.</returns>
     protected virtual IProgressTimer CreateProgressTimer(IProgress<TProgress> progress)
     {
-        var timer = new SystemProgressTimer(ReportProgress, progress);
+        var timer = TimerCoreFactory is null
+            ? new SystemProgressTimer(ReportProgress, progress)
+            : new SystemProgressTimer(ReportProgress, progress, TimerCoreFactory);
         timer.Start(ReportingInterval);
         return timer;
     }
 
 
+    // Test seam: when set, CreateProgressTimer builds its SystemProgressTimer over this timer core
+    // (a deterministic fake) instead of a real System.Threading.Timer.
+    internal Func<System.Threading.TimerCallback, ITimerCore>? TimerCoreFactory;
+
+
+
 
     private async IAsyncEnumerable<TSource> ExtractWithResetAsync
     (
@@ -495,8 +513,9 @@ private void EnsureStarted()
             return;
         }
 
-        var now = DateTimeOffset.UtcNow;
-        var timestamp = Stopwatch.GetTimestamp();
+        var source = TimeSource ?? SystemTimeSource.Instance;
+        var now = source.UtcNow;
+        var timestamp = source.GetTimestamp();
         if (Interlocked.CompareExchange(ref _startTimestamp, timestamp, 0) == 0)
         {
             _startedAtUtc = now;

src/Wolfgang.Etl.Abstractions/ITimeSource.cs

@@ -0,0 +1,22 @@
+using System;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// Internal seam over the wall-clock and monotonic timer the base classes read when they compute
+/// their <see cref="Report"/> timing metrics (<c>StartedAt</c>, <c>Elapsed</c>, and the throughput
+/// values derived from them). The default is the system clock; a fake can be injected in tests — via
+/// <c>InternalsVisibleTo</c> — so those metrics become deterministic. Deliberately internal: no
+/// public API surface.
+/// </summary>
+internal interface ITimeSource
+{
+    /// <summary>The current UTC wall-clock time (used to capture <c>StartedAt</c>).</summary>
+    DateTimeOffset UtcNow { get; }
+
+    /// <summary>A monotonic timestamp tick count (used to measure <c>Elapsed</c>).</summary>
+    long GetTimestamp();
+
+    /// <summary>The number of <see cref="GetTimestamp"/> ticks per second.</summary>
+    long TimestampFrequency { get; }
+}

src/Wolfgang.Etl.Abstractions/ITimerCore.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace Wolfgang.Etl.Abstractions;
+
+/// <summary>
+/// The minimal periodic-timer primitive that <see cref="SystemProgressTimer"/> drives. Production
+/// code uses a wrapper over <see cref="System.Threading.Timer"/>; tests inject a deterministic fake
+/// (one whose ticks fire on demand and that records <see cref="Change"/> / <see cref="IDisposable.Dispose"/>
+/// calls), which makes the timer's Start / StopTimer / Dispose contract observable without relying on
+/// real wall-clock ticks.
+/// </summary>
+internal interface ITimerCore : IDisposable
+{
+    /// <summary>
+    /// Arms or disarms the timer. <see cref="System.Threading.Timeout.Infinite"/> for both arguments
+    /// disarms it; a positive period arms it to fire repeatedly at that interval.
+    /// </summary>
+    void Change(int dueTime, int period);
+}

src/Wolfgang.Etl.Abstractions/LoaderBase.cs

@@ -32,6 +32,15 @@ public abstract class LoaderBase<TDestination, TProgress>
 
 
 
+    // Test seam (#338): when set, StartedAt/Elapsed derive from this time source instead of the
+    // system clock, so the timing-derived Report metrics can be driven deterministically. Left null
+    // in production (real clock). Internal + InternalsVisibleTo, mirroring the IProgressTimer
+    // injection pattern, so Test-Kit doubles can advance a fake clock.
+    internal ITimeSource? TimeSource;
+
+
+
+
     /// <summary>
     /// The UTC time at which the first item was processed (loaded or skipped), or
     /// <c>null</c> if loading has not produced any items yet. Captured automatically
@@ -60,8 +69,9 @@ protected TimeSpan Elapsed
                 return TimeSpan.Zero;
             }
 
-            var ticks = Stopwatch.GetTimestamp() - start;
-            return TimeSpan.FromSeconds(ticks / (double)Stopwatch.Frequency);
+            var source = TimeSource ?? SystemTimeSource.Instance;
+            var ticks = source.GetTimestamp() - start;
+            return TimeSpan.FromSeconds(ticks / (double)source.TimestampFrequency);
         }
     }
 
@@ -291,12 +301,20 @@ public virtual Task LoadAsync(IAsyncEnumerable<TDestination> items, IProgress<TP
     /// <returns>A started <see cref="IProgressTimer"/> instance.</returns>
     protected virtual IProgressTimer CreateProgressTimer(IProgress<TProgress> progress)
     {
-        var timer = new SystemProgressTimer(ReportProgress, progress);
+        var timer = TimerCoreFactory is null
+            ? new SystemProgressTimer(ReportProgress, progress)
+            : new SystemProgressTimer(ReportProgress, progress, TimerCoreFactory);
         timer.Start(ReportingInterval);
         return timer;
     }
 
 
+    // Test seam: when set, CreateProgressTimer builds its SystemProgressTimer over this timer core
+    // (a deterministic fake) instead of a real System.Threading.Timer.
+    internal Func<System.Threading.TimerCallback, ITimerCore>? TimerCoreFactory;
+
+
+
 
     private Task LoadWithResetAsync
     (
@@ -493,8 +511,9 @@ private void EnsureStarted()
             return;
         }
 
-        var now = DateTimeOffset.UtcNow;
-        var timestamp = Stopwatch.GetTimestamp();
+        var source = TimeSource ?? SystemTimeSource.Instance;
+        var now = source.UtcNow;
+        var timestamp = source.GetTimestamp();
         if (Interlocked.CompareExchange(ref _startTimestamp, timestamp, 0) == 0)
         {
             _startedAtUtc = now;

src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt

@@ -3,12 +3,12 @@ Wolfgang.Etl.Abstractions.EtlPipeline
 Wolfgang.Etl.Abstractions.EtlPipelineProgress
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.Elapsed.get -> System.TimeSpan
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.Elapsed.init -> void
-Wolfgang.Etl.Abstractions.EtlPipelineProgress.ErrorItemCount.get -> int
+Wolfgang.Etl.Abstractions.EtlPipelineProgress.ErrorItemCount.get -> long
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.ErrorItemCount.init -> void
-Wolfgang.Etl.Abstractions.EtlPipelineProgress.EtlPipelineProgress(int ExtractedItemCount, int LoadedItemCount, System.TimeSpan Elapsed) -> void
-Wolfgang.Etl.Abstractions.EtlPipelineProgress.ExtractedItemCount.get -> int
+Wolfgang.Etl.Abstractions.EtlPipelineProgress.EtlPipelineProgress(long ExtractedItemCount, long LoadedItemCount, System.TimeSpan Elapsed) -> void
+Wolfgang.Etl.Abstractions.EtlPipelineProgress.ExtractedItemCount.get -> long
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.ExtractedItemCount.init -> void
-Wolfgang.Etl.Abstractions.EtlPipelineProgress.LoadedItemCount.get -> int
+Wolfgang.Etl.Abstractions.EtlPipelineProgress.LoadedItemCount.get -> long
 Wolfgang.Etl.Abstractions.EtlPipelineProgress.LoadedItemCount.init -> void
 Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions
 Wolfgang.Etl.Abstractions.EtlPipelineSourceExtensions
@@ -133,6 +133,7 @@ Wolfgang.Etl.Abstractions.Report.EstimatedRemaining.get -> System.TimeSpan?
 Wolfgang.Etl.Abstractions.Report.ItemsPerSecond.get -> double
 Wolfgang.Etl.Abstractions.Report.PercentComplete.get -> double?
 Wolfgang.Etl.Abstractions.Report.Report(int currentItemCount) -> void
+Wolfgang.Etl.Abstractions.Report.Report(int currentItemCount, System.DateTimeOffset? startedAt, System.TimeSpan elapsed, int? totalItemCount = null) -> void
 Wolfgang.Etl.Abstractions.Report.StartedAt.get -> System.DateTimeOffset?
 Wolfgang.Etl.Abstractions.Report.StartedAt.init -> void
 Wolfgang.Etl.Abstractions.Report.TotalItemCount.get -> int?

src/Wolfgang.Etl.Abstractions/PublicAPI.Unshipped.txt

@@ -1,2 +1 @@
 #nullable enable
-Wolfgang.Etl.Abstractions.Report.Report(int currentItemCount, System.DateTimeOffset? startedAt, System.TimeSpan elapsed, int? totalItemCount = null) -> void

src/Wolfgang.Etl.Abstractions/SystemProgressTimer.cs

@@ -6,22 +6,18 @@ namespace Wolfgang.Etl.Abstractions;
 
 /// <summary>
 /// The default <see cref="IProgressTimer"/> implementation that wraps
-/// <see cref="System.Threading.Timer"/> to drive progress callbacks on a
-/// background thread-pool thread at a regular interval.
+/// <see cref="System.Threading.Timer"/> (via <see cref="ITimerCore"/>) to drive progress callbacks on
+/// a background thread-pool thread at a regular interval.
 /// </summary>
 /// <remarks>
-/// This class is used internally by the ETL base classes. In production code
-/// it is created automatically by
-/// <c>ExtractorBase.CreateProgressTimer</c>,
-/// <c>TransformerBase.CreateProgressTimer</c>, and
-/// <c>LoaderBase.CreateProgressTimer</c>.
-/// In unit tests, override <c>CreateProgressTimer</c> to return a custom
-/// <see cref="IProgressTimer"/> implementation (for example, a manually
-/// controlled fake whose ticks the test fires on demand).
+/// This class is used internally by the ETL base classes. In production code it is created
+/// automatically by <c>ExtractorBase.CreateProgressTimer</c>, <c>TransformerBase.CreateProgressTimer</c>,
+/// and <c>LoaderBase.CreateProgressTimer</c>. In unit tests, inject a fake <see cref="ITimerCore"/> via
+/// the test constructor so the Start / StopTimer / Dispose contract can be verified deterministically.
 /// </remarks>
 internal sealed class SystemProgressTimer : IProgressTimer
 {
-    private readonly Timer _timer;
+    private readonly ITimerCore _timer;
     private bool _disposed;
 
 
@@ -32,25 +28,24 @@ internal sealed class SystemProgressTimer : IProgressTimer
 
 
     /// <summary>
-    /// Initialises a new <see cref="SystemProgressTimer"/> and immediately
-    /// wires the supplied <paramref name="callback"/> to fire on each tick.
+    /// Initialises a new <see cref="SystemProgressTimer"/> backed by a real
+    /// <see cref="System.Threading.Timer"/>, wiring <paramref name="callback"/> to fire on each tick.
     /// </summary>
-    internal SystemProgressTimer
-    (
-        TimerCallback callback,
-        object? state
-    )
+    internal SystemProgressTimer(TimerCallback callback, object? state)
+        : this(callback, state, onTick => new SystemTimerCore(onTick))
     {
-        // Timer is created stopped (Timeout.Infinite) — Start() arms it.
-#pragma warning disable MA0042 // Timer does not implement IAsyncDisposable
-        _timer = new Timer
-        (
-            _ => OnTick(callback, state),
-            state: null,
-            Timeout.Infinite,
-            Timeout.Infinite
-        );
-#pragma warning restore MA0042
+    }
+
+
+
+    /// <summary>
+    /// Test seam: initialises a new <see cref="SystemProgressTimer"/> whose underlying timer is produced
+    /// by <paramref name="coreFactory"/> (the factory receives the per-tick callback to invoke).
+    /// </summary>
+    internal SystemProgressTimer(TimerCallback callback, object? state, Func<TimerCallback, ITimerCore> coreFactory)
+    {
+        // The core is created stopped; Start() arms it.
+        _timer = coreFactory(_ => OnTick(callback, state));
     }
 
 
@@ -106,8 +101,31 @@ public void Dispose()
         }
         _disposed = true;
         Elapsed = null;
-#pragma warning disable CA1849, VSTHRD103 // Timer.Dispose() is correct here
         _timer.Dispose();
+    }
+
+
+
+    /// <summary>The production <see cref="ITimerCore"/> — a thin wrapper over <see cref="Timer"/>.</summary>
+    private sealed class SystemTimerCore : ITimerCore
+    {
+        private readonly Timer _timer;
+
+        internal SystemTimerCore(TimerCallback onTick)
+        {
+#pragma warning disable MA0042 // Timer does not implement IAsyncDisposable
+            _timer = new Timer(onTick, state: null, Timeout.Infinite, Timeout.Infinite);
+#pragma warning restore MA0042
+        }
+
+        public void Change(int dueTime, int period) => _timer.Change(dueTime, period);
+
+        [ExcludeFromCodeCoverage] // thin BCL delegation
+        public void Dispose()
+        {
+#pragma warning disable CA1849, VSTHRD103 // Timer.Dispose() is correct here
+            _timer.Dispose();
 #pragma warning restore CA1849, VSTHRD103
+        }
     }
 }