[WIP] Fix race condition in PayloadAssembler OnTimedEvent method

src/InformaticsGateway/Services/Connectors/PayloadAssembler.cs

@@ -152,9 +152,20 @@ private async void OnTimedEvent(Object? source, System.Timers.ElapsedEventArgs e
                 {
                     _logger.BucketsActive(_payloads.Count);
                 }
-                foreach (var key in _payloads.Keys)
+
+                // Take a snapshot of the keys to avoid InvalidOperationException
+                // when the collection is modified during iteration by concurrent Queue() calls.
+                var keysSnapshot = _payloads.Keys.ToList();
+
+                foreach (var key in keysSnapshot)
                 {
-                    var payload = await _payloads[key].WithCancellation(_tokenSource.Token).ConfigureAwait(false);
+                    // Key may have been removed by another thread between snapshot and access
+                    if (!_payloads.TryGetValue(key, out var lazyPayload))
+                    {
+                        continue;
+                    }
+
+                    var payload = await lazyPayload.WithCancellation(_tokenSource.Token).ConfigureAwait(false);
                     using var loggerScope = _logger.BeginScope(new LoggingDataDictionary<string, object> { { "CorrelationId", payload.CorrelationId } });
 
                     _logger.BucketElapsedTime(key, payload.Timeout, payload.ElapsedTime().TotalSeconds, payload.Files.Count, payload.FilesUploaded, payload.FilesFailedToUpload);

src/InformaticsGateway/Test/Services/Connectors/PayloadAssemblerTest.cs

@@ -15,6 +15,9 @@
  */
 
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
@@ -157,5 +160,125 @@ public async Task GivenAPayloadThatHasCompletedUploads_WhenProcessedByTimedEvent
             Assert.Single(result.Files);
             _logger.VerifyLoggingMessageBeginsWith($"Bucket A sent to processing queue with {result.Count} files", LogLevel.Information, Times.AtLeastOnce());
         }
+
+        [RetryFact(5, 500)]
+        public async Task GivenConcurrentQueueOperations_WhenTimerProcessesBuckets_ExpectNoInvalidOperationException()
+        {
+            // This test verifies the race condition fix: concurrent Queue() calls
+            // should not cause InvalidOperationException in OnTimedEvent when it
+            // iterates over the _payloads dictionary.
+            var payloadAssembler = new PayloadAssembler(_logger.Object, _serviceScopeFactory.Object);
+            var exceptions = new ConcurrentBag<Exception>();
+            var tasks = new List<Task>();
+
+            // Simulate many concurrent Queue operations across different buckets
+            // while the timer is actively processing (fires every 1 second).
+            for (int i = 0; i < 50; i++)
+            {
+                var bucketName = $"concurrent-bucket-{i}";
+                var file = new TestStorageInfo(
+                    Guid.NewGuid().ToString(),
+                    Guid.NewGuid().ToString(),
+                    $"file-{i}",
+                    ".dcm",
+                    new DataOrigin
+                    {
+                        DataService = Messaging.Events.DataService.DIMSE,
+                        Destination = "dest",
+                        Source = "source"
+                    });
+
+                tasks.Add(Task.Run(async () =>
+                {
+                    try
+                    {
+                        await payloadAssembler.Queue(
+                            bucketName,
+                            file,
+                            new DataOrigin
+                            {
+                                DataService = Messaging.Events.DataService.DIMSE,
+                                Destination = "dest",
+                                Source = "source"
+                            },
+                            1);
+                    }
+                    catch (Exception ex)
+                    {
+                        exceptions.Add(ex);
+                    }
+                }));
+            }
+
+            await Task.WhenAll(tasks);
+
+            // Allow the timer to fire and process buckets while new ones may still be settling
+            await Task.Delay(3000);
+
+            payloadAssembler.Dispose();
+
+            // The critical assertion: no InvalidOperationException should have occurred
+            // in either the Queue tasks or the timer's OnTimedEvent processing.
+            var raceConditionErrors = exceptions.Where(ex => ex is InvalidOperationException).ToList();
+            Assert.Empty(raceConditionErrors);
+        }
+
+        [RetryFact(5, 500)]
+        public async Task GivenConcurrentQueueAndDequeue_WhenUnderLoad_ExpectStableOperation()
+        {
+            // Stress test: rapidly add files to overlapping buckets with short timeouts,
+            // ensuring concurrent modification of the dictionary does not crash the timer.
+            var payloadAssembler = new PayloadAssembler(_logger.Object, _serviceScopeFactory.Object);
+            var exceptions = new ConcurrentBag<Exception>();
+            var tasks = new List<Task>();
+
+            for (int i = 0; i < 100; i++)
+            {
+                // Use only 5 distinct bucket names so multiple tasks hit the same bucket concurrently
+                var bucketName = $"stress-bucket-{i % 5}";
+                var file = new TestStorageInfo(
+                    Guid.NewGuid().ToString(),
+                    Guid.NewGuid().ToString(),
+                    $"stress-file-{i}",
+                    ".dcm",
+                    new DataOrigin
+                    {
+                        DataService = Messaging.Events.DataService.DIMSE,
+                        Destination = "dest",
+                        Source = "source"
+                    });
+
+                tasks.Add(Task.Run(async () =>
+                {
+                    try
+                    {
+                        await payloadAssembler.Queue(
+                            bucketName,
+                            file,
+                            new DataOrigin
+                            {
+                                DataService = Messaging.Events.DataService.DIMSE,
+                                Destination = "dest",
+                                Source = "source"
+                            },
+                            1);
+                    }
+                    catch (Exception ex)
+                    {
+                        exceptions.Add(ex);
+                    }
+                }));
+            }
+
+            await Task.WhenAll(tasks);
+
+            // Let timer fire several cycles
+            await Task.Delay(4000);
+
+            payloadAssembler.Dispose();
+
+            var raceConditionErrors = exceptions.Where(ex => ex is InvalidOperationException).ToList();
+            Assert.Empty(raceConditionErrors);
+        }
     }
 }