Merge pull request #12916 from overleaf/bg-move-stream-buffer-code-to-library

move stream-related code to separate  `@overleaf/stream-utils` library

GitOrigin-RevId: a79a873109b927b4fc0ae36f47d5c67e0df58041

libraries/stream-utils/.dockerignore

@@ -0,0 +1 @@
+node_modules/

libraries/stream-utils/.gitignore

@@ -0,0 +1,3 @@
+
+# managed by monorepo$ bin/update_build_scripts
+.npmrc

libraries/stream-utils/.mocharc.json

@@ -0,0 +1,5 @@
+{
+  "ui": "bdd",
+  "recursive": "true",
+  "reporter": "spec"
+}

libraries/stream-utils/.nvmrc

@@ -0,0 +1 @@
+16.17.1

libraries/stream-utils/Dockerfile

@@ -0,0 +1,5 @@
+FROM node:16.17.1
+
+WORKDIR /app
+
+USER node

libraries/stream-utils/buildscript.txt

@@ -0,0 +1,9 @@
+stream-utils
+--dependencies=None
+--docker-repos=gcr.io/overleaf-ops
+--env-add=
+--env-pass-through=
+--is-library=True
+--node-version=16.17.1
+--public-repo=False
+--script-version=4.1.0

libraries/stream-utils/index.js

@@ -0,0 +1,158 @@
+const { Writable, Readable, PassThrough, Transform } = require('stream')
+
+/**
+ * A writable stream that stores all data written to it in a node Buffer.
+ * @extends stream.Writable
+ * @example
+ * const { WritableBuffer } = require('@overleaf/stream-utils')
+ * const bufferStream = new WritableBuffer()
+ * bufferStream.write('hello')
+ * bufferStream.write('world')
+ * bufferStream.end()
+ * bufferStream.contents().toString() // 'helloworld'
+ */
+class WritableBuffer extends Writable {
+  constructor(options) {
+    super(options)
+    this._buffers = []
+    this._size = 0
+  }
+
+  _write(chunk, encoding, callback) {
+    this._buffers.push(chunk)
+    this._size += chunk.length
+    callback()
+  }
+
+  _final(callback) {
+    callback()
+  }
+
+  size() {
+    return this._size
+  }
+
+  getContents() {
+    return Buffer.concat(this._buffers)
+  }
+
+  contents() {
+    return Buffer.concat(this._buffers)
+  }
+}
+
+/**
+ * A readable stream created from a string.
+ * @extends stream.Readable
+ * @example
+ * const { ReadableString } = require('@overleaf/stream-utils')
+ * const stringStream = new ReadableString('hello world')
+ * stringStream.on('data', chunk => console.log(chunk.toString()))
+ * stringStream.on('end', () => console.log('done'))
+ */
+class ReadableString extends Readable {
+  constructor(string, options) {
+    super(options)
+    this._string = string
+  }
+
+  _read(size) {
+    this.push(this._string)
+    this.push(null)
+  }
+}
+
+class SizeExceededError extends Error {}
+
+/**
+ * Limited size stream which will emit a SizeExceededError if the size is exceeded
+ * @extends stream.Transform
+ */
+class LimitedStream extends Transform {
+  constructor(maxSize) {
+    super()
+    this.maxSize = maxSize
+    this.size = 0
+  }
+
+  _transform(chunk, encoding, callback) {
+    this.size += chunk.byteLength
+    if (this.size > this.maxSize) {
+      callback(
+        new SizeExceededError(
+          `exceeded stream size limit of ${this.maxSize}: ${this.size}`
+        )
+      )
+    } else {
+      callback(null, chunk)
+    }
+  }
+}
+
+class AbortError extends Error {}
+
+/**
+ * TimeoutStream which will emit an AbortError if it exceeds a user specified timeout
+ * @extends stream.PassThrough
+ */
+class TimeoutStream extends PassThrough {
+  constructor(timeout) {
+    super()
+    this.t = setTimeout(() => {
+      this.destroy(new AbortError('stream timed out'))
+    }, timeout)
+  }
+
+  _final(callback) {
+    clearTimeout(this.t)
+    callback()
+  }
+}
+
+/**
+ * LoggerStream which will call the provided logger function when the stream exceeds a user specified limit. It will call the provided function again when flushing the stream and it exceeded the user specified limit before.
+* @extends stream.Transform
+ */
+class LoggerStream extends Transform {
+  /**
+   * Constructor.
+   * @param {number} maxSize
+   * @param {function(currentSizeOfStream: number, isFlush: boolean)} fn
+   * @param {Object?} options optional options for the Transform stream
+   */
+  constructor(maxSize, fn, options) {
+    super(options)
+    this.fn = fn
+    this.size = 0
+    this.maxSize = maxSize
+    this.logged = false
+  }
+
+  _transform(chunk, encoding, callback) {
+    this.size += chunk.byteLength
+    if (this.size > this.maxSize && !this.logged) {
+      this.fn(this.size)
+      this.logged = true
+    }
+    callback(null, chunk)
+  }
+
+  _flush(callback) {
+    if (this.size > this.maxSize) {
+      this.fn(this.size, true)
+    }
+    callback()
+  }
+}
+
+// Export our classes
+
+module.exports = {
+  WritableBuffer,
+  ReadableString,
+  LoggerStream,
+  LimitedStream,
+  TimeoutStream,
+  SizeExceededError,
+  AbortError,
+}

