chore: debug problems with worker getting stuck

Squashed commit of the following: commit 0e7f1c7e85bcb7dbfdc737e5e12cc59315ea13a3 Author: Wojtek Majewski <[email protected]> Date: Wed Jan 22 16:18:51 2025 +0100 chore(deps): update deno.lock with p-queue dependency Add npm p-queue package to project dependencies Update lock file with new package and its transitive dependencies commit aa17c507bb840461e0ca2a55795d27ec044098e2 Author: Wojtek Majewski <[email protected]> Date: Wed Jan 22 16:18:28 2025 +0100 refactor(executor): Simplify message execution and error handling Modify message execution flow to use direct signal abort check and update archiving method. Remove complex Promise.race implementation in favor of a more straightforward execution approach. commit 6c1b1c278319951b6e1814ab5a056da7226cc4ef Author: Wojtek Majewski <[email protected]> Date: Wed Jan 22 16:15:47 2025 +0100 refactor(edge-worker): Adjust max concurrency and delay settings Modify max concurrent workers and randomize delay time for test sequence increment to improve worker behavior and testing
pgflow-dev · Jan 22, 2025 · da12151 · da12151
1 parent 97b7e3e
commit da12151
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 21 deletions.
diff --git a/pkgs/edge-worker/deno.lock b/pkgs/edge-worker/deno.lock
diff --git a/pkgs/edge-worker/src/MessageExecutor.ts b/pkgs/edge-worker/src/MessageExecutor.ts
@@ -39,26 +39,16 @@ export class MessageExecutor<MessagePayload extends Json> {
         throw new AbortError();
       }
 
-      await Promise.race([
-        this.messageHandler(this.record.message!),
-        new Promise((_, reject) => {
-          this.signal.addEventListener(
-            'abort',
-            () => {
-              console.log(
-                `################ MessageExecutor aborted during execution: ${this.msgId} ##`
-              );
-              reject(new AbortError());
-            },
-            { once: true }
-          );
-        }),
-      ]);
+      // Check if already aborted before starting
+      this.signal.throwIfAborted();
+
+      await this.messageHandler(this.record.message!);
 
       console.log(
         `[MessageExecutor] Task ${this.msgId} completed successfully, archiving...`
       );
-      await this.batchArchiver.add(this.msgId);
+      await this.queue.archive(this.msgId);
+      // await this.batchArchiver.add(this.msgId);
     } catch (error) {
       await this.handleExecutionError(error);
     }

diff --git a/pkgs/edge-worker/supabase/functions/max_concurrency/index.ts b/pkgs/edge-worker/supabase/functions/max_concurrency/index.ts
@@ -8,9 +8,9 @@ console.log('EDGE_WORKER_DB_URL', EDGE_WORKER_DB_URL);
 const sql = postgres(EDGE_WORKER_DB_URL, { prepare: true });
 
 async function incrementSeq() {
-  await delay(100);
-  // const randTimeMs = Math.floor(Math.random() * 10);
-  // await delay(randTimeMs);
+  // await delay(1000);
+  const randTimeMs = Math.floor(Math.random() * 100 + 50);
+  await delay(randTimeMs);
   console.log(
     '[max_concurrency] last_val =',
     await sql`SELECT nextval('test_seq')`
@@ -19,6 +19,6 @@ async function incrementSeq() {
 
 EdgeWorker.start(incrementSeq, {
   queueName: 'max_concurrency',
-  maxConcurrent: 40,
+  maxConcurrent: 10,
   maxPgConnections: 4,
 });