Rework Repository Entry Tracking

README.md

@@ -65,7 +65,7 @@ The general architecture of the core application goes along these lines:
     * For this general template, the job logic runner implementation is currently a stub that shall sleep for the
       requested number of seconds.
 * In the event that the job source implementation requires application-level heartbeats for long-running messages, a
-  heartbeat maintainer (`IHeartbeatMaintainer`) worker thread shall periodically heartbeat messages according to the job
+  heartbeat monitor (`IHeartbeatMonitor`) worker thread shall periodically heartbeat messages according to the job
   source's recommendation.
     * Generally, job sources that require heartbeats are told to recommend a heartbeat interval of 75% of the maximum
       in-flight time for a message without heartbeats. For example, an SQS consumer configured with a visibility timeout

src/RedShirt.Example.JobWorker.Core/Extensions/ServiceCollectionExtensions.cs

@@ -28,24 +28,23 @@ public static IServiceCollection AddCoreJobManagement(this IServiceCollection se
 
         services = services
             .AddCommon()
+            // Execution end arbiters
+            .AddSingleton<IExecutionEndArbiter, ExecutionEndArbiter>()
+            .AddSingleton<IExecutorExecutionEndArbiter, ExecutorExecutionEndArbiter>()
+            .AddSingleton<IHeartbeatMonitorExecutionEndArbiter, HeartbeatMonitorExecutionEndArbiter>()
+            .AddSingleton<IIdempotencyMonitorExecutionEndArbiter, IdempotencyMonitorExecutionEndArbiter>()
             // General
             .AddSingleton<IHandler, Handler>()
             .AddSingleton<IJobLoaderLoop, JobLoaderLoop>()
             .AddSingleton<IJobSubscriberManager, JobSubscriberManager>()
             .AddSingleton<IJobSubscriberIntakeQueue, JobSubscriberIntakeQueue>()
             .AddSingleton<IJobExecutor, JobExecutor>()
-            .AddSingleton<AppliedExecutionEndArbiter>()
-            .AddSingleton<IAppliedMaintainerExecutionEndArbiter>(provider =>
-                provider.GetRequiredService<AppliedExecutionEndArbiter>())
-            .AddSingleton<IAppliedExecutorExecutionEndArbiter>(provider =>
-                provider.GetRequiredService<AppliedExecutionEndArbiter>())
-            .AddSingleton<IHeartbeatMaintainer, HeartbeatMaintainer>()
+            .AddSingleton<IHeartbeatMonitor, HeartbeatMonitor>()
             .AddSingleton<IHeartbeatCalculator, HeartbeatCalculator>()
             .AddSingleton<ISafeJobRunner, SafeJobRunner>()
             .AddSingleton<ITimeBorderWrapperService, TimeBorderWrapperService>()
             .AddSingleton<ISafeJobAcknowledgementService, SafeJobAcknowledgementService>()
             .AddSingleton<IJobIntakeService, JobIntakeService>()
-            .AddSingleton<IExecutionEndArbiter, ExecutionEndArbiter>()
             .AddSingleton<IJobRepository, JobRepository>()
             .AddSingleton<ICoreConfigurationService, CoreConfigurationService>()
             .Configure<JobRepository.ConfigurationModel>(coreSection)

src/RedShirt.Example.JobWorker.Core/Models/JobRepositoryEntry.cs

@@ -8,96 +8,192 @@ internal interface ISortableJobWrapper
     IJobModel JobModel { get; }
 }
 
