Scope orphaned-event cleanup to recent ingest window to eliminate 8-hour ES CPU spikes

src/Exceptionless.Core/Jobs/CleanupOrphanedDataJob.cs

@@ -23,6 +23,7 @@ namespace Exceptionless.Core.Jobs;
 [Job(Description = "Deletes orphaned data.", IsContinuous = false)]
 public class CleanupOrphanedDataJob : JobWithLockBase, IHealthCheck
 {
+    internal static readonly TimeSpan OrphanedEventLookback = TimeSpan.FromDays(3);
     private readonly ExceptionlessElasticConfiguration _config;
     private readonly ElasticsearchClient _elasticClient;
     private readonly IStackRepository _stackRepository;
@@ -58,21 +59,28 @@ ILoggerFactory loggerFactory
 
     protected override async Task<JobResult> RunInternalAsync(JobContext context)
     {
-        await DeleteOrphanedEventsByStackAsync(context);
-        await DeleteOrphanedEventsByProjectAsync(context);
-        await DeleteOrphanedEventsByOrganizationAsync(context);
+        var orphanedEventCutoffUtc = GetOrphanedEventCutoffUtc();
+        await DeleteOrphanedEventsByStackAsync(context, orphanedEventCutoffUtc);
+        await DeleteOrphanedEventsByProjectAsync(context, orphanedEventCutoffUtc);
+        await DeleteOrphanedEventsByOrganizationAsync(context, orphanedEventCutoffUtc);
 
         await FixDuplicateStacks(context);
 
         return JobResult.Success;
     }
 
-    public async Task DeleteOrphanedEventsByStackAsync(JobContext context)
+    public Task DeleteOrphanedEventsByStackAsync(JobContext context)
+    {
+        return DeleteOrphanedEventsByStackAsync(context, GetOrphanedEventCutoffUtc());
+    }
+
+    private async Task DeleteOrphanedEventsByStackAsync(JobContext context, DateTime orphanedEventCutoffUtc)
     {
         // get approximate number of unique stack ids
         var stackCardinality = await _elasticClient.SearchAsync<PersistentEvent>(s => s
-            .Indices(GetEventIndexPattern())
+            .Indices(GetRecentEventIndex())
             .Size(0)
+            .Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
             .AddAggregation("cardinality_stack_id", a => a.Cardinality(c => c.Field(f => f.StackId).PrecisionThreshold(40000))));
 
         double? uniqueStackIdCount = stackCardinality.Aggregations?.GetCardinality("cardinality_stack_id")?.Value;
@@ -91,8 +99,9 @@ public async Task DeleteOrphanedEventsByStackAsync(JobContext context)
             await RenewLockAsync(context);
 
             var stackIdTerms = await _elasticClient.SearchAsync<PersistentEvent>(s => s
-                .Indices(GetEventIndexPattern())
+                .Indices(GetRecentEventIndex())
                 .Size(0)
+                .Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
                 .AddAggregation("terms_stack_id", a => a.Terms(c => c.Field(f => f.StackId).Include(new TermsInclude(batchNumber, buckets)).Size(batchSize * 2))));
 
             string[] stackIds = stackIdTerms.Aggregations?.GetStringTerms("terms_stack_id")?.Buckets.Select(b => b.Key.ToString()!).ToArray() ?? [];
@@ -114,19 +123,27 @@ public async Task DeleteOrphanedEventsByStackAsync(JobContext context)
             totalOrphanedEventCount += missingStackIds.Length;
             _logger.LogInformation("{BatchNumber}/{BatchCount}: Found {OrphanedEventCount} orphaned events from missing stacks {MissingStackIds} out of {StackIdCount}", batchNumber, buckets, missingStackIds.Length, missingStackIds, stackIds.Length);
             await _elasticClient.DeleteByQueryAsync<PersistentEvent>(r => r
-                .Indices(GetEventIndexPattern())
-                .Query(q => q.Terms(t => t.Field(f => f.StackId).Terms(new TermsQueryField(missingStackIds.Select(FieldValueHelper.ToFieldValue).ToList())))));
+                .Indices(GetRecentEventIndex())
+                .Query(q => q.Bool(b => b.Filter(
+                    f => f.Terms(t => t.Field(e => e.StackId).Terms(new TermsQueryField(missingStackIds.Select(FieldValueHelper.ToFieldValue).ToList()))),
+                    f => RecentEventQuery(f, orphanedEventCutoffUtc)))));
         }
 
-        _logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing stacks out of {StackIdCount}", totalOrphanedEventCount, totalStackIds);
+        _logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing stacks out of {StackIdCount} since {OrphanedEventCutoffUtc}", totalOrphanedEventCount, totalStackIds, orphanedEventCutoffUtc);
     }
 