libraries/stream-utils/package.json

@@ -0,0 +1,22 @@
+{
+  "name": "@overleaf/stream-utils",
+  "version": "0.1.0",
+  "description": "stream handling utilities",
+  "main": "index.js",
+  "scripts": {
+    "test": "npm run lint && npm run format && npm run test:unit",
+    "test:unit": "mocha",
+    "lint": "eslint --max-warnings 0 --format unix .",
+    "lint:fix": "eslint --fix .",
+    "format": "prettier --list-different $PWD/'**/*.js'",
+    "format:fix": "prettier --write $PWD/'**/*.js'",
+    "test:ci": "npm run test:unit"
+  },
+  "author": "Overleaf (https://www.overleaf.com)",
+  "license": "AGPL-3.0-only",
+  "devDependencies":{
+    "chai": "^4.3.6",
+    "chai-as-promised": "^7.1.1",
+    "mocha": "^10.2.0"
+  }
+}

libraries/stream-utils/test/unit/LimitedStreamTests.js

@@ -0,0 +1,30 @@
+const { expect } = require('chai')
+const { LimitedStream, SizeExceededError } = require('../../index')
+
+describe('LimitedStream', function () {
+  it('should emit an error if the stream size exceeds the limit', function (done) {
+    const maxSize = 10
+    const limitedStream = new LimitedStream(maxSize)
+    limitedStream.on('error', err => {
+      expect(err).to.be.an.instanceOf(SizeExceededError)
+      done()
+    })
+    limitedStream.write(Buffer.alloc(maxSize + 1))
+  })
+
+  it('should pass through data if the stream size does not exceed the limit', function (done) {
+    const maxSize = 15
+    const limitedStream = new LimitedStream(maxSize)
+    let data = ''
+    limitedStream.on('data', chunk => {
+      data += chunk.toString()
+    })
+    limitedStream.on('end', () => {
+      expect(data).to.equal('hello world')
+      done()
+    })
+    limitedStream.write('hello')
+    limitedStream.write(' world')
+    limitedStream.end()
+  })
+})

libraries/stream-utils/test/unit/LoggerStreamTests.js

@@ -0,0 +1,36 @@
+const { expect } = require('chai')
+const { LoggerStream } = require('../../index')
+
+describe('LoggerStream', function () {
+  it('should log the size of the stream when it exceeds the limit', function (done) {
+    const maxSize = 10
+    const loggedSizes = []
+    const loggerStream = new LoggerStream(maxSize, (size, isFlush) => {
+      loggedSizes.push([size, isFlush])
+      if (isFlush) {
+        expect(loggedSizes).to.deep.equal([
+          [11, undefined],
+          [11, true],
+        ])
+        done()
+      }
+    })
+    loggerStream.write(Buffer.alloc(maxSize))
+    loggerStream.write(Buffer.alloc(1))
+    loggerStream.end()
+  })
+
+  it('should not log the size of the stream if it does not exceed the limit', function (done) {
+    const maxSize = 10
+    const loggedSizes = []
+    const loggerStream = new LoggerStream(maxSize, (size, isFlush) => {
+      loggedSizes.push(size)
+    })
+    loggerStream.write(Buffer.alloc(maxSize))
+    loggerStream.end()
+    loggerStream.on('finish', () => {
+      expect(loggedSizes).to.deep.equal([])
+      done()
+    })
+  })
+})

libraries/stream-utils/test/unit/ReadableStringTests.js

@@ -0,0 +1,16 @@
+const { expect } = require('chai')
+const { ReadableString } = require('../../index')
+
+describe('ReadableString', function () {
+  it('should emit the string passed to it', function (done) {
+    const stringStream = new ReadableString('hello world')
+    let data = ''
+    stringStream.on('data', chunk => {
+      data += chunk.toString()
+    })
+    stringStream.on('end', () => {
+      expect(data).to.equal('hello world')
+      done()
+    })
+  })
+})

