UBERF-7260 Handle storage errors in collaborator (#5806)

Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>

server/collaboration/src/index.ts

@@ -17,5 +17,5 @@ export * from './history/branch'
 export * from './history/history'
 export * from './history/snapshot'
 export * from './utils/collaborative-doc'
-export * from './utils/minio'
+export * from './utils/storage'
 export * from './utils/ydoc'

server/collaboration/src/utils/collaborative-doc.ts

@@ -28,7 +28,7 @@ import { StorageAdapter } from '@hcengineering/server-core'
 import { yDocBranch } from '../history/branch'
 import { YDocVersion } from '../history/history'
 import { createYdocSnapshot, restoreYdocSnapshot } from '../history/snapshot'
-import { yDocFromStorage, yDocToStorage } from './minio'
+import { yDocFromStorage, yDocToStorage } from './storage'
 
 /** @public */
 export function collaborativeHistoryDocId (id: string): string {

server/collaboration/src/utils/storage.ts

@@ -27,19 +27,18 @@ export async function yDocFromStorage (
   minioDocumentId: string,
   ydoc?: YDoc
 ): Promise<YDoc | undefined> {
+  // stat the object to ensure it exists, because read will throw an error in this case
+  const blob = await storageAdapter.stat(ctx, workspace, minioDocumentId)
+  if (blob === undefined) {
+    return undefined
+  }
+
   // no need to apply gc because we load existing document
   // it is either already gc-ed, or gc not needed and it is disabled
   ydoc ??= new YDoc({ gc: false })
 
-  try {
   const buffer = await storageAdapter.read(ctx, workspace, minioDocumentId)
   return yDocFromBuffer(Buffer.concat(buffer), ydoc)
-  } catch (err: any) {
-    if (err?.code === 'NoSuchKey' || err?.code === 'NotFound') {
-      return undefined
-    }
-    throw err
-  }
 }
 
 /** @public */

server/collaborator/src/extensions/storage.ts

@@ -107,8 +107,8 @@ export class StorageExtension implements Extension {
         return await adapter.loadDocument(ctx, documentId, context)
       })
     } catch (err) {
-      ctx.error('failed to load document content', { documentId, error: err })
+      ctx.error('failed to load document', { documentId, error: err })
-      return undefined
+      throw new Error('Failed to load document')
     }
   }
 
@@ -120,8 +120,8 @@ export class StorageExtension implements Extension {
         await adapter.saveDocument(ctx, documentId, document, context)
       })
     } catch (err) {
-      ctx.error('failed to save document content', { documentId, error: err })
+      ctx.error('failed to save document', { documentId, error: err })
-      return undefined
+      throw new Error('Failed to save document')
     }
   }
 }

server/collaborator/src/storage/platform.ts

@@ -52,7 +52,6 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
   ) {}
 
   async loadDocument (ctx: MeasureContext, documentId: DocumentId, context: Context): Promise<YDoc | undefined> {
-    try {
     // try to load document content
     try {
       ctx.info('load document content', { documentId })
@@ -63,6 +62,7 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
       }
     } catch (err) {
       ctx.error('failed to load document content', { documentId, error: err })
+      throw err
     }
 
     // then try to load from inital content
@@ -81,6 +81,7 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
         }
       } catch (err) {
         ctx.error('failed to load initial document content', { documentId, initialContentId, error: err })
+        throw err
       }
     }
 
@@ -93,6 +94,7 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
           return await this.loadDocumentFromPlatform(ctx, platformDocumentId, context)
         } catch (err) {
           ctx.error('failed to load platform document', { documentId, platformDocumentId, error: err })
+          throw err
         }
       })
 
@@ -107,9 +109,6 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
 
     // nothing found
     return undefined
-    } catch (err) {
-      ctx.error('failed to load document', { documentId, error: err })
-    }
   }
 
   async saveDocument (ctx: MeasureContext, documentId: DocumentId, document: YDoc, context: Context): Promise<void> {
@@ -133,6 +132,9 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
         await this.saveDocumentToStorage(ctx, documentId, document, context)
       } catch (err) {
         ctx.error('failed to save document', { documentId, error: err })
+        // raise an error if failed to save document to storage
+        // this will prevent document from being unloaded from memory
+        throw err
       }
 
       const { platformDocumentId } = context
@@ -166,12 +168,9 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
     const adapter = this.getStorageAdapter(storage)
 
     return await ctx.with('load-document', { storage }, async (ctx) => {
-      try {
+      return await withRetry(ctx, 5, async () => {
         return await loadCollaborativeDoc(adapter, context.workspaceId, collaborativeDoc, ctx)
-      } catch (err) {
+      })
-        ctx.error('failed to load storage document', { documentId, collaborativeDoc, error: err })
-        return undefined
-      }
     })
   }
 
@@ -185,8 +184,10 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
     const adapter = this.getStorageAdapter(storage)
 
     await ctx.with('save-document', {}, async (ctx) => {
+      await withRetry(ctx, 5, async () => {
         await saveCollaborativeDoc(adapter, context.workspaceId, collaborativeDoc, document, ctx)
       })
+    })
   }
 
   async takeSnapshot (
@@ -291,3 +292,25 @@ export class PlatformStorageAdapter implements CollabStorageAdapter {
     }
   }
 }
+
+async function withRetry<T> (
+  ctx: MeasureContext,
+  retries: number,
+  op: () => Promise<T>,
+  delay: number = 100
+): Promise<T> {
+  let error: any
+  while (retries > 0) {
+    retries--
+    try {
+      return await op()
+    } catch (err: any) {
+      error = err
+      ctx.error('error', err)
+      if (retries !== 0) {
+        await new Promise((resolve) => setTimeout(resolve, delay))
+      }
+    }
+  }
+  throw error
+}