-    public async Task DeleteOrphanedEventsByProjectAsync(JobContext context)
+    public Task DeleteOrphanedEventsByProjectAsync(JobContext context)
+    {
+        return DeleteOrphanedEventsByProjectAsync(context, GetOrphanedEventCutoffUtc());
+    }
+
+    private async Task DeleteOrphanedEventsByProjectAsync(JobContext context, DateTime orphanedEventCutoffUtc)
     {
         // get approximate number of unique project ids
         var projectCardinality = await _elasticClient.SearchAsync<PersistentEvent>(s => s
-            .Indices(GetEventIndexPattern())
+            .Indices(GetRecentEventIndex())
             .Size(0)
+            .Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
             .AddAggregation("cardinality_project_id", a => a.Cardinality(c => c.Field(f => f.ProjectId).PrecisionThreshold(40000))));
 
         double? uniqueProjectIdCount = projectCardinality.Aggregations?.GetCardinality("cardinality_project_id")?.Value;
@@ -145,8 +162,9 @@ public async Task DeleteOrphanedEventsByProjectAsync(JobContext context)
             await RenewLockAsync(context);
 
             var projectIdTerms = await _elasticClient.SearchAsync<PersistentEvent>(s => s
-                .Indices(GetEventIndexPattern())
+                .Indices(GetRecentEventIndex())
                 .Size(0)
+                .Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
                 .AddAggregation("terms_project_id", a => a.Terms(c => c.Field(f => f.ProjectId).Include(new TermsInclude(batchNumber, buckets)).Size(batchSize * 2))));
 
             string[] projectIds = projectIdTerms.Aggregations?.GetStringTerms("terms_project_id")?.Buckets.Select(b => b.Key.ToString()!).ToArray() ?? [];
@@ -168,19 +186,27 @@ public async Task DeleteOrphanedEventsByProjectAsync(JobContext context)
             totalOrphanedEventCount += missingProjectIds.Length;
             _logger.LogInformation("{BatchNumber}/{BatchCount}: Found {OrphanedEventCount} orphaned events from missing projects {MissingProjectIds} out of {ProjectIdCount}", batchNumber, buckets, missingProjectIds.Length, missingProjectIds, projectIds.Length);
             await _elasticClient.DeleteByQueryAsync<PersistentEvent>(r => r
-                .Indices(GetEventIndexPattern())
-                .Query(q => q.Terms(t => t.Field(f => f.ProjectId).Terms(new TermsQueryField(missingProjectIds.Select(FieldValueHelper.ToFieldValue).ToList())))));
+                .Indices(GetRecentEventIndex())
+                .Query(q => q.Bool(b => b.Filter(
+                    f => f.Terms(t => t.Field(e => e.ProjectId).Terms(new TermsQueryField(missingProjectIds.Select(FieldValueHelper.ToFieldValue).ToList()))),
+                    f => RecentEventQuery(f, orphanedEventCutoffUtc)))));
         }
 
-        _logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing projects out of {ProjectIdCount}", totalOrphanedEventCount, totalProjectIds);
+        _logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing projects out of {ProjectIdCount} since {OrphanedEventCutoffUtc}", totalOrphanedEventCount, totalProjectIds, orphanedEventCutoffUtc);
+    }
+
+    public Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
+    {
+        return DeleteOrphanedEventsByOrganizationAsync(context, GetOrphanedEventCutoffUtc());
     }
 
-    public async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
+    private async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context, DateTime orphanedEventCutoffUtc)
     {
         // get approximate number of unique organization ids
         var organizationCardinality = await _elasticClient.SearchAsync<PersistentEvent>(s => s
-            .Indices(GetEventIndexPattern())
+            .Indices(GetRecentEventIndex())
             .Size(0)
+            .Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
             .AddAggregation("cardinality_organization_id", a => a.Cardinality(c => c.Field(f => f.OrganizationId).PrecisionThreshold(40000))));
 
         double? uniqueOrganizationIdCount = organizationCardinality.Aggregations?.GetCardinality("cardinality_organization_id")?.Value;
@@ -199,8 +225,9 @@ public async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
             await RenewLockAsync(context);
 
             var organizationIdTerms = await _elasticClient.SearchAsync<PersistentEvent>(s => s
-                .Indices(GetEventIndexPattern())
+                .Indices(GetRecentEventIndex())
                 .Size(0)
+                .Query(q => RecentEventQuery(q, orphanedEventCutoffUtc))
                 .AddAggregation("terms_organization_id", a => a.Terms(c => c.Field(f => f.OrganizationId).Include(new TermsInclude(batchNumber, buckets)).Size(batchSize * 2))));
 
             string[] organizationIds = organizationIdTerms.Aggregations?.GetStringTerms("terms_organization_id")?.Buckets.Select(b => b.Key.ToString()!).ToArray() ?? [];
@@ -222,11 +249,13 @@ public async Task DeleteOrphanedEventsByOrganizationAsync(JobContext context)
             totalOrphanedEventCount += missingOrganizationIds.Length;
             _logger.LogInformation("{BatchNumber}/{BatchCount}: Found {OrphanedEventCount} orphaned events from missing organizations {MissingOrganizationIds} out of {OrganizationIdCount}", batchNumber, buckets, missingOrganizationIds.Length, missingOrganizationIds, organizationIds.Length);
             await _elasticClient.DeleteByQueryAsync<PersistentEvent>(r => r
-                .Indices(GetEventIndexPattern())
-                .Query(q => q.Terms(t => t.Field(f => f.OrganizationId).Terms(new TermsQueryField(missingOrganizationIds.Select(FieldValueHelper.ToFieldValue).ToList())))));
+                .Indices(GetRecentEventIndex())
+                .Query(q => q.Bool(b => b.Filter(
+                    f => f.Terms(t => t.Field(e => e.OrganizationId).Terms(new TermsQueryField(missingOrganizationIds.Select(FieldValueHelper.ToFieldValue).ToList()))),
+                    f => RecentEventQuery(f, orphanedEventCutoffUtc)))));
         }
 
