fix: Update configuration and implement job execution timeout handling

Author: VisualBean

README.md

@@ -111,7 +111,8 @@ services.AddGusto<JobRecord, InMemoryJobStorageProvider>(
   "Gusto": {
     "BatchSize": 50,
     "Concurrency": 8,
-    "PollInterval": "00:00:10"
+    "PollInterval": "00:00:10",
+    "JobExecutionTimeout": "00:05:00"
   }
 }
 ```
@@ -167,6 +168,7 @@ public class UserController : ControllerBase
 - **BatchSize**: Jobs to process per batch (default: 10)
 - **Concurrency**: Max parallel jobs (default: Environment.ProcessorCount)
 - **PollInterval**: Polling frequency (default: 10 seconds)
+- **JobExecutionTimeout**: Max runtime for a single job execution before it's treated as a failure and passed to `OnHandlerExecutionFailureAsync` (default: 5 minutes)
 
 ## Advanced Patterns
 

src/ByteBard.GUSTO.JobQueue/GustoConfig.cs

@@ -9,4 +9,6 @@ public class GustoConfig
     public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(10);
 
     public int Concurrency { get; set; } = Environment.ProcessorCount;
+
+    public TimeSpan JobExecutionTimeout { get; set; } = TimeSpan.FromMinutes(5);
 }

src/ByteBard.GUSTO.JobQueue/JobQueueWorker.cs

@@ -101,6 +101,10 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                 await ProcessBatchCycleAsync(parallelOptions, stoppingToken);
                 SignalBatchCompletedIfSet();
             }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
             catch (Exception ex)
             {
                 _logger.LogError(ex, "Unexpected failure in JobQueueWorker.");
@@ -195,18 +199,33 @@ private async Task ExecuteJobAsync(
 
         using var scope = _serviceProvider.CreateScope();
         var storage = scope.ServiceProvider.GetRequiredService<IJobStorageProvider<TStorageRecord>>();
+        using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        timeoutCts.CancelAfter(_config.JobExecutionTimeout);
 
         try
         {
             var jobType = Type.GetType(storedJob.JobType);
             var arguments = JsonConvert.DeserializeObject<object[]>(storedJob.ArgumentsJson, _settings);
             var jobInstance = ActivatorUtilities.CreateInstance(scope.ServiceProvider, jobType);
             var method = jobType.GetMethod(storedJob.MethodName);
-            await (Task)method.Invoke(jobInstance, arguments);
+            var handlerTask = (Task)method.Invoke(jobInstance, arguments);
+
+            await handlerTask.WaitAsync(timeoutCts.Token);
 
             await storage.MarkJobAsCompleteAsync(storedJob, ct);
             RecordJobSuccess(storedJob, jobStopwatch, jobActivity);
         }
+        catch (OperationCanceledException ex) when (timeoutCts.IsCancellationRequested && !ct.IsCancellationRequested)
+        {
+            var timeoutException = new TimeoutException(
+                $"Job execution timed out after {_config.JobExecutionTimeout}.",
+                ex);
+            await RecordJobFailureAsync(storedJob, storage, timeoutException, jobStopwatch, jobActivity, ct);
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            throw;
+        }
         catch (Exception ex)
         {
             await RecordJobFailureAsync(storedJob, storage, ex, jobStopwatch, jobActivity, ct);

src/ByteBard.GUSTO.Tests/JobQueueWorkerTests.cs

@@ -27,7 +27,7 @@ public class TestableJob
 
         public static void CreateSemaphore()
         {
-            semaphoreSlim = new SemaphoreSlim(0, 1);
+            semaphoreSlim = new SemaphoreSlim(0, int.MaxValue);
         }
 
         public virtual Task ExecuteAsyncSuccess(string input)
@@ -53,15 +53,49 @@ public virtual Task ExecuteAsyncFail(string input)
                 semaphoreSlim.Release();
             }
         }
+
+        public virtual async Task ExecuteAsyncHang(string input)
+        {
+            semaphoreSlim.Release();
+            await Task.Delay(Timeout.InfiniteTimeSpan);
+        }
     }
 
     private GustoConfig GetTestConfig() => new GustoConfig
     {
         Concurrency = 1,
         PollInterval = TimeSpan.FromMilliseconds(10),
-        BatchSize = 1
+        BatchSize = 1,
+        JobExecutionTimeout = TimeSpan.FromSeconds(30)
     };
 
+    private static async Task AssertEventuallyAsync(Action assertion, TimeSpan timeout)
+    {
+        var started = DateTime.UtcNow;
+        Exception? lastException = null;
+
+        while (DateTime.UtcNow - started < timeout)
+        {
+            try
+            {
+                assertion();
+                return;
+            }
+            catch (Exception ex)
+            {
+                lastException = ex;
+                await Task.Delay(25);
+            }
+        }
+
+        if (lastException != null)
+        {
+            throw lastException;
+        }
+
+        assertion();
+    }
+
     [Fact]
     public async Task ExecuteAsync_WhenNoJobsAvailable_DelaysAndContinuesLoop()
     {
@@ -162,6 +196,145 @@ public async Task ExecuteAsync_WhenJobThrowsException_LogsAndHandlesFailure()
         await storage.Received().OnHandlerExecutionFailureAsync(job, Arg.Any<Exception>(), Arg.Any<CancellationToken>());
     }
 
+    [Fact]
+    public async Task ExecuteAsync_WhenOneJobHangs_OtherJobsCompleteAndHangingJobTimesOut()
+    {
+        // Arrange
+        var hangingJob = new TestJob
+        {
+            TrackingId = Guid.NewGuid(),
+            JobType = typeof(TestableJob).AssemblyQualifiedName,
+            MethodName = nameof(TestableJob.ExecuteAsyncHang),
+            ArgumentsJson = JsonConvert.SerializeObject(new object[] { "hang" }),
+            ExecuteAfter = DateTime.UtcNow,
+            IsComplete = false
+        };
+
+        var healthyJob = new TestJob
+        {
+            TrackingId = Guid.NewGuid(),
+            JobType = typeof(TestableJob).AssemblyQualifiedName,
+            MethodName = nameof(TestableJob.ExecuteAsyncSuccess),
+            ArgumentsJson = JsonConvert.SerializeObject(new object[] { "ok" }),
+            ExecuteAfter = DateTime.UtcNow,
+            IsComplete = false
+        };
+
+        var storage = Substitute.For<IJobStorageProvider<TestJob>>();
+        storage.GetBatchAsync(Arg.Any<JobSearchParams<TestJob>>(), Arg.Any<CancellationToken>())
+            .Returns(new List<TestJob> { hangingJob, healthyJob }, new List<TestJob>());
+
+        var services = new ServiceCollection();
+        services.AddScoped<IJobStorageProvider<TestJob>>(_ => storage);
+        var serviceProvider = services.BuildServiceProvider();
+
+        var logger = Substitute.For<ILogger<JobQueueWorker<TestJob>>>();
+        var config = Options.Create(new GustoConfig
+        {
+            Concurrency = 2,
+            PollInterval = TimeSpan.FromMilliseconds(10),
+            BatchSize = 2,
+            JobExecutionTimeout = TimeSpan.FromMilliseconds(100)
+        });
+
+        var worker = new JobQueueWorker<TestJob>(serviceProvider, config, logger);
+
+        // Act
+        TestableJob.CreateSemaphore();
+        await worker.StartAsync(CancellationToken.None);
+
+        try
+        {
+            await AssertEventuallyAsync(
+                () =>
+                {
+                    storage.Received().MarkJobAsCompleteAsync(healthyJob, Arg.Any<CancellationToken>());
+                },
+                TimeSpan.FromSeconds(2));
+
+            await AssertEventuallyAsync(
+                () =>
+                {
+                    storage.Received().OnHandlerExecutionFailureAsync(
+                        hangingJob,
+                        Arg.Is<Exception>(ex => ex is TimeoutException),
+                        Arg.Any<CancellationToken>());
+                },
+                TimeSpan.FromSeconds(2));
+        }
+        finally
+        {
+            await worker.StopAsync(CancellationToken.None);
+        }
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_WhenJobTimesOut_WorkerContinuesPollingNextCycles()
+    {
+        // Arrange
+        var hangingJob = new TestJob
+        {
+            TrackingId = Guid.NewGuid(),
+            JobType = typeof(TestableJob).AssemblyQualifiedName,
+            MethodName = nameof(TestableJob.ExecuteAsyncHang),
+            ArgumentsJson = JsonConvert.SerializeObject(new object[] { "hang" }),
+            ExecuteAfter = DateTime.UtcNow,
+            IsComplete = false
+        };
+
+        var getBatchCallCount = 0;
+        var storage = Substitute.For<IJobStorageProvider<TestJob>>();
+        storage.GetBatchAsync(Arg.Any<JobSearchParams<TestJob>>(), Arg.Any<CancellationToken>())
+            .Returns(_ =>
+            {
+                var callCount = Interlocked.Increment(ref getBatchCallCount);
+                IEnumerable<TestJob> jobs = callCount == 1
+                    ? new List<TestJob> { hangingJob }
+                    : new List<TestJob>();
+                return Task.FromResult(jobs);
+            });
+
+        var services = new ServiceCollection();
+        services.AddScoped<IJobStorageProvider<TestJob>>(_ => storage);
+        var serviceProvider = services.BuildServiceProvider();
+
+        var logger = Substitute.For<ILogger<JobQueueWorker<TestJob>>>();
+        var config = Options.Create(new GustoConfig
+        {
+            Concurrency = 1,
+            PollInterval = TimeSpan.FromMilliseconds(10),
+            BatchSize = 1,
+            JobExecutionTimeout = TimeSpan.FromMilliseconds(100)
+        });
+
+        var worker = new JobQueueWorker<TestJob>(serviceProvider, config, logger);
+
+        // Act
+        TestableJob.CreateSemaphore();
+        await worker.StartAsync(CancellationToken.None);
+
+        try
+        {
+            await AssertEventuallyAsync(
+                () => Assert.True(Volatile.Read(ref getBatchCallCount) >= 2),
+                TimeSpan.FromSeconds(2));
+
+            await AssertEventuallyAsync(
+                () =>
+                {
+                    storage.Received().OnHandlerExecutionFailureAsync(
+                        hangingJob,
+                        Arg.Is<Exception>(ex => ex is TimeoutException),
+                        Arg.Any<CancellationToken>());
+                },
+                TimeSpan.FromSeconds(2));
+        }
+        finally
+        {
+            await worker.StopAsync(CancellationToken.None);
+        }
+    }
+
     [Fact]
     public async Task ExecuteAsync_WhenStorageThrowsException_LogsAndDelays()
     {