qfix: connect timeout + service ws info cache (#9622)

1. Fix connect timeout in case of maitenance
2. Introduce workspace info cache for service accounts
3. Fix null in github integration

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

Author: haiodo

plugins/client-resources/src/index.ts

@@ -36,7 +36,8 @@ import core, {
   type PluginConfiguration,
   type Ref,
   type TxCUD,
-  platformNow
+  platformNow,
+  ClientConnectEvent
 } from '@hcengineering/core'
 import platform, { Severity, Status, getMetadata, getPlugins, setPlatformStatus } from '@hcengineering/platform'
 import { connect } from './connection'
@@ -140,10 +141,20 @@ export default async () => {
                 }
               }, connectTimeout)
               newOpt.onConnect = async (event, lastTx, data) => {
-                // Any event is fine, it means server is alive.
-                clearTimeout(connectTO)
-                await opt?.onConnect?.(event, lastTx, data)
-                resolve()
+                try {
+                  await opt?.onConnect?.(event, lastTx, data)
+                } catch (error) {
+                  void clientConnection?.close()
+                  void opt?.onDialTimeout?.()
+                  reject(error)
+                  return
+                }
+
+                if (event !== ClientConnectEvent.Maintenance) {
+                  // Any event is fine, it means server is alive.
+                  clearTimeout(connectTO)
+                  resolve()
+                }
               }
             })
           }

server/server/src/sessionManager.ts

@@ -63,6 +63,7 @@ import {
   type AddSessionResponse,
   type ClientSessionCtx,
   type ConnectionSocket,
+  type ConsumerHandle,
   type GetWorkspaceResponse,
   LOGGING_ENABLED,
   pingConst,
@@ -72,6 +73,7 @@ import {
   type PlatformQueueProducer,
   QueueTopic,
   type QueueUserMessage,
+  QueueWorkspaceEvent,
   type QueueWorkspaceMessage,
   type Session,
   type SessionManager,
@@ -116,6 +118,7 @@ export class TSessionManager implements SessionManager {
 
   workspaceProducer: PlatformQueueProducer<QueueWorkspaceMessage>
   usersProducer: PlatformQueueProducer<QueueUserMessage>
+  workspaceConsumer: ConsumerHandle
 
   now: number = Date.now()
 
@@ -141,8 +144,28 @@ export class TSessionManager implements SessionManager {
         this.handleTick()
       }, 1000 / ticksPerSecond)
     }
-    this.workspaceProducer = this.queue.getProducer(ctx.newChild('queue', {}, { span: false }), QueueTopic.Workspace)
-    this.usersProducer = this.queue.getProducer(ctx.newChild('queue', {}, { span: false }), QueueTopic.Users)
+    this.workspaceProducer = this.queue.getProducer(ctx.newChild('ws-queue', {}, { span: false }), QueueTopic.Workspace)
+    this.usersProducer = this.queue.getProducer(ctx.newChild('user-queue', {}, { span: false }), QueueTopic.Users)
+
+    this.workspaceConsumer = this.queue.createConsumer<QueueWorkspaceMessage>(
+      ctx.newChild('ws-queue-consume', {}, { span: false }),
+      QueueTopic.Workspace,
+      generateId(),
+      async (messages) => {
+        for (const msg of messages) {
+          for (const m of msg.value) {
+            if (
+              m.type === QueueWorkspaceEvent.Upgraded ||
+              m.type === QueueWorkspaceEvent.Restored ||
+              m.type === QueueWorkspaceEvent.Deleted
+            ) {
+              // Handle workspace messages
+              this.workspaceInfoCache.delete(msg.workspace)
+            }
+          }
+        }
+      }
+    )
 
     this.ticksContext = ctx.newChild('ticks', {}, { span: false })
   }