-internal interface IJobRepositoryEntry : ISortableJobWrapper
+internal interface IJobRepositoryEntry : ISortableJobWrapper, IDisposable
 {
+    bool IsDisposed { get; }
     IRawJobModel RawJobModel { get; }
-    bool CanHeartbeat { get; }
-    DateTime LastHeartbeatTime { get; }
-    JobState State { get; }
 
-    Task SetAsCannotHeartbeatAsync(CancellationToken cancellationToken = default);
-    Task SetLastHeartbeatTimeAsync(DateTime lastHeartbeatTime, CancellationToken cancellationToken = default);
-    Task SetStateAsync(JobState state, CancellationToken cancellationToken = default);
+    /// <summary>
+    ///     Whether this job can still receive heartbeats.
+    ///     May only be set to <c>false</c>.
+    /// </summary>
+    bool CanHeartbeat { get; set; }
+
+    DateTime LastHeartbeatTime { get; set; }
+
+    /// <summary>
+    ///     Current processing state of this job.
+    ///     May not be set to <c>null</c>.
+    /// </summary>
+    /// <exception cref="ArgumentNullException">Thrown when the setter is given <c>null</c>.</exception>
+    JobState? State { get; set; }
 }
 
 internal sealed class JobRepositoryEntry : IJobRepositoryEntry
 {
     /// <summary>
-    ///     Thread-safety for mutable field access from maintainer, executor, and repository threads.
+    ///     Thread-safety for mutable field access from executor, monitor, and repository threads.
     /// </summary>
     private readonly Lock _lock = new();
 
-    private bool _canHeartbeat = true;
-    private DateTime _lastHeartbeatTime;
-    private JobState _state = JobState.Inactive;
+    private bool _disposed;
 
-    public required IRawJobModel RawJobModel { get; init; }
-    public required IJobModel JobModel { get; init; }
+    private Action<IJobRepositoryEntry, JobState?, JobState>? _stateCallbacks;
 
-    public bool CanHeartbeat
+    private void Dispose(bool disposing)
     {
-        get
+        if (!disposing)
         {
-            lock (_lock)
+            return;
+        }
+
+        lock (_lock)
+        {
+            if (_disposed)
             {
-                return _canHeartbeat;
+                return;
             }
+
+            _disposed = true;
+        }
+
+        // Confirm that the job that we're removing is marked as complete,
+        //   for the sake of subscriber callbacks in the underlying JobRepositoryEntry.
+        State = JobState.Complete;
+
+        lock (_lock)
+        {
+            _stateCallbacks = null;
         }
     }
 
-    public DateTime LastHeartbeatTime
+    /// <summary>
+    ///     Threadsafe indicator of being disposed.
+    /// </summary>
+    public bool IsDisposed
     {
         get
         {
             lock (_lock)
             {
-                return _lastHeartbeatTime;
+                return _disposed;
             }
         }
     }
 
-    public JobState State
+    public void Dispose()
+    {
+        Dispose(true);
+        // ReSharper disable once GCSuppressFinalizeForTypeWithoutDestructor
+        GC.SuppressFinalize(this);
+    }
+
+    public required IRawJobModel RawJobModel { get; init; }
+    public required IJobModel JobModel { get; init; }
+
+    public bool CanHeartbeat
     {
         get
         {
             lock (_lock)
             {
-                return _state;
+                return field;
             }
         }
-    }
+        set
+        {
+            if (value)
+            {
+                throw new ArgumentException("CanHeartbeat can only be set to false.", nameof(value));
+            }
+
+            lock (_lock)
+            {
+                field = false;
+            }
+        }
+    } = true;
 
-    public Task SetAsCannotHeartbeatAsync(CancellationToken cancellationToken = default)
+    public required DateTime LastHeartbeatTime
     {
-        cancellationToken.ThrowIfCancellationRequested();
-        lock (_lock)
+        get
         {
-            _canHeartbeat = false;
+            lock (_lock)
+            {
+                return field;
+            }
+        }
+        set
+        {
+            lock (_lock)
+            {
+                field = value;
+            }
         }
-
-        return Task.CompletedTask;
     }
 
-    public Task SetLastHeartbeatTimeAsync(DateTime lastHeartbeatTime,
-        CancellationToken cancellationToken = default)
+    public required JobState? State
     {
-        cancellationToken.ThrowIfCancellationRequested();
-        lock (_lock)
+        get
         {
-            _lastHeartbeatTime = lastHeartbeatTime;
+            lock (_lock)
+            {
+                return field;
+            }
         }
+        set
+        {
+            if (value is not { } newState)
+            {
+                throw new ArgumentNullException(nameof(value));
+            }
 
-        return Task.CompletedTask;
-    }
+            Action<IJobRepositoryEntry, JobState?, JobState>? callbacks;
+            JobState? original;
+            lock (_lock)
+            {
+                if (field == newState)
+                {
+                    return;
+                }
+
+                original = field;
+                field = newState;
+                callbacks = _stateCallbacks;
+            }
+
+            callbacks?.Invoke(this, original, newState);
+        }
+    } = JobState.Inactive;
 
-    public Task SetStateAsync(JobState state, CancellationToken cancellationToken = default)
+    /// <summary>
+    ///     Register a callback invoked with this entry plus the original and current <see cref="State" />.
+    ///     Invoked immediately on subscribe when the current state is not <c>null</c>;
+    ///     the original state is <c>null</c> for that first invocation.
+    /// </summary>
+    /// <param name="action">
+    ///     Receives this entry, the original state (possibly <c>null</c>), and the current non-null state.
+    /// </param>
+    /// <exception cref="ObjectDisposedException">Thrown when this entry has already been disposed.</exception>
+    public void SubscribeToState(Action<IJobRepositoryEntry, JobState?, JobState> action)
     {
-        cancellationToken.ThrowIfCancellationRequested();
+        ArgumentNullException.ThrowIfNull(action);
+
+        JobState? current;
         lock (_lock)
         {
-            _state = state;
+            if (_disposed)
+            {
+                throw new ObjectDisposedException(nameof(JobRepositoryEntry));
+            }
+
+            _stateCallbacks += action;
+            current = State;
         }
 
-        return Task.CompletedTask;
+        if (current is { } state)
+        {
+            action(this, null, state);
+        }
     }
 }