Merge pull request #17378 from overleaf/td-stream-buffer-consolidation

Stream and buffer consolidation

GitOrigin-RevId: 284f411e6403e190d2dda3d9ebf806b1935b3949

services/docstore/app/js/DocArchiveManager.js

@@ -8,6 +8,7 @@ const { ReadableString } = require('@overleaf/stream-utils')
 const RangeManager = require('./RangeManager')
 const PersistorManager = require('./PersistorManager')
 const pMap = require('p-map')
+const { streamToBuffer } = require('./StreamToBuffer').promises
 const { BSON } = require('mongodb-legacy')
 
 const PARALLEL_JOBS = Settings.parallelArchiveJobs
@@ -136,7 +137,7 @@ async function getDoc(projectId, docId) {
     key
   )
   stream.resume()
-  const buffer = await _streamToBuffer(projectId, docId, stream)
+  const buffer = await streamToBuffer(projectId, docId, stream)
   const md5 = crypto.createHash('md5').update(buffer).digest('hex')
   if (sourceMd5 !== md5) {
     throw new Errors.Md5MismatchError('md5 mismatch when downloading doc', {
@@ -187,34 +188,6 @@ async function destroyProject(projectId) {
   await Promise.all(tasks)
 }
 
-async function _streamToBuffer(projectId, docId, stream) {
-  const chunks = []
-  let size = 0
-  let logged = false
-  const logIfTooLarge = finishedReading => {
-    if (size <= Settings.max_doc_length) return
-    // Log progress once and then again at the end.
-    if (logged && !finishedReading) return
-    logger.warn(
-      { projectId, docId, size, finishedReading },
-      'potentially large doc pulled down from gcs'
-    )
-    logged = true
-  }
-  return await new Promise((resolve, reject) => {
-    stream.on('data', chunk => {
-      size += chunk.byteLength
-      logIfTooLarge(false)
-      chunks.push(chunk)
-    })
-    stream.on('error', reject)
-    stream.on('end', () => {
-      logIfTooLarge(true)
-      resolve(Buffer.concat(chunks))
-    })
-  })
-}
-
 function _deserializeArchivedDoc(buffer) {
   const doc = JSON.parse(buffer)
 

services/docstore/app/js/StreamToBuffer.js

@@ -0,0 +1,28 @@
+const { LoggerStream, WritableBuffer } = require('@overleaf/stream-utils')
+const Settings = require('@overleaf/settings')
+const logger = require('@overleaf/logger/logging-manager')
+const { pipeline } = require('node:stream/promises')
+const { callbackify } = require('node:util')
+
+module.exports = {
+  streamToBuffer: callbackify(streamToBuffer),
+  promises: {
+    streamToBuffer,
+  },
+}
+
+async function streamToBuffer(projectId, docId, stream) {
+  const loggerTransform = new LoggerStream(
+    Settings.max_doc_length,
+    (size, isFlush) => {
+      logger.warn(
+        { projectId, docId, size, finishedReading: isFlush },
+        'potentially large doc pulled down from gcs'
+      )
+    }
+  )
+
+  const buffer = new WritableBuffer()
+  await pipeline(stream, loggerTransform, buffer)
+  return buffer.contents()
+}