fix: handle entity message actions in InMemoryOrchestrationBackend

packages/durabletask-js/src/testing/in-memory-backend.ts

@@ -551,6 +551,9 @@ export class InMemoryOrchestrationBackend {
       case pb.OrchestratorAction.OrchestratoractiontypeCase.SENDEVENT:
         this.processSendEventAction(action.getSendevent()!);
         break;
+      case pb.OrchestratorAction.OrchestratoractiontypeCase.SENDENTITYMESSAGE:
+        this.processSendEntityMessageAction(instance, action.getSendentitymessage()!);
+        break;
       case pb.OrchestratorAction.OrchestratoractiontypeCase.REWINDORCHESTRATION:
         this.processRewindOrchestrationAction(instance, action.getRewindorchestration()!);
         break;
@@ -739,6 +742,42 @@ export class InMemoryOrchestrationBackend {
       });
   }
 
+  /**
+   * Handles an entity message emitted by an orchestration.
+   *
+   * This backend has no entity worker, so only fire-and-forget messages can be honoured:
+   * `signalEntity` and the unlock sent when a critical section ends expect no reply and are
+   * dropped. `callEntity` and `lockEntities` block on a response that would never arrive, so
+   * they fail fast with an explanatory error rather than hanging the orchestration forever.
+   */
+  private processSendEntityMessageAction(
+    instance: OrchestrationInstance,
+    entityMessage: pb.SendEntityMessageAction,
+  ): void {
+    const messageType = entityMessage.getEntitymessagetypeCase();
+    const messageTypeCase = pb.SendEntityMessageAction.EntitymessagetypeCase;
+
+    switch (messageType) {
+      case messageTypeCase.ENTITYOPERATIONSIGNALED:
+      case messageTypeCase.ENTITYUNLOCKSENT:
+        break;
+      case messageTypeCase.ENTITYOPERATIONCALLED:
+      case messageTypeCase.ENTITYLOCKREQUESTED:
+        throw new Error(
+          `Orchestration '${instance.instanceId}' used ` +
+            `${messageType === messageTypeCase.ENTITYOPERATIONCALLED ? "callEntity" : "lockEntities"}, which ` +
+            `the in-memory test backend does not support because it has no entity worker to produce a response. ` +
+            `Only fire-and-forget entity messages (signalEntity and the unlock that ends a critical section) ` +
+            `are supported here; use a real backend or the emulator to test entity calls and locks.`,
+        );
+      default:
+        throw new Error(
+          `Unknown entity message type '${messageType}' for orchestration '${instance.instanceId}'. ` +
+            `This likely means the in-memory backend needs to be updated to handle a newly introduced entity message type.`,
+        );
+    }
+  }
+
   private prepareRewind(instance: OrchestrationInstance, reason?: string): void {
     // Reset instance state so it can be re-processed.
     instance.status = pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING;

packages/durabletask-js/test/in-memory-backend.spec.ts

@@ -12,6 +12,7 @@ import {
   OrchestrationContext,
   Task,
   TOrchestrator,
+  EntityInstanceId,
 } from "../src";
 import * as pb from "../src/proto/orchestrator_service_pb";
 
@@ -933,4 +934,116 @@ describe("In-Memory Backend", () => {
       expect(backend.toClientStatus(suspendedInstance!.status)).toEqual(OrchestrationStatus.SUSPENDED);
     });
   });
