diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f93f23..0bcab12 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. 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 90d3a2e..de314cf 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 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") @@ -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 3864e48..41080fe 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java @@ -330,6 +330,28 @@ 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. 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; + * 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, RestartInstanceOptions options) { + 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 8a107aa..0d19248 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java @@ -455,6 +455,23 @@ public String restartInstance(String instanceId, boolean restartWithNewInstanceI } } + @Override + 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(); + + CloneInstanceRequest.Builder request = CloneInstanceRequest.newBuilder() + .setSourceInstanceId(instanceId) + .setNewInstanceId(newInstanceId); + if (version != null) { + request.setVersion(StringValue.of(version)); + } + return this.sidecarClient.cloneInstance(request.build()).getInstanceId(); + } + private PurgeResult toPurgeResult(PurgeInstancesResponse response){ return new PurgeResult(response.getDeletedInstanceCount()); } 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 0000000..845bea9 --- /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/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java index 71f9c9d..2bd0984 100644 --- a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java +++ b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java @@ -93,6 +93,18 @@ default Logger createReplaySafeLogger(Logger logger) { */ String getVersion(); + /** + * Gets the immediate source instance ID when this orchestration was created as a clone. + * 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 expose it in tags + */ + @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 5bdbad2..c390d7c 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 tagSourceInstanceId = startedEvent.getTagsMap().get("MS_DurableFunctions_SourceInstanceId"); + this.sourceInstanceId = !startedEvent.hasParentInstance() && !Helpers.isNullOrEmpty(tagSourceInstanceId) + ? tagSourceInstanceId : 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 0000000..7324bab --- /dev/null +++ b/client/src/test/java/com/microsoft/durabletask/DurableTaskGrpcClientRestartTest.java @@ -0,0 +1,229 @@ +// 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.ServerMethodDefinition; +import io.grpc.ServerServiceDefinition; +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 CloneInstanceRequest cloneRequest; + private RestartInstanceRequest restartRequest; + private CreateInstanceRequest createRequest; + private int metadataRequests; + private Status cloneError; + + private DurableTaskClient startClient() throws Exception { + return startClient(true); + } + + private DurableTaskClient startClient(boolean supportsCloning) throws Exception { + String name = InProcessServerBuilder.generateName(); + 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; + observer.onNext(RestartInstanceResponse.newBuilder() + .setInstanceId("legacy-restart").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(); + } + }; + 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(); + } + + @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", + new RestartInstanceOptions().setNewInstanceId("order-123-retry").setVersion(version))); + 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.cloneRequest.getVersion().getValue()); + } + assertNull(this.restartRequest); + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); + } + + @Test + void omittedVersionAndChainedCloneUseCloneRpc() throws Exception { + DurableTaskClient client = startClient(); + assertEquals("order-123-retry", client.restartInstance("order-123", + new RestartInstanceOptions().setNewInstanceId("order-123-retry"))); + 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.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 + void explicitTargetMayBeTheSourceId() throws Exception { + DurableTaskClient client = startClient(); + assertEquals("order-123", client.restartInstance("order-123", + new RestartInstanceOptions().setNewInstanceId("order-123").setVersion(""))); + 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 + @NullAndEmptySource + @ValueSource(strings = {" ", "\t"}) + void invalidSourceOrTargetIsRejectedBeforeRpc(String id) throws Exception { + DurableTaskClient client = startClient(); + 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.cloneRequest); + assertNull(this.restartRequest); + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); + } + + @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.cloneRequest); + assertNull(this.restartRequest); + assertEquals(0, this.metadataRequests); + assertNull(this.createRequest); + } + + @ParameterizedTest + @ValueSource(strings = {"INVALID_ARGUMENT", "NOT_FOUND", "ALREADY_EXISTS", "UNIMPLEMENTED"}) + void backendErrorsAreNotReplacedWithSchedulingFallback(String status) throws Exception { + DurableTaskClient client = startClient(); + this.cloneError = Status.fromCode(Status.Code.valueOf(status)); + StatusRuntimeException error = assertThrows(StatusRuntimeException.class, + () -> client.restartInstance("source", new RestartInstanceOptions().setNewInstanceId("target"))); + 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); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void legacyRestartStillSchedulesOriginalNameAndInput(boolean newId) throws Exception { + DurableTaskClient client = startClient(false); + 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.cloneRequest); + 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 0000000..929eee1 --- /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 a981c68..6b27130 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 +0a4f8075e383f6ac783acfd5aee9ecb791334e87 \ No newline at end of file diff --git a/internal/durabletask-protobuf/README.md b/internal/durabletask-protobuf/README.md index 6b31046..51643eb 100644 --- a/internal/durabletask-protobuf/README.md +++ b/internal/durabletask-protobuf/README.md @@ -11,10 +11,15 @@ 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 +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 3d9194a..1d6f393 100644 --- a/internal/durabletask-protobuf/protos/orchestrator_service.proto +++ b/internal/durabletask-protobuf/protos/orchestrator_service.proto @@ -542,6 +542,32 @@ message RestartInstanceResponse { string instanceId = 1; } +// 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 CloneInstanceResponse { + // The requested target instance ID. + string instanceId = 1; +} + message CreateTaskHubRequest { bool recreateIfExists = 1; } @@ -706,6 +732,7 @@ message StartNewOrchestrationAction { google.protobuf.Timestamp scheduledTime = 5; google.protobuf.Timestamp requestTime = 6; TraceContext parentTraceContext = 7; + map tags = 8; } message AbandonActivityTaskRequest { @@ -759,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);