Enable PurgeInstanceStateAsync to delete more than 1000 instances (#308)

Author: usemam

src/DurableTask.SqlServer/SqlOrchestrationService.cs

@@ -729,7 +729,7 @@ public override async Task<PurgeResult> PurgeInstanceStateAsync(PurgeInstanceFil
                 FetchInput = false,
                 FetchOutput = false,
             };
-            
+
             if (purgeInstanceFilter.CreatedTimeTo != null)
             {
                 purgeQuery.CreatedTimeTo = purgeInstanceFilter.CreatedTimeTo.Value;
@@ -739,12 +739,23 @@ public override async Task<PurgeResult> PurgeInstanceStateAsync(PurgeInstanceFil
             {
                 purgeQuery.StatusFilter = new HashSet<OrchestrationStatus>(purgeInstanceFilter.RuntimeStatus);
             }
-            
-            IReadOnlyCollection<OrchestrationState> results = await this.GetManyOrchestrationsAsync(purgeQuery, CancellationToken.None);
-                                                                                                  
-            IEnumerable<string> instanceIds = results.Select(r => r.OrchestrationInstance.InstanceId);
-            int purgedInstanceCount = await this.PurgeOrchestrationHistoryAsync(instanceIds);
-            return new PurgeResult(purgedInstanceCount);
+
+            int totalPurgedCount = 0;
+            while (true)
+            {
+                IReadOnlyCollection<OrchestrationState> results =
+                    await this.GetManyOrchestrationsAsync(purgeQuery, CancellationToken.None);
+
+                if (results.Count == 0)
+                {
+                    break;
+                }
+
+                IEnumerable<string> instanceIds = results.Select(r => r.OrchestrationInstance.InstanceId);
+                totalPurgedCount += await this.PurgeOrchestrationHistoryAsync(instanceIds);
+            }
+
+            return new PurgeResult(totalPurgedCount);
         }
 
         public override async Task<OrchestrationQueryResult> GetOrchestrationWithQueryAsync(OrchestrationQuery query, CancellationToken cancellationToken)

test/DurableTask.SqlServer.Tests/Integration/DataRetentionTests.cs

@@ -125,6 +125,42 @@ public async Task PurgesInstancesByStatus(OrchestrationStateTimeRangeFilterType
             await this.testService.PurgeAsync(DateTime.MaxValue, filterType);
         }
 
+        /// <summary>
+        /// Verifies that PurgeInstanceStateAsync with a PurgeInstanceFilter purges all matching instances,
+        /// even when the count exceeds the previous 1000-instance limit.
+        /// This is a regression test for https://github.com/microsoft/durabletask-mssql/issues/297.
+        /// </summary>
+        [Fact]
+        public async Task PurgesMoreThanOneThousandInstances()
+        {
+            const int InstanceCount = 1100;
+
+            IReadOnlyList<TestInstance<string>> instances = await this.testService.RunOrchestrations(
+                count: InstanceCount,
+                instanceIdGenerator: i => $"BulkPurge_{i:0000}",
+                inputGenerator: i => $"Input_{i}",
+                orchestrationName: "QuickComplete",
+                version: string.Empty,
+                implementation: (ctx, input) => Task.FromResult(input));
+
+            TimeSpan timeout = TimeSpan.FromSeconds(60);
+            await Task.WhenAll(instances.Select(instance => instance.WaitForCompletion(timeout)));
+
+            var filter = new PurgeInstanceFilter(
+                DateTime.MinValue,
+                DateTime.UtcNow.AddMinutes(1),
+                new[] { OrchestrationStatus.Completed });
+
+            PurgeResult result = await this.testService.PurgeAsync(filter);
+            Assert.Equal(InstanceCount, result.DeletedInstanceCount);
+
+            foreach (TestInstance<string> instance in instances)
+            {
+                OrchestrationState purgedState = await instance.GetStateAsync();
+                Assert.Null(purgedState);
+            }
+        }
+
         /// <summary>
         /// Validates that external events sent to a completed orchestration are eventually removed from the database
         /// and that a log message is emitted for each discarded event.

test/DurableTask.SqlServer.Tests/Utils/TestService.cs

@@ -107,6 +107,11 @@ public Task PurgeAsync(DateTime maximumThreshold, OrchestrationStateTimeRangeFil
                 filterType);
         }
 
+        public Task<PurgeResult> PurgeAsync(PurgeInstanceFilter filter)
+        {
+            return this.OrchestrationServiceMock.Object.PurgeInstanceStateAsync(filter);
+        }
+
         public async Task DisposeAsync()
         {
             await this.worker.StopAsync(isForced: true);