Skip to content

Commit c0aa90b

Browse files
JPeer264claude
andcommitted
fix(cloudflare): Bind pass-through Queue producer methods to the binding
The Queue producer proxy returned methods other than send and sendBatch unbound, so calling a native method like metrics() ran it with the proxy as `this` and workerd threw "Illegal invocation". Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1 parent 8b31bf1 commit c0aa90b

4 files changed

Lines changed: 38 additions & 1 deletion

File tree

‎dev-packages/cloudflare-integration-tests/suites/queue/index.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@ export default Sentry.withSentry(
3535
return new Response('enqueued batch');
3636
}
3737

38+
if (url.pathname === '/metrics') {
39+
return Response.json(await env.MY_QUEUE.metrics());
40+
}
41+
3842
return new Response('not found', { status: 404 });
3943
},
4044
async queue(batch: MessageBatch<{ trigger?: 'error'; payload?: string }>) {

‎dev-packages/cloudflare-integration-tests/suites/queue/test.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -126,3 +126,19 @@ it('emits a queue.publish span with batch attributes on env.MY_QUEUE.sendBatch',
126126
await runner.makeRequest('post', '/enqueue/batch');
127127
await runner.completed();
128128
});
129+
130+
it('calls non-instrumented Queue methods like metrics() on the underlying binding', async ({ signal }) => {
131+
const runner = createRunner(__dirname)
132+
.expect((envelope: Envelope) => {
133+
expect(envelopeItemType(envelope)).toBe('transaction');
134+
expect(envelopeItem(envelope)).toMatchObject({
135+
transaction: 'GET /metrics',
136+
contexts: { trace: { status: 'ok' } },
137+
});
138+
})
139+
.start(signal);
140+
141+
const metrics = await runner.makeRequest('get', '/metrics');
142+
expect(metrics).toMatchObject({ backlogCount: expect.any(Number), backlogBytes: expect.any(Number) });
143+
await runner.completed();
144+
});

‎packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,9 @@ export function instrumentQueueProducer<T extends Queue>(queue: T, bindingName:
9999
};
100100
}
101101

102-
return Reflect.get(target, prop, receiver);
102+
const value = Reflect.get(target, prop) as unknown;
103+
104+
return typeof value === 'function' ? value.bind(target) : value;
103105
},
104106
});
105107
}

‎packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -178,4 +178,19 @@ describe('instrumentQueueProducer', () => {
178178
const wrapped = instrumentQueueProducer(queue, 'MY_QUEUE') as Queue & { customMethod: () => string };
179179
expect(wrapped.customMethod()).toBe('hi');
180180
});
181+
182+
test('calls pass-through methods with the underlying queue as `this`', async () => {
183+
const metrics = { backlogCount: 3, backlogBytes: 42 };
184+
const queue = Object.assign(createMockQueue(), {
185+
metrics(this: unknown) {
186+
if (this !== queue) {
187+
throw new TypeError('Illegal invocation: function called with incorrect `this` reference.');
188+
}
189+
return Promise.resolve(metrics);
190+
},
191+
}) as unknown as Queue & { metrics: () => Promise<typeof metrics> };
192+
const wrapped = instrumentQueueProducer(queue, 'MY_QUEUE') as typeof queue;
193+
194+
await expect(wrapped.metrics()).resolves.toEqual(metrics);
195+
});
181196
});

0 commit comments

Comments
 (0)