From 1e7b5315643adadca9bcfa929abb70dd83615844 Mon Sep 17 00:00:00 2001 From: Andrey Sobolev Date: Thu, 18 Sep 2025 23:23:28 +0700 Subject: [PATCH] UBERF-13725: Fix Github service oom (#9891) Signed-off-by: Andrey Sobolev --- .vscode/launch.json | 1 + plugins/client-resources/src/connection.ts | 55 ++++---- services/github/pod-github/run.sh | 8 +- services/github/pod-github/src/platform.ts | 151 +++++++++++---------- services/github/pod-github/src/worker.ts | 18 ++- 5 files changed, 125 insertions(+), 108 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index 3c237f20d0..c13806bcbb 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -617,6 +617,7 @@ }, "runtimeArgs": ["--nolazy", "-r", "ts-node/register"], "sourceMaps": true, + "nodeVersionHint": 22, "cwd": "${workspaceRoot}/services/github/pod-github", "protocol": "inspector", "outputCapture": "std" diff --git a/plugins/client-resources/src/connection.ts b/plugins/client-resources/src/connection.ts index 456c881ed6..fd4b2bcd08 100644 --- a/plugins/client-resources/src/connection.ts +++ b/plugins/client-resources/src/connection.ts @@ -72,7 +72,7 @@ class RequestPromise { reject!: (reason?: any) => void reconnect?: () => void - // Required to proeprly handle rate limits + // Required to properly handle rate limits sendData: () => void = () => {} constructor ( @@ -92,6 +92,11 @@ class RequestPromise { const globalRPCHandler: RPCHandler = new RPCHandler() +interface OnConnectHandler { + resolve: () => void + reject: (err: Error) => void +} + class Connection implements ClientConnection { private websocket: ClientSocket | null = null binaryMode = false @@ -99,7 +104,7 @@ class Connection implements ClientConnection { private readonly requests = new Map() private lastId = 0 private interval: number | undefined - private dialTimer: any | undefined + private dialTimer: number | undefined private sockets = 0 private openAction: any @@ -166,12 +171,13 @@ class Connection implements ClientConnection { } private schedulePing (socketId: number): void { - clearInterval(this.interval) this.pingResponse = Date.now() const wsocket = this.websocket - const interval = setInterval(() => { + + clearInterval(this.interval) + this.interval = setInterval(() => { if (wsocket !== this.websocket) { - clearInterval(interval) + clearInterval(this.interval) return } if (!this.upgrading && this.pingResponse !== 0 && Date.now() - this.pingResponse > hangTimeout) { @@ -203,7 +209,6 @@ class Connection implements ClientConnection { clearInterval(this.interval) } }, pingTimeout) - this.interval = interval } async close (): Promise { @@ -211,6 +216,12 @@ class Connection implements ClientConnection { clearTimeout(this.openAction) clearTimeout(this.dialTimer) clearInterval(this.interval) + for (const handler of this.onConnectHandlers) { + handler.reject(new Error('Connection closed')) + } + for (const req of this.requests.values()) { + req.reject(new Error('Connection closed')) + } if (this.websocket !== null) { this.websocket.close(1000) this.websocket = null @@ -222,7 +233,7 @@ class Connection implements ClientConnection { } delay = 0 - onConnectHandlers: (() => void)[] = [] + onConnectHandlers: OnConnectHandler[] = [] private waitOpenConnection (ctx: MeasureContext): Promise | undefined { if (this.isConnected()) { @@ -233,9 +244,10 @@ class Connection implements ClientConnection { 'wait-connection', {}, (ctx) => - new Promise((resolve) => { - this.onConnectHandlers.push(() => { - resolve() + new Promise((resolve, reject) => { + this.onConnectHandlers.push({ + resolve, + reject }) // Websocket is null for first time this.scheduleOpen(ctx, false) @@ -294,7 +306,7 @@ class Connection implements ClientConnection { if (resp.error !== undefined) { if (resp.error?.code === UNAUTHORIZED.code || resp.terminate === true) { if ( - resp.error.code !== platform.status.WorkspaceArchived || + resp.error.code !== platform.status.WorkspaceArchived && resp.error.code !== platform.status.WorkspaceNotFound ) { Analytics.handleError(new PlatformError(resp.error)) @@ -350,7 +362,7 @@ class Connection implements ClientConnection { // We need to clear dial timer, since we recieve hello response. clearTimeout(this.dialTimer) - this.dialTimer = null + this.dialTimer = undefined this.lastHash = (resp as HelloResponse).lastHash const serverVersion = helloResp.serverVersion @@ -374,7 +386,7 @@ class Connection implements ClientConnection { // Notify all waiting connection listeners const handlers = this.onConnectHandlers.splice(0, this.onConnectHandlers.length) for (const h of handlers) { - h() + h.resolve() } for (const [, v] of this.requests.entries()) { @@ -540,12 +552,10 @@ class Connection implements ClientConnection { return } this.websocket = wsocket - const opened = false - - if (this.dialTimer != null) { + if (this.dialTimer === undefined) { this.dialTimer = setTimeout(() => { - this.dialTimer = null - if (!opened && !this.closed) { + this.dialTimer = undefined + if (!this.closed) { void this.opt?.onDialTimeout?.()?.catch((err) => { this.ctx.error('failed to handle dial timeout', { err }) }) @@ -652,16 +662,15 @@ class Connection implements ClientConnection { ctx.withSync('send-hello', {}, () => this.websocket?.send(this.rpcHandler.serialize(helloRequest, false))) } - wsocket.onerror = (event: any) => { + // FIX: remove undefined variable 'opened' + wsocket.onerror = () => { if (this.websocket !== wsocket) { return } if (this.delay < 3) { this.delay += 1 } - if (opened) { - console.error('client websocket error:', socketId, this.url, this.workspace, this.user) - } + console.error('client websocket error:', socketId, this.url, this.workspace, this.user) } } @@ -761,7 +770,7 @@ class Connection implements ClientConnection { getAccount (): Promise { if (this.account !== undefined) { - return clone(this.account) + return Promise.resolve(clone(this.account)) } return this.sendRequest({ method: 'getAccount', params: [] }) } diff --git a/services/github/pod-github/run.sh b/services/github/pod-github/run.sh index 620f7fd770..bf9c19eed8 100755 --- a/services/github/pod-github/run.sh +++ b/services/github/pod-github/run.sh @@ -4,10 +4,8 @@ export CLIENT_SECRET="$POD_GITHUB_CLIENT_SECRET" export PRIVATE_KEY="$POD_GITHUB_PRIVATE_KEY" export SERVER_SECRET=secret export ACCOUNTS_URL=http://localhost:3000 -export COLLABORATOR_URL=http://localhost:3078 -export MINIO_ACCESS_KEY=minioadminchmo -export MINIO_SECRET_KEY=minioadmin -export MINIO_ENDPOINT=localhost -export MONGO_URL=mongodb://localhost:27017 +export COLLABORATOR_URL=ws://huly.local:3078 +export STORAGE_CONFIG="datalake|http://huly.local:4030" +export OTEL_EXPORTER_OTLP_ENDPOINT=http://huly.local:4318/v1/traces rush bundle --to @hcengineering/pod-github node $@ bundle/bundle.js $@ \ No newline at end of file diff --git a/services/github/pod-github/src/platform.ts b/services/github/pod-github/src/platform.ts index 36892bb5d3..7c7d6d4a75 100644 --- a/services/github/pod-github/src/platform.ts +++ b/services/github/pod-github/src/platform.ts @@ -214,6 +214,8 @@ export class PlatformWorker { let oldErrors = '' let sameErrors = 1 + let lastTimeout: any + while (!this.canceled) { let errors: string[] = [] try { @@ -227,6 +229,7 @@ export class PlatformWorker { this.triggerCheckWorkspaces = () => { this.ctx.info('Workspaces check triggered') this.triggerCheckWorkspaces = () => {} + clearTimeout(lastTimeout) resolve() } if (errors.length > 0) { @@ -320,9 +323,10 @@ export class PlatformWorker { try { ;({ client } = await createPlatformClient(ctx, oldWorkspace, 30000)) await this.removeInstallationFromWorkspace(oldWorker, installationId) - await client.close() } catch (err: any) { ctx.error('failed to remove old installation from workspace', { workspace: oldWorkspace, installationId }) + } finally { + await client?.close() } } } @@ -944,26 +948,18 @@ export class PlatformWorker { checkedWorkspaces = new Set() - async checkWorkspaceIsActive ( - workspace: WorkspaceUuid, - workspaceInfo?: WorkspaceInfoWithStatus, - needRecheck = false - ): Promise<{ workspaceInfo: WorkspaceInfoWithStatus | undefined, needRecheck: boolean }> { - if (workspaceInfo === undefined && needRecheck) { - const token = generateToken(systemAccountUuid, workspace, { service: 'github', mode: 'github' }) - workspaceInfo = await getAccountClient(config.AccountsURL, token).getWorkspaceInfo() - } + checkWorkspaceIsActive (workspace: WorkspaceUuid, workspaceInfo: WorkspaceInfoWithStatus): boolean { if (workspaceInfo?.uuid === undefined) { this.ctx.error('No workspace exists for workspaceId', { workspace }) - return { workspaceInfo: undefined, needRecheck: false } + return false } if (workspaceInfo?.isDisabled === true || isDeletingMode(workspaceInfo?.mode)) { this.ctx.warn('Workspace is disabled', { workspace }) - return { workspaceInfo: undefined, needRecheck: false } + return false } if (!isActiveMode(workspaceInfo?.mode)) { this.ctx.warn('Workspace is in maitenance, skipping for now.', { workspace, mode: workspaceInfo?.mode }) - return { workspaceInfo: undefined, needRecheck: true } + return true } const lastVisit = (Date.now() - (workspaceInfo.lastVisit ?? 0)) / (3600 * 24 * 1000) // In days @@ -973,9 +969,38 @@ export class PlatformWorker { this.checkedWorkspaces.add(workspace) this.ctx.warn('Workspace is inactive for too long, skipping for now.', { workspace }) } - return { workspaceInfo: undefined, needRecheck: true } + return true } - return { workspaceInfo, needRecheck: true } + return false + } + + checkReconnect (workspace: WorkspaceUuid, event: ClientConnectEvent, worker: GithubWorker): void { + if (event === ClientConnectEvent.Refresh || event === ClientConnectEvent.Upgraded) { + void this.clients + .get(workspace) + ?.refreshClient(event === ClientConnectEvent.Upgraded) + ?.catch((err) => { + worker.ctx.error('Failed to refresh', { error: err }) + }) + } + + // We need to check if workspace is inactive + const token = generateToken(systemAccountUuid, workspace, { service: 'github', mode: 'github' }) + getAccountClient(config.AccountsURL, token) + .getWorkspaceInfo() + .then((wsInfo) => { + const res = this.checkWorkspaceIsActive(workspace, wsInfo) + if (!res) { + this.ctx.warn('Workspace is inactive, removing from clients list.', { workspace }) + this.clients.delete(workspace) + void worker?.close().catch((err) => { + this.ctx.error('Failed to close workspace', { workspace, error: err }) + }) + } + }) + .catch((err) => { + this.ctx.error('Failed to check workspace is active', { workspace, error: err }) + }) } private async checkWorkspaces (): Promise { @@ -1008,6 +1033,7 @@ export class PlatformWorker { this.ctx.info('connecting to workspace', { workspace: c, time: Date.now() - d.time, version: d.version }) } }, 5000) + try { const token = generateToken(systemAccountUuid, undefined, { service: 'github', mode: 'github' }) const infos = new Map( @@ -1022,82 +1048,57 @@ export class PlatformWorker { toDelete.delete(workspace) continue } + const returnedInfo = infos.get(workspace) + if (returnedInfo === undefined) { + rechecks.push(workspace) + continue + } + const needRecheck = this.checkWorkspaceIsActive(workspace, returnedInfo) + if (needRecheck) { + rechecks.push(workspace) + continue + } await rateLimiter.add(async () => { - const { workspaceInfo, needRecheck } = await this.checkWorkspaceIsActive(workspace, infos.get(workspace)) - if (workspaceInfo === undefined) { - if (needRecheck) { - rechecks.push(workspace) - } - return - } try { - const branding = Object.values(this.brandingMap).find((b) => b.key === workspaceInfo?.branding) ?? null - const workerCtx = this.ctx.newChild('worker', { workspace: workspaceInfo.uuid }, { span: false }) + const branding = Object.values(this.brandingMap).find((b) => b.key === returnedInfo?.branding) ?? null + const workerCtx = this.ctx.newChild('worker', { workspace: returnedInfo.uuid }, { span: false }) - connecting.set(workspaceInfo.uuid, { + connecting.set(returnedInfo.uuid, { time: Date.now(), version: versionToString({ - major: workspaceInfo.versionMajor, - minor: workspaceInfo.versionMinor, - patch: workspaceInfo.versionPatch + major: returnedInfo.versionMajor, + minor: returnedInfo.versionMinor, + patch: returnedInfo.versionPatch }) }) workerCtx.info('************************* Register worker ************************* ', { - workspaceId: workspaceInfo.uuid, - workspaceUrl: workspaceInfo.url, - versionMajor: workspaceInfo.versionMajor, - versionMinor: workspaceInfo.versionMinor, - versionPatch: workspaceInfo.versionPatch, - mode: workspaceInfo.mode, + workspaceId: returnedInfo.uuid, + workspaceUrl: returnedInfo.url, + versionMajor: returnedInfo.versionMajor, + versionMinor: returnedInfo.versionMinor, + versionPatch: returnedInfo.versionPatch, + mode: returnedInfo.mode, index: widx, total: workspaces.length }) - let initialized = false const worker = await GithubWorker.create( this, workerCtx, this.installations, { - dataId: workspaceInfo.dataId, - url: workspaceInfo.url, - uuid: workspaceInfo.uuid + dataId: returnedInfo.dataId, + url: returnedInfo.url, + uuid: returnedInfo.uuid }, branding, this.app, - this.storageAdapter, - (workspace, event) => { - if (event === ClientConnectEvent.Refresh || event === ClientConnectEvent.Upgraded) { - void this.clients - .get(workspace) - ?.refreshClient(event === ClientConnectEvent.Upgraded) - ?.catch((err) => { - workerCtx.error('Failed to refresh', { error: err }) - }) - } - if (initialized) { - // We need to check if workspace is inactive - void this.checkWorkspaceIsActive(workspace, undefined, true) - .then((res) => { - if (res === undefined) { - this.ctx.warn('Workspace is inactive, removing from clients list.', { workspace }) - this.clients.delete(workspace) - void worker?.close().catch((err) => { - this.ctx.error('Failed to close workspace', { workspace, error: err }) - }) - } - }) - .catch((err) => { - this.ctx.error('Failed to check workspace is active', { workspace, error: err }) - }) - } - } + this.storageAdapter ) if (worker !== undefined) { - initialized = true workerCtx.info('************************* Register worker Done ************************* ', { - workspaceId: workspaceInfo.uuid, - workspaceUrl: workspaceInfo.url, + workspaceId: returnedInfo.uuid, + workspaceUrl: returnedInfo.url, index: widx, total: workspaces.length }) @@ -1107,12 +1108,12 @@ export class PlatformWorker { workerCtx.info( '************************* Failed Register worker, timeout or integrations removed *************************', { - workspaceId: workspaceInfo.uuid, - workspaceUrl: workspaceInfo.url, - versionMajor: workspaceInfo.versionMajor, - versionMinor: workspaceInfo.versionMinor, - versionPatch: workspaceInfo.versionPatch, - lastVisit: (Date.now() - (workspaceInfo.lastVisit ?? 0)) / (24 * 60 * 60 * 1000), + workspaceId: returnedInfo.uuid, + workspaceUrl: returnedInfo.url, + versionMajor: returnedInfo.versionMajor, + versionMinor: returnedInfo.versionMinor, + versionPatch: returnedInfo.versionPatch, + lastVisit: (Date.now() - (returnedInfo.lastVisit ?? 0)) / (24 * 60 * 60 * 1000), index: widx, total: workspaces.length } @@ -1124,7 +1125,7 @@ export class PlatformWorker { this.ctx.info("Couldn't create WS worker", { workspace, error: e }) rechecks.push(workspace) } finally { - connecting.delete(workspaceInfo.uuid) + connecting.delete(returnedInfo.uuid) } }) } diff --git a/services/github/pod-github/src/worker.ts b/services/github/pod-github/src/worker.ts index d1ff76ebfb..ca15bb32b4 100644 --- a/services/github/pod-github/src/worker.ts +++ b/services/github/pod-github/src/worker.ts @@ -1766,13 +1766,13 @@ export class GithubWorker implements IntegrationManager { workspace: WorkspaceIds, branding: Branding | null, app: App, - storageAdapter: StorageAdapter, - reconnect: (workspaceId: WorkspaceUuid, event: ClientConnectEvent) => void + storageAdapter: StorageAdapter ): Promise { ctx.info('Connecting to', { workspace }) let client: Client | undefined let endpoint: string | undefined let maitenanceState = false + let worker: GithubWorker | undefined try { ;({ client, endpoint } = await createPlatformClient( ctx, @@ -1784,7 +1784,9 @@ export class GithubWorker implements IntegrationManager { maitenanceState = true throw new Error('Workspace in maintenance') } - reconnect(workspace.uuid, event) + if (worker !== undefined) { + platformWorker.checkReconnect(workspace.uuid, event, worker) + } } )) ctx.info('connected to github', { workspace: workspace.uuid, endpoint }) @@ -1796,7 +1798,7 @@ export class GithubWorker implements IntegrationManager { await GithubWorker.checkIntegrations(client, installations) - const worker = new GithubWorker( + worker = new GithubWorker( ctx, platformWorker.getRateLimiter(endpoint ?? ''), platformWorker, @@ -1808,10 +1810,16 @@ export class GithubWorker implements IntegrationManager { branding ) ctx.info('Init worker', { workspace: workspace.url, workspaceId: workspace.uuid }) - void worker.init() + void worker.init().catch((err) => { + ctx.error('failed to init worker', { error: err, workspace: workspace.uuid }) + void client?.close().catch((err) => { + ctx.error('failed to close client after init error', { error: err, workspace: workspace.uuid }) + }) + }) return worker } catch (err: any) { await client?.close() + void worker?.close() if (maitenanceState) { ctx.info('workspace in maintenance, schedule recheck', { workspace: workspace.uuid, endpoint }) return