colossus: remove sync fs i/o calls

mnaamani · Jan 22, 2024 · 3e836f4 · 3e836f4
1 parent d9f7f18
commit 3e836f4
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 35 deletions.
diff --git a/storage-node/src/services/sync/acceptPendingObjects.ts b/storage-node/src/services/sync/acceptPendingObjects.ts
@@ -67,8 +67,11 @@ export class AcceptPendingObjectsService {
     return this.instance
   }
 
-  pendingObjectExists(id: string): boolean {
-    return fs.existsSync(path.join(this.pendingDataObjectsDir, id))
+  async pendingObjectExists(id: string): Promise<boolean> {
+    return fsPromises
+      .access(path.join(this.pendingDataObjectsDir, id), fs.constants.F_OK)
+      .then(() => true)
+      .catch(() => false)
   }
 
   private async getPendingObjectsFromFolder(): Promise<string[]> {

diff --git a/storage-node/src/services/sync/tasks.ts b/storage-node/src/services/sync/tasks.ts
@@ -94,6 +94,11 @@ export class DownloadFileTask implements SyncTask {
   async execute(): Promise<void> {
     const operatorUrlIndices: number[] = _.shuffle(_.range(this.operatorUrls.length))
 
+    if (operatorUrlIndices.length === 0) {
+      logger.warn(`Sync - No operator URLs for ${this.dataObjectId}`)
+      return
+    }
+
     for (const randomUrlIndex of operatorUrlIndices) {
       const chosenBaseUrl = this.operatorUrls[randomUrlIndex]
       logger.debug(`Sync - random storage node URL was chosen ${chosenBaseUrl}`)
@@ -104,15 +109,18 @@ export class DownloadFileTask implements SyncTask {
         await this.tryDownload(chosenBaseUrl, filepath)
 
         // if download succeeds, break the loop
-        if (fs.existsSync(filepath)) {
+        try {
+          await fsPromises.access(filepath, fs.constants.F_OK)
           return
+        } catch (err) {
+          continue
         }
       } catch (err) {
         logger.error(`Sync - fetching data error for ${this.dataObjectId}: ${err}`, { err })
       }
     }
 
-    logger.warn(`Sync - cannot get operator URLs for ${this.dataObjectId}`)
+    logger.warn(`Sync - Failed to download ${this.dataObjectId}`)
   }
 
   async tryDownload(url: string, filepath: string): Promise<void> {

diff --git a/storage-node/src/services/webApi/app.ts b/storage-node/src/services/webApi/app.ts
@@ -251,13 +251,13 @@ async function validateUploadFileParams(req: express.Request, res: express.Respo
     throw new WebApiError(`Data object ${dataObjectId} doesn't exist in storage bag ${parsedBagId}`, 400)
   }
 
-  const isObjectPending = res.locals.acceptPendingObjectsService.pendingObjectExists(dataObjectId.toString())
-  if (isObjectPending) {
-    throw new WebApiError(`Data object ${dataObjectId} already exists`, 400)
-  }
-
   const isInStorage = getDataObjectIdFromCache(dataObjectId.toString())
   if (isInStorage) {
-    throw new WebApiError(`Data object ${dataObjectId} already exists`, 400)
+    throw new WebApiError(`Data object ${dataObjectId} already exists (in storage)`, 400)
+  }
+
+  const isObjectPending = await res.locals.acceptPendingObjectsService.pendingObjectExists(dataObjectId.toString())
+  if (isObjectPending) {
+    throw new WebApiError(`Data object ${dataObjectId} already exists (pending)`, 400)
   }
 }
diff --git a/storage-node/src/services/webApi/controllers/stateApi.ts b/storage-node/src/services/webApi/controllers/stateApi.ts
@@ -60,39 +60,33 @@ export async function getLocalDataStats(
     const pendingObjectsDir = res.locals.pendingDataObjectsDir
     const fastFolderSizeAsync = promisify(fastFolderSize)
 
-    const tempFolderExists = fs.existsSync(tempFileDir)
-    const statsPromise = fsPromises.readdir(uploadsDir)
+    const statsPromise = fsPromises.readdir(uploadsDir, { withFileTypes: true })
     const sizePromise = fastFolderSizeAsync(uploadsDir)
 
     const [stats, totalSize] = await Promise.all([statsPromise, sizePromise])
 
-    let objectNumber = stats.length
+    const objectNumber = stats.filter((entry) => entry.isFile).length
     let tempDownloads = 0
     let tempDirSize = 0
     let pendingObjects = 0
     let pendingDirSize = 0
-    if (tempFolderExists) {
-      if (objectNumber > 0) {
-        objectNumber--
-      }
-
-      const tempDirStatsPromise = fsPromises.readdir(tempFileDir)
-      const tempDirSizePromise = fastFolderSizeAsync(tempFileDir)
-      const pendingDirStatsPromise = fsPromises.readdir(pendingObjectsDir)
-      const pendingDirSizePromise = fastFolderSizeAsync(pendingObjectsDir)
-
-      const [tempDirStats, tempSize, pendingDirStats, pendingSize] = await Promise.all([
-        tempDirStatsPromise,
-        tempDirSizePromise,
-        pendingDirStatsPromise,
-        pendingDirSizePromise,
-      ])
-
-      tempDirSize = tempSize ?? 0
-      tempDownloads = tempDirStats.length
-      pendingDirSize = pendingSize ?? 0
-      pendingObjects = pendingDirStats.length
-    }
+
+    const tempDirStatsPromise = fsPromises.readdir(tempFileDir, { withFileTypes: true })
+    const tempDirSizePromise = fastFolderSizeAsync(tempFileDir)
+    const pendingDirStatsPromise = fsPromises.readdir(pendingObjectsDir, { withFileTypes: true })
+    const pendingDirSizePromise = fastFolderSizeAsync(pendingObjectsDir)
+
+    const [tempDirStats, tempSize, pendingDirStats, pendingSize] = await Promise.all([
+      tempDirStatsPromise,
+      tempDirSizePromise,
+      pendingDirStatsPromise,
+      pendingDirSizePromise,
+    ])
+
+    tempDirSize = tempSize ?? 0
+    tempDownloads = tempDirStats.filter((entry) => entry.isFile).length
+    pendingDirSize = pendingSize ?? 0
+    pendingObjects = pendingDirStats.filter((entry) => entry.isFile).length
 
     res.status(200).json({
       objectNumber,