Simplify queue spans and fix missing span.end() calls

Author: onurtemizkan

dev-packages/e2e-tests/test-applications/supabase-nextjs/pages/api/queue/purge.ts

@@ -0,0 +1,36 @@
+import { NextApiRequest, NextApiResponse } from 'next';
+import { createClient } from '@supabase/supabase-js';
+
+// These are the default development keys for a local Supabase instance
+const NEXT_PUBLIC_SUPABASE_URL = 'http://localhost:54321';
+const SUPABASE_SERVICE_ROLE_KEY =
+  'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU';
+
+const supabaseClient = createClient(NEXT_PUBLIC_SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY);
+
+// NOTE: Not instrumenting with Sentry intentionally - this is just a cleanup helper
+
+export default async function handler(req: NextApiRequest, res: NextApiResponse) {
+  // Purge all messages from the todos queue by consuming them in a loop
+  let purgedCount = 0;
+  const maxIterations = 100; // Safety limit
+
+  for (let i = 0; i < maxIterations; i++) {
+    const { data, error } = await supabaseClient.schema('pgmq_public').rpc('pop', {
+      queue_name: 'todos',
+    });
+
+    if (error) {
+      return res.status(500).json({ error: error.message, purgedCount });
+    }
+
+    // No more messages to pop
+    if (!data || (Array.isArray(data) && data.length === 0)) {
+      break;
+    }
+
+    purgedCount++;
+  }
+
+  return res.status(200).json({ purgedCount });
+}

dev-packages/e2e-tests/test-applications/supabase-nextjs/tests/performance.test.ts

@@ -368,6 +368,9 @@ test('Sends queue publish spans with `rpc(...)`', async ({ page, baseURL }) => {
 });
 
 test('Sends queue process spans with `schema(...).rpc(...)`', async ({ page, baseURL }) => {
+  // Purge any stale messages from previous tests to ensure we get the message we just produced
+  await fetch(`${baseURL}/api/queue/purge`);
+
   const producerTransactionPromise = waitForTransaction('supabase-nextjs', transactionEvent => {
     return Boolean(
       transactionEvent?.contexts?.trace?.op === 'http.server' &&
@@ -426,7 +429,7 @@ test('Sends queue process spans with `schema(...).rpc(...)`', async ({ page, bas
     data: expect.objectContaining({
       'messaging.destination.name': 'todos',
       'messaging.system': 'supabase',
-      'messaging.message.id': '1',
+      'messaging.message.id': expect.any(String),
       'messaging.operation.type': 'process',
       'messaging.operation.name': 'pop',
       'messaging.message.body.size': expect.any(Number),
@@ -471,13 +474,16 @@ test('Sends queue process spans with `schema(...).rpc(...)`', async ({ page, bas
     message: 'queue.process(todos)',
     data: {
       'messaging.destination.name': 'todos',
-      'messaging.message.id': '1',
+      'messaging.message.id': expect.any(String),
       'messaging.message.body.size': expect.any(Number),
     },
   });
 });
 
 test('Sends queue process spans with `rpc(...)`', async ({ page, baseURL }) => {
+  // Purge any stale messages from previous tests to ensure we get the message we just produced
+  await fetch(`${baseURL}/api/queue/purge`);
+
   const producerTransactionPromise = waitForTransaction('supabase-nextjs', transactionEvent => {
     return !!(
       transactionEvent?.contexts?.trace?.op === 'http.server' &&
@@ -528,15 +534,15 @@ test('Sends queue process spans with `rpc(...)`', async ({ page, baseURL }) => {
   expect(queueMessage.message._sentry).toBeUndefined();
 
   const consumerSpan = transactionEvent.spans?.find(
-    span => span.op === 'queue.process' && span.data?.['messaging.message.id'] === '2',
+    span => span.op === 'queue.process' && span.description === 'process todos',
   );
   expect(consumerSpan).toBeDefined();
 
   expect(consumerSpan).toMatchObject({
     data: expect.objectContaining({
       'messaging.destination.name': 'todos',
       'messaging.system': 'supabase',
-      'messaging.message.id': '2',
+      'messaging.message.id': expect.any(String),
       'messaging.operation.type': 'process',
       'messaging.operation.name': 'pop',
       'messaging.message.body.size': expect.any(Number),
@@ -581,7 +587,7 @@ test('Sends queue process spans with `rpc(...)`', async ({ page, baseURL }) => {
     message: 'queue.process(todos)',
     data: {
       'messaging.destination.name': 'todos',
-      'messaging.message.id': '2',
+      'messaging.message.id': expect.any(String),
       'messaging.message.body.size': expect.any(Number),
     },
   });