UBERF-11409 Postpone collab doc save on storage failure

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

server/collaborator/src/extensions/storage.ts

@@ -1,5 +1,5 @@
 //
-// Copyright © 2023 Hardcore Engineering Inc.
+// Copyright © 2025 Hardcore Engineering Inc.
 //
 // Licensed under the Eclipse Public License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License. You may
@@ -18,9 +18,11 @@ import { type Markup, MeasureContext } from '@hcengineering/core'
 import {
   Document,
   Extension,
+  Hocuspocus,
   afterLoadDocumentPayload,
   afterUnloadDocumentPayload,
   onChangePayload,
+  onConfigurePayload,
   onConnectPayload,
   onDisconnectPayload,
   onLoadDocumentPayload,
@@ -35,6 +37,7 @@ export interface StorageConfiguration {
   ctx: MeasureContext
   adapter: CollabStorageAdapter
   transformer: Transformer
+  retryIntervalMs?: number
 }
 
 type DocumentName = string
@@ -50,9 +53,26 @@ export class StorageExtension implements Extension {
   private readonly configuration: StorageConfiguration
   private readonly updates = new Map<DocumentName, DocumentUpdates>()
   private readonly markups = new Map<DocumentName, Record<Markup, Markup>>()
+  private readonly failedDocuments = new Map<DocumentName, Context>()
+  private readonly retryInterval
+  private instance: Hocuspocus | undefined
 
   constructor (configuration: StorageConfiguration) {
     this.configuration = configuration
+
+    const retryIntervalMs = configuration.retryIntervalMs ?? 1000 * 60
+    this.retryInterval = setInterval(() => {
+      void this.retrySaveDocuments()
+    }, retryIntervalMs)
+  }
+
+  async onDestroy (): Promise<any> {
+    clearInterval(this.retryInterval)
+    await this.retrySaveDocuments()
+  }
+
+  async onConfigure ({ instance }: onConfigurePayload): Promise<any> {
+    this.instance = instance
   }
 
   async onChange ({ context, document, documentName }: withContext<onChangePayload>): Promise<any> {
@@ -108,8 +128,16 @@ export class StorageExtension implements Extension {
       return
     }
 
-    updates.collaborators.clear()
+    const now = Date.now()
+
     await this.storeDocument(documentName, document, updates.context)
+
+    // Remove collaborators that were not updated from before save
+    for (const [connectionId, updatedAt] of updates.collaborators.entries()) {
+      if (updatedAt < now) {
+        updates.collaborators.delete(connectionId)
+      }
+    }
   }
 
   async onConnect ({ context, documentName, instance }: withContext<onConnectPayload>): Promise<any> {
@@ -138,14 +166,23 @@ export class StorageExtension implements Extension {
       return
     }
 
-    updates.collaborators.clear()
-    await this.storeDocument(documentName, document, context)
+    const now = Date.now()
+
+    await this.storeDocument(documentName, document, updates.context)
+
+    // Remove collaborators that were not updated from before save
+    for (const [connectionId, updatedAt] of updates.collaborators.entries()) {
+      if (updatedAt < now) {
+        updates.collaborators.delete(connectionId)
+      }
+    }
   }
 
   async afterUnloadDocument ({ documentName }: afterUnloadDocumentPayload): Promise<any> {
     this.configuration.ctx.info('unload document', { documentName })
     this.updates.delete(documentName)
     this.markups.delete(documentName)
+    this.failedDocuments.delete(documentName)
   }
 
   private async loadDocument (documentName: string, context: Context): Promise<YDoc | undefined> {
@@ -175,9 +212,68 @@ export class StorageExtension implements Extension {
 
       this.markups.set(documentName, currMarkup ?? {})
     } catch (err: any) {
+      this.failedDocuments.set(documentName, context)
+
       Analytics.handleError(err)
       ctx.error('failed to save document', { documentName, error: err })
-      throw new Error('Failed to save document')
+      throw new Error(`Failed to save document ${documentName}`)
     }
   }
+
+  private async retrySaveDocuments (): Promise<void> {
+    const ctx = this.configuration.ctx
+
+    const count = this.failedDocuments.size
+    if (count === 0) {
+      return
+    }
+
+    ctx.info('retry failed documents', { count })
+
+    const hocuspocus = this.instance
+    if (hocuspocus === undefined) {
+      ctx.warn('instance is not set, cannot retry failed documents')
+      return
+    }
+
+    const promises: Promise<void>[] = []
+
+    for (const [documentName, context] of this.failedDocuments.entries()) {
+      const document = hocuspocus.documents.get(documentName)
+
+      if (document === undefined) {
+        ctx.warn('document not found', { documentName })
+        this.failedDocuments.delete(documentName)
+        continue
+      }
+
+      const connections = document.getConnectionsCount()
+      if (connections > 0) {
+        // Someone is still connected to the document
+        // We will retry later, when onStoreDocument or onDisconnect hook is called
+        ctx.info('document is connected, skipping', { documentName, connections })
+        this.failedDocuments.delete(documentName)
+        continue
+      }
+
+      promises.push(
+        ctx.with('retry-failed-document', {}, async (ctx) => {
+          try {
+            await this.storeDocument(documentName, document, context)
+            this.failedDocuments.delete(documentName)
+            if (document.getConnectionsCount() === 0) {
+              await hocuspocus.unloadDocument(document)
+            }
+            ctx.info('successfully retried save document', { documentName })
+          } catch (err: any) {
+            ctx.error('failed to retry save document', { documentName, error: err })
+          }
+        })
+      )
+    }
+
+    await ctx.with('retry-failed', {}, async () => {
+      await Promise.all(promises)
+    })
+  }
 }