Fix continuous history export skipping out-of-order completions

samples/ExportHistoryWebApp/README.md

@@ -66,6 +66,10 @@ Request body (see `Models/CreateExportJobRequest.cs`):
 
 Notes:
 - For `Batch` mode, `completedTimeTo` must be greater than `completedTimeFrom` and cannot be in the future.
+- `Continuous` mode pages through fixed completion-time windows, then resets the instance-key cursor for the next window. This allows earlier-created instances that complete after the current window's upper bound to be picked up in a subsequent window, provided they are query-visible when that window is scanned.
+- Window upper bounds use the orchestrator's clock, not a server snapshot watermark. Delayed query visibility or clock skew can still leave an instance behind the pagination cursor or advanced completion-time bound, causing an export gap. If the backend does not guarantee timely query visibility and compatible timestamps, retain source histories and reconcile exports before purging; continuous export alone does not guarantee that every terminal instance has been exported.
+- Completion-time bounds are inclusive, so instances exactly on a window boundary can be exported again. Retries can also repeat exports; deterministic blob names keep these writes idempotent. Progress counters count successful export operations, not distinct blobs.
+- After upgrading the worker, existing continuous jobs adopt completion-time windows at their next continue-as-new (after at most five polling/page cycles). Legacy checkpoints restart from the job's original `completedTimeFrom` to recover previously skipped instances, which can re-export existing blobs. Keep source histories available until reconciliation finishes; already-purged histories cannot be recovered.
 
 ### Get a job
 

src/ExportHistory/Entity/ExportJob.cs

@@ -272,7 +272,7 @@ void StartExportOrchestration(TaskEntityContext context)
 
             context.ScheduleNewOrchestration(
                 new TaskName(nameof(ExportJobOrchestrator)),
-                new ExportJobRunRequest(context.Id),
+                new ExportJobRunRequest(context.Id) { UseCompletionTimeWindows = true },
                 startOrchestrationOptions);
 
             this.State.OrchestratorInstanceId = instanceId;

src/ExportHistory/Models/ExportCheckpoint.cs

@@ -6,4 +6,17 @@ namespace Microsoft.DurableTask.ExportHistory;
 /// <summary>
 /// Checkpoint information used to resume export.
 /// </summary>
-public sealed record ExportCheckpoint(string? LastInstanceKey = null);
+public sealed record ExportCheckpoint(string? LastInstanceKey = null)
+{
+    /// <summary>
+    /// Gets the inclusive lower completion-time bound for continuous export.
+    /// A null value identifies a legacy instance-key-only checkpoint.
+    /// </summary>
+    public DateTimeOffset? CompletedTimeFrom { get; init; }
+
+    /// <summary>
+    /// Gets the fixed inclusive upper completion-time bound while paging a continuous export window.
+    /// A null value indicates that the next window has not started.
+    /// </summary>
+    public DateTimeOffset? CompletedTimeTo { get; init; }
+}

src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs

@@ -9,7 +9,14 @@ namespace Microsoft.DurableTask.ExportHistory;
 /// <summary>
 /// Orchestrator input to start a runner for a given job.
 /// </summary>
-public sealed record ExportJobRunRequest(EntityInstanceId JobEntityId, int ProcessedCycles = 0);
+public sealed record ExportJobRunRequest(EntityInstanceId JobEntityId, int ProcessedCycles = 0)
+{
+    /// <summary>
+    /// Gets a value indicating whether continuous export uses completion-time windows.
+    /// Defaults to false to preserve replay of existing histories until continue-as-new.
+    /// </summary>
+    public bool UseCompletionTimeWindows { get; init; }
+}
 
 /// <summary>
 /// Orchestrator that performs the actual export work by querying orchestration instances
@@ -68,7 +75,10 @@ public class ExportJobOrchestrator : TaskOrchestrator<ExportJobRunRequest, objec
                 processedCycles++;
                 if (processedCycles > ContinueAsNewFrequency)
                 {
-                    context.ContinueAsNew(new ExportJobRunRequest(input.JobEntityId, ProcessedCycles: 0));
+                    context.ContinueAsNew(new ExportJobRunRequest(input.JobEntityId, ProcessedCycles: 0)
+                    {
+                        UseCompletionTimeWindows = true,
+                    });
                     return null!;
                 }
 
