[bug] Refactor pipeStreams to prevent infinite printFile calls

Common/sources/storage/storage-fs.js

@@ -36,6 +36,7 @@ const { cp, rm, mkdir } = require('fs/promises');
 const { stat, readFile, writeFile } = require('fs/promises');
 var path = require('path');
 var utils = require("../utils");
+const { pipeline } = require('node:stream/promises');
 
 
 function getFilePath(storageCfg, strPath) {
@@ -76,7 +77,7 @@ async function putObject(storageCfg, strPath, buffer, contentLength) {
     await writeFile(fsPath, buffer);
   } else {
     let writable = await utils.promiseCreateWriteStream(fsPath);
-    await utils.pipeStreams(buffer, writable, true);
+    await pipeline(buffer, writable);
   }
 }
 

Common/sources/utils.js

@@ -926,24 +926,22 @@ function changeOnlyOfficeUrl(inputUrl, strPath, optFilename) {
   return inputUrl + constants.ONLY_OFFICE_URL_PARAM + '=' + constants.OUTPUT_NAME + path.extname(optFilename || strPath);
 }
 exports.changeOnlyOfficeUrl = changeOnlyOfficeUrl;
-function pipeStreams(from, to, isEnd) {
-  return new Promise(function(resolve, reject) {
-    from.pipe(to, {end: isEnd});
-    from.on('end', function() {
-      resolve();
-    });
-    from.on('error', function(e) {
-      reject(e);
-    });
+/**
+ * Pipe streams for HTTP responses, swallowing client abort errors.
+ * @param {NodeJS.ReadableStream} from - source stream
+ * @param {NodeJS.WritableStream} to - HTTP response stream
+ * @returns {Promise<void>}
+ */
+function pipeHttpStreams(from, to) {
+  return pipeline(from, to).catch((err) => {
+    // Treat client abort/connection reset as non-fatal to keep "End" logs parity.
+    if (err && (err.code === 'ERR_STREAM_PREMATURE_CLOSE' || err.code === 'ECONNRESET' || err.code === 'EPIPE')) {
+      return;
+    }
+    throw err;
   });
 }
-exports.pipeStreams = pipeStreams;
-function* pipeFiles(from, to) {
-  var fromStream = yield promiseCreateReadStream(from);
-  var toStream = yield promiseCreateWriteStream(to);
-  yield pipeStreams(fromStream, toStream, true);
-}
-exports.pipeFiles = co.wrap(pipeFiles);
+exports.pipeHttpStreams = pipeHttpStreams;
 function checkIpFilter(ctx, ipString, opt_hostname) {
   const tenIpFilterRules = ctx.getCfg('services.CoAuthoring.ipfilter.rules', cfgIpFilterRules);
 

DocService/sources/canvasservice.js

@@ -1638,7 +1638,7 @@ exports.printFile = function(req, res) {
       res.setHeader('Content-Disposition', utils.getContentDisposition(filename, null, constants.CONTENT_DISPOSITION_INLINE));
       res.setHeader('Content-Length', streamObj.contentLength);
       res.setHeader('Content-Type', 'application/pdf');
-      yield utils.pipeStreams(streamObj.readStream, res, true);
+      yield utils.pipeHttpStreams(streamObj.readStream, res);
 
       if (clientStatsD) {
         clientStatsD.timing('coauth.printFile', new Date() - startDate);

DocService/sources/converterservice.js

@@ -607,7 +607,7 @@ function convertTo(req, res) {
         res.setHeader('Content-Disposition', utils.getContentDisposition(filename, null, constants.CONTENT_DISPOSITION_INLINE));
         res.setHeader('Content-Length', streamObj.contentLength);
         res.setHeader('Content-Type', mime.getType(filename));
-        yield utils.pipeStreams(streamObj.readStream, res, true);
+        yield utils.pipeHttpStreams(streamObj.readStream, res);
       } else {
         ctx.logger.error('convert-to error status:%j', status);
         res.sendStatus(400);

FileConverter/sources/converter.js

@@ -41,6 +41,7 @@ var spawnAsync = require('@expo/spawn-async');
 const bytes = require('bytes');
 const lcid = require('lcid');
 const ms = require('ms');
+const { pipeline } = require('node:stream/promises');
 
 var commonDefines = require('./../../Common/sources/commondefines');
 var storage = require('./../../Common/sources/storage/storage-base');
@@ -566,23 +567,20 @@ function* concatFiles(source, template) {
   //concatenate EditorN.ext parts in Editor.ext
   let list = yield utils.listObjects(source, true);
   list.sort(utils.compareStringByLength);
-  let writeStreams = {};
+  const createdTargets = new Set();
   for (let i = 0; i < list.length; ++i) {
     let file = list[i];
     if (file.match(new RegExp(`${template}\\d+\\.`))) {
       let target = file.replace(new RegExp(`(${template})\\d+(\\..*)`), '$1$2');
-      let writeStream = writeStreams[target];
-      if (!writeStream) {
-        writeStream = yield utils.promiseCreateWriteStream(target);
-        writeStreams[target] = writeStream;
-      }
+      const isFirst = !createdTargets.has(target);
+      const writeOpts = isFirst ? undefined : { flags: 'a' };
+      let writeStream = yield utils.promiseCreateWriteStream(target, writeOpts);
       let readStream = yield utils.promiseCreateReadStream(file);
-      yield utils.pipeStreams(readStream, writeStream, false);
-    }
-  }
-  for (let i in writeStreams) {
-    if (writeStreams.hasOwnProperty(i)) {
-      writeStreams[i].end();
+      // Use raw pipeline for file-to-file operations to surface real errors
+      yield pipeline(readStream, writeStream);
+      if (isFirst) {
+        createdTargets.add(target);
+      }
     }
   }
 }