From 8ad5e988d197540d61211f77c6661ab82c61fadf Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Tue, 28 Apr 2026 09:12:58 +0000 Subject: [PATCH] Fix: Handle EntityUnlockSent events in orchestration executor replay Add missing ENTITYUNLOCKSENT case to the orchestration executor's processEvent switch statement. Previously, EntityUnlockSent history events were silently dropped during replay, causing stale unlock actions to accumulate in _pendingActions and be re-sent to the sidecar on every subsequent replay of orchestrations using entity locks. The fix follows the same validateEntityAction pattern used by all other entity event handlers (EntityOperationCalled, EntityLockRequested, etc.) to properly validate and clean up the pending action. Fixes #235 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/worker/orchestration-executor.ts | 13 +++ .../test/entity-locking.spec.ts | 108 ++++++++++++++++++ 2 files changed, 121 insertions(+) diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index e081d32..a72951e 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -189,6 +189,9 @@ export class OrchestrationExecutor { case pb.HistoryEvent.EventtypeCase.ENTITYLOCKGRANTED: await this.handleEntityLockGranted(ctx, event); break; + case pb.HistoryEvent.EventtypeCase.ENTITYUNLOCKSENT: + await this.handleEntityUnlockSent(ctx, event); + break; default: WorkerLogs.orchestrationUnknownEvent(this._logger, eventTypeName, eventType); } @@ -650,6 +653,16 @@ export class OrchestrationExecutor { await ctx.resume(); } + private async handleEntityUnlockSent(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { + this.validateEntityAction( + ctx, + event, + "lockRelease", + (msg) => msg.hasEntityunlocksent(), + "lockRelease (EntityUnlockSent)", + ); + } + private validateEntityAction( ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent, diff --git a/packages/durabletask-js/test/entity-locking.spec.ts b/packages/durabletask-js/test/entity-locking.spec.ts index e2ea7da..c7df395 100644 --- a/packages/durabletask-js/test/entity-locking.spec.ts +++ b/packages/durabletask-js/test/entity-locking.spec.ts @@ -79,6 +79,30 @@ function createEntityLockRequestedEvent( return event; } +function createEntityUnlockSentEvent( + eventId: number, + criticalSectionId: string, + targetInstanceId: string, + parentInstanceId: string, +): pb.HistoryEvent { + const event = new pb.HistoryEvent(); + event.setEventid(eventId); + const ts = new Timestamp(); + ts.fromDate(new Date()); + event.setTimestamp(ts); + + const unlockSent = new pb.EntityUnlockSentEvent(); + unlockSent.setCriticalsectionid(criticalSectionId); + const targetSv = new StringValue(); + targetSv.setValue(targetInstanceId); + unlockSent.setTargetinstanceid(targetSv); + const parentSv = new StringValue(); + parentSv.setValue(parentInstanceId); + unlockSent.setParentinstanceid(parentSv); + event.setEntityunlocksent(unlockSent); + return event; +} + describe("Entity Locking (Critical Sections)", () => { describe("lockEntities", () => { it("should throw when entity list is empty", async () => { @@ -238,6 +262,90 @@ describe("Entity Locking (Critical Sections)", () => { pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, ); }); + + it("should handle EntityUnlockSent events during replay and not produce stale unlock actions", async () => { + // This test verifies that EntityUnlockSent events in the history are properly + // handled during replay. Before this fix, EntityUnlockSent events were silently + // dropped, leaving stale unlock actions in _pendingActions that would be re-sent + // to the sidecar on every replay. + + // Arrange + const registry = new Registry(); + const orchestratorStartTime = new Date(); + + registry.addOrchestrator(async function* testOrchestration(ctx: any) { + const entity = new EntityInstanceId("counter", "test"); + const lock: LockHandle = yield ctx.entities.lockEntities(entity); + lock.release(); + return "completed"; + }); + + // Step 1: First execution - request lock + const executor1 = new OrchestrationExecutor(registry); + const newEvents1 = [ + createOrchestratorStartedEvent(orchestratorStartTime), + createExecutionStartedEvent("testOrchestration"), + ]; + const result1 = await executor1.execute("test-instance", [], newEvents1); + + // Extract the lock request details + const lockAction = result1.actions.find((a) => a.getSendentitymessage()?.hasEntitylockrequested()); + expect(lockAction).toBeDefined(); + const lockRequest = lockAction!.getSendentitymessage()!.getEntitylockrequested()!; + const criticalSectionId = lockRequest.getCriticalsectionid(); + const lockActionId = lockAction!.getId(); + const lockSet = lockRequest.getLocksetList(); + + // Step 2: Second execution - lock granted, orchestration completes + const executor2 = new OrchestrationExecutor(registry); + const oldEvents2 = [ + createOrchestratorStartedEvent(orchestratorStartTime), + createExecutionStartedEvent("testOrchestration"), + createEntityLockRequestedEvent(lockActionId, criticalSectionId, lockSet), + ]; + const newEvents2 = [ + createOrchestratorStartedEvent(), + createEntityLockGrantedEvent(criticalSectionId), + ]; + const result2 = await executor2.execute("test-instance", oldEvents2, newEvents2); + + // Find unlock action(s) from result2 + const unlockActions = result2.actions.filter((a) => a.getSendentitymessage()?.hasEntityunlocksent()); + expect(unlockActions.length).toBeGreaterThanOrEqual(1); + + // Step 3: Third execution (replay of all events including EntityUnlockSent). + // This simulates a replay where the sidecar sends the full history including + // the EntityUnlockSent event that was produced in step 2. + const executor3 = new OrchestrationExecutor(registry); + const unlockActionId = unlockActions[0].getId(); + const entityInstanceId = lockSet[0]; // e.g. "@counter@test" + + const oldEvents3 = [ + createOrchestratorStartedEvent(orchestratorStartTime), + createExecutionStartedEvent("testOrchestration"), + createEntityLockRequestedEvent(lockActionId, criticalSectionId, lockSet), + createOrchestratorStartedEvent(), + createEntityLockGrantedEvent(criticalSectionId), + createEntityUnlockSentEvent(unlockActionId, criticalSectionId, entityInstanceId, "test-instance"), + ]; + const newEvents3 = [ + createOrchestratorStartedEvent(), + ]; + + // Act + const result3 = await executor3.execute("test-instance", oldEvents3, newEvents3); + + // Assert: The stale unlock actions should NOT be in the returned actions. + // Only the completion action should be present. + const staleUnlockActions = result3.actions.filter((a) => a.getSendentitymessage()?.hasEntityunlocksent()); + expect(staleUnlockActions).toHaveLength(0); + + const completionAction = result3.actions.find((a) => a.hasCompleteorchestration()); + expect(completionAction).toBeDefined(); + expect(completionAction!.getCompleteorchestration()!.getOrchestrationstatus()).toBe( + pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, + ); + }); }); describe("isInCriticalSection", () => {