UBERF-8899: Fix Reconnect performance (#7597)

Signed-off-by: Andrey Sobolev <haiodo@gmail.com>

packages/core/src/client.ts

@@ -21,12 +21,10 @@ import { Hierarchy } from './hierarchy'
 import { MeasureContext, MeasureMetricsContext } from './measurements'
 import { ModelDb } from './memdb'
 import type { DocumentQuery, FindOptions, FindResult, FulltextStorage, Storage, TxResult, WithLookup } from './storage'
-import { SearchOptions, SearchQuery, SearchResult, SortingOrder } from './storage'
-import { Tx, TxCUD } from './tx'
+import { SearchOptions, SearchQuery, SearchResult } from './storage'
+import { Tx, TxCUD, type TxWorkspaceEvent } from './tx'
 import { toFindResult } from './utils'
 
-const transactionThreshold = 500
-
 /**
  * @public
  */
@@ -85,11 +83,13 @@ export interface ClientConnection extends Storage, FulltextStorage, BackupClient
   isConnected: () => boolean
 
   close: () => Promise<void>
-  onConnect?: (event: ClientConnectEvent, data: any) => Promise<void>
+  onConnect?: (event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise<void>
 
   // If hash is passed, will return LoadModelResponse
   loadModel: (last: Timestamp, hash?: string) => Promise<Tx[] | LoadModelResponse>
   getAccount: () => Promise<Account>
+
+  getLastHash?: (ctx: MeasureContext) => Promise<string | undefined>
 }
 
 class ClientImpl implements AccountClient, BackupClient {
@@ -236,7 +236,7 @@ export async function createClient (
   let hierarchy = new Hierarchy()
   let model = new ModelDb(hierarchy)
 
-  let lastTx: number = 0
+  let lastTx: string | undefined
 
   function txHandler (...tx: Tx[]): void {
     if (tx == null || tx.length === 0) {
@@ -248,7 +248,11 @@ export async function createClient (
       // eslint-disable-next-line @typescript-eslint/no-floating-promises
       client.updateFromRemote(...tx)
     }
-    lastTx = tx.reduce((cur, it) => (it.modifiedOn > cur ? it.modifiedOn : cur), 0)
+    for (const t of tx) {
+      if (t._class === core.class.TxWorkspaceEvent) {
+        lastTx = (t as TxWorkspaceEvent).params.lastTx
+      }
+    }
   }
   const conn = await ctx.with('connect', {}, () => connect(txHandler))
 
@@ -264,11 +268,14 @@ export async function createClient (
   txHandler(...txBuffer)
   txBuffer = undefined
 
-  const oldOnConnect: ((event: ClientConnectEvent, data: any) => Promise<void>) | undefined = conn.onConnect
-  conn.onConnect = async (event, data) => {
+  const oldOnConnect:
+  | ((event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise<void>)
+  | undefined = conn.onConnect
+  conn.onConnect = async (event, _lastTx, data) => {
     console.log('Client: onConnect', event)
     if (event === ClientConnectEvent.Maintenance) {
-      await oldOnConnect?.(ClientConnectEvent.Maintenance, data)
+      lastTx = _lastTx
+      await oldOnConnect?.(ClientConnectEvent.Maintenance, _lastTx, data)
       return
     }
     // Find all new transactions and apply
@@ -282,51 +289,27 @@ export async function createClient (
       model = new ModelDb(hierarchy)
 
       await ctx.with('build-model', {}, (ctx) => buildModel(ctx, loadModelResponse, modelFilter, hierarchy, model))
-      await oldOnConnect?.(ClientConnectEvent.Upgraded, data)
-
+      await oldOnConnect?.(ClientConnectEvent.Upgraded, _lastTx, data)
+      lastTx = _lastTx
       // No need to fetch more stuff since upgrade was happened.
       return
     }
 
-    if (event === ClientConnectEvent.Connected) {
+    if (event === ClientConnectEvent.Connected && _lastTx !== lastTx && lastTx === undefined) {
       // No need to do anything here since we connected.
-      await oldOnConnect?.(event, data)
+      await oldOnConnect?.(event, _lastTx, data)
+      lastTx = _lastTx
       return
     }
 
-    // We need to look for last {transactionThreshold} transactions and if it is more since lastTx one we receive, we need to perform full refresh.
-    if (lastTx === 0) {
-      await oldOnConnect?.(ClientConnectEvent.Refresh, data)
+    if (_lastTx === lastTx) {
+      // Same lastTx, no need to refresh
+      await oldOnConnect?.(ClientConnectEvent.Reconnected, _lastTx, data)
       return
     }
-    const atxes = await ctx.with('find-atx', {}, () =>
-      conn.findAll(
-        core.class.Tx,
-        { modifiedOn: { $gt: lastTx }, objectSpace: { $ne: core.space.Model } },
-        { sort: { modifiedOn: SortingOrder.Ascending, _id: SortingOrder.Ascending }, limit: transactionThreshold }
-      )
-    )
-
-    let needFullRefresh = false
-    // if we have attachment document create/delete we need to full refresh, since some derived data could be missing
-    for (const tx of atxes) {
-      if (
-        (tx as TxCUD<Doc>).attachedTo !== undefined &&
-        (tx._class === core.class.TxCreateDoc || tx._class === core.class.TxRemoveDoc)
-      ) {
-        needFullRefresh = true
-        break
-      }
-    }
-
-    if (atxes.length < transactionThreshold && !needFullRefresh) {
-      console.log('applying input transactions', atxes.length)
-      txHandler(...atxes)
-      await oldOnConnect?.(ClientConnectEvent.Reconnected, data)
-    } else {
-      // We need to trigger full refresh on queries, etc.
-      await oldOnConnect?.(ClientConnectEvent.Refresh, data)
-    }
+    lastTx = _lastTx
+    // We need to trigger full refresh on queries, etc.
+    await oldOnConnect?.(ClientConnectEvent.Refresh, lastTx, data)
   }
 
   return client
@@ -344,6 +327,10 @@ async function tryLoadModel (
     hash: ''
   }
 
+  if (conn.getLastHash !== undefined && (await conn.getLastHash(ctx)) === current.hash) {
+    // We have same model hash.
+    return current
+  }
   const lastTxTime = getLastTxTime(current.transactions)
   const result = await ctx.with('connection-load-model', { hash: current.hash !== '' }, (ctx) =>
     conn.loadModel(lastTxTime, current.hash)
@@ -357,21 +344,23 @@ async function tryLoadModel (
       hash: ''
     }
   }
-
-  // Save concatenated
-  void ctx
-    .with('persistence-store', {}, (ctx) =>
-      persistence?.store({
-        ...result,
-        transactions: !result.full ? current.transactions.concat(result.transactions) : result.transactions
+  const transactions = current.transactions.concat(result.transactions)
+  if (result.hash !== current.hash) {
+    // Save concatenated, if have some more of them.
+    void ctx
+      .with('persistence-store', {}, (ctx) =>
+        persistence?.store({
+          ...result,
+          transactions: !result.full ? transactions : result.transactions
+        })
+      )
+      .catch((err) => {
+        Analytics.handleError(err)
       })
-    )
-    .catch((err) => {
-      Analytics.handleError(err)
-    })
+  }
 
   if (!result.full && !reload) {
-    result.transactions = current.transactions.concat(result.transactions)
+    result.transactions = transactions
   }
 
   return result

packages/core/src/tx.ts

@@ -51,7 +51,8 @@ export enum WorkspaceEvent {
   IndexingUpdate,
   SecurityChange,
   MaintenanceNotification,
-  BulkUpdate
+  BulkUpdate,
+  LastTx
 }
 
 /**

packages/ui/src/components/DropdownLabels.svelte

@@ -50,9 +50,11 @@
   let container: HTMLElement
   let opened: boolean = false
 
-  $: selectedItem = multiselect ? items.filter((p) => selected?.includes(p.id)) : items.find((x) => x.id === selected)
-  $: if (autoSelect && selected === undefined && items[0] !== undefined) {
-    selected = multiselect ? [items[0].id] : items[0].id
+  $: selectedItem = multiselect
+    ? (items ?? []).filter((p) => selected?.includes(p.id))
+    : (items ?? []).find((x) => x.id === selected)
+  $: if (autoSelect && selected === undefined && items?.[0] !== undefined) {
+    selected = multiselect ? [items?.[0]?.id] : items?.[0]?.id
   }
 
   const dispatch = createEventDispatcher()
@@ -111,7 +113,7 @@
         <slot name="content" />
       {:else if Array.isArray(selectedItem)}
         {#if selectedItem.length > 0}
-          {#each selectedItem as seleceted, i}
+          {#each selectedItem as seleceted}
             <span class="step-row">{seleceted.label}</span>
           {/each}
         {:else}

plugins/client-resources/src/connection.ts

@@ -108,10 +108,12 @@ class Connection implements ClientConnection {
 
   private helloRecieved: boolean = false
 
-  onConnect?: (event: ClientConnectEvent, data: any) => Promise<void>
+  onConnect?: (event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise<void>
 
   rpcHandler = new RPCHandler()
 
+  lastHash?: string
+
   constructor (
     private readonly ctx: MeasureContext,
     private readonly url: string,
@@ -144,6 +146,11 @@ class Connection implements ClientConnection {
     this.scheduleOpen(this.ctx, false)
   }
 
+  async getLastHash (ctx: MeasureContext): Promise<string | undefined> {
+    await this.waitOpenConnection(ctx)
+    return this.lastHash
+  }
+
   private schedulePing (socketId: number): void {
     clearInterval(this.interval)
     this.pingResponse = Date.now()
@@ -272,7 +279,7 @@ class Connection implements ClientConnection {
     if (resp.id === -1) {
       this.delay = 0
       if (resp.result?.state === 'upgrading') {
-        void this.onConnect?.(ClientConnectEvent.Maintenance, resp.result.stats)
+        void this.onConnect?.(ClientConnectEvent.Maintenance, undefined, resp.result.stats)
         this.upgrading = true
         this.delay = 3
         return
@@ -286,6 +293,7 @@ class Connection implements ClientConnection {
         // We need to clear dial timer, since we recieve hello response.
         clearTimeout(this.dialTimer)
         this.dialTimer = null
+        this.lastHash = (resp as HelloResponse).lastHash
 
         const serverVersion = helloResp.serverVersion
         console.log('Connected to server:', serverVersion)
@@ -315,6 +323,7 @@ class Connection implements ClientConnection {
 
         void this.onConnect?.(
           (resp as HelloResponse).reconnect === true ? ClientConnectEvent.Reconnected : ClientConnectEvent.Connected,
+          (resp as HelloResponse).lastTx,
           this.sessionId
         )
         this.schedulePing(socketId)

plugins/client-resources/src/index.ts

@@ -118,10 +118,10 @@ export default async () => {
                   reject(new Error(`Connection timeout, and no connection established to ${endpoint}`))
                 }
               }, connectTimeout)
-              newOpt.onConnect = async (event, data) => {
+              newOpt.onConnect = async (event, lastTx, data) => {
                 // Any event is fine, it means server is alive.
                 clearTimeout(connectTO)
-                await opt?.onConnect?.(event, data)
+                await opt?.onConnect?.(event, lastTx, data)
                 resolve()
               }
             })

plugins/client/src/index.ts

@@ -62,7 +62,7 @@ export interface ClientFactoryOptions {
   onUpgrade?: () => void
   onUnauthorized?: () => void
   onArchived?: () => void
-  onConnect?: (event: ClientConnectEvent, data: any) => Promise<void>
+  onConnect?: (event: ClientConnectEvent, lastTx: string | undefined, data: any) => Promise<void>
   ctx?: MeasureContext
   onDialTimeout?: () => void | Promise<void>
 }

plugins/recruit-resources/src/components/VacancyPresenter.svelte

@@ -31,7 +31,7 @@
   export let type: ObjectPresenterType = 'link'
 
   const dispatch = createEventDispatcher()
-  $: accentColor = getPlatformAvatarColorForTextDef(value.name, $themeStore.dark)
+  $: accentColor = getPlatformAvatarColorForTextDef(value?.name ?? '', $themeStore.dark)
 
   $: dispatch('accent-color', accentColor)
   onMount(() => {

server/core/src/types.ts

@@ -162,6 +162,11 @@ export interface DBAdapterManager {
 
 export interface PipelineContext {
   workspace: WorkspaceIdWithUrl
+
+  lastTx?: string
+
+  lastHash?: string
+
   hierarchy: Hierarchy
   modelDb: ModelDb
   branding: Branding | null

server/middleware/src/model.ts

@@ -115,6 +115,7 @@ export class ModelMiddleware extends BaseMiddleware implements Middleware {
 
   private setLastHash (hash: string): void {
     this.lastHash = hash
+    this.context.lastHash = this.lastHash
     this.lastHashResponse = Promise.resolve({
       full: false,
       hash,

server/middleware/src/txPush.ts

@@ -16,12 +16,16 @@
 import core, {
   DOMAIN_TRANSIENT,
   DOMAIN_TX,
+  generateId,
   TxProcessor,
+  WorkspaceEvent,
   type Doc,
   type MeasureContext,
+  type SessionData,
   type Tx,
   type TxCUD,
-  type TxResult
+  type TxResult,
+  type TxWorkspaceEvent
 } from '@hcengineering/core'
 import { PlatformError, unknownError } from '@hcengineering/platform'
 import type { DBAdapterManager, Middleware, PipelineContext, TxMiddlewareResult } from '@hcengineering/server-core'
@@ -68,6 +72,22 @@ export class TxMiddleware extends BaseMiddleware implements Middleware {
           txes: Array.from(new Set(txToStore.map((it) => it._class)))
         }
       )
+      // We need to remember last Tx Id in context, so it will be used during reconnect to track a requirement for refresh.
+      this.context.lastTx = txToStore[txToStore.length - 1]._id
+      // We need to deliver information to all clients so far.
+      const evt: TxWorkspaceEvent = {
+        _class: core.class.TxWorkspaceEvent,
+        _id: generateId(),
+        event: WorkspaceEvent.LastTx,
+        modifiedBy: core.account.System,
+        modifiedOn: Date.now(),
+        objectSpace: core.space.DerivedTx,
+        space: core.space.DerivedTx,
+        params: {
+          lastTx: this.context.lastTx
+        }
+      }
+      ;(ctx.contextData as SessionData).broadcast.txes.push(evt)
     }
     if (txPromise !== undefined) {
       await txPromise

server/rpc/src/rpc.ts

@@ -46,6 +46,8 @@ export interface HelloResponse extends Response<any> {
   binary: boolean
   reconnect?: boolean
   serverVersion: string
+  lastTx?: string
+  lastHash?: string // Last model hash
 }
 
 function replacer (key: string, value: any): any {

server/server/src/sessionManager.ts

@@ -990,7 +990,7 @@ class TSessionManager implements SessionManager {
           return
         }
         if (request.id === -1 && request.method === 'hello') {
-          this.handleHello<S>(request, service, ctx, workspace, ws, requestCtx)
+          await this.handleHello<S>(request, service, ctx, workspace, ws, requestCtx)
           return
         }
         if (request.id === -2 && request.method === 'forceClose') {
@@ -1053,14 +1053,14 @@ class TSessionManager implements SessionManager {
       })
   }
 
-  private handleHello<S extends Session>(
+  private async handleHello<S extends Session>(
     request: Request<any>,
     service: S,
     ctx: MeasureContext<any>,
     workspace: string,
     ws: ConnectionSocket,
     requestCtx: MeasureContext<any>
-  ): void {
+  ): Promise<void> {
     const hello = request as HelloRequest
     service.binaryMode = hello.binary ?? false
     service.useCompression = hello.compression ?? false
@@ -1080,12 +1080,16 @@ class TSessionManager implements SessionManager {
     if (reconnect) {
       this.reconnectIds.delete(service.sessionId)
     }
+    const pipeline =
+      service.workspace.pipeline instanceof Promise ? await service.workspace.pipeline : service.workspace.pipeline
     const helloResponse: HelloResponse = {
       id: -1,
       result: 'hello',
       binary: service.binaryMode,
       reconnect,
-      serverVersion: this.serverVersion
+      serverVersion: this.serverVersion,
+      lastTx: pipeline.context.lastTx,
+      lastHash: pipeline.context.lastHash
     }
     ws.send(requestCtx, helloResponse, false, false)
   }