diff --git a/.vscode/launch.json b/.vscode/launch.json index 1dbdd7ed62..3bbdf9f43c 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -20,7 +20,8 @@ "request": "launch", "args": ["src/index.ts"], "env": { - "PORT": "37371" + "PORT": "37371", + "DEVELOPMENT": "true" }, "runtimeArgs": ["--nolazy", "-r", "ts-node/register"], "runtimeVersion": "22", @@ -34,7 +35,7 @@ "name": "Debug Tool", "type": "node", "request": "launch", - "args": ["src/index.ts", "list-agents"], + "args": ["src/index.ts", "bench-agent"], "env": { "NETWORK_HOST": "localhost:37371" }, diff --git a/packages/backrpc/src/client.ts b/packages/backrpc/src/client.ts index 81042ffec9..18a83767b5 100644 --- a/packages/backrpc/src/client.ts +++ b/packages/backrpc/src/client.ts @@ -68,12 +68,13 @@ export class BackRPCClient { this.observer = new zmq.Observer(this.dealer) - this.dealer.connect(`tcp://${host}:${port}`) this.observer.on('connect', (data) => { + console.log('Connected to server', data) void this.sendHello().catch((err) => { console.error('Failed to send hello', err) }) }) + this.dealer.connect(`tcp://${host}:${port}`) void this.start().catch((err) => { console.error('Failed to start BackRPCClient', err) }) @@ -101,7 +102,7 @@ export class BackRPCClient { } async checkAlive (): Promise { - await this.doSend([backrpcOperations.ping, this.clientId as string, '', '']) + await this.doSend([backrpcOperations.ping, '', '', '']) } private async sendHello (): Promise { @@ -199,6 +200,10 @@ export class BackRPCClient { } private async resendRequests (): Promise { + if (this.closed) { + console.error('Client is closed, cannot resend requests') + return + } for (const [reqId, req] of Array.from(this.requests.entries())) { try { await this.doSend([backrpcOperations.request, reqId, JSON.stringify([req.method, req.params])]) @@ -214,6 +219,8 @@ export class BackRPCClient { } } + rCount = 0 + async request(method: string, params: any): Promise { if (this.serverId instanceof Promise) { await this.serverId diff --git a/packages/backrpc/src/server.ts b/packages/backrpc/src/server.ts index 7def73442e..c589ac8562 100644 --- a/packages/backrpc/src/server.ts +++ b/packages/backrpc/src/server.ts @@ -66,9 +66,15 @@ export class BackRPCServer { private readonly handlers: BackRPCServerHandler, private readonly tickMgr: TickManager, readonly host: string = '*', - private readonly port: number = 0 + private readonly port: number = 0, + private readonly options: zmq.SocketOptions = {} ) { - this.router = new zmq.Router({ context }) + this.router = new zmq.Router({ + ...options, + context, + // linger: 0, + tcpKeepalive: 1 + }) this.stopTick = this.tickMgr.register(() => { void this.checkAlive().catch((err) => { diff --git a/packages/client/src/agent.ts b/packages/client/src/agent.ts index 711c93deb7..004136f274 100644 --- a/packages/client/src/agent.ts +++ b/packages/client/src/agent.ts @@ -2,7 +2,7 @@ import { agentDirectRef, type ClientUuid, type ContainerConnection, - type ContainerEvent, + type NetworkEvent, type ContainerUuid, type NetworkAgent, type TickManager @@ -42,7 +42,7 @@ export class NetworkAgentServer implements BackRPCServerHandler { await this.agent.onAgentUpdate?.() } - async onContainerUpdate (event: ContainerEvent): Promise { + async onContainerUpdate (event: NetworkEvent): Promise { // Handle container update } diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index 5914c988bf..613011d6a5 100644 --- a/packages/client/src/client.ts +++ b/packages/client/src/client.ts @@ -1,25 +1,28 @@ +import { BackRPCClient, type BackRPCResponseSend } from '@hcengineering/network-backrpc' import { + agentDirectRef, + EndpointKind, + parseEndpointRef, type AgentEndpointRef, - type AgentRecord, + type AgentRecordInfo, type AgentUuid, type ClientUuid, type ContainerConnection, type ContainerEndpointRef, - type ContainerEvent, + type NetworkEvent, type ContainerKind, type ContainerRecord, type ContainerReference, - type ContainerRequest, - type ContainerUpdateListener, + type NetworkUpdateListener, type ContainerUuid, + type GetOptions, type NetworkAgent, type NetworkClient, - type TickManager + type TickManager, + NetworkEventKind } from '@hcengineering/network-core' import { v4 as uuidv4 } from 'uuid' import { ContainerConnectionImpl, NetworkDirectConnectionImpl, RoutedNetworkAgentConnectionImpl } from './agent' -import { BackRPCClient, type BackRPCResponseSend } from '@hcengineering/network-backrpc' -import { agentDirectRef, EndpointKind, parseEndpointRef } from '@hcengineering/network-core' import { opNames } from './types' interface ClientAgentRecord { @@ -31,8 +34,7 @@ interface ClientAgentRecord { class ContainerReferenceImpl implements ContainerReference { constructor ( readonly uuid: ContainerUuid, - private readonly client: NetworkClientImpl, - private readonly _request: ContainerRequest + private readonly client: NetworkClientImpl ) {} get endpoint (): ContainerEndpointRef { @@ -57,16 +59,16 @@ class ContainerReferenceImpl implements ContainerReference { if (conn !== undefined) { return conn } - const endpoint = await this.client.getContainerRef(this.uuid, this._request) - conn = this.client.establishConnection(this.uuid, endpoint) + conn = this.client.establishConnection(this.uuid, this.endpoint) await conn.connect() return conn } } -interface ContainereRef { +interface ContainerRef { ref: ContainerReference - request: ContainerRequest + kind: ContainerKind + request: GetOptions endpoint: ContainerEndpointRef } @@ -86,9 +88,10 @@ export class NetworkClientImpl implements NetworkClient { containerConnections = new Map() agentConnections = new Map>() - containerListeners: ContainerUpdateListener[] = [] + cid: number = 0 + containerListeners = new Map() - references = new Map() + references = new Map() registered: boolean = false @@ -124,6 +127,9 @@ export class NetworkClientImpl implements NetworkClient { } async close (): Promise { + for (const refs of this.references.values()) { + await refs.ref.close() + } for (const directConn of this.containerConnections.values()) { await directConn.close() } @@ -166,12 +172,12 @@ export class NetworkClientImpl implements NetworkClient { } } - async onEvent (event: ContainerEvent): Promise { + async onEvent (event: NetworkEvent): Promise { // Handle container events // In case of container stopped, agent stopped or endpoint changed, we need to update direct connections to be re-established. await this.handleConnectionUpdates(event) - for (const listener of this.containerListeners) { + for (const listener of this.containerListeners.values()) { try { await listener(event) } catch (error) { @@ -180,32 +186,52 @@ export class NetworkClientImpl implements NetworkClient { } } - async handleRefUpdate ( - uuid: ContainerUuid, - endpoint: ContainerEndpointRef | Promise - ): Promise { + async handleRefUpdate (uuid: ContainerUuid, endpoint: ContainerEndpointRef): Promise { const ref = this.references.get(uuid) if (ref !== undefined) { - const refEndpoint = endpoint instanceof Promise ? await endpoint : endpoint const conn = this.containerConnections.get(ref.ref.uuid) - if (conn !== undefined && ref.endpoint !== refEndpoint) { - conn.setConnection(this.establishConnection(ref.ref.uuid, refEndpoint)) + if (conn !== undefined && ref.endpoint !== endpoint) { + conn.setConnection(this.establishConnection(ref.ref.uuid, endpoint)) } else { - ref.endpoint = refEndpoint + ref.endpoint = endpoint } } } - async onRegister (): Promise { - // TODO: Add retry in container requests on re-connect to new network. - for (const [uuid, ref] of this.references.entries()) { - await this.handleRefUpdate(uuid, await this.retryGetContainerRef(uuid, ref.request)) + async handleNewContainer (oldUuid: ContainerUuid, uuid: ContainerUuid, endpoint: ContainerEndpointRef): Promise { + const ref = this.references.get(oldUuid) + this.references.delete(oldUuid) + if (ref !== undefined) { + const conn = this.containerConnections.get(oldUuid) + this.containerConnections.delete(oldUuid) + if (conn !== undefined) { + this.containerConnections.set(uuid, conn) + if (ref.endpoint !== endpoint) { + conn.setConnection(this.establishConnection(uuid, endpoint)) + } + } + ref.ref.uuid = uuid + ref.endpoint = endpoint + + this.references.set(uuid, ref) } + } + + async onRegister (): Promise { this.registered = true - // We need to re-register all our managed agents + // We need to re-register all our managed agents, since we could provide containers we request to our selfs for (const agent of this._agents.values()) { await this.doRegister(agent.agent) } + + for (const [uuid, ref] of this.references.entries()) { + const [newUuid, newEndpoint] = await this.retryGetContainerRef(ref.kind, ref.request) + if (uuid !== newUuid) { + await this.handleNewContainer(uuid, newUuid, newEndpoint) + } else { + await this.handleRefUpdate(uuid, newEndpoint) + } + } } /** @@ -258,23 +284,27 @@ export class NetworkClientImpl implements NetworkClient { this._agents.get(agent.uuid)?.resolve() } - async agents (): Promise { + async agents (): Promise { // Return actual list of agents - return await this.client.request(opNames.getAgents, {}) + return await this.client.request(opNames.getAgents, {}) } async kinds (): Promise { return await this.client.request(opNames.getKinds, {}) } - async get (uuid: ContainerUuid, request: ContainerRequest): Promise { - const existing = this.references.get(uuid) - if (existing !== undefined) { - return existing.ref + async get (kind: ContainerKind, request: GetOptions): Promise { + // TODO: Wait for all pending requests to finish + + if (request.uuid !== undefined) { + const existing = this.references.get(request.uuid) + if (existing !== undefined) { + return existing.ref + } } - const endpoint = await this.getContainerRef(uuid, request) - const ref: ContainerReference = new ContainerReferenceImpl(uuid, this, request) - this.references.set(uuid, { ref, request, endpoint }) + const [uuid, endpoint] = await this.retryGetContainerRef(kind, request) + const ref: ContainerReference = new ContainerReferenceImpl(uuid, this) + this.references.set(uuid, { kind, ref, request, endpoint }) return ref } @@ -325,38 +355,59 @@ export class NetworkClientImpl implements NetworkClient { return conn } - async handleConnectionUpdates (event: ContainerEvent): Promise { + async handleConnectionUpdates (event: NetworkEvent): Promise { // Handle connection updates - for (const updated of event.updated) { - await this.handleRefUpdate(updated.uuid, updated.endpoint) - } - - for (const deleted of event.deleted) { - await this.handleRefUpdate(deleted.uuid, deleted.endpoint) - } - } - - async getContainerRef (uuid: ContainerUuid, request: ContainerRequest): Promise { - return await this.client.request(opNames.getContainer, { uuid, request }) - } - - async retryGetContainerRef (uuid: ContainerUuid, request: ContainerRequest): Promise { - let waitTimeout: number = 1 - while (true) { - try { - const ref = await this.getContainerRef(uuid, request) - if (waitTimeout > 1) { - console.log(`Successfully got container ref for ${uuid} after ${waitTimeout - 1} retries.`) - } - return ref - } catch (err) { - console.warn(`Error getting container ref for ${uuid}. Will retry...`) - await this.tickMgr.waitTick(waitTimeout) - waitTimeout++ + for (const e of event.containers ?? []) { + if (e.event === NetworkEventKind.removed || e.event === NetworkEventKind.updated) { + await this.handleRefUpdate(e.container.uuid, e.container.endpoint) } } } + private async getContainerRef ( + kind: ContainerKind, + request: GetOptions + ): Promise<[ContainerUuid, ContainerEndpointRef]> { + return await this.client.request<[ContainerUuid, ContainerEndpointRef]>(opNames.getContainer, { kind, request }) + } + + async retryGetContainerRef (kind: ContainerKind, request: GetOptions): Promise<[ContainerUuid, ContainerEndpointRef]> { + let waitTimeout: number = 1 + let earlyRetry = (): void => {} + const stop = this.onUpdate(async (event) => { + // We agent is appear with a required kind, we can retry immediately + if (event.agents.some((e) => e.event === NetworkEventKind.added && e.kinds.includes(kind))) { + waitTimeout = 0 + earlyRetry() + } + }) + try { + while (true) { + try { + const ref = await this.getContainerRef(kind, request) + if (waitTimeout > 1) { + console.log(`Successfully got container ref for ${kind} after ${waitTimeout - 1} retries.`) + } + return ref + } catch (err) { + console.warn(`Error getting container ref for ${kind}. Will retry...`) + + await Promise.any([ + this.tickMgr.waitTick(waitTimeout), + new Promise((resolve) => { + earlyRetry = resolve + }) + ]) + if (waitTimeout < this.tickMgr.tps * 5) { + waitTimeout++ + } + } + } + } finally { + stop() + } + } + async release (uuid: ContainerUuid): Promise { await this.client.request(opNames.releaseContainer, { uuid }) } @@ -372,7 +423,11 @@ export class NetworkClientImpl implements NetworkClient { return await this.client.request(opNames.sendContainer, [target, operation, data]) } - onContainerUpdate (listener: ContainerUpdateListener): void { - this.containerListeners.push(listener) + onUpdate (listener: NetworkUpdateListener): () => void { + const cid = this.cid++ + this.containerListeners.set(cid, listener) + return () => { + this.containerListeners.delete(cid) + } } } diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 6ea38503e4..1a652e5351 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -17,7 +17,6 @@ export * from './client' export * from './types' const tickMgr = new TickManagerImpl(timeouts.pingInterval * 2) -tickMgr.start() export function shutdownNetworkTickMgr (): void { tickMgr.stop() @@ -30,6 +29,7 @@ process.on('exit', () => { export function createNetworkClient (url: string): NetworkClient { const [host, portStr] = url.split(':') const port = portStr != null ? parseInt(portStr, 10) : 3737 + tickMgr.start() return new NetworkClientImpl(host, port, tickMgr) } @@ -42,6 +42,7 @@ export async function createAgent ( const [host, portStr] = endpointUrl.split(':') const port = portStr != null ? parseInt(portStr, 10) : 3738 + tickMgr.start() const server = new NetworkAgentServer(tickMgr, host, '*', port) await server.start(agent) diff --git a/packages/client/src/types.ts b/packages/client/src/types.ts index 63eafa5931..75c77dbb86 100644 --- a/packages/client/src/types.ts +++ b/packages/client/src/types.ts @@ -1,16 +1,16 @@ export const opNames = { // NetworkOperations - register: 'r', - unregister: 'u', - getAgents: 'a', - getKinds: 'k', - listContainers: 'l', - getContainer: 'g', - releaseContainer: 'r', - sendContainer: 's', + register: 'ar', + unregister: 'au', + getAgents: 'al', + getKinds: 'ak', + listContainers: 'cl', + getContainer: 'gg', + releaseContainer: 'cr', + sendContainer: 'cs', // Agent operations - containerUpdate: 'c', - terminate: 't', - connect: 'c!', - disconnect: 'd' + containerUpdate: 'cu', + terminate: 'ct', + connect: 'cc', + disconnect: 'cd' } diff --git a/packages/core/src/__test__/network.spec.ts b/packages/core/src/__test__/network.spec.ts index 5f71e69694..355c511f47 100644 --- a/packages/core/src/__test__/network.spec.ts +++ b/packages/core/src/__test__/network.spec.ts @@ -1,6 +1,6 @@ import { AgentImpl } from '../agent' -import type { AgentUuid, ClientUuid, ContainerEndpointRef, ContainerKind } from '../api/types' -import type { Container } from '../containers' +import type { AgentUuid, ClientUuid, ContainerEndpointRef, ContainerKind, GetOptions } from '../api/types' +import { containerUuid, type Container } from '../containers' import { NetworkImpl } from '../network' import { TickManagerImpl } from '../utils' @@ -55,12 +55,16 @@ describe('network tests', () => { const network = new NetworkImpl(tickManager) const agent1 = new AgentImpl(agents.agent1, { - [kinds.session]: () => Promise.resolve([new DummyContainer(), '' as ContainerEndpointRef]) + [kinds.session]: async (opt: GetOptions) => ({ + uuid: opt.uuid ?? containerUuid(), + container: new DummyContainer(), + endpoint: '' as ContainerEndpointRef + }) }) await agent1.register(network) expect((network as any)._agents.size).toBe(1) - const s1 = await network.get(agents.agent1 as any, 's1' as any, { kind: kinds.session }) + const s1 = await network.get(agents.agent1 as any, kinds.session, { uuid: 's1' as any }) expect(s1).toBeDefined() }) }) diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index 2ad41739db..427d41718a 100644 --- a/packages/core/src/agent.ts +++ b/packages/core/src/agent.ts @@ -6,7 +6,7 @@ import type { ContainerEndpointRef, ContainerKind, ContainerRecord, - ContainerRequest, + GetOptions, ContainerUuid } from './api/types' import type { Container, ContainerFactory } from './containers' @@ -17,14 +17,25 @@ interface ContainerRecordImpl { endpoint: ContainerEndpointRef kind: ContainerKind + labels?: string[] + lastVisit: number } +interface AgentPendingContainer { + kind: ContainerKind + options: GetOptions + promise: Promise +} + export class AgentImpl implements NetworkAgent { // Own, managed containers - private readonly _byId = new Map>() + private readonly _byId = new Map() + private readonly _byKind = new Map>() - private readonly _containers = new Map() + // Containers pending startup + pidCounter: number = 0 + private readonly pendings = new Map() endpoint?: AgentEndpointRef | undefined @@ -49,7 +60,7 @@ export class AgentImpl implements NetworkAgent { } async list (kind?: ContainerKind): Promise { - return Array.from(this._containers.values()) + return Array.from(kind !== undefined ? (this._byKind.get(kind)?.values() ?? []) : this._byId.values()) .filter((it) => !(it instanceof Promise) && (kind === undefined || it.kind === kind)) .map((it) => ({ agentId: this.uuid, @@ -64,47 +75,86 @@ export class AgentImpl implements NetworkAgent { return Object.keys(this.factory) as ContainerKind[] } - async getContainerImpl (uuid: ContainerUuid): Promise { - let current = this._byId.get(uuid) - if (current instanceof Promise) { - current = await current - this._byId.set(uuid, current) + selectContainer (kind: ContainerKind, labels?: string[]): ContainerRecordImpl | undefined { + const list = this._byKind.get(kind) + if (list !== undefined) { + let l = list.values() + if (labels !== undefined) { + l = l.filter((it) => it.labels !== undefined && labels.every((l) => (it.labels ?? []).includes(l))) + } + return l.next()?.value } - return current } async getContainer (uuid: ContainerUuid): Promise { - return (await this.getContainerImpl(uuid))?.container + return this._byId.get(uuid)?.container } - async get (uuid: ContainerUuid, request: ContainerRequest): Promise { - const current = await this.getContainerImpl(uuid) - if (current !== undefined) { - return current.endpoint + async get (kind: ContainerKind, options: GetOptions): Promise<[ContainerUuid, ContainerEndpointRef]> { + // If uuid is fixed, we must return that one + + if (options.uuid !== undefined) { + const current = this._byId.get(options.uuid) + if (current !== undefined) { + return [current.uuid, current.endpoint] + } + // Check pending requests + for (const p of this.pendings.values()) { + if (p.kind === kind && p.options.uuid === options.uuid) { + const containerImpl = await p.promise + return [containerImpl.uuid, containerImpl.endpoint] + } + } + } else { + // If we have one with kind, return it. + const existing = this.selectContainer(kind, options.labels) + if (existing !== undefined) { + return [existing.uuid, existing.endpoint] + } } - let container: ContainerRecordImpl | Promise = this.factory[request.kind](uuid, request).then( - (r) => ({ - container: r[0], - endpoint: r[1], - kind: request.kind, - lastVisit: Date.now(), - uuid - }) - ) - this._byId.set(uuid, container) - container = await container - this._containers.set(uuid, container) - this._byId.set(uuid, container) + // Check pending container by kind + for (const p of this.pendings.values()) { + if ( + p.kind === kind && + (options.labels === undefined || + (p.options.labels !== undefined && options.labels.every((l) => (p.options.labels ?? []).includes(l)))) + ) { + const containerImpl = await p.promise + return [containerImpl.uuid, containerImpl.endpoint] + } + } - return container.endpoint + const pendingId = this.pidCounter++ + + const container = this.factory[kind](options).then(({ uuid, container, endpoint }) => ({ + container, + endpoint, + kind, + lastVisit: Date.now(), + uuid + })) + + this.pendings.set(pendingId, { kind, promise: container, options }) + const containerImpl = await container + this.pendings.delete(pendingId) + + this._byId.set(containerImpl.uuid, containerImpl) + const byKind = this._byKind.get(kind) + if (byKind !== undefined) { + byKind.set(containerImpl.uuid, containerImpl) + } else { + this._byKind.set(kind, new Map([[containerImpl.uuid, containerImpl]])) + } + + return [containerImpl.uuid, containerImpl.endpoint] } async terminate (uuid: ContainerUuid): Promise { const current = this._byId.get(uuid) if (current !== undefined) { - this._containers.delete(uuid) this._byId.delete(uuid) + this._byKind.get(current.kind)?.delete(uuid) if (current instanceof Promise) { await (await current).container.terminate() // Await promise before terminating } else { diff --git a/packages/core/src/api/agent.ts b/packages/core/src/api/agent.ts index bb1b9ee903..52129717a9 100644 --- a/packages/core/src/api/agent.ts +++ b/packages/core/src/api/agent.ts @@ -3,11 +3,11 @@ import type { AgentEndpointRef, AgentUuid, ContainerEndpointRef, - ContainerEvent, + NetworkEvent, ContainerKind, ContainerRecord, - ContainerRequest, - ContainerUuid + ContainerUuid, + GetOptions } from './types' export interface AgentRecord { @@ -21,6 +21,8 @@ export interface AgentRecord { containers: ContainerRecord[] kinds: ContainerKind[] } +export type AgentRecordInfo = Omit & { containers: number } + /** * Interface to Huly Agent on agent. */ @@ -35,13 +37,13 @@ export interface NetworkAgent { kinds: ContainerKind[] // event handled from agent to network events. - onUpdate?: (event: ContainerEvent) => Promise + onUpdate?: (event: NetworkEvent) => Promise // Send agent update info to network, if applicable. onAgentUpdate?: () => Promise // Get/Start of required container kind on agent - get: (uuid: ContainerUuid, request: ContainerRequest) => Promise + get: (kind: ContainerKind, options: GetOptions) => Promise<[ContainerUuid, ContainerEndpointRef]> // A low level reference to container getContainer: (uuid: ContainerUuid) => Promise diff --git a/packages/core/src/api/client.ts b/packages/core/src/api/client.ts index 5da3bccdb6..f937e98d8a 100644 --- a/packages/core/src/api/client.ts +++ b/packages/core/src/api/client.ts @@ -1,14 +1,14 @@ -import type { AgentRecord, NetworkAgent } from './agent' +import type { AgentRecordInfo, NetworkAgent } from './agent' import type { ContainerEndpointRef, - ContainerEvent, + NetworkEvent, ContainerKind, ContainerRecord, - ContainerRequest, - ContainerUuid + ContainerUuid, + GetOptions } from './types' -export type ContainerUpdateListener = (event: ContainerEvent) => Promise +export type NetworkUpdateListener = (event: NetworkEvent) => Promise /** * Interface to Huly network. @@ -24,7 +24,7 @@ export interface NetworkClient { */ register: (agent: NetworkAgent) => Promise - agents: () => Promise + agents: () => Promise // A full uniq set of supported container kinds. kinds: () => Promise @@ -33,7 +33,7 @@ export interface NetworkClient { * Get/Start of required container kind on agent * Will start a required container on agent, if not already started. */ - get: (uuid: ContainerUuid, request: ContainerRequest) => Promise + get: (kind: ContainerKind, options: GetOptions) => Promise list: (kind?: ContainerKind) => Promise @@ -41,7 +41,8 @@ export interface NetworkClient { request: (target: ContainerUuid, operation: string, data?: any) => Promise // Register on container update listener - onContainerUpdate: (listener: ContainerUpdateListener) => void + // Return unsubscribe function + onUpdate: (listener: NetworkUpdateListener) => () => void // We could wait for a connection for a time period. // If timeout === 0, we wait indefinitely. diff --git a/packages/core/src/api/network.ts b/packages/core/src/api/network.ts index 456122a3d2..4d43411c58 100644 --- a/packages/core/src/api/network.ts +++ b/packages/core/src/api/network.ts @@ -1,13 +1,13 @@ -import type { AgentRecord, NetworkAgent } from './agent' +import type { AgentRecord, AgentRecordInfo, NetworkAgent } from './agent' import type { - ContainerEndpointRef, AgentUuid, ClientUuid, + ContainerEndpointRef, + NetworkEvent, ContainerKind, - ContainerUuid, - ContainerRequest, ContainerRecord, - ContainerEvent + ContainerUuid, + GetOptions } from './types' /** @@ -27,7 +27,7 @@ export interface Network { // Mark an agent as alive (updates lastSeen timestamp) ping: (agentId: AgentUuid | ClientUuid) => void - agents: () => Promise + agents: () => Promise // A full uniq set of supported container kinds. kinds: () => Promise @@ -36,7 +36,7 @@ export interface Network { * Get/Start of required container kind on agent * Will start a required container on agent, if not already started. */ - get: (client: ClientUuid, uuid: ContainerUuid, request: ContainerRequest) => Promise + get: (client: ClientUuid, kind: ContainerKind, options: GetOptions) => Promise<[ContainerUuid, ContainerEndpointRef]> /** * Release a container for a client, if container is not used anymore it will be shutdown with a shutdown delay. @@ -52,7 +52,7 @@ export interface Network { } export interface NetworkWithClients { - addClient: (clientUuid: ClientUuid, onContainer?: (event: ContainerEvent) => Promise) => void + addClient: (clientUuid: ClientUuid, onContainer?: (event: NetworkEvent) => Promise) => void removeClient: (clientUuid: ClientUuid) => void // When client is registering agent. diff --git a/packages/core/src/api/timeouts.ts b/packages/core/src/api/timeouts.ts index 108f38589c..1da76a7556 100644 --- a/packages/core/src/api/timeouts.ts +++ b/packages/core/src/api/timeouts.ts @@ -1,4 +1,5 @@ export const timeouts = { - aliveTimeout: 3, // seconds - timeout for detecting dead agents + aliveTimeout: 3, // seconds - timeout for detecting dead agents/clients + unusedContainerTimeout: 5, // 10 seconds for container to be terminated because of being unused pingInterval: 1 // seconds - how often to ping agents } diff --git a/packages/core/src/api/types.ts b/packages/core/src/api/types.ts index b34447a88f..581e408881 100644 --- a/packages/core/src/api/types.ts +++ b/packages/core/src/api/types.ts @@ -17,14 +17,19 @@ export interface ContainerRecord { labels?: string[] } -export interface ContainerEvent { - added: ContainerRecord[] - deleted: ContainerRecord[] - updated: ContainerRecord[] +export enum NetworkEventKind { + added = 0, + updated = 1, + removed = 2 +} +export interface NetworkEvent { + agents: { id: AgentUuid, kinds: ContainerKind[], event: NetworkEventKind }[] + containers: { container: ContainerRecord, event: NetworkEventKind }[] } -export interface ContainerRequest { - kind: ContainerKind +export interface GetOptions { + uuid?: ContainerUuid + extra?: Record // Extra parameters for container start labels?: string[] diff --git a/packages/core/src/containers.ts b/packages/core/src/containers.ts index 27795bc2a5..967be957cc 100644 --- a/packages/core/src/containers.ts +++ b/packages/core/src/containers.ts @@ -1,4 +1,5 @@ -import type { ClientUuid, ContainerEndpointRef, ContainerRecord, ContainerRequest, ContainerUuid } from './api/types' +import type { ClientUuid, ContainerEndpointRef, ContainerUuid, GetOptions } from './api/types' +import { v4 as uuidv4 } from 'uuid' export interface Container { request: (operation: string, data?: any, clientId?: ClientUuid) => Promise @@ -15,13 +16,9 @@ export interface Container { } export type ContainerFactory = ( - uuid: ContainerUuid, - request: ContainerRequest -) => Promise<[Container, ContainerEndpointRef]> + request: GetOptions +) => Promise<{ uuid: ContainerUuid, container: Container, endpoint: ContainerEndpointRef }> -export interface ContainerRecordImpl { - record: ContainerRecord - endpoint: ContainerEndpointRef | Promise - - clients: Set +export function containerUuid (): ContainerUuid { + return uuidv4() as ContainerUuid } diff --git a/packages/core/src/network.ts b/packages/core/src/network.ts index 0c6763d9fe..48cd37d06a 100644 --- a/packages/core/src/network.ts +++ b/packages/core/src/network.ts @@ -1,23 +1,32 @@ -import type { AgentRecord, NetworkAgent } from './api/agent' +import type { AgentRecord, AgentRecordInfo, NetworkAgent } from './api/agent' import type { Network, NetworkWithClients } from './api/network' import { timeouts } from './api/timeouts' +import { NetworkEventKind } from './api/types' import type { AgentEndpointRef, AgentUuid, ClientUuid, ContainerEndpointRef, - ContainerEvent, + NetworkEvent, ContainerKind, ContainerRecord, - ContainerRequest, - ContainerUuid + ContainerUuid, + GetOptions } from './api/types' import type { TickManager } from './api/utils' -import type { ContainerRecordImpl } from './containers' + +export interface ContainerRecordImpl { + record: ContainerRecord + endpoint: ContainerEndpointRef + agent: AgentRecordImpl + + clients: Set +} interface AgentRecordImpl { + id: AgentUuid api: NetworkAgent - containers: Map + containers: Set endpoint?: AgentEndpointRef kinds: ContainerKind[] @@ -27,10 +36,18 @@ interface AgentRecordImpl { interface ClientRecordImpl { lastSeen: number containers: Set - onContainer?: (event: ContainerEvent) => Promise + onContainer?: (event: NetworkEvent) => Promise agents: Set } + +interface PendingContainer { + agent: AgentUuid + kind: ContainerKind + options: GetOptions + promise: Promise + clients: Set +} /** * Server network implementation. */ @@ -39,13 +56,17 @@ export class NetworkImpl implements Network, NetworkWithClients { private readonly _agents = new Map() - private readonly _containers = new Map() + private readonly _containers = new Map() + + private pidCounter: number = 0 + + private readonly pending = new Map() private readonly _clients = new Map() - private readonly _orphanedContainers = new Map() + private readonly _orphanedContainers = new Map() - private eventQueue: ContainerEvent[] = [] + private eventQueue: NetworkEvent[] = [] private readonly stopTick?: () => void @@ -64,15 +85,13 @@ export class NetworkImpl implements Network, NetworkWithClients { this.stopTick?.() } - async agents (): Promise { + async agents (): Promise { return Array.from( this._agents.values().map(({ api, containers }) => ({ agentId: api.uuid, endpoint: api.endpoint, kinds: api.kinds, - containers: Array.from(containers.values()) - .filter((it) => !(it.endpoint instanceof Promise)) - .map(({ record, endpoint }) => ({ ...record, endpoint: endpoint as ContainerEndpointRef })) + containers: containers.size })) ) } @@ -87,99 +106,100 @@ export class NetworkImpl implements Network, NetworkWithClients { } async list (kind?: ContainerKind): Promise { - return Array.from(this._agents.values()) - .flatMap((it) => Array.from(it.containers.values())) + return Array.from(this._containers.values()) .filter((it) => kind === undefined || it.record.kind === kind) .map((it) => it.record) } async request (target: ContainerUuid, operation: string, data?: any): Promise { - const agentId = this._containers.get(target) - if (agentId === undefined) { + const container = this._containers.get(target) + if (container === undefined) { throw new Error(`Container ${target} not found`) } - const agent = this._agents.get(agentId) - if (agent === undefined) { - throw new Error(`Agent ${agentId} not found for container ${target}`) - } - const container = agent.containers.get(target) - if (container === undefined) { - throw new Error(`Container ${target} not registered on agent ${agentId}`) - } - return await agent.api.request(target, operation, data) + return await container.agent?.api.request(target, operation, data) } async register (record: AgentRecord, agent: NetworkAgent): Promise { - const containers: ContainerRecord[] = record.containers - const newContainers = new Map( - containers.map((record) => [ + const newContainers: ContainerRecord[] = record.containers + const newContainersMap = new Map( + newContainers.map((record) => [ record.uuid, { record, request: { kind: record.kind }, endpoint: record.endpoint, - clients: new Set([]) + clients: new Set([]), + agent: null as any // Temporarily } ]) ) - const containerEvent: ContainerEvent = { - added: [], - deleted: [], - updated: [] - } - - // Register agent record - const oldAgent = this._agents.get(record.agentId) - if (oldAgent !== undefined) { - // In case re-register or reconnect is happened. - // Check if some of container changed endpoints. - for (const rec of containers) { - const oldRec = oldAgent.containers.get(rec.uuid) - if (oldRec !== undefined) { - if (oldRec.record.endpoint !== rec.endpoint) { - oldRec.endpoint = rec.endpoint // Update endpoint - containerEvent.updated.push(rec) - } - } - } - // Handle remove of containers - for (const oldC of oldAgent.containers.values()) { - if (newContainers.get(oldC.record.uuid) === undefined) { - containerEvent.deleted.push(oldC.record) - this._containers.delete(oldC.record.uuid) // Remove from active container registry - } - } - } - - const containersToShutdown: ContainerUuid[] = [] - - // Update active container registry. - for (const rec of containers) { - const oldAgentId = this._containers.get(rec.uuid) - if (oldAgentId === undefined) { - containerEvent.added.push(rec) - const containerImpl = newContainers.get(rec.uuid) - if (containerImpl !== undefined) { - this._orphanedContainers.set(rec.endpoint, containerImpl) - } - this._containers.set(rec.uuid, record.agentId) - } - if (oldAgentId !== undefined && oldAgentId !== record.agentId) { - containersToShutdown.push(rec.uuid) - } + const containerEvent: NetworkEvent = { + agents: [], + containers: [] } // update agent record - - this._agents.set(record.agentId, { + const agentRecord: AgentRecordImpl = { + id: record.agentId, api: agent, - containers: newContainers, + containers: new Set(), endpoint: record.endpoint, kinds: record.kinds, lastSeen: this.tickManager.now() - }) + } + const oldAgent = this._agents.get(record.agentId) + this._agents.set(record.agentId, agentRecord) + + const containersToShutdown: ContainerUuid[] = [] + + // Find removed containers + for (const cid of oldAgent?.containers ?? []) { + if (!newContainersMap.has(cid)) { + const container = this._containers.get(cid) + if (container !== undefined) { + containerEvent.containers.push({ + container: container.record, + event: NetworkEventKind.removed + }) + this._containers.delete(cid) + } + } + } + + // Update active container registry. + for (const containerImpl of newContainersMap.values()) { + containerImpl.agent = agentRecord + + const existingContainer = this._containers.get(containerImpl.record.uuid) + if (existingContainer === undefined) { + containerEvent.containers.push({ + container: containerImpl.record, + event: NetworkEventKind.added + }) + this._containers.set(containerImpl.record.uuid, containerImpl) + agentRecord.containers.add(containerImpl.record.uuid) + } else { + if (existingContainer.agent.id !== record.agentId) { + // Container already started on different agent, need to shutdown old one. + containersToShutdown.push(containerImpl.record.uuid) + } + + if (existingContainer.record.endpoint !== containerImpl.record.endpoint) { + containerEvent.containers.push({ + container: containerImpl.record, + event: NetworkEventKind.updated + }) + } + } + } + + containerEvent.agents.push({ + id: agent.uuid, + kinds: record.kinds, + event: oldAgent !== undefined ? NetworkEventKind.updated : NetworkEventKind.added + }) this.eventQueue.push(containerEvent) // Send notification to all agents about containers update. @@ -191,7 +211,7 @@ export class NetworkImpl implements Network, NetworkWithClients { if (agent === undefined) { return // Fine to ignore } - await this.processDeadAgent(agentId) + await this.processAgentRemove(agentId) } async sendEvents (): Promise { @@ -202,27 +222,38 @@ export class NetworkImpl implements Network, NetworkWithClients { } // Combine events - const finalEvent: ContainerEvent = { - added: [], - deleted: [], - updated: [] - } + const agents = new Map() + const containers = new Map() + for (const event of events) { - finalEvent.added.push(...event.added) - finalEvent.deleted.push(...event.deleted) - finalEvent.updated.push(...event.updated) + for (const agent of event.agents) { + if (agent.event === NetworkEventKind.removed) { + agents.delete(agent.id) + } else { + agents.set(agent.id, agent) + } + } + for (const container of event.containers) { + if (container.event === NetworkEventKind.removed) { + containers.delete(container.container.uuid) + } else { + containers.set(container.container.uuid, container) + } + } } - // Skip deleted events. - const deletedIds = finalEvent.deleted.map((c) => c.uuid) - finalEvent.added = finalEvent.added.filter((c) => !deletedIds.includes(c.uuid)) - finalEvent.updated = finalEvent.updated.filter((c) => !deletedIds.includes(c.uuid)) + const finalEvent: NetworkEvent = { + agents: Array.from(agents.values()), + containers: Array.from(containers.values()) + } - for (const [clientUuid, client] of Object.entries(this._clients)) { + for (const [clientUuid, client] of this._clients.entries()) { if (client.onContainer !== undefined) { try { // We should not block on broadcast to clients. - void client.onContainer(finalEvent) + void client.onContainer?.(finalEvent)?.catch((err) => { + console.error(`Error in client ${clientUuid} onContainer callback:`, err) + }) } catch (err: any) { console.error(`Error in client ${clientUuid} onContainer callback:`, err) } @@ -230,7 +261,7 @@ export class NetworkImpl implements Network, NetworkWithClients { } } - addClient (clientUuid: ClientUuid, onContainer?: (event: ContainerEvent) => Promise): void { + addClient (clientUuid: ClientUuid, onContainer?: (event: NetworkEvent) => Promise): void { const info = this._clients.get(clientUuid) ?? { lastSeen: this.tickManager.now(), containers: new Set(), @@ -268,77 +299,111 @@ export class NetworkImpl implements Network, NetworkWithClients { } } - async get (clientUuid: ClientUuid, uuid: ContainerUuid, request: ContainerRequest): Promise { + async get ( + clientUuid: ClientUuid, + kind: ContainerKind, + options: GetOptions + ): Promise<[ContainerUuid, ContainerEndpointRef]> { let client = this._clients.get(clientUuid) if (client === undefined) { client = { lastSeen: this.tickManager.now(), containers: new Set(), agents: new Set() } this._clients.set(clientUuid, client) } - client.containers.add(uuid) - const record = await this.getContainer(uuid, request, [clientUuid]) - if (record.endpoint instanceof Promise) { - return await record.endpoint - } - return record.endpoint + const record: ContainerRecordImpl = await this.getContainer(kind, options, [clientUuid]) + client.containers.add(record.record.uuid) + return [record.record.uuid, record.endpoint] } - async getContainer ( - uuid: ContainerUuid, - request: ContainerRequest, - clients: ClientUuid[] - ): Promise { - const existing = this._containers.get(uuid) - if (existing !== undefined) { - const agent = this._agents.get(existing) - const containerImpl = agent?.containers?.get(uuid) - if (containerImpl !== undefined) { - if (!(containerImpl.endpoint instanceof Promise)) { - this._orphanedContainers.delete(containerImpl.endpoint) - } + async getContainer (kind: ContainerKind, options: GetOptions, clients: ClientUuid[]): Promise { + // Reuse existing container if uuid is provided and container exists + if (options.uuid !== undefined) { + const existing = this._containers.get(options.uuid) + if (existing !== undefined) { + this._orphanedContainers.delete(existing.record.uuid) for (const cl of clients) { - containerImpl.clients.add(cl) + existing.clients.add(cl) + } + return existing + } + // Find if container is pending starting for our it + for (const p of this.pending.values()) { + if (p.kind === kind && p.options.uuid === options.uuid) { + // Add to pendings list to properly track orphaned + for (const cl of clients) { + p.clients.add(cl) + } + const containerImpl = await p.promise + for (const cl of clients) { + containerImpl.clients.add(cl) + } + return containerImpl + } + } + } else { + // Check if we have a container pending if no uuid is provided + for (const p of this.pending.values()) { + if ( + p.kind === kind && + (options.labels === undefined || + (p.options.labels !== undefined && options.labels.every((l) => (p.options.labels ?? []).includes(l)))) + ) { + // Add to pendings list to properly track orphaned + for (const cl of clients) { + p.clients.add(cl) + } + const containerImpl = await p.promise + for (const cl of clients) { + containerImpl.clients.add(cl) + } + return containerImpl } - return containerImpl } } // Select agent using round/robin and register it in agent - const suitableAgents = Array.from(this._agents.values().filter((it) => it.kinds.includes(request.kind))) + const suitableAgents = Array.from(this._agents.values().filter((it) => it.kinds.includes(kind))) if (suitableAgents.length === 0) { - throw new Error(`No suitable agents found for container ${uuid}`) + throw new Error(`No suitable agents found for container ${kind}`) } const agent = Array.from(suitableAgents)[++this.idx % suitableAgents.length] - const record: ContainerRecordImpl = { + const record: Promise = agent.api.get(kind, options).then(([uuid, endpoint]) => ({ + agent, record: { uuid, agentId: agent.api.uuid, - kind: request.kind, + kind, lastVisit: this.tickManager.now(), endpoint: '' as ContainerEndpointRef, // Placeholder, will be updated later - labels: request.labels, - extra: request.extra + labels: options.labels, + extra: options.extra }, clients: new Set(clients), - endpoint: agent.api.get(uuid, request) - } - agent.containers.set(uuid, record) - this._containers.set(uuid, agent.api.uuid) + endpoint + })) + + const pid = ++this.pidCounter + this.pending.set(pid, { agent: agent.id, kind, options, promise: record, clients: new Set(clients) }) // Wait for endpoint to be established try { - const endpointRef = await record.endpoint - record.endpoint = endpointRef + const recordImpl = await record + + agent.containers.add(recordImpl.record.uuid) + this.eventQueue.push({ - added: [record.record], - deleted: [], - updated: [] + agents: [], + containers: [{ container: recordImpl.record, event: NetworkEventKind.added }] }) - return record + + // TODO: What if container started with same id? + this._containers.set(recordImpl.record.uuid, recordImpl) + return recordImpl } catch (err: any) { - this._containers.delete(uuid) // Remove from active container registry - throw new Error(`Failed to get endpoint for container ${uuid}: ${err.message}`) + throw new Error(`Failed to get endpoint for container ${kind}: ${err.message}`) + } finally { + this.pending.delete(pid) } } @@ -348,29 +413,26 @@ export class NetworkImpl implements Network, NetworkWithClients { const existing = this._containers.get(uuid) if (existing !== undefined) { - const agent = this._agents.get(existing) - const containerImpl = agent?.containers?.get(uuid) - if (containerImpl !== undefined) { - containerImpl.clients.delete(client) - if (containerImpl.clients.size === 0 && !(containerImpl.endpoint instanceof Promise)) { - this._orphanedContainers.set(containerImpl.endpoint, containerImpl) - } + existing.clients.delete(client) + if (existing.clients.size === 0) { + this._orphanedContainers.set(existing.record.uuid, { + container: existing, + time: this.tickManager.now() + }) } } } async terminate (container: ContainerRecordImpl): Promise { this._containers.delete(container.record.uuid) // Remove from active container registry - this._orphanedContainers.delete(container.record.endpoint) + this._orphanedContainers.delete(container.record.uuid) this.eventQueue.push({ - added: [], - deleted: [container.record], - updated: [] + agents: [], + containers: [{ container: container.record, event: NetworkEventKind.removed }] }) - const agent = this._agents.get(container.record.agentId) - agent?.containers.delete(container.record.uuid) + container.agent.containers.delete(container.record.uuid) - await agent?.api.terminate(container.record.uuid) + await container.agent.api.terminate(container.record.uuid) } /** @@ -414,49 +476,60 @@ export class NetworkImpl implements Network, NetworkWithClients { // Remove dead agents and their containers for (const agentId of deadAgents) { - await this.processDeadAgent(agentId) + await this.processAgentRemove(agentId) } // Handle termination of orphaned containers - for (const container of [...this._orphanedContainers.values()]) { - void this.terminate(container).catch((err) => { - console.error(`Failed to terminate orphaned container ${container.record.uuid}: ${err.message}`) - }) + for (const { container, time } of [...this._orphanedContainers.values()]) { + if (now - time > timeouts.unusedContainerTimeout * 1000) { + void this.terminate(container).catch((err) => { + console.error(`Failed to terminate orphaned container ${container.record.uuid}: ${err.message}`) + }) + } } } /** * Remove a dead agent and clean up its containers */ - private async processDeadAgent (agentId: AgentUuid): Promise { + private async processAgentRemove (agentId: AgentUuid): Promise { const agent = this._agents.get(agentId) if (agent == null) { return } - console.log(`Removing dead agent ${agentId} and its ${agent.containers.size} containers`) + console.log(`Removing agent ${agentId} and its ${agent.containers.size} containers`) // Collect containers to remove const affectedContainers: ContainerRecordImpl[] = [] - for (const [containerId, containerRecord] of agent.containers.entries()) { - affectedContainers.push(containerRecord) + for (const containerId of agent.containers.values()) { + const c = this._containers.get(containerId) + if (c !== undefined) { + affectedContainers.push(c) + } this._containers.delete(containerId) } + // We need to clean pending ones + for (const [pid, p] of this.pending.entries()) { + if (p.agent === agentId) { + this.pending.delete(pid) + } + } + // Remove agent this._agents.delete(agentId) - const containerEvent: ContainerEvent = { - added: [], - deleted: [], - updated: [] + const containerEvent: NetworkEvent = { + agents: [{ id: agentId, kinds: agent.kinds, event: NetworkEventKind.removed }], + containers: [] } // We need to add requests for all used containers for (const container of affectedContainers) { - this._orphanedContainers.delete(container.record.endpoint) + this._orphanedContainers.delete(container.record.uuid) // We just send container is deleted, so clients should re-request whem again. - containerEvent.deleted.push(container.record) + containerEvent.containers.push({ container: container.record, event: NetworkEventKind.removed }) } if (affectedContainers.length > 0) { this.eventQueue.push(containerEvent) diff --git a/packages/core/src/utils.ts b/packages/core/src/utils.ts index 27f8969272..de27a8fba9 100644 --- a/packages/core/src/utils.ts +++ b/packages/core/src/utils.ts @@ -29,6 +29,8 @@ export class TickManagerImpl implements TickManager { tickListeners = new Map void)[]>() + started: boolean = false + constructor (readonly tps: number) { if (tps > 1000 || tps < 1) { throw new Error('Ticks per second has an invalid value: must be >= 1 && <= 1000') @@ -88,6 +90,10 @@ export class TickManagerImpl implements TickManager { stop: () => void = () => {} start (): void { + if (this.started) { + return + } + this.started = true const to = setInterval( () => { this.tick().catch((err) => { @@ -97,6 +103,7 @@ export class TickManagerImpl implements TickManager { Math.round(1000 / this.tps) ) this.stop = () => { + this.started = false clearInterval(to) } } diff --git a/packages/server/src/__test__/network.spec.ts b/packages/server/src/__test__/network.spec.ts index c4f3d8fa6a..dc13c6e760 100644 --- a/packages/server/src/__test__/network.spec.ts +++ b/packages/server/src/__test__/network.spec.ts @@ -2,6 +2,7 @@ import { AgentImpl, composeCID, containerOnAgentEndpointRef, + containerUuid, EndpointKind, NetworkImpl, parseEndpointRef, @@ -29,13 +30,19 @@ const kinds = { function createAgent1 (tickMgr: TickManagerImpl, networkClient: NetworkClient): AgentImpl { const agent: AgentImpl = new AgentImpl(agents.agent1, { - [kinds.session]: async (uuid) => { - return [new DummySessionContainer(), containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid)] + [kinds.session]: async (options) => { + const uuid = options.uuid ?? containerUuid() + return { + uuid, + container: new DummySessionContainer(), + endpoint: containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid) + } }, - [kinds.workspace]: async (uuid) => { + [kinds.workspace]: async (options) => { + const uuid = options.uuid ?? containerUuid() const container = new DummyWorkspaceContainer(uuid, agent.uuid, networkClient) const endpoint = await container.start(tickMgr) - return [container, endpoint] + return { uuid, container, endpoint } } }) return agent @@ -59,6 +66,7 @@ describe('check network server is working fine', () => { await networkClient.close() await network.close() + tickMgr.stop() }) it('check routed connection and requests', async () => { @@ -83,10 +91,10 @@ describe('check network server is working fine', () => { const _agents = await networkClient.agents() expect(_agents.length).toEqual(1) expect(_agents[0].agentId).toEqual(agents.agent1) - expect(_agents[0].containers.length).toEqual(0) + expect(_agents[0].containers).toEqual(0) // Start a new container and check if messaging works - const containerRef = await networkClient.get(composeCID('session', 'user1'), { kind: kinds.session }) + const containerRef = await networkClient.get(kinds.session, { uuid: composeCID('session', 'user1') }) const data = parseEndpointRef(containerRef.endpoint) expect(data.kind).toEqual(EndpointKind.routed) @@ -113,8 +121,9 @@ describe('check network server is working fine', () => { expect(events.length).toEqual(1) expect(events[0]).toEqual('event') - await agentServer.close() await networkClient.close() + await agentServer.close() await network.close() + tickMgr.stop() }) }) diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 7c837f4edc..ac58c132ed 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -6,7 +6,7 @@ import { type ContainerEndpointRef, type ContainerKind, type ContainerRecord, - type ContainerRequest, + type GetOptions, type ContainerUuid, type Network, type NetworkAgent, @@ -25,8 +25,8 @@ class AgentCallbackHandler implements NetworkAgent { readonly client: ClientUuid ) {} - async get (uuid: ContainerUuid, request: ContainerRequest): Promise { - return await this.rpcServer.request(this.client, opNames.getContainer, [this.uuid, [uuid, request]]) + async get (kind: ContainerKind, request: GetOptions): Promise<[ContainerUuid, ContainerEndpointRef]> { + return await this.rpcServer.request(this.client, opNames.getContainer, [this.uuid, [kind, request]]) } async list (kind?: ContainerKind): Promise { @@ -92,9 +92,9 @@ export class NetworkServer implements BackRPCServerHandler { break } case opNames.getContainer: { - const uuid: ContainerUuid = params.uuid - const request: ContainerRequest = params.request - await send(await this.network.get(client, uuid, request)) + const kind: ContainerKind = params.kind + const request: GetOptions = params.request + await send(await this.network.get(client, kind, request)) break } case opNames.releaseContainer: { @@ -164,6 +164,7 @@ export class NetworkServer implements BackRPCServerHandler { new AgentCallbackHandler(server, agentUuid, endpoint, kinds, client) ) this.network.mapAgent(client, agentUuid) + console.log('Client registered agent', agentUuid, 'with containers', containers.length, 'kinds', kinds) await send(res) } @@ -175,6 +176,7 @@ export class NetworkServer implements BackRPCServerHandler { const agentUuid: AgentUuid = params.uuid await this.network.unregister(agentUuid) this.network.unmapAgent(client, agentUuid) + console.log('Client unregistered agent', agentUuid) await send('ok') } } diff --git a/pods/network-pod/Dockerfile b/pods/network-pod/Dockerfile index 379738a6a5..1156a8b5e3 100644 --- a/pods/network-pod/Dockerfile +++ b/pods/network-pod/Dockerfile @@ -1,4 +1,4 @@ -FROM hardcoreeng/base:v20250310 +FROM hardcoreeng/base-slim:latest WORKDIR /usr/src/app diff --git a/pods/network-tool/Dockerfile b/pods/network-tool/Dockerfile index 379738a6a5..1156a8b5e3 100644 --- a/pods/network-tool/Dockerfile +++ b/pods/network-tool/Dockerfile @@ -1,4 +1,4 @@ -FROM hardcoreeng/base:v20250310 +FROM hardcoreeng/base-slim:latest WORKDIR /usr/src/app diff --git a/pods/network-tool/bench_cli.md b/pods/network-tool/bench_cli.md new file mode 100644 index 0000000000..d52607dfab --- /dev/null +++ b/pods/network-tool/bench_cli.md @@ -0,0 +1,17 @@ +## Agent + +```bash +rushx run bench-agent -n localhost:37371 +``` + +## List agents + +```bash +rushx run list-agents -n localhost:37371 +``` + +# Client + +```bash +rushx run bench -n localhost:37371 +``` diff --git a/pods/network-tool/package.json b/pods/network-tool/package.json index 4390fd4558..e094416b11 100644 --- a/pods/network-tool/package.json +++ b/pods/network-tool/package.json @@ -22,6 +22,7 @@ "docker:abuild": "docker build -t hardcoreeng/network-tool . --platform=linux/arm64 && ../../common/scripts/docker_tag_push.sh hardcoreeng/network-tool", "docker:staging": "../../common/scripts/docker_tag.sh hardcoreeng/network-tool staging", "docker:push": "../../common/scripts/docker_tag.sh hardcoreeng/network-tool", + "run": "(rushx bundle || true) && node bundle/bundle.js", "build:watch": "compile", "format": "format src", "test": "jest --passWithNoTests --silent --forceExit", diff --git a/pods/network-tool/src/agents.ts b/pods/network-tool/src/agents.ts index 57cc72df1e..e271459767 100644 --- a/pods/network-tool/src/agents.ts +++ b/pods/network-tool/src/agents.ts @@ -13,10 +13,7 @@ export function registerAgentOperations (): void { const agents = await client.agents() console.log(`Active agents: ${agents.length}`) for (const agent of agents) { - console.log(` - Agent: ${agent.agentId} at ${agent.endpoint} with ${agent.containers.length} containers\n`) - for (const container of agent.containers) { - console.log(` - Container: ${container.uuid} kind: ${container.kind} endpoint: ${container.endpoint}\n`) - } + console.log(` - Agent: ${agent.agentId} at ${agent.endpoint} with ${agent.containers} containers\n`) } await client.close() diff --git a/pods/network-tool/src/benchmark.ts b/pods/network-tool/src/benchmark.ts index 1374d0edf9..7d02ac4c60 100644 --- a/pods/network-tool/src/benchmark.ts +++ b/pods/network-tool/src/benchmark.ts @@ -1,12 +1,14 @@ import { createAgent, createNetworkClient } from '@hcengineering/network-client' import { containerOnAgentEndpointRef, + containerUuid, type AgentEndpointRef, type ClientUuid, type Container, type ContainerKind, type ContainerReference, - type ContainerUuid + type ContainerUuid, + type GetOptions } from '@hcengineering/network-core' import { program } from 'commander' import { addShutdownHandler, tickManager } from './utils' @@ -53,9 +55,14 @@ export function registerBenchmark (): void { const client = createNetworkClient(network) const { agent, server } = await createAgent(cmd.endpoint, { - [benchmarkContainer]: async (uuid: ContainerUuid) => { + [benchmarkContainer]: async (options: GetOptions) => { console.log('Starting bench container') - return [new BenchmarkContainer(uuid), containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid)] + const uuid = options.uuid ?? containerUuid() + return { + uuid, + container: new BenchmarkContainer(uuid), + endpoint: containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid) + } } }) @@ -82,12 +89,16 @@ export function registerBenchmark (): void { .option('-e, --exit ', 'Exit after . If 1 exit on end', '0') .action(async (cmd: { network: string, count: number, requests: number, exit: number }) => { const network = process.env.NETWORK_HOST ?? cmd.network + console.log('Benchmark agent') const client = createNetworkClient(network) + console.log('Connected to network') + const st = tickManager.now() const containers: ContainerReference[] = [] for (let i = 0; i < cmd.count; i++) { - const container = await client.get(`benchmark-${i}` as ContainerUuid, { kind: benchmarkContainer }) + console.log('request container:' + i) + const container = await client.get(benchmarkContainer, { uuid: `benchmark-${i}` as ContainerUuid }) console.log('container obtained', container.endpoint) containers.push(container) } diff --git a/pods/network-tool/src/request.ts b/pods/network-tool/src/request.ts index 9ec327575b..8a36ad12cd 100644 --- a/pods/network-tool/src/request.ts +++ b/pods/network-tool/src/request.ts @@ -1,13 +1,19 @@ import { createNetworkClient } from '@hcengineering/network-client' -import type { ContainerEvent, ContainerKind, ContainerUuid } from '@hcengineering/network-core' +import { + type NetworkEvent, + type ContainerKind, + type ContainerUuid, + NetworkEventKind +} from '@hcengineering/network-core' import { program } from 'commander' import { addShutdownHandler, tickManager } from './utils' export function registerRequest (): void { program - .command('request ') + .command('request ') .description('Connect to network and request a container undefinitely') .option('-n, --network ', 'Network address', 'localhost:3737') + .option('-u, --uuid ', 'Container UUID to request') .option( '-l, --label