From 18844338754d5d974be5dc9e58f4a99b015aceea Mon Sep 17 00:00:00 2001 From: Kristiyan Ivanov Date: Thu, 15 Oct 2026 19:41:21 +0300 Subject: [PATCH 01/17] feat(key-analytics): scan every non-empty logical database and honor dbIndex Customer feedback (BetterDB 0.50.0): instances whose keys live outside db0 produced "No keys found in database, skipping analytics" on every scan, and the configured dbIndex was accepted but never applied to the connection. - Pass config.dbIndex through createAdapter() into the iovalkey `db` option, so connections (and the CLI client) actually SELECT the configured database. - collectKeyAnalytics now discovers non-empty logical databases via INFO keyspace and scans each one, visiting sibling databases through short-lived clients so the shared client never switches databases. Falls back to the old single-db scan when the keyspace section is unavailable, and keeps scanning the remaining databases when one fails. - The port returns one KeyAnalyticsResult per database; agent connections keep their single-db scan (normalized to the array shape) until the agent binary learns multi-db. - db is now a stored dimension: key_pattern_snapshots and hot_key_stats gain a `db` column (lazy-migrated, DEFAULT 0 backfill) across sqlite/turso and postgres; snapshots merge per (db, pattern) instead of folding databases together, hot/largest/composite keys carry the db of each key, and key identity (dedupe, composite ranking, rank deltas) includes the database. - Summary/patterns/trends/hot-keys/largest-keys/composite-keys endpoints accept a `db` filter; new GET /key-analytics/databases lists the databases with stored analytics. The Key Analytics page gets a database selector (shown only when more than one database has data) and per-row db badges. Covered by unit tests (multi-db discovery/scan, dbIndex pass-through, per-db snapshots and ranking, storage filters and migration) plus a live e2e against the valkey container reproducing the empty-db0 report. --- .../interfaces/database-port.interface.ts | 7 +- .../interfaces/storage-port.interface.ts | 4 + .../connection-registry.service.ts | 1 + .../database/adapters/unified.adapter.spec.ts | 178 +++++++++++++++++- .../src/database/adapters/unified.adapter.ts | 99 +++++++++- .../__tests__/key-analytics-multi-db.spec.ts | 144 ++++++++++++++ .../src/storage/adapters/base-sql.adapter.ts | 1 + .../src/storage/adapters/memory.adapter.ts | 5 + .../src/storage/adapters/postgres.adapter.ts | 110 ++++++++--- .../src/storage/adapters/sqlite.adapter.ts | 106 ++++++++--- .../test/key-analytics-multi-db.e2e-spec.ts | 107 +++++++++++ apps/web/src/api/keyAnalytics.ts | 22 ++- apps/web/src/pages/KeyAnalytics.tsx | 94 +++++++-- packages/shared/src/types/key-analytics.ts | 12 ++ proprietary/agent/agent-database-adapter.ts | 6 +- .../__tests__/composite-key-ranker.spec.ts | 26 +++ .../__tests__/key-analytics-topology.spec.ts | 43 ++++- .../__tests__/key-analytics.service.spec.ts | 5 +- .../__tests__/key-pattern-snapshots.spec.ts | 32 +++- .../key-analytics/composite-key-ranker.ts | 22 ++- .../key-analytics/key-analytics.controller.ts | 22 ++- .../key-analytics/key-analytics.service.ts | Bin 21539 -> 22720 bytes .../key-analytics/key-pattern-snapshots.ts | 8 +- 23 files changed, 967 insertions(+), 87 deletions(-) create mode 100644 apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts create mode 100644 apps/api/test/key-analytics-multi-db.e2e-spec.ts diff --git a/apps/api/src/common/interfaces/database-port.interface.ts b/apps/api/src/common/interfaces/database-port.interface.ts index 7b51294b2..a9a620c57 100644 --- a/apps/api/src/common/interfaces/database-port.interface.ts +++ b/apps/api/src/common/interfaces/database-port.interface.ts @@ -96,7 +96,12 @@ export interface DatabasePort { getConfigValues(pattern: string): Promise; getDbSize(): Promise; getLastSaveTime(): Promise; - collectKeyAnalytics(options: KeyAnalyticsOptions): Promise; + /** + * One result per scanned logical database. Adapters that can discover + * sibling databases (via `INFO keyspace`) scan every non-empty one; others + * return a single result for the connected database. + */ + collectKeyAnalytics(options: KeyAnalyticsOptions): Promise; getVectorIndexList(): Promise; getVectorIndexInfo(indexName: string): Promise; getHashFieldBuffer(key: string, field: string): Promise; diff --git a/apps/api/src/common/interfaces/storage-port.interface.ts b/apps/api/src/common/interfaces/storage-port.interface.ts index 5535fac1d..3ecf7d4ac 100644 --- a/apps/api/src/common/interfaces/storage-port.interface.ts +++ b/apps/api/src/common/interfaces/storage-port.interface.ts @@ -507,12 +507,14 @@ export interface StoragePort { startTime?: number, endTime?: number, connectionId?: string, + db?: number, ): Promise; getKeyPatternTrends( pattern: string, startTime: number, endTime: number, connectionId?: string, + db?: number, ): Promise< Array<{ timestamp: number; @@ -522,6 +524,8 @@ export interface StoragePort { }> >; pruneOldKeyPatternSnapshots(cutoffTimestamp: number, connectionId?: string): Promise; + /** Distinct logical database indexes present in stored key-pattern snapshots. */ + getKeyAnalyticsDatabases(connectionId?: string): Promise; // Hot Key Stats Methods - connectionId required for writes, optional filter for reads saveHotKeys(entries: HotKeyEntry[], connectionId: string): Promise; diff --git a/apps/api/src/connections/connection-registry.service.ts b/apps/api/src/connections/connection-registry.service.ts index a46b4ff08..609022a2c 100644 --- a/apps/api/src/connections/connection-registry.service.ts +++ b/apps/api/src/connections/connection-registry.service.ts @@ -356,6 +356,7 @@ export class ConnectionRegistry implements OnModuleInit, OnModuleDestroy { port: config.port, username: config.username || 'default', password: config.password || '', + dbIndex: config.dbIndex, connectionName, tls: config.tls, tlsServername: config.membership?.origin === 'auto' ? config.membership.hostname : undefined, diff --git a/apps/api/src/database/adapters/unified.adapter.spec.ts b/apps/api/src/database/adapters/unified.adapter.spec.ts index 28f301621..fa36d2944 100644 --- a/apps/api/src/database/adapters/unified.adapter.spec.ts +++ b/apps/api/src/database/adapters/unified.adapter.spec.ts @@ -179,8 +179,10 @@ describe('UnifiedDatabaseAdapter.collectKeyAnalytics — stale keys', () => { (adapter as unknown as { _client: unknown })._client = { dbsize: jest.fn().mockResolvedValue(keys.length), scan: jest.fn().mockResolvedValue(['0', keys]), + info: jest.fn().mockResolvedValue(`db0:keys=${keys.length},expires=0,avg_ttl=0`), pipeline, }; + (adapter as unknown as { config: unknown }).config = {}; return adapter; } @@ -193,10 +195,184 @@ describe('UnifiedDatabaseAdapter.collectKeyAnalytics — stale keys', () => { 'session:5': DAY, }); - const result = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + const [result] = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); const [sessions] = result.patterns; expect(sessions.totalIdleTime / sessions.count).toBeGreaterThan(DAY); expect(sessions.staleCount).toBe(1); }); }); + +describe('UnifiedDatabaseAdapter.collectKeyAnalytics — multiple logical databases', () => { + function scanChain() { + const chain = { + memory: () => chain, + object: () => chain, + ttl: () => chain, + type: () => chain, + strlen: () => chain, + llen: () => chain, + hlen: () => chain, + scard: () => chain, + zcard: () => chain, + xlen: () => chain, + exec: async () => [ + [null, 100], + [null, 10], + [null, null], + [null, -1], + [null, 'string'], + [null, 5], + ], + }; + return chain; + } + + function fakeScanClient(keys: string[]) { + return { + dbsize: jest.fn().mockResolvedValue(keys.length), + scan: jest.fn().mockResolvedValue(['0', keys]), + pipeline: scanChain, + connect: jest.fn().mockResolvedValue(undefined), + quit: jest.fn().mockResolvedValue('OK'), + disconnect: jest.fn(), + }; + } + + function makeMultiDbAdapter(options: { + dbIndex?: number; + keyspaceInfo: string; + ownKeys: string[]; + scratchKeysByDb?: Record; + }) { + const adapter = Object.create(UnifiedDatabaseAdapter.prototype) as UnifiedDatabaseAdapter; + const ownClient = { + ...fakeScanClient(options.ownKeys), + info: jest.fn().mockResolvedValue(options.keyspaceInfo), + }; + (adapter as unknown as { _client: unknown })._client = ownClient; + (adapter as unknown as { config: unknown }).config = { dbIndex: options.dbIndex }; + // Object.create bypasses the constructor, so the logger field is unset. + (adapter as unknown as { logger: unknown }).logger = { + debug: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + log: jest.fn(), + }; + const scratchClients: Record> = {}; + const createValkeyClient = jest.fn((_name: string, db?: number) => { + const client = fakeScanClient(options.scratchKeysByDb?.[db ?? 0] ?? []); + scratchClients[db ?? 0] = client; + return client; + }); + (adapter as unknown as { createValkeyClient: unknown }).createValkeyClient = + createValkeyClient; + return { adapter, ownClient, scratchClients, createValkeyClient }; + } + + it('scans every non-empty database from INFO keyspace, visiting siblings with a scratch client', async () => { + const { adapter, ownClient, scratchClients, createValkeyClient } = makeMultiDbAdapter({ + dbIndex: 0, + keyspaceInfo: + 'db0:keys=2,expires=0,avg_ttl=0\r\ndb2:keys=3,expires=1,avg_ttl=0\r\n', + ownKeys: ['a:1', 'a:2'], + scratchKeysByDb: { 2: ['b:1', 'b:2', 'b:3'] }, + }); + + const results = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + + expect(results.map((r) => r.db)).toEqual([0, 2]); + expect(results.map((r) => r.dbSize)).toEqual([2, 3]); + // db0 is the connection's own database: scanned on the shared client. + expect(ownClient.scan).toHaveBeenCalled(); + // db2 is visited through a short-lived client SELECTed onto db 2… + expect(createValkeyClient).toHaveBeenCalledTimes(1); + expect(createValkeyClient.mock.calls[0][1]).toBe(2); + expect(scratchClients[2].connect).toHaveBeenCalled(); + // …which is closed afterwards. + expect(scratchClients[2].quit).toHaveBeenCalled(); + }); + + it('finds keys outside the configured database even when that database is empty (BetterDB feedback)', async () => { + const { adapter, ownClient } = makeMultiDbAdapter({ + dbIndex: 0, + // db0 holds nothing, so INFO keyspace does not list it at all. + keyspaceInfo: 'db1:keys=3,expires=0,avg_ttl=0\r\n', + ownKeys: [], + scratchKeysByDb: { 1: ['user:1', 'user:2', 'user:3'] }, + }); + + const results = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + + // The old implementation DBSIZE'd db0, got 0 and gave up. Now db1 is found. + expect(results.map((r) => r.db)).toEqual([1]); + expect(results[0].dbSize).toBe(3); + expect(results[0].scanned).toBe(3); + expect(ownClient.scan).not.toHaveBeenCalled(); + }); + + it('falls back to scanning the connected database when INFO keyspace is unavailable', async () => { + const { adapter, ownClient, createValkeyClient } = makeMultiDbAdapter({ + dbIndex: 4, + keyspaceInfo: '', + ownKeys: ['k:1'], + }); + ownClient.info.mockRejectedValue(new Error('NOPERM')); + + const results = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + + expect(results).toHaveLength(1); + expect(results[0].db).toBe(4); + expect(ownClient.scan).toHaveBeenCalled(); + expect(createValkeyClient).not.toHaveBeenCalled(); + }); + + it('keeps scanning the remaining databases when one database scan fails', async () => { + const { adapter, scratchClients } = makeMultiDbAdapter({ + dbIndex: 0, + keyspaceInfo: + 'db0:keys=1,expires=0,avg_ttl=0\r\ndb1:keys=1,expires=0,avg_ttl=0\r\ndb2:keys=1,expires=0,avg_ttl=0\r\n', + ownKeys: ['a:1'], + scratchKeysByDb: { 1: ['b:1'], 2: ['c:1'] }, + }); + // db1's scratch connect blows up; db0 and db2 must still come back. + const original = (adapter as unknown as { createValkeyClient: jest.Mock }) + .createValkeyClient; + original.mockImplementationOnce((_name: string, _db?: number) => { + const broken = fakeScanClient([]); + broken.connect.mockRejectedValue(new Error('connection refused')); + return broken; + }); + + const results = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + + expect(results.map((r) => r.db)).toEqual([0, 2]); + expect(scratchClients[2].quit).toHaveBeenCalled(); + }); +}); + +describe('UnifiedDatabaseAdapter — configured dbIndex', () => { + it('passes dbIndex through to the Valkey client options (BetterDB feedback)', () => { + const adapter = new UnifiedDatabaseAdapter({ + host: 'localhost', + port: 6379, + username: 'default', + password: '', + dbIndex: 3, + }); + // lazyConnect: no socket is opened; only the options are inspected. + const client = adapter.getClient() as unknown as { options: { db?: number } }; + expect(client.options.db).toBe(3); + }); + + it('defaults to db0 when no dbIndex is configured', () => { + const adapter = new UnifiedDatabaseAdapter({ + host: 'localhost', + port: 6379, + username: 'default', + password: '', + }); + const client = adapter.getClient() as unknown as { options: { db?: number } }; + expect(client.options.db).toBe(0); + }); +}); diff --git a/apps/api/src/database/adapters/unified.adapter.ts b/apps/api/src/database/adapters/unified.adapter.ts index 531c89536..526446b67 100644 --- a/apps/api/src/database/adapters/unified.adapter.ts +++ b/apps/api/src/database/adapters/unified.adapter.ts @@ -60,6 +60,8 @@ export interface UnifiedDatabaseAdapterConfig { username: string; password: string; connectionName?: string; + /** Logical database index the clients SELECT on connect (defaults to 0). */ + dbIndex?: number; tls?: boolean; /** Hostname the TLS certificate is verified against when `host` is a bare IP. */ tlsServername?: string; @@ -124,12 +126,16 @@ export class UnifiedDatabaseAdapter implements DatabasePort { private connectHost: string; private connectPort: number; - private createValkeyClient(connectionName: string): Valkey { + private createValkeyClient(connectionName: string, db?: number): Valkey { return new Valkey({ host: this.connectHost, port: this.connectPort, username: this.config.username, password: this.config.password, + // Logical database the client SELECTs on connect. Defaults to the + // connection's configured dbIndex; key-analytics scans override it to + // visit sibling databases discovered via INFO keyspace. + db: db ?? this.config.dbIndex ?? 0, lazyConnect: true, enableOfflineQueue: false, connectionName, @@ -805,10 +811,90 @@ export class UnifiedDatabaseAdapter implements DatabasePort { return await this.client.lastsave(); } - async collectKeyAnalytics(options: KeyAnalyticsOptions): Promise { - const dbSize = await this.client.dbsize(); + /** + * Logical databases that currently hold keys, from `INFO keyspace` + * (`db:keys=...` lines — the section only lists non-empty databases). + * Returns [] when the section is empty or unavailable (e.g. cluster mode + * proxies that hide it), in which case the caller falls back to scanning + * the connection's own database. + */ + private async listKeyspaceDatabases(): Promise { + try { + const raw = await this.client.info('keyspace'); + const dbs: number[] = []; + for (const line of raw.split('\n')) { + const match = /^db(\d+):keys=(\d+)/.exec(line.trim()); + if (match && parseInt(match[2], 10) > 0) { + dbs.push(parseInt(match[1], 10)); + } + } + return dbs.sort((a, b) => a - b); + } catch (error) { + this.logger.debug( + `Could not list keyspace databases: ${error instanceof Error ? error.message : error}`, + ); + return []; + } + } + + /** + * Scan every non-empty logical database of this instance, one result per + * database. The databases are discovered via `INFO keyspace` rather than + * `DBSIZE` on the connected database alone, so an instance whose keys live + * outside the configured dbIndex (e.g. everything in db1, nothing in db0) + * is still analyzed. Databases other than the connection's own are visited + * through a short-lived client SELECTed onto that database — the shared + * client never switches databases, so concurrent monitoring is unaffected. + */ + async collectKeyAnalytics(options: KeyAnalyticsOptions): Promise { + const ownDb = this.config.dbIndex ?? 0; + const databases = await this.listKeyspaceDatabases(); + if (databases.length === 0) { + // Empty instance or no keyspace section: preserve the single-db + // behaviour (the dbSize === 0 early-return keeps this cheap). + return [await this.scanDatabase(this.client, ownDb, options)]; + } + + const results: KeyAnalyticsResult[] = []; + const failures: unknown[] = []; + // Sequential on purpose: SCAN + per-key pipelines are heavy, and running + // the databases one after another keeps the load on the server no higher + // than today's single-db scan. + for (const db of databases) { + if (db === ownDb) { + results.push(await this.scanDatabase(this.client, db, options)); + continue; + } + const scratch = this.createValkeyClient( + `${this.config.connectionName ?? 'BetterDB-Monitor'}-keyscan-db${db}`, + db, + ); + try { + await scratch.connect(); + results.push(await this.scanDatabase(scratch, db, options)); + } catch (error) { + failures.push(error); + this.logger.warn( + `Key analytics scan failed on db${db}: ${error instanceof Error ? error.message : error}`, + ); + } finally { + await closeClient(scratch); + } + } + if (results.length === 0 && failures.length > 0) { + throw failures[0]; + } + return results; + } + + private async scanDatabase( + client: Valkey, + db: number, + options: KeyAnalyticsOptions, + ): Promise { + const dbSize = await client.dbsize(); if (dbSize === 0) { - return { dbSize: 0, scanned: 0, patterns: [] }; + return { dbSize: 0, scanned: 0, patterns: [], db }; } const patternsMap = new Map(); @@ -817,7 +903,7 @@ export class UnifiedDatabaseAdapter implements DatabasePort { let scanned = 0; do { - const [newCursor, keys] = await this.client.scan(cursor, 'COUNT', options.scanBatchSize); + const [newCursor, keys] = await client.scan(cursor, 'COUNT', options.scanBatchSize); cursor = newCursor; for (const key of keys) { @@ -841,7 +927,7 @@ export class UnifiedDatabaseAdapter implements DatabasePort { }; try { - const pipeline = this.client.pipeline(); + const pipeline = client.pipeline(); pipeline.memory('USAGE', key); pipeline.object('IDLETIME', key); pipeline.object('FREQ', key); @@ -968,6 +1054,7 @@ export class UnifiedDatabaseAdapter implements DatabasePort { scanned, patterns: Array.from(patternsMap.values()), keyDetails: pruneKeyDetails(keyDetails), + db, }; } diff --git a/apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts b/apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts new file mode 100644 index 000000000..ce554f653 --- /dev/null +++ b/apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts @@ -0,0 +1,144 @@ +import { unlinkSync } from 'fs'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import type { HotKeyEntry, KeyPatternSnapshot } from '@betterdb/shared'; +import { SqliteAdapter } from '../sqlite.adapter'; + +const CONN = 'conn-a'; + +function snapshot(over: Partial & { id: string }): KeyPatternSnapshot { + return { + timestamp: 1_000, + pattern: 'user:*', + keyCount: 10, + sampledKeyCount: 10, + keysWithTtl: 0, + keysExpiringSoon: 0, + totalMemoryBytes: 100, + avgMemoryBytes: 10, + maxMemoryBytes: 20, + ...over, + }; +} + +function hotKey(over: Partial & { id: string }): HotKeyEntry { + return { + keyName: 'user:1', + connectionId: CONN, + capturedAt: 1_000, + signalType: 'lfu', + rank: 1, + ...over, + }; +} + +describe('SqliteAdapter key analytics per logical database', () => { + let storage: SqliteAdapter; + let path: string; + + beforeEach(async () => { + path = join(tmpdir(), `key-analytics-multi-db-${Date.now()}-${Math.random()}.db`); + storage = new SqliteAdapter({ filepath: path }); + await storage.initialize(); + }); + + afterEach(async () => { + await storage.close(); + try { + unlinkSync(path); + } catch { + return; + } + }); + + it('stores and returns the db of each pattern snapshot, defaulting to 0', async () => { + await storage.saveKeyPatternSnapshots( + [snapshot({ id: 's0' }), snapshot({ id: 's1', db: 1 })], + CONN, + ); + + const rows = await storage.getKeyPatternSnapshots({ connectionId: CONN }); + + expect(rows.map((row) => [row.id, row.db]).sort()).toEqual([ + ['s0', 0], + ['s1', 1], + ]); + }); + + it('filters pattern snapshots by db', async () => { + await storage.saveKeyPatternSnapshots( + [snapshot({ id: 's0', db: 0 }), snapshot({ id: 's1', db: 1 }), snapshot({ id: 's2', db: 1 , pattern: 'job:*' })], + CONN, + ); + + const db1 = await storage.getKeyPatternSnapshots({ connectionId: CONN, db: 1 }); + + expect(db1.map((row) => row.id).sort()).toEqual(['s1', 's2']); + }); + + it('aggregates the summary across databases per pattern, and scopes it when filtered', async () => { + await storage.saveKeyPatternSnapshots( + [ + snapshot({ id: 's0', db: 0, keyCount: 10, totalMemoryBytes: 100 }), + snapshot({ id: 's1', db: 1, keyCount: 30, totalMemoryBytes: 300 }), + ], + CONN, + ); + + const all = await storage.getKeyAnalyticsSummary(undefined, undefined, CONN); + expect(all?.totalKeys).toBe(40); + expect(all?.totalPatterns).toBe(1); + expect(all?.byPattern['user:*'].keyCount).toBe(40); + expect(all?.byPattern['user:*'].memoryBytes).toBe(400); + + const db1 = await storage.getKeyAnalyticsSummary(undefined, undefined, CONN, 1); + expect(db1?.totalKeys).toBe(30); + expect(db1?.byPattern['user:*'].keyCount).toBe(30); + }); + + it('aggregates trends per timestamp across databases and respects the db filter', async () => { + await storage.saveKeyPatternSnapshots( + [ + snapshot({ id: 's0', db: 0, keyCount: 10 }), + snapshot({ id: 's1', db: 1, keyCount: 30 }), + ], + CONN, + ); + + const all = await storage.getKeyPatternTrends('user:*', 0, 2_000, CONN); + expect(all).toEqual([{ timestamp: 1_000, keyCount: 40, memoryBytes: 200, staleCount: 0 }]); + + const db1 = await storage.getKeyPatternTrends('user:*', 0, 2_000, CONN, 1); + expect(db1).toEqual([{ timestamp: 1_000, keyCount: 30, memoryBytes: 100, staleCount: 0 }]); + }); + + it('lists the distinct databases that have analytics data', async () => { + await storage.saveKeyPatternSnapshots( + [snapshot({ id: 's0', db: 0 }), snapshot({ id: 's1', db: 2 }), snapshot({ id: 's2', db: 2, pattern: 'job:*' })], + CONN, + ); + await storage.saveKeyPatternSnapshots([snapshot({ id: 'other', db: 5 })], 'conn-b'); + + expect(await storage.getKeyAnalyticsDatabases(CONN)).toEqual([0, 2]); + expect(await storage.getKeyAnalyticsDatabases()).toEqual([0, 2, 5]); + }); + + it('stores and filters hot keys by db, including the latest-snapshot pin', async () => { + await storage.saveHotKeys( + [ + hotKey({ id: 'h0', db: 0, keyName: 'user:1' }), + hotKey({ id: 'h1', db: 1, keyName: 'user:1' }), + hotKey({ id: 'h2', db: 1, keyName: 'user:2', capturedAt: 2_000 }), + ], + CONN, + ); + + const db1 = await storage.getHotKeys({ connectionId: CONN, db: 1 }); + expect(db1.map((entry) => entry.id).sort()).toEqual(['h1', 'h2']); + expect(db1.every((entry) => entry.db === 1)).toBe(true); + + // latest pin with a db filter looks at that db's newest capture only. + const latestDb0 = await storage.getHotKeys({ connectionId: CONN, db: 0, latest: true }); + expect(latestDb0.map((entry) => entry.id)).toEqual(['h0']); + }); +}); diff --git a/apps/api/src/storage/adapters/base-sql.adapter.ts b/apps/api/src/storage/adapters/base-sql.adapter.ts index b8d4cd82f..ff94c9bc9 100644 --- a/apps/api/src/storage/adapters/base-sql.adapter.ts +++ b/apps/api/src/storage/adapters/base-sql.adapter.ts @@ -253,6 +253,7 @@ export class RowMappers { minTtlSeconds: row.min_ttl_seconds, maxTtlSeconds: row.max_ttl_seconds, connectionId: row.connection_id, + db: row.db ?? 0, }; } diff --git a/apps/api/src/storage/adapters/memory.adapter.ts b/apps/api/src/storage/adapters/memory.adapter.ts index df935fd08..13d11c8f7 100644 --- a/apps/api/src/storage/adapters/memory.adapter.ts +++ b/apps/api/src/storage/adapters/memory.adapter.ts @@ -800,6 +800,11 @@ export class MemoryAdapter implements StoragePort, RawDatabaseHandleProvider { return 0; } + async getKeyAnalyticsDatabases(_connectionId?: string): Promise { + // Nothing is ever stored here, so no databases have analytics data. + return []; + } + async saveHotKeys(_entries: HotKeyEntry[], _connectionId: string): Promise { throw new Error('Hot key stats not supported in memory adapter'); } diff --git a/apps/api/src/storage/adapters/postgres.adapter.ts b/apps/api/src/storage/adapters/postgres.adapter.ts index c7a8a4a79..e89627e2d 100644 --- a/apps/api/src/storage/adapters/postgres.adapter.ts +++ b/apps/api/src/storage/adapters/postgres.adapter.ts @@ -1411,8 +1411,10 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { min_ttl_seconds INTEGER, max_ttl_seconds INTEGER, connection_id TEXT NOT NULL DEFAULT 'env-default', + db INTEGER NOT NULL DEFAULT 0, created_at TIMESTAMPTZ DEFAULT NOW() ); + ALTER TABLE key_pattern_snapshots ADD COLUMN IF NOT EXISTS db INTEGER NOT NULL DEFAULT 0; CREATE INDEX IF NOT EXISTS idx_kps_timestamp ON key_pattern_snapshots(timestamp DESC); CREATE INDEX IF NOT EXISTS idx_kps_pattern ON key_pattern_snapshots(pattern, timestamp DESC); @@ -1431,10 +1433,12 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { cardinality BIGINT, key_type TEXT, ttl INTEGER, - rank INTEGER NOT NULL + rank INTEGER NOT NULL, + db INTEGER NOT NULL DEFAULT 0 ); ALTER TABLE hot_key_stats ADD COLUMN IF NOT EXISTS cardinality BIGINT; ALTER TABLE hot_key_stats ADD COLUMN IF NOT EXISTS key_type TEXT; + ALTER TABLE hot_key_stats ADD COLUMN IF NOT EXISTS db INTEGER NOT NULL DEFAULT 0; CREATE INDEX IF NOT EXISTS idx_hot_key_captured_at_global ON hot_key_stats(captured_at); @@ -2680,7 +2684,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { for (const snapshot of snapshots) { values.push( - `($${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++})`, + `($${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++})`, ); params.push( snapshot.id, @@ -2702,6 +2706,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { snapshot.minTtlSeconds ?? null, snapshot.maxTtlSeconds ?? null, connectionId, + snapshot.db ?? 0, ); } @@ -2712,7 +2717,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { keys_with_ttl, keys_expiring_soon, total_memory_bytes, avg_memory_bytes, max_memory_bytes, avg_access_frequency, hot_key_count, cold_key_count, avg_idle_time_seconds, - stale_key_count, avg_ttl_seconds, min_ttl_seconds, max_ttl_seconds, connection_id + stale_key_count, avg_ttl_seconds, min_ttl_seconds, max_ttl_seconds, connection_id, db ) VALUES ${values.join(', ')} `, params, @@ -2746,6 +2751,10 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { conditions.push(`pattern = $${paramIndex++}`); params.push(options.pattern); } + if (options.db !== undefined) { + conditions.push(`db = $${paramIndex++}`); + params.push(options.db); + } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; const limit = options.limit ?? 100; @@ -2768,6 +2777,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { startTime?: number, endTime?: number, connectionId?: string, + db?: number, ): Promise { if (!this.pool) throw new Error('Database not initialized'); @@ -2787,6 +2797,10 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { conditions.push(`timestamp <= $${paramIndex++}`); params.push(endTime); } + if (db !== undefined) { + conditions.push(`db = $${paramIndex++}`); + params.push(db); + } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; @@ -2802,9 +2816,9 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { if (latestSnapshotsResult.rows.length === 0) return null; - const patternConditions = latestSnapshotsResult.rows - .map(() => '(pattern = ? AND timestamp = ?)') - .join(' OR '); + // A (pattern, timestamp) pair can match several rows — one per logical + // database — so the db filter must be re-applied here, and the unfiltered + // per-pattern breakdown below aggregates across the per-database rows. const patternParams: any[] = []; for (const row of latestSnapshotsResult.rows) { patternParams.push(row.pattern, row.latest_timestamp); @@ -2814,6 +2828,11 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { const patternPlaceholders = latestSnapshotsResult.rows .map(() => `(pattern = $${pIdx++} AND timestamp = $${pIdx++})`) .join(' OR '); + const patternWhere = + db !== undefined ? `(${patternPlaceholders}) AND db = $${pIdx++}` : patternPlaceholders; + if (db !== undefined) { + patternParams.push(db); + } const summaryResult = await this.pool.query( ` @@ -2826,7 +2845,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { SUM(cold_key_count) as cold_key_count, SUM(keys_expiring_soon) as keys_expiring_soon FROM key_pattern_snapshots - WHERE ${patternPlaceholders} + WHERE ${patternWhere} `, patternParams, ); @@ -2835,22 +2854,33 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { const patternRowsResult = await this.pool.query( ` - SELECT pattern, key_count, total_memory_bytes, avg_memory_bytes, stale_key_count, hot_key_count, cold_key_count + SELECT + pattern, + SUM(key_count) as key_count, + SUM(total_memory_bytes) as total_memory_bytes, + SUM(stale_key_count) as stale_key_count, + SUM(hot_key_count) as hot_key_count, + SUM(cold_key_count) as cold_key_count, + SUM(sampled_key_count) as sampled_key_count, + SUM(avg_memory_bytes * sampled_key_count) as weighted_avg_memory FROM key_pattern_snapshots - WHERE ${patternPlaceholders} + WHERE ${patternWhere} + GROUP BY pattern `, patternParams, ); const byPattern: Record = {}; for (const row of patternRowsResult.rows) { + const sampledKeyCount = parseInt(row.sampled_key_count) || 0; byPattern[row.pattern] = { - keyCount: row.key_count, - memoryBytes: parseInt(row.total_memory_bytes), - avgMemoryBytes: row.avg_memory_bytes, - staleCount: row.stale_key_count ?? null, - hotCount: row.hot_key_count ?? 0, - coldCount: row.cold_key_count ?? 0, + keyCount: parseInt(row.key_count) || 0, + memoryBytes: parseInt(row.total_memory_bytes) || 0, + avgMemoryBytes: + sampledKeyCount > 0 ? Math.round(parseInt(row.weighted_avg_memory) / sampledKeyCount) : 0, + staleCount: row.stale_key_count == null ? null : parseInt(row.stale_key_count), + hotCount: row.hot_key_count == null ? 0 : parseInt(row.hot_key_count), + coldCount: row.cold_key_count == null ? 0 : parseInt(row.cold_key_count), }; } @@ -2888,6 +2918,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { startTime: number, endTime: number, connectionId?: string, + db?: number, ): Promise< Array<{ timestamp: number; @@ -2902,15 +2933,26 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { const params: any[] = [pattern, startTime, endTime]; if (connectionId) { - conditions.push('connection_id = $4'); params.push(connectionId); + conditions.push(`connection_id = $${params.length}`); + } + if (db !== undefined) { + params.push(db); + conditions.push(`db = $${params.length}`); } + // One collection writes one row per logical database; aggregate them per + // timestamp so an unfiltered trend stays a single series. const result = await this.pool.query( ` - SELECT timestamp, key_count, total_memory_bytes, stale_key_count + SELECT + timestamp, + SUM(key_count) as key_count, + SUM(total_memory_bytes) as total_memory_bytes, + SUM(stale_key_count) as stale_key_count FROM key_pattern_snapshots WHERE ${conditions.join(' AND ')} + GROUP BY timestamp ORDER BY timestamp ASC `, params, @@ -2918,9 +2960,9 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { return result.rows.map((row) => ({ timestamp: parseInt(row.timestamp), - keyCount: row.key_count, - memoryBytes: parseInt(row.total_memory_bytes), - staleCount: row.stale_key_count ?? 0, + keyCount: parseInt(row.key_count) || 0, + memoryBytes: parseInt(row.total_memory_bytes) || 0, + staleCount: row.stale_key_count == null ? 0 : parseInt(row.stale_key_count), })); } @@ -2949,7 +2991,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { for (const entry of entries) { values.push( - `($${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++})`, + `($${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++})`, ); params.push( entry.id, @@ -2964,6 +3006,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { entry.keyType ?? null, entry.ttl ?? null, entry.rank, + entry.db ?? 0, ); } @@ -2971,7 +3014,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { ` INSERT INTO hot_key_stats ( id, key_name, connection_id, captured_at, signal_type, - freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank + freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank, db ) VALUES ${values.join(', ')} `, params, @@ -3005,6 +3048,10 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { conditions.push(`captured_at <= $${paramIndex++}`); params.push(options.endTime); } + if (options.db !== undefined) { + conditions.push(`db = $${paramIndex++}`); + params.push(options.db); + } if (options.latest || options.oldest) { const agg = options.latest ? 'MAX' : 'MIN'; const subSignalPlaceholders = signalTypes.map(() => `$${paramIndex++}`).join(', '); @@ -3022,6 +3069,10 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { subConditions.push(`captured_at <= $${paramIndex++}`); params.push(options.endTime); } + if (options.db !== undefined) { + subConditions.push(`db = $${paramIndex++}`); + params.push(options.db); + } const subWhere = subConditions.length > 0 ? `WHERE ${subConditions.join(' AND ')}` : ''; conditions.push(`captured_at = (SELECT ${agg}(captured_at) FROM hot_key_stats ${subWhere})`); } @@ -3033,7 +3084,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { const result = await this.pool.query( ` SELECT id, key_name, connection_id, captured_at, signal_type, - freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank + freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank, db FROM hot_key_stats ${whereClause} ORDER BY captured_at DESC, rank ASC @@ -3055,9 +3106,22 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { keyType: row.key_type ?? undefined, ttl: row.ttl ?? undefined, rank: row.rank, + db: row.db ?? 0, })); } + async getKeyAnalyticsDatabases(connectionId?: string): Promise { + if (!this.pool) throw new Error('Database not initialized'); + + const where = connectionId ? 'WHERE connection_id = $1' : ''; + const params = connectionId ? [connectionId] : []; + const result = await this.pool.query( + `SELECT DISTINCT db FROM key_pattern_snapshots ${where} ORDER BY db ASC`, + params, + ); + return result.rows.map((row: { db: number | null }) => row.db ?? 0); + } + async pruneOldHotKeys(cutoffTimestamp: number, connectionId?: string): Promise { if (!this.pool) throw new Error('Database not initialized'); diff --git a/apps/api/src/storage/adapters/sqlite.adapter.ts b/apps/api/src/storage/adapters/sqlite.adapter.ts index a359b101e..1465ca37e 100644 --- a/apps/api/src/storage/adapters/sqlite.adapter.ts +++ b/apps/api/src/storage/adapters/sqlite.adapter.ts @@ -585,7 +585,8 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { } } - // Lazy ALTERs for hot_key_stats columns added to support largest-key ranking. + // Lazy ALTERs for hot_key_stats columns added to support largest-key ranking + // and the per-logical-database dimension. const hotKeyInfo = this.db.prepare('PRAGMA table_info(hot_key_stats)').all() as { name: string; }[]; @@ -593,6 +594,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { const hotKeyMigrations = [ { name: 'cardinality', type: 'INTEGER' }, { name: 'key_type', type: 'TEXT' }, + { name: 'db', type: 'INTEGER NOT NULL DEFAULT 0' }, ]; for (const col of hotKeyMigrations) { if (!hotKeyColumns.has(col.name)) { @@ -600,6 +602,16 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { } } + // Lazy ALTER for the key_pattern_snapshots logical-database dimension. + // Pre-existing rows were all collected from the connected database, which + // was db0 before dbIndex support — the DEFAULT 0 backfill is accurate. + const kpsInfo = this.db.prepare('PRAGMA table_info(key_pattern_snapshots)').all() as { + name: string; + }[]; + if (!kpsInfo.some((col) => col.name === 'db')) { + this.db.exec('ALTER TABLE key_pattern_snapshots ADD COLUMN db INTEGER NOT NULL DEFAULT 0'); + } + // Migrate connection_id to all data tables for multi-database support this.migrateConnectionId(); } @@ -1523,6 +1535,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { min_ttl_seconds INTEGER, max_ttl_seconds INTEGER, connection_id TEXT NOT NULL DEFAULT 'env-default', + db INTEGER NOT NULL DEFAULT 0, created_at INTEGER DEFAULT (strftime('%s', 'now') * 1000) ); @@ -1543,7 +1556,8 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { cardinality INTEGER, key_type TEXT, ttl INTEGER, - rank INTEGER NOT NULL + rank INTEGER NOT NULL, + db INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_hks_connection_captured @@ -2580,8 +2594,8 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { keys_with_ttl, keys_expiring_soon, total_memory_bytes, avg_memory_bytes, max_memory_bytes, avg_access_frequency, hot_key_count, cold_key_count, avg_idle_time_seconds, - stale_key_count, avg_ttl_seconds, min_ttl_seconds, max_ttl_seconds, connection_id - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + stale_key_count, avg_ttl_seconds, min_ttl_seconds, max_ttl_seconds, connection_id, db + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `); const insertMany = this.db.transaction((snapshots: KeyPatternSnapshot[], connId: string) => { @@ -2606,6 +2620,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { snapshot.minTtlSeconds ?? null, snapshot.maxTtlSeconds ?? null, connId, + snapshot.db ?? 0, ); } }); @@ -2638,6 +2653,10 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { conditions.push('pattern = ?'); params.push(options.pattern); } + if (options.db !== undefined) { + conditions.push('db = ?'); + params.push(options.db); + } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; const limit = options.limit ?? 100; @@ -2661,6 +2680,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { startTime?: number, endTime?: number, connectionId?: string, + db?: number, ): Promise { if (!this.db) throw new Error('Database not initialized'); @@ -2679,6 +2699,10 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { conditions.push('timestamp <= ?'); params.push(endTime); } + if (db !== undefined) { + conditions.push('db = ?'); + params.push(db); + } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; @@ -2701,7 +2725,10 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { return null; } - // Build aggregation query for latest snapshots only + // Build aggregation query for latest snapshots only. A (pattern, timestamp) + // pair can match several rows — one per logical database — so the db filter + // must be re-applied here, and the unfiltered per-pattern breakdown must + // aggregate across the per-database rows rather than let the last row win. const patternConditions = latestSnapshots .map(() => '(pattern = ? AND timestamp = ?)') .join(' OR '); @@ -2709,6 +2736,10 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { for (const snapshot of latestSnapshots) { patternParams.push(snapshot.pattern, snapshot.latest_timestamp); } + const patternWhere = db !== undefined ? `(${patternConditions}) AND db = ?` : patternConditions; + if (db !== undefined) { + patternParams.push(db); + } const summaryQuery = ` SELECT @@ -2720,23 +2751,26 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { SUM(cold_key_count) as cold_key_count, SUM(keys_expiring_soon) as keys_expiring_soon FROM key_pattern_snapshots - WHERE ${patternConditions} + WHERE ${patternWhere} `; const summary = this.db.prepare(summaryQuery).get(...patternParams) as any; - // Get per-pattern breakdown + // Get per-pattern breakdown, aggregated across the per-database rows of + // each pattern (avg weighted by key count via the summed totals). const byPatternQuery = ` SELECT pattern, - key_count, - total_memory_bytes, - avg_memory_bytes, - stale_key_count, - hot_key_count, - cold_key_count + SUM(key_count) as key_count, + SUM(total_memory_bytes) as total_memory_bytes, + SUM(stale_key_count) as stale_key_count, + SUM(hot_key_count) as hot_key_count, + SUM(cold_key_count) as cold_key_count, + SUM(sampled_key_count) as sampled_key_count, + SUM(avg_memory_bytes * sampled_key_count) as weighted_avg_memory FROM key_pattern_snapshots - WHERE ${patternConditions} + WHERE ${patternWhere} + GROUP BY pattern `; const patternRows = this.db.prepare(byPatternQuery).all(...patternParams) as any[]; @@ -2746,7 +2780,8 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { byPattern[row.pattern] = { keyCount: row.key_count, memoryBytes: row.total_memory_bytes, - avgMemoryBytes: row.avg_memory_bytes, + avgMemoryBytes: + row.sampled_key_count > 0 ? Math.round(row.weighted_avg_memory / row.sampled_key_count) : 0, staleCount: row.stale_key_count ?? null, hotCount: row.hot_key_count ?? 0, coldCount: row.cold_key_count ?? 0, @@ -2783,6 +2818,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { startTime: number, endTime: number, connectionId?: string, + db?: number, ): Promise< Array<{ timestamp: number; @@ -2800,15 +2836,22 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { conditions.push('connection_id = ?'); params.push(connectionId); } + if (db !== undefined) { + conditions.push('db = ?'); + params.push(db); + } + // One collection writes one row per logical database; aggregate them per + // timestamp so an unfiltered trend stays a single series. const query = ` SELECT timestamp, - key_count, - total_memory_bytes, - stale_key_count + SUM(key_count) as key_count, + SUM(total_memory_bytes) as total_memory_bytes, + SUM(stale_key_count) as stale_key_count FROM key_pattern_snapshots WHERE ${conditions.join(' AND ')} + GROUP BY timestamp ORDER BY timestamp ASC `; @@ -2844,8 +2887,8 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { const stmt = this.db.prepare(` INSERT INTO hot_key_stats ( id, key_name, connection_id, captured_at, signal_type, - freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank, db + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `); const insertMany = this.db.transaction((entries: HotKeyEntry[], connId: string) => { @@ -2863,6 +2906,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { entry.keyType ?? null, entry.ttl ?? null, entry.rank, + entry.db ?? 0, ); } }); @@ -2895,6 +2939,10 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { conditions.push('captured_at <= ?'); params.push(options.endTime); } + if (options.db !== undefined) { + conditions.push('db = ?'); + params.push(options.db); + } if (options.latest || options.oldest) { const agg = options.latest ? 'MAX' : 'MIN'; const subConditions: string[] = [`signal_type IN (${signalPlaceholders})`]; @@ -2911,6 +2959,10 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { subConditions.push('captured_at <= ?'); subParams.push(options.endTime); } + if (options.db !== undefined) { + subConditions.push('db = ?'); + subParams.push(options.db); + } const subWhere = subConditions.length > 0 ? `WHERE ${subConditions.join(' AND ')}` : ''; conditions.push(`captured_at = (SELECT ${agg}(captured_at) FROM hot_key_stats ${subWhere})`); params.push(...subParams); @@ -2925,7 +2977,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { .prepare( ` SELECT id, key_name, connection_id, captured_at, signal_type, - freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank + freq_score, idle_seconds, memory_bytes, cardinality, key_type, ttl, rank, db FROM hot_key_stats ${whereClause} ORDER BY captured_at DESC, rank ASC @@ -2947,9 +2999,21 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { keyType: row.key_type ?? undefined, ttl: row.ttl ?? undefined, rank: row.rank, + db: row.db ?? 0, })); } + async getKeyAnalyticsDatabases(connectionId?: string): Promise { + if (!this.db) throw new Error('Database not initialized'); + + const where = connectionId ? 'WHERE connection_id = ?' : ''; + const params = connectionId ? [connectionId] : []; + const rows = this.db + .prepare(`SELECT DISTINCT db FROM key_pattern_snapshots ${where} ORDER BY db ASC`) + .all(...params) as Array<{ db: number | null }>; + return rows.map((row) => row.db ?? 0); + } + async pruneOldHotKeys(cutoffTimestamp: number, connectionId?: string): Promise { if (!this.db) throw new Error('Database not initialized'); diff --git a/apps/api/test/key-analytics-multi-db.e2e-spec.ts b/apps/api/test/key-analytics-multi-db.e2e-spec.ts new file mode 100644 index 000000000..37432df62 --- /dev/null +++ b/apps/api/test/key-analytics-multi-db.e2e-spec.ts @@ -0,0 +1,107 @@ +import Valkey from 'iovalkey'; +import { UnifiedDatabaseAdapter } from '../src/database/adapters/unified.adapter'; + +/** + * Live multi-logical-database key analytics (BetterDB feedback): an instance + * whose keys live outside the connected database (e.g. everything in db13, + * nothing in db0) must still be discovered and scanned, with each result + * tagged by its database index. + * + * Uses the shared betterdb-test-valkey container. The seeded databases are + * high indexes no other e2e suite touches; db0 is deliberately left alone, so + * assertions check inclusion rather than exact equality. + */ + +const HOST = 'localhost'; +const PORT = 6390; +const PASSWORD = 'devpassword'; + +const DB_A = 13; +const DB_B = 14; + +describe('Key analytics across logical databases (E2E)', () => { + let seeder: Valkey; + + beforeAll(async () => { + seeder = new Valkey({ host: HOST, port: PORT, password: PASSWORD, db: DB_A }); + await seeder.flushdb(); + const pipelineA = seeder.pipeline(); + for (let i = 0; i < 25; i++) { + pipelineA.set(`e2emulti:user:${i}`, `value-${i}`); + } + await pipelineA.exec(); + + await seeder.select(DB_B); + await seeder.flushdb(); + const pipelineB = seeder.pipeline(); + for (let i = 0; i < 10; i++) { + pipelineB.set(`e2emulti:job:${i}`, `value-${i}`); + } + await pipelineB.exec(); + await seeder.select(DB_A); + }); + + afterAll(async () => { + try { + await seeder.flushdb(); + await seeder.select(DB_B); + await seeder.flushdb(); + } finally { + await seeder.quit(); + } + }); + + it('honors the configured dbIndex when connecting', async () => { + const adapter = new UnifiedDatabaseAdapter({ + host: HOST, + port: PORT, + username: 'default', + password: PASSWORD, + dbIndex: DB_A, + }); + try { + await adapter.connect(); + // DBSIZE runs against the SELECTed database, not db0. + expect(await adapter.getDbSize()).toBe(25); + } finally { + await adapter.disconnect(); + } + }); + + it('discovers and scans non-empty sibling databases from a db0 connection', async () => { + const adapter = new UnifiedDatabaseAdapter({ + host: HOST, + port: PORT, + username: 'default', + password: PASSWORD, + // Connected to db0 — the seeded keys live in db13/db14 only. + dbIndex: 0, + }); + try { + await adapter.connect(); + const results = await adapter.collectKeyAnalytics({ + sampleSize: 1000, + scanBatchSize: 100, + fullScan: true, + }); + + const byDb = new Map(results.map((result) => [result.db, result])); + const dbA = byDb.get(DB_A); + const dbB = byDb.get(DB_B); + + expect(dbA).toBeDefined(); + expect(dbA?.dbSize).toBe(25); + expect(dbA?.patterns.some((p) => p.pattern.startsWith('e2emulti:user'))).toBe(true); + + expect(dbB).toBeDefined(); + expect(dbB?.dbSize).toBe(10); + expect(dbB?.patterns.some((p) => p.pattern.startsWith('e2emulti:job'))).toBe(true); + + // The shared client must still sit on its configured database afterwards + // (sibling scans go through short-lived clients, never SELECT). + expect(await adapter.getDbSize()).not.toBe(25); + } finally { + await adapter.disconnect(); + } + }); +}); diff --git a/apps/web/src/api/keyAnalytics.ts b/apps/web/src/api/keyAnalytics.ts index 02159710e..d5b948316 100644 --- a/apps/web/src/api/keyAnalytics.ts +++ b/apps/web/src/api/keyAnalytics.ts @@ -4,29 +4,36 @@ import type { KeyPatternSnapshot, KeyAnalyticsSummary, PatternTrend, HotKeyEntry export type { KeyPatternSnapshot, KeyAnalyticsSummary, PatternTrend, HotKeyEntry, KeySizeDistribution }; export const keyAnalyticsApi = { - getSummary: (startTime?: number, endTime?: number) => { + getSummary: (startTime?: number, endTime?: number, db?: number) => { const params = new URLSearchParams(); if (startTime) params.append('startTime', startTime.toString()); if (endTime) params.append('endTime', endTime.toString()); + if (db !== undefined) params.append('db', db.toString()); const query = params.toString(); return fetchApi(query ? `/key-analytics/summary?${query}` : '/key-analytics/summary'); }, - getPatterns: (options?: { pattern?: string; startTime?: number; endTime?: number; limit?: number }) => { + getPatterns: (options?: { pattern?: string; startTime?: number; endTime?: number; limit?: number; db?: number }) => { const params = new URLSearchParams(); if (options?.pattern) params.append('pattern', options.pattern); if (options?.startTime) params.append('startTime', options.startTime.toString()); if (options?.endTime) params.append('endTime', options.endTime.toString()); if (options?.limit) params.append('limit', options.limit.toString()); + if (options?.db !== undefined) params.append('db', options.db.toString()); const query = params.toString(); return fetchApi(query ? `/key-analytics/patterns?${query}` : '/key-analytics/patterns'); }, - getPatternTrends: (pattern: string, startTime: number, endTime: number) => { + getPatternTrends: (pattern: string, startTime: number, endTime: number, db?: number) => { const params = new URLSearchParams({ pattern, startTime: startTime.toString(), endTime: endTime.toString() }); + if (db !== undefined) params.append('db', db.toString()); return fetchApi(`/key-analytics/trends?${params}`); }, + getDatabases: () => { + return fetchApi('/key-analytics/databases'); + }, + triggerCollection: (deep?: boolean) => { const path = deep ? '/key-analytics/collect?deep=true' : '/key-analytics/collect'; return fetchApi<{ message: string; status: string }>(path, { method: 'POST' }); @@ -36,35 +43,38 @@ export const keyAnalyticsApi = { return fetchApi('/key-analytics/key-sizes'); }, - getLargestKeys: (options?: { limit?: number; startTime?: number; endTime?: number; latest?: boolean; oldest?: boolean }) => { + getLargestKeys: (options?: { limit?: number; startTime?: number; endTime?: number; latest?: boolean; oldest?: boolean; db?: number }) => { const params = new URLSearchParams(); if (options?.limit) params.append('limit', options.limit.toString()); if (options?.startTime) params.append('startTime', options.startTime.toString()); if (options?.endTime) params.append('endTime', options.endTime.toString()); if (options?.latest !== undefined) params.append('latest', options.latest ? 'true' : 'false'); if (options?.oldest) params.append('oldest', 'true'); + if (options?.db !== undefined) params.append('db', options.db.toString()); const query = params.toString(); return fetchApi(query ? `/key-analytics/largest-keys?${query}` : '/key-analytics/largest-keys'); }, - getCompositeKeys: (options?: { limit?: number; startTime?: number; endTime?: number; latest?: boolean; oldest?: boolean }) => { + getCompositeKeys: (options?: { limit?: number; startTime?: number; endTime?: number; latest?: boolean; oldest?: boolean; db?: number }) => { const params = new URLSearchParams(); if (options?.limit) params.append('limit', options.limit.toString()); if (options?.startTime) params.append('startTime', options.startTime.toString()); if (options?.endTime) params.append('endTime', options.endTime.toString()); if (options?.latest !== undefined) params.append('latest', options.latest ? 'true' : 'false'); if (options?.oldest) params.append('oldest', 'true'); + if (options?.db !== undefined) params.append('db', options.db.toString()); const query = params.toString(); return fetchApi(query ? `/key-analytics/composite-keys?${query}` : '/key-analytics/composite-keys'); }, - getHotKeys: (options?: { limit?: number; startTime?: number; endTime?: number; latest?: boolean; oldest?: boolean }) => { + getHotKeys: (options?: { limit?: number; startTime?: number; endTime?: number; latest?: boolean; oldest?: boolean; db?: number }) => { const params = new URLSearchParams(); if (options?.limit) params.append('limit', options.limit.toString()); if (options?.startTime) params.append('startTime', options.startTime.toString()); if (options?.endTime) params.append('endTime', options.endTime.toString()); if (options?.latest) params.append('latest', 'true'); if (options?.oldest) params.append('oldest', 'true'); + if (options?.db !== undefined) params.append('db', options.db.toString()); const query = params.toString(); return fetchApi(query ? `/key-analytics/hot-keys?${query}` : '/key-analytics/hot-keys'); }, diff --git a/apps/web/src/pages/KeyAnalytics.tsx b/apps/web/src/pages/KeyAnalytics.tsx index 7d72e67f7..37416db29 100644 --- a/apps/web/src/pages/KeyAnalytics.tsx +++ b/apps/web/src/pages/KeyAnalytics.tsx @@ -40,9 +40,12 @@ function getRankDelta( currentRank: number, keyName: string, prevKeys: HotKeyEntry[] | null, + db?: number, ): { label: string; className: string } { if (!prevKeys || prevKeys.length === 0) return { label: '', className: '' }; - const prev = prevKeys.find((k) => k.keyName === keyName); + // Key identity includes the logical database: the same name in two + // databases is two distinct keys. + const prev = prevKeys.find((k) => k.keyName === keyName && (k.db ?? 0) === (db ?? 0)); if (!prev) return { label: 'NEW', className: 'text-emerald-600 font-semibold' }; const delta = prev.rank - currentRank; if (delta > 0) return { label: `\u2191${delta}`, className: 'text-emerald-600' }; @@ -122,8 +125,28 @@ export function KeyAnalytics() { const [sortField, setSortField] = useState('keyCount'); const [sortDirection, setSortDirection] = useState('desc'); const [selectedPattern, setSelectedPattern] = useState(null); + // Logical database filter: 'all' aggregates across every scanned database. + const [rawSelectedDb, setSelectedDb] = useState('all'); const detailRef = useRef(null); + // Databases that have analytics data for this connection. The filter is only + // shown when there is more than one (or the single one is not db0). + const { data: availableDbs } = usePolling({ + fetcher: () => keyAnalyticsApi.getDatabases(), + interval: 60000, + refetchKey: currentConnection?.id, + }); + // A db no longer present (e.g. after switching connections) must not keep + // filtering everything down to an empty view — fall back to 'all'. + const selectedDb = + rawSelectedDb !== 'all' && availableDbs && !availableDbs.includes(rawSelectedDb) + ? 'all' + : rawSelectedDb; + const dbFilter = selectedDb === 'all' ? undefined : selectedDb; + const showDbFilter = + (availableDbs?.length ?? 0) > 1 || (availableDbs?.length === 1 && availableDbs[0] !== 0); + const showDbBadges = (availableDbs?.length ?? 0) > 1 && selectedDb === 'all'; + // Scroll to detail panel when a pattern is selected useEffect(() => { if (selectedPattern && detailRef.current) { @@ -138,9 +161,9 @@ export function KeyAnalytics() { loading: _summaryLoading, refresh: refetchSummary, } = usePolling({ - fetcher: () => keyAnalyticsApi.getSummary(), + fetcher: () => keyAnalyticsApi.getSummary(undefined, undefined, dbFilter), interval: 60000, // 1 minute - refetchKey: currentConnection?.id, + refetchKey: `${currentConnection?.id}-${selectedDb}`, }); const { @@ -148,9 +171,9 @@ export function KeyAnalytics() { loading: patternsLoading, refresh: refetchPatterns, } = usePolling({ - fetcher: () => keyAnalyticsApi.getPatterns({ limit: 100 }), + fetcher: () => keyAnalyticsApi.getPatterns({ limit: 100, db: dbFilter }), interval: 60000, - refetchKey: currentConnection?.id, + refetchKey: `${currentConnection?.id}-${selectedDb}`, }); const [hotKeyDateRange, setHotKeyDateRange] = useState(undefined); @@ -171,9 +194,10 @@ export function KeyAnalytics() { latest: true, startTime: hotKeyStartTime, endTime: hotKeyEndTime, + db: dbFilter, }), interval: 60000, - refetchKey: `${currentConnection?.id}-${hotKeyStartTime}-${hotKeyEndTime}`, + refetchKey: `${currentConnection?.id}-${hotKeyStartTime}-${hotKeyEndTime}-${selectedDb}`, enabled: activeTab === 'hot-keys', }); @@ -185,13 +209,16 @@ export function KeyAnalytics() { enabled: activeTab === 'hot-keys' && isHotKeyTimeFiltered, }); - // Deduplicate: keep only the latest snapshot per pattern (results are ordered by timestamp DESC) + // Deduplicate: keep only the latest snapshot per (db, pattern) — results are + // ordered by timestamp DESC, and the same pattern can legitimately exist in + // several logical databases at once. const patterns = useMemo(() => { if (!rawPatterns) return null; const seen = new Set(); return rawPatterns.filter((p) => { - if (seen.has(p.pattern)) return false; - seen.add(p.pattern); + const identity = `${p.db ?? 0}:${p.pattern}`; + if (seen.has(identity)) return false; + seen.add(identity); return true; }); }, [rawPatterns]); @@ -212,9 +239,9 @@ export function KeyAnalytics() { loading: largestKeysLoading, refresh: refetchLargestKeys, } = usePolling({ - fetcher: () => keyAnalyticsApi.getLargestKeys({ limit: 50, latest: true }), + fetcher: () => keyAnalyticsApi.getLargestKeys({ limit: 50, latest: true, db: dbFilter }), interval: 60000, - refetchKey: currentConnection?.id, + refetchKey: `${currentConnection?.id}-${selectedDb}`, enabled: activeTab === 'largest-keys', }); @@ -344,6 +371,23 @@ export function KeyAnalytics() {

+ {showDbFilter && ( + + )} Patterns Hot Keys @@ -556,7 +600,14 @@ export function KeyAnalytics() { }`} onClick={() => setSelectedPattern(pattern.pattern)} > - {pattern.pattern} + + {pattern.pattern} + {showDbBadges && ( + + db{pattern.db ?? 0} + + )} + {formatNumber(pattern.keyCount)} {formatNumber(pattern.sampledKeyCount)} @@ -812,6 +863,7 @@ export function KeyAnalytics() { entry.rank, entry.keyName, baselineHotKeys ?? null, + entry.db, ); const isTop10 = entry.rank <= 10; return ( @@ -825,7 +877,14 @@ export function KeyAnalytics() { {delta.label} - {entry.keyName} + + {entry.keyName} + {showDbBadges && ( + + db{entry.db ?? 0} + + )} + {entry.signalType === 'lfu' ? (
@@ -986,7 +1045,14 @@ export function KeyAnalytics() { {entry.rank} - {entry.keyName} + + {entry.keyName} + {showDbBadges && ( + + db{entry.db ?? 0} + + )} + {entry.keyType ? ( {entry.keyType} diff --git a/packages/shared/src/types/key-analytics.ts b/packages/shared/src/types/key-analytics.ts index 77b998d6d..ee9d83caf 100644 --- a/packages/shared/src/types/key-analytics.ts +++ b/packages/shared/src/types/key-analytics.ts @@ -18,6 +18,8 @@ export interface KeyPatternSnapshot { minTtlSeconds?: number; maxTtlSeconds?: number; connectionId?: string; + /** Logical database index the snapshot was collected from (defaults to 0). */ + db?: number; } export interface KeyPatternQueryOptions { @@ -27,6 +29,8 @@ export interface KeyPatternQueryOptions { limit?: number; offset?: number; connectionId?: string; + /** Restrict to one logical database index. */ + db?: number; } export interface KeyAnalyticsSummary { @@ -106,6 +110,8 @@ export interface KeyPatternData { export interface KeyDetail { keyName: string; keyType: string | null; + /** Logical database index the key lives in (defaults to 0). */ + db?: number; /** #elements for collections, byte length for strings (valkey #1827 "bigkey" metric) */ cardinality: number | null; freqScore: number | null; @@ -119,6 +125,8 @@ export interface KeyAnalyticsResult { scanned: number; patterns: KeyPatternData[]; keyDetails?: KeyDetail[]; + /** Logical database index the result was collected from (defaults to 0). */ + db?: number; } /** @@ -145,6 +153,8 @@ export interface HotKeyEntry { keyType?: string; ttl?: number; rank: number; + /** Logical database index the key lives in (defaults to 0). */ + db?: number; } export interface HotKeyQueryOptions { @@ -157,4 +167,6 @@ export interface HotKeyQueryOptions { oldest?: boolean; /** Restrict to these signal types. Defaults to access signals (`lfu`, `idletime`). */ signalTypes?: Array; + /** Restrict to one logical database index. */ + db?: number; } diff --git a/proprietary/agent/agent-database-adapter.ts b/proprietary/agent/agent-database-adapter.ts index c62479c33..7f37235c5 100644 --- a/proprietary/agent/agent-database-adapter.ts +++ b/proprietary/agent/agent-database-adapter.ts @@ -477,9 +477,11 @@ export class AgentDatabaseAdapter implements DatabasePort { return (await this.sendCommand('LASTSAVE')) as number; } - async collectKeyAnalytics(options: KeyAnalyticsOptions): Promise { + async collectKeyAnalytics(options: KeyAnalyticsOptions): Promise { const response = await this.sendCommand('COLLECT_KEY_ANALYTICS', [JSON.stringify(options)]); - return JSON.parse(response as string); + // The agent scans its connected database only; normalize to the + // one-result-per-database shape of the port. + return [JSON.parse(response as string)]; } async getHashFieldBuffer(key: string, field: string): Promise { diff --git a/proprietary/key-analytics/__tests__/composite-key-ranker.spec.ts b/proprietary/key-analytics/__tests__/composite-key-ranker.spec.ts index a401eb836..9a590c0fd 100644 --- a/proprietary/key-analytics/__tests__/composite-key-ranker.spec.ts +++ b/proprietary/key-analytics/__tests__/composite-key-ranker.spec.ts @@ -144,3 +144,29 @@ describe('rankCompositeKeys', () => { ).toEqual([]); }); }); + +describe('rankCompositeKeys across logical databases', () => { + it('treats the same key name in two databases as two distinct keys', () => { + const candidates: CompositeCandidate[] = [ + candidate({ keyName: 'shared', db: 0, freqScore: 200, cardinality: 5000 }), + candidate({ keyName: 'shared', db: 1, freqScore: 100, cardinality: 2500 }), + ]; + + const ranked = rankCompositeKeys(candidates, 5); + + // Both place on both dimensions independently — no cross-db merge. + expect(ranked).toHaveLength(2); + expect(ranked.map((r) => [r.keyName, r.db])).toEqual([ + ['shared', 0], + ['shared', 1], + ]); + }); + + it('passes the db through on the ranked entry', () => { + const ranked = rankCompositeKeys( + [candidate({ keyName: 'hotbig', db: 3, freqScore: 10, cardinality: 10 })], + 5, + ); + expect(ranked[0].db).toBe(3); + }); +}); diff --git a/proprietary/key-analytics/__tests__/key-analytics-topology.spec.ts b/proprietary/key-analytics/__tests__/key-analytics-topology.spec.ts index de68890c1..75e6a2839 100644 --- a/proprietary/key-analytics/__tests__/key-analytics-topology.spec.ts +++ b/proprietary/key-analytics/__tests__/key-analytics-topology.spec.ts @@ -191,7 +191,8 @@ interface FakeClient { function fakeClient(role: 'master' | 'slave' | 'sentinel', result: KeyAnalyticsResult | Error, keySizes = ''): FakeClient { return { getRole: jest.fn().mockResolvedValue({ role }), - collectKeyAnalytics: result instanceof Error ? jest.fn().mockRejectedValue(result) : jest.fn().mockResolvedValue(result), + // The port returns one result per scanned logical database. + collectKeyAnalytics: result instanceof Error ? jest.fn().mockRejectedValue(result) : jest.fn().mockResolvedValue([result]), call: jest.fn().mockResolvedValue(keySizes), getCapabilities: () => ({ isSentinel: role === 'sentinel' }), }; @@ -346,6 +347,46 @@ describe('KeyAnalyticsService with topology members', () => { expect(snapshots[0]).toMatchObject({ pattern: 'user:*', keyCount: 50 }); }); + it('keeps the per-database results of one instance separate and db-tags the hot keys', async () => { + const { poll, clients, storage } = cluster(); + // The standalone instance holds keys in db0 and db1 — one result each. + clients.standalone.collectKeyAnalytics.mockResolvedValue([ + { ...scanOf(50), db: 0 }, + { ...scanOf(70, 'job:*'), db: 1 }, + ]); + + await poll('standalone'); + + const [snapshots] = storage.saveKeyPatternSnapshots.mock.calls[0] as [KeyPatternSnapshot[]]; + expect(snapshots.map((s) => [s.pattern, s.db, s.keyCount]).sort()).toEqual([ + ['job:*', 1, 70], + ['user:*', 0, 50], + ]); + + const [hotKeys] = storage.saveHotKeys.mock.calls[0] as [Array<{ keyName: string; signalType: string; db?: number }>]; + const cardinality = hotKeys.filter((row) => row.signalType === 'cardinality'); + expect(cardinality.map((row) => [row.keyName, row.db])).toEqual([ + ['job:*:70', 1], + ['user:*:50', 0], + ]); + }); + + it('keeps the same pattern apart across logical databases instead of merging it', async () => { + const { poll, clients, storage } = cluster(); + clients.standalone.collectKeyAnalytics.mockResolvedValue([ + { ...scanOf(50), db: 0 }, + { ...scanOf(70), db: 1 }, + ]); + + await poll('standalone'); + + const [snapshots] = storage.saveKeyPatternSnapshots.mock.calls[0] as [KeyPatternSnapshot[]]; + expect(snapshots.map((s) => [s.pattern, s.db, s.keyCount]).sort()).toEqual([ + ['user:*', 0, 50], + ['user:*', 1, 70], + ]); + }); + it('scans a sentinel primary but not its replicas or a retired node', async () => { const connections = [ connection('sentinel'), diff --git a/proprietary/key-analytics/__tests__/key-analytics.service.spec.ts b/proprietary/key-analytics/__tests__/key-analytics.service.spec.ts index 3cbbd61a9..2fdfae60b 100644 --- a/proprietary/key-analytics/__tests__/key-analytics.service.spec.ts +++ b/proprietary/key-analytics/__tests__/key-analytics.service.spec.ts @@ -94,7 +94,8 @@ describe('KeyAnalyticsService.collect composite persistence', () => { client: { collectKeyAnalytics: jest.Mock }; } { const client = { - collectKeyAnalytics: jest.fn().mockResolvedValue({ + // One result per scanned logical database. + collectKeyAnalytics: jest.fn().mockResolvedValue([{ dbSize: 3, scanned: 3, patterns: [], @@ -106,7 +107,7 @@ describe('KeyAnalyticsService.collect composite persistence', () => { // Big but LFU-cold (freq 0 is dropped) -> cardinality dimension only. { keyName: 'onlybig', keyType: 'hash', freqScore: 0, idleSeconds: null, memoryBytes: 900, cardinality: 4000, ttl: null }, ], - }), + }]), }; const storage = { saveHotKeys, diff --git a/proprietary/key-analytics/__tests__/key-pattern-snapshots.spec.ts b/proprietary/key-analytics/__tests__/key-pattern-snapshots.spec.ts index cb1f83e84..aa54a5c02 100644 --- a/proprietary/key-analytics/__tests__/key-pattern-snapshots.spec.ts +++ b/proprietary/key-analytics/__tests__/key-pattern-snapshots.spec.ts @@ -1,5 +1,5 @@ import type { KeyPatternData } from '@betterdb/shared'; -import { buildPatternSnapshots } from '../key-pattern-snapshots'; +import { buildPatternSnapshots, mergePatternSnapshots } from '../key-pattern-snapshots'; const DAY = 86400; @@ -76,3 +76,33 @@ describe('buildPatternSnapshots stale key count', () => { expect(snapshot.coldKeyCount).toBe(2); }); }); + +describe('pattern snapshots across logical databases', () => { + it('tags each snapshot with the logical database of its scan result', () => { + const [snapshot] = buildPatternSnapshots( + { dbSize: 4, scanned: 4, patterns: [pattern()], db: 2 }, + 1, + ); + expect(snapshot.db).toBe(2); + }); + + it('defaults the snapshot db to 0 for results without one (older agents)', () => { + const [snapshot] = buildPatternSnapshots({ dbSize: 4, scanned: 4, patterns: [pattern()] }, 1); + expect(snapshot.db).toBe(0); + }); + + it('merges a pattern across cluster nodes but never across logical databases', () => { + const nodeA = buildPatternSnapshots({ dbSize: 4, scanned: 4, patterns: [pattern()], db: 0 }, 1); + const nodeB = buildPatternSnapshots({ dbSize: 8, scanned: 8, patterns: [pattern({ count: 8, totalMemory: 800 })], db: 0 }, 1); + const otherDb = buildPatternSnapshots({ dbSize: 2, scanned: 2, patterns: [pattern({ count: 2, totalMemory: 200 })], db: 1 }, 1); + + const merged = mergePatternSnapshots([nodeA, nodeB, otherDb]); + + // session:* in db0 (merged across the two nodes) and session:* in db1. + expect(merged).toHaveLength(2); + const db0 = merged.find((s) => s.db === 0); + const db1 = merged.find((s) => s.db === 1); + expect(db0?.keyCount).toBe(12); + expect(db1?.keyCount).toBe(2); + }); +}); diff --git a/proprietary/key-analytics/composite-key-ranker.ts b/proprietary/key-analytics/composite-key-ranker.ts index eef73dc48..544322790 100644 --- a/proprietary/key-analytics/composite-key-ranker.ts +++ b/proprietary/key-analytics/composite-key-ranker.ts @@ -39,6 +39,8 @@ export type CompositeDimension = 'hotness' | 'cardinality'; export interface CompositeCandidate { keyName: string; keyType: string | null; + /** Logical database index the key lives in; part of the key's identity. */ + db?: number; /** LFU access frequency (OBJECT FREQ). Null unless an LFU maxmemory-policy is set. */ freqScore: number | null; /** Idle seconds (OBJECT IDLETIME). The recency fallback when freqScore is absent. */ @@ -52,6 +54,8 @@ export interface CompositeCandidate { export interface CompositeKeyRank { keyName: string; keyType: string | null; + /** Logical database index the key lives in, passed through from the candidate. */ + db?: number; ttl: number | null; /** Dimensions on which the key placed in that dimension's top-N (length >= 2). */ dimensions: CompositeDimension[]; @@ -131,7 +135,12 @@ export function rankCompositeKeys( return []; } - // keyName -> (dimension -> ranking value) for the dimensions where the key made top-N. + // The same key name in two logical databases is two distinct keys, so the + // identity used for membership / dedupe includes the database index. + const identityOf = (candidate: CompositeCandidate): string => + `${candidate.db ?? 0} ${candidate.keyName}`; + + // key identity -> (dimension -> ranking value) for the dimensions where the key made top-N. const membership = new Map>(); const dimensionMax = new Map(); @@ -147,10 +156,10 @@ export function rankCompositeKeys( dimensionMax.set(dimension, withValue[0].value); for (const { candidate, value } of withValue.slice(0, perDimensionTopN)) { - let dims = membership.get(candidate.keyName); + let dims = membership.get(identityOf(candidate)); if (!dims) { dims = new Map(); - membership.set(candidate.keyName, dims); + membership.set(identityOf(candidate), dims); } // Keep the largest value if a key somehow appears twice for a dimension. dims.set(dimension, Math.max(value, dims.get(dimension) ?? value)); @@ -161,14 +170,14 @@ export function rankCompositeKeys( const seen = new Set(); for (const candidate of candidates) { - if (seen.has(candidate.keyName)) { + if (seen.has(identityOf(candidate))) { continue; } - const dims = membership.get(candidate.keyName); + const dims = membership.get(identityOf(candidate)); if (!dims || dims.size < COMPOSITE_MIN_DIMENSIONS) { continue; } - seen.add(candidate.keyName); + seen.add(identityOf(candidate)); let score = 0; for (const [dimension, value] of dims) { @@ -182,6 +191,7 @@ export function rankCompositeKeys( ranked.push({ keyName: candidate.keyName, keyType: candidate.keyType, + db: candidate.db, ttl: candidate.ttl, dimensions: [...dims.keys()], // Report the raw signal the hotness placement came from, not the derived magnitude. diff --git a/proprietary/key-analytics/key-analytics.controller.ts b/proprietary/key-analytics/key-analytics.controller.ts index f2a65216d..63660cdef 100644 --- a/proprietary/key-analytics/key-analytics.controller.ts +++ b/proprietary/key-analytics/key-analytics.controller.ts @@ -26,10 +26,11 @@ export class KeyAnalyticsController { @ConnectionId() connectionId?: string, @Query('startTime') startTime?: string, @Query('endTime') endTime?: string, + @Query('db') db?: string, ) { const start = startTime ? parseInt(startTime, 10) : undefined; const end = endTime ? parseInt(endTime, 10) : undefined; - return this.keyAnalytics.getSummary(start, end, connectionId); + return this.keyAnalytics.getSummary(start, end, connectionId, safeInt(db)); } @Get('patterns') @@ -42,6 +43,7 @@ export class KeyAnalyticsController { @Query('startTime') startTime?: string, @Query('endTime') endTime?: string, @Query('limit') limit?: string, + @Query('db') db?: string, ) { return this.keyAnalytics.getPatternSnapshots({ pattern, @@ -49,6 +51,7 @@ export class KeyAnalyticsController { endTime: endTime ? parseInt(endTime, 10) : undefined, limit: limit ? parseInt(limit, 10) : undefined, connectionId, + db: safeInt(db), }); } @@ -63,6 +66,7 @@ export class KeyAnalyticsController { @Query('endTime') endTime?: string, @Query('latest') latest?: string, @Query('oldest') oldest?: string, + @Query('db') db?: string, ) { return this.keyAnalytics.getHotKeys({ connectionId, @@ -71,6 +75,7 @@ export class KeyAnalyticsController { endTime: safeInt(endTime), latest: latest === 'true', oldest: oldest === 'true', + db: safeInt(db), }); } @@ -85,6 +90,7 @@ export class KeyAnalyticsController { @Query('endTime') endTime?: string, @Query('latest') latest?: string, @Query('oldest') oldest?: string, + @Query('db') db?: string, ) { const wantOldest = oldest === 'true'; return this.keyAnalytics.getLargestKeys({ @@ -95,6 +101,7 @@ export class KeyAnalyticsController { // Default to the latest snapshot unless an explicit time range or oldest is requested. latest: wantOldest ? false : latest !== 'false', oldest: wantOldest, + db: safeInt(db), }); } @@ -109,6 +116,7 @@ export class KeyAnalyticsController { @Query('endTime') endTime?: string, @Query('latest') latest?: string, @Query('oldest') oldest?: string, + @Query('db') db?: string, ) { const wantOldest = oldest === 'true'; return this.keyAnalytics.getCompositeKeys({ @@ -119,6 +127,7 @@ export class KeyAnalyticsController { // Default to the latest snapshot unless an explicit time range or oldest is requested. latest: wantOldest ? false : latest !== 'false', oldest: wantOldest, + db: safeInt(db), }); } @@ -131,11 +140,12 @@ export class KeyAnalyticsController { @Query('pattern') pattern?: string, @Query('startTime') startTime?: string, @Query('endTime') endTime?: string, + @Query('db') db?: string, ) { if (!pattern || !startTime || !endTime) { throw new Error('pattern, startTime, and endTime are required'); } - return this.keyAnalytics.getPatternTrends(pattern, parseInt(startTime, 10), parseInt(endTime, 10), connectionId); + return this.keyAnalytics.getPatternTrends(pattern, parseInt(startTime, 10), parseInt(endTime, 10), connectionId, safeInt(db)); } @Post('collect') @@ -151,6 +161,14 @@ export class KeyAnalyticsController { }; } + @Get('databases') + @UseGuards(LicenseGuard) + @RequiresFeature('keyAnalytics') + @ApiHeader({ name: CONNECTION_ID_HEADER, required: false, description: 'Connection ID to filter by' }) + async getDatabases(@ConnectionId() connectionId?: string) { + return this.keyAnalytics.getDatabases(connectionId); + } + @Get('key-sizes') @UseGuards(LicenseGuard) @RequiresFeature('keyAnalytics') diff --git a/proprietary/key-analytics/key-analytics.service.ts b/proprietary/key-analytics/key-analytics.service.ts index 1dd625caec41297291659bf61b92175f92afc91b..6c4a9e634fce5ef43d4cc2496b484e1bf7b779bd 100644 GIT binary patch delta 1190 zcmZXUyKfUg5XLcih+LEuP=G;!35xJJa?U^qiDD%dC?XQChICet*S8yc?eng6yCE^b z3O!AvrK3U!geZ9EI%uGxrK6{aXb}GbvuB?Zk!yV3{l3|m-^|{tx2dn6QXg+RyZJn< z)fgz!WYA!35cb6<0@yZJ8n^((9&yReje7d!ZhK5vJ^|}S+7Z2uG^qlSXu^m2M|qz=73ffdnF~np_9$#dhxOa z$1&pSM5HG7hejb*Hz+Ejdzs;Z!waUKpNQ!6-_(eW7;DUpOX=ug?o6Wd8`fCQ?+u*o zc3t7*2}V~c3YI{yEqK7R`3C7by&8HGki{(QxB>KRt&kB4I0L<>D_35?L={`N3|IPF zV1CY?-<{eyILcG}F|!&dLitFCtxkc9Cr|B^5Yd5e-BU8uu1j)Ii5U`@!8flixo|TU zBxJIkY#A0I(+YW$*luS!WlUu(Nffs-1@mY8v~gy?46C?~Xz&_Mjah$j5yWP*!b0d6NKRpdA$xS^{ zb9@_dGUK^dT84}Q5hQ49?km`kVSE{oy(); for (const snapshot of perNode.flat()) { - byPattern.set(snapshot.pattern, [...(byPattern.get(snapshot.pattern) ?? []), snapshot]); + const identity = `${snapshot.db ?? 0} ${snapshot.pattern}`; + byPattern.set(identity, [...(byPattern.get(identity) ?? []), snapshot]); } return Array.from(byPattern.values()).map((group) => (group.length === 1 ? group[0] : mergeGroup(group))); } @@ -103,6 +108,7 @@ function mergeGroup(group: KeyPatternSnapshot[]): KeyPatternSnapshot { id: randomUUID(), timestamp: group[0].timestamp, pattern: group[0].pattern, + db: group[0].db, keyCount: sum(keyCount), sampledKeyCount: sum((snapshot) => snapshot.sampledKeyCount), keysWithTtl: sum((snapshot) => snapshot.keysWithTtl), From 9ea709b1627623d3294b1b095d38dd832a79702f Mon Sep 17 00:00:00 2001 From: Kristiyan Ivanov Date: Thu, 15 Oct 2026 20:27:48 +0300 Subject: [PATCH 02/17] fix(key-analytics): use a plain space in the dedupe identity, not a raw NUL byte The embedded \x00 made git treat the file as binary. --- .../key-analytics/key-analytics.service.ts | Bin 22720 -> 22720 bytes 1 file changed, 0 insertions(+), 0 deletions(-) diff --git a/proprietary/key-analytics/key-analytics.service.ts b/proprietary/key-analytics/key-analytics.service.ts index 6c4a9e634fce5ef43d4cc2496b484e1bf7b779bd..aaec04093293650eb9938df3430569b432d9c1b5 100644 GIT binary patch delta 16 YcmX@Gk@3Jr#tmP&7!@{u<2vOF078BTe*gdg delta 16 YcmX@Gk@3Jr#tmP&7#TKy<2vOF071S7UjP6A From 51bd46d4c1731ac72c054e8bd9af866546ec71d7 Mon Sep 17 00:00:00 2001 From: Kristiyan Ivanov Date: Fri, 16 Oct 2026 07:27:07 +0300 Subject: [PATCH 03/17] fix(key-analytics): apply code-review fixes for multi-db support (#542) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(api): avoid int4 overflow in key analytics weighted avg memory query SUM(avg_memory_bytes * sampled_key_count) multiplies two int4 columns, which Postgres evaluates as int4 and can overflow to "integer out of range" on large patterns. Cast avg_memory_bytes to bigint first. * fix(api): harden multi-db key analytics scan and connection dbIndex handling - collectKeyAnalytics: wrap the own-db scan in the same try/catch as the scratch-client branch so a failure on the connection's own database no longer aborts the loop and discards sibling results. - collectKeyAnalytics: attach an error listener to each scratch client (iovalkey otherwise logs raw errors via console.error), explicitly SELECT after connecting so a SELECT iovalkey would silently swallow (e.g. restricted ACL) surfaces as a caught per-db failure instead of silently scanning db0, and skip a database when the SSH tunnel is down instead of dialing the raw host. - doConnect: explicitly SELECT when dbIndex > 0 so an invalid dbIndex (out of range, or illegal in cluster mode) surfaces as a connection error instead of silently running on db0. - testConnection: thread dbIndex into the adapter it builds, so Test Connection validates the same logical database the saved connection will use. - adoptChild: force dbIndex to 0 when adopting a connection as a cluster member, matching addManagedChild, since SELECT > 0 is illegal in cluster mode. * fix(api): stable ordering and per-connection grouping for key pattern queries - getKeyPatternSnapshots (sqlite + postgres): ORDER BY timestamp DESC had no tiebreaker, so when one collection writes patterns x dbs rows at a single timestamp, LIMIT truncated an arbitrary subset of them. Add db ASC, pattern ASC as tiebreakers. - getKeyPatternTrends (sqlite + postgres): GROUP BY timestamp merged rows from different connections together when connectionId is not provided. Group by (timestamp, connection_id) in that case so each connection keeps its own series, same as before the multi-db change; keep plain GROUP BY timestamp when a connectionId scopes the query. * perf(api): index key_pattern_snapshots(connection_id, db) The database selector polls SELECT DISTINCT db every 60s; back it with a composite index on (connection_id, db) in both adapters. Added after the db column's own migration/ALTER so it only runs once the column exists. * fix(web): scope hot-key baseline and pattern selection to the active db - useBaselineHotKeys: accept a db option, pass it to getHotKeys and include it in the query key; KeyAnalytics now passes the active dbFilter so rank deltas compare the same logical database instead of mixing dbs into one baseline. - KeyAnalytics: the patterns table renders one row per (db, pattern) but selection tracked only the pattern name, so selecting a pattern that exists in more than one db highlighted/opened the wrong row. Selection state now carries { db, pattern }, compared on both fields for the row highlight and the detail lookup; the detail card title shows the db when the connection has more than one. - KeyAnalytics: reset the db filter to 'all' whenever the active connection changes, so switching connections doesn't keep querying with a stale numeric db filter while its database list reloads, or silently snap back from 'all' if the remembered db happens to also exist on the new connection. * test(api): cover own-db failure isolation, scratch SELECT hardening, and stable snapshot ordering - unified.adapter.spec.ts: a failing own-db scan no longer discards sibling results; a scratch client whose SELECT is rejected (e.g. restricted ACL) is treated as a failed scan instead of silently tagging its keys with the wrong db; a sibling db is skipped rather than dialing the raw host when the SSH tunnel is down. - key-analytics-multi-db.spec.ts: snapshots tied on timestamp are returned in a deterministic (db, pattern) order instead of an arbitrary LIMIT-truncated subset; trends from different connections stay separate series when no connectionId scopes the query. * fix(api): disconnect the client when a post-connect step fails doConnect's catch block only tore down the SSH tunnel, not the client itself. The new dbIndex SELECT (and the pre-existing detectCapabilities call) run after the socket has already connected and authenticated, so a failure there left a live, authenticated session open on a direct connection — e.g. every failed Test Connection attempt with an invalid or cluster-illegal dbIndex built and abandoned its own adapter. Disconnect and discard the client on any connect failure, not just a tunneled one. Found by Cursor Bugbot on PR #542. --------- --- .../connection-registry.service.ts | 12 +++ .../database/adapters/unified.adapter.spec.ts | 89 +++++++++++++++++++ .../src/database/adapters/unified.adapter.ts | 44 ++++++++- .../__tests__/key-analytics-multi-db.spec.ts | 32 +++++++ .../src/storage/adapters/postgres.adapter.ts | 12 ++- .../src/storage/adapters/sqlite.adapter.ts | 14 ++- apps/web/src/hooks/useBaselineHotKeys.ts | 5 +- apps/web/src/pages/KeyAnalytics.tsx | 30 +++++-- 8 files changed, 224 insertions(+), 14 deletions(-) diff --git a/apps/api/src/connections/connection-registry.service.ts b/apps/api/src/connections/connection-registry.service.ts index 609022a2c..ed26445b2 100644 --- a/apps/api/src/connections/connection-registry.service.ts +++ b/apps/api/src/connections/connection-registry.service.ts @@ -803,6 +803,7 @@ export class ConnectionRegistry implements OnModuleInit, OnModuleDestroy { username: request.username || 'default', password: request.password || '', tls: request.tls, + dbIndex: request.dbIndex, // Unique id so a test tunnel never collides with a live connection's tunnel. connectionId: `test:${randomUUID()}`, sshTunnel: this.sanitizeSshTunnelInput(request.sshTunnel), @@ -1153,6 +1154,17 @@ export class ConnectionRegistry implements OnModuleInit, OnModuleDestroy { async adoptChild(id: string, seedId: string, node: DesiredNode): Promise { await this.setMembership(id, { seedId, origin: 'adopted', ...membershipFields(node) }); + if (node.source === 'cluster') { + // SELECT on a non-zero db is illegal in cluster mode; force db0 like + // addManagedChild does for auto-registered cluster members, since an + // adopted connection may have been configured with a non-zero dbIndex + // before joining the cluster. + const config = this.configs.get(id); + if (config && (config.dbIndex ?? 0) !== 0) { + await this.storage.updateConnection(id, { dbIndex: 0 }); + this.configs.set(id, { ...config, dbIndex: 0 }); + } + } } async refreshChild(id: string, node: DesiredNode): Promise { diff --git a/apps/api/src/database/adapters/unified.adapter.spec.ts b/apps/api/src/database/adapters/unified.adapter.spec.ts index fa36d2944..f27daee9b 100644 --- a/apps/api/src/database/adapters/unified.adapter.spec.ts +++ b/apps/api/src/database/adapters/unified.adapter.spec.ts @@ -234,6 +234,8 @@ describe('UnifiedDatabaseAdapter.collectKeyAnalytics — multiple logical databa scan: jest.fn().mockResolvedValue(['0', keys]), pipeline: scanChain, connect: jest.fn().mockResolvedValue(undefined), + select: jest.fn().mockResolvedValue('OK'), + on: jest.fn(), quit: jest.fn().mockResolvedValue('OK'), disconnect: jest.fn(), }; @@ -349,6 +351,63 @@ describe('UnifiedDatabaseAdapter.collectKeyAnalytics — multiple logical databa expect(results.map((r) => r.db)).toEqual([0, 2]); expect(scratchClients[2].quit).toHaveBeenCalled(); }); + + it('keeps scanning sibling databases when the own-db scan fails', async () => { + const { adapter, ownClient, scratchClients } = makeMultiDbAdapter({ + dbIndex: 0, + keyspaceInfo: 'db0:keys=1,expires=0,avg_ttl=0\r\ndb1:keys=1,expires=0,avg_ttl=0\r\n', + ownKeys: ['a:1'], + scratchKeysByDb: { 1: ['b:1'] }, + }); + ownClient.scan.mockRejectedValue(new Error('own-db scan failed')); + + const results = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + + // The old implementation let an own-db failure propagate out of the loop + // and discard sibling results. Now db1 still comes back. + expect(results.map((r) => r.db)).toEqual([1]); + expect(scratchClients[1].quit).toHaveBeenCalled(); + }); + + it('treats a scratch client that cannot SELECT the target db as a failed scan', async () => { + const { adapter, scratchClients } = makeMultiDbAdapter({ + dbIndex: 0, + keyspaceInfo: 'db0:keys=1,expires=0,avg_ttl=0\r\ndb1:keys=1,expires=0,avg_ttl=0\r\n', + ownKeys: ['a:1'], + scratchKeysByDb: { 1: ['b:1'] }, + }); + // iovalkey silently swallows a failed handshake SELECT; the explicit + // post-connect select() is what must surface this as a failure. + const original = (adapter as unknown as { createValkeyClient: jest.Mock }) + .createValkeyClient; + original.mockImplementationOnce((_name: string, _db?: number) => { + const restricted = fakeScanClient(['b:1']); + restricted.select.mockRejectedValue(new Error('NOPERM this user has no permissions')); + return restricted; + }); + + const results = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + + // db1 must not appear tagged with data actually scanned from db0. + expect(results.map((r) => r.db)).toEqual([0]); + expect(scratchClients[1]).toBeUndefined(); + }); + + it('skips a sibling database instead of dialing the raw host when the SSH tunnel is down', async () => { + const { adapter, createValkeyClient } = makeMultiDbAdapter({ + dbIndex: 0, + keyspaceInfo: 'db0:keys=1,expires=0,avg_ttl=0\r\ndb1:keys=1,expires=0,avg_ttl=0\r\n', + ownKeys: ['a:1'], + scratchKeysByDb: { 1: ['b:1'] }, + }); + (adapter as unknown as { usesTunnel: boolean }).usesTunnel = true; + (adapter as unknown as { tunnelActive: boolean }).tunnelActive = false; + + const results = await adapter.collectKeyAnalytics({ sampleSize: 100, scanBatchSize: 100 }); + + expect(results.map((r) => r.db)).toEqual([0]); + expect(createValkeyClient).not.toHaveBeenCalled(); + }); }); describe('UnifiedDatabaseAdapter — configured dbIndex', () => { @@ -375,4 +434,34 @@ describe('UnifiedDatabaseAdapter — configured dbIndex', () => { const client = adapter.getClient() as unknown as { options: { db?: number } }; expect(client.options.db).toBe(0); }); + + it('surfaces a SELECT failure on an invalid dbIndex and disconnects the leaked client', async () => { + const adapter = Object.create(UnifiedDatabaseAdapter.prototype) as UnifiedDatabaseAdapter; + const client = { + options: { host: 'localhost', port: 6379 }, + connect: jest.fn().mockResolvedValue(undefined), + select: jest.fn().mockRejectedValue(new Error('ERR DB index is out of range')), + disconnect: jest.fn(), + }; + (adapter as unknown as { _client: unknown })._client = client; + (adapter as unknown as { config: unknown }).config = { dbIndex: 5 }; + (adapter as unknown as { usesTunnel: boolean }).usesTunnel = false; + (adapter as unknown as { tunnelActive: boolean }).tunnelActive = false; + (adapter as unknown as { logger: unknown }).logger = { + debug: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + log: jest.fn(), + }; + + await expect( + (adapter as unknown as { doConnect(): Promise }).doConnect(), + ).rejects.toThrow('ERR DB index is out of range'); + + // The socket connected and authenticated before SELECT failed; without an + // explicit disconnect here it would stay open, leaking one live session + // per failed attempt (e.g. each failed Test Connection with a bad dbIndex). + expect(client.disconnect).toHaveBeenCalled(); + expect((adapter as unknown as { _client: unknown })._client).toBeNull(); + }); }); diff --git a/apps/api/src/database/adapters/unified.adapter.ts b/apps/api/src/database/adapters/unified.adapter.ts index 526446b67..38958a705 100644 --- a/apps/api/src/database/adapters/unified.adapter.ts +++ b/apps/api/src/database/adapters/unified.adapter.ts @@ -297,6 +297,14 @@ export class UnifiedDatabaseAdapter implements DatabasePort { } this.logger.log(`Connecting to ${this.client.options.host}:${this.client.options.port}...`); await this.client.connect(); + // iovalkey silently swallows a failed handshake SELECT (silentEmit), so an + // invalid dbIndex (out of range, or SELECT rejected in cluster mode) would + // otherwise leave the client connected and quietly running on db0. Select + // explicitly so that failure surfaces here as a connection error. + const dbIndex = this.config.dbIndex ?? 0; + if (dbIndex > 0) { + await this.client.select(dbIndex); + } this.connected = true; this.logger.log('Connected successfully'); await this.detectCapabilities(); @@ -304,6 +312,16 @@ export class UnifiedDatabaseAdapter implements DatabasePort { } catch (error) { this.connected = false; await this.teardownTunnel(); + // A later setup step (the dbIndex SELECT, detectCapabilities) can fail + // after the socket already connected and authenticated. teardownTunnel + // only discards the client for a tunneled connection, so a direct + // connection's client would otherwise stay open — leaking a live + // session per failed attempt (e.g. each failed Test Connection call + // with a bad dbIndex builds its own throwaway adapter). + if (this._client) { + this._client.disconnect(); + this._client = null; + } this.logger.error(`Connection failed: ${error instanceof Error ? error.message : error}`); throw error; } @@ -862,15 +880,39 @@ export class UnifiedDatabaseAdapter implements DatabasePort { // than today's single-db scan. for (const db of databases) { if (db === ownDb) { - results.push(await this.scanDatabase(this.client, db, options)); + try { + results.push(await this.scanDatabase(this.client, db, options)); + } catch (error) { + failures.push(error); + this.logger.warn( + `Key analytics scan failed on db${db}: ${error instanceof Error ? error.message : error}`, + ); + } + continue; + } + // After a tunnel drop, connectHost/connectPort point at the real (possibly + // bastion-only) endpoint; a scratch client built now would dial it directly + // and bypass the tunnel. Treat the database as failed instead, mirroring + // getCliClient's guard. + if (this.usesTunnel && !this.tunnelActive) { + failures.push(new Error('SSH tunnel is not established')); + this.logger.warn(`Key analytics scan skipped on db${db}: SSH tunnel is not established`); continue; } const scratch = this.createValkeyClient( `${this.config.connectionName ?? 'BetterDB-Monitor'}-keyscan-db${db}`, db, ); + scratch.on('error', (err) => { + this.logger.debug(`Scratch client error on db${db}: ${err.message}`); + }); try { await scratch.connect(); + // iovalkey silently swallows a failed handshake SELECT (silentEmit), so + // a client that cannot SELECT this db (e.g. a restricted ACL) would + // otherwise keep scanning db0 while results are tagged with `db`. Select + // explicitly so that failure surfaces here as a caught per-db failure. + await scratch.select(db); results.push(await this.scanDatabase(scratch, db, options)); } catch (error) { failures.push(error); diff --git a/apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts b/apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts index ce554f653..b1ad9e69c 100644 --- a/apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts +++ b/apps/api/src/storage/adapters/__tests__/key-analytics-multi-db.spec.ts @@ -112,6 +112,38 @@ describe('SqliteAdapter key analytics per logical database', () => { expect(db1).toEqual([{ timestamp: 1_000, keyCount: 30, memoryBytes: 100, staleCount: 0 }]); }); + it('orders tied timestamps deterministically by db then pattern instead of truncating an arbitrary subset', async () => { + // One collection across 2 dbs x 2 patterns writes 4 rows at the same + // timestamp. A LIMIT smaller than that must keep a stable, deterministic + // prefix (db ASC, pattern ASC) rather than whatever order SQLite happens + // to return ties in. + await storage.saveKeyPatternSnapshots( + [ + snapshot({ id: 's-0-b', db: 0, pattern: 'b:*' }), + snapshot({ id: 's-1-a', db: 1, pattern: 'a:*' }), + snapshot({ id: 's-0-a', db: 0, pattern: 'a:*' }), + snapshot({ id: 's-1-b', db: 1, pattern: 'b:*' }), + ], + CONN, + ); + + const page = await storage.getKeyPatternSnapshots({ connectionId: CONN, limit: 2 }); + + expect(page.map((row) => [row.db, row.pattern])).toEqual([ + [0, 'a:*'], + [0, 'b:*'], + ]); + }); + + it('keeps trends from different connections as separate series when connectionId is not provided', async () => { + await storage.saveKeyPatternSnapshots([snapshot({ id: 's0', keyCount: 10 })], CONN); + await storage.saveKeyPatternSnapshots([snapshot({ id: 's1', keyCount: 30 })], 'conn-b'); + + const unscoped = await storage.getKeyPatternTrends('user:*', 0, 2_000); + + expect(unscoped.map((row) => row.keyCount).sort((a, b) => a - b)).toEqual([10, 30]); + }); + it('lists the distinct databases that have analytics data', async () => { await storage.saveKeyPatternSnapshots( [snapshot({ id: 's0', db: 0 }), snapshot({ id: 's1', db: 2 }), snapshot({ id: 's2', db: 2, pattern: 'job:*' })], diff --git a/apps/api/src/storage/adapters/postgres.adapter.ts b/apps/api/src/storage/adapters/postgres.adapter.ts index e89627e2d..d214a228a 100644 --- a/apps/api/src/storage/adapters/postgres.adapter.ts +++ b/apps/api/src/storage/adapters/postgres.adapter.ts @@ -1420,6 +1420,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { CREATE INDEX IF NOT EXISTS idx_kps_pattern ON key_pattern_snapshots(pattern, timestamp DESC); CREATE INDEX IF NOT EXISTS idx_kps_pattern_timestamp ON key_pattern_snapshots(pattern, timestamp); CREATE INDEX IF NOT EXISTS idx_kps_connection_id ON key_pattern_snapshots(connection_id); + CREATE INDEX IF NOT EXISTS idx_kps_connection_db ON key_pattern_snapshots(connection_id, db); CREATE TABLE IF NOT EXISTS hot_key_stats ( id UUID PRIMARY KEY, @@ -2764,7 +2765,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { ` SELECT * FROM key_pattern_snapshots ${whereClause} - ORDER BY timestamp DESC + ORDER BY timestamp DESC, db ASC, pattern ASC LIMIT $${paramIndex++} OFFSET $${paramIndex++} `, [...params, limit, offset], @@ -2862,7 +2863,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { SUM(hot_key_count) as hot_key_count, SUM(cold_key_count) as cold_key_count, SUM(sampled_key_count) as sampled_key_count, - SUM(avg_memory_bytes * sampled_key_count) as weighted_avg_memory + SUM(avg_memory_bytes::bigint * sampled_key_count) as weighted_avg_memory FROM key_pattern_snapshots WHERE ${patternWhere} GROUP BY pattern @@ -2942,7 +2943,10 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { } // One collection writes one row per logical database; aggregate them per - // timestamp so an unfiltered trend stays a single series. + // timestamp so an unfiltered trend stays a single series. When no + // connectionId is given, also group by connection_id so trends from + // different connections aren't merged into one series. + const groupBy = connectionId ? 'timestamp' : 'timestamp, connection_id'; const result = await this.pool.query( ` SELECT @@ -2952,7 +2956,7 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { SUM(stale_key_count) as stale_key_count FROM key_pattern_snapshots WHERE ${conditions.join(' AND ')} - GROUP BY timestamp + GROUP BY ${groupBy} ORDER BY timestamp ASC `, params, diff --git a/apps/api/src/storage/adapters/sqlite.adapter.ts b/apps/api/src/storage/adapters/sqlite.adapter.ts index 1465ca37e..25e29c77a 100644 --- a/apps/api/src/storage/adapters/sqlite.adapter.ts +++ b/apps/api/src/storage/adapters/sqlite.adapter.ts @@ -611,6 +611,11 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { if (!kpsInfo.some((col) => col.name === 'db')) { this.db.exec('ALTER TABLE key_pattern_snapshots ADD COLUMN db INTEGER NOT NULL DEFAULT 0'); } + // Backs the 60s-polled SELECT DISTINCT db query used to populate the + // database selector; the column must exist before the index can be created. + this.db.exec( + 'CREATE INDEX IF NOT EXISTS idx_kps_connection_db ON key_pattern_snapshots(connection_id, db)', + ); // Migrate connection_id to all data tables for multi-database support this.migrateConnectionId(); @@ -2665,7 +2670,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { const query = ` SELECT * FROM key_pattern_snapshots ${whereClause} - ORDER BY timestamp DESC + ORDER BY timestamp DESC, db ASC, pattern ASC LIMIT ? OFFSET ? `; @@ -2842,7 +2847,10 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { } // One collection writes one row per logical database; aggregate them per - // timestamp so an unfiltered trend stays a single series. + // timestamp so an unfiltered trend stays a single series. When no + // connectionId is given, also group by connection_id so trends from + // different connections aren't merged into one series. + const groupBy = connectionId ? 'timestamp' : 'timestamp, connection_id'; const query = ` SELECT timestamp, @@ -2851,7 +2859,7 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { SUM(stale_key_count) as stale_key_count FROM key_pattern_snapshots WHERE ${conditions.join(' AND ')} - GROUP BY timestamp + GROUP BY ${groupBy} ORDER BY timestamp ASC `; diff --git a/apps/web/src/hooks/useBaselineHotKeys.ts b/apps/web/src/hooks/useBaselineHotKeys.ts index 45309a6cc..35ba3e5eb 100644 --- a/apps/web/src/hooks/useBaselineHotKeys.ts +++ b/apps/web/src/hooks/useBaselineHotKeys.ts @@ -5,6 +5,7 @@ interface UseBaselineHotKeysOptions { connectionId?: string; startTime?: number; endTime?: number; + db?: number; enabled?: boolean; } @@ -12,16 +13,18 @@ export function useBaselineHotKeys({ connectionId, startTime, endTime, + db, enabled = true, }: UseBaselineHotKeysOptions) { return useQuery({ - queryKey: ['hot-keys-baseline', connectionId, startTime, endTime], + queryKey: ['hot-keys-baseline', connectionId, startTime, endTime, db], queryFn: async () => { const entries = await keyAnalyticsApi.getHotKeys({ limit: 50, startTime, endTime, oldest: true, + db, }); return entries.length > 0 ? entries : null; }, diff --git a/apps/web/src/pages/KeyAnalytics.tsx b/apps/web/src/pages/KeyAnalytics.tsx index 37416db29..b625323a2 100644 --- a/apps/web/src/pages/KeyAnalytics.tsx +++ b/apps/web/src/pages/KeyAnalytics.tsx @@ -124,11 +124,21 @@ export function KeyAnalytics() { >('patterns'); const [sortField, setSortField] = useState('keyCount'); const [sortDirection, setSortDirection] = useState('desc'); - const [selectedPattern, setSelectedPattern] = useState(null); + const [selectedPattern, setSelectedPattern] = useState<{ db: number; pattern: string } | null>( + null, + ); // Logical database filter: 'all' aggregates across every scanned database. const [rawSelectedDb, setSelectedDb] = useState('all'); const detailRef = useRef(null); + // Reset the db filter when the connection changes, so (a) the first fetches + // for a new connection don't go out with the old numeric filter while + // availableDbs is still loading for it, and (b) the filter doesn't silently + // snap back from 'all' if the remembered db happens to also exist there. + useEffect(() => { + setSelectedDb('all'); + }, [currentConnection?.id]); + // Databases that have analytics data for this connection. The filter is only // shown when there is more than one (or the single one is not db0). const { data: availableDbs } = usePolling({ @@ -206,6 +216,7 @@ export function KeyAnalytics() { connectionId: currentConnection?.id, startTime: hotKeyStartTime, endTime: hotKeyEndTime, + db: dbFilter, enabled: activeTab === 'hot-keys' && isHotKeyTimeFiltered, }); @@ -594,11 +605,12 @@ export function KeyAnalytics() { setSelectedPattern(pattern.pattern)} + onClick={() => setSelectedPattern({ db: pattern.db ?? 0, pattern: pattern.pattern })} > {pattern.pattern} @@ -643,7 +655,13 @@ export function KeyAnalytics() {
- Pattern Detail: {selectedPattern} + Pattern Detail:{' '} + {selectedPattern.pattern} + {(availableDbs?.length ?? 0) > 1 && ( + + db{selectedPattern.db} + + )}