Stream iterator throws an ApiError if fetch fails

Author: aron

index.test.ts

@@ -1189,12 +1189,12 @@ describe("Replicate client", () => {
   // Continue with tests for other methods
 
   describe("Stream", () => {
-    function createStream(body: string | NodeJS.ReadableStream) {
+    function createStream(body: string | NodeJS.ReadableStream, status = 200) {
       const streamEndpoint = "https://stream.replicate.com";
       nock(streamEndpoint)
         .get("/fake_stream")
         .matchHeader("Accept", "text/event-stream")
-        .reply(200, body);
+        .reply(status, body);
 
       return new Stream({ url: `${streamEndpoint}/fake_stream`, fetch });
     }
@@ -1468,5 +1468,41 @@ describe("Replicate client", () => {
       });
       expect(await iterator.next()).toEqual({ done: true });
     });
+
+    test("an error event in the stream raises an exception", async () => {
+      const stream = createStream(
+        `
+        event: output
+        id: EVENT_1
+        data: hello world
+
+        event: error
+        id: EVENT_2
+        data: An unexpected error occurred
+
+      `
+          .trim()
+          .replace(/^[ ]+/gm, "")
+      );
+
+      const iterator = stream[Symbol.asyncIterator]();
+      expect(await iterator.next()).toEqual({
+        done: false,
+        value: { event: "output", id: "EVENT_1", data: "hello world" },
+      });
+      await expect(iterator.next()).rejects.toThrowError(
+        "An unexpected error occurred"
+      );
+      expect(await iterator.next()).toEqual({ done: true });
+    });
+
+    test("an error when fetching the stream raises an exception", async () => {
+      const stream = createStream("{}", 500);
+      const iterator = stream[Symbol.asyncIterator]();
+      await expect(iterator.next()).rejects.toThrowError(
+        "Request to https://stream.replicate.com/fake_stream failed with status 500 Internal Server Error: {}."
+      );
+      expect(await iterator.next()).toEqual({ done: true });
+    });
   });
 });

lib/stream.js

@@ -1,4 +1,7 @@
 // Attempt to use readable-stream if available, attempt to use the built-in stream module.
+
+const ApiError = require("./error");
+
 let Readable;
 try {
   Readable = require("readable-stream").Readable;
@@ -107,15 +110,23 @@ class Stream extends Readable {
   }
 
   async *[Symbol.asyncIterator]() {
-    const response = await this.fetch(this.url, {
+    const init = {
       ...this.options,
       headers: {
         Accept: "text/event-stream",
       },
-    });
+    };
+    const response = await this.fetch(this.url, init);
 
     if (!response.ok) {
-      throw new Error();
+      // The cross-fetch shim doesn't accept Request objects so we create one here.
+      const request = new Request(this.url, init);
+      const text = await response.text();
+      throw new ApiError(
+        `Request to ${request.url} failed with status ${response.status} ${response.statusText}: ${text}.`,
+        request,
+        response
+      );
     }
 
     let partialChunk = "";