@@ -94,15 +104,74 @@ public class ExportJobOrchestrator : TaskOrchestrator<ExportJobRunRequest, objec
                     LastInstanceKey: currentState.Checkpoint?.LastInstanceKey,
                     MaxInstancesPerBatch: currentState.Config.MaxInstancesPerBatch);
 
+                bool useCompletionTimeWindows = input.UseCompletionTimeWindows && config.Mode == ExportMode.Continuous;
+                if (useCompletionTimeWindows)
+                {
+                    // A creation-order cursor is only safe within a fixed completion-time window.
+                    // Legacy checkpoints must restart from the filter to recover skipped instances.
+                    ExportCheckpoint? checkpoint = currentState.Checkpoint;
+                    listRequest = listRequest with
+                    {
+                        CompletedTimeFrom = checkpoint?.CompletedTimeFrom ?? listRequest.CompletedTimeFrom,
+                        CompletedTimeTo = checkpoint?.CompletedTimeTo ?? new DateTimeOffset(context.CurrentUtcDateTime),
+                        LastInstanceKey = checkpoint?.CompletedTimeTo.HasValue == true ? checkpoint?.LastInstanceKey : null,
+                    };
+
+                    if (listRequest.CompletedTimeFrom > listRequest.CompletedTimeTo)
+                    {
+                        await context.CreateTimer(ContinuousExportIdleDelay, default);
+                        continue;
+                    }
+                }
+
                 InstancePage pageResult = await context.CallActivityAsync<InstancePage>(
                     nameof(ListTerminalInstancesActivity),
                     listRequest);
 
                 List<string> instancesToExport = pageResult.InstanceIds;
                 long scannedCount = instancesToExport.Count;
