diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs index 9c8c7a42..4c7fd8a9 100644 --- a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs +++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlRunState.cs @@ -1,5 +1,4 @@ using System; -using System.Diagnostics; namespace Wolfgang.Etl.Abstractions; @@ -12,7 +11,8 @@ namespace Wolfgang.Etl.Abstractions; /// internal sealed class EtlRunState { - private readonly Stopwatch _stopwatch = Stopwatch.StartNew(); + private readonly ITimeSource _timeSource; + private readonly long _startTimestamp; public long ExtractedItemCount; @@ -23,9 +23,33 @@ internal sealed class EtlRunState public Func? 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, }; diff --git a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs index 40aff614..2b795481 100644 --- a/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs +++ b/src/Wolfgang.Etl.Abstractions/ExtractorBase.cs @@ -33,6 +33,15 @@ public abstract class ExtractorBase + // 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; + + + + /// /// The UTC time at which the first item was processed (extracted or skipped), or /// null 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); } } @@ -503,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; diff --git a/src/Wolfgang.Etl.Abstractions/ITimeSource.cs b/src/Wolfgang.Etl.Abstractions/ITimeSource.cs new file mode 100644 index 00000000..f2a0df2d --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/ITimeSource.cs @@ -0,0 +1,22 @@ +using System; + +namespace Wolfgang.Etl.Abstractions; + +/// +/// Internal seam over the wall-clock and monotonic timer the base classes read when they compute +/// their timing metrics (StartedAt, Elapsed, and the throughput +/// values derived from them). The default is the system clock; a fake can be injected in tests — via +/// InternalsVisibleTo — so those metrics become deterministic. Deliberately internal: no +/// public API surface. +/// +internal interface ITimeSource +{ + /// The current UTC wall-clock time (used to capture StartedAt). + DateTimeOffset UtcNow { get; } + + /// A monotonic timestamp tick count (used to measure Elapsed). + long GetTimestamp(); + + /// The number of ticks per second. + long TimestampFrequency { get; } +} diff --git a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs index 184d9fa1..a2971fd9 100644 --- a/src/Wolfgang.Etl.Abstractions/LoaderBase.cs +++ b/src/Wolfgang.Etl.Abstractions/LoaderBase.cs @@ -32,6 +32,15 @@ public abstract class LoaderBase + // 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; + + + + /// /// The UTC time at which the first item was processed (loaded or skipped), or /// null 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); } } @@ -501,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; diff --git a/src/Wolfgang.Etl.Abstractions/SystemTimeSource.cs b/src/Wolfgang.Etl.Abstractions/SystemTimeSource.cs new file mode 100644 index 00000000..aaa06d1b --- /dev/null +++ b/src/Wolfgang.Etl.Abstractions/SystemTimeSource.cs @@ -0,0 +1,28 @@ +using System; +using System.Diagnostics; + +namespace Wolfgang.Etl.Abstractions; + +/// +/// The default : reads the real system clock via +/// and . A shared stateless singleton — +/// the base classes use it whenever no test time source has been injected. +/// +internal sealed class SystemTimeSource : ITimeSource +{ + internal static readonly SystemTimeSource Instance = new(); + + + private SystemTimeSource() + { + } + + + public DateTimeOffset UtcNow => DateTimeOffset.UtcNow; + + + public long GetTimestamp() => Stopwatch.GetTimestamp(); + + + public long TimestampFrequency => Stopwatch.Frequency; +} diff --git a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs index 993f0901..045eaafd 100644 --- a/src/Wolfgang.Etl.Abstractions/TransformerBase.cs +++ b/src/Wolfgang.Etl.Abstractions/TransformerBase.cs @@ -35,6 +35,15 @@ public abstract class TransformerBase + // 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; + + + + /// /// The UTC time at which the first item was processed (transformed or skipped), or /// null if transformation has not produced any items yet. Captured automatically @@ -63,8 +72,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); } } @@ -511,8 +521,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; diff --git a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj index 878463d2..6bb45565 100644 --- a/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj +++ b/src/Wolfgang.Etl.Abstractions/Wolfgang.Etl.Abstractions.csproj @@ -38,6 +38,8 @@ + + diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/ClockSeamTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/ClockSeamTests.cs new file mode 100644 index 00000000..ffc38236 --- /dev/null +++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/BaseClassTests/ClockSeamTests.cs @@ -0,0 +1,201 @@ +using System; +using System.Collections.Generic; +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.BaseClassTests; + +/// +/// Covers the #338 clock seam: when an ITimeSource is injected, the base classes' timing +/// metrics (StartedAt / Elapsed) and EtlRunState's elapsed derive from it +/// deterministically; with no injection the real system clock is used. The seam is internal (surfaced +/// to tests via InternalsVisibleTo); no public API change. +/// +public class ClockSeamTests +{ + private static readonly DateTimeOffset Start = new(2020, 1, 1, 0, 0, 0, TimeSpan.Zero); + + + [Fact] + public async Task Extractor_StartedAt_and_Elapsed_derive_from_the_injected_time_source() + { + var clock = new FakeTimeSource(); + var sut = new ClockExtractor { TimeSource = clock }; + + await Drain(sut.ExtractAsync(CancellationToken.None)); // first item captures start + + Assert.Equal(Start, sut.PeekStartedAt); + Assert.Equal(TimeSpan.Zero, sut.PeekElapsed); + + clock.Advance(TimeSpan.FromSeconds(5)); + + Assert.Equal(TimeSpan.FromSeconds(5), sut.PeekElapsed); + Assert.Equal(Start, sut.PeekStartedAt); // StartedAt is the captured value, not "now" + } + + + [Fact] + public async Task Loader_Elapsed_derives_from_the_injected_time_source() + { + var clock = new FakeTimeSource(); + var sut = new ClockLoader { TimeSource = clock }; + + await sut.LoadAsync(AsyncSource(1), CancellationToken.None); + clock.Advance(TimeSpan.FromSeconds(2)); + + Assert.Equal(Start, sut.PeekStartedAt); + Assert.Equal(TimeSpan.FromSeconds(2), sut.PeekElapsed); + } + + + [Fact] + public async Task Transformer_Elapsed_derives_from_the_injected_time_source() + { + var clock = new FakeTimeSource(); + var sut = new ClockTransformer { TimeSource = clock }; + + await Drain(sut.TransformAsync(AsyncSource(1), CancellationToken.None)); + clock.Advance(TimeSpan.FromSeconds(9)); + + Assert.Equal(Start, sut.PeekStartedAt); + Assert.Equal(TimeSpan.FromSeconds(9), sut.PeekElapsed); + } + + + [Fact] + public void Before_the_first_item_StartedAt_is_null_and_Elapsed_is_zero() + { + var sut = new ClockExtractor { TimeSource = new FakeTimeSource() }; + + Assert.Null(sut.PeekStartedAt); + Assert.Equal(TimeSpan.Zero, sut.PeekElapsed); + } + + + [Fact] + public async Task With_no_injected_source_the_real_system_clock_is_used() + { + var sut = new ClockExtractor(); // TimeSource null -> SystemTimeSource + + var before = DateTimeOffset.UtcNow; + await Drain(sut.ExtractAsync(CancellationToken.None)); + var after = DateTimeOffset.UtcNow; + + Assert.NotNull(sut.PeekStartedAt); + Assert.InRange(sut.PeekStartedAt!.Value, before.AddSeconds(-1), after.AddSeconds(1)); + Assert.True(sut.PeekElapsed >= TimeSpan.Zero); + } + + + [Fact] + public void EtlRunState_elapsed_derives_from_the_injected_time_source() + { + var clock = new FakeTimeSource(); + var state = new EtlRunState(clock); + + clock.Advance(TimeSpan.FromSeconds(3)); + + Assert.Equal(TimeSpan.FromSeconds(3), state.Snapshot().Elapsed); + } + + + // ---------- 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) + { + await foreach (var _ in source.ConfigureAwait(false)) + { + } + } + + + // ---------- doubles ---------- + + // A fake clock: UtcNow and a monotonic tick counter the test advances by hand. The tick counter + // starts non-zero so a captured start never collides with the base's "not started" sentinel (0). + private sealed class FakeTimeSource : ITimeSource + { + public DateTimeOffset UtcNow { get; private set; } = Start; + + public long Timestamp { get; private set; } = TimeSpan.TicksPerSecond; + + public long TimestampFrequency => TimeSpan.TicksPerSecond; + + public long GetTimestamp() => Timestamp; + + public void Advance(TimeSpan by) + { + UtcNow += by; + Timestamp += (long)(by.TotalSeconds * TimestampFrequency); + } + } + + + private sealed class ClockExtractor : ExtractorBase + { + public DateTimeOffset? PeekStartedAt => StartedAt; + + public TimeSpan PeekElapsed => Elapsed; + + protected override async IAsyncEnumerable ExtractWorkerAsync([EnumeratorCancellation] CancellationToken token) + { + await Task.Yield(); + IncrementCurrentItemCount(); + yield return 1; + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + private sealed class ClockLoader : LoaderBase + { + public DateTimeOffset? PeekStartedAt => StartedAt; + + public TimeSpan PeekElapsed => Elapsed; + + protected override async Task LoadWorkerAsync(IAsyncEnumerable items, CancellationToken token) + { + await foreach (var item in items.WithCancellation(token).ConfigureAwait(false)) + { + IncrementCurrentItemCount(); + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } + + + private sealed class ClockTransformer : TransformerBase + { + public DateTimeOffset? PeekStartedAt => StartedAt; + + public TimeSpan PeekElapsed => Elapsed; + + protected override async IAsyncEnumerable TransformWorkerAsync( + IAsyncEnumerable items, [EnumeratorCancellation] CancellationToken token) + { + await foreach (var item in items.WithCancellation(token).ConfigureAwait(false)) + { + IncrementCurrentItemCount(); + yield return item; + } + } + + protected override EtlProgress CreateProgressReport() => new(CurrentItemCount); + } +}