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

src/Exceptionless.Core/Jobs/CleanupDataJob.cs

@@ -93,7 +93,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)
@@ -210,8 +210,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
                 {
@@ -238,8 +243,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
                 {
@@ -426,8 +436,13 @@ private async Task EnforceRetentionAsync(JobContext context, bool canCleanupSour
         var results = await _organizationRepository.FindAsync(q => q.Include(o => o.Id, o => o.Name, o => o.PlanId, 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;
@@ -521,6 +536,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;
@@ -39,22 +38,19 @@ protected override async Task<JobResult> RunInternalAsync(JobContext context)
         _logger.LogTrace("Start save stack event counts");
 
         // Get list of stacks where snooze has expired
-        var results = await _stackRepository.GetExpiredSnoozedStatuses(_timeProvider.GetUtcNow().UtcDateTime, o => o.PageLimit(LIMIT));
+        var results = await _stackRepository.GetExpiredSnoozedStatuses(
+            _timeProvider.GetUtcNow().UtcDateTime,
+            o => o.SearchAfterPaging().PageLimit(LIMIT));
         while (results.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
-            foreach (var stack in results.Documents)
-                stack.MarkOpen();
-
-            await _stackRepository.SaveAsync(results.Documents);
+            await context.RenewLockAsync();
+            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);
 
             if (context.CancellationToken.IsCancellationRequested || !await results.NextPageAsync())
                 break;
-
-            if (results.Documents.Count > 0)
-                await context.RenewLockAsync();
         }
 
         _logger.LogTrace("Finished save stack event counts");

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,41 @@ 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>();
+
+    /// <summary>
+    /// Marks a redirect tombstone whose counters changed after its events may already have been
+    /// reassigned. With no source events left, the orphan scan cannot rediscover that late update;
+    /// this durable marker lets cleanup replay the idempotent contribution ledger instead.
+    /// </summary>
+    [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))

src/Exceptionless.Core/Repositories/EventRepository.cs

@@ -1,21 +1,29 @@
-using Elastic.Clients.Elasticsearch.QueryDsl;
+using System.Linq.Expressions;
+using Elastic.Clients.Elasticsearch;
+using Elastic.Clients.Elasticsearch.Aggregations;
+using Elastic.Clients.Elasticsearch.QueryDsl;
+using Elastic.Transport.Products.Elasticsearch;
 using Exceptionless.Core.Models;
 using Exceptionless.Core.Repositories.Configuration;
 using Exceptionless.Core.Repositories.Queries;
 using Exceptionless.Core.Validation;
 using Exceptionless.DateTimeExtensions;
 using Foundatio.Repositories;
+using Foundatio.Repositories.Elasticsearch.Extensions;
+using Foundatio.Repositories.Exceptions;
 using Foundatio.Repositories.Models;
 
 namespace Exceptionless.Core.Repositories;
 
 public class EventRepository : RepositoryOwnedByOrganizationAndProject<PersistentEvent>, IEventRepository
 {
+    private readonly ExceptionlessElasticConfiguration _configuration;
     private readonly TimeProvider _timeProvider;
 
     public EventRepository(ExceptionlessElasticConfiguration configuration, AppOptions options, MiniValidationValidator validator)
         : base(configuration.Events, validator, options)
     {
+        _configuration = configuration;
         _timeProvider = configuration.TimeProvider;
 
         DisableCache(); // NOTE: If cache is ever enabled, then fast paths for patching/deleting with scripts will be super slow!
@@ -191,9 +199,180 @@ public override Task<FindResults<PersistentEvent>> GetByProjectIdAsync(string pr
     public Task<long> RemoveAllByStackIdsAsync(string[] stackIds)
     {
         ArgumentNullException.ThrowIfNull(stackIds);
-        if (stackIds.Length == 0)
+        if (stackIds is [])
             throw new ArgumentOutOfRangeException(nameof(stackIds));
 
         return RemoveAllAsync(q => q.Stack(stackIds));
     }
+
+    private static bool IsIndexNotFound(ElasticsearchServerError serverError)
+    {
+        if (serverError.Status != 404 || serverError.Error is null)
+            return false;
+
+        return String.Equals(serverError.Error.Type, "index_not_found_exception", StringComparison.Ordinal)
+            || serverError.Error.RootCause?.Any(IsIndexNotFound) == true;
+    }
+
+    private static bool IsIndexNotFound(Elastic.Transport.Products.Elasticsearch.ErrorCause? cause)
+    {
+        return cause is not null
+            && (String.Equals(cause.Type, "index_not_found_exception", StringComparison.Ordinal)
+                || cause.CausedBy is not null && IsIndexNotFound(cause.CausedBy));
+    }
+
+    /// <summary>
+    /// Reassigns all events from the source stacks to the target stack using a parameterized
+    /// Painless script (no string interpolation) to prevent script injection.
+    /// </summary>
+    /// <remarks>
+    /// Foundatio's update-by-query task cannot be interrupted safely once submitted. The token is
+    /// observed before submission and by both strict verification reads; the caller must retain its
+    /// lease until this method returns so cancellation cannot leave an unobserved background write.
+    /// </remarks>
+    public async Task<long> ReassignStackAsync(IEnumerable<string> sourceStackIds, string targetStackId, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(sourceStackIds);
+        ArgumentException.ThrowIfNullOrEmpty(targetStackId);
+
+        // Materialize to avoid multiple enumeration and guard against empty; an empty
+        // .Stack() filter would match ALL events and reassign them to the target stack.
+        var sourceIds = sourceStackIds.Distinct(StringComparer.Ordinal).ToList();
+        if (sourceIds.Count is 0)
+            return 0;
+        if (sourceIds.Contains(targetStackId, StringComparer.Ordinal))
+            throw new ArgumentException("Source and target stack ids must be different.", nameof(sourceStackIds));
+
+        cancellationToken.ThrowIfCancellationRequested();
+
+        const int maxAttempts = 5;
+        long remaining = await CountEventsByStackIdsStrictAsync(sourceIds, cancellationToken);
+        long affected = 0;
+
+        for (int attempt = 1; remaining > 0 && attempt <= maxAttempts; attempt++)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            affected += await PatchAllAsync(
+                q => q.Stack(sourceIds),
+                new ScriptPatch("ctx._source.stack_id = params.targetStackId")
+                {
+                    Params = new Dictionary<string, object> { ["targetStackId"] = targetStackId }
+                },
+                o => o.ImmediateConsistency().Notifications(false));
+
+            // ScriptPatch uses update-by-query with Conflicts.Proceed. Verify strictly against
+            // the same alias and retry any documents skipped by version conflicts.
+            remaining = await CountEventsByStackIdsStrictAsync(sourceIds, cancellationToken);
+        }
+
+        if (remaining > 0)
+            throw new DocumentException($"Unable to reassign {remaining} event(s) after {maxAttempts} attempts.");
+
+        return affected;
+    }
+
+    private async Task<long> CountEventsByStackIdsStrictAsync(IReadOnlyCollection<string> stackIds, CancellationToken cancellationToken)
+    {
+        var count = await _configuration.Client.CountAsync<PersistentEvent>(s => s
+            .Indices(_configuration.Events.Name)
+            .AllowNoIndices(true)
+            .IgnoreUnavailable(false)
+            .ExpandWildcards(ExpandWildcard.All)
+            .Query(q => q.Terms(t => t
+                .Field(e => e.StackId)
+                .Terms(new TermsQueryField(stackIds.Select(id => (FieldValue)id).ToList())))), cancellationToken);
+
+        // A brand-new installation can have no event alias yet. Fall back to every concrete
+        // version only for that specific state so an alias transition can never look like zero.
+        if (!count.IsValidResponse && count.ElasticsearchServerError is not null && IsIndexNotFound(count.ElasticsearchServerError))
+        {
+            count = await _configuration.Client.CountAsync<PersistentEvent>(s => s
+                .Indices($"{_configuration.Events.Name}-v*-*")
+                .AllowNoIndices(true)
+                .IgnoreUnavailable(false)
+                .ExpandWildcards(ExpandWildcard.All)
+                .Query(q => q.Terms(t => t
+                    .Field(e => e.StackId)
+                    .Terms(new TermsQueryField(stackIds.Select(id => (FieldValue)id).ToList())))), cancellationToken);
+        }
+
+        if (!count.IsValidResponse || count.Shards.Failed > 0)
+            throw new DocumentException($"Unable to verify event reassignment through the event alias: {count.DebugInformation}", count.ApiCallDetails.OriginalException);
+
+        return count.Count;
+    }
+
+    public Task<DistinctValuePage> GetDistinctStackIdsAsync(int batchSize, string? afterValue = null, CancellationToken cancellationToken = default)
+    {
+        return GetDistinctFieldValuesAsync("stack_id", e => e.StackId, batchSize, afterValue, cancellationToken);
+    }
+
+    public Task<DistinctValuePage> GetDistinctProjectIdsAsync(int batchSize, string? afterValue = null, CancellationToken cancellationToken = default)
+    {
+        return GetDistinctFieldValuesAsync("project_id", e => e.ProjectId, batchSize, afterValue, cancellationToken);
+    }
+
+    public Task<DistinctValuePage> GetDistinctOrganizationIdsAsync(int batchSize, string? afterValue = null, CancellationToken cancellationToken = default)
+    {
+        return GetDistinctFieldValuesAsync("organization_id", e => e.OrganizationId, batchSize, afterValue, cancellationToken);
+    }
+
+    /// <summary>
+    /// Uses a composite aggregation to paginate through all distinct values of a field.
+    /// Composite aggregations are preferred over terms aggregations for high-cardinality fields
+    /// because terms aggregations can silently miss values when the unique count exceeds the
+    /// configured size parameter. Composite aggregations guarantee correct iteration via an
+    /// after_key cursor, at the cost of requiring sequential page fetches.
+    /// </summary>
+    private async Task<DistinctValuePage> GetDistinctFieldValuesAsync(
+        string fieldName,
+        Expression<Func<PersistentEvent, object>> fieldExpression,
+        int batchSize,
+        string? afterValue,
+        CancellationToken cancellationToken)
+    {
+        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(batchSize);
+
+        string aggregationName = $"composite_{fieldName}";
+        var sources = new List<KeyValuePair<string, CompositeAggregationSource>>
+        {
+            new(fieldName, new CompositeAggregationSource
+            {
+                Terms = new CompositeTermsAggregation { Field = fieldExpression }
+            })
+        };
+
+        var search = await _configuration.Client.SearchAsync<PersistentEvent>(s =>
+        {
+            s.Indices($"{_configuration.Events.VersionedName}-*")
+                .Size(0)
+                .AddAggregation(aggregationName, a => a.Composite(c =>
+                {
+                    c.Size(batchSize)
+                        .Sources(sources);
+
+                    if (!String.IsNullOrEmpty(afterValue))
+                        c.After(new Dictionary<Field, FieldValue> { [fieldName] = afterValue });
+                }));
+        }, cancellationToken);
+
+        if (!search.IsValidResponse)
+        {
+            if (search.ElasticsearchServerError is not null && IsIndexNotFound(search.ElasticsearchServerError))
+                return new DistinctValuePage([], null);
+
+            throw new InvalidOperationException($"Error retrieving distinct event values for '{fieldName}': {search.DebugInformation}", search.ApiCallDetails.OriginalException);
+        }
+
+        var composite = search.Aggregations?.GetComposite(aggregationName);
+        var values = composite?.Buckets is { Count: > 0 }
+            ? composite.Buckets.Select(bucket => bucket.Key[fieldName].ToString()!).ToArray()
+            : [];
+        string? nextValue = composite?.AfterKey is not null && composite.AfterKey.TryGetValue(fieldName, out var next)
+            ? next.ToString()
+            : null;
+
+        return new DistinctValuePage(values, nextValue);
+    }
 }