From 2d04d6e9e2df43346725258dabd96532424c1ab6 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Fri, 21 Aug 2026 17:20:06 -0700 Subject: [PATCH] Reduce trace log noise in monitors. --- .../AppliedExecutionEndArbiter.cs | 51 ++- .../Heartbeats/HeartbeatMaintainer.cs | 4 +- .../Idempotency/IdempotencyMonitor.cs | 5 +- .../AppliedExecutionEndArbiterTests.cs | 293 ++++++++++++++---- .../Services/Heartbeats/MaintainerTests.cs | 3 +- .../Idempotency/IdempotencyMonitorTests.cs | 30 +- 6 files changed, 291 insertions(+), 95 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/AppliedExecutionEndArbiter.cs b/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/AppliedExecutionEndArbiter.cs index 2e5cc39..de95db5 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/AppliedExecutionEndArbiter.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/AppliedExecutionEndArbiter.cs @@ -1,22 +1,31 @@ +using Microsoft.Extensions.Logging; using RedShirt.Example.JobWorker.Common.Services.Utility; using RedShirt.Example.JobWorker.Core.Services.Jobs; +using RedShirt.Example.JobWorker.Core.Utility; namespace RedShirt.Example.JobWorker.Core.Services.ExecutionState; /// /// Dictates if maintainer workers should continue running. /// Extends the functionality of the base IExecutionEndArbiter by accessing the job repository. -/// Written as a test-friendly alternative to `while(true){}` +/// Initially written as a test-friendly alternative to `while(true){}` /// internal interface IAppliedMaintainerExecutionEndArbiter { /// /// Delays for , honouring both and /// an internal interrupt signal that triggers when the worker is stopping. + /// If there are no watched jobs to monitor, then also delays until there are watched jobs to monitor before + /// delaying for . /// Intended for maintainer workers only. /// Cancellation caused by the internal interrupt signal is ignored and treated as a completed delay. /// - Task DelayMaintainerWithStopAwarenessAsync(TimeSpan delay, CancellationToken cancellationToken = default); + /// How long to wait when the skip-wait event is not set. + /// Label for future wait-related log messages (unused for now). + /// Description for future wait-related log messages (unused for now). + /// Caller cancellation. + Task MaintainerDelayWaitAsync(TimeSpan delay, string loggerLabel, string loggerDescription, + CancellationToken cancellationToken = default); bool MaintainerShouldKeepRunning(); } @@ -37,7 +46,9 @@ internal sealed class AppliedExecutionEndArbiter : IAppliedMaintainerExecutionEn private readonly IExecutionEndArbiter _executionEndArbiter; private readonly CancellationTokenSource _interruptCts = new(); private readonly Lock _lock = new(); + private readonly ILogger _logger; private readonly ISleepService _sleepService; + private readonly AsyncManualResetEvent _watchedJobsToMaintainEvent = new(); private bool _disposed; private int _inactiveJobsCount; @@ -89,6 +100,8 @@ private void OnInactiveJobChange(int inactiveJobCount) private void OnWatchedJobChange(int watchedJobCount) { bool shouldInterrupt; + int previousValue; + lock (_lock) { if (_disposed) @@ -96,10 +109,25 @@ private void OnWatchedJobChange(int watchedJobCount) return; } + previousValue = _watchedJobsCount; _watchedJobsCount = watchedJobCount; shouldInterrupt = ShouldSendMaintainerInterruptSignalUnsafe(); } + if (previousValue != watchedJobCount) + { + // Confirmed a change + + if (watchedJobCount == 0) + { + _watchedJobsToMaintainEvent.Reset(); + } + else + { + _watchedJobsToMaintainEvent.Set(); + } + } + if (shouldInterrupt) { TryCancelInterrupt(); @@ -109,10 +137,12 @@ private void OnWatchedJobChange(int watchedJobCount) public AppliedExecutionEndArbiter( IExecutionEndArbiter executionEndArbiter, IJobRepository jobRepository, - ISleepService sleepService) + ISleepService sleepService, + ILogger logger) { _executionEndArbiter = executionEndArbiter; _sleepService = sleepService; + _logger = logger; jobRepository.SubscribeToInactiveCountUpdate(OnInactiveJobChange); jobRepository.SubscribeToWatchedJobsUpdate(OnWatchedJobChange); } @@ -127,7 +157,7 @@ public bool ExecutorsShouldKeepRunning() } } - public async Task DelayMaintainerWithStopAwarenessAsync(TimeSpan delay, + public async Task MaintainerDelayWaitAsync(TimeSpan delay, string loggerLabel, string loggerDescription, CancellationToken cancellationToken = default) { CancellationToken interruptToken; @@ -146,6 +176,19 @@ public async Task DelayMaintainerWithStopAwarenessAsync(TimeSpan delay, try { + // Immediately check to see if the event is set. + if (!await _watchedJobsToMaintainEvent.WaitAsync(TimeSpan.Zero, linkedCts.Token)) + { + // Event is not set, so wait until it is + // This wait prevents the maintainer from creating noise (trace-level though it may be) + // when there are no watched jobs to maintain. + _logger.LogTrace("{Label}: Waiting for watchable events", loggerLabel); + await _watchedJobsToMaintainEvent.WaitAsync(linkedCts.Token); + return; + } + + _logger.LogTrace("{Label}: {Time} until next {Description}", loggerLabel, delay, + loggerDescription); await _sleepService.DelayAsync(delay, linkedCts.Token); } catch (OperationCanceledException) when (interruptToken.IsCancellationRequested diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Heartbeats/HeartbeatMaintainer.cs b/src/RedShirt.Example.JobWorker.Core/Services/Heartbeats/HeartbeatMaintainer.cs index 61dc6cb..f06cb14 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Heartbeats/HeartbeatMaintainer.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Heartbeats/HeartbeatMaintainer.cs @@ -56,8 +56,8 @@ internal sealed class HeartbeatMaintainer( /// private async Task LogAndWaitAsync(TimeSpan timeToWait, CancellationToken cancellationToken = default) { - logger.LogTrace("Heartbeat Monitor: Waiting for {Time} until next heartbeat check", timeToWait); - await appliedExecutionEndArbiter.DelayMaintainerWithStopAwarenessAsync(timeToWait, cancellationToken); + await appliedExecutionEndArbiter.MaintainerDelayWaitAsync(timeToWait, "Heartbeat Monitor", "heartbeat check", + cancellationToken); } /// diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Idempotency/IdempotencyMonitor.cs b/src/RedShirt.Example.JobWorker.Core/Services/Idempotency/IdempotencyMonitor.cs index 80691e6..78d39c8 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Idempotency/IdempotencyMonitor.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Idempotency/IdempotencyMonitor.cs @@ -117,9 +117,8 @@ await idempotencyExecutionService.SetResultInCacheAsync(blockedJob.RawJobModel, /// private async Task LogAndWaitAsync(TimeSpan timeToWait, CancellationToken cancellationToken = default) { - logger.LogTrace("Idempotency Monitor: {Time} until next follow-up check", - timeToWait); - await executionEndArbiter.DelayMaintainerWithStopAwarenessAsync(timeToWait, cancellationToken); + await executionEndArbiter.MaintainerDelayWaitAsync(timeToWait, "Idempotency Monitor", "follow-up check", + cancellationToken); } public async Task RunAsync(CancellationToken cancellationToken = default) diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/AppliedExecutionEndArbiterTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/AppliedExecutionEndArbiterTests.cs index 12d06de..23b6ef8 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/AppliedExecutionEndArbiterTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/AppliedExecutionEndArbiterTests.cs @@ -1,6 +1,8 @@ +using Microsoft.Extensions.Logging.Abstractions; using RedShirt.Example.JobWorker.Common.Services.Utility; using RedShirt.Example.JobWorker.Core.Services.ExecutionState; using RedShirt.Example.JobWorker.Core.Services.Jobs; +using System.Reflection; namespace RedShirt.Example.JobWorker.Core.UnitTests.Tests.Services.ExecutionState; @@ -51,7 +53,7 @@ public void CountCallbacks_AfterDispose_AreIgnored() var arbiter = new AppliedExecutionEndArbiter( innerArbiter.Object, CreateJobRepository(out var notifier, 1, 1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.MaintainerShouldKeepRunning()); Assert.True(arbiter.ExecutorsShouldKeepRunning()); @@ -75,7 +77,7 @@ public void CountCallbacks_UpdateKeepRunningDecisions() using var arbiter = new AppliedExecutionEndArbiter( innerArbiter.Object, CreateJobRepository(out var notifier, 1, 1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.ExecutorsShouldKeepRunning()); Assert.True(arbiter.MaintainerShouldKeepRunning()); @@ -90,11 +92,36 @@ public void CountCallbacks_UpdateKeepRunningDecisions() } [Fact] - public async Task DelayMaintainerWithStopAwarenessAsync_CompletesNormallyWhenNeitherTokenCancels() + public void Dispose_IsIdempotent() + { + var innerArbiter = new Mock(MockBehavior.Strict); + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); + + var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1, 1).Object, + CreateSleepService().Object, NullLogger.Instance); + + arbiter.Dispose(); + arbiter.Dispose(); + } + + [Fact] + public void ExecutorsShouldKeepRunning_WhenInnerTrueAndNoInactive_ReturnsTrue() + { + var innerArbiter = new Mock(MockBehavior.Strict); + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); + + using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(0, 5).Object, + CreateSleepService().Object, NullLogger.Instance); + + Assert.True(arbiter.ExecutorsShouldKeepRunning()); + } + + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_CompletesNormallyWhenNeitherTokenCancels() { var delay = TimeSpan.FromSeconds(5); var innerArbiter = new Mock(MockBehavior.Strict); - // Keep jobs present so the interrupt signal is not sent. + // Keep jobs present so the interrupt signal is not sent and the wait event is set. innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); var sleepService = CreateSleepService(); @@ -103,15 +130,47 @@ public async Task DelayMaintainerWithStopAwarenessAsync_CompletesNormallyWhenNei .Returns(Task.CompletedTask); using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1, 1).Object, - sleepService.Object); + sleepService.Object, NullLogger.Instance); - await arbiter.DelayMaintainerWithStopAwarenessAsync(delay, TestContext.Current.CancellationToken); + await arbiter.MaintainerDelayWaitAsync(delay, "test", "test", TestContext.Current.CancellationToken); sleepService.Verify(s => s.DelayAsync(delay, It.IsAny()), Times.Once); } - [Fact] - public async Task DelayMaintainerWithStopAwarenessAsync_WhenCallerCancels_PropagatesCancellation() + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenCallerCancelsDuringSleep_PropagatesCancellation() + { + var delay = TimeSpan.FromSeconds(5); + using var callerCts = new CancellationTokenSource(); + + var innerArbiter = new Mock(MockBehavior.Strict); + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); + + var delayStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var sleepService = CreateSleepService(); + sleepService + .Setup(s => s.DelayAsync(delay, It.IsAny())) + .Returns((TimeSpan _, CancellationToken token) => + { + delayStarted.SetResult(); + return Task.Delay(Timeout.Infinite, token); + }); + + using var arbiter = new AppliedExecutionEndArbiter( + innerArbiter.Object, + CreateJobRepository(1, 1).Object, + sleepService.Object, NullLogger.Instance); + + var delayTask = arbiter.MaintainerDelayWaitAsync(delay, "test", "test", callerCts.Token); + await delayStarted.Task; + + await callerCts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => delayTask); + } + + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenCallerCancels_PropagatesCancellation() { var delay = TimeSpan.FromSeconds(5); using var callerCts = new CancellationTokenSource(); @@ -127,14 +186,14 @@ public async Task DelayMaintainerWithStopAwarenessAsync_WhenCallerCancels_Propag .Returns((TimeSpan _, CancellationToken token) => Task.FromCanceled(token)); using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1, 1).Object, - sleepService.Object); + sleepService.Object, NullLogger.Instance); await Assert.ThrowsAnyAsync(() => - arbiter.DelayMaintainerWithStopAwarenessAsync(delay, callerCts.Token)); + arbiter.MaintainerDelayWaitAsync(delay, "test", "test", callerCts.Token)); } - [Fact] - public async Task DelayMaintainerWithStopAwarenessAsync_WhenCountsDropToEmptyWhileStopping_InterruptsAndCompletes() + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenCountsDropToEmptyWhileStopping_InterruptsAndCompletes() { var delay = TimeSpan.FromSeconds(5); var innerArbiter = new Mock(MockBehavior.Strict); @@ -157,9 +216,9 @@ public async Task DelayMaintainerWithStopAwarenessAsync_WhenCountsDropToEmptyWhi using var arbiter = new AppliedExecutionEndArbiter( innerArbiter.Object, CreateJobRepository(out var notifier, 1, 1).Object, - sleepService.Object); + sleepService.Object, NullLogger.Instance); - var delayTask = arbiter.DelayMaintainerWithStopAwarenessAsync(delay, CancellationToken.None); + var delayTask = arbiter.MaintainerDelayWaitAsync(delay, "test", "test", CancellationToken.None); await delayStarted.Task; // Both counts must be empty before the interrupt fires. @@ -172,8 +231,8 @@ public async Task DelayMaintainerWithStopAwarenessAsync_WhenCountsDropToEmptyWhi await delayTask; } - [Fact] - public async Task DelayMaintainerWithStopAwarenessAsync_WhenDisposed_ReturnsWithoutSleeping() + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenDisposed_ReturnsWithoutSleeping() { var delay = TimeSpan.FromSeconds(5); var innerArbiter = new Mock(MockBehavior.Strict); @@ -181,81 +240,162 @@ public async Task DelayMaintainerWithStopAwarenessAsync_WhenDisposed_ReturnsWith var sleepService = CreateSleepService(); var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1, 1).Object, - sleepService.Object); + sleepService.Object, NullLogger.Instance); arbiter.Dispose(); - await arbiter.DelayMaintainerWithStopAwarenessAsync(delay, CancellationToken.None); + await arbiter.MaintainerDelayWaitAsync(delay, "test", "test", CancellationToken.None); sleepService.Verify(s => s.DelayAsync(It.IsAny(), It.IsAny()), Times.Never); } - [Fact] - public async Task DelayMaintainerWithStopAwarenessAsync_WhenEmptyButInnerSaysKeepRunning_DoesNotInterrupt() + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenInterrupted_IgnoresCancellation() { var delay = TimeSpan.FromSeconds(5); var innerArbiter = new Mock(MockBehavior.Strict); - innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); + // Empty job counts while stopping cancels the internal interrupt token on subscribe. + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(false); var sleepService = CreateSleepService(); - sleepService - .Setup(s => s.DelayAsync(delay, It.IsAny())) - .Returns((TimeSpan _, CancellationToken token) => - { - Assert.False(token.IsCancellationRequested); - return Task.CompletedTask; - }); using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository().Object, - sleepService.Object); + sleepService.Object, NullLogger.Instance); - await arbiter.DelayMaintainerWithStopAwarenessAsync(delay, CancellationToken.None); + await arbiter.MaintainerDelayWaitAsync(delay, "test", "test", CancellationToken.None); - sleepService.Verify(s => s.DelayAsync(delay, It.IsAny()), Times.Once); + sleepService.Verify(s => s.DelayAsync(It.IsAny(), It.IsAny()), Times.Never); } - [Fact] - public async Task DelayMaintainerWithStopAwarenessAsync_WhenInterrupted_IgnoresCancellation() + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenNoWatchedJobsAndKeepRunning_DoesNotInterrupt() { var delay = TimeSpan.FromSeconds(5); var innerArbiter = new Mock(MockBehavior.Strict); - // Empty job counts while stopping cancels the internal interrupt token on subscribe. - innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(false); + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); var sleepService = CreateSleepService(); - sleepService - .Setup(s => s.DelayAsync(delay, It.IsAny())) - .Returns((TimeSpan _, CancellationToken token) => Task.FromCanceled(token)); - using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository().Object, - sleepService.Object); + using var arbiter = new AppliedExecutionEndArbiter( + innerArbiter.Object, + CreateJobRepository(out var notifier).Object, + sleepService.Object, NullLogger.Instance); + + var delayTask = arbiter.MaintainerDelayWaitAsync(delay, "test", "test", CancellationToken.None); - await arbiter.DelayMaintainerWithStopAwarenessAsync(delay, CancellationToken.None); + await Task.Delay(50, TestContext.Current.CancellationToken); + Assert.False(delayTask.IsCompleted); + + // Empty counts while keep-running must not fire the interrupt; only watched jobs unblock. + notifier.NotifyInactive(0); + await Task.Delay(50, TestContext.Current.CancellationToken); + Assert.False(delayTask.IsCompleted); + + notifier.NotifyWatched(1); + await delayTask; } - [Fact] - public void Dispose_IsIdempotent() + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenNoWatchedJobs_WaitsUntilWatchedThenReturnsWithoutSleeping() { + var delay = TimeSpan.FromSeconds(5); var innerArbiter = new Mock(MockBehavior.Strict); innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); - var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1, 1).Object, - CreateSleepService().Object); + var sleepService = CreateSleepService(); - arbiter.Dispose(); - arbiter.Dispose(); + using var arbiter = new AppliedExecutionEndArbiter( + innerArbiter.Object, + CreateJobRepository(out var notifier).Object, + sleepService.Object, NullLogger.Instance); + + var delayTask = arbiter.MaintainerDelayWaitAsync(delay, "test", "test", CancellationToken.None); + + await Task.Delay(50, TestContext.Current.CancellationToken); + Assert.False(delayTask.IsCompleted); + + notifier.NotifyWatched(1); + await delayTask; + + sleepService.Verify(s => s.DelayAsync(It.IsAny(), It.IsAny()), Times.Never); } - [Fact] - public void ExecutorsShouldKeepRunning_WhenInnerTrueAndNoInactive_ReturnsTrue() + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenWaitingForWatched_CallerCancelPropagates() { + var delay = TimeSpan.FromSeconds(5); + using var callerCts = new CancellationTokenSource(); + var innerArbiter = new Mock(MockBehavior.Strict); innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); - using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(0, 5).Object, - CreateSleepService().Object); + var sleepService = CreateSleepService(); - Assert.True(arbiter.ExecutorsShouldKeepRunning()); + using var arbiter = new AppliedExecutionEndArbiter( + innerArbiter.Object, + CreateJobRepository().Object, + sleepService.Object, NullLogger.Instance); + + var delayTask = arbiter.MaintainerDelayWaitAsync(delay, "test", "test", callerCts.Token); + + await Task.Delay(50, TestContext.Current.CancellationToken); + Assert.False(delayTask.IsCompleted); + + await callerCts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => delayTask); + sleepService.Verify(s => s.DelayAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenWaitingForWatched_InterruptCompletesWithoutThrowing() + { + var delay = TimeSpan.FromSeconds(5); + var innerArbiter = new Mock(MockBehavior.Strict); + // Stopping with inactive work present: interrupt is deferred until counts clear. + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(false); + + var sleepService = CreateSleepService(); + + using var arbiter = new AppliedExecutionEndArbiter( + innerArbiter.Object, + CreateJobRepository(out var notifier, 1).Object, + sleepService.Object, NullLogger.Instance); + + var delayTask = arbiter.MaintainerDelayWaitAsync(delay, "test", "test", CancellationToken.None); + + await Task.Delay(50, TestContext.Current.CancellationToken); + Assert.False(delayTask.IsCompleted); + + notifier.NotifyInactive(0); + await delayTask; + + sleepService.Verify(s => s.DelayAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact(Timeout = 5000)] + public async Task MaintainerDelayWaitAsync_WhenWatchedCountUnchanged_StillTakesSleepPath() + { + var delay = TimeSpan.FromSeconds(5); + var innerArbiter = new Mock(MockBehavior.Strict); + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); + + var sleepService = CreateSleepService(); + sleepService + .Setup(s => s.DelayAsync(delay, It.IsAny())) + .Returns(Task.CompletedTask); + + using var arbiter = new AppliedExecutionEndArbiter( + innerArbiter.Object, + CreateJobRepository(out var notifier, 0, 1).Object, + sleepService.Object, NullLogger.Instance); + + // Same count must not Reset the wait event; sleep path should remain available. + notifier.NotifyWatched(1); + + await arbiter.MaintainerDelayWaitAsync(delay, "test", "test", CancellationToken.None); + + sleepService.Verify(s => s.DelayAsync(delay, It.IsAny()), Times.Once); } [Fact] @@ -265,7 +405,7 @@ public void MaintainerShouldKeepRunning_WhenInnerTrueAndNoJobs_ReturnsTrue() innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository().Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.MaintainerShouldKeepRunning()); } @@ -282,7 +422,7 @@ public void TestExecutorStopRunningWeird() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(-1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.False(arbiter.ExecutorsShouldKeepRunning()); } @@ -299,7 +439,7 @@ public void TestExecutorsKeepRunningA() .Returns(true); using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.ExecutorsShouldKeepRunning()); } @@ -313,7 +453,7 @@ public void TestExecutorsKeepRunningBecauseInactive() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.ExecutorsShouldKeepRunning()); } @@ -327,7 +467,7 @@ public void TestExecutorsKeepRunningDespiteInner() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.ExecutorsShouldKeepRunning()); } @@ -341,7 +481,7 @@ public void TestExecutorsKeepRunningDespiteWatched() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(0, 1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); // Confirming that we're ignoring watched jobs Assert.False(arbiter.ExecutorsShouldKeepRunning()); @@ -356,7 +496,7 @@ public void TestExecutorsStopRunning() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository().Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.False(arbiter.ExecutorsShouldKeepRunning()); } @@ -373,7 +513,7 @@ public void TestMaintainerKeepRunningA() .Returns(true); using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1, 1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.MaintainerShouldKeepRunning()); } @@ -387,7 +527,7 @@ public void TestMaintainerKeepRunningBecauseInactive() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.MaintainerShouldKeepRunning()); } @@ -401,7 +541,7 @@ public void TestMaintainerKeepRunningBecauseWatched() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(0, 1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.MaintainerShouldKeepRunning()); } @@ -415,7 +555,7 @@ public void TestMaintainerKeepRunningDespiteInner() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(1, 1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.True(arbiter.MaintainerShouldKeepRunning()); } @@ -429,7 +569,7 @@ public void TestMaintainerStopRunning() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository().Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); Assert.False(arbiter.MaintainerShouldKeepRunning()); } @@ -446,7 +586,30 @@ public void TestMaintainerStopRunningWeird() .Returns(false); // Inner arbiter says no using var arbiter = new AppliedExecutionEndArbiter(innerArbiter.Object, CreateJobRepository(-1, -1).Object, - CreateSleepService().Object); + CreateSleepService().Object, NullLogger.Instance); + + Assert.False(arbiter.MaintainerShouldKeepRunning()); + } + + [Fact] + public void TryCancelInterrupt_WhenCtsAlreadyDisposed_SwallowsObjectDisposedException() + { + var innerArbiter = new Mock(MockBehavior.Strict); + innerArbiter.Setup(a => a.ShouldKeepRunning()).Returns(false); + + using var arbiter = new AppliedExecutionEndArbiter( + innerArbiter.Object, + CreateJobRepository(out var notifier, 1).Object, + CreateSleepService().Object, NullLogger.Instance); + + var field = typeof(AppliedExecutionEndArbiter).GetField("_interruptCts", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(field); + var cts = (CancellationTokenSource) field.GetValue(arbiter)!; + cts.Dispose(); + + // Clearing the last active count would cancel the interrupt CTS; it is already disposed. + notifier.NotifyInactive(0); Assert.False(arbiter.MaintainerShouldKeepRunning()); } diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Heartbeats/MaintainerTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Heartbeats/MaintainerTests.cs index 9ec7dfc..bd2d3fc 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Heartbeats/MaintainerTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Heartbeats/MaintainerTests.cs @@ -35,7 +35,8 @@ private static ISleepService CreateSleepService() private static void SetupMaintainerDelay(Mock arbiter) { arbiter - .Setup(a => a.DelayMaintainerWithStopAwarenessAsync(It.IsAny(), It.IsAny())) + .Setup(a => a.MaintainerDelayWaitAsync(It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny())) .Returns(Task.CompletedTask); } diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Idempotency/IdempotencyMonitorTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Idempotency/IdempotencyMonitorTests.cs index cbf4dbe..70ba3c8 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Idempotency/IdempotencyMonitorTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Idempotency/IdempotencyMonitorTests.cs @@ -1,4 +1,3 @@ -using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using RedShirt.Example.JobWorker.Common.Distributed.Models; @@ -59,17 +58,18 @@ private static (Mock Entry, Mock JobModel, Mock< private static void SetupMaintainerDelay(Mock arbiter) { arbiter - .Setup(a => a.DelayMaintainerWithStopAwarenessAsync(It.IsAny(), It.IsAny())) + .Setup(a => a.MaintainerDelayWaitAsync(It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny())) .Returns(Task.CompletedTask); } [Fact(Timeout = 1000)] - public async Task RunAsync_LogsWaitBeforeDelayingBetweenLoops() + public async Task RunAsync_PassesWaitLabelAndDescriptionToDelay() { var doQuit = false; var executionEndArbiter = new Mock(MockBehavior.Strict); executionEndArbiter - .Setup(a => a.DelayMaintainerWithStopAwarenessAsync(TimeSpan.FromSeconds(3), + .Setup(a => a.MaintainerDelayWaitAsync(TimeSpan.FromSeconds(3), "Idempotency Monitor", "follow-up check", TestContext.Current.CancellationToken)) .Returns(Task.CompletedTask); executionEndArbiter @@ -90,27 +90,17 @@ public async Task RunAsync_LogsWaitBeforeDelayingBetweenLoops() .Setup(r => r.GetAllIdempotencyBlockedJobsAsync(TestContext.Current.CancellationToken)) .ReturnsAsync([]); - var logger = new Mock>(); - logger.Setup(l => l.IsEnabled(LogLevel.Trace)).Returns(true); - var monitor = new IdempotencyMonitor(executionEndArbiter.Object, jobRepository.Object, new Mock(MockBehavior.Strict).Object, new Mock(MockBehavior.Strict).Object, Options.Create(CreateOptions(monitorIntervalSeconds: 1)), CreateStatisticsService(), - logger.Object); + new NullLogger()); await monitor.RunAsync(TestContext.Current.CancellationToken); - logger.Verify( - l => l.Log( - LogLevel.Trace, - It.IsAny(), - It.Is((state, _) => - state.ToString()!.Contains("Idempotency Monitor:", StringComparison.Ordinal) - && state.ToString()!.Contains("until next follow-up check", StringComparison.Ordinal)), - It.IsAny(), - It.IsAny>()), - Times.Once); + executionEndArbiter.Verify( + a => a.MaintainerDelayWaitAsync(TimeSpan.FromSeconds(3), "Idempotency Monitor", "follow-up check", + TestContext.Current.CancellationToken), Times.Once); } [Fact(Timeout = 1000)] @@ -119,7 +109,7 @@ public async Task RunAsync_SleepsUsingEffectiveMonitorIntervalBetweenLoops() var doQuit = false; var executionEndArbiter = new Mock(MockBehavior.Strict); executionEndArbiter - .Setup(a => a.DelayMaintainerWithStopAwarenessAsync(TimeSpan.FromSeconds(3), + .Setup(a => a.MaintainerDelayWaitAsync(TimeSpan.FromSeconds(3), It.IsAny(), It.IsAny(), TestContext.Current.CancellationToken)) .Returns(Task.CompletedTask); executionEndArbiter @@ -149,7 +139,7 @@ public async Task RunAsync_SleepsUsingEffectiveMonitorIntervalBetweenLoops() await monitor.RunAsync(TestContext.Current.CancellationToken); executionEndArbiter.Verify( - a => a.DelayMaintainerWithStopAwarenessAsync(TimeSpan.FromSeconds(3), + a => a.MaintainerDelayWaitAsync(TimeSpan.FromSeconds(3), It.IsAny(), It.IsAny(), TestContext.Current.CancellationToken), Times.Once); }