UBERF-8150: Update to fresh mongo (#6603)

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

.vscode/launch.json

@@ -242,7 +242,7 @@
       "name": "Debug tool upgrade",
       "type": "node",
       "request": "launch",
-      "args": ["src/__start.ts", "create-workspace", "sanity-ws", "-w sanity-ws"],
+      "args": ["src/__start.ts", "upgrade", "--force"],
       "env": {
         "SERVER_SECRET": "secret",
         "MINIO_ACCESS_KEY": "minioadmin",

dev/tool/package.json

@@ -155,7 +155,7 @@
     "got": "^11.8.3",
     "libphonenumber-js": "^1.9.46",
     "mime-types": "~2.1.34",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "pg": "8.12.0",
     "ws": "^8.18.0"
   }

dev/tool/src/index.ts

@@ -86,7 +86,7 @@ import core, {
 } from '@hcengineering/core'
 import { consoleModelLogger, type MigrateOperation } from '@hcengineering/model'
 import contact from '@hcengineering/model-contact'
-import { getMongoClient, getWorkspaceDB } from '@hcengineering/mongo'
+import { getMongoClient, getWorkspaceDB, shutdown } from '@hcengineering/mongo'
 import type { StorageAdapter, StorageAdapterEx } from '@hcengineering/server-core'
 import { deepEqual } from 'fast-equals'
 import { createWriteStream, readFileSync } from 'fs'
@@ -205,6 +205,7 @@ export function devTool (
     }
     client.close()
     console.log(`closing database connection to '${uri}'...`)
+    await shutdown()
   }
 
   async function withStorage (mongodbUri: string, f: (storageAdapter: StorageAdapter) => Promise<any>): Promise<void> {
@@ -552,11 +553,12 @@ export function devTool (
         const measureCtx = new MeasureMetricsContext('upgrade', {})
 
         for (const ws of workspaces) {
-          try {
-            const logger = cmd.console
-              ? consoleModelLogger
-              : new FileModelLogger(path.join(cmd.logs, `${ws.workspace}.log`))
+          console.warn('UPGRADING', ws.workspaceName)
+          const logger = cmd.console
+            ? consoleModelLogger
+            : new FileModelLogger(path.join(cmd.logs, `${ws.workspace}.log`))
 
+          try {
             await upgradeWorkspace(
               measureCtx,
               version,
@@ -577,10 +579,10 @@ export function devTool (
               attempts: 0
             })
           } catch (err: any) {
-            console.error(err)
+            toolCtx.error('failed to upgrade', { err, workspace: ws.workspace, workspaceName: ws.workspaceName })
+            continue
           }
         }
-
         console.log('upgrade done')
       })
     })

packages/core/src/measurements/context.ts

