From: Ardis Lu <ardisglu@gmail.com>
Date: Fri, 4 Jul 2025 05:02:25 +0000 (-0700)
Subject: Increase `highWaterMark` for `CSVWriter`'s queuing strategy
X-Git-Tag: v0.1.2~4
X-Git-Url: https://git.ardis.lu/?a=commitdiff_plain;h=c6eeb119fcf503aa030839c6ecfe55394bcc2bee;p=csvio

Increase `highWaterMark` for `CSVWriter`'s queuing strategy

This change improves the 100MB Write benchmark from ~9s to ~7s.

Bumping `highWaterMark` to >1 also changes the stream's interaction with `AbortController`, so now the chunk processed during an abort is included in the write.

I believe this behavior has to do with the task queue order of the callback for the `AbortSignal`'s `abort` event - so now with multiple chunks in the queue, the last chunk is written before the `abort` event is processed. I was unable to force this writing code to occur after the `abort` event in a deterministic way.

This situation is a rare edge case so letting it through.
---

diff --git a/examples/data/ex1_7-out.csv b/examples/data/ex1_7-out.csv
index 0e472ca..f14cd79 100644
--- a/examples/data/ex1_7-out.csv
+++ b/examples/data/ex1_7-out.csv
@@ -1,3 +1,4 @@
 Header A
 abc
 def
+abort
diff --git a/examples/ex1_7.js b/examples/ex1_7.js
index dd2a643..abfadd3 100644
--- a/examples/ex1_7.js
+++ b/examples/ex1_7.js
@@ -1,8 +1,8 @@
 // ## Example 1.7: Abort a transformation
 // 
 // Pass an [`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal) to `.pipeTo()` to gracefully
-// abort a transformation. All transformations enqueued up until the row causing the abort (but *not* including the abort
-// row itself) will be saved to disk.
+// abort a transformation. All transformations enqueued up until the row causing the abort (including the abort row
+// itself) will be saved to disk.
 // 
 // Aborting throws an error, so you must also add a `.catch()` to handle the abort.
 
diff --git a/src/core.js b/src/core.js
index c76004f..8bbb274 100644
--- a/src/core.js
+++ b/src/core.js
@@ -509,7 +509,7 @@ export class CSVWriter extends WritableStream {
       async abort() {
         await this.close();
       }
-    });
+    }, new CountQueuingStrategy({ highWaterMark: 4 })); // "4" picked based on manual benchmarks
 
     this.#status = status;
   }