Enhance crontab job execution with re-entry protection

Author: aden.chen

src/Infrastructure/BotSharp.Abstraction/Crontab/ICrontabService.cs

@@ -4,4 +4,5 @@ public interface ICrontabService
 {
     Task<List<CrontabItem>> GetCrontable();
     Task ScheduledTimeArrived(CrontabItem item);
+    Task ExecuteTimeArrivedItemWithReentryProtection(CrontabItem item);
 }

src/Infrastructure/BotSharp.Abstraction/Crontab/Models/CrontabItem.cs

@@ -32,6 +32,9 @@ public class CrontabItem : ScheduleTaskArgs
     [JsonPropertyName("trigger_type")]
     public CronTabItemTriggerType TriggerType { get; set; } = CronTabItemTriggerType.BackgroundWatcher;
 
+    [JsonPropertyName("reentry_protection")]
+    public bool ReentryProtection { get; set; } = true;
+
     public override string ToString()
     {
         return $"{Title}: {Description} [AgentId: {AgentId}, UserId: {UserId}]";

src/Infrastructure/BotSharp.Core.Crontab/Services/CrontabService.cs

@@ -15,6 +15,7 @@ limitations under the License.
 ******************************************************************************/
 
 using BotSharp.Abstraction.Agents.Models;
+using BotSharp.Abstraction.Infrastructures;
 using BotSharp.Abstraction.Repositories;
 using BotSharp.Abstraction.Repositories.Filters;
 using BotSharp.Abstraction.Tasks;
@@ -127,4 +128,62 @@ await HookEmitter.Emit<ICrontabHook>(_services, async hook =>
             }
         }, item.AgentId);
     }
+
+    public async Task ExecuteTimeArrivedItemWithReentryProtection(CrontabItem item)
+    {
+        if (!item.ReentryProtection)
+        {
+            await ExecuteTimeArrivedItem(item);
+            return;
+        }
+
+        var lockKey = $"crontab:execution:{item.Title}";
+        using var scope = _services.CreateScope();
+        var locker = scope.ServiceProvider.GetRequiredService<IDistributedLocker>();
+        var acquired = false;
+        var lockAcquired = false;
+
+        try
+        {
+            acquired = await locker.LockAsync(lockKey, async () =>
+            {
+                lockAcquired = true;
+                _logger.LogInformation("Crontab: {0}, Distributed lock acquired, beginning execution...", item.Title);
+                await ExecuteTimeArrivedItem(item);
+            }, timeout: 600);
+
+            if (!acquired)
+            {
+                _logger.LogWarning("Crontab: {0}, Failed to acquire distributed lock, task is still executing, skipping this occurrence to prevent re-entry.", item.Title);
+            }
+        }
+        catch (Exception ex)
+        {
+            if (!lockAcquired)
+            {
+                _logger.LogWarning("Crontab: {0}, Redis exception occurred before acquiring lock: {1}, executing without lock protection (re-entry protection disabled).", item.Title, ex.Message);
+                await ExecuteTimeArrivedItem(item);
+            }
+            else
+            {
+                _logger.LogWarning("Crontab: {0}, Redis exception occurred after lock acquired: {1}, task execution completed but lock release failed.", item.Title, ex.Message);
+            }
+        }
+    } 
+    
+    private async Task<bool> ExecuteTimeArrivedItem(CrontabItem item)
+    {
+        try
+        {
+            _logger.LogInformation($"Start running crontab {item.Title}");
+            await ScheduledTimeArrived(item);
+            _logger.LogInformation($"Complete running crontab {item.Title}");
+            return true;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, $"Error when running crontab {item.Title}");
+            return false;
+        }
+    }
 }

src/Infrastructure/BotSharp.OpenAPI/Controllers/Crontab/CrontabController.cs

@@ -60,7 +60,7 @@ public async Task<CrontabSchedulingResult> SchedulingCrontab()
         {
             if (item.CheckNextOccurrenceEveryOneMinute())
             {
-                _logger.LogInformation("Crontab: {0}, One occurrence was matched, Beginning execution...", item.Title);
+                _logger.LogInformation($"Crontab: {item.Title}, One occurrence was matched, attempting to execute...");
                 Task.Run(() => ExecuteTimeArrivedItem(item, _services));
                 result.OccurrenceMatchedItems.Add(item.Title);
             }
@@ -84,21 +84,10 @@ private async Task<List<CrontabItem>> GetCrontabItems(string? title = null)
         return allowedCrons.Where(cron => cron.Title.IsEqualTo(title)).ToList();
     }
 
-    private async Task<bool> ExecuteTimeArrivedItem(CrontabItem item, IServiceProvider services)
+    private async Task ExecuteTimeArrivedItem(CrontabItem item, IServiceProvider services)
     {
-        try
-        {
-            using var scope = services.CreateScope();
-            var crontabService = scope.ServiceProvider.GetRequiredService<ICrontabService>();
-            _logger.LogInformation($"Start running crontab {item.Title}");
-            await crontabService.ScheduledTimeArrived(item);
-            _logger.LogInformation($"Complete running crontab {item.Title}");
-            return true;
-        }
-        catch (Exception ex)
-        {
-            _logger.LogError(ex, $"Error when running crontab {item.Title}");
-            return false;
-        }
+        using var scope = services.CreateScope();
+        var crontabService = scope.ServiceProvider.GetRequiredService<ICrontabService>();
+        await crontabService.ExecuteTimeArrivedItemWithReentryProtection(item);
     }
 }