diff --git a/packages/cloudflare/src/flush.ts b/packages/cloudflare/src/flush.ts index 77911cd8752c..fe86e21dbd62 100644 --- a/packages/cloudflare/src/flush.ts +++ b/packages/cloudflare/src/flush.ts @@ -35,7 +35,7 @@ const flushLockRegistries = new WeakMap { - const waitUntil = context.waitUntil.bind(context); + const waitUntil = getOriginalWaitUntil(context).bind(context); const client = init({ ...options, ctx: context, enableDedupe: false }); isolationScope.setClient(client); diff --git a/packages/cloudflare/test/flush.test.ts b/packages/cloudflare/test/flush.test.ts index 49ce15dc5153..bcef56a8c101 100644 --- a/packages/cloudflare/test/flush.test.ts +++ b/packages/cloudflare/test/flush.test.ts @@ -165,7 +165,7 @@ describe('getOriginalWaitUntil', () => { expect(result).not.toBe(context.waitUntil); expect(result).toBeDefined(); - result!(Promise.resolve()); + result(Promise.resolve()); expect(originalWaitUntil).toHaveBeenCalled(); }); @@ -183,7 +183,7 @@ describe('getOriginalWaitUntil', () => { const result = getOriginalWaitUntil(context); expect(result).not.toBe(context.waitUntil); - result!(Promise.resolve()); + result(Promise.resolve()); expect(originalWaitUntil).toHaveBeenCalled(); }); @@ -207,7 +207,7 @@ describe('getOriginalWaitUntil', () => { } as unknown as Client; const originalWaitUntil = getOriginalWaitUntil(context); - originalWaitUntil!.call(context, flushAndDispose(mockClient)); + originalWaitUntil.call(context, flushAndDispose(mockClient)); await vi.waitFor(() => Promise.all(waitUntilPromises)); expect(mockClient.flush).toHaveBeenCalled(); diff --git a/packages/cloudflare/test/workflow.test.ts b/packages/cloudflare/test/workflow.test.ts index 5ea96ba8449d..933d721ed93b 100644 --- a/packages/cloudflare/test/workflow.test.ts +++ b/packages/cloudflare/test/workflow.test.ts @@ -209,6 +209,117 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { await expect(drainWaitUntilLikeCloudflareVitestPool(waitUntilPromises)).resolves.toBeUndefined(); }); + test('teardown does not deadlock when a workflow instance is reused across runs', async () => { + const waitUntilPromises: Promise[] = []; + const context: ExecutionContext = { + waitUntil: vi.fn((promise: Promise) => { + waitUntilPromises.push(promise); + }), + passThroughOnException: vi.fn(), + props: {}, + }; + + let runCount = 0; + let releaseAppWork: () => void = () => undefined; + + class ReusedWorkflow { + public constructor(private _ctx: ExecutionContext) {} + + public async run(_event: Readonly>, step: WorkflowStep): Promise { + runCount += 1; + await step.do('reused step', async () => { + if (runCount === 2) { + this._ctx.waitUntil( + new Promise(resolve => { + releaseAppWork = resolve; + }), + ); + } + }); + } + } + + const TestWorkflowInstrumented = instrumentWorkflowWithSentry(getSentryOptions, ReusedWorkflow as any); + // Cloudflare reuses a Workflow instance across runs, so the context + // captured at construction is instrumented by the first run's init() + const workflow = new TestWorkflowInstrumented(context, {}) as ReusedWorkflow; + const event = { payload: {}, timestamp: new Date(), instanceId: INSTANCE_ID }; + + await workflow.run(event, mockStep); + await drainWaitUntilLikeCloudflareVitestPool(waitUntilPromises); + + await workflow.run(event, mockStep); + + releaseAppWork(); + + // Both the application work and the teardown promise must settle + await expect(drainWaitUntilLikeCloudflareVitestPool(waitUntilPromises)).resolves.toBeUndefined(); + }); + + test('step errors are still captured when a workflow instance is reused across runs', async () => { + const waitUntilPromises: Promise[] = []; + const context: ExecutionContext = { + waitUntil: vi.fn((promise: Promise) => { + waitUntilPromises.push(promise); + }), + passThroughOnException: vi.fn(), + props: {}, + }; + + let shouldThrow = false; + + class ReusedErrorWorkflow { + public constructor(private _ctx: ExecutionContext) {} + + public async run(_event: Readonly>, step: WorkflowStep): Promise { + await step.do('flaky step', async () => { + if (shouldThrow) { + shouldThrow = false; + throw new Error('second run error'); + } + }); + } + } + + const TestWorkflowInstrumented = instrumentWorkflowWithSentry(getSentryOptions, ReusedErrorWorkflow as any); + const workflow = new TestWorkflowInstrumented(context, {}) as ReusedErrorWorkflow; + const event = { payload: {}, timestamp: new Date(), instanceId: INSTANCE_ID }; + + await workflow.run(event, mockStep); + await drainWaitUntilLikeCloudflareVitestPool(waitUntilPromises); + + shouldThrow = true; + await workflow.run(event, mockStep); + await drainWaitUntilLikeCloudflareVitestPool(waitUntilPromises); + + expect(mockTransport.send).toHaveBeenCalledWith([ + expect.objectContaining({ + trace: expect.objectContaining({ + transaction: 'flaky step', + trace_id: TRACE_ID, + }), + }), + [ + [ + { + type: 'event', + }, + expect.objectContaining({ + exception: { + values: [ + expect.objectContaining({ + type: 'Error', + value: 'second run error', + mechanism: { type: 'auto.faas.cloudflare.workflow', handled: true }, + }), + ], + }, + }), + ], + ], + ]); + }); + test('Wraps env with instrumentEnv', async () => { class EnvTestWorkflow { constructor(_ctx: ExecutionContext, _env: unknown) {}