fix: Optimize cleanup jobs, eliminate raw ES client usage, add test coverage

src/Exceptionless.Core/Jobs/CleanupDataJob.cs

@@ -86,7 +86,7 @@ ILoggerFactory loggerFactory
 
     protected override Task<ILock?> GetLockAsync(CancellationToken cancellationToken = default)
     {
-        return _lockProvider.TryAcquireAsync(nameof(CleanupDataJob), TimeSpan.FromMinutes(15), cancellationToken);
+        return _lockProvider.TryAcquireAsync(nameof(CleanupDataJob), TimeSpan.FromHours(2), cancellationToken);
     }
 
     protected override async Task<JobResult> RunInternalAsync(JobContext context)
@@ -201,8 +201,13 @@ private async Task CleanupSoftDeletedOrganizationsAsync(JobContext context)
 
         while (organizationResults.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
+            await RenewLockAsync(context);
+
             foreach (var organization in organizationResults.Documents)
             {
+                if (context.CancellationToken.IsCancellationRequested)
+                    break;
+
                 using var _ = _logger.BeginScope(new ExceptionlessState().Organization(organization.Id));
                 try
                 {
@@ -229,8 +234,13 @@ private async Task CleanupSoftDeletedProjectsAsync(JobContext context)
 
         while (projectResults.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
+            await RenewLockAsync(context);
+
             foreach (var project in projectResults.Documents)
             {
+                if (context.CancellationToken.IsCancellationRequested)
+                    break;
+
                 using var _ = _logger.BeginScope(new ExceptionlessState().Organization(project.OrganizationId).Project(project.Id));
                 try
                 {
@@ -383,8 +393,13 @@ private async Task EnforceRetentionAsync(JobContext context)
         var results = await _organizationRepository.FindAsync(q => q.Include(o => o.Id, o => o.Name, o => o.RetentionDays), o => o.SearchAfterPaging().PageLimit(100));
         while (results.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
+            await RenewLockAsync(context);
+
             foreach (var organization in results.Documents)
             {
+                if (context.CancellationToken.IsCancellationRequested)
+                    break;
+
                 using var _ = _logger.BeginScope(new ExceptionlessState().Organization(organization.Id));
 
                 int retentionDays = _billingManager.GetBillingPlanByUpsellingRetentionPeriod(organization.RetentionDays)?.RetentionDays ?? _appOptions.MaximumRetentionDays;
@@ -454,6 +469,8 @@ private async Task EnforceEventRetentionDaysAsync(Organization organization, int
 
     private Task RenewLockAsync(JobContext context)
     {
+        // Called at each page boundary to prevent the distributed lock from expiring
+        // during long-running bulk cleanup operations that span multiple pages.
         _lastRun = _timeProvider.GetUtcNow().UtcDateTime;
         return context.RenewLockAsync();
     }

src/Exceptionless.Core/Jobs/StackStatusJob.cs

@@ -1,5 +1,4 @@
-using Exceptionless.Core.Extensions;
-using Exceptionless.Core.Repositories;
+using Exceptionless.Core.Repositories;
 using Foundatio.Caching;
 using Foundatio.Jobs;
 using Foundatio.Lock;
@@ -42,10 +41,7 @@ protected override async Task<JobResult> RunInternalAsync(JobContext context)
         var results = await _stackRepository.GetExpiredSnoozedStatuses(_timeProvider.GetUtcNow().UtcDateTime, o => o.PageLimit(LIMIT));
         while (results.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
-            foreach (var stack in results.Documents)
-                stack.MarkOpen();
-
-            await _stackRepository.SaveAsync(results.Documents);
+            await _stackRepository.MarkOpenAsync(results.Documents.Select(stack => stack.Id));
 
             // Sleep so we are not hammering the backend.
             await Task.Delay(TimeSpan.FromSeconds(2.5), _timeProvider);

src/Exceptionless.Core/Models/Stack.cs

@@ -4,12 +4,13 @@
 using System.Runtime.Serialization;
 using System.Text.Json.Serialization;
 using Exceptionless.Core.Attributes;
+using Exceptionless.Core.Serialization;
 using Foundatio.Repositories.Models;
 
 namespace Exceptionless.Core.Models;
 
 [DebuggerDisplay("Id={Id} Type={Type} Status={Status} IsDeleted={IsDeleted} Title={Title} TotalOccurrences={TotalOccurrences}")]
-public class Stack : IOwnedByOrganizationAndProjectWithIdentity, IHaveDates, ISupportSoftDeletes, IValidatableObject
+public class Stack : IOwnedByOrganizationAndProjectWithIdentity, IHaveDates, ISupportSoftDeletes, IVersioned, IValidatableObject
 {
     /// <summary>
     /// Unique id that identifies a stack.
@@ -120,6 +121,36 @@ public class Stack : IOwnedByOrganizationAndProjectWithIdentity, IHaveDates, ISu
     public DateTime UpdatedUtc { get; set; }
     public bool IsDeleted { get; set; }
 
+    /// <summary>
+    /// The canonical stack for events that still reference this duplicate stack.
+    /// This is internal cleanup state and is not part of the public API contract.
+    /// </summary>
+    [JsonInclude]
+    [JsonIgnoreForExternalSerialization]
+    internal string? RedirectToStackId { get; set; }
+
+    /// <summary>
+    /// Tracks how many occurrences from each duplicate stack have already been merged.
+    /// This is internal cleanup state and is not part of the public API contract.
+    /// </summary>
+    [JsonInclude]
+    [JsonIgnoreForExternalSerialization]
+    internal IDictionary<string, int> MergedDuplicateStackTotals { get; set; } = new Dictionary<string, int>();
+
+    [JsonInclude]
+    [JsonIgnoreForExternalSerialization]
+    internal bool NeedsRedirectReconciliation { get; set; }
+
+    [JsonInclude]
+    [JsonIgnoreForExternalSerialization]
+    internal string ElasticVersion { get; set; } = null!;
+
+    string IVersioned.Version
+    {
+        get => ElasticVersion;
+        set => ElasticVersion = value;
+    }
+
     public bool AllowNotifications => Status != StackStatus.Fixed && Status != StackStatus.Ignored && Status != StackStatus.Discarded && Status != StackStatus.Snoozed;
 
     public static class KnownTypes

src/Exceptionless.Core/Pipeline/010_AssignToStackAction.cs

@@ -111,7 +111,7 @@ public override async Task ProcessBatchAsync(ICollection<EventContext> contexts)
             }
             else
             {
-                ctx.Stack = await _stackRepository.GetByIdAsync(ctx.Event.StackId, o => o.Cache());
+                ctx.Stack = await _stackRepository.GetCanonicalStackAsync(ctx.Event.StackId);
                 if (ctx.Stack is null || ctx.Stack.ProjectId != ctx.Event.ProjectId)
                 {
                     ctx.SetError("Invalid StackId.");
@@ -168,8 +168,8 @@ await _publisher.PublishAsync(new EntityChanged
         }
 
         var stacksToSave = stacks.Where(s => s.Value.ShouldSave).Select(kvp => kvp.Value.Stack).ToList();
-        if (stacksToSave.Count > 0)
-            await _stackRepository.SaveAsync(stacksToSave, o => o.Cache().Notifications(false)); // notification will get sent later in the update stats step
+        foreach (var stack in stacksToSave)
+            await _stackRepository.AddEventTagsAsync(stack.Id, stack.Tags); // notification will get sent later in the update stats step
 
         // Set stack ids after they have been saved and created
         contexts.ForEach(ctx =>

src/Exceptionless.Core/Repositories/Configuration/Indexes/StackIndex.cs

@@ -47,6 +47,8 @@ public override void ConfigureIndexMapping(TypeMappingDescriptor<Stack> map)
                 .Keyword(e => e.SignatureHash, k => k.IgnoreAbove(1024))
                     .FieldAlias(Alias.SignatureHash, a => a.Path(f => f.SignatureHash))
                 .Keyword(e => e.DuplicateSignature)
+                .Keyword(e => e.RedirectToStackId)
+                .Boolean(e => e.NeedsRedirectReconciliation)
                 .Keyword(e => e.Type, k => k.IgnoreAbove(1024))
                 .Date(e => e.FirstOccurrence)
                     .FieldAlias(Alias.FirstOccurrence, a => a.Path(f => f.FirstOccurrence))