fix: storage adapter throw error on stat (#8893) (#8898)

Signed-off-by: Alexander Onnikov <[email protected]>

Author: aonnikov

common/config/rush/pnpm-lock.yaml

@@ -1470,16 +1470,20 @@ async function rebuildSizeInfo (
 
   const addFileSize = async (file: string | undefined | null): Promise<void> => {
     if (file != null) {
-      const sz = sizeInfo[file]
-      const fileSize = sz ?? (await storage.stat(file))
-      if (sz === undefined) {
-        sizeInfo[file] = fileSize
-        processed++
-        if (processed % 10 === 0) {
-          ctx.info('Calculate size processed', { processed, size: Math.round(result.backupSize / (1024 * 1024)) })
+      try {
+        const sz = sizeInfo[file]
+        const fileSize = sz ?? (await storage.stat(file))
+        if (sz === undefined) {
+          sizeInfo[file] = fileSize
+          processed++
+          if (processed % 10 === 0) {
+            ctx.info('Calculate size processed', { processed, size: Math.round(result.backupSize / (1024 * 1024)) })
+          }
         }
+        result.backupSize += fileSize
+      } catch (err: any) {
+        ctx.error('failed to calculate size', { file, err })
       }
-      result.backupSize += fileSize
     }
   }
 
@@ -1551,9 +1555,13 @@ export async function backupSize (storage: BackupStorage): Promise<void> {
   console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version)
   const addFileSize = async (file: string | undefined | null): Promise<void> => {
     if (file != null && (await storage.exists(file))) {
-      const fileSize = await storage.stat(file)
-      console.log(file, fileSize)
-      size += fileSize
+      try {
+        const fileSize = await storage.stat(file)
+        console.log(file, fileSize)
+        size += fileSize
+      } catch (err: any) {
+        console.error('failed to calculate size', { file, err })
+      }
     }
   }
 
@@ -1606,20 +1614,24 @@ export async function backupDownload (storage: BackupStorage, storeIn: string):
       const serverSize: number | undefined = sizeInfo[file]
 
       if (!existsSync(target) || force || (serverSize !== undefined && serverSize !== statSync(target).size)) {
-        const fileSize = serverSize ?? (await storage.stat(file))
-        console.log('downloading', file, fileSize)
-        const readStream = await storage.load(file)
-        const outp = createWriteStream(target)
+        try {
+          const fileSize = serverSize ?? (await storage.stat(file))
+          console.log('downloading', file, fileSize)
+          const readStream = await storage.load(file)
+          const outp = createWriteStream(target)
 
-        readStream.pipe(outp)
-        await new Promise<void>((resolve) => {
-          readStream.on('end', () => {
-            readStream.destroy()
-            outp.close()
-            resolve()
+          readStream.pipe(outp)
+          await new Promise<void>((resolve) => {
+            readStream.on('end', () => {
+              readStream.destroy()
+              outp.close()
+              resolve()
+            })
           })
-        })
-        size += fileSize
+          size += fileSize
+        } catch (err: any) {
+          console.error('failed to calculate size', { file, err })
+        }
       } else {
         console.log('file-same', file)
       }

server/backup/src/backup.ts

@@ -46,22 +46,37 @@ export interface DatalakeConfig extends StorageConfig {
 /**
  * @public
  */
-export function createDatalakeClient (opt: DatalakeConfig, token: string): DatalakeClient {
-  const endpoint = Number.isInteger(opt.port) ? `${opt.endpoint}:${opt.port}` : opt.endpoint
+export function createDatalakeClient (cfg: DatalakeConfig, token: string): DatalakeClient {
+  const endpoint = Number.isInteger(cfg.port) ? `${cfg.endpoint}:${cfg.port}` : cfg.endpoint
   return new DatalakeClient(endpoint, token)
 }
 
 export const CONFIG_KIND = 'datalake'
 
+/**
+ * @public
+ */
+export interface DatalakeClientOptions {
+  retryCount?: number
+  retryInterval?: number
+}
+
 /**
  * @public
  */
 export class DatalakeService implements StorageAdapter {
   private readonly client: DatalakeClient
+  private readonly retryCount: number
+  private readonly retryInterval: number
 
-  constructor (readonly opt: DatalakeConfig) {
+  constructor (
+    readonly cfg: DatalakeConfig,
+    readonly options: DatalakeClientOptions = {}
+  ) {
     const token = generateToken(systemAccountEmail, { name: '' }, { service: 'datalake' })
-    this.client = createDatalakeClient(opt, token)
+    this.client = createDatalakeClient(cfg, token)
+    this.retryCount = options.retryCount ?? 5
+    this.retryInterval = options.retryInterval ?? 50
   }
 
   async initialize (ctx: MeasureContext, workspaceId: WorkspaceId): Promise<void> {}
@@ -86,7 +101,7 @@ export class DatalakeService implements StorageAdapter {
   async remove (ctx: MeasureContext, workspaceId: WorkspaceId, objectNames: string[]): Promise<void> {
     await Promise.all(
       objectNames.map(async (objectName) => {
-        await this.client.deleteObject(ctx, workspaceId, objectName)
+        await this.retry(ctx, () => this.client.deleteObject(ctx, workspaceId, objectName))
       })
     )
   }
@@ -106,7 +121,7 @@ export class DatalakeService implements StorageAdapter {
       next: async () => {
         try {
           while (hasMore && buffer.length < 50) {
-            const res = await this.client.listObjects(ctx, workspaceId, cursor)
+            const res = await this.retry(ctx, () => this.client.listObjects(ctx, workspaceId, cursor))
             hasMore = res.cursor !== undefined
             cursor = res.cursor
 
@@ -116,7 +131,7 @@ export class DatalakeService implements StorageAdapter {
                 _class: core.class.Blob,
                 etag: blob.etag,
                 size: (typeof blob.size === 'string' ? parseInt(blob.size) : blob.size) ?? 0,
-                provider: this.opt.name,
+                provider: this.cfg.name,
                 space: core.space.Configuration,
                 modifiedBy: core.account.ConfigUser,
                 modifiedOn: 0
@@ -134,32 +149,26 @@ export class DatalakeService implements StorageAdapter {
 
   @withContext('stat')
   async stat (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<Blob | undefined> {
-    return await withRetry(ctx, 5, async () => {
-      try {
-        const result = await this.client.statObject(ctx, workspaceId, objectName)
-        if (result !== undefined) {
-          return {
-            provider: '',
-            _class: core.class.Blob,
-            _id: objectName as Ref<Blob>,
-            contentType: result.type,
-            size: result.size ?? 0,
-            etag: result.etag ?? '',
-            space: core.space.Configuration,
-            modifiedBy: core.account.System,
-            modifiedOn: result.lastModified,
-            version: null
-          }
-        }
-      } catch (err) {
-        ctx.error('failed to stat object', { error: err, objectName, workspaceId: workspaceId.name })
+    const result = await this.retry(ctx, () => this.client.statObject(ctx, workspaceId, objectName))
+    if (result !== undefined) {
+      return {
+        provider: '',
+        _class: core.class.Blob,
+        _id: objectName as Ref<Blob>,
+        contentType: result.type,
+        size: result.size ?? 0,
+        etag: result.etag ?? '',
+        space: core.space.Configuration,
+        modifiedBy: core.account.System,
+        modifiedOn: result.lastModified,
+        version: null
       }
-    })
+    }
   }
 
   @withContext('get')
   async get (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<Readable> {
-    return await this.client.getObject(ctx, workspaceId, objectName)
+    return await this.retry(ctx, () => this.client.getObject(ctx, workspaceId, objectName))
   }
 
   @withContext('put')
@@ -178,7 +187,7 @@ export class DatalakeService implements StorageAdapter {
     }
 
     const { etag } = await ctx.with('put', {}, (ctx) =>
-      withRetry(ctx, 5, () => this.client.putObject(ctx, workspaceId, objectName, stream, params))
+      this.retry(ctx, () => this.client.putObject(ctx, workspaceId, objectName, stream, params))
     )
 
     return {
@@ -189,7 +198,7 @@ export class DatalakeService implements StorageAdapter {
 
   @withContext('read')
   async read (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<Buffer[]> {
-    const data = await this.client.getObject(ctx, workspaceId, objectName)
+    const data = await this.retry(ctx, () => this.client.getObject(ctx, workspaceId, objectName))
     const chunks: Buffer[] = []
 
     for await (const chunk of data) {
@@ -207,12 +216,16 @@ export class DatalakeService implements StorageAdapter {
     offset: number,
     length?: number
   ): Promise<Readable> {
-    return await this.client.getPartialObject(ctx, workspaceId, objectName, offset, length)
+    return await this.retry(ctx, () => this.client.getPartialObject(ctx, workspaceId, objectName, offset, length))
   }
 
   async getUrl (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<string> {
     return this.client.getObjectUrl(ctx, workspaceId, objectName)
   }
+
+  async retry<T>(ctx: MeasureContext, op: () => Promise<T>): Promise<T> {
+    return await withRetry(ctx, this.retryCount, op, this.retryInterval)
+  }
 }
 
 export function processConfigFromEnv (storageConfig: StorageConfiguration): string | undefined {
@@ -244,7 +257,7 @@ async function withRetry<T> (
     } catch (err: any) {
       error = err
       ctx.error('error', { err })
-      if (retries !== 0) {
+      if (retries !== 0 && delay > 0) {
         await new Promise((resolve) => setTimeout(resolve, delay))
       }
     }

server/datalake/src/index.ts

@@ -31,13 +31,12 @@
     "@types/node": "~20.11.16",
     "jest": "^29.7.0",
     "ts-jest": "^29.1.1",
-    "@types/jest": "^29.5.5",
-    "@types/minio": "~7.0.11"
+    "@types/jest": "^29.5.5"
   },
   "dependencies": {
     "@hcengineering/core": "^0.6.32",
     "@hcengineering/platform": "^0.6.11",
     "@hcengineering/server-core": "^0.6.1",
-    "minio": "^8.0.0"
+    "minio": "^8.0.5"
   }
 }