Reindexing fixes

Signed-off-by: Andrey Sobolev <[email protected]>

Author: haiodo

.vscode/launch.json

@@ -158,17 +158,17 @@
         // "PORT": "4700",// For mongo
         "PORT": "4710", // for cockroach
         "FULLTEXT_DB_URL": "http://localhost:9201",
-        "DB_URL": "mongodb://localhost:27018",
+        // "DB_URL": "mongodb://localhost:27018",
         // "DB_URL": "postgresql://postgres:example@localhost:5432",
-        // "DB_URL": "postgresql://[email protected]:26257/defaultdb?sslmode=disable",
+        "DB_URL": "postgresql://[email protected]:26258/defaultdb?sslmode=disable",
         "STORAGE_CONFIG": "minio|localhost?accessKey=minioadmin&secretKey=minioadmin",
         "SERVER_SECRET": "secret",
         "REKONI_URL": "http://localhost:4004",
         "MODEL_JSON": "${workspaceRoot}/models/all/bundle/model.json",
         "ELASTIC_INDEX_NAME": "local_storage_index",
         "REGION": "",
         "STATS_URL": "http://huly.local:4901",
-        "ACCOUNTS_URL": "http://localhost:3003",
+        "ACCOUNTS_URL": "http://huly.local:3003",
         "QUEUE_CONFIG": "localhost:19093;-staging"
       },
       "runtimeArgs": ["--nolazy", "-r", "ts-node/register"],

dev/tool/src/index.ts