libraries/stream-utils/test/unit/TimeoutStreamTests.js

@@ -0,0 +1,22 @@
+const { expect } = require('chai')
+const { TimeoutStream, AbortError } = require('../../index')
+
+describe('TimeoutStream', function () {
+  it('should emit an error if the stream times out', function (done) {
+    const timeout = 10
+    const timeoutStream = new TimeoutStream(timeout)
+    timeoutStream.on('error', err => {
+      expect(err).to.be.an.instanceOf(AbortError)
+      done()
+    })
+  })
+
+  it('should not emit an error if the stream does not time out', function (done) {
+    const timeout = 100
+    const timeoutStream = new TimeoutStream(timeout)
+    setTimeout(() => {
+      timeoutStream.end()
+      done()
+    }, 1)
+  })
+})

libraries/stream-utils/test/unit/WritableBufferTests.js

@@ -0,0 +1,20 @@
+const { expect } = require('chai')
+const { WritableBuffer } = require('../../index')
+
+describe('WritableBuffer', function () {
+  it('should store all data written to it in a node Buffer', function () {
+    const bufferStream = new WritableBuffer()
+    bufferStream.write('hello')
+    bufferStream.write('world')
+    bufferStream.end()
+    expect(bufferStream.contents().toString()).to.equal('helloworld')
+  })
+
+  it('should return the size of the data written to it', function () {
+    const bufferStream = new WritableBuffer()
+    bufferStream.write('hello')
+    bufferStream.write('world')
+    bufferStream.end()
+    expect(bufferStream.size()).to.equal(10)
+  })
+})

services/docstore/app/js/DocArchiveManager.js

@@ -4,7 +4,7 @@ const Errors = require('./Errors')
 const logger = require('@overleaf/logger')
 const Settings = require('@overleaf/settings')
 const crypto = require('crypto')
-const Streamifier = require('streamifier')
+const { ReadableString } = require('@overleaf/stream-utils')
 const RangeManager = require('./RangeManager')
 const PersistorManager = require('./PersistorManager')
 const pMap = require('p-map')
@@ -92,7 +92,7 @@ async function archiveDoc(projectId, docId) {
   }
 
   const md5 = crypto.createHash('md5').update(json).digest('hex')
-  const stream = Streamifier.createReadStream(json)
+  const stream = new ReadableString(json)
   await PersistorManager.sendStream(Settings.docstore.bucket, key, stream, {
     sourceMd5: md5,
   })

services/docstore/package.json

@@ -21,6 +21,7 @@
     "@overleaf/o-error": "*",
     "@overleaf/object-persistor": "*",
     "@overleaf/settings": "*",
+    "@overleaf/stream-utils": "^0.1.0",
     "async": "^3.2.2",
     "body-parser": "^1.19.0",
     "bson": "^1.1.4",
@@ -30,8 +31,7 @@
     "lodash": "^4.17.21",
     "mongodb": "^4.11.0",
     "p-map": "^4.0.0",
