diff --git a/plugins/client-resources/src/index.ts b/plugins/client-resources/src/index.ts index 5e7cea3ecb..905ceefb6d 100644 --- a/plugins/client-resources/src/index.ts +++ b/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() + } } }) } diff --git a/server/server/src/sessionManager.ts b/server/server/src/sessionManager.ts index 533288dbca..5ef5a66798 100644 --- a/server/server/src/sessionManager.ts +++ b/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 usersProducer: PlatformQueueProducer + 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( + 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() + workspaceInfoCache = new Map() + 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 { 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. diff --git a/services/github/pod-github/src/platform.ts b/services/github/pod-github/src/platform.ts index 3695189444..047174a02f 100644 --- a/services/github/pod-github/src/platform.ts +++ b/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 *************************', { diff --git a/services/github/pod-github/src/worker.ts b/services/github/pod-github/src/worker.ts index 2d2a9d55b8..fe2f41d280 100644 --- a/services/github/pod-github/src/worker.ts +++ b/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 } }