@@ -141,10 +141,9 @@ export class MeasureMetricsContext implements MeasureContext {
       const value = op(c)
       if (value != null && value instanceof Promise) {
         needFinally = false
-        void value.finally(() => {
+        return value.finally(() => {
           c.end()
         })
-        return value
       } else {
         return Promise.resolve(value)
       }

pods/account/package.json

@@ -57,7 +57,7 @@
     "@hcengineering/platform": "^0.6.11",
     "@hcengineering/auth-providers": "^0.6.0",
     "@hcengineering/core": "^0.6.32",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "koa": "^2.15.3",
     "koa-router": "^12.0.1",
     "koa-bodyparser": "^4.4.1",

pods/authProviders/package.json

@@ -46,7 +46,7 @@
     "@types/querystringify": "^2.0.2"
   },
   "dependencies": {
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "@hcengineering/core": "^0.6.32",
     "@hcengineering/account": "^0.6.0",
     "passport-custom": "~1.1.1",

pods/workspace/package.json

@@ -51,7 +51,7 @@
     "@hcengineering/workspace-service": "^0.6.0",
     "@hcengineering/platform": "^0.6.11",
     "@hcengineering/core": "^0.6.32",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "@hcengineering/server-token": "^0.6.11",
     "@hcengineering/server-core": "^0.6.1",
     "@hcengineering/model-all": "^0.6.0",

server-plugins/ai-bot-resources/package.json

@@ -50,6 +50,6 @@
     "@hcengineering/ai-bot": "^0.6.0",
     "@hcengineering/server-ai-bot": "^0.6.0",
     "@hcengineering/analytics-collector": "^0.6.0",
-    "mongodb": "^6.8.0"
+    "mongodb": "^6.9.0"
   }
 }

server/account-service/package.json

@@ -48,7 +48,7 @@
     "@hcengineering/auth-providers": "^0.6.0",
     "@hcengineering/core": "^0.6.32",
     "@hcengineering/mongo": "^0.6.1",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "koa": "^2.15.3",
     "koa-router": "^12.0.1",
     "koa-bodyparser": "^4.4.1",

server/account/package.json

@@ -40,7 +40,7 @@
     "@types/otp-generator": "^4.0.2"
   },
   "dependencies": {
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "@hcengineering/platform": "^0.6.11",
     "@hcengineering/core": "^0.6.32",
     "@hcengineering/contact": "^0.6.24",

server/collaborator/package.json

@@ -66,7 +66,7 @@
     "@hocuspocus/transformer": "^2.13.5",
     "@tiptap/core": "^2.2.4",
     "@tiptap/html": "^2.2.4",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "yjs": "^13.5.52",
     "y-prosemirror": "^1.2.1",
     "express": "^4.19.2",

server/core/src/fulltext.ts

@@ -72,7 +72,9 @@ export class FullTextIndex implements WithFind {
   }
 
   async close (): Promise<void> {
-    await this.indexer.cancel()
+    if (!this.upgrade) {
+      await this.indexer.cancel()
+    }
   }
 
   async tx (ctx: MeasureContext, txes: Tx[]): Promise<TxResult> {

server/core/src/pipeline.ts

@@ -74,11 +74,18 @@ class PipelineImpl implements Pipeline {
     let current: Middleware | undefined
     for (let index = constructors.length - 1; index >= 0; index--) {
       const element = constructors[index]
-      const newCur = await element(ctx, context, current)
-      if (newCur != null) {
-        this.middlewares.push(newCur)
+      try {
+        const newCur = await element(ctx, context, current)
+        if (newCur != null) {
+          this.middlewares.push(newCur)
+        }
+        current = newCur ?? current
+      } catch (err: any) {
+        ctx.error('failed to initialize pipeline', { err, workspace: context.workspace.name })
+        // We need to call close for all items.
+        await this.close()
+        throw err
       }
-      current = newCur ?? current
     }
     this.middlewares.reverse()
 

server/middleware/src/model.ts

@@ -46,8 +46,8 @@ export class ModelMiddleware extends BaseMiddleware implements Middleware {
     }
     const txAdapter = this.context.adapterManager.getAdapter(DOMAIN_TX, true) as TxAdapter
 
-    this.model = await ctx.with('get model', {}, async (ctx) => {
-      const model = await ctx.with('fetch-model', {}, async (ctx) => await txAdapter.getModel(ctx))
+    this.model = await ctx.with('get-model', {}, async (ctx) => {
+      const model = await ctx.with('fetch-model', {}, (ctx) => txAdapter.getModel(ctx))
       for (const tx of model) {
         try {
           this.context.hierarchy.tx(tx)

server/mongo/package.json

@@ -37,7 +37,7 @@
     "@hcengineering/core": "^0.6.32",
     "@hcengineering/platform": "^0.6.11",
     "@hcengineering/server-core": "^0.6.1",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "bson": "^6.8.0"
   }
 }

server/mongo/src/utils.ts

@@ -81,7 +81,10 @@ class MongoClientReferenceImpl {
       }
       this.onclose()
       void (async () => {
-        const cl = await this.client
+        let cl = this.client
+        if (cl instanceof Promise) {
+          cl = await cl
+        }
         await cl.close()
       })()
     }

server/server-pipeline/src/pipeline.ts

@@ -1,16 +1,16 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 import {
-  type Branding,
   DOMAIN_BENCHMARK,
   DOMAIN_BLOB,
   DOMAIN_FULLTEXT_BLOB,
   DOMAIN_MODEL,
   DOMAIN_TRANSIENT,
   DOMAIN_TX,
-  type WorkspaceIdWithUrl,
   Hierarchy,
   ModelDb,
-  type MeasureContext
+  type Branding,
+  type MeasureContext,
+  type WorkspaceIdWithUrl
 } from '@hcengineering/core'
 import { createElasticAdapter, createElasticBackupDataAdapter } from '@hcengineering/elastic'
 import {
@@ -35,8 +35,8 @@ import {
   TriggersMiddleware,
   TxMiddleware
 } from '@hcengineering/middleware'
-import { createPostgresAdapter, createPostgresTxAdapter } from '@hcengineering/postgres'
 import { createMongoAdapter, createMongoTxAdapter } from '@hcengineering/mongo'
+import { createPostgresAdapter, createPostgresTxAdapter } from '@hcengineering/postgres'
 import {
   buildStorageFromConfig,
   createNullAdapter,
@@ -49,18 +49,18 @@ import {
   createBenchmarkAdapter,
   createInMemoryAdapter,
   createPipeline,
-  type Middleware,
-  type DbAdapterFactory,
+  DummyFullTextAdapter,
   FullTextMiddleware,
+  type AggregatorStorageAdapter,
+  type DbAdapterFactory,
   type DbConfiguration,
+  type Middleware,
   type MiddlewareCreator,
+  type Pipeline,
   type PipelineContext,
   type PipelineFactory,
   type StorageAdapter,
-  type Pipeline,
-  type StorageConfiguration,
-  DummyFullTextAdapter,
-  type AggregatorStorageAdapter
+  type StorageConfiguration
 } from '@hcengineering/server-core'
 import { createIndexStages } from './indexing'
 
@@ -200,9 +200,14 @@ export async function getServerPipeline (
     }
   )
 
-  return {
-    pipeline: await pipelineFactory(ctx, wsUrl, true, () => {}, null),
-    storageAdapter
+  try {
+    return {
+      pipeline: await pipelineFactory(ctx, wsUrl, true, () => {}, null),
+      storageAdapter
+    }
+  } catch (err: any) {
+    await storageAdapter.close()
+    throw err
   }
 }
 

server/tool/package.json

@@ -40,7 +40,7 @@
     "@types/js-yaml": "^4.0.9"
   },
   "dependencies": {
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "@hcengineering/platform": "^0.6.11",
     "@hcengineering/core": "^0.6.32",
     "@hcengineering/contact": "^0.6.24",

server/tool/src/index.ts

@@ -253,7 +253,7 @@ export async function upgradeModel (
   workspaceId: WorkspaceIdWithUrl,
   txes: Tx[],
   pipeline: Pipeline,
-  storageAdapter: AggregatorStorageAdapter,
+  storageAdapter: StorageAdapter,
   migrateOperations: [string, MigrateOperation][],
   logger: ModelLogger = consoleModelLogger,
   skipTxUpdate: boolean = false,
@@ -413,9 +413,7 @@ export async function upgradeModel (
       let i = 0
       for (const op of migrateOperations) {
         const t = Date.now()
-        await ctx.with(op[0], {}, async () => {
-          await op[1].upgrade(migrateState, getUpgradeClient, logger)
-        })
+        await ctx.with(op[0], {}, () => op[1].upgrade(migrateState, getUpgradeClient, logger))
         logger.log('upgrade:', { operation: op[0], time: Date.now() - t, workspaceId: workspaceId.name })
         await progress(60 + ((100 / migrateOperations.length) * i * 30) / 100)
         i++
@@ -474,11 +472,11 @@ export async function fetchModel (
   const modelDb = new ModelDb(hierarchy)
 
   if (model === undefined) {
-    const res = await ctx.with('get-model', {}, async (ctx) => await pipeline.loadModel(ctx, 0))
+    const res = await ctx.with('load-model', {}, (ctx) => pipeline.loadModel(ctx, 0))
     model = Array.isArray(res) ? res : res.transactions
   }
 
-  await ctx.with('build local model', {}, async () => {
+  ctx.withSync('build local model', {}, () => {
     for (const tx of model ?? []) {
       try {
         hierarchy.tx(tx)

server/workspace-service/src/ws-operations.ts

@@ -21,6 +21,7 @@ import {
   SessionDataImpl,
   type Pipeline,
   type PipelineFactory,
+  type StorageAdapter,
   type StorageConfiguration
 } from '@hcengineering/server-core'
 import {
@@ -284,22 +285,23 @@ export async function upgradeWorkspace (
     workspaceName: ws.workspaceName ?? '',
     workspaceUrl: ws.workspaceUrl ?? ''
   }
-
-  const { pipeline, storageAdapter } = await getServerPipeline(ctx, mongodbUri, dbUrl, wsUrl)
-  const contextData = new SessionDataImpl(
-    systemAccountEmail,
-    'backup',
-    true,
-    { targets: {}, txes: [] },
-    wsUrl,
-    null,
-    false,
-    new Map(),
-    new Map(),
-    pipeline.context.modelDb
-  )
-  ctx.contextData = contextData
+  let pipeline: Pipeline | undefined
+  let storageAdapter: StorageAdapter | undefined
   try {
+    ;({ pipeline, storageAdapter } = await getServerPipeline(ctx, mongodbUri, dbUrl, wsUrl))
+    const contextData = new SessionDataImpl(
+      systemAccountEmail,
+      'backup',
+      true,
+      { targets: {}, txes: [] },
+      wsUrl,
+      null,
+      false,
+      new Map(),
+      new Map(),
+      pipeline.context.modelDb
+    )
+    ctx.contextData = contextData
     await handleWsEvent?.('upgrade-started', version, 0)
 
     await upgradeModel(
@@ -322,9 +324,10 @@ export async function upgradeWorkspace (
   } catch (err: any) {
     ctx.error('upgrade-failed', { message: err.message })
     await handleWsEvent?.('ping', version, 0, `Upgrade failed: ${err.message}`)
+    throw err
   } finally {
-    await pipeline.close()
-    await storageAdapter.close()
+    await pipeline?.close()
+    await storageAdapter?.close()
     clearInterval(updateProgressHandle)
   }
 }

services/ai-bot/pod-ai-bot/package.json

@@ -79,7 +79,7 @@
     "express": "^4.19.2",
     "fast-equals": "^5.0.1",
     "form-data": "^4.0.0",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "node-fetch": "^2.6.6",
     "openai": "^4.56.0",
     "tiktoken": "^1.0.16",

services/analytics-collector/pod-analytics-collector/package.json

@@ -101,7 +101,7 @@
     "cors": "^2.8.5",
     "dotenv": "~16.0.0",
     "express": "^4.19.2",
-    "mongodb": "^6.8.0",
+    "mongodb": "^6.9.0",
     "puppeteer": "^22.6.1",
     "ws": "^8.18.0"
   }