Repository: Azure/azure-functions-durable-js
Pull request #578 (2024-01-31 21:15:03)

Add suspend/resume support
As titled.  _Please add @Julio-Arryo as contributor in the release notes._

This PR merges work contribtued by @Julio-Arryo in PR https://github.com/Azure/azure-functions-durable-js/pull/376.  His forked repo is too outdated and couldn't be auto merged. And the target branch of the old PR is outdated too. So I opened this one instead. 

nytian: Do I neede to merge this changes to branch v2.x too? Not sure if this branch is still in support
ejizba: v2 is in support yes, but we generally only ship new features on the latest version. So no, I would not merge to v2

Patch:
diff --git a/src/durableClient/DurableClient.ts b/src/durableClient/DurableClient.ts
index a75bb02..601767a 100644
--- a/src/durableClient/DurableClient.ts
+++ b/src/durableClient/DurableClient.ts
@@ -529,6 +529,62 @@ export class DurableClient implements types.DurableClient {
         }
     }
 
+    public async suspend(instanceId: string, reason: string): Promise<void> {
+        const idPlaceholder = this.clientData.managementUrls.id;
+        let requestUrl: string;
+        if (this.clientData.rpcBaseUrl) {
+            // Fast local RPC path
+            requestUrl = new URL(
+                `instances/${instanceId}/suspend?reason=${reason}`,
+                this.clientData.rpcBaseUrl
+            ).href;
+        } else {
+            // Legacy app frontend path
+            requestUrl = this.clientData.managementUrls.suspendPostUri
+                .replace(idPlaceholder, instanceId)
+                .replace(this.reasonPlaceholder, reason);
+        }
+
+        const response = await this.axiosInstance.post(requestUrl);
+        switch (response.status) {
+            case 202: // suspend accepted
+            case 410: // instance completed or failed
+                return;
+            case 404:
+                return Promise.reject(new Error(`No instance with ID '${instanceId}' found.`));
+            default:
+                return Promise.reject(this.createGenericError(response));
+        }
+    }
+
+    public async resume(instanceId: string, reason: string): Promise<void> {
+        const idPlaceholder = this.clientData.managementUrls.id;
+        let requestUrl: string;
+        if (this.clientData.rpcBaseUrl) {
+            // Fast local RPC path
+            requestUrl = new URL(
+                `instances/${instanceId}/resume?reason=${reason}`,
+                this.clientData.rpcBaseUrl
+            ).href;
+        } else {
+            // Legacy app frontend path
+            requestUrl = this.clientData.managementUrls.resumePostUri
+                .replace(idPlaceholder, instanceId)
+                .replace(this.reasonPlaceholder, reason);
+        }
+
+        const response = await this.axiosInstance.post(requestUrl);
+        switch (response.status) {
+            case 202: // resume accepted
+            case 410: // instance completed or failed
+                return;
+            case 404:
+                return Promise.reject(new Error(`No instance with ID '${instanceId}' found.`));
+            default:
+                return Promise.reject(this.createGenericError(response));
+        }
+    }
+
     public async waitForCompletionOrCreateCheckStatusResponse(
         request: HttpRequest,
         instanceId: string,
diff --git a/src/history/HistoryEventType.ts b/src/history/HistoryEventType.ts
index c3e189e..fb13458 100644
--- a/src/history/HistoryEventType.ts
+++ b/src/history/HistoryEventType.ts
@@ -22,4 +22,6 @@ export enum HistoryEventType {
     ContinueAsNew = 16,
     GenericEvent = 17,
     HistoryState = 18,
+    ExecutionSuspended = 19,
+    ExecutionResumed = 20,
 }
diff --git a/src/http/HttpManagementPayload.ts b/src/http/HttpManagementPayload.ts
index d76349d..94fb0ca 100644
--- a/src/http/HttpManagementPayload.ts
+++ b/src/http/HttpManagementPayload.ts
@@ -11,6 +11,8 @@ export class HttpManagementPayload implements types.HttpManagementPayload {
         public readonly sendEventPostUri: string,
         public readonly terminatePostUri: string,
         public readonly rewindPostUri: string,
-        public readonly purgeHistoryDeleteUri: string
+        public readonly purgeHistoryDeleteUri: string,
+        public readonly suspendPostUri: string,
+        public readonly resumePostUri: string
     ) {}
 }
