feat(worker): enhance serial worker with sequence tracking and concur…

…rency control

Modify serial sleep worker to add database sequence tracking and improve worker configuration.
Add end-to-end test to verify maxConcurrent behavior with precise timing checks.

- Update worker to use postgres sequence for tracking message processing
- Modify sleep function to include sequence value retrieval
- Expand worker configuration with additional parameters
- Add comprehensive test for maxConcurrent functionality

pkgs/supaworker/supabase/functions/serial-sleep-worker/index.ts

@@ -1,6 +1,19 @@
 import { Supaworker } from '../_supaworker/index.ts';
 import { delay } from 'jsr:@std/async';
+import postgres from 'postgres';
 
-const sleep1s = () => delay(1000);
+const DB_POOL_URL = Deno.env.get('DB_POOL_URL')!;
+const sql = postgres(DB_POOL_URL);
+await sql`CREATE SEQUENCE IF NOT EXISTS test_seq`;
+await sql`ALTER SEQUENCE test_seq RESTART WITH 1`;
 
-Supaworker.start(sleep1s, { maxConcurrent: 1 });
+const sleep1s = async () => {
+  await sql`SELECT nextval('test_seq')`;
+  await delay(1000);
+};
+
+Supaworker.start(sleep1s, {
+  batchSize: 1,
+  maxConcurrent: 1,
+  visibilityTimeout: 1,
+});

pkgs/supaworker/tests/e2e/README.md

@@ -59,4 +59,4 @@ functions/
 
 ✅ Concurrency
 
-- [ ] Worker respects maxConcurrent (does not read new messages if there are no empty slots, reads max empty-slots-count of messages)
+- [x] Worker respects maxConcurrent and processes messages in serial when set to 1

pkgs/supaworker/tests/e2e/maxConcurrent_works.test.ts

@@ -0,0 +1,43 @@
+import { sql } from '../sql.ts';
+import {
+  assertEquals,
+  assertGreaterOrEqual,
+  assertLess,
+} from 'jsr:@std/assert';
+import { delay } from 'jsr:@std/async';
+import { seqLastValue, waitForSeqValue } from './_helpers.ts';
+
+Deno.test('worker respect maxConcurrent settings', async () => {
+  await sql`CREATE SEQUENCE IF NOT EXISTS test_seq`;
+  await sql`ALTER SEQUENCE test_seq RESTART WITH 1`;
+  await sql`DELETE FROM pgmq.q_pgflow`;
+  await sql`DELETE FROM pgmq.a_pgflow`;
+  await sql`SELECT supaworker.spawn('serial-sleep-worker')`;
+
+  try {
+    // worker sleeps for 1s for each message
+    // se we will expect roughly 1 message per second
+    const startTime = Date.now();
+
+    await sql`SELECT pgmq.send_batch('pgflow', ARRAY['{}', '{}', '{}', '{}', '{}']::jsonb[])`;
+
+    await waitForSeqValue(5);
+
+    const endTime = Date.now();
+    const totalMs = Math.floor(endTime - startTime);
+
+    assertGreaterOrEqual(
+      totalMs,
+      5000,
+      'Should take at least 5 seconds to process all messages'
+    );
+    assertLess(
+      totalMs,
+      7000,
+      'Should take less than 7 seconds to process all messages'
+    );
+  } finally {
+    // Clean up connection
+    await sql.end();
+  }
+});