-        _logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing organizations out of {OrganizationIdCount}", totalOrphanedEventCount, totalOrganizationIds);
+        _logger.LogInformation("Found {OrphanedEventCount} orphaned events from missing organizations out of {OrganizationIdCount} since {OrphanedEventCutoffUtc}", totalOrphanedEventCount, totalOrganizationIds, orphanedEventCutoffUtc);
     }
 
     public async Task FixDuplicateStacks(JobContext context)
@@ -401,6 +430,21 @@ private Task RenewLockAsync(JobContext context)
         return context.RenewLockAsync();
     }
 
+    private DateTime GetOrphanedEventCutoffUtc()
+    {
+        return _timeProvider.GetUtcNow().UtcDateTime.Subtract(OrphanedEventLookback);
+    }
+
+    private static QueryDescriptor<PersistentEvent> RecentEventQuery(QueryDescriptor<PersistentEvent> query, DateTime orphanedEventCutoffUtc)
+    {
+        return query.Range(r => r.Date(d => d.Field(e => e.CreatedUtc).Gte(orphanedEventCutoffUtc)));
+    }
+
+    private string GetRecentEventIndex()
+    {
+        return $"{_config.Events.Name}-last3days";
+    }
+
     private string GetEventIndexPattern()
     {
         return $"{_config.Events.VersionedName}-*";

tests/Exceptionless.Tests/Jobs/CleanupOrphanedDataJobTests.cs

@@ -67,7 +67,7 @@
         Assert.Equal(200, totalCount);
     }
 
     [Fact]
     public async Task DeleteOrphanedEventsByStack_WithMixedOrphanedAndValid_OnlyDeletesOrphaned()
     {
         // Arrange - Tenant 1 has valid events; Tenant 2 has orphaned events (stack doesn't exist)
@@ -106,7 +106,7 @@
         Assert.Equal(100, totalAfter);
     }
 
     [Fact]
     public async Task DeleteOrphanedEventsByStack_LargeVolume_PreservesAllValidEvents()
     {
         // Arrange - Large volume across two tenants: 5000 valid + 10000 orphaned
@@ -173,7 +173,7 @@
         Assert.Equal(200, totalAfter);
     }
 
     [Fact]
     public async Task DeleteOrphanedEventsByStack_OnlyOrphanedEventsInOneTenant_OtherTenantUnaffected()
     {
         // Arrange - Tenant 1 has all orphaned events (will be deleted); Tenant 2 has all valid events
@@ -233,7 +233,7 @@
         Assert.Equal(100, totalAfter);
     }
 
     [Fact]
     public async Task DeleteOrphanedEventsByProject_WithOrphanedProject_DeletesEventsForMissingProject()
     {
         // Arrange - Events reference a project that doesn't exist
@@ -265,7 +265,7 @@
         Assert.Equal(75, totalAfter);
     }
 
     [Fact]
     public async Task DeleteOrphanedEventsByProject_MultiTenant_EachTenantIndependent()
     {
         // Arrange - Tenant 1 has valid project, Tenant 2 has orphaned project
@@ -326,7 +326,7 @@
         Assert.Equal(160, totalAfter);
     }
 
     [Fact]
     public async Task DeleteOrphanedEventsByOrganization_WithOrphanedOrganization_DeletesEventsForMissingOrganization()
     {
         // Arrange - Valid organization1 + events referencing non-existent organization
@@ -358,7 +358,7 @@
         Assert.Equal(100, totalAfter);
     }
 
     [Fact]
     public async Task DeleteOrphanedEventsByOrganization_TwoTenantsOneDeleted_OnlyDeletesOrphanedTenantEvents()
     {
         // Arrange - Organization 1 exists, Organization 2 does NOT exist (never created) but has events
@@ -390,6 +390,66 @@
         Assert.Equal(120, totalAfter);
     }
 
