Feature/issues 1009 998 986 984

backend/elasticsearch/__tests__/connectionPool.test.ts

@@ -0,0 +1,278 @@
+/**
+ * Tests — ElasticsearchConnectionPool (Issue #986)
+ */
+
+import {
+  ElasticsearchConnectionPool,
+  DnsCache,
+  resetDefaultPool,
+  type ConnectionPoolConfig,
+} from '../connectionPool';
+
+const BASE_CONFIG: ConnectionPoolConfig = {
+  primaryHost: 'es-primary',
+  primaryPort: 9200,
+  poolSize: 5,
+  acquireTimeoutMs: 200,
+  idleTimeoutMs: 500,
+  leakThresholdMs: 300,
+  dnsCacheTtlMs: 1_000,
+  maintenanceIntervalMs: 100,
+};
+
+describe('ElasticsearchConnectionPool', () => {
+  let pool: ElasticsearchConnectionPool;
+
+  beforeEach(() => {
+    jest.useFakeTimers();
+    pool = new ElasticsearchConnectionPool(BASE_CONFIG);
+  });
+
+  afterEach(() => {
+    pool.shutdown();
+    resetDefaultPool();
+    jest.useRealTimers();
+  });
+
+  // ── Pool initialisation ───────────────────────────────────────────────────
+
+  it('initialises pool with correct size', () => {
+    expect(pool.size()).toBe(BASE_CONFIG.poolSize);
+    expect(pool.idleCount()).toBe(BASE_CONFIG.poolSize);
+    expect(pool.activeCount()).toBe(0);
+  });
+
+  it('assigns primary role to all connections when no replicas', () => {
+    const conns = pool.listConnections();
+    expect(conns.every((c) => c.role === 'primary')).toBe(true);
+  });
+
+  it('assigns replica roles when replicas configured', () => {
+    const p = new ElasticsearchConnectionPool({
+      ...BASE_CONFIG,
+      replicas: [{ host: 'replica-1', port: 9201 }],
+      poolSize: 4,
+    });
+    const roles = p.listConnections().map((c) => c.role);
+    expect(roles).toContain('replica');
+    expect(roles).toContain('primary');
+    p.shutdown();
+  });
+
+  // ── Acquire / Release ─────────────────────────────────────────────────────
+
+  it('acquire returns a connection and marks it in-use', async () => {
+    const { connection } = await pool.acquire();
+    expect(connection.inUse).toBe(true);
+    expect(connection.acquiredAt).toBeDefined();
+  });
+
+  it('release frees connection back to pool', async () => {
+    const { connection } = await pool.acquire();
+    expect(pool.activeCount()).toBe(1);
+    pool.release(connection.id);
+    expect(pool.activeCount()).toBe(0);
+    expect(pool.idleCount()).toBe(BASE_CONFIG.poolSize);
+  });
+
+  it('queued acquires are resolved after release', async () => {
+    // Exhaust pool
+    const acquired: Array<{ connection: { id: string } }> = [];
+    for (let i = 0; i < BASE_CONFIG.poolSize; i++) {
+      acquired.push(await pool.acquire());
+    }
+    expect(pool.idleCount()).toBe(0);
+
+    // Queue a waiter
+    const waiting = pool.acquire();
+    // Release one connection
+    pool.release(acquired[0]!.connection.id);
+
+    const resolved = await waiting;
+    expect(resolved.connection.inUse).toBe(true);
+    // Cleanup
+    for (let i = 1; i < acquired.length; i++) pool.release(acquired[i]!.connection.id);
+    pool.release(resolved.connection.id);
+  });
+
+  it('acquire times out when pool exhausted', async () => {
+    // Exhaust all connections
+    const acquired: Array<{ connection: { id: string } }> = [];
+    for (let i = 0; i < BASE_CONFIG.poolSize; i++) {
+      acquired.push(await pool.acquire());
+    }
+
+    const p = pool.acquire();
+    jest.advanceTimersByTime(BASE_CONFIG.acquireTimeoutMs + 10);
+    await expect(p).rejects.toThrow('acquire timed out');
+
+    // Cleanup
+    for (const a of acquired) pool.release(a.connection.id);
+  });
+
+  it('release is a no-op for unknown connection id', () => {
+    expect(() => pool.release('nonexistent')).not.toThrow();
+  });
+
+  // ── withConnection ────────────────────────────────────────────────────────
+
+  it('withConnection releases connection even when fn throws', async () => {
+    await expect(
+      pool.withConnection(async () => {
+        throw new Error('query failed');
+      }),
+    ).rejects.toThrow('query failed');
+    expect(pool.idleCount()).toBe(BASE_CONFIG.poolSize);
+  });
+
+  it('withConnection passes connection to fn', async () => {
+    const result = await pool.withConnection(async (conn) => conn.id);
+    expect(typeof result).toBe('string');
+  });
+
+  // ── Read routing ──────────────────────────────────────────────────────────
+
+  it('readOnly acquire prefers replica connections', async () => {
+    const p = new ElasticsearchConnectionPool({
+      ...BASE_CONFIG,
+      replicas: [{ host: 'replica-1', port: 9201 }],
+      poolSize: 4,
+    });
+    const { connection } = await p.acquire(true);
+    expect(connection.role).toBe('replica');
+    p.release(connection.id);
+    p.shutdown();
+  });
+
+  it('readOnly falls back to primary when replicas exhausted', async () => {
+    const p = new ElasticsearchConnectionPool({
+      ...BASE_CONFIG,
+      replicas: [{ host: 'replica-1', port: 9201 }],
+      poolSize: 4,
+    });
+    const replicas = p.listConnections().filter((c) => c.role === 'replica');
+    // Acquire all replicas
+    for (const r of replicas) await p.acquire(true);
+
+    // Next read-only acquire should fall back to primary
+    const { connection } = await p.acquire(true);
+    expect(connection.role).toBe('primary');
+    p.shutdown();
+  });
+
+  // ── Leak detection ────────────────────────────────────────────────────────
+
+  it('emits leak event for long-held connections', async () => {
+    const leaks: string[] = [];
+    pool.on('leak', ({ connectionId }) => leaks.push(connectionId));
+
+    const { connection } = await pool.acquire();
+    jest.advanceTimersByTime(BASE_CONFIG.leakThresholdMs + BASE_CONFIG.maintenanceIntervalMs + 50);
+    await Promise.resolve(); // flush
+
+    expect(leaks).toContain(connection.id);
+    expect(pool.getMetrics().leaksDetected).toBeGreaterThanOrEqual(1);
+    pool.release(connection.id);
+  });
+
+  // ── Idle teardown ─────────────────────────────────────────────────────────
+
+  it('emits idle-teardown for long-idle connections', async () => {
+    const teardowns: string[] = [];
+    pool.on('idle-teardown', (id) => teardowns.push(id));
+
+    const { connection } = await pool.acquire();
+    pool.release(connection.id);
+    jest.advanceTimersByTime(BASE_CONFIG.idleTimeoutMs + BASE_CONFIG.maintenanceIntervalMs + 50);
+    await Promise.resolve();
+
+    expect(teardowns.length).toBeGreaterThan(0);
+  });
+
+  // ── Metrics ───────────────────────────────────────────────────────────────
+
+  it('tracks acquires and releases in metrics', async () => {
+    const { connection } = await pool.acquire();
+    pool.release(connection.id);
+    const m = pool.getMetrics();
+    expect(m.totalAcquires).toBe(1);
+    expect(m.totalReleases).toBe(1);
+    expect(m.peakActiveConnections).toBe(1);
+  });
+
+  it('tracks acquire timeouts in metrics', async () => {
+    for (let i = 0; i < BASE_CONFIG.poolSize; i++) await pool.acquire();
+    const p = pool.acquire();
+    jest.advanceTimersByTime(BASE_CONFIG.acquireTimeoutMs + 10);
+    await expect(p).rejects.toThrow();
+    expect(pool.getMetrics().acquireTimeouts).toBe(1);
+  });
+
+  it('prometheusMetrics returns valid format', async () => {
+    const prom = pool.prometheusMetrics();
+    expect(prom).toContain('subtrackr_es_pool_connections_total');
+    expect(prom).toContain('subtrackr_es_pool_acquire_timeouts_total');
+  });
+
+  // ── Tuning recommendations ────────────────────────────────────────────────
+
+  it('recommends pool increase at high utilisation', async () => {
+    // Acquire all connections to simulate peak
+    const acquired = await Promise.all(
+      Array.from({ length: BASE_CONFIG.poolSize }, () => pool.acquire()),
+    );
+    for (const a of acquired) pool.release(a.connection.id);
+
+    const recs = pool.getTuningRecommendations();
+    // At 100% peak utilisation should recommend increase
+    expect(recs.some((r) => r.includes('poolSize'))).toBe(true);
+  });
+
+  it('reports healthy when under-utilised', () => {
+    const recs = pool.getTuningRecommendations();
+    // Fresh pool with 0 acquires — low utilisation
+    expect(recs.length).toBeGreaterThan(0);
+  });
+
+  // ── Shutdown ──────────────────────────────────────────────────────────────
+
+  it('shutdown rejects pending waiters', async () => {
+    for (let i = 0; i < BASE_CONFIG.poolSize; i++) await pool.acquire();
+    const p = pool.acquire();
+    pool.shutdown();
+    await expect(p).rejects.toThrow('shutdown');
+  });
+});
+
+// ---------------------------------------------------------------------------
+// DnsCache
+// ---------------------------------------------------------------------------
+
+describe('DnsCache', () => {
+  it('caches resolved addresses', async () => {
+    const cache = new DnsCache();
+    await cache.resolve('es-primary', 1_000);
+    await cache.resolve('es-primary', 1_000);
+    expect(cache.stats().hits).toBe(1);
+    expect(cache.stats().lookups).toBe(1);
+  });
+
+  it('invalidates cache entry', async () => {
+    const cache = new DnsCache();
+    await cache.resolve('es-primary', 1_000);
+    cache.invalidate('es-primary');
+    await cache.resolve('es-primary', 1_000);
+    expect(cache.stats().lookups).toBe(2);
+    expect(cache.stats().hits).toBe(0);
+  });
+
+  it('re-resolves after TTL expires', async () => {
+    jest.useFakeTimers();
+    const cache = new DnsCache();
+    await cache.resolve('es-node', 100);
+    jest.advanceTimersByTime(200);
+    await cache.resolve('es-node', 100);
+    expect(cache.stats().lookups).toBe(2);
+    jest.useRealTimers();
+  });
+});

