Monitor LMCache KV cache footprint in Valkey (Pro)

.github/workflows/api-tests.yml

@@ -240,6 +240,8 @@ jobs:
 
       - name: Run API tests
         working-directory: apps/api
+        env:
+          KV_CACHE_IT_VALKEY_PORT: '6390'
         run: pnpm test
 
       - name: Run entitlement unit tests

apps/api/src/__tests__/kv-cache-feature.spec.ts

@@ -0,0 +1,13 @@
+import { Feature, TIER_FEATURES, Tier, DEFAULT_KV_CACHE_HIT_RATE_THRESHOLD } from '@betterdb/shared';
+
+describe('kv cache feature', () => {
+  it('is a Pro feature', () => {
+    expect(Feature.KV_CACHE_MONITORING).toBe('kvCacheMonitoring');
+    expect(TIER_FEATURES[Tier.pro]).toContain(Feature.KV_CACHE_MONITORING);
+    expect(TIER_FEATURES[Tier.community]).not.toContain(Feature.KV_CACHE_MONITORING);
+  });
+
+  it('defaults the hit rate threshold to 0.2', () => {
+    expect(DEFAULT_KV_CACHE_HIT_RATE_THRESHOLD).toBe(0.2);
+  });
+});

apps/api/src/app.module.ts

@@ -47,6 +47,7 @@ let LatencyRegressionModule: any = null;
 let WebhookProModule: any = null;
 let InferenceLatencyProModule: any = null;
 let ScalingReadinessProModule: any = null;
+let KvCacheModule: any = null;
 let CacheProposalsModule: any = null;
 let MemoryProposalsModule: any = null;
 let AgentModule: any = null;
@@ -126,6 +127,14 @@ try {
   // Proprietary module not available
 }
 
