Skip to content
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
126 changes: 126 additions & 0 deletions apps/api/src/__tests__/grpc-probe-pool.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import {
CLIENT_POOL_LIMIT,
__clientPoolSize,
__resetClientPool,
getClient,
} from "../services/grpc-probe.js";

/**
* Regression test for the gRPC clientPool memory leak: the pool created one
* client per endpoint and never closed/evicted any, so churning provider
* endpoints accumulated dead HTTP/2 channels until the task OOM-killed.
*
* These tests exercise the eviction + close logic through the injectable
* createClient factory — no real network/gRPC channel is opened.
*/

interface FakeClient {
endpoint: string;
close: ReturnType<typeof vi.fn>;
}

function makeFactory() {
const created: FakeClient[] = [];
const create = (endpoint: string) => {
const client = { endpoint, close: vi.fn() };
created.push(client);
// The cast matches the real grpc ServiceClient instance shape closely
// enough for the pool, which only ever calls .close() on it here.
return client as unknown as InstanceType<
typeof import("@grpc/grpc-js").ServiceClientConstructor
>;
};
return { created, create };
}

afterEach(() => {
__resetClientPool();
});

describe("grpc-probe clientPool", () => {
it("reuses the same client for a repeated endpoint (no new channel)", () => {
const { created, create } = makeFactory();

const a = getClient("provider-1.lava.build:443", create);
const b = getClient("provider-1.lava.build:443", create);

expect(a).toBe(b);
expect(created).toHaveLength(1);
expect(__clientPoolSize()).toBe(1);
});

it("creates distinct clients for distinct endpoints", () => {
const { created, create } = makeFactory();

getClient("provider-1.lava.build:443", create);
getClient("provider-2.lava.build:443", create);

expect(created).toHaveLength(2);
expect(__clientPoolSize()).toBe(2);
});

it("is bounded — never exceeds CLIENT_POOL_LIMIT entries", () => {
const { create } = makeFactory();

// Push well past the cap with unique endpoints (the leak scenario).
for (let i = 0; i < CLIENT_POOL_LIMIT + 50; i++) {
getClient(`provider-${i}.lava.build:443`, create);
}

expect(__clientPoolSize()).toBe(CLIENT_POOL_LIMIT);
});

it("closes the channel of every evicted client", () => {
const { created, create } = makeFactory();

const overflow = 50;
for (let i = 0; i < CLIENT_POOL_LIMIT + overflow; i++) {
getClient(`provider-${i}.lava.build:443`, create);
}

// The first `overflow` endpoints are the least-recently-used and must have
// been evicted AND closed.
const closed = created.filter((c) => c.close.mock.calls.length > 0);
expect(closed).toHaveLength(overflow);
for (let i = 0; i < overflow; i++) {
expect(created[i]!.close).toHaveBeenCalledTimes(1);
}
});

it("keeps recently-used endpoints warm (LRU touch survives eviction)", () => {
const { create } = makeFactory();

// Fill the pool exactly to the cap.
for (let i = 0; i < CLIENT_POOL_LIMIT; i++) {
getClient(`provider-${i}.lava.build:443`, create);
}

// Touch endpoint 0 so it becomes most-recently-used.
const warm = getClient("provider-0.lava.build:443", create);

// One more unique endpoint forces a single eviction — it must NOT be
// endpoint 0 (which we just touched); endpoint 1 is now the LRU.
getClient("provider-new.lava.build:443", create);

// Endpoint 0 still resolves to the same live client (no re-create).
const stillWarm = getClient("provider-0.lava.build:443", create);
expect(stillWarm).toBe(warm);
expect(__clientPoolSize()).toBe(CLIENT_POOL_LIMIT);
});

it("__resetClientPool closes all channels and empties the pool", () => {
const { created, create } = makeFactory();

getClient("provider-1.lava.build:443", create);
getClient("provider-2.lava.build:443", create);
expect(__clientPoolSize()).toBe(2);

__resetClientPool();

expect(__clientPoolSize()).toBe(0);
for (const c of created) {
expect(c.close).toHaveBeenCalledTimes(1);
}
});
});
84 changes: 78 additions & 6 deletions apps/api/src/services/grpc-probe.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,18 +27,90 @@ const proto = grpc.loadPackageDefinition(packageDef) as unknown as {

const RelayerClient = proto.lavanet.lava.pairing.Relayer;

// Reuse gRPC clients per endpoint — avoids TLS handshake per probe call
const clientPool = new Map<string, InstanceType<grpc.ServiceClientConstructor>>();
type RelayerClientInstance = InstanceType<grpc.ServiceClientConstructor>;

function getClient(endpoint: string): InstanceType<grpc.ServiceClientConstructor> {
let client = clientPool.get(endpoint);
if (client) return client;
/**
* Max number of gRPC clients kept warm at once. Each entry owns a live HTTP/2
* channel (socket + keepalive timers + native buffers), so an unbounded pool
* leaks memory and file descriptors as provider endpoints churn (restarts,
* re-IPs, stake changes) over the process lifetime. We bound the pool and
* close the least-recently-used channel on overflow.
*
* Override via HEALTH_PROBE_CLIENT_POOL_MAX. The provider universe in flight at
* any moment is a few hundred endpoints; 256 keeps the active set warm while
* capping retained channels.
*/
const CLIENT_POOL_MAX = (() => {
const raw = process.env.HEALTH_PROBE_CLIENT_POOL_MAX;
const n = raw !== undefined ? parseInt(raw, 10) : NaN;
return Number.isFinite(n) && n > 0 ? n : 256;
})();

client = new RelayerClient(endpoint, grpc.credentials.createSsl());
// Reuse gRPC clients per endpoint — avoids TLS handshake per probe call.
// Map insertion order gives us LRU ordering: a hit re-inserts the key to the
// tail, so the least-recently-used entry is always Map's first key.
const clientPool = new Map<string, RelayerClientInstance>();

/**
* Build a real SSL-backed Relayer client. Injectable so the pool's
* eviction/closing behaviour can be unit-tested without network I/O.
*/
function defaultCreateClient(endpoint: string): RelayerClientInstance {
return new RelayerClient(endpoint, grpc.credentials.createSsl());
}

function closeClient(client: RelayerClientInstance): void {
try {
// grpc-js clients expose close() to tear down the channel; guard in case a
// test double or future client shape omits it.
(client as { close?: () => void }).close?.();
} catch {
// A failed close must not crash the probe loop — the channel will be GC'd
// once unreferenced regardless.
}
}

export function getClient(
endpoint: string,
createClient: (endpoint: string) => RelayerClientInstance = defaultCreateClient,
): RelayerClientInstance {
const existing = clientPool.get(endpoint);
if (existing) {
// LRU touch: move to tail so active endpoints survive eviction.
clientPool.delete(endpoint);
clientPool.set(endpoint, existing);
return existing;
}

// Evict least-recently-used entries (and close their channels) until there's
// room for the newcomer.
while (clientPool.size >= CLIENT_POOL_MAX) {
const lruKey = clientPool.keys().next().value;
if (lruKey === undefined) break;
const lru = clientPool.get(lruKey);
clientPool.delete(lruKey);
if (lru) closeClient(lru);
}

const client = createClient(endpoint);
clientPool.set(endpoint, client);
return client;
}

/** Test hook: close every pooled channel and empty the pool. */
export function __resetClientPool(): void {
for (const client of clientPool.values()) closeClient(client);
clientPool.clear();
}

/** Test hook: current pool occupancy. */
export function __clientPoolSize(): number {
return clientPool.size;
}

/** The configured pool cap — exported for tests/observability. */
export const CLIENT_POOL_LIMIT = CLIENT_POOL_MAX;

export interface ProbeResult {
latestBlock: number;
lavaEpoch: number;
Expand Down
Loading