[world-vercel] Retry throttled stream WebSocket writes

.changeset/ws-stream-retry-429.md

@@ -0,0 +1,5 @@
+---
+"@workflow/world-vercel": patch
+---
+
+Retry throttled (429) stream WebSocket writes and closes after `Retry-After`, moving to HTTP when the socket closes or the wait budget runs out, and retry a close 5xx over HTTP instead of failing the writer.

docs/content/docs/v5/configuration/worlds.mdx

@@ -290,6 +290,7 @@ Platform-provided values such as `VERCEL_DEPLOYMENT_ID`, `VERCEL_PROJECT_ID`, an
 - Default: `http`
 - Experimental stream-write transport capability. Set to exactly `ws` to attempt `workflow-stream-ws/v1`. The server authoritatively accepts or declines each upgrade; a decline uses HTTP directly for that writer lifetime. Stream reads remain HTTP and demand-driven.
 - This is not tenant rollout policy or a package-version check. HTTP remains the compatibility path. `/websockets/v1` is independent of REST v2/v4 and persisted workflow `specVersion` values.
+- A throttled (429) socket write or close is retried after the server's `Retry-After`, as over HTTP. It moves to HTTP if the connection ends during the wait or the cumulative wait passes 30 seconds.
 
 ### `WORKFLOW_DISABLE_ANALYTICS_READS`
 

docs/content/worlds/v5/vercel.mdx