+                ExportCheckpoint nextCheckpoint = pageResult.NextCheckpoint;
+                bool windowComplete = false;
+
+                if (useCompletionTimeWindows)
+                {
+                    string? nextKey = nextCheckpoint.LastInstanceKey;
+                    windowComplete = string.IsNullOrEmpty(nextKey) ||
+                        (scannedCount == 0 && nextKey == listRequest.LastInstanceKey);
+                    nextCheckpoint = windowComplete
+                        ? new ExportCheckpoint
+                        {
+                            // Bounds are inclusive; overlap the boundary rather than risk missing an instance.
+                            CompletedTimeFrom = listRequest.CompletedTimeTo,
+                        }
+                        : new ExportCheckpoint(nextKey)
+                        {
+                            CompletedTimeFrom = listRequest.CompletedTimeFrom,
+                            CompletedTimeTo = listRequest.CompletedTimeTo,
+                        };
+                }
 
                 if (scannedCount == 0)
                 {
+                    if (useCompletionTimeWindows)
+                    {
+                        await this.CommitCheckpointAsync(
+                            context,
+                            input.JobEntityId,
+                            scannedInstances: 0,
+                            exportedInstances: 0,
+                            checkpoint: nextCheckpoint,
+                            failures: null);
+
+                        if (windowComplete)
+                        {
+                            logger.ExportJobOperationInfo(jobId, nameof(ExportJobOrchestrator), "Completion-time window exported - waiting for the next poll.");
+                            await context.CreateTimer(ContinuousExportIdleDelay, default);
+                        }
+
+                        continue;
+                    }
+
                     if (config.Mode == ExportMode.Continuous)
                     {
                         logger.ExportJobOperationInfo(jobId, nameof(ExportJobOrchestrator), "No more instances to export - export complete.");
@@ -136,8 +205,13 @@ await this.CommitCheckpointAsync(
                         input.JobEntityId,
                         scannedInstances: scannedCount,
                         exportedInstances: batchResult.ExportedCount,
-                        checkpoint: pageResult.NextCheckpoint,
+                        checkpoint: nextCheckpoint,
                         failures: null);
+
+                    if (windowComplete)
+                    {
+                        await context.CreateTimer(ContinuousExportIdleDelay, default);
+                    }
                 }
                 else
                 {

test/ExportHistory.Tests/Entity/ExportJobTests.cs

@@ -1,7 +1,9 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using Microsoft.DurableTask.Entities;
 using Microsoft.DurableTask.Entities.Tests;
+using Moq;
 using Xunit;
 
 namespace Microsoft.DurableTask.ExportHistory.Tests.Entity;
@@ -17,6 +19,33 @@ public ExportJobTests()
         this.exportJob = new ExportJob(this.logger);
     }
 
+    [Fact]
+    public async Task Run_ActiveJob_EnablesCompletionTimeWindows()
+    {
+        // Arrange
+        EntityInstanceId entityId = new(nameof(ExportJob), "test-job");
+        ExportJobState state = new()
+        {
+            Status = ExportJobStatus.Active,
+            Config = new ExportJobConfiguration(
+                ExportMode.Continuous,
+                new ExportFilter(DateTimeOffset.UtcNow.AddDays(-1), null, null),
+                new ExportDestination("container"),
+                ExportFormat.Default),
+        };
+        TestEntityOperation operation = new(nameof(ExportJob.Run), new TestEntityState(state), null);
+        Mock.Get(operation.Context).Setup(c => c.Id).Returns(entityId);
+
+        // Act
+        await this.exportJob.RunAsync(operation);
+
+        // Assert
+        Mock.Get(operation.Context).Verify(c => c.ScheduleNewOrchestration(
+            new TaskName(nameof(ExportJobOrchestrator)),
+            It.Is<ExportJobRunRequest>(r => r.JobEntityId == entityId && r.UseCompletionTimeWindows),
+            It.IsAny<StartOrchestrationOptions>()), Times.Once);
+    }
+
     [Fact]
     public async Task Create_WithValidOptions_CreatesJob()
     {
@@ -367,4 +396,3 @@ public async Task Create_WhenFailed_CanRecreate()
         jobState.Status.Should().Be(ExportJobStatus.Active);
     }
 }
-

test/ExportHistory.Tests/Models/ExportCheckpointTests.cs

@@ -2,12 +2,46 @@
 // Licensed under the MIT License.
 
 using FluentAssertions;
+using Microsoft.DurableTask.Converters;
 using Xunit;
 
 namespace Microsoft.DurableTask.ExportHistory.Tests.Models;
 
 public class ExportCheckpointTests
 {
+    [Fact]
+    public void Deserialize_LegacyCheckpoint_PreservesKeyWithoutWindow()
+    {
+        // Arrange
+        string json = """{"LastInstanceKey":"legacy-key"}""";
+
+        // Act
+        ExportCheckpoint checkpoint = JsonDataConverter.Default.Deserialize<ExportCheckpoint>(json)!;
+
+        // Assert
+        Assert.Equal("legacy-key", checkpoint.LastInstanceKey);
+        Assert.Null(checkpoint.CompletedTimeFrom);
+        Assert.Null(checkpoint.CompletedTimeTo);
+    }
+
+    [Fact]
+    public void Serialize_WindowCheckpoint_RoundTrips()
+    {
+        // Arrange
+        ExportCheckpoint checkpoint = new("page-key")
+        {
+            CompletedTimeFrom = new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero),
+            CompletedTimeTo = new DateTimeOffset(2026, 1, 2, 0, 0, 0, TimeSpan.Zero),
+        };
+
+        // Act
+        ExportCheckpoint? result = JsonDataConverter.Default.Deserialize<ExportCheckpoint>(
+            JsonDataConverter.Default.Serialize(checkpoint));
+
+        // Assert
+        Assert.Equal(checkpoint, result);
+    }
+
     [Fact]
     public void Constructor_WithNullLastInstanceKey_CreatesInstance()
     {
@@ -68,4 +102,3 @@ public void Record_Inequality_Works()
         checkpoint1.Should().NotBe(checkpoint2);
     }
 }
-