Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions packages/durabletask-js/src/testing/in-memory-backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -175,13 +175,19 @@ export class InMemoryOrchestrationBackend {
return;
}

// Update status immediately to match real sidecar behavior, where the
// suspend RPC transitions the orchestration to SUSPENDED right away.
instance.status = pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED;

const event = pbh.newSuspendEvent();
instance.pendingEvents.push(event);
instance.lastUpdatedAt = new Date();

if (!this.orchestrationQueueSet.has(instanceId)) {
this.enqueueOrchestration(instanceId);
}

this.notifyWaiters(instanceId);
}

/**
Expand All @@ -193,13 +199,21 @@ export class InMemoryOrchestrationBackend {
throw new Error(`Orchestration instance '${instanceId}' not found`);
}

// Transition from SUSPENDED back to RUNNING to match real sidecar behavior.
// Only update if the instance was actually suspended.
if (instance.status === pb.OrchestrationStatus.ORCHESTRATION_STATUS_SUSPENDED) {
instance.status = pb.OrchestrationStatus.ORCHESTRATION_STATUS_RUNNING;
}

Comment on lines +202 to +207
const event = pbh.newResumeEvent();
instance.pendingEvents.push(event);
instance.lastUpdatedAt = new Date();

if (!this.orchestrationQueueSet.has(instanceId)) {
this.enqueueOrchestration(instanceId);
}

this.notifyWaiters(instanceId);
}

/**
Expand Down
116 changes: 116 additions & 0 deletions packages/durabletask-js/test/in-memory-backend.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -377,4 +377,120 @@ describe("In-Memory Backend", () => {
expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED);
expect(state?.serializedOutput).toEqual(JSON.stringify(42));
});

describe("suspend and resume status", () => {
it("should update status to SUSPENDED when suspend is called", async () => {
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
yield ctx.waitForExternalEvent("proceed");
return "done";
};

worker.addOrchestrator(orchestrator);
await worker.start();

const id = await client.scheduleNewOrchestration(orchestrator);
await client.waitForOrchestrationStart(id, false, 10);

await client.suspendOrchestration(id);

const state = await client.getOrchestrationState(id);
expect(state?.runtimeStatus).toEqual(OrchestrationStatus.SUSPENDED);
});

it("should update status to RUNNING when resume is called after suspend", async () => {
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
yield ctx.waitForExternalEvent("proceed");
return "done";
};

worker.addOrchestrator(orchestrator);
await worker.start();

const id = await client.scheduleNewOrchestration(orchestrator);
await client.waitForOrchestrationStart(id, false, 10);

await client.suspendOrchestration(id);
let state = await client.getOrchestrationState(id);
expect(state?.runtimeStatus).toEqual(OrchestrationStatus.SUSPENDED);

await client.resumeOrchestration(id);
state = await client.getOrchestrationState(id);
expect(state?.runtimeStatus).toEqual(OrchestrationStatus.RUNNING);
});
Comment on lines +400 to +419

it("should complete successfully after suspend and resume", async () => {
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
const val: number = yield ctx.waitForExternalEvent("proceed");
return val * 2;
};

worker.addOrchestrator(orchestrator);
await worker.start();

const id = await client.scheduleNewOrchestration(orchestrator);
await client.waitForOrchestrationStart(id, false, 10);

// Suspend the orchestration
await client.suspendOrchestration(id);
let state = await client.getOrchestrationState(id);
expect(state?.runtimeStatus).toEqual(OrchestrationStatus.SUSPENDED);

// Send an event while suspended (will be buffered)
await client.raiseOrchestrationEvent(id, "proceed", 21);

// Resume the orchestration
await client.resumeOrchestration(id);

// Wait for completion — the buffered event should be processed
state = await client.waitForOrchestrationCompletion(id, true, 10);
expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED);
expect(state?.serializedOutput).toEqual(JSON.stringify(42));
});

it("should be idempotent when suspend is called twice", async () => {
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
yield ctx.waitForExternalEvent("proceed");
return "done";
};

worker.addOrchestrator(orchestrator);
await worker.start();

const id = await client.scheduleNewOrchestration(orchestrator);
await client.waitForOrchestrationStart(id, false, 10);

// Call suspend twice — should not throw
await client.suspendOrchestration(id);
await client.suspendOrchestration(id);

const state = await client.getOrchestrationState(id);
expect(state?.runtimeStatus).toEqual(OrchestrationStatus.SUSPENDED);
});

it("should notify state waiters on suspend", async () => {
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
yield ctx.waitForExternalEvent("proceed");
return "done";
};

worker.addOrchestrator(orchestrator);
await worker.start();

const id = await client.scheduleNewOrchestration(orchestrator);
await client.waitForOrchestrationStart(id, false, 10);

// Set up a waiter for SUSPENDED status, then suspend
const suspendedPromise = backend.waitForState(
id,
(inst) => backend.toClientStatus(inst.status) === OrchestrationStatus.SUSPENDED,
5000,
);

await client.suspendOrchestration(id);

const suspendedInstance = await suspendedPromise;
expect(suspendedInstance).toBeDefined();
expect(backend.toClientStatus(suspendedInstance!.status)).toEqual(OrchestrationStatus.SUSPENDED);
});
});
});
Loading