fix(csv-parse): propagate date cast errors through streams

packages/csv-parse/lib/api/index.js

@@ -777,8 +777,12 @@ const transform = function (original_options = {}) {
       if (this.__isFloat(field)) {
         return [undefined, Number(field)];
       } else if (this.options.cast_date !== false) {
-        const info = this.__infoField();
-        return [undefined, this.options.cast_date.call(null, field, info)];
+        try {
+          const info = this.__infoField();
+          return [undefined, this.options.cast_date.call(null, field, info)];
+        } catch (err) {
+          return [err];
+        }
       }
       return [undefined, field];
     },

packages/csv-parse/lib/stream.js

@@ -19,13 +19,13 @@ const parse = (opts) => {
       transform(chunk) {
         const error = api.parse(chunk, false, enqueue, terminate);
         if (error) {
-          controller.error(error);
+          throw error;
         }
       },
       flush() {
         const error = api.parse(undefined, true, enqueue, terminate);
         if (error) {
-          controller.error(error);
+          throw error;
         }
       },
     },

packages/csv-parse/test/api.web_stream.ts

@@ -1,8 +1,62 @@
 import "should";
+import assert from "node:assert/strict";
 import { parse as parseStream } from "../lib/stream.js";
+import type { Options } from "../lib/stream.js";
 import { CsvError } from "../lib/index.js";
 
 describe("API Web Stream", function () {
+  for (const castOption of ["cast", "cast_date"] as const) {
+    for (const phase of ["write", "close"]) {
+      it(`rejects ${phase} with the original ${castOption} error`, async function () {
+        const error = new Error("conversion failed");
+        const options: Options = { cast: true };
+        options[castOption] = () => {
+          throw error;
+        };
+        const stream = parseStream(options);
+        const writer = stream.writable.getWriter();
+        const reader = stream.readable.getReader();
+        const readError = reader.read().catch((err) => err);
+        const closedError = writer.closed.catch((err) => err);
+        let operation;
+        if (phase === "write") {
+          operation = writer.write(Buffer.from("bad,next\n"));
+        } else {
+          await writer.write(Buffer.from("bad"));
+          operation = writer.close();
+        }
+        await assert.rejects(operation, (err) => err === error);
+        assert.equal(await readError, error);
+        assert.equal(await closedError, error);
+      });
+    }
+  }
+
+  for (const phase of ["write", "close"]) {
+    it(`rejects ${phase} with the original CSV syntax error`, async function () {
+      const stream = parseStream();
+      const writer = stream.writable.getWriter();
+      const reader = stream.readable.getReader();
+      const readError = reader.read().catch((err) => err);
+      const closedError = writer.closed.catch((err) => err);
+      let operation;
+      if (phase === "write") {
+        operation = writer.write(Buffer.from('bad"quote,next\n'));
+      } else {
+        await writer.write(Buffer.from('"unterminated'));
+        operation = writer.close();
+      }
+      const error = await readError;
+      await assert.rejects(operation, (err) => err === error);
+      assert.ok(error instanceof CsvError);
+      assert.equal(
+        error.code,
+        phase === "write" ? "INVALID_OPENING_QUOTE" : "CSV_QUOTE_NOT_CLOSED",
+      );
+      assert.equal(await closedError, error);
+    });
+  }
+
   describe("stream/web/TransformStream", function () {
     it("simple parse", async function () {
       const stream = parseStream();

packages/csv-parse/test/option.cast_date.errors.js

@@ -0,0 +1,48 @@
+import assert from "node:assert/strict";
+import { parse } from "../lib/index.js";
+import { parse as parseSync } from "../lib/sync.js";
+
+describe("Option `cast_date` errors", function () {
+  for (const input of ["bad,next\n", "good,bad", "good,date\nbad,next\n"]) {
+    for (const skip_records_with_error of [false, true]) {
+      it(`reports an error from ${JSON.stringify(input)} with skip=${skip_records_with_error}`, async function () {
+        const error = new Error("date conversion failed");
+        const callbackErrors = [];
+        const streamErrors = [];
+        const parser = parse(
+          {
+            cast: true,
+            skip_records_with_error,
+            cast_date(value) {
+              if (value === "bad") throw error;
+              return value;
+            },
+          },
+          (err) => callbackErrors.push(err),
+        );
+        parser.on("error", (err) => streamErrors.push(err));
+        const closed = new Promise((resolve) => parser.on("close", resolve));
+        assert.doesNotThrow(() => parser.end(input));
+        await closed;
+        assert.deepEqual(callbackErrors, [error]);
+        assert.deepEqual(streamErrors, [error]);
+      });
+    }
+  }
+
+  it("continues to throw the original error from the sync API", function () {
+    const error = new Error("date conversion failed");
+    let thrown;
+    try {
+      parseSync("bad", {
+        cast: true,
+        cast_date() {
+          throw error;
+        },
+      });
+    } catch (err) {
+      thrown = err;
+    }
+    assert.equal(thrown, error);
+  });
+});