fix: better support for scoped services.

added more tests as well

Author: VisualBean

src/ByteBard.GUSTO.JobQueue/JobQueueWorker.cs

@@ -66,11 +66,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
                 await Parallel.ForEachAsync(jobStorageRecords, parallelOptions, async (storedJob, ct) =>
                 {
+                    using var scope = _serviceProvider.CreateScope();
                     try
                     {
                         var jobType = Type.GetType(storedJob.JobType);
                         var arguments = JsonConvert.DeserializeObject<object[]>(storedJob.ArgumentsJson, _settings);
-                        var jobInstance = ActivatorUtilities.CreateInstance(_serviceProvider, jobType);
+                        var jobInstance = ActivatorUtilities.CreateInstance(scope.ServiceProvider, jobType);
                         var method = jobType.GetMethod(storedJob.MethodName);
                         await (Task)method.Invoke(jobInstance, arguments);
 

src/ByteBard.GUSTO.Tests/JobQueueTests.cs

@@ -31,7 +31,7 @@ public virtual Task DoSomethingAsync(string input)
     }
 
     [Fact]
-    public async Task EnqueueAsync_ValidInterfaceJobType_StoresSerializedJobWithRealType()
+    public async Task EnqueueAsync_WhenValidInterfaceJobTypeProvided_StoresSerializedJobWithRealType()
     {
         // Arrange
         var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
@@ -54,7 +54,7 @@ await storageProvider.Received(1).StoreJobAsync(
     }
 
     [Fact]
-    public async Task EnqueueAsync_ValidMethodCall_StoresSerializedJob()
+    public async Task EnqueueAsync_WhenValidMethodCallProvided_StoresSerializedJob()
     {
         // Arrange
         var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
@@ -78,7 +78,7 @@ await storageProvider.Received(1).StoreJobAsync(
 
 
     [Fact]
-    public async Task EnqueueAsyncGeneric_ValidMethodCall_StoresSerializedJob()
+    public async Task EnqueueAsyncGeneric_WhenValidMethodCallProvided_StoresSerializedJob()
     {
         // Arrange
         var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
@@ -100,7 +100,7 @@ await storageProvider.Received(1).StoreJobAsync(
     }
 
     [Fact]
-    public async Task EnqueueAsync_WithExecuteAfter_UsesProvidedTime()
+    public async Task EnqueueAsync_WhenExecuteAfterTimeProvided_UsesProvidedTime()
     {
         // Arrange
         var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
@@ -119,7 +119,7 @@ await storageProvider.Received(1).StoreJobAsync(
     }
 
     [Fact]
-    public async Task EnqueueAsync_WithoutExecuteAfter_SetsToUtcNow()
+    public async Task EnqueueAsync_WhenExecuteAfterNotProvided_SetsToUtcNow()
     {
         // Arrange
         var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
@@ -137,4 +137,99 @@ await storageProvider.Received(1).StoreJobAsync(
                 record.ExecuteAfter >= beforeCall && record.ExecuteAfter <= afterCall),
             cancellationToken);
     }
+
+    public class JobWithMultipleParameters
+    {
+        public Task ProcessAsync(string message, int count, bool isEnabled)
+        {
+            return Task.CompletedTask;
+        }
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_WhenMultipleParametersProvided_SerializesAllArguments()
+    {
+        // Arrange
+        var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
+        var jobQueue = new JobQueue<TestJobStorageRecord>(storageProvider);
+        var cancellationToken = CancellationToken.None;
+
+        // Act
+        var trackingId = await jobQueue.EnqueueAsync<JobWithMultipleParameters>(
+            job => job.ProcessAsync("hello", 42, true),
+            null,
+            cancellationToken);
+
+        // Assert
+        await storageProvider.Received(1).StoreJobAsync(
+            Arg.Is<TestJobStorageRecord>(record =>
+                record.TrackingId == trackingId &&
+                record.MethodName == "ProcessAsync" &&
+                record.JobType == typeof(JobWithMultipleParameters).AssemblyQualifiedName &&
+                JsonConvert.DeserializeObject<object[]>(record.ArgumentsJson).Length == 3 &&
+                JsonConvert.DeserializeObject<object[]>(record.ArgumentsJson)[0].ToString() == "hello" &&
+                JsonConvert.DeserializeObject<object[]>(record.ArgumentsJson)[1].ToString() == "42" &&
+                JsonConvert.DeserializeObject<object[]>(record.ArgumentsJson)[2].ToString() == "True"),
+            cancellationToken);
+    }
+
+    [Fact]
+    public async Task ConstructRecordFromExpression_WhenCalled_ReturnsValidRecord()
+    {
+        // Arrange
+        var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
+        var jobQueue = new JobQueue<TestJobStorageRecord>(storageProvider);
+        var executeAfter = DateTime.UtcNow.AddMinutes(5);
+
+        // Act
+        var record = jobQueue.ConstructRecordFromExpression<TestJob>(
+            job => job.DoSomethingAsync("test"),
+            executeAfter);
+
+        // Assert
+        Assert.NotEqual(Guid.Empty, record.TrackingId);
+        Assert.Equal("DoSomethingAsync", record.MethodName);
+        Assert.Equal(typeof(TestJob).AssemblyQualifiedName, record.JobType);
+        Assert.False(record.IsComplete);
+        Assert.Equal(executeAfter, record.ExecuteAfter);
+        Assert.True(record.CreatedOn <= DateTime.UtcNow);
+
+        var args = JsonConvert.DeserializeObject<object[]>(record.ArgumentsJson);
+        Assert.Single(args);
+        Assert.Equal("test", args[0]);
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_WhenStorageProviderThrowsException_PropagatesException()
+    {
+        // Arrange
+        var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
+        storageProvider.StoreJobAsync(Arg.Any<TestJobStorageRecord>(), Arg.Any<CancellationToken>())
+            .Returns<Task>(_ => throw new InvalidOperationException("Storage error"));
+
+        var jobQueue = new JobQueue<TestJobStorageRecord>(storageProvider);
+        var cancellationToken = CancellationToken.None;
+
+        // Act & Assert
+        await Assert.ThrowsAsync<InvalidOperationException>(async () =>
+            await jobQueue.EnqueueAsync<TestJob>(job => job.DoSomethingAsync("test"), null, cancellationToken));
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_WhenCancellationRequested_PropagatesCancellationToStorage()
+    {
+        // Arrange
+        var storageProvider = Substitute.For<IJobStorageProvider<TestJobStorageRecord>>();
+        storageProvider.StoreJobAsync(Arg.Any<TestJobStorageRecord>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromCanceled(new CancellationToken(true)));
+
+        var jobQueue = new JobQueue<TestJobStorageRecord>(storageProvider);
+
+        using var cts = new CancellationTokenSource();
+        cts.Cancel();
+
+        // Act & Assert
+        await Assert.ThrowsAsync<TaskCanceledException>(async () =>
+            await jobQueue.EnqueueAsync<TestJob>(job => job.DoSomethingAsync("test"), null, cts.Token));
+    }
 }

src/ByteBard.GUSTO.Tests/JobQueueWorkerTests.cs

@@ -61,7 +61,7 @@ public virtual Task ExecuteAsyncFail(string input)
     };
 
     [Fact]
-    public async Task ExecuteAsync_NoJobs_DelaysAndContinuesLoop()
+    public async Task ExecuteAsync_WhenNoJobsAvailable_DelaysAndContinuesLoop()
     {
         // Arrange
         var storage = Substitute.For<IJobStorageProvider<TestJob>>();
@@ -85,7 +85,7 @@ public async Task ExecuteAsync_NoJobs_DelaysAndContinuesLoop()
     }
 
     [Fact]
-    public async Task ExecuteAsync_ValidJob_InvokesJobAndMarksComplete()
+    public async Task ExecuteAsync_WhenValidJobExists_InvokesJobAndMarksComplete()
     {
         // Arrange
         var job = new TestJob
@@ -120,7 +120,7 @@ public async Task ExecuteAsync_ValidJob_InvokesJobAndMarksComplete()
     }
 
     [Fact]
-    public async Task ExecuteAsync_JobThrowsException_LogsAndHandlesFailure()
+    public async Task ExecuteAsync_WhenJobThrowsException_LogsAndHandlesFailure()
     {
         // Arrange
         var job = new TestJob
@@ -154,7 +154,7 @@ public async Task ExecuteAsync_JobThrowsException_LogsAndHandlesFailure()
     }
 
     [Fact]
-    public async Task ExecuteAsync_StorageThrowsException_LogsAndDelays()
+    public async Task ExecuteAsync_WhenStorageThrowsException_LogsAndDelays()
     {
         // Arrange
         var storage = Substitute.For<IJobStorageProvider<TestJob>>();
@@ -178,7 +178,7 @@ public async Task ExecuteAsync_StorageThrowsException_LogsAndDelays()
     }
 
     [Fact]
-    public async Task ExecuteAsync_EnqueuedViaJobQueue_JobIsExecutedAndMarkedComplete()
+    public async Task ExecuteAsync_WhenEnqueuedViaJobQueue_JobIsExecutedAndMarkedComplete()
     {
         // Arrange
         var storage = Substitute.For<IJobStorageProvider<TestJob>>();
@@ -214,4 +214,191 @@ public async Task ExecuteAsync_EnqueuedViaJobQueue_JobIsExecutedAndMarkedComplet
         await storage.Received().MarkJobAsCompleteAsync(capturedJob, Arg.Any<CancellationToken>());
         Assert.Equal("from real queue", JsonConvert.DeserializeObject<string[]>(capturedJob.ArgumentsJson)[0]);
     }
+
+    public interface IScopedTestService
+    {
+        string ProcessData(string input);
+        bool IsDisposed { get; }
+    }
+
+    public class ScopedTestService : IScopedTestService, IDisposable
+    {
+        public bool IsDisposed { get; private set; }
+
+        public string ProcessData(string input)
+        {
+            if (IsDisposed) throw new ObjectDisposedException(nameof(ScopedTestService));
+            return $"Processed: {input}";
+        }
+
+        public void Dispose()
+        {
+            IsDisposed = true;
+        }
+    }
+
+    public class JobWithScopedDependency
+    {
+        private readonly IScopedTestService _scopedService;
+        private readonly ITestResultCollector _resultCollector;
+
+        public JobWithScopedDependency(IScopedTestService scopedService, ITestResultCollector resultCollector)
+        {
+            _scopedService = scopedService;
+            _resultCollector = resultCollector;
+        }
+
+        public Task ExecuteWithScopedService(string input)
+        {
+            try
+            {
+                var result = _scopedService.ProcessData(input);
+                _resultCollector.AddResult(result);
+                return Task.CompletedTask;
+            }
+            finally
+            {
+                _resultCollector.SignalCompletion();
+            }
+        }
+    }
+
+    public interface ITestResultCollector
+    {
+        void AddResult(string result);
+        void SignalCompletion();
+        Task WaitForCompletionAsync(int expectedCount, TimeSpan timeout);
+        List<string> GetResults();
+    }
+
+    public class TestResultCollector : ITestResultCollector
+    {
+        private readonly List<string> _results = new();
+        private readonly SemaphoreSlim _semaphore = new(0, 10);
+
+        public void AddResult(string result)
+        {
+            lock (_results)
+            {
+                _results.Add(result);
+            }
+        }
+
+        public void SignalCompletion()
+        {
+            _semaphore.Release();
+        }
+
+        public async Task WaitForCompletionAsync(int expectedCount, TimeSpan timeout)
+        {
+            for (int i = 0; i < expectedCount; i++)
+            {
+                await _semaphore.WaitAsync(timeout);
+            }
+        }
+
+        public List<string> GetResults()
+        {
+            lock (_results)
+            {
+                return new List<string>(_results);
+            }
+        }
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_WhenJobRequiresScopedService_CreatesAndUsesServiceFromScope()
+    {
+        // Arrange
+        var resultCollector = new TestResultCollector();
+
+        var job = new TestJob
+        {
+            TrackingId = Guid.NewGuid(),
+            JobType = typeof(JobWithScopedDependency).AssemblyQualifiedName,
+            MethodName = nameof(JobWithScopedDependency.ExecuteWithScopedService),
+            ArgumentsJson = JsonConvert.SerializeObject(new object[] { "test-data" }),
+            ExecuteAfter = DateTime.UtcNow,
+            IsComplete = false
+        };
+
+        var storage = Substitute.For<IJobStorageProvider<TestJob>>();
+        storage.GetBatchAsync(Arg.Any<JobSearchParams<TestJob>>(), Arg.Any<CancellationToken>())
+            .Returns(new List<TestJob> { job }, new List<TestJob>());
+
+        var services = new ServiceCollection();
+        services.AddScoped<IScopedTestService, ScopedTestService>();
+        services.AddSingleton<ITestResultCollector>(resultCollector);
+        var serviceProvider = services.BuildServiceProvider();
+
+        var logger = Substitute.For<ILogger<JobQueueWorker<TestJob>>>();
+        var config = Options.Create(GetTestConfig());
+
+        var worker = new JobQueueWorker<TestJob>(serviceProvider, storage, config, logger);
+
+        // Act
+        var running = worker.StartAsync(CancellationToken.None);
+        await resultCollector.WaitForCompletionAsync(1, TimeSpan.FromSeconds(5));
+        await worker.StopAsync(CancellationToken.None);
+
+        // Assert
+        await storage.Received().MarkJobAsCompleteAsync(job, Arg.Any<CancellationToken>());
+        var results = resultCollector.GetResults();
+        Assert.Single(results);
+        Assert.Equal("Processed: test-data", results[0]);
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_WhenMultipleJobsWithScopedServices_EachJobGetsOwnServiceInstance()
+    {
+        // Arrange
+        var resultCollector = new TestResultCollector();
+
+        var job1 = new TestJob
+        {
+            TrackingId = Guid.NewGuid(),
+            JobType = typeof(JobWithScopedDependency).AssemblyQualifiedName,
+            MethodName = nameof(JobWithScopedDependency.ExecuteWithScopedService),
+            ArgumentsJson = JsonConvert.SerializeObject(new object[] { "job1" }),
+            ExecuteAfter = DateTime.UtcNow,
+            IsComplete = false
+        };
+
+        var job2 = new TestJob
+        {
+            TrackingId = Guid.NewGuid(),
+            JobType = typeof(JobWithScopedDependency).AssemblyQualifiedName,
+            MethodName = nameof(JobWithScopedDependency.ExecuteWithScopedService),
+            ArgumentsJson = JsonConvert.SerializeObject(new object[] { "job2" }),
+            ExecuteAfter = DateTime.UtcNow,
+            IsComplete = false
+        };
+
+        var storage = Substitute.For<IJobStorageProvider<TestJob>>();
+        storage.GetBatchAsync(Arg.Any<JobSearchParams<TestJob>>(), Arg.Any<CancellationToken>())
+            .Returns(new List<TestJob> { job1, job2 }, new List<TestJob>());
+
+        var services = new ServiceCollection();
+        services.AddScoped<IScopedTestService, ScopedTestService>();
+        services.AddSingleton<ITestResultCollector>(resultCollector);
+        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 });
+
+        var worker = new JobQueueWorker<TestJob>(serviceProvider, storage, config, logger);
+
+        // Act
+        var running = worker.StartAsync(CancellationToken.None);
+        await resultCollector.WaitForCompletionAsync(2, TimeSpan.FromSeconds(5));
+        await worker.StopAsync(CancellationToken.None);
+
+        // Assert
+        await storage.Received().MarkJobAsCompleteAsync(job1, Arg.Any<CancellationToken>());
+        await storage.Received().MarkJobAsCompleteAsync(job2, Arg.Any<CancellationToken>());
+        var results = resultCollector.GetResults();
+        Assert.Equal(2, results.Count);
+        Assert.Contains("Processed: job1", results);
+        Assert.Contains("Processed: job2", results);
+    }
 }