Merge pull request #40 from Blazor-Data-Orchestrator/feature/agent-single-job-execution-guard

Prevent concurrent job execution on single agent and ensure memory cl…

Author: imsaadawais

src/BlazorDataOrchestrator.Core/Services/CodeExecutorService.cs

@@ -775,6 +775,24 @@ import json
 
         return null;
     }
+
+    /// <summary>
+    /// Cleans up state between job executions to prevent memory leaks and state bleed.
+    /// Resets the CSScript evaluator and forces unloading of collectible AssemblyLoadContexts.
+    /// Must be called after every job execution, regardless of success or failure.
+    /// </summary>
+    public void CleanupAfterExecution()
+    {
+        try
+        {
+            // Reset the CSScript evaluator to clear all cached references and compiled assemblies
+            CSScript.Evaluator.Reset();
+        }
+        catch
+        {
+            // Evaluator may not have been initialized — safe to ignore
+        }
+    }
 }
 
 /// <summary>

src/BlazorOrchestrator.Agent/Worker.cs

@@ -26,6 +26,10 @@ public class Worker : BackgroundService
     // Visibility timeout configuration for long-running jobs
     private static readonly TimeSpan VisibilityTimeout = TimeSpan.FromMinutes(5);
     private static readonly TimeSpan RenewalInterval = TimeSpan.FromMinutes(3);
+    
+    // Ensures only one job runs at a time per agent instance.
+    // Prevents picking up a new job while the current one is still executing.
+    private static readonly SemaphoreSlim _jobExecutionLock = new(1, 1);
 
     public Worker(
         ILogger<Worker> logger, 
@@ -96,6 +100,10 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                 using var renewalCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
                 Task<string>? renewalTask = null;
 
+                // Acquire the execution lock — blocks if a previous job is somehow still running
+                await _jobExecutionLock.WaitAsync(stoppingToken);
+                _logger.LogInformation("Execution lock acquired for message {MessageId}", message.MessageId);
+
                 try
                 {
                     // Parse the queue message, handling both Base64-encoded and plain text formats
@@ -183,6 +191,18 @@ await _jobManager.ProcessJobInstanceAsync(
                     // Message will become visible again after visibility timeout
                     // After too many failures, it will go to poison queue (if configured)
                 }
+                finally
+                {
+                    // Release the execution lock so the next job can be picked up
+                    _jobExecutionLock.Release();
+                    _logger.LogInformation("Execution lock released for message {MessageId}", message.MessageId);
+
+                    // Force cleanup between jobs to prevent memory leaks and state bleed
+                    _codeExecutor.CleanupAfterExecution();
+                    GC.Collect(2, GCCollectionMode.Aggressive, blocking: true);
+                    GC.WaitForPendingFinalizers();
+                    _logger.LogInformation("Post-job memory cleanup completed");
+                }
             }
             catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
             {