diff --git a/src/orchestrations/OrchestrationRuntimeStatus.ts b/src/orchestrations/OrchestrationRuntimeStatus.ts
index f4dbcf5..f1468e2 100644
--- a/src/orchestrations/OrchestrationRuntimeStatus.ts
+++ b/src/orchestrations/OrchestrationRuntimeStatus.ts
@@ -6,4 +6,5 @@ export enum OrchestrationRuntimeStatus {
     Canceled = "Canceled",
     Terminated = "Terminated",
     Pending = "Pending",
+    Suspended = "Suspended",
 }
diff --git a/types/orchestration.d.ts b/types/orchestration.d.ts
index 3eaeb96..611512f 100644
--- a/types/orchestration.d.ts
+++ b/types/orchestration.d.ts
@@ -388,6 +388,12 @@ export declare enum OrchestrationRuntimeStatus {
      * running.
      */
     Pending = "Pending",
+
+    /**
+     * The orchestration instance has been suspended and
+     * may be resumed to running later.
+     */
+    Suspended = "Suspended",
 }
 
 /**


Test patch:
diff --git a/test/testobjects/testconstants.ts b/test/testobjects/testconstants.ts
index 398ad74..373a8c7 100644
--- a/test/testobjects/testconstants.ts
+++ b/test/testobjects/testconstants.ts
@@ -23,6 +23,8 @@ export class TestConstants {
     public static readonly terminatePostUriTemplate: string = `${TestConstants.hostPlaceholder}${TestConstants.webhookPath}instances/${TestConstants.idPlaceholder}/terminate?reason=${TestConstants.reasonPlaceholder}&${TestConstants.uriSuffix}`; // tslint:disable-line max-line-length
     public static readonly rewindPostUriTemplate: string = `${TestConstants.hostPlaceholder}${TestConstants.webhookPath}instances/${TestConstants.idPlaceholder}/rewind?reason=${TestConstants.reasonPlaceholder}&${TestConstants.uriSuffix}`; // tslint:disable-line max-line-length
     public static readonly purgeDeleteUriTemplate: string = `${TestConstants.hostPlaceholder}${TestConstants.webhookPath}instances/${TestConstants.idPlaceholder}?${TestConstants.uriSuffix}`; // tslint:disable-line max-line-length
+    public static readonly suspendPostUriTemplate: string = `${TestConstants.hostPlaceholder}${TestConstants.webhookPath}instances/${TestConstants.idPlaceholder}/suspend?reason=${TestConstants.reasonPlaceholder}&${TestConstants.uriSuffix}`; // tslint:disable-line max-line-length
+    public static readonly resumePostUriTemplate: string = `${TestConstants.hostPlaceholder}${TestConstants.webhookPath}instances/${TestConstants.idPlaceholder}/resume?reason=${TestConstants.reasonPlaceholder}&${TestConstants.uriSuffix}`; // tslint:disable-line max-line-length
 
     public static readonly createPostUriTemplate: string = `${TestConstants.hostPlaceholder}${TestConstants.webhookPath}orchestrators/${TestConstants.functionPlaceholder}${TestConstants.idPlaceholder}?${TestConstants.testCode}`; // tslint:disable-line max-line-length
     public static readonly waitOnPostUriTemplate: string = `${TestConstants.hostPlaceholder}${TestConstants.webhookPath}orchestrators/${TestConstants.functionPlaceholder}${TestConstants.idPlaceholder}?timeout=${TestConstants.timeoutPlaceholder}&pollingInterval=${TestConstants.intervalPlaceholder}&${TestConstants.testCode}`; // tslint:disable-line max-line-length
diff --git a/test/testobjects/testutils.ts b/test/testobjects/testutils.ts
index 7ff5451..44a4367 100644
--- a/test/testobjects/testutils.ts
+++ b/test/testobjects/testutils.ts
@@ -81,6 +81,16 @@ export class TestUtils {
                 .replace(TestConstants.taskHubPlaceholder, taskHub)
                 .replace(TestConstants.connectionPlaceholder, connection),
             TestConstants.purgeDeleteUriTemplate
+                .replace(TestConstants.hostPlaceholder, host)
+                .replace(TestConstants.idPlaceholder, id)
+                .replace(TestConstants.taskHubPlaceholder, taskHub)
+                .replace(TestConstants.connectionPlaceholder, connection),
+            TestConstants.suspendPostUriTemplate
+                .replace(TestConstants.hostPlaceholder, host)
+                .replace(TestConstants.idPlaceholder, id)
+                .replace(TestConstants.taskHubPlaceholder, taskHub)
+                .replace(TestConstants.connectionPlaceholder, connection),
+            TestConstants.resumePostUriTemplate
                 .replace(TestConstants.hostPlaceholder, host)
                 .replace(TestConstants.idPlaceholder, id)
                 .replace(TestConstants.taskHubPlaceholder, taskHub)
diff --git a/test/unit/durableclient-spec.ts b/test/unit/durableclient-spec.ts
index 5b17e42..7db2e84 100644
--- a/test/unit/durableclient-spec.ts
+++ b/test/unit/durableclient-spec.ts
@@ -35,6 +35,8 @@ const durableClientBindingInputJson = JSON.stringify({
         terminatePostUri: `${externalBaseUrl}/instances/INSTANCEID/?taskHub=${testTaskHubName}&connection=${testConnectionName}`,
         rewindPostUri: `${externalBaseUrl}/instances/INSTANCEID/?taskHub=${testTaskHubName}&connection=${testConnectionName}`,
         purgeHistoryDeleteUri: `${externalBaseUrl}/instances/INSTANCEID/?taskHub=${testTaskHubName}&connection=${testConnectionName}`,
+        suspendPostUri: `${externalBaseUrl}/instances/INSTANCEID/?taskHub=${testTaskHubName}&connection=${testConnectionName}`,
+        resumePostUri: `${externalBaseUrl}/instances/INSTANCEID/?taskHub=${testTaskHubName}&connection=${testConnectionName}`,
     },
     baseUrl: externalBaseUrl,
     rpcBaseUrl: testRpcBaseUrl,
@@ -89,6 +91,8 @@ describe("Durable client RPC endpoint", () => {
             expect(payload.sendEventPostUri).to.startWith(externalBaseUrl);
             expect(payload.statusQueryGetUri).to.startWith(externalBaseUrl);
             expect(payload.terminatePostUri).to.startWith(externalBaseUrl);
+            expect(payload.suspendPostUri).to.startWith(externalBaseUrl);
+            expect(payload.resumePostUri).to.startWith(externalBaseUrl);
         });
     });
 
@@ -231,11 +235,11 @@ describe("Durable client RPC endpoint", () => {
             const input = JSON.parse(durableClientBindingInputJson) as OrchestrationClientInputData;
             const client = new DurableClient(input);
 
-            // The getStatusBy() method should do a GET to http://127.0.0.1:17071/durabletask/instances/?createdTimeFrom=2020-01-01T00:00:00Z&createdTimeTo=2020-01-01T23:59:59Z&runtimeStatus=Pending,Running,Completed,Terminated,Failed
+            // The getStatusBy() method should do a GET to http://127.0.0.1:17071/durabletask/instances/?createdTimeFrom=2020-01-01T00:00:00Z&createdTimeTo=2020-01-01T23:59:59Z&runtimeStatus=Pending,Running,Completed,Terminated,Failed,Suspended
             const expectedUrl = new URL(`${testRpcOrigin}/durabletask/instances/`);
             const createdTimeFrom = "2020-01-01T00:00:00.000Z";
             const createdTimeTo = "2020-01-01T23:59:59.000Z";
-            const runtimeStatus = "Pending,Running,Completed,Terminated,Failed";
+            const runtimeStatus = "Pending,Running,Completed,Terminated,Failed,Suspended";
 
             const scope = nock(expectedUrl.origin)
                 .get(expectedUrl.pathname)
@@ -263,11 +267,11 @@ describe("Durable client RPC endpoint", () => {
             const input = JSON.parse(durableClientBindingInputJson) as OrchestrationClientInputData;
             const client = new DurableClient(input);
 
-            // The getStatusBy() method should do a GET to http://127.0.0.1:17071/durabletask/instances/?createdTimeFrom=2020-01-01T00:00:00Z&createdTimeTo=2020-01-01T23:59:59Z&runtimeStatus=Pending,Running,Completed,Terminated,Failed
+            // The getStatusBy() method should do a GET to http://127.0.0.1:17071/durabletask/instances/?createdTimeFrom=2020-01-01T00:00:00Z&createdTimeTo=2020-01-01T23:59:59Z&runtimeStatus=Pending,Running,Completed,Terminated,Failed,Suspended
             const expectedUrl = new URL(`${testRpcOrigin}/durabletask/instances/`);
             const createdTimeFrom = "2020-01-01T00:00:00.000Z";
             const createdTimeTo = "2020-01-01T23:59:59.000Z";
-            const runtimeStatus = "Pending,Running,Completed,Terminated,Failed";
+            const runtimeStatus = "Pending,Running,Completed,Terminated,Failed,Suspended";
 
             // create dummy orchestration status for response
             const dummyDate = new Date();
@@ -339,6 +343,48 @@ describe("Durable client RPC endpoint", () => {
         });
     });
 
+    describe("suspend()", () => {
+        it("uses the RPC endpoint", async () => {
+            const input = JSON.parse(durableClientBindingInputJson) as OrchestrationClientInputData;
+            const client = new DurableClient(input);
+
+            // The suspend() method should do a POST to http://127.0.0.1:17071/durabletask/instances/abc123/suspend?reason=because
+            const instanceId = "abc123";
+            const reason = "because";
+            const expectedUrl = new URL(
+                `${testRpcOrigin}/durabletask/instances/${instanceId}/suspend?reason=${reason}`
+            );
+
+            const scope = nock(expectedUrl.origin)
+                .post(expectedUrl.pathname + expectedUrl.search)
+                .reply(202);
+
+            await client.suspend(instanceId, reason);
+            expect(scope.isDone()).to.be.equal(true);
+        });
+    });
+
+    describe("resume()", () => {
+        it("uses the RPC endpoint", async () => {
+            const input = JSON.parse(durableClientBindingInputJson) as OrchestrationClientInputData;
+            const client = new DurableClient(input);
+
+            // The resume() method should do a POST to http://127.0.0.1:17071/durabletask/instances/abc123/resume?reason=because
+            const instanceId = "abc123";
+            const reason = "because";
+            const expectedUrl = new URL(
+                `${testRpcOrigin}/durabletask/instances/${instanceId}/resume?reason=${reason}`
+            );
+
+            const scope = nock(expectedUrl.origin)
+                .post(expectedUrl.pathname + expectedUrl.search)
+                .reply(202);
+
+            await client.resume(instanceId, reason);
+            expect(scope.isDone()).to.be.equal(true);
+        });
+    });
+
     describe("purgeInstanceHistory[By]()", () => {
         it("uses the RPC endpoint (single instance)", async () => {
             const input = JSON.parse(durableClientBindingInputJson) as OrchestrationClientInputData;
@@ -362,11 +408,11 @@ describe("Durable client RPC endpoint", () => {
             const client = new DurableClient(input);
 
             // The purgeInstanceHistoryBy() method should do a DELETE to
-            // http://127.0.0.1:17071/durabletask/instances/?createdTimeFrom=2020-01-01T00:00:00Z&createdTimeTo=2020-01-01T23:59:59Z&runtimeStatus=Pending,Running,Completed,Terminated,Failed
+            // http://127.0.0.1:17071/durabletask/instances/?createdTimeFrom=2020-01-01T00:00:00Z&createdTimeTo=2020-01-01T23:59:59Z&runtimeStatus=Pending,Running,Completed,Terminated,Failed,Suspended
             const expectedUrl = new URL(`${testRpcOrigin}/durabletask/instances/`);
             const createdTimeFrom = "2020-01-01T00:00:00.000Z";
             const createdTimeTo = "2020-01-01T23:59:59.000Z";
-            const runtimeStatus = "Pending,Running,Completed,Terminated,Failed";
+            const runtimeStatus = "Pending,Running,Completed,Terminated,Failed,Suspended";
 
             const scope = nock(expectedUrl.origin)
                 .delete(expectedUrl.pathname)
diff --git a/test/unit/orchestrationclient-spec.ts b/test/unit/orchestrationclient-spec.ts
index da28b91..e1ffbfb 100644
--- a/test/unit/orchestrationclient-spec.ts
+++ b/test/unit/orchestrationclient-spec.ts
@@ -1200,7 +1200,10 @@ describe("Orchestration Client", () => {
                 .query((actualQueryObject: object) =>
                     urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
                 )
-                .reply(202, new HttpManagementPayload(defaultInstanceId, "", "", "", "", ""));
+                .reply(
+                    202,
+                    new HttpManagementPayload(defaultInstanceId, "", "", "", "", "", "", "")
+                );
 
             const result = await client.startNew(functionName);
             expect(scope.isDone()).to.be.equal(true);
@@ -1223,7 +1226,10 @@ describe("Orchestration Client", () => {
                 .query((actualQueryObject: object) =>
                     urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
                 )
-                .reply(202, new HttpManagementPayload(defaultInstanceId, "", "", "", "", ""));
+                .reply(
+                    202,
+                    new HttpManagementPayload(defaultInstanceId, "", "", "", "", "", "", "")
+                );
 
             const result = await client.startNew(functionName, {
                 instanceId: defaultInstanceId,
@@ -1331,6 +1337,160 @@ describe("Orchestration Client", () => {
         });
     });
 
+    describe("suspend()", () => {
+        beforeEach(async () => {
+            nock.cleanAll();
+        });
+
+        afterEach(async () => {
+            nock.cleanAll();
+        });
+
+        it("calls expected webhook and completes for valid instance", async () => {
+            const client = new DurableClient(defaultClientInputData);
+
+            const testReason = "test";
+            const expectedWebhookUrl = new url.URL(
+                defaultClientInputData.managementUrls.suspendPostUri
+                    .replace(TestConstants.idPlaceholder, defaultInstanceId)
+                    .replace(TestConstants.reasonPlaceholder, testReason)
+            );
+            const scope = nock(expectedWebhookUrl.origin, requiredPostHeaders)
+                .post(expectedWebhookUrl.pathname)
+                .query((actualQueryObject: object) =>
+                    urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
+                )
+                .reply(202);
+
+            const result = await client.suspend(defaultInstanceId, testReason);
+            expect(scope.isDone()).to.be.equal(true);
+            expect(result).to.be.equal(undefined);
+        });
+
+        it(`throws when webhook returns invalid status code 404`, async () => {
+            const client = new DurableClient(defaultClientInputData);
+
+            const id = "badId";
+            const testReason = "test";
+            const expectedWebhookUrl = new url.URL(
+                defaultClientInputData.managementUrls.suspendPostUri
+                    .replace(TestConstants.idPlaceholder, id)
+                    .replace(TestConstants.reasonPlaceholder, testReason)
+            );
+            const scope = nock(expectedWebhookUrl.origin, requiredPostHeaders)
+                .post(expectedWebhookUrl.pathname)
+                .query((actualQueryObject: object) =>
+                    urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
+                )
+                .reply(404);
+
+            await expect(client.suspend(id, testReason)).to.be.rejectedWith(
+                `No instance with ID '${id}' found.`
+            );
+            expect(scope.isDone()).to.be.equal(true);
+        });
+
+        it(`throws when webhook returns invalid status code 500`, async () => {
+            const client = new DurableClient(defaultClientInputData);
+
+            const id = "badId";
+            const testReason = "test";
+            const expectedWebhookUrl = new url.URL(
+                defaultClientInputData.managementUrls.suspendPostUri
+                    .replace(TestConstants.idPlaceholder, id)
+                    .replace(TestConstants.reasonPlaceholder, testReason)
+            );
+            const scope = nock(expectedWebhookUrl.origin, requiredPostHeaders)
+                .post(expectedWebhookUrl.pathname)
+                .query((actualQueryObject: object) =>
+                    urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
+                )
+                .reply(500, "Kah-BOOOM!!");
+
+            await expect(client.suspend(id, testReason)).to.be.rejectedWith(
+                `The operation failed with an unexpected status code: 500. Details: "Kah-BOOOM!!"`
+            );
+            expect(scope.isDone()).to.be.equal(true);
+        });
+    });
+
+    describe("resume()", () => {
+        beforeEach(async () => {
+            nock.cleanAll();
+        });
+
+        afterEach(async () => {
+            nock.cleanAll();
+        });
+
+        it("calls expected webhook and completes for valid instance", async () => {
+            const client = new DurableClient(defaultClientInputData);
+
+            const testReason = "test";
+            const expectedWebhookUrl = new url.URL(
+                defaultClientInputData.managementUrls.resumePostUri
+                    .replace(TestConstants.idPlaceholder, defaultInstanceId)
+                    .replace(TestConstants.reasonPlaceholder, testReason)
+            );
+            const scope = nock(expectedWebhookUrl.origin, requiredPostHeaders)
+                .post(expectedWebhookUrl.pathname)
+                .query((actualQueryObject: object) =>
+                    urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
+                )
+                .reply(202);
+
+            const result = await client.resume(defaultInstanceId, testReason);
+            expect(scope.isDone()).to.be.equal(true);
+            expect(result).to.be.equal(undefined);
+        });
+
+        it(`throws when webhook returns invalid status code 404`, async () => {
+            const client = new DurableClient(defaultClientInputData);
+
+            const id = "badId";
+            const testReason = "test";
+            const expectedWebhookUrl = new url.URL(
+                defaultClientInputData.managementUrls.resumePostUri
+                    .replace(TestConstants.idPlaceholder, id)
+                    .replace(TestConstants.reasonPlaceholder, testReason)
+            );
+            const scope = nock(expectedWebhookUrl.origin, requiredPostHeaders)
+                .post(expectedWebhookUrl.pathname)
+                .query((actualQueryObject: object) =>
+                    urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
+                )
+                .reply(404);
+
+            await expect(client.resume(id, testReason)).to.be.rejectedWith(
+                `No instance with ID '${id}' found.`
+            );
+            expect(scope.isDone()).to.be.equal(true);
+        });
+
+        it(`throws when webhook returns invalid status code 500`, async () => {
+            const client = new DurableClient(defaultClientInputData);
+
+            const id = "badId";
+            const testReason = "test";
+            const expectedWebhookUrl = new url.URL(
+                defaultClientInputData.managementUrls.resumePostUri
+                    .replace(TestConstants.idPlaceholder, id)
+                    .replace(TestConstants.reasonPlaceholder, testReason)
+            );
+            const scope = nock(expectedWebhookUrl.origin, requiredPostHeaders)
+                .post(expectedWebhookUrl.pathname)
+                .query((actualQueryObject: object) =>
+                    urlQueryEqualsQueryObject(expectedWebhookUrl, actualQueryObject)
+                )
+                .reply(500, "Kah-BOOOM!!");
+
+            await expect(client.resume(id, testReason)).to.be.rejectedWith(
+                `The operation failed with an unexpected status code: 500. Details: "Kah-BOOOM!!"`
+            );
+            expect(scope.isDone()).to.be.equal(true);
+        });
+    });
+
     describe("waitForCompletionOrCreateCheckStatusResponse()", () => {
         const defaultRequest: HttpRequest = new HttpRequest({
             url: defaultRequestUrl,