@@ -246,7 +246,7 @@ Experimental stream-write transport capability. Default: `http`. Set `WORKFLOW_S
 
 This setting does not own tenant rollout policy and does not infer server support from package versions. HTTP remains the compatibility path. The protocol route (`/websockets/v1`) is versioned independently from REST v2/v4 and persisted workflow `specVersion` values.
 
-Writes and close requests execute serially. The first operation waits up to 250 ms for an opted-in socket to open; this is an implementation-level measurement knob, not protocol semantics. If the budget expires, or the upgrade fails or is declined before acceptance, the writer uses HTTP without racing the same operation over both transports. After the socket accepts a write, a missing acknowledgement has an unknown outcome: the writer fails rather than replaying the write over HTTP and risking a duplicate append.
+Writes and close requests execute serially. The first operation waits up to 250 ms for an opted-in socket to open; this is an implementation-level measurement knob, not protocol semantics. If the budget expires, or the upgrade fails or is declined before acceptance, the writer uses HTTP without racing the same operation over both transports. After the socket accepts a write, a missing acknowledgement has an unknown outcome: the writer fails rather than replaying the write over HTTP and risking a duplicate append. A throttled (429) write or close did not apply, so the writer retries it after the server's `Retry-After`, as the HTTP writer does, on the socket if it stays open (or its replacement, after a drain) and otherwise over HTTP. Past 30 seconds of cumulative wait, the write moves to HTTP, whose own 429 retry applies. A close that fails with a 5xx is retried over HTTP, since close is idempotent.
 
 Before routine authentication expiry or server max duration, the server sends a v1 `drain` control. The client stops sending, lets an already-admitted request receive its reply, and reconnects after the server closes with code 1001. Authentication drains re-resolve a fresh bearer. A request that was sent but receives no reply before close still has an unknown outcome and is never replayed.
 

packages/world-vercel/src/__fixtures__/workflow-stream-ws-v1.json

@@ -55,6 +55,18 @@
       "bodyHex": "",
       "frameHex": "00000031b900046474797065656572726f7265726571496409667374617475731901ad676d65737361676569747279206c6174657200000000"
     },
+    {
+      "direction": "server_to_client",
+      "meta": {
+        "type": "error",
+        "reqId": 10,
+        "status": 429,
+        "message": "Too many requests",
+        "retryAfter": "2"
+      },
+      "bodyHex": "",
+      "frameHex": "00000046b900056474797065656572726f726572657149640a667374617475731901ad676d65737361676571546f6f206d616e792072657175657374736a72657472794166746572613200000000"
+    },
     {
       "direction": "server_to_client",
       "meta": {

packages/world-vercel/src/event-retry.ts

@@ -327,8 +327,23 @@ export function isRetryableEventPostError(err: unknown): boolean {
   return collectErrorMarkers(err).some((m) => TRANSIENT_CODES.has(m));
 }
 
-const sleep = (ms: number): Promise<void> =>
-  new Promise((resolve) => setTimeout(resolve, ms));
+/** Rejects with `signal.reason`, clearing the timer, when `signal` aborts. */
+const sleep = (ms: number, signal?: AbortSignal): Promise<void> =>
+  new Promise((resolve, reject) => {
+    if (signal?.aborted) {
+      reject(signal.reason);
+      return;
+    }
+    const onAbort = () => {
+      clearTimeout(timer);
+      reject(signal?.reason);
+    };
+    const timer = setTimeout(() => {
+      signal?.removeEventListener('abort', onAbort);
+      resolve();
+    }, ms);
+    signal?.addEventListener('abort', onAbort, { once: true });
+  });
 
 /** Gated like the rest of world-vercel's HTTP layer (`DEBUG=workflow:*`). Keeps
  * in-process retries visible during a latency/outage investigation; otherwise a
@@ -389,17 +404,19 @@ export interface EventPostRetryOptions {
 }
 
 /**
- * Per-POST throttle-wait accounting. The returned function waits out a 429's
- * `retryAfter` in-process (so the caller can re-attempt), or rethrows the
- * ThrottleError once the cumulative wait would exceed
- * THROTTLE_RETRY_BUDGET_MS, at which point the queue's delivery-counted
- * redelivery takes over.
+ * Cumulative 429 wait accounting for one logical write: an event POST, or a
+ * stream WebSocket write/close (`ws-stream-session.ts`). Both share this so a
+ * throttled write waits the same way whichever path it takes. Each call waits
+ * out one 429's `retryAfter` (seconds; DEFAULT_THROTTLE_RETRY_AFTER_SECONDS
+ * when absent) so the caller can re-attempt, or rethrows the 429 without
+ * waiting once the cumulative wait would exceed THROTTLE_RETRY_BUDGET_MS.
+ * Aborting `signal` ends a wait early by rejecting with its reason.
  */
-function createThrottleWaiter(
-  eventType: WorkflowEventType
-): (err: ThrottleError) => Promise<void> {
+export function createThrottleWaiter(
+  target: string
+): (err: { retryAfter?: number }, signal?: AbortSignal) => Promise<void> {
   let waitedMs = 0;
-  return async (err) => {
+  return async (err, signal) => {
     const waitMs =
       Math.max(
         1,
@@ -408,8 +425,8 @@ function createThrottleWaiter(
           : DEFAULT_THROTTLE_RETRY_AFTER_SECONDS
       ) * 1000;
     if (waitedMs + waitMs > THROTTLE_RETRY_BUDGET_MS) {
-      logRetry('throttle retry budget exhausted; surfacing to the queue', {
-        eventType,
+      logRetry('throttle retry budget exhausted', {
+        target,
         waitedMs,
         retryAfter: err.retryAfter,
       });
@@ -419,9 +436,9 @@ function createThrottleWaiter(
     // Visible (not debug-gated): this stalls the invocation for whole
     // seconds, which would otherwise read as unexplained latency.
     console.warn(
-      `[workflow] Throttled (429) writing ${eventType} event; retrying in-process in ${waitMs / 1000}s`
+      `[workflow] Throttled (429) writing ${target}; retrying in-process in ${waitMs / 1000}s`
     );
-    await sleep(waitMs);
+    await sleep(waitMs, signal);
   };
 }
 
@@ -460,7 +477,9 @@ export async function withEventPostRetry<T>(
   // Throttle waits draw on a shared per-POST budget instead of the transient
   // attempt counter, so a throttled write keeps its full transient-blip
   // allowance (and vice versa).
-  const waitOutThrottle = createThrottleWaiter(eventType);
+  // Beyond the budget the ThrottleError surfaces and the queue's
+  // (delivery-counted, backed-off) redelivery takes over.
+  const waitOutThrottle = createThrottleWaiter(`${eventType} event`);
   for (let attempt = 0; ; ) {
     try {
       return await fn();

packages/world-vercel/src/stream-ws-protocol-v1.test.ts

@@ -49,10 +49,10 @@ describe('workflow-stream-ws/v1 contract', () => {
     const fixtureBytes = await readFile(
       new URL('./__fixtures__/workflow-stream-ws-v1.json', import.meta.url)
     );
-    // Canonical source: workflow-server#876
+    // Canonical source: workflow-server#876, #1051
     // test/fixtures/workflow-stream-ws-v1.json
     expect(createHash('sha256').update(fixtureBytes).digest('hex')).toBe(
-      '908d0be8ca85661d6b0b666d7e6fc36c581cb70f9561def37c3c93c61bada009'
+      '485d36a30bac6dfee3f3d4f62dc23cd2ac866ee8d1826bce9cdca2c2394c0001'
     );
 
     expect(fixture.protocol).toBe(STREAM_WS_PROTOCOL_V1);

packages/world-vercel/src/stream-ws-protocol-v1.ts

@@ -68,12 +68,17 @@ export const StreamWsCloseAckMetaSchema = z.object({
   reqId: RequestIdSchema,
 });
 
-/** An absent request id makes the error connection-fatal. */
+/**
+ * An absent request id makes the error connection-fatal. A correlated 429 is
+ * the one retryable rejection: the request did not apply. `retryAfter` is the
+ * server handler's `Retry-After` header, verbatim, when it set one.
+ */
 export const StreamWsErrorMetaSchema = z.object({
   type: z.literal('error'),
   reqId: RequestIdSchema.optional(),
   status: z.number().int(),
   message: z.string().optional(),
+  retryAfter: z.string().optional(),
 });
 
 /**
@@ -96,8 +101,9 @@ export const StreamWsReplyMetaSchema = z.discriminatedUnion('type', [
 /*
  * v1 pipelining is ordered and fail-stop. The server executes requests serially
  * in receive order. Its first failed request prevents later queued writes or
- * closes from executing and poisons the writer. An unknown client outcome is
- * likewise never replayed. The whole WebSocket-message ceiling is deliberately
+ * closes from executing and poisons the writer. The exception is a correlated
+ * 429, which did not apply: a client with no other request outstanding may
+ * resend it after `retryAfter`. An unknown client outcome is never replayed. The whole WebSocket-message ceiling is deliberately
  * implementation- and measurement-defined beyond the per-chunk/count bounds.
  */