self-hosted supabase guidde

Author: jumski

pkgs/edge-worker/src/core/BatchProcessor.ts

@@ -18,8 +18,14 @@ export class BatchProcessor<TMessage extends IMessage> {
   }
 
   async processBatch() {
+    const availableSlots = this.executionController.availableSlots;
+    if (availableSlots <= 0) {
+      await this.executionController.waitForSlot();
+      return;
+    }
+
     this.logger.polling();
-    const messageRecords = await this.poller.poll();
+    const messageRecords = await this.poller.poll(availableSlots);
 
     if (this.signal.aborted) {
       this.logger.info('Discarding messageRecords because worker is stopping');
@@ -28,10 +34,12 @@ export class BatchProcessor<TMessage extends IMessage> {
 
     this.logger.taskCount(messageRecords.length);
 
-    const startPromises = messageRecords.map((message) =>
-      this.executionController.start(message)
-    );
-    await Promise.all(startPromises);
+    for (const message of messageRecords) {
+      void this.executionController.start(message).catch(() => {
+        // ExecutionController already logs task failures; swallow here so
+        // refilling the next slot does not produce unhandled rejections.
+      });
+    }
   }
 
   async awaitCompletion() {

pkgs/edge-worker/src/core/ExecutionController.ts

@@ -11,6 +11,8 @@ export class ExecutionController<TMessage extends IMessage> {
   private promiseQueue: PromiseQueue;
   private signal: AbortSignal;
   private createExecutor: (record: TMessage, signal: AbortSignal) => IExecutor;
+  private readonly maxConcurrent: number;
+  private slotWaiters = new Set<() => void>();
 
   constructor(
     executorFactory: (record: TMessage, signal: AbortSignal) => IExecutor,
@@ -20,23 +22,59 @@ export class ExecutionController<TMessage extends IMessage> {
   ) {
     this.signal = abortSignal;
     this.createExecutor = executorFactory;
+    this.maxConcurrent = config.maxConcurrent;
     this.promiseQueue = newQueue(config.maxConcurrent);
     this.logger = logger;
   }
 
-  async start(record: TMessage) {
+  get availableSlots(): number {
+    return Math.max(0, this.maxConcurrent - this.promiseQueue.size());
+  }
+
+  start(record: TMessage) {
     const executor = this.createExecutor(record, this.signal);
 
     this.logger.debug(`Scheduling execution of task ${executor.msgId}`);
 
-    return await this.promiseQueue.add(async () => {
+    return this.promiseQueue.add(async () => {
       try {
         this.logger.debug(`Executing task ${executor.msgId}...`);
         await executor.execute();
         this.logger.debug(`Execution successful for ${executor.msgId}`);
       } catch (error) {
         this.logger.error(`Execution failed for ${executor.msgId}`, error);
         throw error;
+      } finally {
+        this.notifySlotWaiters();
+      }
+    });
+  }
+
+  async waitForSlot(): Promise<void> {
+    if (this.signal.aborted || this.availableSlots > 0) {
+      return;
+    }
+
+    await new Promise<void>((resolve) => {
+      const onAbort = () => {
+        cleanup();
+        resolve();
+      };
+      const onSlotFreed = () => {
+        cleanup();
+        resolve();
+      };
+      const cleanup = () => {
+        this.slotWaiters.delete(onSlotFreed);
+        this.signal.removeEventListener('abort', onAbort);
+      };
+
+      this.slotWaiters.add(onSlotFreed);
+      this.signal.addEventListener('abort', onAbort, { once: true });
+
+      if (this.signal.aborted || this.availableSlots > 0) {
+        cleanup();
+        resolve();
       }
     });
   }
@@ -50,4 +88,12 @@ export class ExecutionController<TMessage extends IMessage> {
     );
     await this.promiseQueue.done();
   }
+
+  private notifySlotWaiters() {
+    const waiters = [...this.slotWaiters];
+    this.slotWaiters.clear();
+    for (const waiter of waiters) {
+      waiter();
+    }
+  }
 }

pkgs/edge-worker/src/core/Worker.ts

@@ -6,6 +6,7 @@ export class Worker {
   private lifecycle: ILifecycle;
   private logger: Logger;
   private abortController = new AbortController();
+  private readonly requestShutdown?: () => void;
 
   private batchProcessor: IBatchProcessor;
   private sql: postgres.Sql;
@@ -16,12 +17,14 @@ export class Worker {
     batchProcessor: IBatchProcessor,
     lifecycle: ILifecycle,
     sql: postgres.Sql,
-    logger: Logger
+    logger: Logger,
+    requestShutdown?: () => void
   ) {
     this.sql = sql;
     this.lifecycle = lifecycle;
     this.batchProcessor = batchProcessor;
     this.logger = logger;
+    this.requestShutdown = requestShutdown;
   }
 
   startOnlyOnce(workerBootstrap: WorkerBootstrap) {
@@ -69,12 +72,13 @@ export class Worker {
       return;
     }
 
-    this.lifecycle.transitionToStopping();
+      this.lifecycle.transitionToStopping();
 
-    try {
-      // Signal deprecation (which includes "Stopped accepting new messages")
-      this.logDeprecation();
-      this.abortController.abort();
+      try {
+        // Signal deprecation (which includes "Stopped accepting new messages")
+        this.logDeprecation();
+        this.requestShutdown?.();
+        this.abortController.abort();
 
       try {
         this.logger.debug('-> Waiting for main loop to complete');

pkgs/edge-worker/src/core/types.ts

@@ -7,7 +7,7 @@ export type { Json } from '@pgflow/core';
 export type Supplier<T> = () => T;
 
 export interface IPoller<IMessage> {
-  poll(): Promise<IMessage[]>;
+  poll(limit?: number): Promise<IMessage[]>;
 }
 
 export interface IExecutor {

pkgs/edge-worker/src/flow/StepTaskPoller.ts

@@ -36,23 +36,24 @@ export class StepTaskPoller<TFlow extends AnyFlow>
     this.logger = logger;
   }
 
-  async poll(): Promise<StepTaskWithMessage<TFlow>[]> {
+  async poll(limit?: number): Promise<StepTaskWithMessage<TFlow>[]> {
     if (this.isAborted()) {
       this.logger.debug('Polling aborted, returning empty array');
       return [];
     }
 
     const workerId = this.getWorkerId();
+    const batchSize = limit ?? this.config.batchSize;
     this.logger.debug(
-      `Two-phase polling for flow tasks with batch size ${this.config.batchSize}, maxPollSeconds: ${this.config.maxPollSeconds}, pollIntervalMs: ${this.config.pollIntervalMs}`
+      `Two-phase polling for flow tasks with batch size ${batchSize}, maxPollSeconds: ${this.config.maxPollSeconds}, pollIntervalMs: ${this.config.pollIntervalMs}`
     );
 
     try {
       // Phase 1: Read messages from queue
       const messages = await this.adapter.readMessages(
         this.config.queueName,
         this.config.visibilityTimeout ?? 2,
-        this.config.batchSize,
+        batchSize,
         this.config.maxPollSeconds,
         this.config.pollIntervalMs
       );

pkgs/edge-worker/src/flow/createFlowWorker.ts

@@ -204,5 +204,11 @@ export function createFlowWorker<
   );
 
   // Return Worker
-  return new Worker(batchProcessor, lifecycle, sql, createLogger('Worker'));
+  return new Worker(
+    batchProcessor,
+    lifecycle,
+    sql,
+    createLogger('Worker'),
+    () => platformAdapter.requestShutdown()
+  );
 }

pkgs/edge-worker/src/platform/SupabasePlatformAdapter.ts

@@ -104,15 +104,17 @@ export class SupabasePlatformAdapter implements PlatformAdapter<SupabaseResource
   }
 
   async stopWorker(): Promise<void> {
-    // Trigger shutdown signal
-    this.abortController.abort();
-
-    // Cleanup resources
-    await this._platformResources.sql.end();
+    this.requestShutdown();
 
     if (this.worker) {
       await this.worker.stop();
     }
+
+    await this._platformResources.sql.end();
+  }
+
+  requestShutdown(): void {
+    this.abortController.abort();
   }
 
   createLogger(module: string): Logger {

pkgs/edge-worker/src/platform/types.ts

@@ -77,6 +77,11 @@ export interface PlatformAdapter<TResources extends Record<string, unknown> = Re
    */
   stopWorker(): Promise<void>;
 
+  /**
+   * Trigger the shared shutdown signal used by pollers, executors, and contexts.
+   */
+  requestShutdown(): void;
+
   /**
    * Get the connection string for the database
    * Returns undefined if sql was provided directly via config

pkgs/edge-worker/src/queue/ReadWithPollPoller.ts

@@ -22,15 +22,17 @@ export class ReadWithPollPoller<TPayload extends Json> {
     this.logger = logger;
   }
 
-  async poll(): Promise<PgmqMessageRecord<TPayload>[]> {
+  async poll(limit?: number): Promise<PgmqMessageRecord<TPayload>[]> {
     if (this.isAborted()) {
       this.logger.debug('Polling aborted, returning empty array');
       return [];
     }
 
-    this.logger.debug(`Polling queue '${this.queue.queueName}' with batch size ${this.config.batchSize}`);
+    const batchSize = limit ?? this.config.batchSize;
+
+    this.logger.debug(`Polling queue '${this.queue.queueName}' with batch size ${batchSize}`);
     const messages = await this.queue.readWithPoll(
-      this.config.batchSize,
+      batchSize,
       this.config.visibilityTimeout,
       this.config.maxPollSeconds,
       this.config.pollIntervalMs

pkgs/edge-worker/src/queue/createQueueWorker.ts

@@ -218,5 +218,11 @@ export function createQueueWorker<TPayload extends Json, TResources extends Reco
     createLogger('BatchProcessor')
   );
 
-  return new Worker(batchProcessor, lifecycle, sql, createLogger('Worker'));
+  return new Worker(
+    batchProcessor,
+    lifecycle,
+    sql,
+    createLogger('Worker'),
+    () => platformAdapter.requestShutdown()
+  );
 }