From 00d9829d99162e59abb6c4124c82227da0238733 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:40:30 +0000 Subject: [PATCH 1/2] Initial plan From 15600dcf6f7ee12fa220afc6fbc977b929b4bef8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:51:44 +0000 Subject: [PATCH 2/2] Add required-target orchestration restart options with optional version Co-authored-by: bachuv <15795068+bachuv@users.noreply.github.com> --- CHANGELOG.md | 3 + README.md | 21 +++ internal/protocol/SOURCE_COMMIT | 2 +- .../protos/orchestrator_service.proto | 14 ++ packages/azure-functions-durable/CHANGELOG.md | 4 +- packages/azure-functions-durable/README.md | 15 +- .../azure-functions-durable/src/client.ts | 6 +- packages/azure-functions-durable/src/index.ts | 1 + .../test/unit/client.spec.ts | 7 + packages/durabletask-js/src/client/client.ts | 35 ++++- packages/durabletask-js/src/index.ts | 1 + .../src/proto/orchestrator_service_pb.d.ts | 17 +++ .../src/proto/orchestrator_service_pb.js | 139 +++++++++++++++++- .../durabletask-js/src/task/options/index.ts | 1 + .../src/task/options/task-options.ts | 10 ++ .../test/client-restart.spec.ts | 99 +++++++++++++ .../test/orchestration_executor.spec.ts | 36 +++++ 17 files changed, 400 insertions(+), 11 deletions(-) create mode 100644 packages/durabletask-js/test/client-restart.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 52f244be..e5864bec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ### New +- Add `restartOrchestration(id, { newInstanceId, version? })` for restart-as-clone with a required + target ID. Omitted versions preserve backend restart defaults; `""` selects unversioned execution. + Boolean restart calls remain unchanged. Target/version support requires the updated backend protocol. - Add payload storage contracts and bounded protobuf payload processing, with inclusive UTF-8 thresholds, explicit storage/size failures, and non-retriable worker upload-failure conversion. This is phase 1 of the external-payload feature; transport, providers, and cleanup follow separately. diff --git a/README.md b/README.md index 2759742a..73aaea3d 100644 --- a/README.md +++ b/README.md @@ -450,6 +450,27 @@ Azure Functions `app.orchestration` / `app.activity` remain host registrations w names, not a multi-version host-routing API. The embedded `DurableFunctionsWorker` inherits core version dispatch and child defaults for integrations that supply versioned protobuf requests. +### Restart as clone + +Restart an existing orchestration from the beginning with its original name and input, not its history: + +```typescript +await client.restartOrchestration("order-123", { + newInstanceId: "order-123-retry", + version: "2.0", +}); +``` + +The options form requires a non-empty `newInstanceId`. Omitting `version` preserves the backend's +restart defaults (not the client's default version); `version: ""` selects an unversioned orchestration. +Existing `restartOrchestration(id, boolean)` calls are unchanged. + +With backend lineage support, the clone sees `ctx.instanceId === "order-123-retry"` and +`ctx.sourceInstanceId === "order-123"`, including during replay. Cloning the clone again references +`"order-123-retry"` as its immediate source. Sub-orchestrations do not inherit this source reference. +Explicit target IDs and version overrides require a backend supporting the updated restart protocol; +older backends may ignore these fields. + ### Continue as new Long-running orchestrations can restart with fresh history and optionally move to a new diff --git a/internal/protocol/SOURCE_COMMIT b/internal/protocol/SOURCE_COMMIT index 7018febb..7149943c 100644 --- a/internal/protocol/SOURCE_COMMIT +++ b/internal/protocol/SOURCE_COMMIT @@ -1 +1 @@ -3145f9337fca9de57d2f89a6ff6f07150d34f1c2 +962a493037e72b5e08f07a55eff0f121c1aaae80 diff --git a/internal/protocol/protos/orchestrator_service.proto b/internal/protocol/protos/orchestrator_service.proto index 3d9194ac..b9f9f63d 100644 --- a/internal/protocol/protos/orchestrator_service.proto +++ b/internal/protocol/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 { diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index 0948e769..022f84bf 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -5,7 +5,9 @@ - Expose readonly `context.df.sourceInstanceId`, forwarding core clone lineage on initial execution and replay. Null or missing metadata returns `undefined`. Actual Functions lineage requires extension lineage support and provider tag persistence. - Restart APIs remain unchanged; destination-ID/version options are not supported by the protocol. +- Add `client.restart(id, { newInstanceId, version? })`, forwarding required-target cloning options + to core. Omitted versions preserve backend restart defaults; `""` selects unversioned execution. + Existing boolean calls remain supported. Target/version options require an updated backend. - Add options-object overloads to `callSubOrchestrator()` and `callSubOrchestratorWithRetry()` so classic orchestrators can attach tags to child orchestration instances. Existing positional `instanceId` and `version` calls are unchanged. diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index 03dd6bdf..d4a73304 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -395,7 +395,20 @@ are `@deprecated`): | `client.suspend(id)` | `client.suspendOrchestration(id)` | | `client.resume(id)` | `client.resumeOrchestration(id)` | | `client.rewind(id, reason)` | `client.rewindInstance(id, reason)` | -| `client.restart(id, restartWithNewInstanceId?)` | `client.restartOrchestration(id, restartWithNewInstanceId?)` | +| `client.restart(id, booleanOrOptions?)` | `client.restartOrchestration(id, booleanOrOptions?)` | + +To restart as a clone with a chosen ID: + +```typescript +await client.restart("order-123", { newInstanceId: "order-123-retry", version: "2.0" }); +``` + +The options form requires `newInstanceId`. Omit `version` to retain backend restart defaults; +use `""` for unversioned execution. Existing boolean calls are unchanged. The clone starts with +the original name/input and fresh history. With backend lineage support, +`context.df.sourceInstanceId` is the immediate source ID on execution and replay; sub-orchestrations +do not inherit it. Target/version options require an updated backend; Functions lineage additionally +requires extension lineage support and provider tag persistence. ## Status diff --git a/packages/azure-functions-durable/src/client.ts b/packages/azure-functions-durable/src/client.ts index 13c9763a..f4103f43 100644 --- a/packages/azure-functions-durable/src/client.ts +++ b/packages/azure-functions-durable/src/client.ts @@ -9,6 +9,7 @@ import { OrchestrationState, OrchestrationStatus, PurgeInstanceCriteria, + RestartOrchestrationOptions, TaskHubGrpcClient, } from "@microsoft/durabletask-js"; import { HttpManagementPayload, createHttpManagementPayload as createPayload } from "./http-management-payload"; @@ -80,6 +81,9 @@ export interface StartNewOptions { dedupeStatuses?: readonly OrchestrationStatus[]; } +/** Options for {@link DurableFunctionsClient.restart} with a chosen target ID. */ +export type RestartOptions = RestartOrchestrationOptions; + /** * @hidden * Removes the input/result payload fields from a core history event, honoring v3's @@ -559,7 +563,7 @@ export class DurableFunctionsClient extends TaskHubGrpcClient { * @param restartWithNewInstanceId - When `true`, the restarted instance is assigned a new ID. * @returns The instance ID of the restarted orchestration. */ - async restart(instanceId: string, restartWithNewInstanceId = false): Promise { + async restart(instanceId: string, restartWithNewInstanceId: boolean | RestartOptions = false): Promise { return this.restartOrchestration(instanceId, restartWithNewInstanceId); } diff --git a/packages/azure-functions-durable/src/index.ts b/packages/azure-functions-durable/src/index.ts index 2ef05bfa..46ac5d30 100644 --- a/packages/azure-functions-durable/src/index.ts +++ b/packages/azure-functions-durable/src/index.ts @@ -14,6 +14,7 @@ export { DurableFunctionsClientInput, DurableOrchestrationClient, StartNewOptions, + RestartOptions, TaskHubOptions, getGrpcHostAddress, } from "./client"; diff --git a/packages/azure-functions-durable/test/unit/client.spec.ts b/packages/azure-functions-durable/test/unit/client.spec.ts index 5e2f52b8..685b5088 100644 --- a/packages/azure-functions-durable/test/unit/client.spec.ts +++ b/packages/azure-functions-durable/test/unit/client.spec.ts @@ -106,6 +106,13 @@ describe("DurableFunctionsClient", () => { const restart = jest.spyOn(client, "restartOrchestration").mockResolvedValue("id-2"); await expect(client.restart("id-1", true)).resolves.toBe("id-2"); expect(restart).toHaveBeenCalledWith("id-1", true); + await expect(client.restart("id-1")).resolves.toBe("id-2"); + expect(restart).toHaveBeenLastCalledWith("id-1", false); + for (const version of [undefined, "", "2.0"]) { + const options = { newInstanceId: "id-2", version }; + await expect(client.restart("id-1", options)).resolves.toBe("id-2"); + expect(restart).toHaveBeenLastCalledWith("id-1", options); + } const purge = jest.spyOn(client, "purgeOrchestration").mockResolvedValue(undefined); const purgeResult = await client.purgeInstanceHistoryBy({ diff --git a/packages/durabletask-js/src/client/client.ts b/packages/durabletask-js/src/client/client.ts index 0b56ad04..66e296d0 100644 --- a/packages/durabletask-js/src/client/client.ts +++ b/packages/durabletask-js/src/client/client.ts @@ -28,7 +28,7 @@ import { convertFailureDetails } from "../utils/failure-details.util"; import { HistoryEvent } from "../orchestration/history-event"; import { convertProtoHistoryEvent } from "../utils/history-event-converter"; import { Logger, ConsoleLogger } from "../types/logger.type"; -import { StartOrchestrationOptions } from "../task/options"; +import { StartOrchestrationOptions, RestartOrchestrationOptions } from "../task/options"; import { toProtobufOrchestrationIdReusePolicy } from "../orchestration/orchestration-id-reuse-policy"; import { OrchestrationAlreadyExistsError } from "../orchestration/exception/orchestration-already-exists-error"; import { mapToRecord } from "../utils/tags.util"; @@ -738,13 +738,16 @@ export class TaskHubGrpcClient { * * Restarting with a new ID is a restart-as-clone, not execution-history cloning. * The context's `sourceInstanceId` is available only if the backend supplies - * lineage metadata. The current protocol does not support choosing a destination ID or version. + * lineage metadata. Pass options to choose a target ID and optionally override the version. + * Omitted versions preserve backend restart defaults; an empty string selects an unversioned orchestration. + * Explicit target IDs and version overrides require a backend supporting these protocol fields. * * @param instanceId - The unique ID of the orchestration instance to restart. * @param restartWithNewInstanceId - If true, the restarted orchestration will be assigned * a new instance ID. If false (default), the same instance ID will be reused. * When reusing the same instance ID, the orchestration must be in a terminal state * (Completed, Failed, or Terminated). + * Alternatively, pass options with a required `newInstanceId` and optional `version`. * @returns A Promise that resolves to the instance ID of the restarted orchestration. * This will be the same as the input instanceId if restartWithNewInstanceId is false, * or a new ID if restartWithNewInstanceId is true. @@ -752,16 +755,38 @@ export class TaskHubGrpcClient { * @throws Error if the orchestration cannot be restarted (e.g., it's still running * and restartWithNewInstanceId is false). */ - async restartOrchestration(instanceId: string, restartWithNewInstanceId: boolean = false): Promise { + async restartOrchestration( + instanceId: string, + restartWithNewInstanceId: boolean | RestartOrchestrationOptions = false, + ): Promise { if (!instanceId) { throw new Error("instanceId cannot be null or empty"); } const req = new pb.RestartInstanceRequest(); req.setInstanceid(instanceId); - req.setRestartwithnewinstanceid(restartWithNewInstanceId); + if (typeof restartWithNewInstanceId === "boolean") { + req.setRestartwithnewinstanceid(restartWithNewInstanceId); + } else { + if ( + !restartWithNewInstanceId || + typeof restartWithNewInstanceId.newInstanceId !== "string" || + !restartWithNewInstanceId.newInstanceId + ) { + throw new TypeError("newInstanceId is required and must be a non-empty string"); + } + const { newInstanceId, version } = restartWithNewInstanceId; + if (version !== undefined && typeof version !== "string") { + throw new TypeError("version must be a string"); + } + req.setRestartwithnewinstanceid(true); + req.setNewinstanceid(new StringValue().setValue(newInstanceId)); + if (version !== undefined) { + req.setVersion(new StringValue().setValue(version)); + } + } - ClientLogs.restartingInstance(this._logger, instanceId, restartWithNewInstanceId); + ClientLogs.restartingInstance(this._logger, instanceId, req.getRestartwithnewinstanceid()); const res = await this._call( this._stub.restartInstance.bind(this._stub), diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index 5e22ec89..7e1d7dcd 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -110,6 +110,7 @@ export { TaskOptions, SubOrchestrationOptions, StartOrchestrationOptions, + RestartOrchestrationOptions, TaskRetryOptions, isRetryPolicy, isRetryHandler, diff --git a/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts b/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts index 59ac1b3b..0d974f85 100644 --- a/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts +++ b/packages/durabletask-js/src/proto/orchestrator_service_pb.d.ts @@ -2647,6 +2647,16 @@ export class RestartInstanceRequest extends jspb.Message { getRestartwithnewinstanceid(): boolean; setRestartwithnewinstanceid(value: boolean): RestartInstanceRequest; + hasNewinstanceid(): boolean; + clearNewinstanceid(): void; + getNewinstanceid(): google_protobuf_wrappers_pb.StringValue | undefined; + setNewinstanceid(value?: google_protobuf_wrappers_pb.StringValue): RestartInstanceRequest; + + hasVersion(): boolean; + clearVersion(): void; + getVersion(): google_protobuf_wrappers_pb.StringValue | undefined; + setVersion(value?: google_protobuf_wrappers_pb.StringValue): RestartInstanceRequest; + serializeBinary(): Uint8Array; toObject(includeInstance?: boolean): RestartInstanceRequest.AsObject; static toObject(includeInstance: boolean, msg: RestartInstanceRequest): RestartInstanceRequest.AsObject; @@ -2661,6 +2671,8 @@ export namespace RestartInstanceRequest { export type AsObject = { instanceid: string, restartwithnewinstanceid: boolean, + newinstanceid?: google_protobuf_wrappers_pb.StringValue.AsObject, + version?: google_protobuf_wrappers_pb.StringValue.AsObject, } } @@ -3505,6 +3517,9 @@ export class StartNewOrchestrationAction extends jspb.Message { getParenttracecontext(): TraceContext | undefined; setParenttracecontext(value?: TraceContext): StartNewOrchestrationAction; + getTagsMap(): jspb.Map; + clearTagsMap(): void; + serializeBinary(): Uint8Array; toObject(includeInstance?: boolean): StartNewOrchestrationAction.AsObject; static toObject(includeInstance: boolean, msg: StartNewOrchestrationAction): StartNewOrchestrationAction.AsObject; @@ -3524,6 +3539,8 @@ export namespace StartNewOrchestrationAction { scheduledtime?: google_protobuf_timestamp_pb.Timestamp.AsObject, requesttime?: google_protobuf_timestamp_pb.Timestamp.AsObject, parenttracecontext?: TraceContext.AsObject, + + tagsMap: Array<[string, string]>, } } diff --git a/packages/durabletask-js/src/proto/orchestrator_service_pb.js b/packages/durabletask-js/src/proto/orchestrator_service_pb.js index 210f5cd2..346b2736 100644 --- a/packages/durabletask-js/src/proto/orchestrator_service_pb.js +++ b/packages/durabletask-js/src/proto/orchestrator_service_pb.js @@ -21816,7 +21816,9 @@ proto.RestartInstanceRequest.prototype.toObject = function(opt_includeInstance) proto.RestartInstanceRequest.toObject = function(includeInstance, msg) { var f, obj = { instanceid: jspb.Message.getFieldWithDefault(msg, 1, ""), - restartwithnewinstanceid: jspb.Message.getBooleanFieldWithDefault(msg, 2, false) + restartwithnewinstanceid: jspb.Message.getBooleanFieldWithDefault(msg, 2, false), + newinstanceid: (f = msg.getNewinstanceid()) && google_protobuf_wrappers_pb.StringValue.toObject(includeInstance, f), + version: (f = msg.getVersion()) && google_protobuf_wrappers_pb.StringValue.toObject(includeInstance, f) }; if (includeInstance) { @@ -21861,6 +21863,16 @@ proto.RestartInstanceRequest.deserializeBinaryFromReader = function(msg, reader) var value = /** @type {boolean} */ (reader.readBool()); msg.setRestartwithnewinstanceid(value); break; + case 3: + var value = new google_protobuf_wrappers_pb.StringValue; + reader.readMessage(value,google_protobuf_wrappers_pb.StringValue.deserializeBinaryFromReader); + msg.setNewinstanceid(value); + break; + case 4: + var value = new google_protobuf_wrappers_pb.StringValue; + reader.readMessage(value,google_protobuf_wrappers_pb.StringValue.deserializeBinaryFromReader); + msg.setVersion(value); + break; default: reader.skipField(); break; @@ -21904,6 +21916,22 @@ proto.RestartInstanceRequest.serializeBinaryToWriter = function(message, writer) f ); } + f = message.getNewinstanceid(); + if (f != null) { + writer.writeMessage( + 3, + f, + google_protobuf_wrappers_pb.StringValue.serializeBinaryToWriter + ); + } + f = message.getVersion(); + if (f != null) { + writer.writeMessage( + 4, + f, + google_protobuf_wrappers_pb.StringValue.serializeBinaryToWriter + ); + } }; @@ -21943,6 +21971,80 @@ proto.RestartInstanceRequest.prototype.setRestartwithnewinstanceid = function(va }; +/** + * optional google.protobuf.StringValue newInstanceId = 3; + * @return {?proto.google.protobuf.StringValue} + */ +proto.RestartInstanceRequest.prototype.getNewinstanceid = function() { + return /** @type{?proto.google.protobuf.StringValue} */ ( + jspb.Message.getWrapperField(this, google_protobuf_wrappers_pb.StringValue, 3)); +}; + + +/** + * @param {?proto.google.protobuf.StringValue|undefined} value + * @return {!proto.RestartInstanceRequest} returns this +*/ +proto.RestartInstanceRequest.prototype.setNewinstanceid = function(value) { + return jspb.Message.setWrapperField(this, 3, value); +}; + + +/** + * Clears the message field making it undefined. + * @return {!proto.RestartInstanceRequest} returns this + */ +proto.RestartInstanceRequest.prototype.clearNewinstanceid = function() { + return this.setNewinstanceid(undefined); +}; + + +/** + * Returns whether this field is set. + * @return {boolean} + */ +proto.RestartInstanceRequest.prototype.hasNewinstanceid = function() { + return jspb.Message.getField(this, 3) != null; +}; + + +/** + * optional google.protobuf.StringValue version = 4; + * @return {?proto.google.protobuf.StringValue} + */ +proto.RestartInstanceRequest.prototype.getVersion = function() { + return /** @type{?proto.google.protobuf.StringValue} */ ( + jspb.Message.getWrapperField(this, google_protobuf_wrappers_pb.StringValue, 4)); +}; + + +/** + * @param {?proto.google.protobuf.StringValue|undefined} value + * @return {!proto.RestartInstanceRequest} returns this +*/ +proto.RestartInstanceRequest.prototype.setVersion = function(value) { + return jspb.Message.setWrapperField(this, 4, value); +}; + + +/** + * Clears the message field making it undefined. + * @return {!proto.RestartInstanceRequest} returns this + */ +proto.RestartInstanceRequest.prototype.clearVersion = function() { + return this.setVersion(undefined); +}; + + +/** + * Returns whether this field is set. + * @return {boolean} + */ +proto.RestartInstanceRequest.prototype.hasVersion = function() { + return jspb.Message.getField(this, 4) != null; +}; + + @@ -27815,7 +27917,8 @@ proto.StartNewOrchestrationAction.toObject = function(includeInstance, msg) { input: (f = msg.getInput()) && google_protobuf_wrappers_pb.StringValue.toObject(includeInstance, f), scheduledtime: (f = msg.getScheduledtime()) && google_protobuf_timestamp_pb.Timestamp.toObject(includeInstance, f), requesttime: (f = msg.getRequesttime()) && google_protobuf_timestamp_pb.Timestamp.toObject(includeInstance, f), - parenttracecontext: (f = msg.getParenttracecontext()) && proto.TraceContext.toObject(includeInstance, f) + parenttracecontext: (f = msg.getParenttracecontext()) && proto.TraceContext.toObject(includeInstance, f), + tagsMap: (f = msg.getTagsMap()) ? f.toObject(includeInstance, undefined) : [] }; if (includeInstance) { @@ -27885,6 +27988,12 @@ proto.StartNewOrchestrationAction.deserializeBinaryFromReader = function(msg, re reader.readMessage(value,proto.TraceContext.deserializeBinaryFromReader); msg.setParenttracecontext(value); break; + case 8: + var value = msg.getTagsMap(); + reader.readMessage(value, function(message, reader) { + jspb.Map.deserializeBinary(message, reader, jspb.BinaryReader.prototype.readString, jspb.BinaryReader.prototype.readString, null, "", ""); + }); + break; default: reader.skipField(); break; @@ -27968,6 +28077,10 @@ proto.StartNewOrchestrationAction.serializeBinaryToWriter = function(message, wr proto.TraceContext.serializeBinaryToWriter ); } + f = message.getTagsMap(true); + if (f && f.getLength() > 0) { + f.serializeBinary(8, writer, jspb.BinaryWriter.prototype.writeString, jspb.BinaryWriter.prototype.writeString); + } }; @@ -28192,6 +28305,28 @@ proto.StartNewOrchestrationAction.prototype.hasParenttracecontext = function() { }; +/** + * map tags = 8; + * @param {boolean=} opt_noLazyCreate Do not create the map if + * empty, instead returning `undefined` + * @return {!jspb.Map} + */ +proto.StartNewOrchestrationAction.prototype.getTagsMap = function(opt_noLazyCreate) { + return /** @type {!jspb.Map} */ ( + jspb.Message.getMapField(this, 8, opt_noLazyCreate, + null)); +}; + + +/** + * Clears values from the map. The map will be non-null. + * @return {!proto.StartNewOrchestrationAction} returns this + */ +proto.StartNewOrchestrationAction.prototype.clearTagsMap = function() { + this.getTagsMap().clear(); + return this;}; + + diff --git a/packages/durabletask-js/src/task/options/index.ts b/packages/durabletask-js/src/task/options/index.ts index 8ea3058e..87f8de65 100644 --- a/packages/durabletask-js/src/task/options/index.ts +++ b/packages/durabletask-js/src/task/options/index.ts @@ -5,6 +5,7 @@ export { TaskOptions, SubOrchestrationOptions, StartOrchestrationOptions, + RestartOrchestrationOptions, TaskRetryOptions, isRetryPolicy, isRetryHandler, diff --git a/packages/durabletask-js/src/task/options/task-options.ts b/packages/durabletask-js/src/task/options/task-options.ts index 691cebf0..f9b98228 100644 --- a/packages/durabletask-js/src/task/options/task-options.ts +++ b/packages/durabletask-js/src/task/options/task-options.ts @@ -91,6 +91,16 @@ export interface StartOrchestrationOptions { dedupeStatuses?: readonly OrchestrationStatus[]; } +/** + * Options for restarting an orchestration with a chosen instance ID. + */ +export interface RestartOrchestrationOptions { + /** The required target instance ID. */ + newInstanceId: string; + /** Omit to preserve backend restart defaults; an empty string selects an unversioned orchestration. */ + version?: string; +} + /** * Type guard to check if the retry option is a RetryPolicy. * diff --git a/packages/durabletask-js/test/client-restart.spec.ts b/packages/durabletask-js/test/client-restart.spec.ts new file mode 100644 index 00000000..766d7198 --- /dev/null +++ b/packages/durabletask-js/test/client-restart.spec.ts @@ -0,0 +1,99 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import * as grpc from "@grpc/grpc-js"; +import { EventEmitter } from "events"; +import { NoOpLogger, RestartOrchestrationOptions, TaskHubGrpcClient } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; + +describe("restartOrchestration", () => { + let client: TaskHubGrpcClient; + let restart: jest.Mock; + let requests: pb.RestartInstanceRequest[]; + + beforeEach(() => { + client = new TaskHubGrpcClient({ defaultVersion: "client-default", logger: new NoOpLogger() }); + requests = []; + restart = jest.fn((request, _metadata, callback) => { + const decoded = pb.RestartInstanceRequest.deserializeBinary(request.serializeBinary()); + requests.push(decoded); + callback( + null, + new pb.RestartInstanceResponse().setInstanceid( + decoded.getNewinstanceid()?.getValue() ?? + (decoded.getRestartwithnewinstanceid() ? "generated-id" : decoded.getInstanceid()), + ), + ); + return Object.assign(new EventEmitter(), { cancel: jest.fn() }); + }); + client["_stub"].restartInstance = restart; + }); + + afterEach(() => { + client["_stub"].close(); + }); + + it.each([undefined, false, true])("preserves the legacy boolean request for %s", async (flag) => { + await expect(client.restartOrchestration("source", flag)).resolves.toBe(flag ? "generated-id" : "source"); + expect(requests[0].getInstanceid()).toBe("source"); + expect(requests[0].getRestartwithnewinstanceid()).toBe(flag ?? false); + expect(requests[0].hasNewinstanceid()).toBe(false); + expect(requests[0].hasVersion()).toBe(false); + }); + + it.each([undefined, "", "2.0"])("preserves target and version presence for %s", async (version) => { + const options: RestartOrchestrationOptions = { newInstanceId: "Source:@Case/retry", version }; + await expect(client.restartOrchestration("source", options)).resolves.toBe(options.newInstanceId); + expect(requests[0].getInstanceid()).toBe("source"); + expect(requests[0].getRestartwithnewinstanceid()).toBe(true); + expect(requests[0].getNewinstanceid()?.getValue()).toBe(options.newInstanceId); + expect(requests[0].hasVersion()).toBe(version !== undefined); + expect(requests[0].getVersion()?.getValue()).toBe(version); + expect(options).toEqual({ newInstanceId: "Source:@Case/retry", version }); + }); + + it("uses the immediate source when cloning a clone", async () => { + await client.restartOrchestration("original", { newInstanceId: "clone" }); + await client.restartOrchestration("clone", { newInstanceId: "next-clone" }); + expect(requests.map((req) => [req.getInstanceid(), req.getNewinstanceid()?.getValue()])).toEqual([ + ["original", "clone"], + ["clone", "next-clone"], + ]); + }); + + it.each([{}, { newInstanceId: "" }, { newInstanceId: null }, { newInstanceId: 42 }, null])( + "rejects invalid target options %j without making an RPC", + async (options) => { + await expect(client.restartOrchestration("source", options as any)).rejects.toThrow("newInstanceId"); + expect(restart).not.toHaveBeenCalled(); + }, + ); + + it.each([null, 42])("rejects an invalid version %j without making an RPC", async (version) => { + await expect( + client.restartOrchestration("source", { newInstanceId: "clone", version: version as any }), + ).rejects.toThrow("version must be a string"); + expect(restart).not.toHaveBeenCalled(); + }); + + it("rejects an empty source ID before making an RPC", async () => { + await expect(client.restartOrchestration("", { newInstanceId: "clone" })).rejects.toThrow("instanceId"); + expect(restart).not.toHaveBeenCalled(); + }); + + it("preserves backend failures for explicit-target requests", async () => { + const error = Object.assign(new Error("missing"), { + code: grpc.status.NOT_FOUND, + metadata: new grpc.Metadata(), + details: "missing", + }); + restart.mockImplementation((_request, _metadata, callback) => { + callback(error); + return Object.assign(new EventEmitter(), { cancel: jest.fn() }); + }); + await expect(client.restartOrchestration("source", { newInstanceId: "clone" })).rejects.toMatchObject({ + message: "An orchestration with the instanceId 'source' was not found.", + cause: error, + }); + }); +}); diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index 4e54aeb1..fa4bba8e 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -62,6 +62,42 @@ describe("Orchestration Context", () => { }); describe("Orchestration Executor", () => { + it("starts clones afresh with immediate lineage without passing lineage to sub-orchestrations", async () => { + const registry = new Registry(); + registry.addNamedOrchestrator("Flow", async function* (ctx, input) { + ctx.setCustomStatus([ctx.name, input, ctx.instanceId, ctx.sourceInstanceId, ctx.version, ctx.isReplaying]); + yield ctx.callSubOrchestrator("Child", input, { instanceId: `${ctx.instanceId}-child` }); + }); + registry.addNamedOrchestrator("Child", async (ctx, input) => [ + input, ctx.instanceId, ctx.sourceInstanceId, ctx.parent?.instanceId, + ]); + for (const [sourceId, cloneId] of [["original", "clone"], ["clone", "next-clone"]]) { + const startEvents = [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent("Flow", cloneId, '"original-input"', undefined, undefined, "2.0"), + ]; + const initial = await new OrchestrationExecutor(registry, testLogger).execute( + cloneId, [], startEvents, undefined, sourceId, + ); + expect(JSON.parse(initial.customStatus!)).toEqual([ + "Flow", "original-input", cloneId, sourceId, "2.0", false, + ]); + expect(initial.actions).toHaveLength(1); + const childAction = initial.actions[0]; + expect(childAction.getId()).toBe(1); + expect(childAction.getCreatesuborchestration()?.getInput()?.getValue()).toBe('"original-input"'); + const childId = childAction.getCreatesuborchestration()!.getInstanceid(); + const child = await new OrchestrationExecutor(registry, testLogger).execute(childId, [], [ + newExecutionStartedEvent("Child", childId, '"original-input"', { + name: "Flow", instanceId: cloneId, taskScheduledId: childAction.getId(), + }), + ]); + expect(getAndValidateSingleCompleteOrchestrationAction(child)?.getResult()?.getValue()).toBe( + JSON.stringify(["original-input", childId, undefined, cloneId]), + ); + } + }); + it.each([undefined, "", "Source:@Case/with.punctuation-123"])( "preserves source %s independently of parent, identity, version, actions, and replay", async (sourceInstanceId) => {