fix(web): stop AsyncQueue.fail from ending a parked consumer normally

bindings/web/packages/core/src/Foundation/AsyncQueue.ts

@@ -25,54 +25,55 @@
 /** Async queue with a single producer + single consumer (for-await). */
 export class AsyncQueue<T> implements AsyncIterable<T> {
   private buffer: T[] = [];
-  private resolveNext: ((v: IteratorResult<T>) => void) | null = null;
+  private wake: (() => void) | null = null;
   private done = false;
   private error: Error | null = null;
 
+  /**
+   * Wake a parked consumer. Producers only ever signal; `next()` re-reads the
+   * queue state itself, which is what keeps `fail()` from having to resolve a
+   * waiter it cannot reject. Mirrors `AsyncQueue` in
+   * `@runanywhere/proto-ts/streams/push`.
+   */
+  private signal(): void {
+    const wake = this.wake;
+    this.wake = null;
+    if (wake) wake();
+  }
+
   /** Producer: push the next value. Discarded if the queue is closed. */
   push(value: T): void {
     if (this.done) return;
-    if (this.resolveNext) {
-      const r = this.resolveNext;
-      this.resolveNext = null;
-      r({ value, done: false });
-    } else {
-      this.buffer.push(value);
-    }
+    this.buffer.push(value);
+    this.signal();
   }
 
   /** Producer: signal normal end-of-stream. Idempotent. */
   complete(): void {
     if (this.done) return;
     this.done = true;
-    if (this.resolveNext) {
-      const r = this.resolveNext;
-      this.resolveNext = null;
-      r({ value: undefined as unknown as T, done: true });
-    }
+    this.signal();
   }
 
   /** Producer: signal abnormal termination. Next consumer await throws. */
   fail(error: Error): void {
     if (this.done) return;
     this.done = true;
     this.error = error;
-    if (this.resolveNext) {
-      const r = this.resolveNext;
-      this.resolveNext = null;
-      r({ value: undefined as unknown as T, done: true });
-    }
+    this.signal();
   }
 
   [Symbol.asyncIterator](): AsyncIterator<T> {
     return {
-      next: (): Promise<IteratorResult<T>> => {
-        if (this.buffer.length > 0) {
-          return Promise.resolve({ value: this.buffer.shift()!, done: false });
+      next: async (): Promise<IteratorResult<T>> => {
+        for (;;) {
+          if (this.buffer.length > 0) {
+            return { value: this.buffer.shift()!, done: false };
+          }
+          if (this.error) throw this.error;
+          if (this.done) return { value: undefined as unknown as T, done: true };
+          await new Promise<void>((r) => { this.wake = r; });
         }
-        if (this.error) return Promise.reject(this.error);
-        if (this.done) return Promise.resolve({ value: undefined as unknown as T, done: true });
-        return new Promise((r) => { this.resolveNext = r; });
       },
     };
   }

bindings/web/packages/core/tests/unit/Foundation/AsyncQueue.test.ts

@@ -0,0 +1,53 @@
+/**
+ * AsyncQueue.test.ts
+ *
+ * `fail()` has to reach the consumer whether or not it is parked. A streaming
+ * consumer is parked on `next()` almost all the time (it drains tokens faster
+ * than inference produces them), so a failure that only surfaces on a later
+ * `next()` is a failure the caller never sees.
+ */
+
+import { describe, it, expect } from 'vitest';
+
+import { AsyncQueue } from '../../../src/Foundation/AsyncQueue.js';
+
+describe('AsyncQueue.fail', () => {
+  it('rejects a consumer that is already parked on next()', async () => {
+    const queue = new AsyncQueue<string>();
+    const iterator = queue[Symbol.asyncIterator]();
+
+    // Park first, then fail — the order a mid-stream inference error takes.
+    const parked = iterator.next();
+    queue.fail(new Error('inference exploded'));
+
+    await expect(parked).rejects.toThrow('inference exploded');
+  });
+
+  it('rejects a consumer that arrives after the failure', async () => {
+    const queue = new AsyncQueue<string>();
+    queue.fail(new Error('inference exploded'));
+
+    const iterator = queue[Symbol.asyncIterator]();
+    await expect(iterator.next()).rejects.toThrow('inference exploded');
+  });
+
+  it('drains buffered values before reporting the failure', async () => {
+    const queue = new AsyncQueue<string>();
+    queue.push('a');
+    queue.fail(new Error('inference exploded'));
+
+    const iterator = queue[Symbol.asyncIterator]();
+    await expect(iterator.next()).resolves.toEqual({ value: 'a', done: false });
+    await expect(iterator.next()).rejects.toThrow('inference exploded');
+  });
+
+  it('still ends normally on complete()', async () => {
+    const queue = new AsyncQueue<string>();
+    const iterator = queue[Symbol.asyncIterator]();
+
+    const parked = iterator.next();
+    queue.complete();
+
+    await expect(parked).resolves.toEqual({ value: undefined, done: true });
+  });
+});