Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

UBERF-7790: Fix connection timeout issue #6301

Merged
merged 1 commit into from
Aug 9, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 17 additions & 11 deletions plugins/client-resources/src/connection.ts
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,8 @@ class Connection implements ClientConnection {

private pingResponse: number = Date.now()

private helloRecieved: boolean = false

rpcHandler = new RPCHandler()

constructor (
Expand Down Expand Up @@ -173,7 +175,7 @@ class Connection implements ClientConnection {
}

isConnected (): boolean {
return this.websocket != null && this.websocket.readyState === ClientSocketReadyState.OPEN
return this.websocket != null && this.websocket.readyState === ClientSocketReadyState.OPEN && this.helloRecieved
}

delay = 0
Expand Down Expand Up @@ -240,6 +242,10 @@ class Connection implements ClientConnection {
return
}
if (resp.result === 'hello') {
// We need to clear dial timer, since we recieve hello response.
clearTimeout(this.dialTimer)
this.dialTimer = null
this.helloRecieved = true
if (this.upgrading) {
// We need to call upgrade since connection is upgraded
this.opt?.onUpgrade?.()
Expand Down Expand Up @@ -394,12 +400,15 @@ class Connection implements ClientConnection {
this.websocket = wsocket
const opened = false

this.dialTimer = setTimeout(() => {
if (!opened && !this.closed) {
void this.opt?.onDialTimeout?.()
this.scheduleOpen(true)
}
}, dialTimeout)
if (this.dialTimer != null) {
this.dialTimer = setTimeout(() => {
this.dialTimer = null
if (!opened && !this.closed) {
void this.opt?.onDialTimeout?.()
this.scheduleOpen(true)
}
}, dialTimeout)
}

wsocket.onmessage = (event: MessageEvent) => {
if (this.closed) {
Expand All @@ -419,10 +428,8 @@ class Connection implements ClientConnection {
}
}
wsocket.onclose = (ev) => {
clearTimeout(this.dialTimer)
if (this.websocket !== wsocket) {
wsocket.close()
clearTimeout(this.dialTimer)
return
}
// console.log('client websocket closed', socketId, ev?.reason)
Expand All @@ -435,7 +442,7 @@ class Connection implements ClientConnection {
}
const useBinary = getMetadata(client.metadata.UseBinaryProtocol) ?? true
const useCompression = getMetadata(client.metadata.UseProtocolCompression) ?? false
clearTimeout(this.dialTimer)
this.helloRecieved = false
const helloRequest: HelloRequest = {
method: 'hello',
params: [],
Expand All @@ -447,7 +454,6 @@ class Connection implements ClientConnection {
}

wsocket.onerror = (event: any) => {
clearTimeout(this.dialTimer)
if (this.websocket !== wsocket) {
return
}
Expand Down
14 changes: 0 additions & 14 deletions server/ws/src/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -287,33 +287,19 @@ class TSessionManager implements SessionManager {
accountsUrl !== '' ? await this.getWorkspaceInfo(ctx, accountsUrl, rawToken) : this.wsFromToken(token)
} catch (err: any) {
this.updateConnectErrorInfo(token)
// No connection to account service, retry from client.
await new Promise<void>((resolve) => {
setTimeout(resolve, 1000)
})
return { error: err }
}

if (workspaceInfo === undefined) {
this.updateConnectErrorInfo(token)
// No connection to account service, retry from client.
await new Promise<void>((resolve) => {
setTimeout(resolve, 1000)
})
return { upgrade: true }
}

if (workspaceInfo?.creating === true && token.email !== systemAccountEmail) {
await new Promise<void>((resolve) => {
setTimeout(resolve, 1000)
})
// No access to workspace for token.
return { error: new Error(`Workspace during creation phase ${token.email} ${token.workspace.name}`) }
}
if (workspaceInfo === undefined && token.extra?.admin !== 'true') {
await new Promise<void>((resolve) => {
setTimeout(resolve, 5000)
})
this.updateConnectErrorInfo(token)
// No access to workspace for token.
return { error: new Error(`No access to workspace for token ${token.email} ${token.workspace.name}`) }
Expand Down
15 changes: 12 additions & 3 deletions server/ws/src/server_http.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

import { Analytics } from '@hcengineering/analytics'
import { generateId, toWorkspaceString, type MeasureContext } from '@hcengineering/core'
import { UNAUTHORIZED } from '@hcengineering/platform'
import { UNAUTHORIZED, unknownStatus } from '@hcengineering/platform'
import { RPCHandler, type Response } from '@hcengineering/rpc'
import { decodeToken, type Token } from '@hcengineering/server-token'
import cors from 'cors'
Expand Down Expand Up @@ -277,13 +277,22 @@ export function startHttpServer (
if (webSocketData.session instanceof Promise) {
void webSocketData.session.then((s) => {
if ('error' in s) {
cs.close()
void cs
.send(ctx, { id: -1, error: unknownStatus(s.error.message ?? 'Unknown error') }, false, false)
.then(() => {
// No connection to account service, retry from client.
setTimeout(() => {
cs.close()
}, 1000)
})
}
if ('upgrade' in s) {
void cs
.send(ctx, { id: -1, result: { state: 'upgrading', stats: (s as any).upgradeInfo } }, false, false)
.then(() => {
cs.close()
setTimeout(() => {
cs.close()
}, 5000)
})
}
})
Expand Down
14 changes: 12 additions & 2 deletions server/ws/src/server_u.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import uWebSockets, { DISABLED, SHARED_COMPRESSOR, type HttpResponse, type WebSo
import { Readable } from 'stream'
import { getFile, getFileRange, type BlobResponse } from './blobs'
import { doSessionOp, processRequest, type WebsocketData } from './utils'
import { unknownStatus } from '@hcengineering/platform'

const rpcHandler = new RPCHandler()

Expand Down Expand Up @@ -139,13 +140,22 @@ export function startUWebsocketServer (
if (data.session instanceof Promise) {
void data.session.then((s) => {
if ('error' in s) {
ctx.error('error', { error: s.error?.message, stack: s.error?.stack })
void cs
.send(ctx, { id: -1, error: unknownStatus(s.error.message ?? 'Unknown error') }, false, false)
.then(() => {
// No connection to account service, retry from client.
setTimeout(() => {
cs.close()
}, 1000)
})
}
if ('upgrade' in s) {
void cs
.send(ctx, { id: -1, result: { state: 'upgrading', stats: (s as any).upgradeInfo } }, false, false)
.then(() => {
cs.close()
setTimeout(() => {
cs.close()
}, 5000)
})
}
})
Expand Down
3 changes: 3 additions & 0 deletions services/github/pod-github/src/platform.ts
Original file line number Diff line number Diff line change
Expand Up @@ -739,6 +739,9 @@ export class PlatformWorker {
}
}
}
this.ctx.info('************************* Check workspaces done ************************* ', {
workspaces: this.clients.size
})
return errors > 0
}

Expand Down
17 changes: 9 additions & 8 deletions services/github/pod-github/src/worker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,6 @@ import core, {
type Blob,
type MigrationState
} from '@hcengineering/core'
import { LiveQuery } from '@hcengineering/query'
import { StorageAdapter } from '@hcengineering/server-core'
import { getPublicLinkUrl } from '@hcengineering/server-guest-resources'
import task, { ProjectType, TaskType } from '@hcengineering/task'
import { MarkupNode, jsonToMarkup, isMarkdownsEquals } from '@hcengineering/text'
import tracker from '@hcengineering/tracker'
import { User } from '@octokit/webhooks-types'
import github, {
DocSyncInfo,
GithubAuthentication,
Expand All @@ -53,6 +46,13 @@ import github, {
GithubUserInfo,
githubId
} from '@hcengineering/github'
import { LiveQuery } from '@hcengineering/query'
import { StorageAdapter } from '@hcengineering/server-core'
import { getPublicLinkUrl } from '@hcengineering/server-guest-resources'
import task, { ProjectType, TaskType } from '@hcengineering/task'
import { MarkupNode, isMarkdownsEquals, jsonToMarkup } from '@hcengineering/text'
import tracker from '@hcengineering/tracker'
import { User } from '@octokit/webhooks-types'
import { App, Octokit } from 'octokit'
import { createPlatformClient } from './client'
import { createCollaboratorClient } from './collaborator'
Expand Down Expand Up @@ -1487,7 +1487,7 @@ export class GithubWorker implements IntegrationManager {
ctx.info('Connecting to', { workspace: workspace.workspaceUrl, workspaceId: workspace.workspaceName })
let client: Client | undefined
try {
client = await createPlatformClient(workspace.name, workspace.productId, 10000, (event: ClientConnectEvent) => {
client = await createPlatformClient(workspace.name, workspace.productId, 30000, (event: ClientConnectEvent) => {
reconnect(workspace.name, event)
})

Expand All @@ -1506,6 +1506,7 @@ export class GithubWorker implements IntegrationManager {
return worker
}
} catch (err: any) {
ctx.error('timeout during to connect', { workspace, error: err })
await client?.close()
}
}
Expand Down