WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-09-12 20:57:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

Resolve connectivity issues

Browse Source
This commit is contained in:
Andrey Sobolev
2025-09-10 15:23:23 +07:00
parent ed68628955
commit ff3a9775bc
15 changed files with 153 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/config/rush/pnpm-lock.yaml
Generated
+4 -4
View File
@@ -84,7 +84,7 @@ importers:
version: 8.3.2
devDependencies:
'@hcengineering/platform-rig':
specifier: 0.7.4
specifier: ^0.7.4
version: 0.7.4(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint-config-standard-with-typescript@40.0.0(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(prettier@3.6.2)
'@types/jest':
specifier: ^29.5.5
@@ -139,7 +139,7 @@ importers:
version: 8.3.2
devDependencies:
'@hcengineering/platform-rig':
specifier: 0.7.4
specifier: ^0.7.4
version: 0.7.4(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint-config-standard-with-typescript@40.0.0(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(prettier@3.6.2)
'@types/jest':
specifier: ^29.5.5
@@ -203,7 +203,7 @@ importers:
version: 8.3.2
devDependencies:
'@hcengineering/platform-rig':
specifier: 0.7.4
specifier: ^0.7.4
version: 0.7.4(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint-config-standard-with-typescript@40.0.0(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(prettier@3.6.2)
'@types/jest':
specifier: ^29.5.5
@@ -267,7 +267,7 @@ importers:
version: 8.3.2
devDependencies:
'@hcengineering/platform-rig':
specifier: 0.7.4
specifier: ^0.7.4
version: 0.7.4(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint-config-standard-with-typescript@40.0.0(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.2))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(typescript@5.9.2))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(prettier@3.6.2)
'@types/dockerode':
specifier: ^3.3.19
packages/backrpc/src/__test__/backrpc.spec.ts
+3 -3
View File
@@ -272,7 +272,7 @@ describe('backrpc', () => {
expect(response).toBe('event-sent')
// Wait a bit for the event to be processed
await new Promise(resolve => setTimeout(resolve, 100))
await new Promise((resolve) => setTimeout(resolve, 100))
expect(eventsReceived).toHaveLength(1)
expect(eventsReceived[0]).toEqual({ type: 'test-event', data: 'test-data' })
@@ -338,7 +338,7 @@ describe('backrpc', () => {
)
// Wait for both clients to connect
await new Promise(resolve => setTimeout(resolve, 100))
await new Promise((resolve) => setTimeout(resolve, 100))
const client1Id = await client1.request('get-client-id', '')
const client2Id = await client2.request('get-client-id', '')
@@ -394,7 +394,7 @@ describe('backrpc', () => {
{
requestHandler: async (client, method, params, send) => {
// Simulate some async work
await new Promise(resolve => setTimeout(resolve, 50))
await new Promise((resolve) => setTimeout(resolve, 50))
await send(`processed-${params}`)
},
helloHandler: async (clientId) => {
packages/backrpc/src/client.ts
+38 -24
View File
@@ -57,7 +57,6 @@ export class BackRPCClient<ClientT extends string = ClientId> {
options?: zmq.SocketOptions<zmq.Dealer>
) {
this.dealer = new zmq.Dealer({ ...options, context })
this.dealer.connect(`tcp://${host}:${port}`)
this.setServerId = () => {}
this.serverId = new Promise<string>((resolve) => {
@@ -68,6 +67,8 @@ export class BackRPCClient<ClientT extends string = ClientId> {
})
this.observer = new zmq.Observer(this.dealer)
this.dealer.connect(`tcp://${host}:${port}`)
this.observer.on('connect', (data) => {
void this.sendHello().catch((err) => {
console.error('Failed to send hello', err)
@@ -78,18 +79,33 @@ export class BackRPCClient<ClientT extends string = ClientId> {
})
this.stopTick = this.tickMgr.register(() => {
void this.checkAlive().catch(err => {
void this.checkAlive().catch((err) => {
console.error(err)
})
}, timeouts.pingInterval)
}
private sendPromise: Promise<void> | undefined
async doSend (msg: any[]): Promise<void> {
while (this.sendPromise !== undefined) {
await this.sendPromise
}
this.sendPromise = this.dealer.send(msg)
try {
await this.sendPromise
} catch (err: any) {
console.error('Failed to send message', err)
}
this.sendPromise = undefined
}
async checkAlive (): Promise<void> {
await this.dealer.send([backrpcOperations.ping, this.clientId as string, '', ''])
await this.doSend([backrpcOperations.ping, this.clientId as string, '', ''])
}
private async sendHello (): Promise<void> {
await this.dealer.send([backrpcOperations.hello, this.clientId as string, '', ''])
await this.doSend([backrpcOperations.hello, this.clientId as string, '', ''])
}
private async start (): Promise<void> {
@@ -120,21 +136,19 @@ export class BackRPCClient<ClientT extends string = ClientId> {
const [method, params] = JSON.parse(payload.toString())
void this.client
.requestHandler(method, params, async (response: any) => {
await this.dealer.send([backrpcOperations.response, reqId, JSON.stringify(response)])
await this.doSend([backrpcOperations.response, reqId, JSON.stringify(response)])
})
.catch((error) => {
void this.dealer
.send([
backrpcOperations.responseError,
reqId,
JSON.stringify({
message: error.message ?? '',
stack: error.stack ?? ''
})
])
.catch((err2) => {
console.error('Failed to send error', err2, err2)
void this.doSend([
backrpcOperations.responseError,
reqId,
JSON.stringify({
message: error.message ?? '',
stack: error.stack ?? ''
})
]).catch((err2) => {
console.error('Failed to send error', err2, err2)
})
})
}
break
@@ -169,11 +183,11 @@ export class BackRPCClient<ClientT extends string = ClientId> {
if (req !== undefined) {
const count = JSON.parse(payload.toString())
void this.tickMgr.waitTick(count).then(() => {
void this.dealer
.send([backrpcOperations.request, reqId, JSON.stringify([req.method, req.params])])
.catch((err) => {
void this.doSend([backrpcOperations.request, reqId, JSON.stringify([req.method, req.params])]).catch(
(err) => {
console.error('Failed to resend request', err)
})
}
)
})
}
}
@@ -187,14 +201,14 @@ export class BackRPCClient<ClientT extends string = ClientId> {
private async resendRequests (): Promise<void> {
for (const [reqId, req] of Array.from(this.requests.entries())) {
try {
await this.dealer.send([backrpcOperations.request, reqId, JSON.stringify([req.method, req.params])])
await this.doSend([backrpcOperations.request, reqId, JSON.stringify([req.method, req.params])])
} catch (err: any) {
console.error('Failed to resend request', err)
}
}
}
async waitConnecting (): Promise<void> {
async waitConnection (): Promise<void> {
if (this.serverId instanceof Promise) {
await this.serverId
}
@@ -208,7 +222,7 @@ export class BackRPCClient<ClientT extends string = ClientId> {
const reqId = this.clientId + '-' + this.requestCounter++
this.requests.set(reqId, { resolve, reject, method, params })
void this.dealer.send([backrpcOperations.request, reqId, JSON.stringify([method, params])]).catch((err) => {
void this.doSend([backrpcOperations.request, reqId, JSON.stringify([method, params])]).catch((err) => {
this.requests.delete(reqId) // Cleanup on failure
reject(err)
})
@@ -216,7 +230,7 @@ export class BackRPCClient<ClientT extends string = ClientId> {
}
async send (body: any): Promise<any> {
await this.dealer.send([backrpcOperations.event, body])
await this.doSend([backrpcOperations.event, body])
}
close (): void {
packages/backrpc/src/server.ts
+60 -18
View File
@@ -70,7 +70,7 @@ export class BackRPCServer<ClientT extends string = ClientId> {
this.router = new zmq.Router({ context })
this.stopTick = this.tickMgr.register(() => {
void this.checkAlive().catch(err => {
void this.checkAlive().catch((err) => {
console.error(err)
})
}, timeouts.pingInterval)
@@ -79,16 +79,23 @@ export class BackRPCServer<ClientT extends string = ClientId> {
}
async checkAlive (): Promise<void> {
console.log('check alive:', this.revClientMapping.size, JSON.stringify(this.stats))
this.stats.hellos = 0
this.stats.pings = 0
this.stats.requests = 0
this.stats.responses = 0
const now = this.tickMgr.now()
// Handle outdated clients
for (const [clientId, clientRecord] of this.revClientMapping.entries()) {
for (const [clientId, clientRecord] of [...this.revClientMapping.entries()]) {
const timeSinceLastSeen = now - clientRecord.lastSeen
if (timeSinceLastSeen > timeouts.aliveTimeout * 1000) {
console.warn(
`Client ${clientId} has been inactive for ${Math.round(timeSinceLastSeen / 1000)}s, marking as dead`
)
await this.handlers.handleTimeout?.(clientRecord.id)
void this.handlers.handleTimeout?.(clientRecord.id).catch((err) => {
console.error('Error in handleTimeout', err)
})
this.revClientMapping.delete(clientId)
this.clientMapping.delete(clientRecord.id)
}
@@ -110,6 +117,33 @@ export class BackRPCServer<ClientT extends string = ClientId> {
return port
}
private sendPromise: Promise<void> | undefined
async doSend (msg: any[]): Promise<void> {
while (this.sendPromise !== undefined) {
await this.sendPromise
}
this.sendPromise = this.router.send(msg)
try {
await this.sendPromise
} catch (err: any) {
console.error('Failed to send message', err)
}
this.sendPromise = undefined
}
stats: {
pings: number
requests: number
responses: number
hellos: number
} = {
pings: 0,
requests: 0,
responses: 0,
hellos: 0
}
private async start (): Promise<void> {
this.bound = this.router.bind(`tcp://${this.host}:${this.port}`)
await this.bound
@@ -130,10 +164,9 @@ export class BackRPCServer<ClientT extends string = ClientId> {
}
switch (operation) {
case backrpcOperations.hello: {
this.stats.hellos++
// Remember clientId to be able to do back requests.
if (!this.clientMapping.has(reqId.toString() as ClientT)) {
await this.handlers.helloHandler?.(reqId.toString() as ClientT)
}
const needHello = !this.clientMapping.has(reqId.toString() as ClientT)
this.clientMapping.set(reqId.toString() as ClientT, clientId)
const clientInfo: RPCClientInfo<ClientT> =
@@ -149,19 +182,26 @@ export class BackRPCServer<ClientT extends string = ClientId> {
clientInfo.helloCounter++
this.revClientMapping.set(clientIdText, clientInfo)
void this.router.send([clientId, backrpcOperations.hello, this.uuid, ''])
void this.doSend([clientId, backrpcOperations.hello, this.uuid, ''])
if (needHello) {
void this.handlers.helloHandler?.(reqId.toString() as ClientT).catch((err) => {
console.error('Error in helloHandler', err)
})
}
break
}
case backrpcOperations.ping: {
void this.router.send([clientId, backrpcOperations.pong, this.uuid, ''])
console.log('ping:' + clientIdText)
this.stats.pings++
void this.doSend([clientId, backrpcOperations.pong, this.uuid, ''])
// console.log('ping:' + clientIdText)
break
}
case backrpcOperations.request:
{
this.stats.requests++
if (client === undefined) {
// No Client, requests are not possible
void this.router.send([clientId, backrpcOperations.retry, reqId, JSON.stringify(1)])
void this.doSend([clientId, backrpcOperations.retry, reqId, JSON.stringify(1)])
continue
}
if (client.requests.has(reqId)) {
@@ -170,7 +210,7 @@ export class BackRPCServer<ClientT extends string = ClientId> {
}
if (client.requests.size > this.requestsLimit) {
// No Client, requests are not possible
void this.router.send([clientId, backrpcOperations.retry, reqId, JSON.stringify(client.requests.size)])
void this.doSend([clientId, backrpcOperations.retry, reqId, JSON.stringify(client.requests.size)])
continue
}
@@ -179,7 +219,7 @@ export class BackRPCServer<ClientT extends string = ClientId> {
const [method, params] = JSON.parse(payload.toString())
const sendError = async (err: Error): Promise<void> => {
await this.router.send([
void this.doSend([
clientId,
backrpcOperations.responseError,
reqId,
@@ -192,7 +232,7 @@ export class BackRPCServer<ClientT extends string = ClientId> {
client.requests.add(reqId)
void this.handlers
.requestHandler(client.id, method, params, async (response: any) => {
await this.router.send([clientId, backrpcOperations.response, reqId, JSON.stringify(response)])
void this.doSend([clientId, backrpcOperations.response, reqId, JSON.stringify(response)])
})
.catch((err) => {
void sendError(err)
@@ -200,6 +240,7 @@ export class BackRPCServer<ClientT extends string = ClientId> {
}
break
case backrpcOperations.response: {
this.stats.responses++
const reqID = reqId.toString()
const req = this.backRequests.get(reqID)
try {
@@ -211,6 +252,7 @@ export class BackRPCServer<ClientT extends string = ClientId> {
break
}
case backrpcOperations.responseError: {
this.stats.responses++
const reqID = reqId.toString()
const req = this.backRequests.get(reqID)
try {
@@ -237,11 +279,11 @@ export class BackRPCServer<ClientT extends string = ClientId> {
const reqId = clientId + '-' + this.requestCounter++
this.backRequests.set(reqId, { resolve, reject })
void this.router
.send([clientIdentity, backrpcOperations.request, reqId, JSON.stringify([method, params])])
.catch((err) => {
void this.doSend([clientIdentity, backrpcOperations.request, reqId, JSON.stringify([method, params])]).catch(
(err) => {
reject(err)
})
}
)
})
}
@@ -250,7 +292,7 @@ export class BackRPCServer<ClientT extends string = ClientId> {
if (clientIdentity === undefined) {
throw new Error(`Client ${clientId as string} not found`)
}
await this.router.send([clientIdentity, backrpcOperations.event, '', JSON.stringify(body)])
await this.doSend([clientIdentity, backrpcOperations.event, '', JSON.stringify(body)])
}
async close (): Promise<void> {
packages/client/package.json
+1 -1
View File
@@ -18,7 +18,7 @@
"_phase:validate": "compile validate"
},
"devDependencies": {
"@hcengineering/platform-rig": "0.7.4",
"@hcengineering/platform-rig": "^0.7.4",
"@typescript-eslint/eslint-plugin": "^6.11.0",
"eslint-plugin-import": "^2.26.0",
"eslint-plugin-promise": "^6.1.1",
packages/client/src/client.ts
+22 -3
View File
@@ -88,7 +88,7 @@ export class NetworkClientImpl implements NetworkClient {
containerListeners: ContainerUpdateListener[] = []
references = new Map< ContainerUuid, ContainereRef>()
references = new Map<ContainerUuid, ContainereRef>()
registered: boolean = false
@@ -100,8 +100,27 @@ export class NetworkClientImpl implements NetworkClient {
this.client = new BackRPCClient<ClientUuid>(this.clientId, this, host, port, tickMgr)
}
waitingForConnection (): Promise<void> {
return this.client.waitConnecting()
async waitConnection (timeout?: number): Promise<void> {
if (timeout !== undefined) {
await new Promise<void>((resolve, reject) => {
const co = setTimeout(() => {
// Timeout reached, we reject the promise by throwing an error
reject(new Error('Connection timeout'))
}, timeout)
this.client
.waitConnection()
.then(() => {
resolve()
clearTimeout(co)
})
.catch((err) => {
reject(err)
})
})
return
}
await this.client.waitConnection()
}
async close (): Promise<void> {
packages/client/src/index.ts
+2 -4
View File
@@ -16,8 +16,6 @@ process.on('exit', () => {
shutdownNetworkTickMgr()
})
export async function createNetworkClient (host: string, port: number): Promise<NetworkClient> {
const client = new NetworkClientImpl(host, port, tickMgr)
await client.waitingForConnection()
return client
export function createNetworkClient (host: string, port: number): NetworkClient {
return new NetworkClientImpl(host, port, tickMgr)
}
packages/core/package.json
+1 -1
View File
@@ -18,7 +18,7 @@
"_phase:validate": "compile validate"
},
"devDependencies": {
"@hcengineering/platform-rig": "0.7.4",
"@hcengineering/platform-rig": "^0.7.4",
"@typescript-eslint/eslint-plugin": "^6.11.0",
"eslint-plugin-import": "^2.26.0",
"eslint-plugin-promise": "^6.1.1",
packages/core/src/api/client.ts
+6
View File
@@ -14,6 +14,8 @@ export type ContainerUpdateListener = (event: ContainerEvent) => Promise<void>
* Interface to Huly network.
*
* Identification is generated during instantions of client.
*
* Client is attempt connecting indefinitely, but helper connect method could be used to be sure we connected on time.
*/
export interface NetworkClient {
/*
@@ -41,6 +43,10 @@ export interface NetworkClient {
// Register on container update listener
onContainerUpdate: (listener: ContainerUpdateListener) => void
// We could wait for a connection for a time period.
// If timeout === 0, we wait indefinitely.
waitConnection: (timeout?: number) => Promise<void>
close: () => Promise<void>
}
packages/core/src/api/utils.ts
+2
View File
@@ -3,6 +3,8 @@ export type TickHandler = () => void | Promise<void>
export interface TickManager {
now: () => number
tps: number
// Interval in seconds
register: (handler: TickHandler, interval: number) => () => void
packages/server/package.json
+1 -1
View File
@@ -18,7 +18,7 @@
"_phase:validate": "compile validate"
},
"devDependencies": {
"@hcengineering/platform-rig": "0.7.4",
"@hcengineering/platform-rig": "^0.7.4",
"@typescript-eslint/eslint-plugin": "^6.11.0",
"eslint-plugin-import": "^2.26.0",
"eslint-plugin-promise": "^6.1.1",
pods/.eslintrc.js
-7
View File
@@ -1,7 +0,0 @@
module.exports = {
extends: ['./node_modules/@hcengineering/platform-rig/profiles/node/eslint.config.json'],
parserOptions: {
tsconfigRootDir: __dirname,
project: './tsconfig.json'
}
}
pods/network-pod/package.json
+1 -1
View File
@@ -31,7 +31,7 @@
"_phase:validate": "compile validate"
},
"devDependencies": {
"@hcengineering/platform-rig": "0.7.4",
"@hcengineering/platform-rig": "^0.7.4",
"@typescript-eslint/eslint-plugin": "^6.11.0",
"eslint-plugin-import": "^2.26.0",
"eslint-plugin-promise": "^6.1.1",
pods/network-pod/src/index.ts
+6 -6
View File
@@ -4,7 +4,7 @@ import { NetworkServer } from '@hcengineering/network-server'
/**
* Main entry point for the network pod
*/
async function main(): Promise<void> {
async function main (): Promise<void> {
console.log('Starting Huly Network Pod...')
// Create tick manager
@@ -21,7 +21,7 @@ async function main(): Promise<void> {
console.log(`Network Pod started on port ${port}`)
const shutdown = async () => {
const shutdown = async (): Promise<void> => {
console.log('Shutting down Network Pod...')
tickManager.stop()
await server.close()
@@ -30,12 +30,12 @@ async function main(): Promise<void> {
}
// Handle graceful shutdown
process.on('SIGINT', async () => {
shutdown()
process.on('SIGINT', (): void => {
void shutdown()
})
process.on('SIGTERM', async () => {
shutdown()
process.on('SIGTERM', (): void => {
void shutdown()
})
}
tests/src/__tests__/client.test.ts
+6 -5
View File
@@ -24,16 +24,17 @@ describe('network client tests', () => {
const clients: NetworkClient[] = []
for (let i = 0; i < count; i++) {
const st = performance.now()
const client = await createNetworkClient('localhost', 37371)
console.log('connecting agents: ' + i)
const client = createNetworkClient('localhost', 37371)
await client.waitConnection()
console.log('listing agents: ' + i)
const agents = await client.agents()
expect(agents).toBeDefined()
// const agents = await client.agents()
// expect(agents).toBeDefined()
clients.push(client)
const ed = performance.now()
total += ed - st
}
console.log('Average connect time: ' + Math.round(10 * total / count) / 10 + ' ms')
console.log('Average connect time: ' + Math.round((10 * total) / count) / 10 + ' ms')
for (const client of clients) {
await client.close()
}