@@ -19,7 +19,6 @@ import accountPlugin, {
   createWorkspaceRecord,
   flattenStatus,
   getAccountDB,
-  getWorkspaceById,
   getWorkspaceInfoWithStatusById,
   getWorkspaces,
   getWorkspacesInfoWithStatusByIds,
@@ -117,9 +116,9 @@ import { performGmailAccountMigrations } from './gmail'
 import { getToolToken, getWorkspace, getWorkspaceTransactorEndpoint } from './utils'
 
 import { createRestClient } from '@hcengineering/api-client'
-import { mkdir, writeFile } from 'fs/promises'
-import { basename, dirname } from 'path'
 import { existsSync } from 'fs'
+import { mkdir, writeFile } from 'fs/promises'
+import { dirname } from 'path'
 import { restoreMarkupRefs } from './markup'
 
 const colorConstants = {
@@ -438,58 +437,67 @@ export function devTool (
   //   })
   // })
 
-  program
-    .command('upgrade-workspace <name>')
-    .description('upgrade workspace')
-    .option('-f|--force [force]', 'Force update', true)
-    .option('-i|--indexes [indexes]', 'Force indexes rebuild', false)
-    .action(async (workspace, cmd: { force: boolean, indexes: boolean }) => {
-      const { version, txes, migrateOperations } = prepareTools()
+  async function doUpgrade (
+    toolCtx: MeasureMetricsContext,
+    workspace: WorkspaceUuid,
+    forceUpdate: boolean,
+    forceIndexes: boolean
+  ): Promise<void> {
+    const { version, txes, migrateOperations } = prepareTools()
 
-      await withAccountDatabase(async (db) => {
-        const info = await getWorkspace(db, workspace)
-        if (info === null) {
-          throw new Error(`workspace ${workspace} not found`)
-        }
+    await withAccountDatabase(async (db) => {
+      const info = await getWorkspace(db, workspace)
+      if (info === null) {
+        throw new Error(`workspace ${workspace} not found`)
+      }
 
-        const wsInfo = await getWorkspaceInfoWithStatusById(db, info.uuid)
-        if (wsInfo === null) {
-          throw new Error(`workspace ${workspace} not found`)
-        }
+      const wsInfo = await getWorkspaceInfoWithStatusById(db, info.uuid)
+      if (wsInfo === null) {
+        throw new Error(`workspace ${workspace} not found`)
+      }
 
-        const coreWsInfo = flattenStatus(wsInfo)
-        const measureCtx = new MeasureMetricsContext('upgrade-workspace', {})
-        const accountClient = getAccountClient(getToolToken(wsInfo.uuid))
-        const queue = getPlatformQueue('tool', info.region)
-        const wsProducer = queue.getProducer<QueueWorkspaceMessage>(toolCtx, QueueTopic.Workspace)
-        await upgradeWorkspace(
-          measureCtx,
-          version,
-          txes,
-          migrateOperations,
-          accountClient,
-          coreWsInfo,
-          consoleModelLogger,
-          wsProducer,
-          async () => {},
-          cmd.force,
-          cmd.indexes,
-          true
-        )
+      const coreWsInfo = flattenStatus(wsInfo)
+      const measureCtx = new MeasureMetricsContext('upgrade-workspace', {})
+      const accountClient = getAccountClient(getToolToken(wsInfo.uuid))
+      const queue = getPlatformQueue('tool', info.region)
+      const wsProducer = queue.getProducer<QueueWorkspaceMessage>(toolCtx, QueueTopic.Workspace)
+      await upgradeWorkspace(
+        measureCtx,
+        version,
+        txes,
+        migrateOperations,
+        accountClient,
+        coreWsInfo,
+        consoleModelLogger,
+        wsProducer,
+        async () => {},
+        forceUpdate,
+        forceIndexes,
+        true
+      )
 
-        await updateWorkspaceInfo(measureCtx, db, null, getToolToken(), {
-          workspaceUuid: info.uuid,
-          event: 'upgrade-done',
-          version,
-          progress: 100
-        })
+      await updateWorkspaceInfo(measureCtx, db, null, getToolToken(), {
+        workspaceUuid: info.uuid,
+        event: 'upgrade-done',
+        version,
+        progress: 100
+      })
 
-        console.log(metricsToString(measureCtx.metrics, 'upgrade', 60))
+      console.log(metricsToString(measureCtx.metrics, 'upgrade', 60))
 
-        await wsProducer.send(info.uuid, [workspaceEvents.upgraded()])
-        await queue.shutdown()
-        console.log('upgrade-workspace done')
-      })
+      await wsProducer.send(info.uuid, [workspaceEvents.upgraded()])
+      await queue.shutdown()
+      console.log('upgrade-workspace done')
+    })
+  }
+
+  program
+    .command('upgrade-workspace <name>')
+    .description('upgrade workspace')
+    .option('-f|--force [force]', 'Force update', true)
+    .option('-i|--indexes [indexes]', 'Force indexes rebuild', false)
+    .action(async (workspace, cmd: { force: boolean, indexes: boolean }) => {
+      await doUpgrade(toolCtx, workspace, cmd.force, cmd.indexes)
     })
 
   // program
@@ -1133,6 +1141,7 @@ export function devTool (
     .option('-i, --include <include>', 'A list of ; separated domain names to include during backup', '*')
     .option('-s, --skip <skip>', 'A list of ; separated domain names to skip during backup', '')
     .option('--use-storage <useStorage>', 'Use workspace storage adapter from env variable', '')
+    .option('--upgrade', 'Upgrade workspace', false)
     .option(
       '--history-file <historyFile>',
       'Store blob send info into file. Will skip already send documents.',
@@ -1152,6 +1161,7 @@ export function devTool (
           skip: string
           useStorage: string
           historyFile: string
+          upgrade: boolean
         }
       ) => {
         await withAccountDatabase(async (db) => {
@@ -1181,6 +1191,11 @@ export function devTool (
             storageAdapter: workspaceStorage,
             historyFile: cmd.historyFile
           })
+
+          if (cmd.upgrade) {
+            await doUpgrade(toolCtx, workspace, true, true)
+          }
+
           const queue = getPlatformQueue('tool', ws.region)
           const wsProducer = queue.getProducer<QueueWorkspaceMessage>(toolCtx, QueueTopic.Workspace)
           await wsProducer.send(ws.uuid, [workspaceEvents.fullReindex()])

models/core/src/migration.ts

@@ -236,8 +236,13 @@ async function processMigrateContentFor (
             try {
               const buffer = Buffer.from(value)
               await storageAdapter.put(client.ctx, client.wsIds, blobId, buffer, 'application/json', buffer.length)
-            } catch (err) {
-              client.logger.error('failed to process document', { _class: doc._class, _id: doc._id, err })
+            } catch (err: any) {
+              client.logger.error('failed to process document', {
+                _class: doc._class,
+                _id: doc._id,
+                err: err.message,
+                stack: err.stack
+              })
             }
 
             update[attributeName] = blobId

packages/account-client/src/client.ts

@@ -267,7 +267,7 @@ class AccountClientImpl implements AccountClient {
       return ws
     }
 
-    const result = { ...ws, ...status }
+    const result = { ...ws, ...status, processingAttemps: status.processingAttempts ?? 0 }
     delete result.status
 
     return result

packages/core/src/classes.ts

@@ -839,6 +839,7 @@ export interface WorkspaceInfoWithStatus extends WorkspaceInfo {
   mode: WorkspaceMode
   processingProgress?: number
   backupInfo?: BackupStatus
+  processingAttemps: number
 }
 
 export interface WorkspaceMemberInfo {

pods/fulltext/src/__tests__/indexing.spec.ts

@@ -45,6 +45,7 @@ class TestWorkspaceManager extends WorkspaceManager {
       dataId: decodedToken.workspace as unknown as WorkspaceDataId,
       mode: 'active',
       processingProgress: 0,
+      processingAttemps: 0,
       backupInfo: {
         dataSize: 0,
         blobsSize: 0,

pods/fulltext/src/manager.ts

@@ -1,6 +1,16 @@
 /* eslint-disable @typescript-eslint/unbound-method */
-import type { Doc, MeasureContext, Space, Tx, TxCUD, WorkspaceInfoWithStatus, WorkspaceUuid } from '@hcengineering/core'
-import { Hierarchy, systemAccountUuid } from '@hcengineering/core'
+import type {
+  Doc,
+  MeasureContext,
+  Space,
+  Tx,
+  TxCreateDoc,
+  TxCUD,
+  Version,
+  WorkspaceInfoWithStatus,
+  WorkspaceUuid
+} from '@hcengineering/core'
+import core, { Hierarchy, systemAccountUuid, TxProcessor, versionToString } from '@hcengineering/core'
 import { getAccountClient, getTransactorEndpoint } from '@hcengineering/server-client'
 import {
   createContentAdapter,
@@ -50,6 +60,11 @@ export class WorkspaceManager {
     for (const tx of model) {
       this.sysHierarchy.tx(tx)
     }
+    this.supportedVersion = TxProcessor.createDoc2Doc(
+      model.find(
+        (it) => TxProcessor.isExtendsCUD(it._class) && (it as TxCUD<Doc>).objectClass === core.class.Version
+      ) as TxCreateDoc<Version>
+    )
 
     this.workspaceProducer = this.opt.queue.getProducer<QueueWorkspaceMessage>(this.ctx, QueueTopic.Workspace)
   }
@@ -64,6 +79,8 @@ export class WorkspaceManager {
 
   txInformer: any
 
+  supportedVersion: Version
+
   async startIndexer (): Promise<void> {
     this.contentAdapter = await this.ctx.with('create content adapter', {}, (ctx) =>
       createContentAdapter(this.opt.config.contentAdapters, this.opt.config.defaultContentAdapter)
@@ -125,7 +142,12 @@ export class WorkspaceManager {
       }
 
       const indexer = await this.getIndexer(this.ctx, ws, token, true)
-      await indexer?.fulltext.processDocuments(this.ctx, m.value, control)
+      // TODO: If workspace is not upgraded, we will loose the fulltext index
+      if (indexer === undefined) {
+        // this.workspaceProducer
+      } else {
+        await indexer?.fulltext.processDocuments(this.ctx, m.value, control)
+      }
     }
   }
 
@@ -159,6 +181,10 @@ export class WorkspaceManager {
               ws,
               classes.map((it) => workspaceEvents.reindex(it.domain, it.classes))
             )
+          } else {
+            await this.workspaceProducer.send(ws, [workspaceEvents.fullReindex()]).catch((err) => {
+              this.ctx.error('error', { err })
+            })
           }
         } else if (
           mm.type === QueueWorkspaceEvent.Deleted ||
@@ -176,6 +202,9 @@ export class WorkspaceManager {
           const indexer = await this.getIndexer(this.ctx, ws, token, true)
           const mmd = mm as QueueWorkspaceReindexMessage
           await indexer?.reindex(this.ctx, mmd.domain, mmd.classes, control)
+        } else if (mm.type === QueueWorkspaceEvent.Upgraded) {
+          console.error('Upgraded', this.supportedVersion)
+          await this.closeWorkspace(ws)
         }
       }
     }
@@ -195,20 +224,38 @@ export class WorkspaceManager {
     return (await getTransactorEndpoint(token, 'internal')).replace('wss://', 'https://').replace('ws://', 'http://')
   }
 
-  async getIndexer (
+  async createIndexer (
     ctx: MeasureContext,
     workspace: WorkspaceUuid,
-    token: string | undefined,
-    create: boolean = false
-  ): Promise<WorkspaceIndexer | undefined> {
-    let idx = this.indexers.get(workspace)
-    if (idx === undefined && create) {
+    token: string | undefined
+  ): Promise<WorkspaceIndexer> {
+    while (true) {
       const workspaceInfo = await this.getWorkspaceInfo(ctx, token)
       if (workspaceInfo === undefined) {
-        return
+        throw new Error('Workspace not found')
+      }
+      if (
+        workspaceInfo.versionMajor !== this.supportedVersion.major ||
+        workspaceInfo.versionMinor !== this.supportedVersion.minor ||
+        workspaceInfo.versionPatch !== this.supportedVersion.patch
+      ) {
+        ctx.warn('wrong version', {
+          workspace,
+          version: versionToString({
+            major: workspaceInfo.versionMajor,
+            minor: workspaceInfo.versionMinor,
+            patch: workspaceInfo.versionPatch
+          }),
+          supportedVersion: this.supportedVersion
+        })
+        if (workspaceInfo.processingAttemps < 4) {
+          await new Promise((resolve) => setTimeout(resolve, 10000))
+          continue
+        }
+        throw new Error('Workspace limit reached')
       }
       ctx.warn('indexer created', { workspace })
-      idx = WorkspaceIndexer.create(
+      return await WorkspaceIndexer.create(
         ctx,
         this.model,
         {
@@ -223,12 +270,32 @@ export class WorkspaceManager {
         (token) => this.getTransactorAPIEndpoint(token),
         this.opt.listener
       )
-      this.indexers.set(workspace, idx)
     }
-    if (idx instanceof Promise) {
-      idx = await idx
+  }
+
+  async getIndexer (
+    ctx: MeasureContext,
+    workspace: WorkspaceUuid,
+    token: string | undefined,
+    create: boolean = false
+  ): Promise<WorkspaceIndexer | undefined> {
+    let idx = this.indexers.get(workspace)
+    if (idx === undefined && create) {
+      idx = this.createIndexer(ctx, workspace, token)
       this.indexers.set(workspace, idx)
     }
+    if (idx === undefined) {
+      return undefined
+    }
+    try {
+      if (idx instanceof Promise) {
+        idx = await idx
+        this.indexers.set(workspace, idx)
+      }
+    } catch (err: any) {
+      this.indexers.delete(workspace)
+      return undefined
+    }
     return idx
   }
 

pods/fulltext/src/server.ts

@@ -221,6 +221,7 @@ export async function startIndexer (
       req.body = {}
 
       ctx.info('reindex', { workspace: decoded.workspace })
+      await manager.closeWorkspace(decoded.workspace)
       const indexer = await manager.getIndexer(ctx, decoded.workspace, token, true)
       if (indexer !== undefined) {
         indexer.lastUpdate = Date.now()