+try {
+  const kvCacheModule = require('../../../proprietary/kv-cache/kv-cache.module');
+  KvCacheModule = kvCacheModule.KvCacheModule;
+  console.log('[KvCache] Proprietary module loaded');
+} catch {
+  // Proprietary module not available
+}
+
 try {
   const cacheProposalsModule = require('../../../proprietary/cache-proposals/cache-proposals.module');
   CacheProposalsModule = cacheProposalsModule.CacheProposalsModule;
@@ -241,6 +250,7 @@ const proprietaryImports = [
   WebhookProModule,
   InferenceLatencyProModule,
   ScalingReadinessProModule,
+  KvCacheModule,
   CacheProposalsModule,
   MemoryProposalsModule,
   AiModule,

apps/api/src/common/interfaces/database-port.interface.ts

@@ -54,6 +54,7 @@ export interface DatabasePort {
   disconnect(): Promise<void>;
   isConnected(): boolean;
   sampleVersion?(): number | null;
+  getActiveDbIndex?(): number;
   ping(): Promise<boolean>;
   getInfo(sections?: string[]): Promise<Record<string, unknown>>;
   getCapabilities(): DatabaseCapabilities;

apps/api/src/common/interfaces/storage-port.interface.ts

@@ -39,6 +39,10 @@ import type {
   ReadinessDimension,
   ReadinessDimensionKey,
   ScalingReadinessSettings,
+  KvCacheEngineSample,
+  KvCacheEngineSource,
+  KvCacheFootprintSnapshot,
+  KvCacheSettings,
 } from '@betterdb/shared';
 export type { MetricForecastSettings, MetricKind } from '@betterdb/shared';
 export type { StoredCveDataset, CveScanResult } from '@betterdb/shared';
@@ -403,6 +407,37 @@ export interface ScalingReadinessScoreQuery {
   limit?: number;
 }
 
+export interface StoredKvCacheEngine {
+  id: string;
+  connectionId: string;
+  name: string;
+  source: KvCacheEngineSource;
+  scrapeUrl: string | null;
+  scrapeAuthHeader: string | null;
+  scrapeAuthEncrypted: boolean;
+  otlpEngineId: string | null;
+  enabled: boolean;
+  createdAt: number;
+  lastSeenAt: number | null;
+  lastError: string | null;
+}
+
+export interface KvCacheRangeQuery {
+  connectionId: string;
+  from?: number;
+  to?: number;
+  limit?: number;
+}
+
+export interface KvCacheFootprintQuery extends KvCacheRangeQuery {
+  bucketMs?: number;
+}
+
+export interface KvCacheSampleQuery extends KvCacheRangeQuery {
+  engineId?: string;
+  modelName?: string;
+}
+
 export interface MemorySnapshotQueryOptions {
   connectionId?: string;
   startTime?: number;
@@ -615,6 +650,19 @@ export interface StoragePort {
   pruneOldScalingReadinessScores(cutoffTimestamp: number, connectionId?: string): Promise<number>;
   getScalingReadinessSettings(connectionId: string): Promise<ScalingReadinessSettings | null>;
   saveScalingReadinessSettings(settings: ScalingReadinessSettings): Promise<ScalingReadinessSettings>;
+  saveKvCacheFootprintSnapshot(snapshot: KvCacheFootprintSnapshot): Promise<void>;
+  getKvCacheFootprintSnapshots(query: KvCacheFootprintQuery): Promise<KvCacheFootprintSnapshot[]>;
+  pruneOldKvCacheFootprintSnapshots(cutoffTimestamp: number, connectionId?: string): Promise<number>;
+  saveKvCacheEngineSamples(samples: KvCacheEngineSample[]): Promise<void>;
+  getKvCacheEngineSamples(query: KvCacheSampleQuery): Promise<KvCacheEngineSample[]>;
+  pruneOldKvCacheEngineSamples(cutoffTimestamp: number, connectionId?: string): Promise<number>;
+  saveKvCacheEngine(engine: StoredKvCacheEngine): Promise<StoredKvCacheEngine>;
+  getKvCacheEngines(connectionId?: string): Promise<StoredKvCacheEngine[]>;
+  getKvCacheEngine(id: string): Promise<StoredKvCacheEngine | null>;
+  deleteKvCacheEngine(id: string): Promise<boolean>;
+  getKvCacheSettings(connectionId: string): Promise<KvCacheSettings | null>;
+  saveKvCacheSettings(settings: KvCacheSettings): Promise<KvCacheSettings>;
+  deleteKvCacheConnectionData(connectionId: string): Promise<void>;
 
   // Command Stats Sample Methods - connectionId required for writes
   saveCommandStatsSamples(

apps/api/src/config/env.schema.kv-cache.spec.ts

@@ -0,0 +1,19 @@
+import { envSchema } from './env.schema';
+
+const KV_CACHE_FOOTPRINT_VARS = [
+  'KV_CACHE_FOOTPRINT_INTERVAL_MS',
+  'KV_CACHE_SCAN_MAX_KEYS',
+  'KV_CACHE_MATCH_MAX_KEYS',
+  'KV_CACHE_SAMPLE_KEYS',
+  'KV_CACHE_UNDETECTED_RECHECK_MS',
+] as const;
+
+describe('KV cache footprint env vars', () => {
+  it.each(KV_CACHE_FOOTPRINT_VARS)('registers %s', (name) => {
+    expect((envSchema.parse({ [name]: '600000' }) as Record<string, unknown>)[name]).toBe('600000');
+  });
+
+  it.each(KV_CACHE_FOOTPRINT_VARS)('leaves an invalid %s to the service fallback instead of failing startup', (name) => {
+    expect(envSchema.safeParse({ [name]: 'abc' }).success).toBe(true);
+  });
+});

apps/api/src/config/env.schema.ts

@@ -163,6 +163,12 @@ export const envSchema = z
     ANOMALY_CACHE_TTL_MS: z.coerce.number().int().min(1000).default(3600000),
     ANOMALY_PROMETHEUS_INTERVAL_MS: z.coerce.number().int().min(1000).default(30000),
 
+    KV_CACHE_FOOTPRINT_INTERVAL_MS: z.string().optional(),
+    KV_CACHE_SCAN_MAX_KEYS: z.string().optional(),
+    KV_CACHE_MATCH_MAX_KEYS: z.string().optional(),
+    KV_CACHE_SAMPLE_KEYS: z.string().optional(),
+    KV_CACHE_UNDETECTED_RECHECK_MS: z.string().optional(),
+
     // License configuration (optional)
     BETTERDB_LICENSE_KEY: z.string().optional(),
     // Offline (air-gapped) license: the signed token itself, or a path to it

apps/api/src/connections/__tests__/connection-registry.membership.spec.ts

@@ -635,4 +635,35 @@ describe('ConnectionRegistry membership', () => {
     await registry.retireChild('adopted');
     expect(registry.getDefaultId()).toBe('seed');
   });
+
+  it('notifies removal listeners after a connection and its auto children are deleted', async () => {
+    const { registry, storage } = build();
+    put(registry, seed);
+    put(registry, { id: 'child', membership: { seedId: 'seed', nodeId: 'c', origin: 'auto', source: 'cluster' } });
+    const removed: string[] = [];
+    registry.addConnectionRemovalListener((id) => {
+      expect(storage.deleteConnection).toHaveBeenCalledWith(id);
+      removed.push(id);
+    });
+    const unsubscribe = registry.addConnectionRemovalListener(() => {
+      throw new Error('boom');
+    });
+    await registry.removeConnection('seed');
+    expect(removed).toEqual(['child', 'seed']);
+    unsubscribe();
+  });
+
+  it('logs a removal listener that rejects asynchronously', async () => {
+    const { registry } = build();
+    put(registry, seed);
+    const warn = jest
+      .spyOn((registry as unknown as { logger: { warn: (message: string) => void } }).logger, 'warn')
+      .mockImplementation(() => undefined);
+    registry.addConnectionRemovalListener(async () => {
+      throw new Error('async boom');
+    });
+    await registry.removeConnection('seed');
+    await new Promise((resolve) => setImmediate(resolve));
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining('async boom'));
+  });
 });

apps/api/src/connections/connection-registry.service.ts

@@ -20,6 +20,8 @@ import { ENV_DEFAULT_ID } from './connection.constants';
 
 export const CHILD_CONNECT_TIMEOUT_MS = 10_000;
 
+export type ConnectionRemovalListener = (id: string) => void | Promise<void>;
+
 function certificateHint(errorMsg: string, node: DesiredNode): string {
   if (node.hostname !== undefined || !/altnames|ERR_TLS_CERT_ALTNAME_INVALID/i.test(errorMsg)) {
     return '';
@@ -91,6 +93,7 @@ export class ConnectionRegistry implements OnModuleInit, OnModuleDestroy {
   private readonly encryption: EnvelopeEncryptionService | null;
   private startupConnectionErrors: Array<{ name: string; host: string; port: number; error: string }> = [];
   private readonly seedLocks = new Map<string, Promise<unknown>>();
+  private readonly removalListeners = new Set<ConnectionRemovalListener>();
 
   constructor(
     @Inject('STORAGE_CLIENT') private readonly storage: StoragePort,
@@ -709,6 +712,27 @@ export class ConnectionRegistry implements OnModuleInit, OnModuleDestroy {
     this.discoveredInstances?.forgetPushed(discoveredHost, discoveredPort, Date.now());
   }
 
+  addConnectionRemovalListener(listener: ConnectionRemovalListener): () => void {
+    this.removalListeners.add(listener);
+    return () => {
+      this.removalListeners.delete(listener);
+    };
+  }
+
+  private notifyRemoved(id: string): void {
+    const report = (error: unknown) => {
+      const reason = error instanceof Error ? error.message : String(error);
+      this.logger.warn(`Connection removal listener failed for ${id}: ${reason}`);
+    };
+    for (const listener of this.removalListeners) {
+      try {
+        void Promise.resolve(listener(id)).catch(report);
+      } catch (error: unknown) {
+        report(error);
+      }
+    }
+  }
+
   async removeConnection(id: string): Promise<void> {
     if (id === ENV_DEFAULT_ID) {
       throw new Error('Cannot remove the default environment connection');
@@ -751,6 +775,7 @@ export class ConnectionRegistry implements OnModuleInit, OnModuleDestroy {
     this.runtimeCapabilityTracker.removeConnection(id);
     this.externalMetricsStore.clear(id);
     await this.storage.deleteConnection(id);
+    this.notifyRemoved(id);
 
     if (removedConfig) {
       this.usageTelemetry?.trackDbRemove({

apps/api/src/database/adapters/unified.adapter.ts

@@ -7,6 +7,7 @@ import {
 } from '../../common/interfaces/database-port.interface';
 import { InfoParser } from '../parsers/info.parser';
 import { MetricsParser } from '../parsers/metrics.parser';
+import { parseKeyspaceDatabases, type KeyspaceDatabase } from '../parsers/keyspace-databases';
 import { CLUSTER_TOTAL_SLOTS } from '../../common/constants/cluster.constants';
 import {
   InfoResponse,
@@ -884,22 +885,13 @@ export class UnifiedDatabaseAdapter implements DatabasePort {
    * proxies that hide it), in which case the caller falls back to scanning
    * the connection's own database.
    */
-  private async listKeyspaceDatabases(): Promise<Array<{ db: number; keys: number }>> {
+  private async listKeyspaceDatabases(): Promise<KeyspaceDatabase[]> {
     try {
       // Reuse the centralized INFO keyspace parser (MetricsParser via
       // getInfoParsed) rather than a second regex: it is the single definition
       // of a db entry and already drops malformed `db<N>:` lines as raw strings.
       const { keyspace } = await this.getInfoParsed(['keyspace']);
-      const dbs: Array<{ db: number; keys: number }> = [];
-      for (const [key, info] of Object.entries(keyspace ?? {})) {
-        const match = /^db(\d+)$/.exec(key);
-        // A raw-string value is a malformed line the parser kept verbatim; only
-        // typed entries with a positive key count are non-empty databases.
-        if (match && typeof info === 'object' && info.keys > 0) {
-          dbs.push({ db: parseInt(match[1], 10), keys: info.keys });
-        }
-      }
-      return dbs.sort((a, b) => a.db - b.db);
+      return parseKeyspaceDatabases(keyspace);
     } catch (error) {
       this.logger.debug(
         `Could not list keyspace databases: ${error instanceof Error ? error.message : error}`,

apps/api/src/database/parsers/keyspace-databases.spec.ts

@@ -0,0 +1,22 @@
+import { parseKeyspaceDatabases } from './keyspace-databases';
+
+describe('parseKeyspaceDatabases', () => {
+  it('lists non-empty databases sorted by index', () => {
+    expect(
+      parseKeyspaceDatabases({
+        db3: { keys: 2, expires: 0, avg_ttl: 0 },
+        db0: { keys: 5, expires: 0, avg_ttl: 0 },
+        db4: { keys: 0, expires: 0, avg_ttl: 0 },
+        junk: 'x',
+        db7: 'malformed',
+      }),
+    ).toEqual([
+      { db: 0, keys: 5 },
+      { db: 3, keys: 2 },
+    ]);
+  });
+
+  it('returns an empty list without a keyspace section', () => {
+    expect(parseKeyspaceDatabases(undefined)).toEqual([]);
+  });
+});

apps/api/src/database/parsers/keyspace-databases.ts

@@ -0,0 +1,17 @@
+import type { KeyspaceInfo } from '../../common/types/metrics.types';
+
+export interface KeyspaceDatabase {
+  db: number;
+  keys: number;
+}
+
+export function parseKeyspaceDatabases(keyspace: KeyspaceInfo | undefined): KeyspaceDatabase[] {
+  const dbs: KeyspaceDatabase[] = [];
+  for (const [key, info] of Object.entries(keyspace ?? {})) {
+    const match = /^db(\d+)$/.exec(key);
+    if (match && typeof info === 'object' && info.keys > 0) {
+      dbs.push({ db: parseInt(match[1], 10), keys: info.keys });
+    }
+  }
+  return dbs.sort((a, b) => a.db - b.db);
+}

apps/api/src/retention/__tests__/local-retention.service.spec.ts

@@ -26,6 +26,8 @@ describe('LocalRetentionService', () => {
       pruneOldLatencyHistograms: jest.fn().mockResolvedValue(0),
       pruneOldMemorySnapshots: jest.fn().mockResolvedValue(0),
       pruneOldScalingReadinessScores: jest.fn().mockResolvedValue(3),
+      pruneOldKvCacheFootprintSnapshots: jest.fn().mockResolvedValue(4),
+      pruneOldKvCacheEngineSamples: jest.fn().mockResolvedValue(5),
       pruneOldCaptureChunks: jest.fn().mockResolvedValue(0),
       pruneOldCaptureSessions: jest.fn().mockResolvedValue(0),
       pruneOldCaptureTriggers: jest.fn().mockResolvedValue(0),
@@ -78,6 +80,15 @@ describe('LocalRetentionService', () => {
     expect(results.scaling_readiness_scores).toBe(3);
   });
 
+  it('reports pruned kv cache rows under their own names', async () => {
+    const results = await runRetentionSweep(storage, 1_234, { error: jest.fn() } as any);
+
+    expect(storage.pruneOldKvCacheFootprintSnapshots).toHaveBeenCalledWith(1_234);
+    expect(storage.pruneOldKvCacheEngineSamples).toHaveBeenCalledWith(1_234);
+    expect(results.kv_cache_footprint_snapshots).toBe(4);
+    expect(results.kv_cache_engine_samples).toBe(5);
+  });
+
   it('does nothing in cloud mode even when a window is configured', async () => {
     process.env.CLOUD_MODE = 'true';
     await makeService(30).runSweep();

apps/api/src/retention/retention-sweep.ts

@@ -7,9 +7,9 @@ import { StoragePort } from '../common/interfaces/storage-port.interface';
  * self-hosted local sweep so the two never drift on which tables are covered.
  *
  * `sampleCutoff` applies to the high-volume sample stores (command/latency
- * stats samples, vector index snapshots), which in cloud keep a tighter 7-day
- * cap than the tier's analytics window. Self-hosted passes one cutoff for
- * everything, so it defaults to `cutoff`.
+ * stats samples, vector index snapshots, KV cache engine samples), which in
+ * cloud keep a tighter 7-day cap than the tier's analytics window.
+ * Self-hosted passes one cutoff for everything, so it defaults to `cutoff`.
  *
  * A store that fails to prune is reported as -1 and does not stop the sweep.
  */
@@ -33,6 +33,8 @@ export async function runRetentionSweep(
     { name: 'latency_histograms', fn: () => storage.pruneOldLatencyHistograms(cutoff) },
     { name: 'memory_snapshots', fn: () => storage.pruneOldMemorySnapshots(cutoff) },
     { name: 'scaling_readiness_scores', fn: () => storage.pruneOldScalingReadinessScores(cutoff) },
+    { name: 'kv_cache_footprint_snapshots', fn: () => storage.pruneOldKvCacheFootprintSnapshots(cutoff) },
+    { name: 'kv_cache_engine_samples', fn: () => storage.pruneOldKvCacheEngineSamples(sampleCutoff) },
     { name: 'capture_chunks', fn: () => storage.pruneOldCaptureChunks(cutoff) },
     { name: 'capture_sessions', fn: () => storage.pruneOldCaptureSessions(cutoff) },
     { name: 'capture_triggers', fn: () => storage.pruneOldCaptureTriggers(cutoff) },