Skip to content

Commit b04fb74

Browse files
pombosilvapombosilva
committed
Fix pause - should not wait for sleeps or waitForEvents
1 parent a28dbc5 commit b04fb74

File tree

10 files changed

+285
-119
lines changed

10 files changed

+285
-119
lines changed

‎fixtures/vitest-pool-workers-examples/workflows/test/integration.test.ts‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -157,11 +157,6 @@ describe("workflow instance lifecycle methods", () => {
157157
const handle = await env.MODERATOR.get(data.id);
158158
await handle.pause();
159159

160-
await handle.sendEvent({
161-
type: "moderation-decision",
162-
payload: { moderatorAction: "approve" },
163-
});
164-
165160
// ASSERTIONS:
166161
await expect(instance.waitForStatus("paused")).resolves.not.toThrow();
167162

‎fixtures/workflow-multiple/tests/index.test.ts‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -121,13 +121,6 @@ describe("Workflows", () => {
121121
// Pause the instance
122122
await fetchJson(`http://${ip}:${port}/pause?workflowName=3&id=${id}`);
123123

124-
// Send event to unblock waitForEvent
125-
await fetchJson(
126-
`http://${ip}:${port}/sendEvent?workflowName=3&id=${id}`,
127-
{ done: true },
128-
"POST"
129-
);
130-
131124
await vi.waitFor(
132125
async () => {
133126
const result = (await fetchJson(
@@ -141,6 +134,12 @@ describe("Workflows", () => {
141134
// Resume the instance
142135
await fetchJson(`http://${ip}:${port}/resume?workflowName=3&id=${id}`);
143136

137+
await fetchJson(
138+
`http://${ip}:${port}/sendEvent?workflowName=3&id=${id}`,
139+
{ done: true },
140+
"POST"
141+
);
142+
144143
await vi.waitFor(
145144
async () => {
146145
const result = (await fetchJson(

‎fixtures/workflow/tests/index.test.ts‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -286,13 +286,6 @@ describe("Workflows", () => {
286286
// Pause the instance
287287
await fetchJson(`http://${ip}:${port}/pause?workflowName=${name}`);
288288

289-
// Send event to unblock waitForEvent
290-
await fetchJson(
291-
`http://${ip}:${port}/sendEvent?workflowName=${name}`,
292-
{ event: true },
293-
"POST"
294-
);
295-
296289
await vi.waitFor(
297290
async () => {
298291
const result = await fetchJson(
@@ -303,17 +296,23 @@ describe("Workflows", () => {
303296
{ timeout: 1500 }
304297
);
305298

299+
// Resume the instance
306300
await fetchJson(`http://${ip}:${port}/resume?workflowName=${name}`);
307301

308-
// After resume, the workflow should eventually complete
302+
await fetchJson(
303+
`http://${ip}:${port}/sendEvent?workflowName=${name}`,
304+
{ event: true },
305+
"POST"
306+
);
307+
309308
await vi.waitFor(
310309
async () => {
311310
const result = await fetchJson(
312311
`http://${ip}:${port}/get2?workflowName=${name}`
313312
);
314313
expect(result.status).toBe("complete");
315314
},
316-
{ timeout: 1500 }
315+
{ timeout: 5000 }
317316
);
318317
});
319318

‎packages/miniflare/test/plugins/workflows/index.spec.ts‎

Lines changed: 7 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -88,12 +88,12 @@ export class LifecycleWorkflow extends WorkflowEntrypoint {
8888
async run(event, step) {
8989
const first = await step.do("first step", async () => "step-1-done");
9090
91-
const eventData = await step.waitForEvent("wait for signal", {
92-
type: "continue",
93-
timeout: "1 day",
91+
await step.do("long step", async () => {
92+
await scheduler.wait(500);
93+
return "long-step-done";
9494
});
9595
96-
const second = await step.do("second step", async () => "step-2-done");
96+
const second = await step.do("third step", async () => "step-3-done");
9797
9898
return "workflow-complete";
9999
}
@@ -224,19 +224,13 @@ describe("workflow instance lifecycle methods", () => {
224224

225225
await waitForStepOutput(mf, "pause-resume-test", "step-1-done");
226226

227-
// Pause the instance
227+
// Pause the instance — waits for the in-flight long step to finish, then pauses
228228
const pauseRes = await mf.dispatchFetch(
229229
"http://localhost/pause?id=pause-resume-test"
230230
);
231231
const pauseData = (await pauseRes.json()) as Record<string, unknown>;
232232
expect(pauseData).toHaveProperty("status");
233233

234-
// Send the event to unblock waitForEvent
235-
const eventRes = await mf.dispatchFetch(
236-
"http://localhost/sendEvent?id=pause-resume-test"
237-
);
238-
expect(((await eventRes.json()) as Record<string, unknown>).ok).toBe(true);
239-
240234
await waitForStatus(mf, "pause-resume-test", "paused");
241235

242236
// Resume the instance
@@ -246,7 +240,7 @@ describe("workflow instance lifecycle methods", () => {
246240
const resumeData = (await resumeRes.json()) as Record<string, unknown>;
247241
expect(resumeData).toHaveProperty("status");
248242

249-
// After resume, the workflow should complete (step-2 runs, then returns)
243+
// After resume, the workflow should complete (third step runs, then returns)
250244
const finalStatus = await waitForStatus(
251245
mf,
252246
"pause-resume-test",
@@ -299,17 +293,7 @@ describe("workflow instance lifecycle methods", () => {
299293
const restartData = (await restartRes.json()) as Record<string, unknown>;
300294
expect(restartData).toHaveProperty("status");
301295

302-
// After restart, the workflow restarts from scratch — wait for "running"
303-
await waitForStatus(mf, "restart-test", "running");
304-
305-
// Send the event to complete the restarted workflow
306-
// (workflow replays: first step is cached, then waitForEvent blocks again)
307-
await waitForStepOutput(mf, "restart-test", "step-1-done");
308-
const eventRes = await mf.dispatchFetch(
309-
"http://localhost/sendEvent?id=restart-test"
310-
);
311-
await eventRes.text(); // consume the body
312-
296+
// After restart, the workflow restarts from scratch and runs to completion
313297
const finalStatus = await waitForStatus(mf, "restart-test", "complete");
314298
expect(finalStatus.output).toBe("workflow-complete");
315299
});

‎packages/workflows-shared/README.md‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,3 @@
11
# `@cloudflare/workflows-shared`
22

3-
This is a package that is used at Cloudflare to power some internal features of [Cloudflare Workflows](https://developers.cloudflare.com/workflows/), as well as their open-source equivalents here in workers-sdk and Wrangler.
4-
5-
> [!NOTE]
6-
> Since code in this package is used by the Workflows infrastructure, it is important that PRs are given careful review with regards to how they could cause a failure in production.
7-
> Ideally, there should be comprehensive tests for changes being made to give extra confidence about the behavior.
3+
This package powers the local development experience for [Cloudflare Workflows](https://developers.cloudflare.com/workflows/) in workers-sdk and Wrangler.

‎packages/workflows-shared/src/context.ts‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,10 @@ export class Context extends RpcTarget {
6767
}
6868

6969
async #checkForPendingPause(): Promise<void> {
70+
if (this.#engine.timeoutHandler.isRunningStep()) {
71+
return;
72+
}
73+
7074
const status = await this.#engine.getStatus();
7175
if (status === InstanceStatus.WaitingForPause) {
7276
await this.#state.storage.put(PAUSE_DATETIME, new Date());
@@ -643,9 +647,6 @@ export class Context extends RpcTarget {
643647

644648
// @ts-expect-error priorityQueue is initiated in init
645649
this.#engine.priorityQueue.remove({ hash: cacheKey, type: "sleep" });
646-
647-
// Check if a pause was requested while this sleep was running
648-
await this.#checkForPendingPause();
649650
}
650651

651652
async sleepUntil(name: string, timestamp: Date | number): Promise<void> {
@@ -832,9 +833,6 @@ export class Context extends RpcTarget {
832833
throw error;
833834
});
834835

835-
// Check if a pause was requested while we were waiting for the event
836-
await this.#checkForPendingPause();
837-
838836
return result as WorkflowStepEvent<T>;
839837
}
840838
}

‎packages/workflows-shared/src/engine.ts‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -553,6 +553,7 @@ export class Engine extends DurableObject<Env> {
553553
const currentStatus = await this.getStatus();
554554
if (currentStatus === InstanceStatus.WaitingForPause) {
555555
// Engine is still running — cancel the pending pause
556+
this.timeoutHandler.cancelWaitingPromisesByType("pause");
556557
await this.setStatus(
557558
metadata.accountId,
558559
metadata.instance.id,
@@ -639,14 +640,21 @@ export class Engine extends DurableObject<Env> {
639640
);
640641
}
641642

642-
// Set to WaitingForPause — the engine keeps running.
643-
// The Context class will check for this between steps, store
644-
// PAUSE_DATETIME, transition to Paused, and then abort.
645643
await this.setStatus(
646644
metadata.accountId,
647645
metadata.instance.id,
648646
InstanceStatus.WaitingForPause
649647
);
648+
649+
void this.timeoutHandler.waitUntilNothingIsRunning("pause", async () => {
650+
await this.ctx.storage.put(PAUSE_DATETIME, new Date());
651+
await this.setStatus(
652+
metadata.accountId,
653+
metadata.instance.id,
654+
InstanceStatus.Paused
655+
);
656+
await this.abort(ABORT_REASONS.USER_PAUSE);
657+
});
650658
}
651659

652660
async userTriggeredRestart() {

‎packages/workflows-shared/src/lib/gracePeriodSemaphore.ts‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,24 @@ export const ENGINE_TIMEOUT = ms("5 minutes" satisfies WorkflowSleepDuration);
77

88
let latestGracePeriodTimestamp: number | undefined = undefined;
99

10+
export type WaitingPromiseType = "pause";
11+
1012
export type GracePeriodCallback = (engine: Engine, timeoutMs: number) => void;
1113

1214
export class GracePeriodSemaphore {
1315
#counter: number = 0;
1416
readonly callback: GracePeriodCallback;
1517
readonly timeoutMs: number;
18+
#waitingPromises: {
19+
rejectCallback: () => void;
20+
resolveCallback: (value: unknown) => void;
21+
type: WaitingPromiseType;
22+
}[] = [];
23+
#canInitiateSteps = true;
24+
#waitingSteps: {
25+
rejectCallback: () => void;
26+
resolveCallback: (value: unknown) => void;
27+
}[] = [];
1628

1729
constructor(callback: GracePeriodCallback, timeoutMs: number) {
1830
this.callback = callback;
@@ -21,6 +33,14 @@ export class GracePeriodSemaphore {
2133

2234
// acquire takes engine to be the same as release
2335
async acquire(_engine: Engine) {
36+
if (!this.#canInitiateSteps) {
37+
await new Promise((resolve, reject) => {
38+
this.#waitingSteps.push({
39+
resolveCallback: resolve,
40+
rejectCallback: reject,
41+
});
42+
});
43+
}
2444
// when the counter goes from 0 to 1 - we can safely reject the previous grace period
2545
if (this.#counter == 0) {
2646
latestGracePeriodTimestamp = undefined;
@@ -34,7 +54,61 @@ export class GracePeriodSemaphore {
3454
// Trigger timeout promise, no need to await here,
3555
// this can be triggered slightly after it's not time sensitive
3656
this.callback(engine, this.timeoutMs);
57+
// Resolve any promises waiting for all steps to finish (e.g. pause)
58+
for (const promise of this.#waitingPromises) {
59+
promise.resolveCallback(undefined);
60+
}
61+
this.#waitingPromises = [];
62+
}
63+
}
64+
65+
async waitUntilNothingIsRunning(
66+
type: WaitingPromiseType,
67+
callback: () => Promise<void>
68+
): Promise<void> {
69+
this.#canInitiateSteps = false;
70+
if (this.#counter > 0) {
71+
try {
72+
await new Promise((resolve, reject) => {
73+
this.#waitingPromises.push({
74+
resolveCallback: resolve,
75+
rejectCallback: reject,
76+
type,
77+
});
78+
});
79+
} catch {
80+
// If the promise gets rejected (e.g. resume cancels the pause),
81+
// allow steps to run again
82+
this.#canInitiateSteps = true;
83+
return;
84+
}
85+
}
86+
await callback();
87+
// Allow steps to run again and unblock any that were waiting
88+
for (const promise of this.#waitingSteps) {
89+
promise.resolveCallback(undefined);
90+
}
91+
this.#waitingSteps = [];
92+
this.#canInitiateSteps = true;
93+
}
94+
95+
cancelWaitingPromisesByType(type: WaitingPromiseType) {
96+
const sameTypePromises = this.#waitingPromises.filter(
97+
(val) => val.type === type
98+
);
99+
if (sameTypePromises.length === 0) {
100+
return;
37101
}
102+
103+
for (const promise of sameTypePromises) {
104+
promise.rejectCallback();
105+
}
106+
107+
this.#waitingPromises = this.#waitingPromises.filter(
108+
(val) => val.type !== type
109+
);
110+
111+
this.#canInitiateSteps = true;
38112
}
39113

40114
isRunningStep() {

0 commit comments

Comments
 (0)