From 74f1daed5b757c1e442352c01a0935e55258e91c Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 12:20:45 -0700 Subject: [PATCH 01/20] Address timeout problem (tests pass), extract logic into separate method for later re-use --- .../Services/Jobs/JobRepository.cs | 117 ++++++++++++------ 1 file changed, 76 insertions(+), 41 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index c520865..a818932 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -76,6 +76,13 @@ internal sealed class JobRepository( /// private readonly AsyncManualResetEvent _jobsAvailableEvent = new(); + /// + /// Guards Set/Reset of together with enqueue onto + /// . + /// Lock order: then this gate. + /// + private readonly Lock _jobsAvailableGate = new(); + /// /// Signalled when the repository has no watched jobs OR the repository was unable to produce an inactive job for a /// worker request. @@ -150,11 +157,40 @@ private void NotifyWatchedJobsUpdate(int count) callbacks?.Invoke(count); } + /// + /// Align with whether inactive jobs or shortlisted jobs exist. + /// Assumed to run while holding . + /// + private void SyncJobsAvailableEvent() + { + lock (_jobsAvailableGate) + { + if (_inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty) + { + if (WatchedJobs.Count > 0) + { + // In-flight jobs remain. Do not park GetNextJobAsync waiters: they must + // keep observing ShouldKeepRunning() without a wait timeout. + _jobsAvailableEvent.Set(); + } + else + { + _jobsAvailableEvent.Reset(); + } + } + else + { + _jobsAvailableEvent.Set(); + } + } + } + private async Task TryGetUnblockedJobAsync(CancellationToken cancellationToken) { IJobRepositoryEntry? result; var iterated = false; + // ReSharper disable once InconsistentlySynchronizedField while (_unblockedJobsQueue.TryDequeue(out result)) { iterated = true; @@ -174,11 +210,7 @@ private async Task TryGetUnblockedJobAsync(CancellationToken await _inactiveJobsListSemaphore.WaitAsync(cancellationToken); try { - if (_inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty) - { - // Jobs are no longer available - _jobsAvailableEvent.Reset(); - } + SyncJobsAvailableEvent(); } finally { @@ -215,12 +247,7 @@ private async Task TryGetInactiveJobAsync(CancellationToken c if (result is not null) { _inactiveJobsList.RemoveAt(0); - - if (_inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty) - { - // Jobs are no longer available - _jobsAvailableEvent.Reset(); - } + SyncJobsAvailableEvent(); } } finally @@ -248,11 +275,14 @@ private void OnEntryStateUpdateUnblocked(IJobRepositoryEntry job, JobState? oldS return; } - // Identified as a newly-unblocked job. - // Shortlist the job for re-execution in memory. - _unblockedJobsQueue.Enqueue(job); - // Tell any active invocations of GetNextJobAsync that there is something available. - _jobsAvailableEvent.Set(); + lock (_jobsAvailableGate) + { + // Identified as a newly-unblocked job. + // Shortlist the job for re-execution in memory. + _unblockedJobsQueue.Enqueue(job); + // Tell any active invocations of GetNextJobAsync that there is something available. + _jobsAvailableEvent.Set(); + } } /// @@ -412,17 +442,17 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo do { // Try shortlist of unblocked jobs - if (await TryGetUnblockedJobAsync(cancellationToken) is {Success: true} unblockedAttemptResult) + if (await TryGetUnblockedJobAsync(cancellationToken) is {Success: true, Result: { } unblockedJob}) { - result = unblockedAttemptResult.Result!; + result = unblockedJob; // Continue out of loop iteration to abort via do-while condition continue; } - if (await TryGetInactiveJobAsync(cancellationToken) is {Success: true} inactiveAttemptResult) + if (await TryGetInactiveJobAsync(cancellationToken) is {Success: true, Result: { } inactiveAttemptResult}) { - result = inactiveAttemptResult.Result!; + result = inactiveAttemptResult; // Continue out of loop iteration to abort via do-while condition continue; @@ -431,14 +461,7 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo // If execution has reached here, then there are currently no available jobs to be handed out. // Is it because we've been asked to stop running? - if ( - // Note: Using the raw IExecutionEndArbiter because we want to avoid a circular dependency - !executionEndArbiter.ShouldKeepRunning() - // Confirm that the job loader thread has finished and will not be loading any more jobs - && jobLoaderStateService.IsLoaderFinished() - // Confirm that there are no more jobs in the background. - // This was already implied by the overall method structure, but now that the loader is finished we want to guarantee it - && await GetInactiveJobCountAsync(cancellationToken) == 0) + if (!HaveReasonToContinue()) { // It IS because we've been asked to stop running! // We have also confirmed that the job loader is fully finished, and no more jobs are incoming @@ -449,10 +472,8 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo // Only the JobLoader should care about this via the IJobRepository.WaitForJobDemandAsync method _jobsDemandEvent.Set(); - // Wait for jobs to arrive - // The milliseconds timeout is necessary due to timing problems that came up during unit testing - // I can't say that I'm thrilled with it, though... - await _jobsAvailableEvent.WaitAsync(TimeSpan.FromMilliseconds(250), cancellationToken); + // ReSharper disable once InconsistentlySynchronizedField + await _jobsAvailableEvent.WaitAsync(cancellationToken); } while (result is null); result.State = JobState.Active; @@ -460,6 +481,26 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo return result; } + private bool HaveReasonToContinue() + { + if ( + // If execution is still running, then we have every reason to believe that there will be more incoming jobs. + // Note: Using the raw IExecutionEndArbiter because we want to avoid a circular dependency. + executionEndArbiter.ShouldKeepRunning() + // If the job loader is not yet finished, then there may be more incoming jobs. + || !jobLoaderStateService.IsLoaderFinished()) + { + return true; + } + + lock (_tallyLock) + { + // Confirm whether there are any inactive jobs, or jobs that may become inactive again. + return _inactiveJobsTally > 0 + || _idempotencyBlockedTally > 0; + } + } + public async Task LoadAsync(IReadOnlyList intakeItems, CancellationToken cancellationToken = default) { @@ -512,14 +553,13 @@ public async Task LoadAsync(IReadOnlyList intakeItems, * * We're assuming that we're not working with enormous datasets for our backlog size. */ _inactiveJobsList = sorter.GetSortedListOfJobs(_inactiveJobsList); + SyncJobsAvailableEvent(); } finally { _inactiveJobsListSemaphore.Release(); } - _jobsAvailableEvent.Set(); - NotifyWatchedJobsUpdate(await GetWatchedJobsCountAsync(cancellationToken)); } @@ -530,13 +570,8 @@ public async Task RemoveJobAsync(IJobRepositoryEntry job, CancellationToken canc await _inactiveJobsListSemaphore.WaitAsync(cancellationToken); try { - if (_inactiveJobsList.Remove(job) - && _inactiveJobsList.Count == 0 - && _unblockedJobsQueue.IsEmpty) - { - // Jobs are no longer available - _jobsAvailableEvent.Reset(); - } + _inactiveJobsList.Remove(job); + SyncJobsAvailableEvent(); } finally { From 33cad93154638634da62b88f522542d7edcc8db3 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 12:38:22 -0700 Subject: [PATCH 02/20] Some organization --- .../Services/Jobs/JobRepository.cs | 73 ++++++++++--------- 1 file changed, 39 insertions(+), 34 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index a818932..1ebb1db 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -165,9 +165,17 @@ private void SyncJobsAvailableEvent() { lock (_jobsAvailableGate) { - if (_inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty) + bool isEmptyCondition; + int watchTally; + lock (_tallyLock) + { + isEmptyCondition = _inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty; + watchTally = _watchedJobsTally; + } + + if (isEmptyCondition) { - if (WatchedJobs.Count > 0) + if (watchTally > 0) { // In-flight jobs remain. Do not park GetNextJobAsync waiters: they must // keep observing ShouldKeepRunning() without a wait timeout. @@ -185,7 +193,7 @@ private void SyncJobsAvailableEvent() } } - private async Task TryGetUnblockedJobAsync(CancellationToken cancellationToken) + private TryGetJobResponse TryGetUnblockedJobAsync() { IJobRepositoryEntry? result; var iterated = false; @@ -207,15 +215,7 @@ private async Task TryGetUnblockedJobAsync(CancellationToken { // Check to see if we emptied the queue, but only if we actually dequeued something // Assume that the event is up to date and doesn't need a redundant reset. - await _inactiveJobsListSemaphore.WaitAsync(cancellationToken); - try - { - SyncJobsAvailableEvent(); - } - finally - { - _inactiveJobsListSemaphore.Release(); - } + SyncJobsAvailableEvent(); } if (result is null @@ -378,6 +378,31 @@ private void OnEntryStateUpdateTallies(IJobRepositoryEntry job, JobState? oldSta { NotifyWatchedJobsUpdate(localTallyWatched); } + + // Note: Although tallies are updated here, should not invoke SyncJobsAvailableEvent here. + // SyncJobsAvailableEvent reads off these tallies that suggest a state, but in practice the events are used + // for more concrete realities. Therefore, SyncJobsAvailableEvent should only be invoked when these + // sources of truth have been updated. + } + + private bool HaveReasonToExpectFutureJobs() + { + if ( + // If execution is still running, then we have every reason to believe that there will be more incoming jobs. + // Note: Using the raw IExecutionEndArbiter because we want to avoid a circular dependency. + executionEndArbiter.ShouldKeepRunning() + // If the job loader is not yet finished, then there may be more incoming jobs. + || !jobLoaderStateService.IsLoaderFinished()) + { + return true; + } + + lock (_tallyLock) + { + // Confirm whether there are any inactive jobs, or jobs that may become inactive again. + return _inactiveJobsTally > 0 + || _idempotencyBlockedTally > 0; + } } internal List WatchedJobs { get; } = []; @@ -442,7 +467,7 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo do { // Try shortlist of unblocked jobs - if (await TryGetUnblockedJobAsync(cancellationToken) is {Success: true, Result: { } unblockedJob}) + if (TryGetUnblockedJobAsync() is {Success: true, Result: { } unblockedJob}) { result = unblockedJob; @@ -461,7 +486,7 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo // If execution has reached here, then there are currently no available jobs to be handed out. // Is it because we've been asked to stop running? - if (!HaveReasonToContinue()) + if (!HaveReasonToExpectFutureJobs()) { // It IS because we've been asked to stop running! // We have also confirmed that the job loader is fully finished, and no more jobs are incoming @@ -481,26 +506,6 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo return result; } - private bool HaveReasonToContinue() - { - if ( - // If execution is still running, then we have every reason to believe that there will be more incoming jobs. - // Note: Using the raw IExecutionEndArbiter because we want to avoid a circular dependency. - executionEndArbiter.ShouldKeepRunning() - // If the job loader is not yet finished, then there may be more incoming jobs. - || !jobLoaderStateService.IsLoaderFinished()) - { - return true; - } - - lock (_tallyLock) - { - // Confirm whether there are any inactive jobs, or jobs that may become inactive again. - return _inactiveJobsTally > 0 - || _idempotencyBlockedTally > 0; - } - } - public async Task LoadAsync(IReadOnlyList intakeItems, CancellationToken cancellationToken = default) { From 16b28a5353e34485ae0b3861685b1eb00ae8170a Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 12:44:58 -0700 Subject: [PATCH 03/20] progress --- .../Services/Jobs/JobRepository.cs | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index 1ebb1db..ee28bb1 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -166,25 +166,17 @@ private void SyncJobsAvailableEvent() lock (_jobsAvailableGate) { bool isEmptyCondition; - int watchTally; lock (_tallyLock) { - isEmptyCondition = _inactiveJobsList.Count == 0 && _unblockedJobsQueue.IsEmpty; - watchTally = _watchedJobsTally; + isEmptyCondition = _inactiveJobsTally == 0; } + isEmptyCondition &= _unblockedJobsQueue.IsEmpty; + if (isEmptyCondition) { - if (watchTally > 0) - { - // In-flight jobs remain. Do not park GetNextJobAsync waiters: they must - // keep observing ShouldKeepRunning() without a wait timeout. - _jobsAvailableEvent.Set(); - } - else - { - _jobsAvailableEvent.Reset(); - } + // Job list is empty + _jobsAvailableEvent.Reset(); } else { From f4459e62285ab048767016995c4515c7016e327e Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 12:48:59 -0700 Subject: [PATCH 04/20] paranoid optimization --- .../Services/Jobs/JobRepository.cs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index ee28bb1..c95231f 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -121,6 +121,13 @@ internal sealed class JobRepository( private int _inactiveJobsTally; + /// + /// Notes if is set. + /// Created out of optimization paranoia to avoid unnecessary event sets/resets to . + /// Use should be gated behind . + /// + private bool _jobsAvailableEventIsSet; + private Action? _watchedJobsCallbacks; private int _watchedJobsTally; @@ -176,11 +183,18 @@ private void SyncJobsAvailableEvent() if (isEmptyCondition) { // Job list is empty - _jobsAvailableEvent.Reset(); + + // ReSharper disable once InvertIf + if (_jobsAvailableEventIsSet) + { + _jobsAvailableEvent.Reset(); + _jobsAvailableEventIsSet = false; + } } - else + else if (!_jobsAvailableEventIsSet) { _jobsAvailableEvent.Set(); + _jobsAvailableEventIsSet = true; } } } From 7919367a431276baca1d86061aeb61e756bcb361 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 12:57:32 -0700 Subject: [PATCH 05/20] comment --- .../Services/Jobs/JobRepository.cs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index c95231f..dc2ab9e 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -385,10 +385,13 @@ private void OnEntryStateUpdateTallies(IJobRepositoryEntry job, JobState? oldSta NotifyWatchedJobsUpdate(localTallyWatched); } - // Note: Although tallies are updated here, should not invoke SyncJobsAvailableEvent here. - // SyncJobsAvailableEvent reads off these tallies that suggest a state, but in practice the events are used - // for more concrete realities. Therefore, SyncJobsAvailableEvent should only be invoked when these - // sources of truth have been updated. + /* + * Note: Although tallies are updated here, SyncJobsAvailableEvent should not be invoked here. + * SyncJobsAvailableEvent reads off these tallies that suggest a state, but in practice the events are used + * for more concrete realities and tallies are set before these realities are implemented (read: before the + * inactive jobs list or unblocked jobs queue is updated). Therefore, SyncJobsAvailableEvent should only be + * invoked when these sources of truth have been updated. + */ } private bool HaveReasonToExpectFutureJobs() From 3b241093e9b3ebe14e7b852caa3d16702e63d797 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 13:00:20 -0700 Subject: [PATCH 06/20] line --- .../Services/Jobs/JobRepository.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index dc2ab9e..61a40b4 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -419,7 +419,6 @@ private bool HaveReasonToExpectFutureJobs() public async Task> GetAllInFlightJobsAsync(CancellationToken cancellationToken = default) { await _watchedJobsListSemaphore.WaitAsync(cancellationToken); - try { var items = WatchedJobs From 87f8b5f528ca6b39631a4d5cd686b094f37b333b Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 14:35:47 -0700 Subject: [PATCH 07/20] Fix order of operations before syncing event state --- .../Services/Jobs/JobRepository.cs | 121 +++++++++++++----- 1 file changed, 89 insertions(+), 32 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index 61a40b4..e8f4d4e 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -78,7 +78,7 @@ internal sealed class JobRepository( /// /// Guards Set/Reset of together with enqueue onto - /// . + /// and . /// Lock order: then this gate. /// private readonly Lock _jobsAvailableGate = new(); @@ -95,6 +95,11 @@ internal sealed class JobRepository( /// private readonly AsyncManualResetEvent _repositoryEmptyEvent = new(true); + /// + /// Guards Set/Reset of . + /// + private readonly Lock _repositoryEmptyGate = new(); + private readonly Lock _tallyLock = new(); /// @@ -128,6 +133,14 @@ internal sealed class JobRepository( /// private bool _jobsAvailableEventIsSet; + /// + /// Notes if is set. + /// Created out of optimization paranoia to avoid unnecessary event sets/resets to + /// . + /// Use should be gated behind . + /// + private bool _repositoryEmptyEventIsSet; + private Action? _watchedJobsCallbacks; private int _watchedJobsTally; @@ -168,6 +181,39 @@ private void NotifyWatchedJobsUpdate(int count) /// Align with whether inactive jobs or shortlisted jobs exist. /// Assumed to run while holding . /// + private void SyncRepositoryEmptyEvent() + { + lock (_repositoryEmptyGate) + { + bool isEmptyCondition; + lock (_tallyLock) + { + isEmptyCondition = _watchedJobsTally == 0; + } + + if (isEmptyCondition) + { + // Job list is empty + + // ReSharper disable once InvertIf + if (_repositoryEmptyEventIsSet) + { + _repositoryEmptyEvent.Reset(); + _repositoryEmptyEventIsSet = false; + } + } + else if (!_repositoryEmptyEventIsSet) + { + _repositoryEmptyEvent.Set(); + _repositoryEmptyEventIsSet = true; + } + } + } + + /// + /// Align with whether inactive jobs or shortlisted jobs exist. + /// Should be run after state has been updated and a data store has been updated. + /// private void SyncJobsAvailableEvent() { lock (_jobsAvailableGate) @@ -217,13 +263,6 @@ private TryGetJobResponse TryGetUnblockedJobAsync() } } - if (iterated) - { - // Check to see if we emptied the queue, but only if we actually dequeued something - // Assume that the event is up to date and doesn't need a redundant reset. - SyncJobsAvailableEvent(); - } - if (result is null // Account for technical race condition, will never happen in practice || result.IsDisposed) @@ -231,6 +270,7 @@ private TryGetJobResponse TryGetUnblockedJobAsync() return new TryGetJobResponse { Success = false, + ModifiedSourceStore = iterated, Result = null }; } @@ -238,6 +278,7 @@ private TryGetJobResponse TryGetUnblockedJobAsync() return new TryGetJobResponse { Success = true, + ModifiedSourceStore = iterated, Result = result }; } @@ -253,7 +294,6 @@ private async Task TryGetInactiveJobAsync(CancellationToken c if (result is not null) { _inactiveJobsList.RemoveAt(0); - SyncJobsAvailableEvent(); } } finally @@ -264,6 +304,7 @@ private async Task TryGetInactiveJobAsync(CancellationToken c return new TryGetJobResponse { Success = result is not null, + ModifiedSourceStore = result is not null, Result = result }; } @@ -474,21 +515,46 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo IJobRepositoryEntry? result = null; do { + Console.WriteLine("NEXT JOB ITERATION"); // Try shortlist of unblocked jobs - if (TryGetUnblockedJobAsync() is {Success: true, Result: { } unblockedJob}) + if (TryGetUnblockedJobAsync() is { } unblockedJobAttempt) { - result = unblockedJob; + if (unblockedJobAttempt is {Success: true, Result: { } unblockedJob}) + { + result = unblockedJob; + result.State = JobState.Active; + } - // Continue out of loop iteration to abort via do-while condition - continue; + if (unblockedJobAttempt.ModifiedSourceStore) + { + SyncJobsAvailableEvent(); + } + + if (result is not null) + { + // Continue out of loop iteration to abort via do-while condition + continue; + } } - if (await TryGetInactiveJobAsync(cancellationToken) is {Success: true, Result: { } inactiveAttemptResult}) + if (await TryGetInactiveJobAsync(cancellationToken) is { } inactiveAttempt) { - result = inactiveAttemptResult; + if (inactiveAttempt is {Success: true, Result: { } inactiveAttemptResult}) + { + result = inactiveAttemptResult; + result.State = JobState.Active; + } - // Continue out of loop iteration to abort via do-while condition - continue; + if (inactiveAttempt.ModifiedSourceStore) + { + SyncJobsAvailableEvent(); + } + + if (result is not null) + { + // Continue out of loop iteration to abort via do-while condition + continue; + } } // If execution has reached here, then there are currently no available jobs to be handed out. @@ -509,8 +575,6 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo await _jobsAvailableEvent.WaitAsync(cancellationToken); } while (result is null); - result.State = JobState.Active; - return result; } @@ -550,7 +614,7 @@ public async Task LoadAsync(IReadOnlyList intakeItems, _jobsDemandEvent.Reset(); // Once jobs are added, then the repository is either no longer empty or continues to not be empty. - _repositoryEmptyEvent.Reset(); + SyncRepositoryEmptyEvent(); } } finally @@ -603,7 +667,7 @@ public async Task RemoveJobAsync(IJobRepositoryEntry job, CancellationToken canc // If there's nothing to grab, then there must be an executor about to demand something. _jobsDemandEvent.Set(); // No watched jobs is the very definition of an empty repository - _repositoryEmptyEvent.Set(); + SyncRepositoryEmptyEvent(); } } finally @@ -673,18 +737,10 @@ public Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken return _jobsDemandEvent.WaitAsync(waitDuration, cancellationToken); } - public async Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) + public Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) { - int count; - do - { - // Short timeout mirrors GetNextJobAsync: avoids missing a Set/Reset edge under concurrency - await _repositoryEmptyEvent.WaitAsync(TimeSpan.FromMilliseconds(250), cancellationToken); - lock (_tallyLock) - { - count = _watchedJobsTally; - } - } while (count > 0); + Console.WriteLine("WaitForEmptyRepositoryAsync"); + return _repositoryEmptyEvent.WaitAsync(cancellationToken); } public int GetBacklogMaxCount() @@ -711,6 +767,7 @@ public async Task GetInactiveJobCountAsync(CancellationToken cancellationTo private sealed class TryGetJobResponse { public required bool Success { get; init; } + public required bool ModifiedSourceStore { get; init; } public required IJobRepositoryEntry? Result { get; init; } } From a846cffd79bcba8c9b5f76ecd1f00d207947cda8 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 14:57:23 -0700 Subject: [PATCH 08/20] comment phrasing --- .../Services/Jobs/JobRepository.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index e8f4d4e..08cc0f9 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -568,7 +568,7 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo } // Note that there's a demand. - // Only the JobLoader should care about this via the IJobRepository.WaitForJobDemandAsync method + // Only the loader mode should care about this via the IJobRepository.WaitForJobDemandAsync method _jobsDemandEvent.Set(); // ReSharper disable once InconsistentlySynchronizedField From 009e88f100a25356c3fa8fb677a80e01cc9c6a3e Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 15:33:14 -0700 Subject: [PATCH 09/20] Empty repository event logic fixes --- .../Services/Jobs/JobRepository.cs | 37 ++++++++++--------- 1 file changed, 20 insertions(+), 17 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index 08cc0f9..b5f27d5 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -128,18 +128,19 @@ internal sealed class JobRepository( /// /// Notes if is set. - /// Created out of optimization paranoia to avoid unnecessary event sets/resets to . + /// Created because of optimization paranoia to avoid unnecessary event sets/resets to + /// . /// Use should be gated behind . /// private bool _jobsAvailableEventIsSet; /// /// Notes if is set. - /// Created out of optimization paranoia to avoid unnecessary event sets/resets to - /// . + /// Created because of optimization paranoia to avoid unnecessary event sets/resets to + /// . /// Use should be gated behind . /// - private bool _repositoryEmptyEventIsSet; + private bool _repositoryEmptyEventIsSet = true; private Action? _watchedJobsCallbacks; private int _watchedJobsTally; @@ -196,16 +197,16 @@ private void SyncRepositoryEmptyEvent() // Job list is empty // ReSharper disable once InvertIf - if (_repositoryEmptyEventIsSet) + if (!_repositoryEmptyEventIsSet) { - _repositoryEmptyEvent.Reset(); - _repositoryEmptyEventIsSet = false; + _repositoryEmptyEvent.Set(); + _repositoryEmptyEventIsSet = true; } } - else if (!_repositoryEmptyEventIsSet) + else if (_repositoryEmptyEventIsSet) { - _repositoryEmptyEvent.Set(); - _repositoryEmptyEventIsSet = true; + _repositoryEmptyEvent.Reset(); + _repositoryEmptyEventIsSet = false; } } } @@ -592,7 +593,6 @@ public async Task LoadAsync(IReadOnlyList intakeItems, try { await _watchedJobsListSemaphore.WaitAsync(cancellationToken); - try { // ReSharper disable once ForeachCanBePartlyConvertedToQueryUsingAnotherGetEnumerator @@ -613,8 +613,6 @@ public async Task LoadAsync(IReadOnlyList intakeItems, WatchedJobs.Add(job); _jobsDemandEvent.Reset(); - // Once jobs are added, then the repository is either no longer empty or continues to not be empty. - SyncRepositoryEmptyEvent(); } } finally @@ -631,6 +629,7 @@ public async Task LoadAsync(IReadOnlyList intakeItems, */ _inactiveJobsList = sorter.GetSortedListOfJobs(_inactiveJobsList); SyncJobsAvailableEvent(); + SyncRepositoryEmptyEvent(); } finally { @@ -655,19 +654,19 @@ public async Task RemoveJobAsync(IJobRepositoryEntry job, CancellationToken canc _inactiveJobsListSemaphore.Release(); } + var watchedIsNowEmpty = false; await _watchedJobsListSemaphore.WaitAsync(cancellationToken); try { - WatchedJobs.Remove(job); - - if (WatchedJobs.Count == 0) + if (WatchedJobs.Remove(job) && WatchedJobs.Count == 0) { + // Just reached zero. // Avoid possible race condition in JobLoader // If there's nothing to grab, then there must be an executor about to demand something. _jobsDemandEvent.Set(); // No watched jobs is the very definition of an empty repository - SyncRepositoryEmptyEvent(); + watchedIsNowEmpty = true; } } finally @@ -676,6 +675,10 @@ public async Task RemoveJobAsync(IJobRepositoryEntry job, CancellationToken canc } job.Dispose(); + if (watchedIsNowEmpty) + { + SyncRepositoryEmptyEvent(); + } } public void SubscribeToInactiveCountUpdate(Action callback) From b8956fd336848057b03fe90167285765309c55d4 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 15:33:40 -0700 Subject: [PATCH 10/20] remove local debug prints --- .../Services/Jobs/JobRepository.cs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index b5f27d5..beddd47 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -516,7 +516,6 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo IJobRepositoryEntry? result = null; do { - Console.WriteLine("NEXT JOB ITERATION"); // Try shortlist of unblocked jobs if (TryGetUnblockedJobAsync() is { } unblockedJobAttempt) { @@ -742,7 +741,6 @@ public Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken public Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) { - Console.WriteLine("WaitForEmptyRepositoryAsync"); return _repositoryEmptyEvent.WaitAsync(cancellationToken); } From 6f97bfc910084efd62949a1c078a90e82e295a20 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 15:41:36 -0700 Subject: [PATCH 11/20] refactor to satisfy Sonar's cognitive complexity settings --- .../Services/Jobs/JobRepository.cs | 62 ++++++++----------- 1 file changed, 25 insertions(+), 37 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index beddd47..d627568 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -264,6 +264,14 @@ private TryGetJobResponse TryGetUnblockedJobAsync() } } + // Need to set state before thinking about syncing events + result?.State = JobState.Active; + + if (iterated) + { + SyncJobsAvailableEvent(); + } + if (result is null // Account for technical race condition, will never happen in practice || result.IsDisposed) @@ -271,7 +279,6 @@ private TryGetJobResponse TryGetUnblockedJobAsync() return new TryGetJobResponse { Success = false, - ModifiedSourceStore = iterated, Result = null }; } @@ -279,7 +286,6 @@ private TryGetJobResponse TryGetUnblockedJobAsync() return new TryGetJobResponse { Success = true, - ModifiedSourceStore = iterated, Result = result }; } @@ -302,10 +308,17 @@ private async Task TryGetInactiveJobAsync(CancellationToken c _inactiveJobsListSemaphore.Release(); } + // Need to set state before thinking about syncing events + result?.State = JobState.Active; + + if (result is not null) + { + SyncJobsAvailableEvent(); + } + return new TryGetJobResponse { Success = result is not null, - ModifiedSourceStore = result is not null, Result = result }; } @@ -517,44 +530,20 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo do { // Try shortlist of unblocked jobs - if (TryGetUnblockedJobAsync() is { } unblockedJobAttempt) + if (TryGetUnblockedJobAsync() is {Success: true, Result: { } formerlyUnblockedJobResult}) { - if (unblockedJobAttempt is {Success: true, Result: { } unblockedJob}) - { - result = unblockedJob; - result.State = JobState.Active; - } + result = formerlyUnblockedJobResult; - if (unblockedJobAttempt.ModifiedSourceStore) - { - SyncJobsAvailableEvent(); - } - - if (result is not null) - { - // Continue out of loop iteration to abort via do-while condition - continue; - } + // Continue out of loop iteration to abort via do-while condition + continue; } - if (await TryGetInactiveJobAsync(cancellationToken) is { } inactiveAttempt) + if (await TryGetInactiveJobAsync(cancellationToken) is + {Success: true, Result: { } formerlyInactiveJobResult}) { - if (inactiveAttempt is {Success: true, Result: { } inactiveAttemptResult}) - { - result = inactiveAttemptResult; - result.State = JobState.Active; - } - - if (inactiveAttempt.ModifiedSourceStore) - { - SyncJobsAvailableEvent(); - } - - if (result is not null) - { - // Continue out of loop iteration to abort via do-while condition - continue; - } + result = formerlyInactiveJobResult; + // Continue out of loop iteration to abort via do-while condition + continue; } // If execution has reached here, then there are currently no available jobs to be handed out. @@ -768,7 +757,6 @@ public async Task GetInactiveJobCountAsync(CancellationToken cancellationTo private sealed class TryGetJobResponse { public required bool Success { get; init; } - public required bool ModifiedSourceStore { get; init; } public required IJobRepositoryEntry? Result { get; init; } } From d3004e5ebe8e30541fa9e813a5dea90b2079aaac Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 16:09:53 -0700 Subject: [PATCH 12/20] groundwork for making WaitAsync shutdown-aware --- .../Services/Jobs/JobRepository.cs | 89 +++++++++++++++++-- 1 file changed, 83 insertions(+), 6 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index d627568..f2f76a7 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -15,7 +15,7 @@ namespace RedShirt.Example.JobWorker.Core.Services.Jobs; /// If you choose to use one polling mode when applying this template by pruning the other one, then you may want to /// prune in this method as well. /// -internal interface IJobRepository +internal interface IJobRepository : IDisposable { Task> GetAllIdempotencyBlockedJobsAsync(CancellationToken cancellationToken = default); Task> GetAllInFlightJobsAsync(CancellationToken cancellationToken = default); @@ -69,6 +69,9 @@ internal sealed class JobRepository( : IJobRepository { private readonly Lock _callbackLock = new(); + + private readonly CancellationTokenSource _cancellationTokenSource = new(); + private readonly Lock _generalGate = new(); private readonly SemaphoreSlim _inactiveJobsListSemaphore = new(1, 1); /// @@ -110,6 +113,8 @@ internal sealed class JobRepository( private readonly SemaphoreSlim _watchedJobsListSemaphore = new(1, 1); + private bool _disposed; + private Action? _idempotencyBlockedJobsCallbacks; private int _idempotencyBlockedTally; @@ -145,6 +150,45 @@ internal sealed class JobRepository( private Action? _watchedJobsCallbacks; private int _watchedJobsTally; + private void Dispose(bool disposing) + { + if (!disposing) + { + return; + } + + lock (_generalGate) + { + if (_disposed) + { + return; + } + + _disposed = true; + } + + _cancellationTokenSource.Cancel(); + _cancellationTokenSource.Dispose(); + } + + private CancellationToken GetLinkedToken(CancellationToken cancellationToken) + { + lock (_generalGate) + { + if (_disposed) + { + using var fallbackCts = + CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + fallbackCts.Cancel(); + return fallbackCts.Token; + } + } + + using var linkedCts = + CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationTokenSource.Token); + return linkedCts.Token; + } + private void NotifyInactiveCountUpdate(int count) { Action? callbacks; @@ -561,7 +605,16 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo _jobsDemandEvent.Set(); // ReSharper disable once InconsistentlySynchronizedField - await _jobsAvailableEvent.WaitAsync(cancellationToken); + var linkedToken = GetLinkedToken(cancellationToken); + try + { + await _jobsAvailableEvent.WaitAsync(linkedToken); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // Exception from a cancelled internal CTS suggests shutdown + // Suppress and let do-while loop continue to drain + } } while (result is null); return result; @@ -723,14 +776,31 @@ public void SubscribeToWatchedJobsUpdate(Action callback) } } - public Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default) + public async Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default) { - return _jobsDemandEvent.WaitAsync(waitDuration, cancellationToken); + var linkedToken = GetLinkedToken(cancellationToken); + try + { + return await _jobsDemandEvent.WaitAsync(waitDuration, linkedToken); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // Suggests exception from a cancelled internal CTS (which suggests shutdown) + return false; + } } - public Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) + public async Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) { - return _repositoryEmptyEvent.WaitAsync(cancellationToken); + var linkedToken = GetLinkedToken(cancellationToken); + try + { + await _repositoryEmptyEvent.WaitAsync(linkedToken); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // Suppress exception from a cancelled internal CTS (suggests shutdown) + } } public int GetBacklogMaxCount() @@ -754,6 +824,13 @@ public async Task GetInactiveJobCountAsync(CancellationToken cancellationTo } } + public void Dispose() + { + Dispose(true); + // ReSharper disable once GCSuppressFinalizeForTypeWithoutDestructor + GC.SuppressFinalize(this); + } + private sealed class TryGetJobResponse { public required bool Success { get; init; } From 102ff12e08492708dc134fa2554700e7466f6b75 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 16:28:20 -0700 Subject: [PATCH 13/20] checkpoint --- .../Services/Jobs/JobRepository.cs | 60 +++++++++++++++---- 1 file changed, 50 insertions(+), 10 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index f2f76a7..7480a41 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -61,18 +61,16 @@ Task LoadAsync(IReadOnlyList intakeItems, Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default); } -internal sealed class JobRepository( - IExecutionEndArbiter executionEndArbiter, - IJobLoaderStateReaderService jobLoaderStateService, - ISourceMessageSorter sorter, - IOptions options) - : IJobRepository +internal sealed class JobRepository : IJobRepository { private readonly Lock _callbackLock = new(); private readonly CancellationTokenSource _cancellationTokenSource = new(); + + private readonly IExecutionEndArbiter _executionEndArbiter; private readonly Lock _generalGate = new(); private readonly SemaphoreSlim _inactiveJobsListSemaphore = new(1, 1); + private readonly IJobLoaderStateReaderService _jobLoaderStateService; /// /// Indicates that jobs are available to be pulled by JobExecutor instances via GetNextJobAsync. @@ -92,6 +90,8 @@ internal sealed class JobRepository( /// private readonly AsyncManualResetEvent _jobsDemandEvent = new(); + private readonly IOptions _options; + /// /// Signalled when the repository has no watched jobs. /// Starts signalled because the repository begins empty. @@ -103,6 +103,8 @@ internal sealed class JobRepository( /// private readonly Lock _repositoryEmptyGate = new(); + private readonly ISourceMessageSorter _sorter; + private readonly Lock _tallyLock = new(); /// @@ -150,6 +152,31 @@ internal sealed class JobRepository( private Action? _watchedJobsCallbacks; private int _watchedJobsTally; + private void OnExecutionEndArbiterStop(Exception? exception) + { + ConsiderInterruptingEventWaits(); + } + + /// + /// Check to see if we should cancel the local CancellationTokenSource to interrupt method invocations that are waiting + /// on an event. + /// + private void ConsiderInterruptingEventWaits() + { + lock (_generalGate) + { + if (_disposed) + { + return; + } + + if (!HaveReasonToExpectFutureJobs()) + { + _cancellationTokenSource.Cancel(); + } + } + } + private void Dispose(bool disposing) { if (!disposing) @@ -498,9 +525,9 @@ private bool HaveReasonToExpectFutureJobs() if ( // If execution is still running, then we have every reason to believe that there will be more incoming jobs. // Note: Using the raw IExecutionEndArbiter because we want to avoid a circular dependency. - executionEndArbiter.ShouldKeepRunning() + _executionEndArbiter.ShouldKeepRunning() // If the job loader is not yet finished, then there may be more incoming jobs. - || !jobLoaderStateService.IsLoaderFinished()) + || !_jobLoaderStateService.IsLoaderFinished()) { return true; } @@ -513,6 +540,19 @@ private bool HaveReasonToExpectFutureJobs() } } + public JobRepository(IExecutionEndArbiter executionEndArbiter, + IJobLoaderStateReaderService jobLoaderStateReaderService, + ISourceMessageSorter sourceMessageSorter, + IOptions options) + { + _executionEndArbiter = executionEndArbiter; + _jobLoaderStateService = jobLoaderStateReaderService; + _sorter = sourceMessageSorter; + _options = options; + + executionEndArbiter.AddOnStopCallback(OnExecutionEndArbiterStop); + } + internal List WatchedJobs { get; } = []; public async Task> GetAllInFlightJobsAsync(CancellationToken cancellationToken = default) @@ -668,7 +708,7 @@ public async Task LoadAsync(IReadOnlyList intakeItems, * * Needs to be compatible with Batch mode, at least for the time being. * * We're assuming that we're not working with enormous datasets for our backlog size. */ - _inactiveJobsList = sorter.GetSortedListOfJobs(_inactiveJobsList); + _inactiveJobsList = _sorter.GetSortedListOfJobs(_inactiveJobsList); SyncJobsAvailableEvent(); SyncRepositoryEmptyEvent(); } @@ -805,7 +845,7 @@ public async Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToke public int GetBacklogMaxCount() { - return options.Value.EffectiveBacklogSize; + return _options.Value.EffectiveBacklogSize; } public async Task GetInactiveJobCountAsync(CancellationToken cancellationToken = default) From 51f940e84e935fe1c6844f8ba589bab9a71ad5fb Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 16:36:05 -0700 Subject: [PATCH 14/20] Update JobLoaderStateService with a callback, increase thread-safety just-in-case --- .../ExecutionState/JobLoaderStateService.cs | 72 +++++++++- .../JobLoaderStateServiceTests.cs | 136 +++++++++++++++--- 2 files changed, 178 insertions(+), 30 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs b/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs index 2573bff..77d68e1 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/ExecutionState/JobLoaderStateService.cs @@ -5,7 +5,12 @@ namespace RedShirt.Example.JobWorker.Core.Services.ExecutionState; /// internal interface IJobLoaderStateReaderService { - bool HasLoaderStarted(); + /// + /// Thread-safe addition of callback actions invoked once the loader has both started and stopped. + /// If the loader is already finished, the callback is invoked immediately. + /// + /// + void AddOnFinishCallback(Action callback); bool IsLoaderFinished(); } @@ -24,8 +29,7 @@ internal interface IJobLoaderStateService : IJobLoaderStateReaderService internal sealed class JobLoaderStateService : IJobLoaderStateService { /// - /// Multithreading protection. - /// Feels a little silly for a service that only sets booleans to true, but it makes automated audits happy. + /// Multithreading protection for start/stop flags and finish callbacks. /// private readonly Lock _lock = new(); @@ -33,35 +37,89 @@ internal sealed class JobLoaderStateService : IJobLoaderStateService private bool _isStarted; + private Action? _onFinishCallbacks; + + private bool IsFinishedUnsafe() + { + return _isStarted && _isFinished; + } + + /// + /// Detaches finish callbacks if the loader is finished. + /// Must be safe to call with or without the caller already holding + /// ( is reentrant). + /// + private Action? TakeCallbacksIfFinished() + { + lock (_lock) + { + if (!IsFinishedUnsafe()) + { + return null; + } + + return Interlocked.Exchange(ref _onFinishCallbacks, null); + } + } + + private static void InvokeCallbacks(Action? callbacks) + { + if (callbacks is null) + { + return; + } + + foreach (var invocation in callbacks.GetInvocationList()) + { + ((Action) invocation)(); + } + } + public void ReportLoaderStart() { + Action? callbacks; lock (_lock) { _isStarted = true; + callbacks = TakeCallbacksIfFinished(); } + + InvokeCallbacks(callbacks); } public void ReportLoaderStop() { + Action? callbacks; lock (_lock) { _isFinished = true; + callbacks = TakeCallbacksIfFinished(); } + + InvokeCallbacks(callbacks); } - public bool HasLoaderStarted() + public bool IsLoaderFinished() { lock (_lock) { - return _isStarted; + return IsFinishedUnsafe(); } } - public bool IsLoaderFinished() + public void AddOnFinishCallback(Action callback) { + ArgumentNullException.ThrowIfNull(callback); + lock (_lock) { - return _isStarted && _isFinished; + if (!IsFinishedUnsafe()) + { + _onFinishCallbacks += callback; + return; + } } + + callback(); } } \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs index 5c696e7..132a59b 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/ExecutionState/JobLoaderStateServiceTests.cs @@ -5,64 +5,154 @@ namespace RedShirt.Example.JobWorker.Core.UnitTests.Tests.Services.ExecutionStat // Generated by Cursor public class JobLoaderStateServiceTests { + [Fact] + public void AddOnFinishCallback_AfterStartAndStop_InvokesImmediately() + { + var service = new JobLoaderStateService(); + var invocations = 0; + + service.ReportLoaderStart(); + service.ReportLoaderStop(); + service.AddOnFinishCallback(() => invocations++); + + Assert.Equal(1, invocations); + } + + [Fact] + public void AddOnFinishCallback_AfterStartOnly_DoesNotInvokeUntilStop() + { + var service = new JobLoaderStateService(); + var invocations = 0; + + service.ReportLoaderStart(); + service.AddOnFinishCallback(() => invocations++); + + Assert.Equal(0, invocations); + + service.ReportLoaderStop(); + + Assert.Equal(1, invocations); + } + + [Fact] + public void AddOnFinishCallback_AfterStopOnly_DoesNotInvokeUntilStart() + { + var service = new JobLoaderStateService(); + var invocations = 0; + + service.ReportLoaderStop(); + service.AddOnFinishCallback(() => invocations++); + + Assert.Equal(0, invocations); + + service.ReportLoaderStart(); + + Assert.Equal(1, invocations); + } + /// - /// Concurrent start/stop/read calls should not throw and should eventually report finished. + /// Subscribers racing with start/stop must each run exactly once (immediate or deferred). /// [Fact(Timeout = 2000)] - public async Task TestConcurrentAccess_EventuallyFinished() + public async Task AddOnFinishCallback_ConcurrentSubscribe_EachRunsOnce() { var service = new JobLoaderStateService(); + var invocations = 0; var cancellationToken = TestContext.Current.CancellationToken; + const int subscriberCount = 64; - var startTask = Task.Run(() => + var subscribeTask = Task.Run(() => { - for (var i = 0; i < 100; i++) - { - service.ReportLoaderStart(); - _ = service.IsLoaderFinished(); - } + Parallel.For(0, subscriberCount, new ParallelOptions {CancellationToken = cancellationToken}, + _ => service.AddOnFinishCallback(() => Interlocked.Increment(ref invocations))); }, cancellationToken); - var stopTask = Task.Run(() => + var reportTask = Task.Run(() => { - for (var i = 0; i < 100; i++) - { - service.ReportLoaderStop(); - _ = service.IsLoaderFinished(); - } + service.ReportLoaderStart(); + service.ReportLoaderStop(); }, cancellationToken); - await Task.WhenAll(startTask, stopTask); + await Task.WhenAll(subscribeTask, reportTask); + Assert.Equal(subscriberCount, invocations); Assert.True(service.IsLoaderFinished()); } [Fact] - public void TestHasLoaderStarted_AfterStart_True() + public void AddOnFinishCallback_InvokesRegisteredCallbacksOnce() { var service = new JobLoaderStateService(); + var invocations = 0; + + service.AddOnFinishCallback(() => invocations++); + service.AddOnFinishCallback(() => invocations++); service.ReportLoaderStart(); + service.ReportLoaderStop(); + service.ReportLoaderStop(); - Assert.True(service.HasLoaderStarted()); + Assert.Equal(2, invocations); } - [Fact] - public void TestHasLoaderStarted_AfterStopOnly_False() + [Fact(Timeout = 2000)] + public async Task AddOnFinishCallback_IsThreadSafeAndRunsOnce() { var service = new JobLoaderStateService(); + var invocations = 0; + var cancellationToken = TestContext.Current.CancellationToken; - service.ReportLoaderStop(); + service.AddOnFinishCallback(() => Interlocked.Increment(ref invocations)); + + await Task.WhenAll( + Enumerable.Range(0, 32).Select(_ => Task.Run(() => + { + service.ReportLoaderStart(); + service.ReportLoaderStop(); + }, cancellationToken))); - Assert.False(service.HasLoaderStarted()); + Assert.Equal(1, invocations); + Assert.True(service.IsLoaderFinished()); } [Fact] - public void TestHasLoaderStarted_InitialState_False() + public void AddOnFinishCallback_WhenNull_ThrowsArgumentNullException() { var service = new JobLoaderStateService(); - Assert.False(service.HasLoaderStarted()); + Assert.Throws(() => service.AddOnFinishCallback(null!)); + } + + /// + /// Concurrent start/stop/read calls should not throw and should eventually report finished. + /// + [Fact(Timeout = 2000)] + public async Task TestConcurrentAccess_EventuallyFinished() + { + var service = new JobLoaderStateService(); + var cancellationToken = TestContext.Current.CancellationToken; + + var startTask = Task.Run(() => + { + for (var i = 0; i < 100; i++) + { + service.ReportLoaderStart(); + _ = service.IsLoaderFinished(); + } + }, cancellationToken); + + var stopTask = Task.Run(() => + { + for (var i = 0; i < 100; i++) + { + service.ReportLoaderStop(); + _ = service.IsLoaderFinished(); + } + }, cancellationToken); + + await Task.WhenAll(startTask, stopTask); + + Assert.True(service.IsLoaderFinished()); } /// From dc7002f64fefcf070fccafc7ff1422cab318301b Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 16:39:19 -0700 Subject: [PATCH 15/20] Finish implementing update callbacks, though Sonar is an issue again --- .../Services/Jobs/JobRepository.cs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index 7480a41..4074d96 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -511,6 +511,12 @@ private void OnEntryStateUpdateTallies(IJobRepositoryEntry job, JobState? oldSta NotifyWatchedJobsUpdate(localTallyWatched); } + if ((updatedInactive && localTallyInactive == 0) + || (updatedIdempotencyBlocked && localTallyIdempotencyBlocked == 0)) + { + ConsiderInterruptingEventWaits(); + } + /* * Note: Although tallies are updated here, SyncJobsAvailableEvent should not be invoked here. * SyncJobsAvailableEvent reads off these tallies that suggest a state, but in practice the events are used @@ -551,6 +557,7 @@ public JobRepository(IExecutionEndArbiter executionEndArbiter, _options = options; executionEndArbiter.AddOnStopCallback(OnExecutionEndArbiterStop); + jobLoaderStateReaderService.AddOnFinishCallback(ConsiderInterruptingEventWaits); } internal List WatchedJobs { get; } = []; From 1d66762de34368c7a42c75a2e07381723b0d238c Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 16:42:17 -0700 Subject: [PATCH 16/20] Address cognitive complexity --- .../Services/Jobs/JobRepository.cs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index 4074d96..3aafd00 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -496,14 +496,17 @@ private void OnEntryStateUpdateTallies(IJobRepositoryEntry job, JobState? oldSta } } + var consideringToConsiderCancellingWaitEvents = false; // Yes, this variable name is very silly if (updatedInactive) { NotifyInactiveCountUpdate(localTallyInactive); + consideringToConsiderCancellingWaitEvents |= localTallyInactive == 0; } if (updatedIdempotencyBlocked) { NotifyIdempotencyBlockedCountUpdate(localTallyIdempotencyBlocked); + consideringToConsiderCancellingWaitEvents |= localTallyIdempotencyBlocked == 0; } if (updatedWatched) @@ -511,8 +514,7 @@ private void OnEntryStateUpdateTallies(IJobRepositoryEntry job, JobState? oldSta NotifyWatchedJobsUpdate(localTallyWatched); } - if ((updatedInactive && localTallyInactive == 0) - || (updatedIdempotencyBlocked && localTallyIdempotencyBlocked == 0)) + if (consideringToConsiderCancellingWaitEvents) { ConsiderInterruptingEventWaits(); } From 6e447ed4f5ce06306b0316bd5bc2a93128e0dc96 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 16:47:11 -0700 Subject: [PATCH 17/20] Address callback-related test failures. Deliberately fixing errors commit by commit to closely monitor test changes. --- .../Tests/Services/Jobs/JobRepositoryTests.cs | 86 ++++++++++++++++++- 1 file changed, 85 insertions(+), 1 deletion(-) diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs index e9098ca..ad2fac9 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs @@ -12,6 +12,22 @@ namespace RedShirt.Example.JobWorker.Core.UnitTests.Tests.Services.Jobs; public class JobRepositoryTests { + private static void SetupConstructionCallbacks( + Mock executionEndArbiter, + Mock jobLoaderStateService) + { + executionEndArbiter.Setup(a => a.AddOnStopCallback(It.IsAny>())); + jobLoaderStateService.Setup(s => s.AddOnFinishCallback(It.IsAny())); + } + + private static void VerifyConstructionCallbacks( + Mock executionEndArbiter, + Mock jobLoaderStateService) + { + executionEndArbiter.Verify(a => a.AddOnStopCallback(It.IsAny>()), Times.Once); + jobLoaderStateService.Verify(s => s.AddOnFinishCallback(It.IsAny()), Times.Once); + } + private static JobRepository CreateRepository( Mock? executionEndArbiter = null, Mock? jobLoaderStateService = null, @@ -25,11 +41,17 @@ private static JobRepository CreateRepository( .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); - return new JobRepository( + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + + var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(new JobRepository.ConfigurationModel {BacklogSize = backlogSize})); + + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + + return jobRepository; } private static Mock CreateJobModel(string messageId) @@ -39,6 +61,26 @@ private static Mock CreateJobModel(string messageId) return jobModel; } + [Fact] + public void Constructor_RegistersStopAndFinishCallbacks() + { + var executionEndArbiter = new Mock(MockBehavior.Strict); + var jobLoaderStateService = new Mock(MockBehavior.Strict); + var sorter = new Mock(MockBehavior.Strict); + + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + + _ = new JobRepository( + executionEndArbiter.Object, + jobLoaderStateService.Object, + sorter.Object, + Options.Create(new JobRepository.ConfigurationModel {BacklogSize = 0})); + + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); + executionEndArbiter.VerifyNoOtherCalls(); + jobLoaderStateService.VerifyNoOtherCalls(); + } + [Fact(Timeout = 2000)] public async Task GetNextJobAsync_WhenOnlyUnblockedJobIsDisposed_FallsBackToInactiveQueue() { @@ -130,6 +172,7 @@ public async Task LoadAsync_WhenResponseHasNoItems_DoesNotTouchWatchedJobs() var jobLoaderStateService = new Mock(MockBehavior.Strict); var sorter = new Mock(MockBehavior.Strict); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, @@ -138,6 +181,7 @@ public async Task LoadAsync_WhenResponseHasNoItems_DoesNotTouchWatchedJobs() { BacklogSize = 0 })); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); await jobRepository.LoadAsync([], TestContext.Current.CancellationToken); @@ -213,11 +257,13 @@ public async Task TestGetAllIdempotencyBlockedJobsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var expectedBlockedJobs = new List>(); @@ -287,11 +333,13 @@ public async Task TestGetAllInFlightJobsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var expectedJobs = new List>(); @@ -366,11 +414,13 @@ public void TestGetBacklogMaxCount(int backlogSize, int expectedEffectiveBatchSi .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); Assert.Equal(expectedEffectiveBatchSize, jobRepository.GetBacklogMaxCount()); } @@ -392,10 +442,12 @@ public async Task TestGetCountsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); Mock job; @@ -451,11 +503,13 @@ public async Task TestGetNextJobAsync_Null() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); Assert.Null(await jobRepository.GetNextJobAsync(TestContext.Current.CancellationToken)); @@ -482,11 +536,13 @@ public async Task TestLoadJobs(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -550,11 +606,13 @@ public async Task TestLoadJobsAndWaitForJob(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -623,11 +681,13 @@ public async Task TestLoadJobsAndWaitForJob_RequeuedBacklog() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -720,11 +780,13 @@ public async Task TestLoadJobsAndWaitForJob_RequeuedBacklogThenEmpty() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -849,11 +911,13 @@ public async Task TestLoadJobsAndWaitForJob_ThenRemove() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -947,11 +1011,13 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -1056,11 +1122,13 @@ public async Task TestLoadJobsAndWaitForJob_VerifySetToActive(int responseSize) .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var envelopes = new List(); @@ -1123,11 +1191,13 @@ public async Task TestRemoveJobsAsync() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var job = new Mock(); job.SetupProperty(j => j.State, JobState.Active); @@ -1161,11 +1231,13 @@ public async Task TestRemoveJobsAsyncB() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var job = new Mock(); job.SetupProperty(j => j.State, JobState.Active); @@ -1209,11 +1281,13 @@ public async Task TestWaitForDemand_False() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); // Start waiting for there to be a job demand var stopwatch = Stopwatch.StartNew(); @@ -1247,11 +1321,13 @@ public async Task TestWaitForDemand_True() var sorter = new Mock(MockBehavior.Strict); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); // Start waiting for there to be a job demand var demandTask = Task.Run( @@ -1280,6 +1356,7 @@ public async Task UnblockingJob_ShortlistsJobAheadOfInactiveQueue() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, @@ -1288,6 +1365,7 @@ public async Task UnblockingJob_ShortlistsJobAheadOfInactiveQueue() { BacklogSize = 0 })); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var unblockedModel = new Mock(MockBehavior.Strict); unblockedModel.Setup(m => m.MessageId).Returns("unblocked"); @@ -1343,11 +1421,13 @@ public async Task WaitForEmptyRepositoryAsync_CompletesWhenLastJobRemoved() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobModel1 = new Mock(MockBehavior.Strict); jobModel1.Setup(m => m.MessageId).Returns(Guid.NewGuid().ToString()); @@ -1412,11 +1492,13 @@ public async Task WaitForEmptyRepositoryAsync_HonorsCancellation() .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobModel = new Mock(MockBehavior.Strict); jobModel.Setup(m => m.MessageId).Returns(Guid.NewGuid().ToString()); @@ -1458,11 +1540,13 @@ public async Task WaitForEmptyRepositoryAsync_WhenAlreadyEmpty_ReturnsImmediatel .Setup(s => s.GetSortedListOfJobs(It.IsAny>())) .Returns((List input) => input); + SetupConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var jobRepository = new JobRepository( executionEndArbiter.Object, jobLoaderStateService.Object, sorter.Object, Options.Create(options)); + VerifyConstructionCallbacks(executionEndArbiter, jobLoaderStateService); var stopwatch = Stopwatch.StartNew(); await jobRepository.WaitForEmptyRepositoryAsync(TestContext.Current.CancellationToken); From 92e0d9c96da286060e4010536a528915d21305f9 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 18:05:03 -0700 Subject: [PATCH 18/20] Fix remaining JobRepository tests --- .../Services/Jobs/JobRepository.cs | 141 ++++++++++++++---- .../Tests/Services/Jobs/JobRepositoryTests.cs | 31 +++- 2 files changed, 141 insertions(+), 31 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index 3aafd00..898c34c 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -198,24 +198,6 @@ private void Dispose(bool disposing) _cancellationTokenSource.Dispose(); } - private CancellationToken GetLinkedToken(CancellationToken cancellationToken) - { - lock (_generalGate) - { - if (_disposed) - { - using var fallbackCts = - CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - fallbackCts.Cancel(); - return fallbackCts.Token; - } - } - - using var linkedCts = - CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationTokenSource.Token); - return linkedCts.Token; - } - private void NotifyInactiveCountUpdate(int count) { Action? callbacks; @@ -548,6 +530,51 @@ private bool HaveReasonToExpectFutureJobs() } } + /// + /// Wait for available jobs. + /// This separate method isn't strictly necessary, even for Sonar complexity warnings, + /// but it makes much more readable. + /// + /// + /// + private async Task WaitForAvailableJobsAsync(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + + /* + * Construct a linked CTS to tie it to _cancellationTokenSource. + * + * Note: During development, I experimented with a GetLinkedToken method in order + * to centralize some of this, but that ended up being invalid. The reason for + * that is that disposing a linked source unregisters it from _cancellationTokenSource. + * The CancellationToken that was being returned was no longer hooked to that source, + * making the end result just the baseline cancellation token with extra steps. + */ + + CancellationTokenSource? linkedCts = null; + try + { + CancellationToken linkedToken; + lock (_generalGate) + { + if (_disposed) + { + throw new OperationCanceledException(); + } + + linkedCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, _cancellationTokenSource.Token); + linkedToken = linkedCts.Token; + } + + await _jobsAvailableEvent.WaitAsync(linkedToken); + } + finally + { + linkedCts?.Dispose(); + } + } + public JobRepository(IExecutionEndArbiter executionEndArbiter, IJobLoaderStateReaderService jobLoaderStateReaderService, ISourceMessageSorter sourceMessageSorter, @@ -653,16 +680,15 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo // Only the loader mode should care about this via the IJobRepository.WaitForJobDemandAsync method _jobsDemandEvent.Set(); - // ReSharper disable once InconsistentlySynchronizedField - var linkedToken = GetLinkedToken(cancellationToken); try { - await _jobsAvailableEvent.WaitAsync(linkedToken); + await WaitForAvailableJobsAsync(cancellationToken); } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { // Exception from a cancelled internal CTS suggests shutdown - // Suppress and let do-while loop continue to drain + // Manually break from loop so that invoking executors can abort + break; } } while (result is null); @@ -827,28 +853,87 @@ public void SubscribeToWatchedJobsUpdate(Action callback) public async Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default) { - var linkedToken = GetLinkedToken(cancellationToken); + cancellationToken.ThrowIfCancellationRequested(); + + /* + * Construct a linked CTS to tie it to _cancellationTokenSource. + * + * Note: During development, I experimented with a GetLinkedToken method in order + * to centralize some of this, but that ended up being invalid. The reason for + * that is that disposing a linked source unregisters it from _cancellationTokenSource. + * The CancellationToken that was being returned was no longer hooked to that source, + * making the end result just the baseline cancellation token with extra steps. + */ + try { - return await _jobsDemandEvent.WaitAsync(waitDuration, linkedToken); + CancellationTokenSource? linkedCts = null; + try + { + CancellationToken linkedToken; + lock (_generalGate) + { + if (_disposed) + { + throw new OperationCanceledException(); + } + + linkedCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, _cancellationTokenSource.Token); + linkedToken = linkedCts.Token; + } + + return await _jobsDemandEvent.WaitAsync(waitDuration, linkedToken); + } + finally + { + linkedCts?.Dispose(); + } } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { // Suggests exception from a cancelled internal CTS (which suggests shutdown) - return false; + // Pass } + + return false; } public async Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) { - var linkedToken = GetLinkedToken(cancellationToken); + cancellationToken.ThrowIfCancellationRequested(); + + /* + * Construct a linked CTS to tie it to _cancellationTokenSource. + * + * Note: During development, I experimented with a GetLinkedToken method in order + * to centralize some of this, but that ended up being invalid. The reason for + * that is that disposing a linked source unregisters it from _cancellationTokenSource. + * The CancellationToken that was being returned was no longer hooked to that source, + * making the end result just the baseline cancellation token with extra steps. + */ + + CancellationTokenSource? linkedCts = null; try { + CancellationToken linkedToken; + lock (_generalGate) + { + if (_disposed) + { + throw new OperationCanceledException(); + } + + linkedCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, _cancellationTokenSource.Token); + linkedToken = linkedCts.Token; + } + await _repositoryEmptyEvent.WaitAsync(linkedToken); } - catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + finally { - // Suppress exception from a cancelled internal CTS (suggests shutdown) + linkedCts?.Dispose(); } } diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs index ad2fac9..81d25f1 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs @@ -7,6 +7,7 @@ using RedShirt.Example.JobWorker.Core.Services.SourceMessages; using RedShirt.Example.JobWorker.Core.Utility; using System.Diagnostics; +using Range = Moq.Range; namespace RedShirt.Example.JobWorker.Core.UnitTests.Tests.Services.Jobs; @@ -35,6 +36,7 @@ private static JobRepository CreateRepository( int backlogSize = 10) { executionEndArbiter ??= new Mock(MockBehavior.Strict); + executionEndArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); jobLoaderStateService ??= new Mock(MockBehavior.Strict); sorter ??= new Mock(); sorter @@ -999,7 +1001,8 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) var jobLoaderStateService = new Mock(MockBehavior.Strict); jobLoaderStateService .Setup(s => s.IsLoaderFinished()) - .Returns(true); + // ReSharper disable once AccessToModifiedClosure + .Returns(() => readyToEnd); var options = new JobRepository.ConfigurationModel { @@ -1073,9 +1076,27 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) DateTime.UtcNow + TimeSpan.FromMilliseconds(250)); } - await Task.Delay(TimeSpan.FromMilliseconds(250), TestContext.Current.CancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); + // Confirm that the task hasn't yet finished after that delay + Assert.False(retrievedJobsTask.IsCompleted); + // Mark things as ready to finish readyToEnd = true; + var jobLoaderStateServiceCallbackInvocation = Assert.Single(jobLoaderStateService.Invocations, + i => i.Method.Name == nameof(jobLoaderStateService.Object.AddOnFinishCallback)); + // At this point I'm just having fun with the ridiculous variable names in this test method. + var jobLoaderStateServiceCallbackInvocationArgumentAsCallback = + jobLoaderStateServiceCallbackInvocation.Arguments[0] as Action; + Assert.NotNull(jobLoaderStateServiceCallbackInvocationArgumentAsCallback); + jobLoaderStateServiceCallbackInvocationArgumentAsCallback(); + + /* + * Now that we've cancelled the mock-application, the task should be finishable through the callbacks. + */ + await Task.Delay(TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); var retrievedJobs = await retrievedJobsTask; + + /* Verify Results */ + Assert.NotNull(retrievedJobs); Assert.NotEmpty(retrievedJobs); @@ -1091,7 +1112,10 @@ public async Task TestLoadJobsAndWaitForJob_UntilEmpty(int responseSize) Assert.Same(expectedEnvelope.RawJobModel, currentJob.RawJobModel); } - jobLoaderStateService.Verify(s => s.IsLoaderFinished(), Times.Once); + // Verify that state service was called a reasonable number of times (1-4). + // It's going to be at least once, during some debugging with some extra statements it was 4. + // At this point in the test, we're happy as long as it wasn't invoked a million times + jobLoaderStateService.Verify(s => s.IsLoaderFinished(), Times.Between(1, 4, Range.Inclusive)); } /// @@ -1411,6 +1435,7 @@ await jobRepository.LoadAsync( public async Task WaitForEmptyRepositoryAsync_CompletesWhenLastJobRemoved() { var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter.Setup(a => a.ShouldKeepRunning()).Returns(true); var jobLoaderStateService = new Mock(MockBehavior.Strict); var options = new JobRepository.ConfigurationModel { From f6fb8e7bcc1ac1272928b71b22d44264bc8f18d6 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 20:52:24 -0700 Subject: [PATCH 19/20] Address warning elsewhere in JobRepositoryTests --- .../Tests/Services/Jobs/JobRepositoryTests.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs index 81d25f1..084d963 100644 --- a/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs +++ b/test/RedShirt.Example.JobWorker.Core.UnitTests/Tests/Services/Jobs/JobRepositoryTests.cs @@ -1427,6 +1427,7 @@ await jobRepository.LoadAsync( var nextJob = await jobRepository.GetNextJobAsync(TestContext.Current.CancellationToken); + Assert.NotNull(nextJob); Assert.Same(unblockedEntry, nextJob); Assert.Equal(JobState.Active, nextJob.State); } From b22b2ddc90d8342a13243b12ea2bb687f926d774 Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Sat, 22 Aug 2026 21:08:52 -0700 Subject: [PATCH 20/20] Centralize linked token operations into private wrapper method. --- .../Services/Jobs/JobRepository.cs | 104 +++++------------- 1 file changed, 26 insertions(+), 78 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs index 898c34c..14606ea 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/JobRepository.cs @@ -530,23 +530,17 @@ private bool HaveReasonToExpectFutureJobs() } } - /// - /// Wait for available jobs. - /// This separate method isn't strictly necessary, even for Sonar complexity warnings, - /// but it makes much more readable. - /// - /// - /// - private async Task WaitForAvailableJobsAsync(CancellationToken cancellationToken) + private async Task DoOperationWithLinkedToken(Func operation, + CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); /* * Construct a linked CTS to tie it to _cancellationTokenSource. * - * Note: During development, I experimented with a GetLinkedToken method in order - * to centralize some of this, but that ended up being invalid. The reason for - * that is that disposing a linked source unregisters it from _cancellationTokenSource. + * Note: During development, I experimented with a GetLinkedToken method, + * but that ended up being invalid. The reason it was invalid is that + * disposing a linked source unregisters it from _cancellationTokenSource. * The CancellationToken that was being returned was no longer hooked to that source, * making the end result just the baseline cancellation token with extra steps. */ @@ -567,7 +561,7 @@ private async Task WaitForAvailableJobsAsync(CancellationToken cancellationToken linkedToken = linkedCts.Token; } - await _jobsAvailableEvent.WaitAsync(linkedToken); + await operation(linkedToken); } finally { @@ -575,6 +569,16 @@ private async Task WaitForAvailableJobsAsync(CancellationToken cancellationToken } } + private Task DoWaitForAvailableJobsAsync(CancellationToken cancellationToken) + { + return _jobsAvailableEvent.WaitAsync(cancellationToken); + } + + private Task DoWaitForEmptyRepositoryAsync(CancellationToken cancellationToken) + { + return _repositoryEmptyEvent.WaitAsync(cancellationToken); + } + public JobRepository(IExecutionEndArbiter executionEndArbiter, IJobLoaderStateReaderService jobLoaderStateReaderService, ISourceMessageSorter sourceMessageSorter, @@ -682,7 +686,7 @@ public async Task GetWatchedJobsCountAsync(CancellationToken cancellationTo try { - await WaitForAvailableJobsAsync(cancellationToken); + await DoOperationWithLinkedToken(DoWaitForAvailableJobsAsync, cancellationToken); } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { @@ -853,87 +857,31 @@ public void SubscribeToWatchedJobsUpdate(Action callback) public async Task WaitForJobDemandAsync(TimeSpan waitDuration, CancellationToken cancellationToken = default) { - cancellationToken.ThrowIfCancellationRequested(); - - /* - * Construct a linked CTS to tie it to _cancellationTokenSource. - * - * Note: During development, I experimented with a GetLinkedToken method in order - * to centralize some of this, but that ended up being invalid. The reason for - * that is that disposing a linked source unregisters it from _cancellationTokenSource. - * The CancellationToken that was being returned was no longer hooked to that source, - * making the end result just the baseline cancellation token with extra steps. - */ + var result = false; try { - CancellationTokenSource? linkedCts = null; - try - { - CancellationToken linkedToken; - lock (_generalGate) - { - if (_disposed) - { - throw new OperationCanceledException(); - } - - linkedCts = CancellationTokenSource.CreateLinkedTokenSource( - cancellationToken, _cancellationTokenSource.Token); - linkedToken = linkedCts.Token; - } - - return await _jobsDemandEvent.WaitAsync(waitDuration, linkedToken); - } - finally - { - linkedCts?.Dispose(); - } + await DoOperationWithLinkedToken( + async linkedToken => { result = await _jobsDemandEvent.WaitAsync(waitDuration, linkedToken); }, + cancellationToken); } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { - // Suggests exception from a cancelled internal CTS (which suggests shutdown) - // Pass + // Pass, use default false } - return false; + return result; } public async Task WaitForEmptyRepositoryAsync(CancellationToken cancellationToken = default) { - cancellationToken.ThrowIfCancellationRequested(); - - /* - * Construct a linked CTS to tie it to _cancellationTokenSource. - * - * Note: During development, I experimented with a GetLinkedToken method in order - * to centralize some of this, but that ended up being invalid. The reason for - * that is that disposing a linked source unregisters it from _cancellationTokenSource. - * The CancellationToken that was being returned was no longer hooked to that source, - * making the end result just the baseline cancellation token with extra steps. - */ - - CancellationTokenSource? linkedCts = null; try { - CancellationToken linkedToken; - lock (_generalGate) - { - if (_disposed) - { - throw new OperationCanceledException(); - } - - linkedCts = CancellationTokenSource.CreateLinkedTokenSource( - cancellationToken, _cancellationTokenSource.Token); - linkedToken = linkedCts.Token; - } - - await _repositoryEmptyEvent.WaitAsync(linkedToken); + await DoOperationWithLinkedToken(DoWaitForEmptyRepositoryAsync, cancellationToken); } - finally + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { - linkedCts?.Dispose(); + // Pass } }