CON-402 CN Remove all synchronous disk ops (#3908)

creator-node/src/components/replicaSet/syncQueueComponentService.js

@@ -4,6 +4,7 @@
  */
 const enqueueSync = async (params) => {
   const { serviceRegistry } = params
+  // eslint-disable-next-line node/no-sync
   await serviceRegistry.syncQueue.enqueueSync(params)
 }
 
@@ -13,6 +14,7 @@ const enqueueSync = async (params) => {
  */
 const processManualImmediateSync = async (params) => {
   const { serviceRegistry } = params
+  // eslint-disable-next-line node/no-sync
   await serviceRegistry.syncImmediateQueue.processManualImmediateSync(params)
 }
 

creator-node/src/components/tracks/TrackTranscodeHandoffManager.js

@@ -1,6 +1,5 @@
 const axios = require('axios')
-const fs = require('fs')
-const fsExtra = require('fs-extra')
+const fs = require('fs-extra')
 const FormData = require('form-data')
 const _ = require('lodash')
 
@@ -380,7 +379,7 @@ class TrackTranscodeHandoffManager {
    * @returns formData object passed in axios to send a transcode and segment request
    */
   static async createFormData(pathToFile) {
-    const fileExists = await fsExtra.pathExists(pathToFile)
+    const fileExists = await fs.pathExists(pathToFile)
     if (!fileExists) {
       throw new Error(`File does not exist at path=${pathToFile}`)
     }

creator-node/src/config.js

@@ -1,6 +1,6 @@
 const axios = require('axios')
 const convict = require('convict')
-const fs = require('fs')
+const fs = require('fs-extra')
 const path = require('path')
 const os = require('os')
 const _ = require('lodash')

creator-node/src/diskManager.js

@@ -1,5 +1,5 @@
 const path = require('path')
-const fs = require('fs')
+const fs = require('fs-extra')
 const config = require('./config')
 const { logger: genericLogger } = require('./logging')
 const CID = require('cids')
@@ -27,14 +27,14 @@ class DiskManager {
    * is we should be able to delete the contents of this folder without scanning through other folders with the
    * naming scheme.
    */
-  static getTmpTrackUploadArtifactsPath() {
+  static async getTmpTrackUploadArtifactsPath() {
     const dirPath = path.join(
       config.get('storagePath'),
       'files',
       'tmp_track_artifacts'
     )
     if (!TMP_TRACK_ARTIFACTS_CREATED) {
-      this.ensureDirPathExists(dirPath)
+      await this.ensureDirPathExists(dirPath)
       TMP_TRACK_ARTIFACTS_CREATED = true
     }
     return dirPath
@@ -51,7 +51,7 @@ class DiskManager {
    *      eg QmYfSQCgCwhxwYcdEwCkFJHicDe6rzCAb7AtLz3GrHmuU6 will be eg /file_storage/muU/QmYfSQCgCwhxwYcdEwCkFJHicDe6rzCAb7AtLz3GrHmuU6
    * @param {String} cid file system destination, either filename or directory
    */
-  static computeFilePath(cid, ensureDirPathExists = true) {
+  static async computeFilePath(cid, ensureDirPathExists = true) {
     try {
       CID.isCID(new CID(cid))
     } catch (e) {
@@ -76,7 +76,7 @@ class DiskManager {
 
     // create the subdirectories in parentDirHash if they don't exist
     if (ensureDirPathExists) {
-      this.ensureDirPathExists(parentDirPath)
+      await this.ensureDirPathExists(parentDirPath)
     }
 
     return path.join(parentDirPath, cid)
@@ -115,7 +115,7 @@ class DiskManager {
    * @param {String} dirName directory name
    * @param {String} fileName file name
    */
-  static computeFilePathInDir(dirName, fileName) {
+  static async computeFilePathInDir(dirName, fileName) {
     if (!dirName || !fileName) {
       genericLogger.error(
         `Invalid dirName and/or fileName, dirName=${dirName}, fileName=${fileName}`
@@ -135,7 +135,7 @@ class DiskManager {
       )
     }
 
-    const parentDirPath = this.computeFilePath(dirName)
+    const parentDirPath = await this.computeFilePath(dirName)
     const absolutePath = path.join(parentDirPath, fileName)
     genericLogger.info(`File path computed, absolutePath=${absolutePath}`)
     return absolutePath
@@ -146,10 +146,10 @@ class DiskManager {
    * If it does exist, it will not overwrite, effectively a no-op
    * @param {*} dirPath fs directory path to create if it does not exist
    */
-  static ensureDirPathExists(dirPath) {
+  static async ensureDirPathExists(dirPath) {
     try {
       // the mkdir recursive option is equivalent to `mkdir -p` and should created nested folders several levels deep
-      fs.mkdirSync(dirPath, { recursive: true })
+      await fs.mkdir(dirPath, { recursive: true })
     } catch (e) {
       genericLogger.error(
         `Error making directory, dirName=${dirPath}, error=${e.toString()}`

creator-node/src/diskManager.test.js

@@ -22,23 +22,23 @@ describe('Test DiskManager', function () {
   /**
    * getTmpTrackUploadArtifactsPath
    */
-  it('Should pass if storagePath is correctly set', function () {
+  it('Should pass if storagePath is correctly set', async function () {
     const tmpTrackArtifactPath = path.join(
       DiskManager.getConfigStoragePath(),
       'files',
       'tmp_track_artifacts'
     )
     assert.deepStrictEqual(
       tmpTrackArtifactPath,
-      DiskManager.getTmpTrackUploadArtifactsPath()
+      await DiskManager.getTmpTrackUploadArtifactsPath()
     )
   })
 
   /**
    * computeFilePath
    */
-  it('Should pass if computeFilePath returns the correct path', function () {
-    const fullPath = DiskManager.computeFilePath(
+  it('Should pass if computeFilePath returns the correct path', async function () {
+    const fullPath = await DiskManager.computeFilePath(
       'QmYfSQCgCwhxwYcdEwCkFJHicDe6rzCAb7AtLz3GrHmuU6'
     )
     const validPath = path.join(
@@ -50,29 +50,29 @@ describe('Test DiskManager', function () {
     assert.deepStrictEqual(fullPath, validPath)
   })
 
-  it('Should fail if fileName is not passed into computeFilePath', function () {
+  it('Should fail if fileName is not passed into computeFilePath', async function () {
     try {
-      DiskManager.computeFilePath()
+      await DiskManager.computeFilePath()
     } catch (e) {
       assert.ok(
         e.message.includes('Please pass in a valid cid to computeFilePath')
       )
     }
   })
 
-  it(`Should fail if fileName doesn't contain the appropriate amount of characters`, function () {
+  it(`Should fail if fileName doesn't contain the appropriate amount of characters`, async function () {
     try {
-      DiskManager.computeFilePath('asd')
+      await DiskManager.computeFilePath('asd')
     } catch (e) {
       assert.ok(
         e.message.includes('Please pass in a valid cid to computeFilePath')
       )
     }
   })
 
-  it(`Should fail if fileName contains a slash`, function () {
+  it(`Should fail if fileName contains a slash`, async function () {
     try {
-      DiskManager.computeFilePath('/file_storage/asdf')
+      await DiskManager.computeFilePath('/file_storage/asdf')
     } catch (e) {
       assert.ok(
         e.message.includes('Please pass in a valid cid to computeFilePath')
@@ -83,8 +83,8 @@ describe('Test DiskManager', function () {
   /**
    * computeFilePathInDir
    */
-  it('Should pass if computeFilePathInDir returns the correct path', function () {
-    const fullPath = DiskManager.computeFilePathInDir(
+  it('Should pass if computeFilePathInDir returns the correct path', async function () {
+    const fullPath = await DiskManager.computeFilePathInDir(
       'QmRSvU8NtadxPPrP4M72wUPBiTqykqziWDuGr6q2arsYW4',
       'QmYfSQCgCwhxwYcdEwCkFJHicDe6rzCAb7AtLz3GrHmuU6'
     )
@@ -98,17 +98,17 @@ describe('Test DiskManager', function () {
     assert.deepStrictEqual(fullPath, validPath)
   })
 
-  it('Should fail if dirName and fileName are not passed into computeFilePathInDir', function () {
+  it('Should fail if dirName and fileName are not passed into computeFilePathInDir', async function () {
     try {
-      DiskManager.computeFilePathInDir()
+      await DiskManager.computeFilePathInDir()
     } catch (e) {
       assert.ok(e.message.includes('Must pass in valid dirName and fileName'))
     }
   })
 
-  it('Should fail if dirName or fileName are not a CID passed into computeFilePathInDir', function () {
+  it('Should fail if dirName or fileName are not a CID passed into computeFilePathInDir', async function () {
     try {
-      DiskManager.computeFilePathInDir(
+      await DiskManager.computeFilePathInDir(
         'Qmdirhash',
         'QmYfSQCgCwhxwYcdEwCkFJHicDe6rzCAb7AtLz3GrHmuU6'
       )

creator-node/src/ffmpeg.js

@@ -1,5 +1,5 @@
 const config = require('./config')
-const fs = require('fs')
+const fs = require('fs-extra')
 const path = require('path')
 const ffmpeg = require('ffmpeg-static').path
 const spawn = require('child_process').spawn
@@ -13,7 +13,7 @@ const { logger: genericLogger } = require('./logging')
  * @param {string} params.fileDir the directory of the uploaded track artifact
  * @param {string} params.fileName the uploaded track artifact filename
  * @param {Object} params.logContext the log context used to instantiate a logger
- * @returns {Object} response in the structure 
+ * @returns {Promise<Object>} response in the structure 
   {
     segments: {
       fileNames: segmentFileNames {string[]}: the segment file names only, 
@@ -24,6 +24,7 @@ const { logger: genericLogger } = require('./logging')
  */
 function segmentFile(fileDir, fileName, { logContext }) {
   const logger = genericLogger.child(logContext)
+
   return new Promise((resolve, reject) => {
     const absolutePath = path.resolve(fileDir, fileName)
     logger.info(`Segmenting file ${absolutePath}...`)
@@ -63,24 +64,27 @@ function segmentFile(fileDir, fileName, { logContext }) {
     proc.stderr.on('data', (data) => (stderr += data.toString()))
 
     proc.on('close', (code) => {
-      if (code === 0) {
-        const segmentFileNames = fs.readdirSync(fileDir + '/segments')
-        const segmentFilePaths = segmentFileNames.map((filename) =>
-          path.resolve(fileDir, 'segments', filename)
-        )
+      async function asyncFn() {
+        if (code === 0) {
+          const segmentFileNames = await fs.readdir(fileDir + '/segments')
+          const segmentFilePaths = segmentFileNames.map((filename) =>
+            path.resolve(fileDir, 'segments', filename)
+          )
 
-        resolve({
-          segments: {
-            fileNames: segmentFileNames,
-            filePaths: segmentFilePaths
-          },
-          m3u8FilePath
-        })
-      } else {
-        logger.error('Error when processing file with ffmpeg')
-        logger.error('Command stdout:', stdout, '\nCommand stderr:', stderr)
-        reject(new Error('FFMPEG Error'))
+          resolve({
+            segments: {
+              fileNames: segmentFileNames,
+              filePaths: segmentFilePaths
+            },
+            m3u8FilePath
+          })
+        } else {
+          logger.error('Error when processing file with ffmpeg')
+          logger.error('Command stdout:', stdout, '\nCommand stderr:', stderr)
+          reject(new Error('FFMPEG Error'))
+        }
       }
+      asyncFn()
     })
   })
 }
@@ -92,21 +96,22 @@ function segmentFile(fileDir, fileName, { logContext }) {
  * @param {string} params.fileDir the directory of the uploaded track artifact
  * @param {string} params.fileName the uploaded track artifact filename
  * @param {Object} params.logContext the log context used to instantiate a logger
- * @returns {string} the path to the transcode
+ * @returns {Promise<string>} the path to the transcode
  */
-function transcodeFileTo320(fileDir, fileName, { logContext }) {
+async function transcodeFileTo320(fileDir, fileName, { logContext }) {
   const logger = genericLogger.child(logContext)
-  return new Promise((resolve, reject) => {
-    const sourcePath = path.resolve(fileDir, fileName)
-    const targetPath = path.resolve(fileDir, fileName.split('.')[0] + '-dl.mp3')
-    logger.info(`Transcoding file ${sourcePath}...`)
 
-    // Exit if dl-copy file already exists at target path.
-    if (fs.existsSync(targetPath)) {
-      logger.info(`Downloadable copy already exists at ${targetPath}.`)
-      resolve(targetPath)
-    }
+  const sourcePath = path.resolve(fileDir, fileName)
+  const targetPath = path.resolve(fileDir, fileName.split('.')[0] + '-dl.mp3')
+  logger.info(`Transcoding file ${sourcePath}...`)
 
+  // Exit if dl-copy file already exists at target path.
+  if (await fs.pathExists(targetPath)) {
+    logger.info(`Downloadable copy already exists at ${targetPath}.`)
+    return targetPath
+  }
+
+  return new Promise((resolve, reject) => {
     // https://ffmpeg.org/ffmpeg-formats.html#hls-2
     const args = [
       '-i',
@@ -130,20 +135,23 @@ function transcodeFileTo320(fileDir, fileName, { logContext }) {
     proc.stderr.on('data', (data) => (stderr += data.toString()))
 
     proc.on('close', (code) => {
-      if (code === 0) {
-        if (fs.existsSync(targetPath)) {
-          logger.info(`Transcoded file ${targetPath}`)
-          resolve(targetPath)
+      async function asyncFn() {
+        if (code === 0) {
+          if (await fs.pathExists(targetPath)) {
+            logger.info(`Transcoded file ${targetPath}`)
+            resolve(targetPath)
+          } else {
+            logger.error('Error when processing file with ffmpeg')
+            logger.error('Command stdout:', stdout, '\nCommand stderr:', stderr)
+            reject(new Error('FFMPEG Error'))
+          }
         } else {
           logger.error('Error when processing file with ffmpeg')
           logger.error('Command stdout:', stdout, '\nCommand stderr:', stderr)
           reject(new Error('FFMPEG Error'))
         }
-      } else {
-        logger.error('Error when processing file with ffmpeg')
-        logger.error('Command stdout:', stdout, '\nCommand stderr:', stderr)
-        reject(new Error('FFMPEG Error'))
       }
+      asyncFn()
     })
   })
 }