diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index c1748da..dce1de2 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/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; diff --git a/packages/durabletask-js/test/in-memory-backend.spec.ts b/packages/durabletask-js/test/in-memory-backend.spec.ts index e2fac55..07963db 100644 --- a/packages/durabletask-js/test/in-memory-backend.spec.ts +++ b/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 => { + // 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 => { + 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"); + }); });