backend/elasticsearch/config.ts

@@ -3,8 +3,60 @@
  * In this mobile-first architecture the "cluster" is an in-process index
  * backed by AsyncStorage, mirroring a real ES setup so the service layer
  * can be swapped for a remote cluster without changing callers.
+ *
+ * Issue #986: Extended with connection pool settings.
  */
 
+// ---------------------------------------------------------------------------
+// Connection pool config (Issue #986)
+// ---------------------------------------------------------------------------
+
+export interface ElasticsearchPoolConfig {
+  /** Primary node host. Default: localhost */
+  primaryHost: string;
+  /** Primary node port. Default: 9200 */
+  primaryPort: number;
+  /** Optional read replicas for query routing. */
+  replicas?: { host: string; port: number }[];
+  /**
+   * Total connections in pool across primary + replicas.
+   * Recommended: (vCPUs * 2) for IO-bound ES workloads.
+   * Default: 10
+   */
+  poolSize: number;
+  /** Milliseconds to wait for a free connection. Default: 5000 */
+  acquireTimeoutMs: number;
+  /** Idle connection teardown threshold (ms). Default: 60_000 */
+  idleTimeoutMs: number;
+  /** Connection-held-too-long leak threshold (ms). Default: 30_000 */
+  leakThresholdMs: number;
+  /** DNS cache TTL (ms). Default: 30_000 */
+  dnsCacheTtlMs: number;
+  /** Maintenance sweep interval (ms). Default: 10_000 */
+  maintenanceIntervalMs: number;
+}
+
+export const DEFAULT_POOL_CONFIG: ElasticsearchPoolConfig = {
+  primaryHost: process.env['ES_PRIMARY_HOST'] ?? 'localhost',
+  primaryPort: Number(process.env['ES_PRIMARY_PORT'] ?? 9200),
+  replicas: process.env['ES_REPLICA_HOSTS']
+    ? process.env['ES_REPLICA_HOSTS'].split(',').map((h) => {
+        const [host, port] = h.split(':');
+        return { host: host ?? 'localhost', port: Number(port ?? 9200) };
+      })
+    : [],
+  poolSize: Number(process.env['ES_POOL_SIZE'] ?? 10),
+  acquireTimeoutMs: Number(process.env['ES_ACQUIRE_TIMEOUT_MS'] ?? 5_000),
+  idleTimeoutMs: Number(process.env['ES_IDLE_TIMEOUT_MS'] ?? 60_000),
+  leakThresholdMs: Number(process.env['ES_LEAK_THRESHOLD_MS'] ?? 30_000),
+  dnsCacheTtlMs: Number(process.env['ES_DNS_CACHE_TTL_MS'] ?? 30_000),
+  maintenanceIntervalMs: Number(process.env['ES_MAINTENANCE_INTERVAL_MS'] ?? 10_000),
+};
+
+// ---------------------------------------------------------------------------
+// Index / Search config
+// ---------------------------------------------------------------------------
+
 export interface ElasticsearchConfig {
   indexName: string;
   fuzzyMaxEdits: number;
@@ -14,6 +66,8 @@ export interface ElasticsearchConfig {
   analyticsEnabled: boolean;
   /** Analyzer locales used for multilingual tokenization */
   analyzerLocales: string[];
+  /** Connection pool settings (Issue #986) */
+  pool?: ElasticsearchPoolConfig;
 }
 
 export const DEFAULT_ES_CONFIG: ElasticsearchConfig = {
@@ -33,6 +87,7 @@ export const DEFAULT_ES_CONFIG: ElasticsearchConfig = {
   maxResults: 100,
   analyticsEnabled: true,
   analyzerLocales: ['en', 'fr', 'de', 'es'],
+  pool: DEFAULT_POOL_CONFIG,
 };
 
 export interface IndexMapping {