diff --git a/samples/ExportHistoryWebApp/README.md b/samples/ExportHistoryWebApp/README.md
index 88f9dc0c2..473034068 100644
--- a/samples/ExportHistoryWebApp/README.md
+++ b/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
diff --git a/src/ExportHistory/Entity/ExportJob.cs b/src/ExportHistory/Entity/ExportJob.cs
index 7ccce1ad1..9dc6b3c73 100644
--- a/src/ExportHistory/Entity/ExportJob.cs
+++ b/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;
diff --git a/src/ExportHistory/Models/ExportCheckpoint.cs b/src/ExportHistory/Models/ExportCheckpoint.cs
index 561dad6ea..5393b79ee 100644
--- a/src/ExportHistory/Models/ExportCheckpoint.cs
+++ b/src/ExportHistory/Models/ExportCheckpoint.cs
@@ -6,4 +6,17 @@ namespace Microsoft.DurableTask.ExportHistory;
///
/// Checkpoint information used to resume export.
///
-public sealed record ExportCheckpoint(string? LastInstanceKey = null);
+public sealed record ExportCheckpoint(string? LastInstanceKey = null)
+{
+ ///
+ /// Gets the inclusive lower completion-time bound for continuous export.
+ /// A null value identifies a legacy instance-key-only checkpoint.
+ ///
+ public DateTimeOffset? CompletedTimeFrom { get; init; }
+
+ ///
+ /// 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.
+ ///
+ public DateTimeOffset? CompletedTimeTo { get; init; }
+}
diff --git a/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs b/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs
index 7112c79eb..3d4fa590f 100644
--- a/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs
+++ b/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs
@@ -9,7 +9,14 @@ namespace Microsoft.DurableTask.ExportHistory;
///
/// Orchestrator input to start a runner for a given job.
///
-public sealed record ExportJobRunRequest(EntityInstanceId JobEntityId, int ProcessedCycles = 0);
+public sealed record ExportJobRunRequest(EntityInstanceId JobEntityId, int ProcessedCycles = 0)
+{
+ ///
+ /// Gets a value indicating whether continuous export uses completion-time windows.
+ /// Defaults to false to preserve replay of existing histories until continue-as-new.
+ ///
+ public bool UseCompletionTimeWindows { get; init; }
+}
///
/// Orchestrator that performs the actual export work by querying orchestration instances
@@ -68,7 +75,10 @@ public class ExportJobOrchestrator : TaskOrchestrator 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 listRequest.CompletedTimeTo)
+ {
+ await context.CreateTimer(ContinuousExportIdleDelay, default);
+ continue;
+ }
+ }
+
InstancePage pageResult = await context.CallActivityAsync(
nameof(ListTerminalInstancesActivity),
listRequest);
List 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
{
diff --git a/test/ExportHistory.Tests/Entity/ExportJobTests.cs b/test/ExportHistory.Tests/Entity/ExportJobTests.cs
index 10a780443..f17bce9c9 100644
--- a/test/ExportHistory.Tests/Entity/ExportJobTests.cs
+++ b/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(r => r.JobEntityId == entityId && r.UseCompletionTimeWindows),
+ It.IsAny()), 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);
}
}
-
diff --git a/test/ExportHistory.Tests/Models/ExportCheckpointTests.cs b/test/ExportHistory.Tests/Models/ExportCheckpointTests.cs
index 2fb12f566..a5d369d9c 100644
--- a/test/ExportHistory.Tests/Models/ExportCheckpointTests.cs
+++ b/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(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(
+ 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);
}
}
-
diff --git a/test/ExportHistory.Tests/Orchestrations/ExportJobOrchestratorTests.cs b/test/ExportHistory.Tests/Orchestrations/ExportJobOrchestratorTests.cs
new file mode 100644
index 000000000..9410ef10d
--- /dev/null
+++ b/test/ExportHistory.Tests/Orchestrations/ExportJobOrchestratorTests.cs
@@ -0,0 +1,389 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.DurableTask.Client;
+using Microsoft.DurableTask.Converters;
+using Microsoft.DurableTask.Entities;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace Microsoft.DurableTask.ExportHistory.Tests.Orchestrations;
+
+public class ExportJobOrchestratorTests
+{
+ static readonly DateTimeOffset PollTime = new(2026, 1, 1, 12, 0, 0, TimeSpan.Zero);
+
+ [Fact]
+ public void Deserialize_LegacyRunRequest_PreservesReplayBehavior()
+ {
+ // Arrange
+ EntityInstanceId entityId = new(nameof(ExportJob), "test-job");
+ string? json = JsonDataConverter.Default.Serialize(new { JobEntityId = entityId, ProcessedCycles = 2 });
+
+ // Act
+ ExportJobRunRequest request = JsonDataConverter.Default.Deserialize(json)!;
+
+ // Assert
+ Assert.Equal(entityId, request.JobEntityId);
+ Assert.Equal(2, request.ProcessedCycles);
+ Assert.False(request.UseCompletionTimeWindows);
+ }
+
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)]
+ public async Task RunAsync_OutOfOrderCompletion_ExportsEarlierCreatedInstance(bool legacyCheckpoint)
+ {
+ // Arrange
+ TestRunner runner = new();
+ runner.State.Config = runner.State.Config! with { MaxInstancesPerBatch = 1 };
+ runner.State.Checkpoint = legacyCheckpoint ? new ExportCheckpoint("2") : null;
+ Mock client = new("test-client");
+ client.Setup(c => c.ListInstanceIdsAsync(
+ It.IsAny?>(),
+ It.IsAny(),
+ It.IsAny(),
+ It.IsAny(),
+ It.IsAny(),
+ It.IsAny()))
+ .Returns((
+ IEnumerable? statuses,
+ DateTimeOffset? from,
+ DateTimeOffset? to,
+ int pageSize,
+ string? key,
+ CancellationToken cancellation) =>
+ {
+ // A has an earlier creation key, but completes after B and after the first poll.
+ (string Id, string Key, DateTimeOffset Completed)[] instances =
+ [
+ ("A", "1", PollTime.AddSeconds(30)),
+ ("B", "2", PollTime.AddSeconds(-10)),
+ ];
+ var matches = instances
+ .Where(i => i.Completed <= runner.Now && i.Completed >= from && i.Completed <= to)
+ .Where(i => key == null || string.CompareOrdinal(i.Key, key) > 0)
+ .Take(pageSize)
+ .ToList();
+ return Task.FromResult(new Page(
+ matches.Select(i => i.Id).ToList(),
+ matches.Count == 0 ? key : matches[^1].Key));
+ });
+ ListTerminalInstancesActivity activity = new(client.Object, NullLogger.Instance);
+ runner.ListPage = request => activity.RunAsync(Mock.Of(), request);
+
+ // Act
+ await runner.RunAsync();
+
+ // Assert
+ Assert.Equal(new[] { "B", "A" }, runner.ExportedIds);
+ Assert.Null(runner.ListRequests[0].LastInstanceKey);
+ Assert.Equal(PollTime, runner.ListRequests[0].CompletedTimeTo);
+ Assert.Equal("2", runner.ListRequests[1].LastInstanceKey);
+ Assert.Equal(PollTime, runner.ListRequests[1].CompletedTimeTo);
+ Assert.Null(runner.ListRequests[2].LastInstanceKey);
+ Assert.Equal(PollTime, runner.ListRequests[2].CompletedTimeFrom);
+ Assert.Equal(PollTime.AddMinutes(1), runner.ListRequests[2].CompletedTimeTo);
+ Assert.Equal(2, runner.State.ExportedInstances);
+ }
+
+ [Fact]
+ public async Task RunAsync_ContinueAsNew_ResumesSameWindowAndCursor()
+ {
+ // Arrange
+ TestRunner runner = new();
+ runner.ListPage = _ => Task.FromResult(new InstancePage(["instance"], new ExportCheckpoint("next-key")));
+
+ // Act
+ await runner.RunAsync(new ExportJobRunRequest(runner.EntityId, ProcessedCycles: 4)
+ {
+ UseCompletionTimeWindows = true,
+ });
+ ExportJobRunRequest continued = Assert.IsType(runner.ContinuedInput);
+ runner.Now = PollTime.AddHours(1);
+ runner.ListPage = _ =>
+ {
+ runner.State.Status = ExportJobStatus.Completed;
+ return Task.FromResult(new InstancePage([], new ExportCheckpoint()));
+ };
+ await runner.RunAsync(continued);
+
+ // Assert
+ Assert.True(continued.UseCompletionTimeWindows);
+ Assert.Equal("next-key", runner.ListRequests[1].LastInstanceKey);
+ Assert.Equal(PollTime.AddHours(-1), runner.ListRequests[1].CompletedTimeFrom);
+ Assert.Equal(PollTime, runner.ListRequests[1].CompletedTimeTo);
+ }
+
+ [Fact]
+ public async Task RunAsync_InclusiveBoundary_IsNotSkippedBetweenWindows()
+ {
+ // Arrange
+ TestRunner runner = new();
+ runner.ListPage = request =>
+ {
+ if (runner.ListRequests.Count == 3)
+ {
+ runner.State.Status = ExportJobStatus.Completed;
+ }
+
+ List ids = request.CompletedTimeFrom <= PollTime && request.CompletedTimeTo >= PollTime
+ ? ["boundary-instance"]
+ : [];
+ return Task.FromResult(new InstancePage(ids, new ExportCheckpoint()));
+ };
+
+ // Act
+ await runner.RunAsync();
+
+ // Assert
+ Assert.Equal(new[] { "boundary-instance", "boundary-instance" }, runner.ExportedIds);
+ Assert.Equal(PollTime, runner.ListRequests[1].CompletedTimeFrom);
+ Assert.Equal(PollTime.AddMinutes(1), runner.ListRequests[2].CompletedTimeFrom);
+ Assert.All(runner.ListRequests, request => Assert.Null(request.LastInstanceKey));
+ }
+
+ [Theory]
+ [InlineData(null)]
+ [InlineData("")]
+ [InlineData("previous-key")]
+ public async Task RunAsync_EmptyFinalPage_AdvancesWatermarkAndResetsCursor(string? nextKey)
+ {
+ // Arrange
+ TestRunner runner = new();
+ runner.State.Checkpoint = new ExportCheckpoint("previous-key")
+ {
+ CompletedTimeFrom = PollTime.AddHours(-1),
+ CompletedTimeTo = PollTime,
+ };
+ runner.ListPage = _ =>
+ {
+ runner.State.Status = ExportJobStatus.Completed;
+ return Task.FromResult(new InstancePage([], new ExportCheckpoint(nextKey)));
+ };
+
+ // Act
+ await runner.RunAsync();
+
+ // Assert
+ Assert.Equal(new ExportCheckpoint { CompletedTimeFrom = PollTime }, runner.State.Checkpoint);
+ Assert.Equal(1, runner.TimerCount);
+ Assert.Empty(runner.ExportedIds);
+ Assert.Equal(0, runner.State.ExportedInstances);
+ }
+
+ [Fact]
+ public async Task RunAsync_EmptyPageWithAdvancingCursor_KeepsPagingSameWindow()
+ {
+ // Arrange
+ TestRunner runner = new();
+ runner.ListPage = _ =>
+ {
+ if (runner.ListRequests.Count == 1)
+ {
+ return Task.FromResult(new InstancePage([], new ExportCheckpoint("scanned-key")));
+ }
+
+ runner.State.Status = ExportJobStatus.Completed;
+ return Task.FromResult(new InstancePage(["instance"], new ExportCheckpoint()));
+ };
+
+ // Act
+ await runner.RunAsync();
+
+ // Assert
+ Assert.Equal("scanned-key", runner.ListRequests[1].LastInstanceKey);
+ Assert.Equal(PollTime, runner.ListRequests[1].CompletedTimeTo);
+ Assert.Equal(1, runner.TimerCount);
+ Assert.Single(runner.ExportedIds);
+ }
+
+ [Fact]
+ public async Task RunAsync_FailedExport_DoesNotAdvanceWindowOrCursor()
+ {
+ // Arrange
+ TestRunner runner = new();
+ ExportCheckpoint checkpoint = new("previous-key")
+ {
+ CompletedTimeFrom = PollTime.AddHours(-1),
+ CompletedTimeTo = PollTime,
+ };
+ runner.State.Checkpoint = checkpoint;
+ runner.ExportSucceeds = false;
+ runner.ListPage = _ => Task.FromResult(new InstancePage(["instance"], new ExportCheckpoint()));
+
+ // Act
+ await Assert.ThrowsAsync(() => runner.RunAsync());
+
+ // Assert
+ Assert.Equal(checkpoint, runner.State.Checkpoint);
+ Assert.Equal(0, runner.State.ExportedInstances);
+ Assert.Equal(ExportJobStatus.Failed, runner.State.Status);
+ }
+
+ [Fact]
+ public async Task RunAsync_FutureLowerBound_WaitsWithoutListing()
+ {
+ // Arrange
+ TestRunner runner = new();
+ runner.State.Config = runner.State.Config! with
+ {
+ Filter = new ExportFilter(PollTime.AddDays(1), null, null),
+ };
+
+ // Act
+ await runner.RunAsync();
+
+ // Assert
+ Assert.Empty(runner.ListRequests);
+ Assert.Equal(5, runner.TimerCount);
+ Assert.Null(runner.State.Checkpoint);
+ }
+
+ [Theory]
+ [InlineData(ExportMode.Continuous, false)]
+ [InlineData(ExportMode.Batch, true)]
+ public async Task RunAsync_LegacyHistoryOrBatch_PreservesListingBehavior(ExportMode mode, bool useWindows)
+ {
+ // Arrange
+ TestRunner runner = new();
+ runner.State.Config = runner.State.Config! with
+ {
+ Mode = mode,
+ Filter = new ExportFilter(PollTime.AddHours(-1), mode == ExportMode.Batch ? PollTime : null, null),
+ };
+ runner.State.Checkpoint = new ExportCheckpoint("legacy-key");
+
+ // Act
+ await runner.RunAsync(new ExportJobRunRequest(runner.EntityId) { UseCompletionTimeWindows = useWindows });
+
+ // Assert
+ Assert.All(runner.ListRequests, request =>
+ {
+ Assert.Equal("legacy-key", request.LastInstanceKey);
+ Assert.Equal(runner.State.Config.Filter.CompletedTimeTo, request.CompletedTimeTo);
+ });
+ if (mode == ExportMode.Continuous)
+ {
+ Assert.True(Assert.IsType(runner.ContinuedInput).UseCompletionTimeWindows);
+ }
+ else
+ {
+ Assert.Equal(ExportJobStatus.Completed, runner.State.Status);
+ Assert.Equal(0, runner.TimerCount);
+ }
+ }
+
+ sealed class TestRunner
+ {
+ readonly Mock context = new(MockBehavior.Strict);
+ readonly Mock entities = new(MockBehavior.Strict);
+
+ public TestRunner()
+ {
+ this.context.Setup(c => c.Entities).Returns(this.entities.Object);
+ this.context.Setup(c => c.CurrentUtcDateTime).Returns(() => this.Now.UtcDateTime);
+ this.context.Setup(c => c.CreateReplaySafeLogger()).Returns(NullLogger.Instance);
+ this.context.Setup(c => c.CreateTimer(It.IsAny(), It.IsAny()))
+ .Returns((TimeSpan delay, CancellationToken cancellation) =>
+ {
+ this.Now += delay;
+ this.TimerCount++;
+ return Task.CompletedTask;
+ });
+ this.context.Setup(c => c.ContinueAsNew(It.IsAny