support set custom status (#1)

* support set custom status

Signed-off-by: Fabian Martinez <[email protected]>

* update license header

Signed-off-by: Fabian Martinez <[email protected]>

* fix test

Signed-off-by: Fabian Martinez <[email protected]>

* fix e2e test

Signed-off-by: Fabian Martinez <[email protected]>

---------

Signed-off-by: Fabian Martinez <[email protected]>

Author: famarting

examples/activity-sequence.ts

@@ -28,6 +28,8 @@ import { TaskHubGrpcWorker } from "../src/worker/task-hub-grpc-worker";
     const result3 = yield ctx.callActivity(hello, "London");
     cities.push(result3);
 
+    ctx.setCustomStatus("sequence done");
+
     return cities;
   };
 

src/task/context/orchestration-context.ts

@@ -86,4 +86,11 @@ export abstract class OrchestrationContext {
    * @param saveEvents {boolean} A flag indicating whether to add any unprocessed external events in the new orchestration history.
    */
   abstract continueAsNew(newInput: any, saveEvents: boolean): void;
+
+  /**
+   * Sets the custom status
+   *
+   * @param status {string} The new custom status
+   */
+  abstract setCustomStatus(status: string): void;
 }

src/worker/orchestration-execute-result.ts

@@ -0,0 +1,14 @@
+// Copyright 2024 The Dapr Authors.
+// Licensed under the MIT License.
+
+import * as pb from "../proto/orchestrator_service_pb";
+
+export class OrchestrationExecuteResult {
+  actions: pb.OrchestratorAction[];
+  customStatus: string;
+
+  constructor(actions: pb.OrchestratorAction[], customStatus: string) {
+    this.actions = actions;
+    this.customStatus = customStatus;
+  }
+}

src/worker/orchestration-executor.ts

@@ -16,6 +16,7 @@ import { enumValueToKey } from "../utils/enum.util";
 import { getOrchestrationStatusStr, isEmpty } from "../utils/pb-helper.util";
 import { OrchestratorNotRegisteredError } from "./exception/orchestrator-not-registered-error";
 import { StopIterationError } from "./exception/stop-iteration-error";
+import { OrchestrationExecuteResult } from "./orchestration-execute-result";
 import { Registry } from "./registry";
 import { RuntimeOrchestrationContext } from "./runtime-orchestration-context";
 
@@ -36,7 +37,7 @@ export class OrchestrationExecutor {
     instanceId: string,
     oldEvents: pb.HistoryEvent[],
     newEvents: pb.HistoryEvent[],
-  ): Promise<pb.OrchestratorAction[]> {
+  ): Promise<OrchestrationExecuteResult> {
     if (!newEvents?.length) {
       throw new OrchestrationStateError("The new history event list must have at least one event in it");
     }
@@ -79,7 +80,7 @@ export class OrchestrationExecutor {
     const actions = ctx.getActions();
     console.log(`${instanceId}: Returning ${actions.length} action(s)`);
 
-    return actions;
+    return new OrchestrationExecuteResult(actions, ctx._customStatus);
   }
 
   private async processEvent(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise<void> {

src/worker/runtime-orchestration-context.ts

@@ -27,6 +27,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext {
   _pendingEvents: Record<string, CompletableTask<any>[]>;
   _newInput?: any;
   _saveEvents: any;
+  _customStatus: string;
 
   constructor(instanceId: string) {
     super();
@@ -45,6 +46,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext {
     this._pendingEvents = {};
     this._newInput = undefined;
     this._saveEvents = false;
+    this._customStatus = "";
   }
 
   get instanceId(): string {
@@ -330,4 +332,8 @@ export class RuntimeOrchestrationContext extends OrchestrationContext {
 
     this.setContinuedAsNew(newInput, saveEvents);
   }
+
+  setCustomStatus(status: string) {
+    this._customStatus = status;
+  }
 }

src/worker/task-hub-grpc-worker.ts

@@ -217,11 +217,14 @@ export class TaskHubGrpcWorker {
 
     try {
       const executor = new OrchestrationExecutor(this._registry);
-      const actions = await executor.execute(req.getInstanceid(), req.getPasteventsList(), req.getNeweventsList());
+      const result = await executor.execute(req.getInstanceid(), req.getPasteventsList(), req.getNeweventsList());
 
       res = new pb.OrchestratorResponse();
       res.setInstanceid(req.getInstanceid());
-      res.setActionsList(actions);
+      res.setActionsList(result.actions);
+      const cs = new StringValue();
+      cs.setValue(result.customStatus);
+      res.setCustomstatus(cs);
     } catch (e: any) {
       console.error(e);
       console.log(`An error occurred while trying to execute instance '${req.getInstanceid()}': ${e.message}`);

test/e2e/orchestration.spec.ts

@@ -64,6 +64,8 @@ describe("Durable Functions", () => {
         numbers.push(current);
       }
 
+      ctx.setCustomStatus("foobaz");
+
       return numbers;
     };
 
@@ -81,6 +83,7 @@ describe("Durable Functions", () => {
     expect(state?.runtimeStatus).toEqual(OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED);
     expect(state?.serializedInput).toEqual(JSON.stringify(1));
     expect(state?.serializedOutput).toEqual(JSON.stringify([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]));
+    expect(state?.serializedCustomStatus).toEqual("foobaz");
   }, 31000);
 
   it("should be able to run fan-out/fan-in", async () => {
@@ -179,13 +182,13 @@ describe("Durable Functions", () => {
     await taskHubWorker.start();
 
     const id = await taskHubClient.scheduleNewOrchestration(orchestratorParent, SUB_ORCHESTRATION_COUNT);
-    const state = await taskHubClient.waitForOrchestrationCompletion(id, undefined, 30);
+    const state = await taskHubClient.waitForOrchestrationCompletion(id, undefined, 43);
 
     expect(state);
     expect(state?.runtimeStatus).toEqual(OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED);
     expect(state?.failureDetails).toBeUndefined();
     expect(activityCounter).toEqual(SUB_ORCHESTRATION_COUNT * ACTIVITY_COUNT);
-  }, 31000);
+  }, 45000);
 
   it("should allow waiting for multiple external events", async () => {
     const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, _: any): any {