fix: handle EntityUnlockSent replay events

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<void> {
+    this.validateEntityAction(
+      ctx,
+      event,
+      "lockRelease",
+      (msg) => msg.hasEntityunlocksent(),
+      "lockRelease (EntityUnlockSent)",
+    );
+  }
+
   private validateEntityAction(
     ctx: RuntimeOrchestrationContext,
     event: pb.HistoryEvent,

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", () => {