Clean up world-postgres LISTEN self-healing for upstream PR

.changeset/world-postgres-listen-self-healing.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Make stream delivery durable across `LISTEN`/`NOTIFY` interruptions: the dedicated `pg.Client` now reconnects with bounded exponential backoff (250 ms → 30 s), and `readFromStream` runs a periodic re-query of `streams WHERE chunk_id > lastChunkId` as a polling safety net for chunks delivered while the LISTEN socket was down. The poll interval is configurable via `PostgresWorldConfig.streamPollIntervalMs` (default 5000 ms; set to 0 to disable). Tracks vercel/workflow#1855.

packages/world-postgres/src/config.ts

@@ -1,5 +1,4 @@
 import type { Pool } from 'pg';
-import type { ListenAdapter } from './listen-adapter.js';
 
 type PgConnectionConfig =
   | { connectionString: string; maxPoolSize?: number; pool?: undefined }
@@ -14,11 +13,11 @@ export type PostgresWorldConfig = PgConnectionConfig & {
    */
   streamFlushIntervalMs?: number;
   /**
-   * Plug in an alternative LISTEN/NOTIFY transport for the streamer.
-   * Defaults to the bundled `pg`-backed adapter (with self-healing reconnect).
-   * See {@link ListenAdapter} for the contract; useful for swapping in
-   * `bun:sql`'s native listen (oven-sh/bun#29710) once it lands, or a
-   * different pub/sub backend entirely.
+   * How often (ms) `readFromStream` re-queries the `streams` table as a
+   * safety net for chunks delivered while the LISTEN client was reconnecting.
+   * Default is 5000. Set to 0 to disable polling entirely.
+   *
+   * See `CreateStreamerOptions.pollIntervalMs` for the full contract.
    */
-  listenAdapter?: ListenAdapter;
+  streamPollIntervalMs?: number;
 };

packages/world-postgres/src/index.ts

@@ -54,7 +54,9 @@ export function createWorld(
   const drizzle = createClient(pool);
   const queue = createQueue(config, pool);
   const storage = createStorage(drizzle);
-  const streamer = createStreamer(pool, drizzle, config.listenAdapter);
+  const streamer = createStreamer(pool, drizzle, {
+    pollIntervalMs: config.streamPollIntervalMs,
+  });
 
   return {
     specVersion: SPEC_VERSION_CURRENT,
@@ -80,10 +82,4 @@ export function createWorld(
 
 // Re-export schema for users who want to extend or inspect the database schema
 export type { PostgresWorldConfig } from './config.js';
-export {
-  createBunSqlListenAdapter,
-  createPgListenAdapter,
-  type ListenAdapter,
-  type ListenSubscription,
-} from './listen-adapter.js';
 export * from './drizzle/schema.js';

packages/world-postgres/src/queue.ts

@@ -256,9 +256,10 @@ export function createQueue(
       `${baseUrl}/.well-known/workflow/v1/${pathname}`,
       {
         method: 'POST',
+        duplex: 'half',
         headers,
         body,
-      }
+      } as any
     );
     const text = await response.text();