From 213649df100e2e95532897b2c6f312b344961acd Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Sun, 1 Nov 2026 00:13:24 -0700 Subject: [PATCH 1/2] Fix continuous export skipping out-of-order completions Scope creation-order pagination to persistent completion-time windows. Adopt the new traversal at a replay-safe continue-as-new boundary and reconcile legacy checkpoints from their original completion filter. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- samples/ExportHistoryWebApp/README.md | 3 + src/ExportHistory/Entity/ExportJob.cs | 2 +- src/ExportHistory/Models/ExportCheckpoint.cs | 15 +- .../Orchestrations/ExportJobOrchestrator.cs | 80 +++- .../Entity/ExportJobTests.cs | 30 +- .../Models/ExportCheckpointTests.cs | 35 +- .../ExportJobOrchestratorTests.cs | 389 ++++++++++++++++++ 7 files changed, 547 insertions(+), 7 deletions(-) create mode 100644 test/ExportHistory.Tests/Orchestrations/ExportJobOrchestratorTests.cs diff --git a/samples/ExportHistoryWebApp/README.md b/samples/ExportHistoryWebApp/README.md index 88f9dc0c2..a90fe52bf 100644 --- a/samples/ExportHistoryWebApp/README.md +++ b/samples/ExportHistoryWebApp/README.md @@ -66,6 +66,9 @@ 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. Instances that finish out of creation order are picked up in a subsequent window. +- 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..91f6202d3 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(), true)) + .Callback((object? input, bool preserveEvents) => this.ContinuedInput = input); + this.entities.Setup(e => e.CallEntityAsync( + this.EntityId, nameof(ExportJob.Get), null, null)) + .ReturnsAsync(() => this.State); + this.entities.Setup(e => e.CallEntityAsync( + this.EntityId, nameof(ExportJob.Get), null)) + .CallBase(); + this.entities.Setup(e => e.CallEntityAsync( + this.EntityId, nameof(ExportJob.CommitCheckpoint), It.IsAny(), null)) + .Returns((EntityInstanceId id, string operation, object? input, CallEntityOptions? options) => + { + CommitCheckpointRequest request = Assert.IsType(input); + if (request.Checkpoint != null) + { + this.State.Checkpoint = JsonDataConverter.Default.Deserialize( + JsonDataConverter.Default.Serialize(request.Checkpoint)); + } + + this.State.ExportedInstances += request.ExportedInstances; + return Task.CompletedTask; + }); + this.entities.Setup(e => e.CallEntityAsync( + this.EntityId, nameof(ExportJob.MarkAsCompleted), null, null)) + .Returns(() => + { + this.State.Status = ExportJobStatus.Completed; + return Task.CompletedTask; + }); + this.entities.Setup(e => e.CallEntityAsync( + this.EntityId, nameof(ExportJob.MarkAsCompleted), null)) + .CallBase(); + this.entities.Setup(e => e.CallEntityAsync( + this.EntityId, nameof(ExportJob.MarkAsFailed), It.IsAny(), null)) + .Returns(() => + { + this.State.Status = ExportJobStatus.Failed; + return Task.CompletedTask; + }); + this.context.Setup(c => c.CallActivityAsync( + nameof(ListTerminalInstancesActivity), It.IsAny(), null)) + .Returns((TaskName name, object? input, TaskOptions? options) => + { + ListTerminalInstancesRequest request = Assert.IsType(input); + this.ListRequests.Add(request); + return this.ListPage(request); + }); + this.context.Setup(c => c.CallActivityAsync( + nameof(ExportInstanceHistoryActivity), It.IsAny(), It.IsAny())) + .Returns((TaskName name, object? input, TaskOptions? options) => + { + ExportRequest request = Assert.IsType(input); + this.ExportedIds.Add(request.InstanceId); + return Task.FromResult(new ExportResult + { + InstanceId = request.InstanceId, + Success = this.ExportSucceeds, + Error = this.ExportSucceeds ? null : "Export failed", + }); + }); + } + + public EntityInstanceId EntityId { get; } = new(nameof(ExportJob), "test-job"); + + public ExportJobState State { get; } = new() + { + Status = ExportJobStatus.Active, + Config = new ExportJobConfiguration( + ExportMode.Continuous, + new ExportFilter(PollTime.AddHours(-1), null, null), + new ExportDestination("container"), + ExportFormat.Default), + }; + + public DateTimeOffset Now { get; set; } = PollTime; + + public Func> ListPage { get; set; } = + _ => Task.FromResult(new InstancePage([], new ExportCheckpoint())); + + public List ListRequests { get; } = []; + + public List ExportedIds { get; } = []; + + public bool ExportSucceeds { get; set; } = true; + + public int TimerCount { get; private set; } + + public object? ContinuedInput { get; private set; } + + public Task RunAsync(ExportJobRunRequest? input = null) => + new ExportJobOrchestrator().RunAsync( + this.context.Object, + input ?? new ExportJobRunRequest(this.EntityId) { UseCompletionTimeWindows = true }); + } +} From c0e73fe76a831215d436e0dd3baf6cd0a536920b Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Sun, 1 Nov 2026 01:37:08 -0700 Subject: [PATCH 2/2] Address export history PR review feedback Use null-conditional checkpoint key access and document potential export gaps from delayed query visibility or clock skew, including reconciliation before purging. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- samples/ExportHistoryWebApp/README.md | 3 ++- src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/samples/ExportHistoryWebApp/README.md b/samples/ExportHistoryWebApp/README.md index a90fe52bf..473034068 100644 --- a/samples/ExportHistoryWebApp/README.md +++ b/samples/ExportHistoryWebApp/README.md @@ -66,7 +66,8 @@ 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. Instances that finish out of creation order are picked up in a subsequent window. +- `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. diff --git a/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs b/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs index 91f6202d3..3d4fa590f 100644 --- a/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs +++ b/src/ExportHistory/Orchestrations/ExportJobOrchestrator.cs @@ -114,7 +114,7 @@ public class ExportJobOrchestrator : TaskOrchestrator listRequest.CompletedTimeTo)