+
+  it("should not crash when orchestration signals an entity (fire-and-forget)", async () => {
+    const entityId = new EntityInstanceId("counter", "mykey");
+
+    const orchestrator: TOrchestrator = async (ctx: OrchestrationContext): Promise<any> => {
+      // Signal an entity (fire-and-forget). The in-memory backend doesn't
+      // process entities, but it must not crash on the SENDENTITYMESSAGE action.
+      ctx.entities.signalEntity(entityId, "add", 1);
+      return "signaled";
+    };
+
+    worker.addOrchestrator(orchestrator);
+    await worker.start();
+
+    const id = await client.scheduleNewOrchestration(orchestrator);
+    const state = await client.waitForOrchestrationCompletion(id, true, 10);
+
+    expect(state).toBeDefined();
+    expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED);
+    expect(state?.serializedOutput).toEqual(JSON.stringify("signaled"));
+  });
+
+  it("should not crash when orchestration signals multiple entities", async () => {
+    const entity1 = new EntityInstanceId("counter", "key1");
+    const entity2 = new EntityInstanceId("counter", "key2");
+
+    const orchestrator: TOrchestrator = async (ctx: OrchestrationContext): Promise<any> => {
+      ctx.entities.signalEntity(entity1, "add", 1);
+      ctx.entities.signalEntity(entity2, "add", 2);
+      return "done";
+    };
+
+    worker.addOrchestrator(orchestrator);
+    await worker.start();
+
+    const id = await client.scheduleNewOrchestration(orchestrator);
+    const state = await client.waitForOrchestrationCompletion(id, true, 10);
+
+    expect(state).toBeDefined();
+    expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED);
+    expect(state?.serializedOutput).toEqual(JSON.stringify("done"));
+  });
+
+  it("should not crash when orchestration signals entity then calls activity", async () => {
+    const entityId = new EntityInstanceId("counter", "mykey");
+
+    const doubleIt = async (_: ActivityContext, input: number) => {
+      return input * 2;
+    };
+
+    const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
+      // Signal an entity, then call an activity. The backend must handle
+      // the SENDENTITYMESSAGE action without crashing before the activity
+      // result can be processed.
+      ctx.entities.signalEntity(entityId, "add", 5);
+      const result: number = yield ctx.callActivity(doubleIt, 21);
+      return result;
+    };
+
+    worker.addOrchestrator(orchestrator);
+    worker.addActivity(doubleIt);
+    await worker.start();
+
+    const id = await client.scheduleNewOrchestration(orchestrator);
+    const state = await client.waitForOrchestrationCompletion(id, true, 10);
+
+    expect(state).toBeDefined();
+    expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED);
+    expect(state?.serializedOutput).toEqual(JSON.stringify(42));
+  });
+
+  it("should fail with an explanatory error when an orchestration calls an entity", async () => {
+    // callEntity waits for a response that no entity worker will ever send here, so the
+    // backend must fail the orchestration rather than let it hang until the test times out.
+    const entityId = new EntityInstanceId("counter", "mykey");
+
+    const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
+      return yield ctx.entities.callEntity(entityId, "get");
+    };
+
+    worker.addOrchestrator(orchestrator);
+    await worker.start();
+
+    const id = await client.scheduleNewOrchestration(orchestrator);
+    const state = await client.waitForOrchestrationCompletion(id, true, 10);
+
+    expect(state).toBeDefined();
+    expect(state?.runtimeStatus).toEqual(OrchestrationStatus.FAILED);
+    expect(state?.failureDetails?.message).toContain("callEntity");
+    expect(state?.failureDetails?.message).toContain("does not support");
+  });
+
+  it("should fail with an explanatory error when an orchestration locks entities", async () => {
+    // lockEntities waits for an EntityLockGranted response, which this backend cannot produce.
+    const entityId = new EntityInstanceId("counter", "mykey");
+
+    const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
+      yield ctx.entities.lockEntities(entityId);
+      return "locked";
+    };
+
+    worker.addOrchestrator(orchestrator);
+    await worker.start();
+
+    const id = await client.scheduleNewOrchestration(orchestrator);
+    const state = await client.waitForOrchestrationCompletion(id, true, 10);
+
+    expect(state).toBeDefined();
+    expect(state?.runtimeStatus).toEqual(OrchestrationStatus.FAILED);
+    expect(state?.failureDetails?.message).toContain("lockEntities");
+    expect(state?.failureDetails?.message).toContain("does not support");
+  });
 });