From 77a84d52f7458c29531df6ffa1159561e3f7efe6 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:32:23 +0000 Subject: [PATCH 1/5] Initial plan From ff41f984f3793fc605d8eac5d256c8fe930efa9d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:42:06 +0000 Subject: [PATCH 2/5] Support explicit orchestration cloning with version and source lineage Co-authored-by: bachuv <15795068+bachuv@users.noreply.github.com> --- client/build.gradle | 7 +- .../durabletask/DurableTaskClient.java | 29 ++++ .../durabletask/DurableTaskGrpcClient.java | 15 ++ .../durabletask/TaskOrchestrationContext.java | 11 ++ .../TaskOrchestrationExecutor.java | 10 ++ .../DurableTaskGrpcClientRestartTest.java | 158 ++++++++++++++++++ .../TaskOrchestrationCloningTest.java | 127 ++++++++++++++ .../PROTO_SOURCE_COMMIT_HASH | 2 +- internal/durabletask-protobuf/README.md | 2 +- .../protos/orchestrator_service.proto | 14 ++ 10 files changed, 372 insertions(+), 3 deletions(-) create mode 100644 client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java create mode 100644 client/src/test/java/com/microsoft/durabletask/TaskOrchestrationCloningTest.java diff --git a/client/build.gradle b/client/build.gradle index 90d3a2e8..5ee84298 100644 --- a/client/build.gradle +++ b/client/build.gradle @@ -75,7 +75,8 @@ compileTestJava { } task downloadProtoFiles { - ext.branch = project.hasProperty('protoBranch') ? project.protoBranch : 'main' + // Includes the explicit restart target/version contract from durabletask-protobuf PR #82. + ext.branch = project.hasProperty('protoBranch') ? project.protoBranch : '962a493037e72b5e08f07a55eff0f121c1aaae80' doLast { def protoDir = file("${rootProject.projectDir}/internal/durabletask-protobuf/protos") @@ -89,6 +90,10 @@ task downloadProtoFiles { // Get and save the commit hash def commitHashFile = new File("${rootProject.projectDir}/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH") + if (ext.branch ==~ /[0-9a-fA-F]{40}/) { + commitHashFile.text = ext.branch + return + } def commitApiUrl = new URL("https://api.github.com/repos/microsoft/durabletask-protobuf/commits?path=protos/orchestrator_service.proto&sha=${ext.branch}&per_page=1") def connection = commitApiUrl.openConnection() connection.setRequestProperty('Accept', 'application/vnd.github.v3+json') diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java index 3864e487..42e9f0b2 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java @@ -330,6 +330,35 @@ public List getOrchestrationHistory(String instanceId) { */ public abstract String restartInstance(String instanceId, boolean restartWithNewInstanceId); + /** + * Restarts an orchestration from the beginning with its original name and input and a required target ID. + * Execution history is not copied. Requires a backend that supports explicit restart targets. + * + * @param instanceId the ID of the source orchestration + * @param newInstanceId the non-empty target instance ID + * @return the ID of the restarted orchestration + */ + public String restartInstance(String instanceId, String newInstanceId) { + return this.restartInstance(instanceId, newInstanceId, null); + } + + /** + * Restarts an orchestration from the beginning with its original name and input and a required target ID. + * Execution history is not copied. When the target differs from the source, the clone exposes the immediate + * source via {@link TaskOrchestrationContext#getSourceInstanceId()} if the backend persists orchestration tags. + * Sub-orchestrations do not inherit this reference. Existing-instance conflicts are handled by the backend. + * Requires a backend that supports explicit restart targets and versions. + * + * @param instanceId the ID of the source orchestration + * @param newInstanceId the non-empty target instance ID + * @param version the version override; {@code null} preserves the backend's default restart behavior, + * and an empty string selects an unversioned orchestration + * @return the ID of the restarted orchestration + */ + public String restartInstance(String instanceId, String newInstanceId, @Nullable String version) { + throw new UnsupportedOperationException("Explicit restart targets are not supported by this client."); + } + /** * Rewinds a failed orchestration instance to the last known good state and replays from there. *

diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java index 8a107aa7..8ac8903b 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java @@ -455,6 +455,21 @@ public String restartInstance(String instanceId, boolean restartWithNewInstanceI } } + @Override + public String restartInstance(String instanceId, String newInstanceId, @Nullable String version) { + Helpers.throwIfArgumentNullOrWhiteSpace(instanceId, "instanceId"); + Helpers.throwIfArgumentNullOrWhiteSpace(newInstanceId, "newInstanceId"); + + RestartInstanceRequest.Builder request = RestartInstanceRequest.newBuilder() + .setInstanceId(instanceId) + .setRestartWithNewInstanceId(true) + .setNewInstanceId(StringValue.of(newInstanceId)); + if (version != null) { + request.setVersion(StringValue.of(version)); + } + return this.sidecarClient.restartInstance(request.build()).getInstanceId(); + } + private PurgeResult toPurgeResult(PurgeInstancesResponse response){ return new PurgeResult(response.getDeletedInstanceCount()); } diff --git a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java index 71f9c9d2..c30bbee1 100644 --- a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java +++ b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java @@ -93,6 +93,17 @@ default Logger createReplaySafeLogger(Logger logger) { */ String getVersion(); + /** + * Gets the immediate source instance ID when this orchestration was created as a clone. + * This reference is available during replay and is not inherited by sub-orchestrations. + * + * @return the source instance ID, or {@code null} if this is not a clone or the backend does not persist it + */ + @Nullable + default String getSourceInstanceId() { + return null; + } + /** * Returns a new {@code Task} that is completed when all tasks in {@code tasks} completes. * See {@link #allOf(Task[])} for more detailed information. diff --git a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java index 5bdbad2e..f0ddba88 100644 --- a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java +++ b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java @@ -122,6 +122,7 @@ private class ContextImplTask implements TaskOrchestrationContext { private String orchestratorName; private String rawInput; private String instanceId; + private String sourceInstanceId; private Instant currentInstant; private boolean isComplete; private boolean isSuspended; @@ -208,6 +209,12 @@ public String getInstanceId() { return this.instanceId; } + @Override + @Nullable + public String getSourceInstanceId() { + return this.sourceInstanceId; + } + @Override @Nullable public ParentOrchestrationInstance getParentInstance() { @@ -1707,6 +1714,9 @@ private void processEvent(HistoryEvent e) { this.setInput(input); String version = startedEvent.getVersion().getValue(); this.setVersion(version); + String sourceInstanceId = startedEvent.getTagsMap().get("MS_DurableFunctions_SourceInstanceId"); + this.sourceInstanceId = !startedEvent.hasParentInstance() && !Helpers.isNullOrEmpty(sourceInstanceId) + ? sourceInstanceId : null; if (startedEvent.hasParentTraceContext()) { this.parentTraceContext = startedEvent.getParentTraceContext(); } else { diff --git a/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java b/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java new file mode 100644 index 00000000..e773327f --- /dev/null +++ b/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java @@ -0,0 +1,158 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask; + +import com.google.protobuf.StringValue; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*; +import com.microsoft.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; + +import static org.junit.jupiter.api.Assertions.*; + +class DurableTaskGrpcClientRestartTest { + private Server server; + private ManagedChannel channel; + private RestartInstanceRequest restartRequest; + private CreateInstanceRequest createRequest; + private int metadataRequests; + private Status restartError; + + private DurableTaskClient startClient() throws Exception { + String name = InProcessServerBuilder.generateName(); + this.server = InProcessServerBuilder.forName(name).directExecutor() + .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void restartInstance(RestartInstanceRequest request, + StreamObserver observer) { + restartRequest = request; + if (restartError != null) { + observer.onError(restartError.asRuntimeException()); + return; + } + observer.onNext(RestartInstanceResponse.newBuilder() + .setInstanceId(request.getNewInstanceId().getValue()).build()); + observer.onCompleted(); + } + + @Override + public void getInstance(GetInstanceRequest request, StreamObserver observer) { + metadataRequests++; + observer.onNext(GetInstanceResponse.newBuilder().setExists(true) + .setOrchestrationState(OrchestrationState.newBuilder() + .setInstanceId(request.getInstanceId()) + .setName("Order") + .setInput(StringValue.of("{\"order\":123}"))).build()); + observer.onCompleted(); + } + + @Override + public void startInstance(CreateInstanceRequest request, + StreamObserver observer) { + createRequest = request; + observer.onNext(CreateInstanceResponse.newBuilder() + .setInstanceId(request.getInstanceId()).build()); + observer.onCompleted(); + } + }).build().start(); + this.channel = InProcessChannelBuilder.forName(name).directExecutor().build(); + return new DurableTaskGrpcClientBuilder().grpcChannel(this.channel).defaultVersion("client-default").build(); + } + + @AfterEach + void tearDown() { + if (this.channel != null) { + this.channel.shutdownNow(); + } + if (this.server != null) { + this.server.shutdownNow(); + } + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {"2.0"}) + void explicitTargetPreservesVersionPresence(String version) throws Exception { + DurableTaskClient client = startClient(); + assertEquals("order-123-retry", client.restartInstance("order-123", "order-123-retry", version)); + assertEquals("order-123", this.restartRequest.getInstanceId()); + assertTrue(this.restartRequest.getRestartWithNewInstanceId()); + assertTrue(this.restartRequest.hasNewInstanceId()); + assertEquals("order-123-retry", this.restartRequest.getNewInstanceId().getValue()); + assertEquals(version != null, this.restartRequest.hasVersion()); + if (version != null) { + assertEquals(version, this.restartRequest.getVersion().getValue()); + } + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); + } + + @Test + void omittedVersionAndChainedCloneUseBackendRestart() throws Exception { + DurableTaskClient client = startClient(); + assertEquals("order-123-retry", client.restartInstance("order-123", "order-123-retry")); + assertFalse(this.restartRequest.hasVersion()); + assertEquals("order-123-retry-again", + client.restartInstance("order-123-retry", "order-123-retry-again")); + assertEquals("order-123-retry", this.restartRequest.getInstanceId()); + assertFalse(this.restartRequest.hasVersion()); + } + + @Test + void explicitTargetMayBeTheSourceId() throws Exception { + DurableTaskClient client = startClient(); + assertEquals("order-123", client.restartInstance("order-123", "order-123", "")); + assertEquals("order-123", this.restartRequest.getNewInstanceId().getValue()); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {" ", "\t"}) + void invalidSourceOrTargetIsRejectedBeforeRpc(String id) throws Exception { + DurableTaskClient client = startClient(); + assertThrows(IllegalArgumentException.class, () -> client.restartInstance("source", id)); + assertThrows(IllegalArgumentException.class, () -> client.restartInstance(id, "target", "2.0")); + assertNull(this.restartRequest); + assertEquals(0, this.metadataRequests); + } + + @ParameterizedTest + @ValueSource(strings = {"NOT_FOUND", "ALREADY_EXISTS", "UNIMPLEMENTED"}) + void backendErrorsAreNotReplacedWithSchedulingFallback(String status) throws Exception { + DurableTaskClient client = startClient(); + this.restartError = Status.fromCode(Status.Code.valueOf(status)); + StatusRuntimeException error = assertThrows(StatusRuntimeException.class, + () -> client.restartInstance("source", "target")); + assertEquals(this.restartError.getCode(), error.getStatus().getCode()); + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void legacyRestartStillSchedulesOriginalNameAndInput(boolean newId) throws Exception { + DurableTaskClient client = startClient(); + String result = client.restartInstance("order-123", newId); + assertEquals(this.createRequest.getInstanceId(), result); + if (newId) { + assertNotEquals("order-123", result); + } else { + assertEquals("order-123", result); + } + assertEquals("Order", this.createRequest.getName()); + assertEquals("{\"order\":123}", this.createRequest.getInput().getValue()); + assertEquals("client-default", this.createRequest.getVersion().getValue()); + assertEquals(1, this.metadataRequests); + assertNull(this.restartRequest); + } +} diff --git a/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationCloningTest.java b/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationCloningTest.java new file mode 100644 index 00000000..929eee1b --- /dev/null +++ b/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationCloningTest.java @@ -0,0 +1,127 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask; + +import com.google.protobuf.StringValue; +import com.google.protobuf.Timestamp; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; +import java.util.*; +import java.util.logging.Logger; + +import static org.junit.jupiter.api.Assertions.*; + +class TaskOrchestrationCloningTest { + private static final String SOURCE_TAG = "MS_DurableFunctions_SourceInstanceId"; + + private TaskOrchestrationExecutor executor(TaskOrchestration orchestration) { + HashMap factories = new HashMap<>(); + factories.put("Order", new TaskOrchestrationFactory() { + @Override + public String getName() { + return "Order"; + } + + @Override + public TaskOrchestration create() { + return orchestration; + } + }); + return new TaskOrchestrationExecutor(factories, new JacksonDataConverter(), Duration.ofDays(3), + Logger.getLogger(TaskOrchestrationCloningTest.class.getName()), null); + } + + private List startEvents(String sourceId, boolean child) { + ExecutionStartedEvent.Builder started = ExecutionStartedEvent.newBuilder() + .setName("Order") + .setInput(StringValue.of("\"original-input\"")) + .setVersion(StringValue.of("2.0")) + .setOrchestrationInstance(OrchestrationInstance.newBuilder().setInstanceId("order-retry")); + if (sourceId != null) { + started.putTags(SOURCE_TAG, sourceId); + } + if (child) { + started.setParentInstance(ParentInstanceInfo.newBuilder() + .setOrchestrationInstance(OrchestrationInstance.newBuilder().setInstanceId("parent-clone"))); + } + return Arrays.asList( + HistoryEvent.newBuilder().setEventId(-1).setTimestamp(Timestamp.getDefaultInstance()) + .setOrchestratorStarted(OrchestratorStartedEvent.getDefaultInstance()).build(), + HistoryEvent.newBuilder().setEventId(-1).setTimestamp(Timestamp.getDefaultInstance()) + .setExecutionStarted(started).build()); + } + + @ParameterizedTest + @ValueSource(strings = {"order-123", "order-123-retry"}) + void immediateSourceIsStableAcrossReplayAndChildCompletion(String sourceId) { + List sources = new ArrayList<>(); + List replayStates = new ArrayList<>(); + List details = new ArrayList<>(); + TaskOrchestrationExecutor executor = executor(ctx -> { + sources.add(ctx.getSourceInstanceId()); + replayStates.add(ctx.getIsReplaying()); + details.add(ctx.getInstanceId()); + details.add(ctx.getVersion()); + details.add(ctx.getInput(String.class)); + ctx.callSubOrchestrator("Child", null, String.class).await(); + sources.add(ctx.getSourceInstanceId()); + replayStates.add(ctx.getIsReplaying()); + ctx.complete("done"); + }); + List history = new ArrayList<>(startEvents(sourceId, false)); + TaskOrchestratorResult initial = executor.execute(Collections.emptyList(), history, null); + OrchestratorAction action = initial.getActions().iterator().next(); + assertTrue(action.hasCreateSubOrchestration()); + assertFalse(action.getCreateSubOrchestration().getTagsMap().containsKey(SOURCE_TAG)); + assertEquals(Collections.singletonList(sourceId), sources); + assertEquals(Collections.singletonList(false), replayStates); + + history.add(HistoryEvent.newBuilder().setEventId(action.getId()) + .setSubOrchestrationInstanceCreated(SubOrchestrationInstanceCreatedEvent.newBuilder() + .setName("Child") + .setInstanceId(action.getCreateSubOrchestration().getInstanceId())).build()); + List newEvents = Collections.singletonList(HistoryEvent.newBuilder().setEventId(-1) + .setSubOrchestrationInstanceCompleted(SubOrchestrationInstanceCompletedEvent.newBuilder() + .setTaskScheduledId(action.getId()).setResult(StringValue.of("\"child-result\""))).build()); + TaskOrchestratorResult replay = executor.execute(history, newEvents, null); + assertEquals(Arrays.asList(sourceId, sourceId, sourceId), sources); + assertEquals(Arrays.asList(false, true, false), replayStates); + assertEquals(Arrays.asList("order-retry", "2.0", "original-input", + "order-retry", "2.0", "original-input"), details); + CompleteOrchestrationAction completion = replay.getActions().iterator().next().getCompleteOrchestration(); + assertEquals(OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, completion.getOrchestrationStatus()); + assertEquals("\"done\"", completion.getResult().getValue()); + } + + @ParameterizedTest + @NullAndEmptySource + void ordinaryOrLegacyHistoryHasNoSourceReference(String sourceId) { + List sources = new ArrayList<>(); + TaskOrchestrationExecutor executor = executor(ctx -> { + sources.add(ctx.getSourceInstanceId()); + ctx.complete(null); + }); + List events = startEvents(sourceId, false); + executor.execute(Collections.emptyList(), events, null); + executor.execute(events, Collections.emptyList(), null); + assertEquals(Arrays.asList(null, null), sources); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void childDoesNotExposeInheritedSourceTagDuringLiveExecutionOrReplay(boolean replay) { + List sources = new ArrayList<>(); + TaskOrchestrationExecutor executor = executor(ctx -> { + sources.add(ctx.getSourceInstanceId()); + ctx.complete(null); + }); + List events = startEvents("original-source", true); + executor.execute(replay ? events : Collections.emptyList(), + replay ? Collections.emptyList() : events, null); + assertEquals(Collections.singletonList(null), sources); + } +} diff --git a/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH b/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH index a981c689..538d9ccb 100644 --- a/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH +++ b/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH @@ -1 +1 @@ -3145f9337fca9de57d2f89a6ff6f07150d34f1c2 \ No newline at end of file +962a493037e72b5e08f07a55eff0f121c1aaae80 \ No newline at end of file diff --git a/internal/durabletask-protobuf/README.md b/internal/durabletask-protobuf/README.md index 6b310469..6cd08a42 100644 --- a/internal/durabletask-protobuf/README.md +++ b/internal/durabletask-protobuf/README.md @@ -11,7 +11,7 @@ This directory contains the protocol buffer definitions used by the Durable Task The proto files are automatically downloaded and updated when running Gradle builds. This is handled by the `downloadProtoFiles` task in the `client/build.gradle` file. The task: -1. Downloads the latest version of `orchestrator_service.proto` +1. Downloads the pinned version of `orchestrator_service.proto` specified in `client/build.gradle` (or the `protoBranch` override) 2. Saves the current commit hash for tracking purposes 3. Updates these files before proto compilation begins diff --git a/internal/durabletask-protobuf/protos/orchestrator_service.proto b/internal/durabletask-protobuf/protos/orchestrator_service.proto index 3d9194ac..b9f9f63d 100644 --- a/internal/durabletask-protobuf/protos/orchestrator_service.proto +++ b/internal/durabletask-protobuf/protos/orchestrator_service.proto @@ -533,9 +533,22 @@ message PurgeInstancesResponse { google.protobuf.BoolValue isComplete = 2; } +// Restarts an orchestration from the beginning with its original name and input, +// without copying its history. message RestartInstanceRequest { string instanceId = 1; + + // When newInstanceId is absent, true generates a new instance ID and false + // restarts using instanceId. bool restartWithNewInstanceId = 2; + + // An explicit target instance ID. When present, takes precedence over + // restartWithNewInstanceId. + google.protobuf.StringValue newInstanceId = 3; + + // Overrides the restarted orchestration's version, including an empty string. + // When absent, preserves the server's default restart behavior. + google.protobuf.StringValue version = 4; } message RestartInstanceResponse { @@ -706,6 +719,7 @@ message StartNewOrchestrationAction { google.protobuf.Timestamp scheduledTime = 5; google.protobuf.Timestamp requestTime = 6; TraceContext parentTraceContext = 7; + map tags = 8; } message AbandonActivityTaskRequest { From ba61c35c497ce4aa4991d1acb4173f33179e7ee4 Mon Sep 17 00:00:00 2001 From: Varshitha Bachu Date: Thu, 5 Nov 2026 17:02:56 -0800 Subject: [PATCH 3/5] Potential fix for pull request finding 'Possible confusion of local and field' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> --- .../microsoft/durabletask/TaskOrchestrationExecutor.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java index f0ddba88..c390d7c6 100644 --- a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java +++ b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java @@ -1714,9 +1714,9 @@ private void processEvent(HistoryEvent e) { this.setInput(input); String version = startedEvent.getVersion().getValue(); this.setVersion(version); - String sourceInstanceId = startedEvent.getTagsMap().get("MS_DurableFunctions_SourceInstanceId"); - this.sourceInstanceId = !startedEvent.hasParentInstance() && !Helpers.isNullOrEmpty(sourceInstanceId) - ? sourceInstanceId : null; + String tagSourceInstanceId = startedEvent.getTagsMap().get("MS_DurableFunctions_SourceInstanceId"); + this.sourceInstanceId = !startedEvent.hasParentInstance() && !Helpers.isNullOrEmpty(tagSourceInstanceId) + ? tagSourceInstanceId : null; if (startedEvent.hasParentTraceContext()) { this.parentTraceContext = startedEvent.getParentTraceContext(); } else { From 27332f2cd9b258bde2b5cfa1dcb2379289d3a9d1 Mon Sep 17 00:00:00 2001 From: Varshi Bachu Date: Thu, 5 Nov 2026 17:14:38 -0800 Subject: [PATCH 4/5] add restartinstanceoptions --- CHANGELOG.md | 1 + .../durabletask/DurableTaskClient.java | 22 ++----- .../durabletask/DurableTaskGrpcClient.java | 5 +- .../durabletask/RestartInstanceOptions.java | 62 +++++++++++++++++++ .../DurableTaskGrpcClientRestartTest.java | 31 +++++++--- 5 files changed, 97 insertions(+), 24 deletions(-) create mode 100644 client/src/main/java/com/microsoft/durabletask/RestartInstanceOptions.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f93f23c..5b1d038a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ ## Unreleased +* Add orchestration cloning via `restartInstance(instanceId, RestartInstanceOptions)` with a required target instance ID and optional version override, preserving the existing boolean overload. Clones expose their immediate source through `getSourceInstanceId()`. Requires a backend supporting explicit restart targets and versions. * Write self-describing v2 blob payload tokens by default while retaining v1 and v2 reads from the configured storage endpoint and container. Require complete token formats for direct downloads and account for URL token length in worker action-size estimates. Upgrade readers before deploying v2 writers; changing storage locations requires explicit migration or a store configured for the original location. * Add explicit primary-account-wide and additional-account read scopes for Azure Blob payloads while retaining configured-container access by default. Additional accounts require identity authentication and exact HTTPS account service endpoints. Preserve named and nested v1/v2 references and ignore legacy v2 URL query and fragment suffixes. diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java index 42e9f0b2..c2cad562 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java @@ -330,18 +330,6 @@ public List getOrchestrationHistory(String instanceId) { */ public abstract String restartInstance(String instanceId, boolean restartWithNewInstanceId); - /** - * Restarts an orchestration from the beginning with its original name and input and a required target ID. - * Execution history is not copied. Requires a backend that supports explicit restart targets. - * - * @param instanceId the ID of the source orchestration - * @param newInstanceId the non-empty target instance ID - * @return the ID of the restarted orchestration - */ - public String restartInstance(String instanceId, String newInstanceId) { - return this.restartInstance(instanceId, newInstanceId, null); - } - /** * Restarts an orchestration from the beginning with its original name and input and a required target ID. * Execution history is not copied. When the target differs from the source, the clone exposes the immediate @@ -350,12 +338,14 @@ public String restartInstance(String instanceId, String newInstanceId) { * Requires a backend that supports explicit restart targets and versions. * * @param instanceId the ID of the source orchestration - * @param newInstanceId the non-empty target instance ID - * @param version the version override; {@code null} preserves the backend's default restart behavior, - * and an empty string selects an unversioned orchestration + * @param options the restart options, including a required nonblank new instance ID and an optional version; + * a {@code null} version preserves the backend's default restart behavior, and an empty + * version selects an unversioned orchestration + * @throws IllegalArgumentException if the source ID is blank, the options are {@code null}, + * or the target ID is missing or blank * @return the ID of the restarted orchestration */ - public String restartInstance(String instanceId, String newInstanceId, @Nullable String version) { + public String restartInstance(String instanceId, RestartInstanceOptions options) { throw new UnsupportedOperationException("Explicit restart targets are not supported by this client."); } diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java index 8ac8903b..caee110d 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java @@ -456,9 +456,12 @@ public String restartInstance(String instanceId, boolean restartWithNewInstanceI } @Override - public String restartInstance(String instanceId, String newInstanceId, @Nullable String version) { + public String restartInstance(String instanceId, RestartInstanceOptions options) { Helpers.throwIfArgumentNullOrWhiteSpace(instanceId, "instanceId"); + Helpers.throwIfArgumentNull(options, "options"); + String newInstanceId = options.getNewInstanceId(); Helpers.throwIfArgumentNullOrWhiteSpace(newInstanceId, "newInstanceId"); + String version = options.getVersion(); RestartInstanceRequest.Builder request = RestartInstanceRequest.newBuilder() .setInstanceId(instanceId) diff --git a/client/src/main/java/com/microsoft/durabletask/RestartInstanceOptions.java b/client/src/main/java/com/microsoft/durabletask/RestartInstanceOptions.java new file mode 100644 index 00000000..845bea9c --- /dev/null +++ b/client/src/main/java/com/microsoft/durabletask/RestartInstanceOptions.java @@ -0,0 +1,62 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask; + +import javax.annotation.Nullable; + +/** + * Options for restarting an orchestration with a required target instance ID and an optional version. + */ +public final class RestartInstanceOptions { + private String newInstanceId; + private String version; + + /** + * Creates a new {@link RestartInstanceOptions}. A target instance ID must be set before restarting. + */ + public RestartInstanceOptions() { + } + + /** + * Sets the required target instance ID. + * + * @param newInstanceId the nonblank ID of the restarted orchestration + * @return this {@link RestartInstanceOptions} object + */ + public RestartInstanceOptions setNewInstanceId(String newInstanceId) { + this.newInstanceId = newInstanceId; + return this; + } + + /** + * Gets the target instance ID. + * + * @return the target instance ID, or {@code null} if none has been set + */ + @Nullable + public String getNewInstanceId() { + return this.newInstanceId; + } + + /** + * Sets the version override. A {@code null} value preserves the backend's default restart behavior; + * an empty string explicitly selects an unversioned orchestration. + * + * @param version the version override + * @return this {@link RestartInstanceOptions} object + */ + public RestartInstanceOptions setVersion(@Nullable String version) { + this.version = version; + return this; + } + + /** + * Gets the version override. + * + * @return the version override, or {@code null} to preserve the backend's default restart behavior + */ + @Nullable + public String getVersion() { + return this.version; + } +} diff --git a/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java b/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java index e773327f..7cc8adf9 100644 --- a/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java +++ b/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java @@ -84,7 +84,8 @@ void tearDown() { @ValueSource(strings = {"2.0"}) void explicitTargetPreservesVersionPresence(String version) throws Exception { DurableTaskClient client = startClient(); - assertEquals("order-123-retry", client.restartInstance("order-123", "order-123-retry", version)); + assertEquals("order-123-retry", client.restartInstance("order-123", + new RestartInstanceOptions().setNewInstanceId("order-123-retry").setVersion(version))); assertEquals("order-123", this.restartRequest.getInstanceId()); assertTrue(this.restartRequest.getRestartWithNewInstanceId()); assertTrue(this.restartRequest.hasNewInstanceId()); @@ -100,10 +101,12 @@ void explicitTargetPreservesVersionPresence(String version) throws Exception { @Test void omittedVersionAndChainedCloneUseBackendRestart() throws Exception { DurableTaskClient client = startClient(); - assertEquals("order-123-retry", client.restartInstance("order-123", "order-123-retry")); + assertEquals("order-123-retry", client.restartInstance("order-123", + new RestartInstanceOptions().setNewInstanceId("order-123-retry"))); assertFalse(this.restartRequest.hasVersion()); assertEquals("order-123-retry-again", - client.restartInstance("order-123-retry", "order-123-retry-again")); + client.restartInstance("order-123-retry", + new RestartInstanceOptions().setNewInstanceId("order-123-retry-again"))); assertEquals("order-123-retry", this.restartRequest.getInstanceId()); assertFalse(this.restartRequest.hasVersion()); } @@ -111,7 +114,8 @@ void omittedVersionAndChainedCloneUseBackendRestart() throws Exception { @Test void explicitTargetMayBeTheSourceId() throws Exception { DurableTaskClient client = startClient(); - assertEquals("order-123", client.restartInstance("order-123", "order-123", "")); + assertEquals("order-123", client.restartInstance("order-123", + new RestartInstanceOptions().setNewInstanceId("order-123").setVersion(""))); assertEquals("order-123", this.restartRequest.getNewInstanceId().getValue()); } @@ -120,19 +124,32 @@ void explicitTargetMayBeTheSourceId() throws Exception { @ValueSource(strings = {" ", "\t"}) void invalidSourceOrTargetIsRejectedBeforeRpc(String id) throws Exception { DurableTaskClient client = startClient(); - assertThrows(IllegalArgumentException.class, () -> client.restartInstance("source", id)); - assertThrows(IllegalArgumentException.class, () -> client.restartInstance(id, "target", "2.0")); + assertThrows(IllegalArgumentException.class, () -> client.restartInstance("source", + new RestartInstanceOptions().setNewInstanceId(id))); + assertThrows(IllegalArgumentException.class, () -> client.restartInstance(id, + new RestartInstanceOptions().setNewInstanceId("target").setVersion("2.0"))); assertNull(this.restartRequest); assertEquals(0, this.metadataRequests); } + @Test + void missingOptionsOrTargetIsRejectedBeforeRpc() throws Exception { + DurableTaskClient client = startClient(); + assertThrows(IllegalArgumentException.class, () -> client.restartInstance("source", null)); + assertThrows(IllegalArgumentException.class, + () -> client.restartInstance("source", new RestartInstanceOptions())); + assertNull(this.restartRequest); + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); + } + @ParameterizedTest @ValueSource(strings = {"NOT_FOUND", "ALREADY_EXISTS", "UNIMPLEMENTED"}) void backendErrorsAreNotReplacedWithSchedulingFallback(String status) throws Exception { DurableTaskClient client = startClient(); this.restartError = Status.fromCode(Status.Code.valueOf(status)); StatusRuntimeException error = assertThrows(StatusRuntimeException.class, - () -> client.restartInstance("source", "target")); + () -> client.restartInstance("source", new RestartInstanceOptions().setNewInstanceId("target"))); assertEquals(this.restartError.getCode(), error.getStatus().getCode()); assertEquals(0, this.metadataRequests); assertNull(this.createRequest); From 76440d8783f7e71a9ee51cd2a9b4e9b05117c09c Mon Sep 17 00:00:00 2001 From: Varshi Bachu Date: Tue, 10 Nov 2026 20:01:25 -0800 Subject: [PATCH 5/5] Update cloning to use the new CloneInstance RPC --- CHANGELOG.md | 2 +- client/build.gradle | 4 +- .../durabletask/DurableTaskClient.java | 11 +- .../durabletask/DurableTaskGrpcClient.java | 9 +- .../durabletask/TaskOrchestrationContext.java | 5 +- .../DurableTaskGrpcClientRestartTest.java | 102 +++++++++++++----- .../PROTO_SOURCE_COMMIT_HASH | 2 +- internal/durabletask-protobuf/README.md | 5 + .../protos/orchestrator_service.proto | 48 ++++++--- 9 files changed, 137 insertions(+), 51 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b1d038a..0bcab128 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ ## Unreleased -* Add orchestration cloning via `restartInstance(instanceId, RestartInstanceOptions)` with a required target instance ID and optional version override, preserving the existing boolean overload. Clones expose their immediate source through `getSourceInstanceId()`. Requires a backend supporting explicit restart targets and versions. +* Add orchestration cloning via `restartInstance(instanceId, RestartInstanceOptions)` with a required target instance ID and optional version override, preserving the existing boolean overload. Uses the dedicated `CloneInstance` RPC without falling back to legacy restart or scheduling APIs on unsupported backends. `getSourceInstanceId()` exposes the immediate source only when the backend supplies it in orchestration tags; the cloning protocol keeps lineage internal. * Write self-describing v2 blob payload tokens by default while retaining v1 and v2 reads from the configured storage endpoint and container. Require complete token formats for direct downloads and account for URL token length in worker action-size estimates. Upgrade readers before deploying v2 writers; changing storage locations requires explicit migration or a store configured for the original location. * Add explicit primary-account-wide and additional-account read scopes for Azure Blob payloads while retaining configured-container access by default. Additional accounts require identity authentication and exact HTTPS account service endpoints. Preserve named and nested v1/v2 references and ignore legacy v2 URL query and fragment suffixes. diff --git a/client/build.gradle b/client/build.gradle index 5ee84298..de314cf1 100644 --- a/client/build.gradle +++ b/client/build.gradle @@ -75,8 +75,8 @@ compileTestJava { } task downloadProtoFiles { - // Includes the explicit restart target/version contract from durabletask-protobuf PR #82. - ext.branch = project.hasProperty('protoBranch') ? project.protoBranch : '962a493037e72b5e08f07a55eff0f121c1aaae80' + // Includes the dedicated CloneInstance RPC from durabletask-protobuf PR #82. + ext.branch = project.hasProperty('protoBranch') ? project.protoBranch : '0a4f8075e383f6ac783acfd5aee9ecb791334e87' doLast { def protoDir = file("${rootProject.projectDir}/internal/durabletask-protobuf/protos") diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java index c2cad562..41080fe0 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java @@ -332,10 +332,13 @@ public List getOrchestrationHistory(String instanceId) { /** * Restarts an orchestration from the beginning with its original name and input and a required target ID. - * Execution history is not copied. When the target differs from the source, the clone exposes the immediate - * source via {@link TaskOrchestrationContext#getSourceInstanceId()} if the backend persists orchestration tags. - * Sub-orchestrations do not inherit this reference. Existing-instance conflicts are handled by the backend. - * Requires a backend that supports explicit restart targets and versions. + * Execution history is not copied. The target may be the source instance ID; existing-instance conflicts + * are handled by the backend. Source lineage is stored internally by the backend and is not exposed by the + * cloning protocol. {@link TaskOrchestrationContext#getSourceInstanceId()} is available only if the backend + * also supplies the source reference in orchestration tags. + * Requires a backend implementing the {@code CloneInstance} RPC and the requested version behavior. + * Unsupported backends reject the request without creating an orchestration. This method does not fall + * back to the legacy restart or scheduling APIs. * * @param instanceId the ID of the source orchestration * @param options the restart options, including a required nonblank new instance ID and an optional version; diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java index caee110d..0d19248e 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java @@ -463,14 +463,13 @@ public String restartInstance(String instanceId, RestartInstanceOptions options) Helpers.throwIfArgumentNullOrWhiteSpace(newInstanceId, "newInstanceId"); String version = options.getVersion(); - RestartInstanceRequest.Builder request = RestartInstanceRequest.newBuilder() - .setInstanceId(instanceId) - .setRestartWithNewInstanceId(true) - .setNewInstanceId(StringValue.of(newInstanceId)); + CloneInstanceRequest.Builder request = CloneInstanceRequest.newBuilder() + .setSourceInstanceId(instanceId) + .setNewInstanceId(newInstanceId); if (version != null) { request.setVersion(StringValue.of(version)); } - return this.sidecarClient.restartInstance(request.build()).getInstanceId(); + return this.sidecarClient.cloneInstance(request.build()).getInstanceId(); } private PurgeResult toPurgeResult(PurgeInstancesResponse response){ diff --git a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java index c30bbee1..2bd09848 100644 --- a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java +++ b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java @@ -95,9 +95,10 @@ default Logger createReplaySafeLogger(Logger logger) { /** * Gets the immediate source instance ID when this orchestration was created as a clone. - * This reference is available during replay and is not inherited by sub-orchestrations. + * The backend must supply this reference in orchestration tags; the cloning protocol does not expose it. + * When supplied, it is available during replay and is not inherited by sub-orchestrations. * - * @return the source instance ID, or {@code null} if this is not a clone or the backend does not persist it + * @return the source instance ID, or {@code null} if this is not a clone or the backend does not expose it in tags */ @Nullable default String getSourceInstanceId() { diff --git a/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java b/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java index 7cc8adf9..7324bab9 100644 --- a/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java +++ b/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java @@ -7,6 +7,8 @@ import com.microsoft.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.Server; +import io.grpc.ServerMethodDefinition; +import io.grpc.ServerServiceDefinition; import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.grpc.inprocess.InProcessChannelBuilder; @@ -23,25 +25,39 @@ class DurableTaskGrpcClientRestartTest { private Server server; private ManagedChannel channel; + private CloneInstanceRequest cloneRequest; private RestartInstanceRequest restartRequest; private CreateInstanceRequest createRequest; private int metadataRequests; - private Status restartError; + private Status cloneError; private DurableTaskClient startClient() throws Exception { + return startClient(true); + } + + private DurableTaskClient startClient(boolean supportsCloning) throws Exception { String name = InProcessServerBuilder.generateName(); - this.server = InProcessServerBuilder.forName(name).directExecutor() - .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase service = + new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void cloneInstance(CloneInstanceRequest request, + StreamObserver observer) { + cloneRequest = request; + if (cloneError != null) { + observer.onError(cloneError.asRuntimeException()); + return; + } + observer.onNext(CloneInstanceResponse.newBuilder() + .setInstanceId(request.getNewInstanceId()).build()); + observer.onCompleted(); + } + @Override public void restartInstance(RestartInstanceRequest request, StreamObserver observer) { restartRequest = request; - if (restartError != null) { - observer.onError(restartError.asRuntimeException()); - return; - } observer.onNext(RestartInstanceResponse.newBuilder() - .setInstanceId(request.getNewInstanceId().getValue()).build()); + .setInstanceId("legacy-restart").build()); observer.onCompleted(); } @@ -64,7 +80,17 @@ public void startInstance(CreateInstanceRequest request, .setInstanceId(request.getInstanceId()).build()); observer.onCompleted(); } - }).build().start(); + }; + ServerServiceDefinition.Builder serviceDefinition = + ServerServiceDefinition.builder(TaskHubSidecarServiceGrpc.SERVICE_NAME); + for (ServerMethodDefinition method : service.bindService().getMethods()) { + if (supportsCloning || !method.getMethodDescriptor().getFullMethodName() + .equals(TaskHubSidecarServiceGrpc.getCloneInstanceMethod().getFullMethodName())) { + serviceDefinition.addMethod(method); + } + } + this.server = InProcessServerBuilder.forName(name).directExecutor() + .addService(serviceDefinition.build()).build().start(); this.channel = InProcessChannelBuilder.forName(name).directExecutor().build(); return new DurableTaskGrpcClientBuilder().grpcChannel(this.channel).defaultVersion("client-default").build(); } @@ -86,29 +112,32 @@ void explicitTargetPreservesVersionPresence(String version) throws Exception { DurableTaskClient client = startClient(); assertEquals("order-123-retry", client.restartInstance("order-123", new RestartInstanceOptions().setNewInstanceId("order-123-retry").setVersion(version))); - assertEquals("order-123", this.restartRequest.getInstanceId()); - assertTrue(this.restartRequest.getRestartWithNewInstanceId()); - assertTrue(this.restartRequest.hasNewInstanceId()); - assertEquals("order-123-retry", this.restartRequest.getNewInstanceId().getValue()); - assertEquals(version != null, this.restartRequest.hasVersion()); + assertEquals("order-123", this.cloneRequest.getSourceInstanceId()); + assertEquals("order-123-retry", this.cloneRequest.getNewInstanceId()); + assertEquals(version != null, this.cloneRequest.hasVersion()); if (version != null) { - assertEquals(version, this.restartRequest.getVersion().getValue()); + assertEquals(version, this.cloneRequest.getVersion().getValue()); } + assertNull(this.restartRequest); assertEquals(0, this.metadataRequests); assertNull(this.createRequest); } @Test - void omittedVersionAndChainedCloneUseBackendRestart() throws Exception { + void omittedVersionAndChainedCloneUseCloneRpc() throws Exception { DurableTaskClient client = startClient(); assertEquals("order-123-retry", client.restartInstance("order-123", new RestartInstanceOptions().setNewInstanceId("order-123-retry"))); - assertFalse(this.restartRequest.hasVersion()); + assertFalse(this.cloneRequest.hasVersion()); assertEquals("order-123-retry-again", client.restartInstance("order-123-retry", new RestartInstanceOptions().setNewInstanceId("order-123-retry-again"))); - assertEquals("order-123-retry", this.restartRequest.getInstanceId()); - assertFalse(this.restartRequest.hasVersion()); + assertEquals("order-123-retry", this.cloneRequest.getSourceInstanceId()); + assertEquals("order-123-retry-again", this.cloneRequest.getNewInstanceId()); + assertFalse(this.cloneRequest.hasVersion()); + assertNull(this.restartRequest); + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); } @Test @@ -116,7 +145,11 @@ void explicitTargetMayBeTheSourceId() throws Exception { DurableTaskClient client = startClient(); assertEquals("order-123", client.restartInstance("order-123", new RestartInstanceOptions().setNewInstanceId("order-123").setVersion(""))); - assertEquals("order-123", this.restartRequest.getNewInstanceId().getValue()); + assertEquals("order-123", this.cloneRequest.getSourceInstanceId()); + assertEquals("order-123", this.cloneRequest.getNewInstanceId()); + assertTrue(this.cloneRequest.hasVersion()); + assertEquals("", this.cloneRequest.getVersion().getValue()); + assertNull(this.restartRequest); } @ParameterizedTest @@ -128,8 +161,10 @@ void invalidSourceOrTargetIsRejectedBeforeRpc(String id) throws Exception { new RestartInstanceOptions().setNewInstanceId(id))); assertThrows(IllegalArgumentException.class, () -> client.restartInstance(id, new RestartInstanceOptions().setNewInstanceId("target").setVersion("2.0"))); + assertNull(this.cloneRequest); assertNull(this.restartRequest); assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); } @Test @@ -138,19 +173,37 @@ void missingOptionsOrTargetIsRejectedBeforeRpc() throws Exception { assertThrows(IllegalArgumentException.class, () -> client.restartInstance("source", null)); assertThrows(IllegalArgumentException.class, () -> client.restartInstance("source", new RestartInstanceOptions())); + assertNull(this.cloneRequest); assertNull(this.restartRequest); assertEquals(0, this.metadataRequests); assertNull(this.createRequest); } @ParameterizedTest - @ValueSource(strings = {"NOT_FOUND", "ALREADY_EXISTS", "UNIMPLEMENTED"}) + @ValueSource(strings = {"INVALID_ARGUMENT", "NOT_FOUND", "ALREADY_EXISTS", "UNIMPLEMENTED"}) void backendErrorsAreNotReplacedWithSchedulingFallback(String status) throws Exception { DurableTaskClient client = startClient(); - this.restartError = Status.fromCode(Status.Code.valueOf(status)); + this.cloneError = Status.fromCode(Status.Code.valueOf(status)); StatusRuntimeException error = assertThrows(StatusRuntimeException.class, () -> client.restartInstance("source", new RestartInstanceOptions().setNewInstanceId("target"))); - assertEquals(this.restartError.getCode(), error.getStatus().getCode()); + assertEquals(this.cloneError.getCode(), error.getStatus().getCode()); + assertNotNull(this.cloneRequest); + assertNull(this.restartRequest); + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {"2.0"}) + void unsupportedServerRejectsCloneWithoutCallingLegacyRestart(String version) throws Exception { + DurableTaskClient client = startClient(false); + StatusRuntimeException error = assertThrows(StatusRuntimeException.class, + () -> client.restartInstance("source", + new RestartInstanceOptions().setNewInstanceId("target").setVersion(version))); + assertEquals(Status.Code.UNIMPLEMENTED, error.getStatus().getCode()); + assertNull(this.cloneRequest); + assertNull(this.restartRequest); assertEquals(0, this.metadataRequests); assertNull(this.createRequest); } @@ -158,7 +211,7 @@ void backendErrorsAreNotReplacedWithSchedulingFallback(String status) throws Exc @ParameterizedTest @ValueSource(booleans = {false, true}) void legacyRestartStillSchedulesOriginalNameAndInput(boolean newId) throws Exception { - DurableTaskClient client = startClient(); + DurableTaskClient client = startClient(false); String result = client.restartInstance("order-123", newId); assertEquals(this.createRequest.getInstanceId(), result); if (newId) { @@ -170,6 +223,7 @@ void legacyRestartStillSchedulesOriginalNameAndInput(boolean newId) throws Excep assertEquals("{\"order\":123}", this.createRequest.getInput().getValue()); assertEquals("client-default", this.createRequest.getVersion().getValue()); assertEquals(1, this.metadataRequests); + assertNull(this.cloneRequest); assertNull(this.restartRequest); } } diff --git a/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH b/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH index 538d9ccb..6b27130c 100644 --- a/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH +++ b/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH @@ -1 +1 @@ -962a493037e72b5e08f07a55eff0f121c1aaae80 \ No newline at end of file +0a4f8075e383f6ac783acfd5aee9ecb791334e87 \ No newline at end of file diff --git a/internal/durabletask-protobuf/README.md b/internal/durabletask-protobuf/README.md index 6cd08a42..51643ebd 100644 --- a/internal/durabletask-protobuf/README.md +++ b/internal/durabletask-protobuf/README.md @@ -15,6 +15,11 @@ The proto files are automatically downloaded and updated when running Gradle bui 2. Saves the current commit hash for tracking purposes 3. Updates these files before proto compilation begins +The pinned schema includes the dedicated `CloneInstance` RPC from +[microsoft/durabletask-protobuf#82](https://github.com/microsoft/durabletask-protobuf/pull/82). +The options-based Java restart API uses this RPC without falling back to legacy restart or scheduling APIs. +The existing `RestartInstanceRequest` message remains unchanged. + ## Manual Update If you need to manually update the proto files, you can run: diff --git a/internal/durabletask-protobuf/protos/orchestrator_service.proto b/internal/durabletask-protobuf/protos/orchestrator_service.proto index b9f9f63d..1d6f3933 100644 --- a/internal/durabletask-protobuf/protos/orchestrator_service.proto +++ b/internal/durabletask-protobuf/protos/orchestrator_service.proto @@ -533,25 +533,38 @@ message PurgeInstancesResponse { google.protobuf.BoolValue isComplete = 2; } -// Restarts an orchestration from the beginning with its original name and input, -// without copying its history. message RestartInstanceRequest { string instanceId = 1; - - // When newInstanceId is absent, true generates a new instance ID and false - // restarts using instanceId. bool restartWithNewInstanceId = 2; +} - // An explicit target instance ID. When present, takes precedence over - // restartWithNewInstanceId. - google.protobuf.StringValue newInstanceId = 3; +message RestartInstanceResponse { + string instanceId = 1; +} - // Overrides the restarted orchestration's version, including an empty string. - // When absent, preserves the server's default restart behavior. - google.protobuf.StringValue version = 4; +// Restarts an orchestration from the beginning with the source's original name +// and input, without copying its history. When source and target IDs differ, the +// clone's source lineage identifies sourceInstanceId (the immediate source), not +// an earlier ancestor. Sub-orchestrations must not inherit this source lineage. +// The backend stores lineage internally. This protocol does not expose it or +// require SDKs to expose it. When IDs match, lineage follows the backend's +// existing in-place restart behavior. +message CloneInstanceRequest { + // Required, nonblank ID of the source orchestration. + string sourceInstanceId = 1; + + // Required, nonblank target ID; may equal sourceInstanceId. + // Target conflicts follow the backend's existing instance-ID conflict behavior. + string newInstanceId = 2; + + // Overrides the cloned orchestration's version when present. An explicit + // empty string selects an unversioned orchestration; absence preserves the + // backend's default restart behavior. + google.protobuf.StringValue version = 3; } -message RestartInstanceResponse { +message CloneInstanceResponse { + // The requested target instance ID. string instanceId = 1; } @@ -773,6 +786,17 @@ service TaskHubSidecarService { // Restarts an orchestration instance. rpc RestartInstance(RestartInstanceRequest) returns (RestartInstanceResponse); + // Clones an orchestration to an explicit target instance ID. + // Errors: + // - INVALID_ARGUMENT: either ID is empty or whitespace-only. + // - NOT_FOUND: the source instance does not exist. + // - UNIMPLEMENTED: cloning or the requested version behavior is unsupported. + // Return these errors before scheduling or creating any orchestration. + // Options must not be ignored, and callers and servers must not fall back + // to RestartInstance. Servers without this RPC return UNIMPLEMENTED without + // invoking RestartInstance. + rpc CloneInstance(CloneInstanceRequest) returns (CloneInstanceResponse); + // Waits for an orchestration instance to reach a running or completion state. rpc WaitForInstanceStart(GetInstanceRequest) returns (GetInstanceResponse);