@@ -523,6 +546,8 @@ export class TSessionManager implements SessionManager {
 
   maintenanceWorkspaces = new Set<WorkspaceUuid>()
 
+  workspaceInfoCache = new Map<WorkspaceUuid, WorkspaceInfoWithStatus>()
+
   async addSession (
     ctx: MeasureContext,
     ws: ConnectionSocket,
@@ -552,10 +577,13 @@ export class TSessionManager implements SessionManager {
       if (wsInfo === undefined) {
         // In case of guest or system account
         // We need to get workspace info for system account.
-        const workspaceInfo = await this.getWorkspaceInfo(ctx, rawToken, false)
+        const workspaceInfo =
+          this.workspaceInfoCache.get(token.workspace) ?? (await this.getWorkspaceInfo(ctx, rawToken, false))
         if (workspaceInfo === undefined) {
           return { error: new Error('Workspace not found or not available'), terminate: true }
         }
+        this.workspaceInfoCache.set(token.workspace, workspaceInfo)
+
         wsInfo = {
           url: workspaceInfo.url,
           mode: workspaceInfo.mode,
@@ -569,6 +597,8 @@ export class TSessionManager implements SessionManager {
           endpoint: { externalUrl: '', internalUrl: '', region: workspaceInfo.region ?? '' },
           progress: workspaceInfo.processingProgress
         }
+      } else {
+        this.workspaceInfoCache.delete(token.workspace)
       }
       const { workspace, resp } = await this.getWorkspace(ctx.parent ?? ctx, token.workspace, wsInfo, token, ws)
       if (resp !== undefined) {
@@ -961,6 +991,7 @@ export class TSessionManager implements SessionManager {
   async forceClose (wsId: WorkspaceUuid, ignoreSocket?: ConnectionSocket): Promise<void> {
     const ws = this.workspaces.get(wsId)
     this.maintenanceWorkspaces.delete(wsId)
+    this.workspaceInfoCache.delete(wsId)
     if (ws !== undefined) {
       this.ctx.warn('force-close', { name: ws.wsId.url })
       ws.maintenance = true // We need to similare upgrade to refresh all clients.

services/github/pod-github/src/platform.ts

@@ -134,6 +134,16 @@ export class PlatformWorker {
           workspace: i.workspaceUuid,
           installationId: Array.isArray(installationId) ? installationId : [installationId]
         })
+      } else {
+        ctx.warn('Integration without installationId', {
+          accountId: i.socialId,
+          workspace: i.workspaceUuid
+        })
+        await accountsClient.deleteIntegration({
+          kind: 'github',
+          workspaceUuid: i.workspaceUuid,
+          socialId: i.socialId
+        })
       }
     }
 
@@ -1076,9 +1086,9 @@ export class PlatformWorker {
                 index: widx,
                 total: workspaces.length
               })
-              // No if no integration, we will try connect one more time in a time period
               this.clients.set(workspace, worker)
             } else {
+              // No if no integration, we will try connect one more time in a time period
               workerCtx.info(
                 '************************* Failed Register worker, timeout or integrations removed *************************',
                 {

services/github/pod-github/src/worker.ts

@@ -1753,8 +1753,14 @@ export class GithubWorker implements IntegrationManager {
     ctx.info('Connecting to', { workspace })
     let client: Client | undefined
     let endpoint: string | undefined
+    let maitenanceState = false
     try {
       ;({ client, endpoint } = await createPlatformClient(workspace.uuid, 30000, async (event: ClientConnectEvent) => {
+        if (event === ClientConnectEvent.Maintenance) {
+          await client?.close()
+          maitenanceState = true
+          throw new Error('Workspace in maintenance')
+        }
         reconnect(workspace.uuid, event)
       }))
       ctx.info('connected to github', { workspace: workspace.uuid, endpoint })
@@ -1781,8 +1787,12 @@ export class GithubWorker implements IntegrationManager {
       void worker.init()
       return worker
     } catch (err: any) {
-      ctx.error('timeout during to connect', { workspace, error: err })
       await client?.close()
+      if (maitenanceState) {
+        ctx.info('workspace in maintenance, schedule recheck', { workspace: workspace.uuid, endpoint })
+        return
+      }
+      ctx.error('timeout during to connect', { workspace, error: err })
       return undefined
     }
   }