From 170a3d3cf79a00f188558eec0ebf38b75c5ea920 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Sun, 19 Dec 2021 03:21:00 +0100 Subject: [PATCH] Use stream.finished for detecting end of streams There are some corner cases with streams that cause them to not emit finish close or end events which in turn means that code liseting for these events directly may end up hanging forever. Use stream.finished to detect end of streams which will handle these cases correctly. --- packages/create/create.js | 3 ++- packages/ctl/ctl.js | 14 +++++++++----- packages/lib/build_mod.js | 6 +++++- packages/lib/logging_utils.js | 14 ++++++++++++-- packages/slave/slave.js | 13 ++++++++----- test/lib/stream.js | 26 +++++++++++++++----------- 6 files changed, 51 insertions(+), 25 deletions(-) diff --git a/packages/create/create.js b/packages/create/create.js index b885f780..b72b6a8e 100755 --- a/packages/create/create.js +++ b/packages/create/create.js @@ -15,6 +15,7 @@ const yargs = require("yargs"); const { levels, logger, setLogLevel } = require("./logging"); let dev = false; const scriptExt = process.platform === "win32" ? ".cmd" : ""; +const finished = util.promisify(stream.finished); const availablePlugins = [ @@ -480,7 +481,7 @@ async function downloadLinuxServer() { const writeStream = fs.createWriteStream(tmpArchivePath); res.pipe(writeStream); - await events.once(res, "end"); + await finished(writeStream); await fs.rename(tmpArchivePath, archivePath); try { diff --git a/packages/ctl/ctl.js b/packages/ctl/ctl.js index 1c4ab74d..64e1ff15 100755 --- a/packages/ctl/ctl.js +++ b/packages/ctl/ctl.js @@ -16,6 +16,8 @@ const setBlocking = require("set-blocking"); const phin = require("phin"); const os = require("os"); const child_process = require("child_process"); +const stream = require("stream"); +const util = require("util"); const libLink = require("@clusterio/lib/link"); const libErrors = require("@clusterio/lib/errors"); @@ -30,6 +32,8 @@ const libHelpers = require("@clusterio/lib/helpers"); const libFactorio = require("@clusterio/lib/factorio"); const libFileOps = require("@clusterio/lib/file_ops"); +const finished = util.promisify(stream.finished); + function print(...content) { // eslint-disable-next-line no-console @@ -640,12 +644,12 @@ instanceCommands.add(new libCommand.Command({ stream: true, }); - let stream; + let writeStream; let tempFilename = args.save.replace(/(\.zip)?$/, ".tmp.zip"); while (true) { try { - stream = fs.createWriteStream(tempFilename, { flags: "wx" }); - await events.once(stream, "open"); + writeStream = fs.createWriteStream(tempFilename, { flags: "wx" }); + await events.once(writeStream, "open"); break; } catch (err) { if (err.code === "EEXIST") { @@ -655,8 +659,8 @@ instanceCommands.add(new libCommand.Command({ } } } - response.pipe(stream); - await events.once(stream, "finish"); + response.pipe(writeStream); + await finished(writeStream); let filename = await libFileOps.findUnusedName(".", args.save, ".zip"); await fs.rename(tempFilename, filename); diff --git a/packages/lib/build_mod.js b/packages/lib/build_mod.js index 13211900..3ac716c6 100644 --- a/packages/lib/build_mod.js +++ b/packages/lib/build_mod.js @@ -5,8 +5,12 @@ const fs = require("fs-extra"); const JSZip = require("jszip"); const klaw = require("klaw"); const path = require("path"); +const stream = require("stream"); +const util = require("util"); const yargs = require("yargs"); +const finished = util.promisify(stream.finished); + async function buildMod(args, info) { if (args.build) { @@ -42,7 +46,7 @@ async function buildMod(args, info) { let modPath = path.join(args.outputDir, `${modName}.zip`); console.log(`Writing ${modPath}`); let writeStream = zip.generateNodeStream().pipe(fs.createWriteStream(modPath)); - await events.once(writeStream, "finish"); + await finished(writeStream); } else { let modDir = path.join(args.outputDir, modName); diff --git a/packages/lib/logging_utils.js b/packages/lib/logging_utils.js index 6b2d3af4..37e818e7 100644 --- a/packages/lib/logging_utils.js +++ b/packages/lib/logging_utils.js @@ -8,6 +8,8 @@ const events = require("events"); const fs = require("fs-extra"); const path = require("path"); +const stream = require("stream"); +const util = require("util"); const winston = require("winston"); const Transport = require("winston-transport"); const { LEVEL, MESSAGE } = require("triple-beam"); @@ -19,6 +21,8 @@ const libFileOps = require("./file_ops"); const { levels, logger } = require("./logging"); const libStream = require("./stream"); +const finished = util.promisify(stream.finished); + /** * Format a parsed Factorio message with colors @@ -435,7 +439,13 @@ async function queryLog(logDirectory, filter, index) { } }); fileStream.pipe(lineStream); - await events.once(lineStream, "close"); + try { + await finished(lineStream); + } catch (err) { + if (log.length < filter.limit || err.code !== "ERR_STREAM_PREMATURE_CLOSE") { + throw err; + } + } if (log.length >= filter.limit) { break; @@ -456,7 +466,7 @@ async function migrateLogs(log, directory, pattern) { async function switchOutput(newOutputPath) { if (output) { output.end(); - await events.once(output, "close"); + await finished(output); } output = fs.createWriteStream(newOutputPath, { flags: "ax" }); diff --git a/packages/slave/slave.js b/packages/slave/slave.js index f33ae4dc..cbcb0098 100755 --- a/packages/slave/slave.js +++ b/packages/slave/slave.js @@ -20,6 +20,7 @@ const events = require("events"); const pidusage = require("pidusage"); const setBlocking = require("set-blocking"); const phin = require("phin"); +const stream = require("stream"); const util = require("util"); const version = require("./package").version; const winston = require("winston"); @@ -40,6 +41,8 @@ const { ConsoleTransport, levels, logger } = require("@clusterio/lib/logging"); const libLoggingUtils = require("@clusterio/lib/logging_utils"); const libHelpers = require("@clusterio/lib/helpers"); +const finished = util.promisify(stream.finished); + const instanceRconCommandDuration = new libPrometheus.Histogram( "clusterio_instance_rcon_command_duration_seconds", @@ -1478,11 +1481,11 @@ class Slave extends libLink.Link { let savesDir = path.join(instanceInfo.path, "saves"); let tempFilename = filename.replace(/(\.zip)?$/, ".tmp.zip"); - let stream; + let writeStream; while (true) { try { - stream = fs.createWriteStream(path.join(savesDir, tempFilename), { flags: "wx" }); - await events.once(stream, "open"); + writeStream = fs.createWriteStream(path.join(savesDir, tempFilename), { flags: "wx" }); + await events.once(writeStream, "open"); break; } catch (err) { if (err.code === "EEXIST") { @@ -1492,8 +1495,8 @@ class Slave extends libLink.Link { } } } - response.pipe(stream); - await events.once(stream, "finish"); + response.pipe(writeStream); + await finished(writeStream); filename = await libFileOps.findUnusedName(savesDir, filename, ".zip"); await fs.rename(path.join(savesDir, tempFilename), path.join(savesDir, filename)); diff --git a/test/lib/stream.js b/test/lib/stream.js index 721ed090..f85af99f 100644 --- a/test/lib/stream.js +++ b/test/lib/stream.js @@ -3,16 +3,20 @@ const assert = require("assert").strict; const events = require("events"); const fs = require("fs-extra"); const path = require("path"); +const stream = require("stream"); +const util = require("util"); const libStream = require("@clusterio/lib/stream"); +const finished = util.promisify(stream.finished); + describe("lib/stream", function() { describe("class LineSplitter", function() { function createSplitter(lines) { - let stream = new libStream.LineSplitter({ readableObjectMode: true }); - stream.on("data", line => lines.push(line.toString("utf-8"))); - return stream; + let lineStream = new libStream.LineSplitter({ readableObjectMode: true }); + lineStream.on("data", line => lines.push(line.toString("utf-8"))); + return lineStream; } it("should split three lines", function() { @@ -57,9 +61,9 @@ describe("lib/stream", function() { describe("class ReverseLineSplitter", function() { function createReverseSplitter(lines) { - let stream = new libStream.ReverseLineSplitter({ readableObjectMode: true }); - stream.on("data", line => lines.push(line.toString("utf-8"))); - return stream; + let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true }); + lineStream.on("data", line => lines.push(line.toString("utf-8"))); + return lineStream; } it("should split three lines", function() { @@ -107,16 +111,16 @@ describe("lib/stream", function() { let content = ""; for (let i = 0; i < 10; i++) { content += String(i).repeat(10); } await fs.outputFile(path.join("temp", "test", "reverse.txt"), content); - let stream = await libStream.createReverseReadStream( + let reverseStream = await libStream.createReverseReadStream( path.join("temp", "test", "reverse.txt"), { encoding: "utf8", highWaterMark: 10 } ); let index = 9; - stream.on("data", data => { + reverseStream.on("data", data => { assert.equal(data, String(index).repeat(10)); index -= 1; }); - await events.once(stream, "end"); + await finished(reverseStream); }); it("should reverse the lines of a file", async function() { @@ -125,7 +129,7 @@ describe("lib/stream", function() { ws.write(Buffer.from(`${i}\n`)); } ws.end(); - await events.once(ws, "finish"); + await finished(ws); let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true }); let index = 99999; @@ -139,7 +143,7 @@ describe("lib/stream", function() { index -= 1; }); rs.pipe(lineStream); - await events.once(lineStream, "end"); + await finished(lineStream); assert.equal(index, 0, "line check did not reach the end"); });