Fix FS drops connection if download is too slow

Only happens in HTTPS because seemingly somewhere in the stack up to
50mb of data is being buffered potentially triggering a timeout at a
higher level if not drained fast enough.

This commit also includes other streams related improvements.

lib/cwmp.ts

@@ -48,7 +48,7 @@ import {
   CpeFault,
 } from "./types";
 import { IncomingMessage, ServerResponse } from "http";
-import { Readable } from "stream";
+import { pipeline, Readable } from "stream";
 import { promisify } from "util";
 import { decode, encodingExists } from "iconv-lite";
 import { parseXmlDeclaration } from "./xml-parser";
@@ -1451,10 +1451,14 @@ async function listenerAsync(
   if (httpRequest.headers["content-encoding"]) {
     switch (httpRequest.headers["content-encoding"]) {
       case "gzip":
-        stream = httpRequest.pipe(zlib.createGunzip());
+        stream = pipeline(stream, zlib.createGunzip(), () => {
+          // Errors are also raised by the async iterator
+        });
         break;
       case "deflate":
-        stream = httpRequest.pipe(zlib.createInflate());
+        stream = pipeline(stream, zlib.createInflate(), () => {
+          // Errors are also raised by the async iterator
+        });
         break;
       default:
         httpResponse.writeHead(415, { Connection: "close" });
@@ -1463,34 +1467,17 @@ async function listenerAsync(
     }
   }
 
-  const body = await new Promise<Buffer>((resolve, reject) => {
+  const chunks: Buffer[] = [];
-    const chunks: Buffer[] = [];
+  try {
-    let bytes = 0;
+    for await (const chunk of stream) chunks.push(chunk);
+    // In Node versions prior to 15, the stream will not emit an error if the
+    // connection is closed before the stream is finished.
+    if (!stream.readableEnded) throw new Error("Connection closed");
+  } catch (err) {
+    return;
+  }
 
-    stream.on("data", (chunk) => {
+  const body = Buffer.concat(chunks);
-      chunks.push(chunk);
-      bytes += chunk.length;
-    });
-
-    stream.on("end", () => {
-      const _body = Buffer.allocUnsafe(bytes);
-      let offset = 0;
-      for (const chunk of chunks) {
-        chunk.copy(_body, offset, 0, chunk.length);
-        offset += chunk.length;
-      }
-      resolve(_body);
-    });
-
-    stream.on("error", reject);
-
-    httpRequest.on("aborted", () => {
-      resolve(null);
-    });
-  });
-
-  // Request aborted
-  if (!body) return;
 
   let sessionContext = currentSessions.get(httpRequest.socket);
 

lib/fs.ts

@@ -19,6 +19,7 @@
 
 import * as url from "url";
 import { IncomingMessage, ServerResponse } from "http";
+import { pipeline, Readable } from "stream";
 import { Collection, GridFSBucket } from "mongodb";
 import { onConnect } from "./db";
 import * as logger from "./logger";
@@ -34,21 +35,19 @@ onConnect(async (db) => {
 });
 
 const getFile = memoize(
-  (uploadDate: number, size: number, filename: string): Promise<Buffer> => {
+  async (
-    return new Promise((resolve, reject) => {
+    uploadDate: number,
-      const buffer = Buffer.allocUnsafe(size);
+    size: number,
-      let i = 0;
+    filename: string
-      const downloadStream = filesBucket.openDownloadStreamByName(filename);
+  ): Promise<Iterable<Buffer>> => {
-      downloadStream.on("error", reject);
+    const chunks: Buffer[] = [];
-      downloadStream.on("data", (data: Buffer) => {
+    const downloadStream = filesBucket.openDownloadStreamByName(filename);
-        data.copy(buffer, i);
+    for await (const chunk of downloadStream) chunks.push(chunk);
-        i += data.length;
+    // Node 12-14 don't throw error when stream is closed prematurely.
-      });
+    // However, we don't need to check for that since we're checking file size.
-      downloadStream.on("end", () => {
+    if (size !== chunks.reduce((a, b) => a + b.length, 0))
-        if (i !== size) reject(new Error("File size mismatch"));
+      throw new Error("File size mismatch");
-        else resolve(buffer);
+    return chunks;
-      });
-    });
   }
 );
 
@@ -81,7 +80,7 @@ export async function listener(
     return;
   }
 
-  const buffer = await getFile(
+  const chunks = await getFile(
     file["uploadDate"].getTime(),
     file.length,
     filename
@@ -92,6 +91,9 @@ export async function listener(
     "Content-Length": file.length,
   });
 
-  response.end(buffer);
+  pipeline(Readable.from(chunks), response, () => {
+    // Ignore errors resulting from client disconnecting
+  });
+
   logger.accessInfo(log);
 }

lib/ui/db.ts

@@ -23,7 +23,7 @@ import { onConnect, optimizeProjection } from "../db";
 import * as mongodbFunctions from "../mongodb-functions";
 import * as expression from "../common/expression";
 import { QueryOptions, Expression } from "../types";
-import { Readable, Writable } from "stream";
+import { Readable } from "stream";
 import { minimize } from "../common/boolean-expression";
 
 const RESOURCE_COLLECTION = {
@@ -342,10 +342,19 @@ export function putFile(
         metadata: metadata,
       }
     );
+
+    contentStream.on("close", () => {
+      if (!contentStream.readableEnded)
+        uploadStream.destroy(new Error("Stream closed prematurely"));
+    });
+
+    contentStream.on("error", (err) => {
+      uploadStream.destroy(err);
+    });
+
     uploadStream.on("error", reject);
-    contentStream.on("error", reject);
     uploadStream.on("finish", resolve);
-    contentStream.pipe(uploadStream as Writable);
+    contentStream.pipe(uploadStream);
   });
 }