+    [Fact]
+    public async Task RunAsync_OrphanedEventsAtAndBeforeLookbackBoundary_DeletesOnlyEventsWithinLookback()
+    {
+        TimeProvider.SetUtcNow(DateTimeOffset.UtcNow);
+
+        var organization = await _organizationRepository.AddAsync(
+            _organizationData.GenerateSampleOrganization(_billingManager, _plans),
+            o => o.ImmediateConsistency());
+        var project = await _projectRepository.AddAsync(
+            _projectData.GenerateSampleProject(),
+            o => o.ImmediateConsistency());
+        var stack = await _stackRepository.AddAsync(
+            _stackData.GenerateSampleStack(),
+            o => o.ImmediateConsistency());
+
+        var cutoffUtc = TimeProvider.GetUtcNow().UtcDateTime.Subtract(CleanupOrphanedDataJob.OrphanedEventLookback);
+        var beforeCutoffUtc = cutoffUtc.AddMilliseconds(-1);
+        var validEvent = _eventData.GenerateEvent(organization.Id, project.Id, stack.Id);
+
+        string missingStackId = ObjectId.GenerateNewId().ToString();
+        var stackOrphanAtCutoff = _eventData.GenerateEvent(organization.Id, project.Id, missingStackId);
+        stackOrphanAtCutoff.CreatedUtc = cutoffUtc;
+        var stackOrphanBeforeCutoff = _eventData.GenerateEvent(organization.Id, project.Id, missingStackId);
+        stackOrphanBeforeCutoff.CreatedUtc = beforeCutoffUtc;
+
+        string missingProjectId = ObjectId.GenerateNewId().ToString();
+        var projectOrphanAtCutoff = _eventData.GenerateEvent(organization.Id, missingProjectId, stack.Id);
+        projectOrphanAtCutoff.CreatedUtc = cutoffUtc;
+        var projectOrphanBeforeCutoff = _eventData.GenerateEvent(organization.Id, missingProjectId, stack.Id);
+        projectOrphanBeforeCutoff.CreatedUtc = beforeCutoffUtc;
+
+        string missingOrganizationId = ObjectId.GenerateNewId().ToString();
+        var organizationOrphanAtCutoff = _eventData.GenerateEvent(missingOrganizationId, project.Id, stack.Id);
+        organizationOrphanAtCutoff.CreatedUtc = cutoffUtc;
+        var organizationOrphanBeforeCutoff = _eventData.GenerateEvent(missingOrganizationId, project.Id, stack.Id);
+        organizationOrphanBeforeCutoff.CreatedUtc = beforeCutoffUtc;
+
+        await _eventRepository.AddAsync([
+            validEvent,
+            stackOrphanAtCutoff,
+            stackOrphanBeforeCutoff,
+            projectOrphanAtCutoff,
+            projectOrphanBeforeCutoff,
+            organizationOrphanAtCutoff,
+            organizationOrphanBeforeCutoff
+        ], o => o.ImmediateConsistency());
+
+        await _job.RunAsync(TestCancellationToken);
+
+        var remainingEvents = await _eventRepository.GetAllAsync(o => o.PageLimit(10).ImmediateConsistency());
+        Assert.Equal(4, remainingEvents.Total);
+        Assert.Contains(remainingEvents.Documents, e => e.Id == validEvent.Id);
+        Assert.Contains(remainingEvents.Documents, e => e.Id == stackOrphanBeforeCutoff.Id);
+        Assert.Contains(remainingEvents.Documents, e => e.Id == projectOrphanBeforeCutoff.Id);
+        Assert.Contains(remainingEvents.Documents, e => e.Id == organizationOrphanBeforeCutoff.Id);
+        Assert.DoesNotContain(remainingEvents.Documents, e => e.Id == stackOrphanAtCutoff.Id);
+        Assert.DoesNotContain(remainingEvents.Documents, e => e.Id == projectOrphanAtCutoff.Id);
+        Assert.DoesNotContain(remainingEvents.Documents, e => e.Id == organizationOrphanAtCutoff.Id);
+    }
+
     [Fact]
     public async Task FixDuplicateStacks_WithDuplicatesAcrossTenants_MergesCorrectly()
     {
@@ -463,7 +523,7 @@
         Assert.Equal(40, totalEvents);
     }
 
     [Fact]
     public async Task RunAsync_AllOrphanTypes_CleansUpCorrectly()
     {
         // Arrange - Complex scenario: valid data, orphaned by stack, orphaned by project, orphaned by organization