-    "request": "^2.88.2",
-    "streamifier": "^0.1.1"
+    "request": "^2.88.2"
   },
   "devDependencies": {
     "@google-cloud/storage": "^6.10.1",

services/docstore/test/acceptance/js/ArchiveDocsTests.js

@@ -19,10 +19,10 @@ const DocstoreApp = require('./helpers/DocstoreApp')
 const DocstoreClient = require('./helpers/DocstoreClient')
 const { Storage } = require('@google-cloud/storage')
 const Persistor = require('../../../app/js/PersistorManager')
-const Streamifier = require('streamifier')
+const { ReadableString } = require('@overleaf/stream-utils')
 
 function uploadContent(path, json, callback) {
-  const stream = Streamifier.createReadStream(JSON.stringify(json))
+  const stream = new ReadableString(JSON.stringify(json))
   Persistor.sendStream(Settings.docstore.bucket, path, stream)
     .then(() => callback())
     .catch(callback)

services/docstore/test/unit/js/DocArchiveManagerTests.js

@@ -12,7 +12,7 @@ describe('DocArchiveManager', function () {
     RangeManager,
     Settings,
     Crypto,
-    Streamifier,
+    StreamUtils,
     HashDigest,
     HashUpdate,
     archivedDocs,
@@ -42,8 +42,8 @@ describe('DocArchiveManager', function () {
     Crypto = {
       createHash: sinon.stub().returns({ update: HashUpdate }),
     }
-    Streamifier = {
-      createReadStream: sinon.stub().returns({ stream: 'readStream' }),
+    StreamUtils = {
+      ReadableString: sinon.stub().returns({ stream: 'readStream' }),
     }
 
     projectId = ObjectId()
@@ -158,7 +158,7 @@ describe('DocArchiveManager', function () {
       requires: {
         '@overleaf/settings': Settings,
         crypto: Crypto,
-        streamifier: Streamifier,
+        '@overleaf/stream-utils': StreamUtils,
         './MongoManager': MongoManager,
         './RangeManager': RangeManager,
         './PersistorManager': PersistorManager,
@@ -185,7 +185,7 @@ describe('DocArchiveManager', function () {
 
     it('should add the schema version', async function () {
       await DocArchiveManager.promises.archiveDoc(projectId, mongoDocs[1]._id)
-      expect(Streamifier.createReadStream).to.have.been.calledWith(
+      expect(StreamUtils.ReadableString).to.have.been.calledWith(
         sinon.match(/"schema_v":1/)
       )
     })
@@ -219,7 +219,7 @@ describe('DocArchiveManager', function () {
 
     it('should create a stream from the encoded json and send it', async function () {
       await DocArchiveManager.promises.archiveDoc(projectId, mongoDocs[0]._id)
-      expect(Streamifier.createReadStream).to.have.been.calledWith(
+      expect(StreamUtils.ReadableString).to.have.been.calledWith(
         archivedDocJson
       )
       expect(PersistorManager.sendStream).to.have.been.calledWith(

services/filestore/app/js/HealthCheckController.js

@@ -1,7 +1,7 @@
 const fs = require('fs')
 const path = require('path')
 const Settings = require('@overleaf/settings')
-const streamBuffers = require('stream-buffers')
+const { WritableBuffer } = require('@overleaf/stream-utils')
 const { promisify } = require('util')
 const Stream = require('stream')
 
@@ -23,9 +23,7 @@ async function checkCanGetFiles() {
   const key = `${projectId}/${fileId}`
   const bucket = Settings.filestore.stores.user_files
 
-  const buffer = new streamBuffers.WritableStreamBuffer({
-    initialSize: 100,
-  })
+  const buffer = new WritableBuffer({ initialSize: 100 })
 
   const sourceStream = await FileHandler.getFile(bucket, key, {})
   try {

services/filestore/package.json

@@ -23,6 +23,7 @@
     "@overleaf/o-error": "*",
     "@overleaf/object-persistor": "*",
     "@overleaf/settings": "*",
+    "@overleaf/stream-utils": "^0.1.0",
     "body-parser": "^1.19.0",
     "bunyan": "^1.8.15",
     "express": "^4.18.2",
@@ -30,7 +31,6 @@
     "lodash.once": "^4.1.1",
     "node-fetch": "^2.6.7",
     "range-parser": "^1.2.1",
-    "stream-buffers": "~0.2.6",
     "tiny-async-pool": "^1.1.0"
   },
   "devDependencies": {

services/filestore/test/acceptance/js/FilestoreTests.js

@@ -137,13 +137,6 @@ describe('Filestore', function () {
         expect(body).to.contain('up')
       })
 
-      it('should send a 200 for the health-check endpoint', async function () {
-        const response = await fetch(`${filestoreUrl}/health_check`)
-        expect(response.status).to.equal(200)
-        const body = await response.text()
-        expect(body).to.equal('OK')
-      })
-
       describe('with a file on the server', function () {
         let fileId, fileUrl, constantFileContent
 
@@ -198,6 +191,17 @@ describe('Filestore', function () {
           expect(body).to.equal(constantFileContent)
         })
 
+        it('should send a 200 for the health-check endpoint using the file', async function () {
+          Settings.health_check = {
+            project_id: projectId,
+            file_id: fileId,
+          }
+          const response = await fetch(`${filestoreUrl}/health_check`)
+          expect(response.status).to.equal(200)
+          const body = await response.text()
+          expect(body).to.equal('OK')
+        })
+
         it('should not leak a socket', async function () {
           await fetch(fileUrl)
           await expectNoSockets()

services/history-v1/package.json

@@ -10,6 +10,7 @@
     "@overleaf/metrics": "*",
     "@overleaf/o-error": "*",
     "@overleaf/object-persistor": "*",
+    "@overleaf/stream-utils": "^0.1.0",
     "archiver": "^5.3.0",
     "basic-auth": "^2.0.1",
     "bluebird": "^3.7.2",
@@ -31,7 +32,6 @@
     "mongodb": "^4.11.0",
     "overleaf-editor-core": "*",
     "pg": "^8.7.1",
-    "string-to-stream": "^1.0.1",
     "swagger-tools": "^0.10.4",
     "temp": "^0.8.3",
     "throng": "^4.0.0",

services/history-v1/storage/lib/blob_store/index.js

@@ -3,7 +3,7 @@
 const config = require('config')
 const fs = require('fs')
 const isValidUtf8 = require('utf-8-validate')
-const stringToStream = require('string-to-stream')
+const { ReadableString } = require('@overleaf/stream-utils')
 
 const core = require('overleaf-editor-core')
 const objectPersistor = require('@overleaf/object-persistor')
@@ -162,9 +162,9 @@ class BlobStore {
       return existingBlob
     }
     const newBlob = new Blob(hash, Buffer.byteLength(string), string.length)
-    // Note: the stringToStream is to work around a bug in the AWS SDK: it won't
+    // Note: the ReadableString is to work around a bug in the AWS SDK: it won't
     // allow Body to be blank.
-    await uploadBlob(this.projectId, newBlob, stringToStream(string))
+    await uploadBlob(this.projectId, newBlob, new ReadableString(string))
     await this.backend.insertBlob(this.projectId, newBlob)
     return newBlob
   }

services/history-v1/storage/lib/streams.js

@@ -7,8 +7,8 @@
 
 const BPromise = require('bluebird')
 const zlib = require('zlib')
-const stringToStream = require('string-to-stream')
-const { pipeline, Writable } = require('stream')
+const { WritableBuffer, ReadableString } = require('@overleaf/stream-utils')
+const { pipeline } = require('stream')
 
 function promisePipe(readStream, writeStream) {
   return new BPromise(function (resolve, reject) {
@@ -33,26 +33,6 @@ function promisePipe(readStream, writeStream) {
  */
 exports.promisePipe = promisePipe
 
-class WritableBuffer extends Writable {
-  constructor(options) {
-    super(options)
-    this.buffers = []
-  }
-
-  _write(chunk, encoding, callback) {
-    this.buffers.push(chunk)
-    callback()
-  }
-
-  _final(callback) {
-    callback()
-  }
-
-  contents() {
-    return Buffer.concat(this.buffers)
-  }
-}
-
 function readStreamToBuffer(readStream) {
   return new BPromise(function (resolve, reject) {
     const bufferStream = new WritableBuffer()
@@ -100,7 +80,7 @@ exports.gunzipStreamToBuffer = gunzipStreamToBuffer
 
 function gzipStringToStream(string) {
   const gzip = zlib.createGzip()
-  return stringToStream(string).pipe(gzip)
+  return new ReadableString(string).pipe(gzip)
 }
 
 /**

services/project-history/package.json

@@ -34,7 +34,6 @@
     "esmock": "^2.1.0",
     "express": "^4.18.2",
     "heap": "^0.2.6",
-    "JSONStream": "^1.3.5",
     "line-reader": "^0.2.4",
     "lodash": "^4.17.20",
     "mongo-uri": "^0.1.2",
@@ -48,7 +47,6 @@
   "devDependencies": {
     "chai": "^4.3.6",
     "chai-as-promised": "^7.1.1",
-    "memorystream": "0.3.1",
     "mocha": "^10.2.0",
     "multer": "^1.4.2",
     "nock": "^12.0.3",

services/web/package.json

@@ -137,7 +137,6 @@
     "body-parser": "^1.19.0",
     "bootstrap": "^3.4.1",
     "bowser": "^2.11.0",
-    "bufferedstream": "1.6.0",
     "bull": "^3.18.0",
     "bunyan": "^1.8.15",
     "cache-flow": "^1.7.4",

services/web/test/unit/src/ThirdPartyDataStore/UpdateMergerTests.js

@@ -1,7 +1,7 @@
 const SandboxedModule = require('sandboxed-module')
 const sinon = require('sinon')
 const { expect } = require('chai')
-const BufferedStream = require('bufferedstream')
+const { Writable } = require('stream')
 const { ObjectId } = require('mongodb')
 
 const MODULE_PATH =
@@ -32,7 +32,7 @@ describe('UpdateMerger :', function () {
 \\date{June 2011}`
     this.docLines = this.fileContents.split('\n')
     this.source = 'dropbox'
-    this.updateRequest = new BufferedStream()
+    this.updateRequest = new Writable()
 
     this.fsPromises = {
       unlink: sinon.stub().resolves(),