Handle backpressure when streaming file responses

scripts/lib/lactoserv/dev

@@ -65,7 +65,7 @@ function target-build {
         && env-minimize \
         && lib node-project build-main-module \
             --out="${outDir}" --modules-dirs="${srcDir}" \
-            --runner-script=run --runner-versions[]='20 21 22 23 24' \
+            --runner-script=run --runner-versions[]='20 21 22 23 24 25' \
             lactoserv
     )
 }

src/net-util/export/FullResponse.js

@@ -775,8 +775,12 @@ export class FullResponse extends BaseResponse {
           throw new Error(`File changed length during response processing: ${path}`);
         }
 
-        // TODO: Handle `drain` requests (based on return value of `write()`).
-        res.write(bytesRead === buffer.length ? buffer : buffer.subarray(0, bytesRead));
+        const chunk = bytesRead === buffer.length ? buffer : buffer.subarray(0, bytesRead);
+        const canContinue = res.write(chunk);
+
+        if (!canContinue) {
+          await FullResponse.#waitForDrain(res);
+        }
 
         at += bytesRead;
         remaining -= bytesRead;
@@ -1030,6 +1034,58 @@ export class FullResponse extends BaseResponse {
     return buffer;
   }
 
+  /**
+   * Waits for the 'drain' event on a response, or returns immediately if the
+   * response is already closed/destroyed.
+   *
+   * @param {TypeNodeResponse} res The low-level response object.
+   * @returns {boolean} `true` if drained successfully, `false` if the response
+   *   was closed/destroyed.
+   * @throws {Error} Any error reported by the response.
+   */
+  static async #waitForDrain(res) {
+    if (res.closed || res.destroyed) {
+      const error = res.errored;
+      if (error) {
+        throw (error instanceof Error) ? error : new Error(`Response error: ${error}`);
+      }
+      return false;
+    }
+
+    const resultMp = new ManualPromise();
+
+    const onDrain = () => {
+      cleanup();
+      resultMp.resolve(true);
+    };
+
+    const onError = (error) => {
+      cleanup();
+      if (error) {
+        resultMp.reject((error instanceof Error) ? error : new Error(`Response error: ${error}`));
+      } else {
+        resultMp.resolve(false);
+      }
+    };
+
+    const onClose = () => {
+      cleanup();
+      resultMp.resolve(false);
+    };
+
+    const cleanup = () => {
+      res.removeListener('drain', onDrain);
+      res.removeListener('error', onError);
+      res.removeListener('close', onClose);
+    };
+
+    res.once('drain', onDrain);
+    res.once('error', onError);
+    res.once('close', onClose);
+
+    return resultMp.promise;
+  }
+
   /**
    * Cleans up response headers for logging.
    *