Harden KV cache cleanup, cluster evictions, sample limits and scrape DNS

apps/api/src/common/utils/__tests__/outbound-url-guard.spec.ts

@@ -1,6 +1,6 @@
 import { BadRequestException } from '@nestjs/common';
 import { promises as dns } from 'dns';
-import { assertSafeOutboundUrl, BlockedAddressError, guardedLookup, isBlockedIp } from '../outbound-url-guard';
+import { assertSafeOutboundUrl, isBlockedIp } from '../outbound-url-guard';
 
 describe('outbound url guard', () => {
   const env = process.env.NODE_ENV;
@@ -160,40 +160,4 @@ describe('outbound url guard', () => {
     ] as any);
     await expect(assertSafeOutboundUrl('http://metrics.example.com/x', { label: 'metrics URL', allowPrivateNetworks: true })).rejects.toThrow('Metrics URL resolves to blocked IP address: fe80::1');
   });
-
-  describe('guardedLookup', () => {
-    const run = (lookup: ReturnType<typeof guardedLookup>, host: string, all: boolean) =>
-      new Promise<{ error: NodeJS.ErrnoException | null; address: unknown; family?: number }>((resolve) =>
-        lookup(host, { all } as any, (error: NodeJS.ErrnoException | null, address: unknown, family?: number) => resolve({ error, address, family })),
-      );
-
-    it('returns the validated address in both callback shapes', async () => {
-      jest.spyOn(dns, 'lookup').mockResolvedValue([{ address: '8.8.8.8', family: 4 }] as any);
-      const lookup = guardedLookup({ label: 'metrics URL' });
-      await expect(run(lookup, 'metrics.example.com', false)).resolves.toEqual({ error: null, address: '8.8.8.8', family: 4 });
-      await expect(run(lookup, 'metrics.example.com', true)).resolves.toEqual({ error: null, address: [{ address: '8.8.8.8', family: 4 }], family: undefined });
-    });
-
-    it('fails the connection when any answer is blocked', async () => {
-      jest.spyOn(dns, 'lookup').mockResolvedValue([
-        { address: '8.8.8.8', family: 4 },
-        { address: '169.254.169.254', family: 4 },
-      ] as any);
-      const { error } = await run(guardedLookup({ label: 'metrics URL', allowPrivateNetworks: true }), 'metrics.example.com', true);
-      expect(error).toBeInstanceOf(BlockedAddressError);
-      expect(error?.message).toBe('Metrics URL resolves to blocked IP address: 169.254.169.254');
-    });
-
-    it('applies the private network setting', async () => {
-      jest.spyOn(dns, 'lookup').mockResolvedValue([{ address: '10.1.2.3', family: 4 }] as any);
-      expect((await run(guardedLookup({ label: 'metrics URL' }), 'h', false)).error).toBeInstanceOf(BlockedAddressError);
-      expect((await run(guardedLookup({ label: 'metrics URL', allowPrivateNetworks: true }), 'h', false)).address).toBe('10.1.2.3');
-    });
-
-    it('passes DNS errors through', async () => {
-      const failure = Object.assign(new Error('getaddrinfo ENOTFOUND h'), { code: 'ENOTFOUND' });
-      jest.spyOn(dns, 'lookup').mockRejectedValue(failure);
-      expect((await run(guardedLookup({ label: 'metrics URL' }), 'h', false)).error).toBe(failure);
-    });
-  });
 });

apps/api/src/common/utils/outbound-url-guard.ts

@@ -1,15 +1,13 @@
 import { BadRequestException, Logger } from '@nestjs/common';
 import { promises as dns } from 'dns';
-import { isIP, type LookupFunction } from 'net';
+import { isIP } from 'net';
 
 export interface OutboundUrlOptions {
   label: string;
   allowPrivateNetworks?: boolean;
   resolveDns?: 'always' | 'production';
 }
 
-export class BlockedAddressError extends Error {}
-
 const logger = new Logger('OutboundUrlGuard');
 
 const BLOCKED_IP_PATTERNS = [
@@ -122,24 +120,3 @@ export async function assertSafeOutboundUrl(rawUrl: string, options: OutboundUrl
     throw new BadRequestException(`Invalid ${label}`);
   }
 }
-
-export function guardedLookup(options: Omit<OutboundUrlOptions, 'resolveDns'>): LookupFunction {
-  const { label, allowPrivateNetworks } = options;
-  return (hostname, lookupOptions, callback) => {
-    dns.lookup(hostname, { family: lookupOptions.family, hints: lookupOptions.hints, all: true, verbatim: true }).then(
-      (answers) => {
-        const blocked = answers.find(({ address }) => isBlockedIp(address, allowPrivateNetworks));
-        if (blocked) {
-          callback(new BlockedAddressError(`${capitalize(label)} resolves to blocked IP address: ${blocked.address}`), '', 0);
-        } else if (answers.length === 0) {
-          callback(Object.assign(new Error(`getaddrinfo ENOTFOUND ${hostname}`), { code: 'ENOTFOUND' }), '', 0);
-        } else if (lookupOptions.all) {
-          (callback as unknown as (error: null, addresses: typeof answers) => void)(null, answers);
-        } else {
-          callback(null, answers[0].address, answers[0].family);
-        }
-      },
-      (error: NodeJS.ErrnoException) => callback(error, '', 0),
-    );
-  };
-}

apps/web/src/pages/KvCache.tsx

@@ -110,6 +110,12 @@ export function KvCache() {
             recentBuckets={recent.data?.buckets ?? []}
           />
           {latest && <KvCacheAdvisories advisories={advisoriesFor(latest)} />}
+          {samples.data?.truncated && (
+            <p className="text-sm text-muted-foreground">
+              This range holds more samples than can be shown, so the charts and table cover only its most recent
+              part. Pick a shorter range to see all of it.
+            </p>
+          )}
           <KvCacheCharts buckets={buckets} history={chartHistory} engines={engines} />
           {latest && <KvCacheModelTable latest={latest} buckets={buckets} />}
         </>

docs/kv-cache-monitoring.md

@@ -41,6 +41,8 @@ that hold keys.
 Page controls:
 
 - A date range picker for the charts, defaulting to the last 6 hours.
+  A range covers at most 31 days and 50,000 stored samples. When a range holds more
+  samples, the page shows only the most recent ones and says so.
 - **Refresh**, which re-collects the footprint now.
 - The alert settings sheet (see [Per-connection switches](#per-connection-switches)).
 - **Rescan now** on the not-detected view, which collects the footprint again.
@@ -183,6 +185,11 @@ addresses as well. Link-local addresses (`169.254.0.0/16`, which
 includes cloud metadata endpoints, and `fe80::/10`) are always refused,
 whatever the setting.
 
+The guard checks every address a hostname resolves to, and the scrape then
+connects only to an address that passed the check. A DNS answer that changes
+between the check and the connection cannot redirect a scrape to a refused
+address.
+
 When you change the metrics URL of a linked engine to a different origin
 (scheme, host or port) and a stored auth header exists, BetterDB refuses the
 update with a `400` unless you also supply an auth header. Send an empty

packages/shared/src/types/kv-cache.types.ts

@@ -87,6 +87,7 @@ export interface KvCacheSampleBucket extends KvCacheEngineSample {
 export interface KvCacheSamplesResponse {
   buckets: KvCacheSampleBucket[];
   rangeHitRate: number | null;
+  truncated: boolean;
 }
 
 export interface KvCacheStatus {

proprietary/key-analytics/scan-nodes.ts

@@ -27,13 +27,24 @@ export function clusterScanNodes(
 }
 
 export async function primaryScanNodes(nodes: ScanNode[], logger: Logger, label: string): Promise<ScanNode[]> {
+  return (await readPrimaryScanNodes(nodes, logger, label)).nodes;
+}
+
+export async function readPrimaryScanNodes(
+  nodes: ScanNode[],
+  logger: Logger,
+  label: string,
+): Promise<{ nodes: ScanNode[]; unreadable: number }> {
   const roles = await Promise.allSettled(nodes.map((node) => node.client.getRole()));
-  return nodes.filter((node, index) => {
+  let unreadable = 0;
+  const primaries = nodes.filter((node, index) => {
     const role = roles[index];
     if (role.status === 'rejected') {
+      unreadable += 1;
       logger.warn(`${label} could not read the role of ${node.name}: ${role.reason instanceof Error ? role.reason.message : role.reason}`);
       return false;
     }
     return role.value.role === 'master';
   });
+  return { nodes: primaries, unreadable };
 }

proprietary/kv-cache/__tests__/footprint-estimate.spec.ts

@@ -180,18 +180,12 @@ describe('buildSnapshot', () => {
   });
 
   it('computes the eviction delta per node and treats a lower counter as a restart', () => {
-    expect(evictedKeysDelta(null, { a: 5 })).toBeNull();
-    expect(evictedKeysDelta({ a: 20 }, { a: 50 })).toBe(30);
-    expect(evictedKeysDelta({ a: 20 }, { a: 5 })).toBe(5);
-    expect(evictedKeysDelta({ a: 10, b: 1000 }, { a: 12 })).toBe(2);
-    expect(evictedKeysDelta({ a: 10 }, { a: 12, b: 1000 })).toBe(2);
-    expect(evictedKeysDelta({ a: 10 }, { b: 1000 })).toBeNull();
-  });
-
-  it('passes the eviction delta through to the snapshot', () => {
-    const node: NodeObservation = { scannedKeys: 0, matchedKeys: [], scanComplete: true, dbSize: 0, samples: [] };
-    expect(buildSnapshot({ ...base, nodes: [node] }).evictedKeysDelta).toBeNull();
-    expect(buildSnapshot({ ...base, nodes: [node], evictedKeysDelta: 7 }).evictedKeysDelta).toBe(7);
+    expect(evictedKeysDelta(undefined, new Map([['a', 5]]))).toBeNull();
+    expect(evictedKeysDelta(new Map([['a', 20]]), new Map([['a', 50]]))).toBe(30);
+    expect(evictedKeysDelta(new Map([['a', 20]]), new Map([['a', 5]]))).toBe(5);
+    expect(evictedKeysDelta(new Map([['a', 20], ['b', 1000]]), new Map([['a', 25]]))).toBe(5);
+    expect(evictedKeysDelta(new Map([['a', 20]]), new Map([['a', 20], ['b', 1000]]))).toBe(0);
+    expect(evictedKeysDelta(new Map([['a', 20]]), new Map([['b', 1000]]))).toBeNull();
   });
 
   it('detects through the GLIDE client even with no keys', () => {
@@ -208,4 +202,16 @@ describe('buildSnapshot', () => {
     const s = buildSnapshot({ ...base, nodes: [n1, n2] });
     expect(s).toMatchObject({ scannedKeys: 10, matchedKeys: 3, chunksEst: 3, bytesEst: 50 });
   });
+
+  it('weights each node sample by the keys it stands for when splitting per model', () => {
+    const a = Array.from({ length: 100 }, (_, i) => key(i));
+    const b = key(0, '', 'other', 'half');
+    const n1: NodeObservation = { scannedKeys: 100, matchedKeys: parse(a), scanComplete: true, dbSize: 100, samples: [sample(a[0], { memoryBytes: 100 })] };
+    const n2: NodeObservation = { scannedKeys: 1, matchedKeys: parse([b]), scanComplete: true, dbSize: 1, samples: [sample(b, { memoryBytes: 100 })] };
+    const s = buildSnapshot({ ...base, nodes: [n1, n2] });
+    expect(s.perModel).toEqual([
+      { model: M, dtype: 'bfloat16', chunksEst: 100, bytesEst: 10_000 },
+      { model: 'other', dtype: 'half', chunksEst: 1, bytesEst: 100 },
+    ]);
+  });
 });

proprietary/kv-cache/__tests__/footprint-scanner.spec.ts

@@ -37,19 +37,54 @@ describe('scanLmcacheKeys', () => {
     expect(client.call).toHaveBeenCalledTimes(2);
     expect(result.scanComplete).toBe(false);
   });
+
+  it('counts glob hits that fail the classifier toward the matched budget', async () => {
+    const lookalikes = (n: number) => Array.from({ length: n }, (_, i) => `user@tenant@region@shard@${i}`);
+    const client = scanClient([['7', lookalikes(3)], ['8', lookalikes(3)], ['9', lookalikes(3)]]);
+    const result = await scanLmcacheKeys(client, { maxScanned: 200_000, maxMatched: 5 }, 10_000);
+    expect(client.call).toHaveBeenCalledTimes(2);
+    expect(result).toEqual({ scannedKeys: 2000, matchedKeys: [], scanComplete: false });
+  });
+
+  it('gives up after the probe pages when nothing matches the glob', async () => {
+    const pages: Array<[string, string[]]> = [['7', []], ['8', []], ['0', [k(0)]]];
+    const probing = scanClient(pages);
+    await scanLmcacheKeys(probing, { maxScanned: 200_000, maxMatched: 2000 }, 10_000, true);
+    expect(probing.call).toHaveBeenCalledTimes(2);
+
+    const full = scanClient(pages);
+    const result = await scanLmcacheKeys(full, { maxScanned: 200_000, maxMatched: 2000 }, 10_000);
+    expect(result.matchedKeys).toEqual(p(k(0)));
+  });
+
+  it('keeps scanning past the probe pages once the glob has hits', async () => {
+    const client = scanClient([['7', ['a@b@c@d@e']], ['8', []], ['0', [k(0)]]]);
+    const result = await scanLmcacheKeys(client, { maxScanned: 200_000, maxMatched: 2000 }, 3000, true);
+    expect(result.matchedKeys).toEqual(p(k(0)));
+  });
 });
 
+function sampleClient(replies: Record<string, unknown | Error>) {
+  const call = jest.fn(async (command: string, args: string[]) => {
+    const reply = replies[`${command} ${args.join(' ')}`];
+    if (reply instanceof Error) throw reply;
+    return reply;
+  });
+  return { call, getClient: () => { throw new Error('Raw client access not available through agent connections'); } } as any;
+}
+
 describe('sampleLmcacheKeys', () => {
-  it('reads memory, ttl and sibling existence in one pipeline', async () => {
-    const pipeline = { call: jest.fn().mockReturnThis(), ttl: jest.fn().mockReturnThis(), exists: jest.fn().mockReturnThis(), exec: jest.fn() };
-    pipeline.exec.mockResolvedValue([
-      [null, 3670112], [null, -1], [null, 0],
-      [null, 120], [null, 50],
-      [null, null], [null, -2],
-    ]);
-    const client = { getClient: () => ({ pipeline: () => pipeline }) } as any;
+  it('reads memory, ttl and sibling existence through call only', async () => {
+    const client = sampleClient({
+      [`MEMORY USAGE ${k(0, 'kv_bytes')}`]: 3670112,
+      [`TTL ${k(0, 'kv_bytes')}`]: -1,
+      [`EXISTS ${k(0, 'metadata')}`]: 0,
+      [`MEMORY USAGE ${k(1)}`]: 120,
+      [`TTL ${k(1)}`]: 50,
+      [`MEMORY USAGE ${k(2)}`]: null,
+      [`TTL ${k(2)}`]: -2,
+    });
     const samples = await sampleLmcacheKeys(client, p(k(0, 'kv_bytes'), k(1), k(2)));
-    expect(pipeline.exists).toHaveBeenCalledWith(k(0, 'metadata'));
     expect(samples.map(({ memoryBytes, ttl, siblingExists }) => ({ memoryBytes, ttl, siblingExists }))).toEqual([
       { memoryBytes: 3670112, ttl: -1, siblingExists: false },
       { memoryBytes: 120, ttl: 50, siblingExists: null },
@@ -58,19 +93,53 @@ describe('sampleLmcacheKeys', () => {
   });
 
   it('treats a cross-slot sibling probe rejected with MOVED as unknown, not orphaned', async () => {
-    const pipeline = { call: jest.fn().mockReturnThis(), ttl: jest.fn().mockReturnThis(), exists: jest.fn().mockReturnThis(), exec: jest.fn() };
-    pipeline.exec.mockResolvedValue([
-      [null, 3670112], [null, -1], [new Error('MOVED 1234 10.0.0.2:6379'), null],
-      [null, 3670112], [null, -1], [null, 1],
-      [null, 3670112], [null, -1], [null, 0],
-    ]);
-    const client = { getClient: () => ({ pipeline: () => pipeline }) } as any;
-    const samples = await sampleLmcacheKeys(client, p(k(0, 'kv_bytes'), k(1, 'kv_bytes'), k(2, 'kv_bytes')));
-    expect(samples.map((s) => s.siblingExists)).toEqual([null, true, false]);
+    const client = sampleClient({
+      [`MEMORY USAGE ${k(0, 'kv_bytes')}`]: 3670112,
+      [`TTL ${k(0, 'kv_bytes')}`]: -1,
+      [`EXISTS ${k(0, 'metadata')}`]: new Error('MOVED 1234 10.0.0.2:6379'),
+      [`MEMORY USAGE ${k(1, 'kv_bytes')}`]: 3670112,
+      [`TTL ${k(1, 'kv_bytes')}`]: -1,
+      [`EXISTS ${k(1, 'metadata')}`]: 1,
+    });
+    const samples = await sampleLmcacheKeys(client, p(k(0, 'kv_bytes'), k(1, 'kv_bytes')));
+    expect(samples.map((s) => s.siblingExists)).toEqual([null, true]);
+  });
+
+  it('reports a key as unsized when MEMORY is unavailable', async () => {
+    const client = sampleClient({
+      [`MEMORY USAGE ${k(0)}`]: new Error("ERR unknown command 'MEMORY'"),
+      [`TTL ${k(0)}`]: -1,
+    });
+    const [sample] = await sampleLmcacheKeys(client, p(k(0)));
+    expect(sample.memoryBytes).toBeNull();
+  });
+
+  it('fails the read when every command for a key is rejected', async () => {
+    const closed = new Error('Connection is closed.');
+    const client = sampleClient({
+      [`MEMORY USAGE ${k(0)}`]: 120,
+      [`TTL ${k(0)}`]: 50,
+      [`MEMORY USAGE ${k(1, 'kv_bytes')}`]: closed,
+      [`TTL ${k(1, 'kv_bytes')}`]: closed,
+      [`EXISTS ${k(1, 'metadata')}`]: closed,
+    });
+    await expect(sampleLmcacheKeys(client, p(k(0), k(1, 'kv_bytes')))).rejects.toThrow('Connection is closed.');
+  });
+
+  it('keeps the ttl of a key whose MEMORY and sibling probes are rejected', async () => {
+    const client = sampleClient({
+      [`MEMORY USAGE ${k(0, 'kv_bytes')}`]: new Error("ERR unknown command 'MEMORY'"),
+      [`TTL ${k(0, 'kv_bytes')}`]: 30,
+      [`EXISTS ${k(0, 'metadata')}`]: new Error('MOVED 1234 10.0.0.2:6379'),
+    });
+    const [sample] = await sampleLmcacheKeys(client, p(k(0, 'kv_bytes')));
+    expect(sample).toMatchObject({ memoryBytes: null, ttl: 30, siblingExists: null });
   });
 
-  it('skips the pipeline for no keys', async () => {
-    expect(await sampleLmcacheKeys({ getClient: () => { throw new Error('unused'); } } as any, [])).toEqual([]);
+  it('issues no commands for no keys', async () => {
+    const client = sampleClient({});
+    expect(await sampleLmcacheKeys(client, [])).toEqual([]);
+    expect(client.call).not.toHaveBeenCalled();
   });
 });
 
@@ -81,6 +150,11 @@ describe('helpers', () => {
     expect(await hasLmcacheGlideClient({ call: jest.fn().mockRejectedValue(new Error('NOPERM')) } as any)).toBe(false);
   });
 
+  it('ignores a client name that imitates the GLIDE library field', async () => {
+    const list = 'id=1 name=lib-name=GlidePySync(lmcache:0.5.5) lib-name=redis-py lib-ver=5.0\n';
+    expect(await hasLmcacheGlideClient({ call: jest.fn().mockResolvedValue(list) } as any)).toBe(false);
+  });
+
   it('reads memory stats and other databases', () => {
     const info = {
       memory: { used_memory: '100', maxmemory: '200', maxmemory_policy: 'volatile-lru' },

proprietary/kv-cache/__tests__/kv-cache-alerts.service.spec.ts

@@ -135,6 +135,33 @@ describe('KvCacheAlertsService eviction risk', () => {
     expect(evictingStates()).toEqual([true, true, true, true, true, false, false]);
   });
 
+  it('keeps the evicting state when a snapshot only read some nodes', async () => {
+    const { service, pro } = setup();
+    const evictingStates = () =>
+      pro.dispatchKvCacheEvictionRisk.mock.calls.filter(([e]) => e.reason === 'evicting').map(([e]) => e.active);
+    await service.onSnapshot(snapshot({ evictedKeysDelta: 50 }), { partial: true });
+    await service.onSnapshot(snapshot({ evictedKeysDelta: 50 }));
+    for (let i = 0; i < 3; i += 1) await service.onSnapshot(snapshot({ evictedKeysDelta: 0 }), { partial: true });
+    await service.onSnapshot(snapshot({ evictedKeysDelta: 0 }));
+    expect(evictingStates()).toEqual([false, true, true, true, true, true]);
+  });
+
+  it('keeps the evicting state when a partial snapshot does not detect LMCache', async () => {
+    const { service, pro } = setup();
+    await service.onSnapshot(snapshot({ evictedKeysDelta: 50 }));
+    await service.onSnapshot(snapshot({ detected: false, evictedKeysDelta: 0 }), { partial: true });
+    expect(pro.dispatchKvCacheEvictionRisk).toHaveBeenLastCalledWith(expect.objectContaining({ reason: 'evicting', active: true }));
+  });
+
+  it('passes the partial flag from footprint snapshots through', async () => {
+    const { service, pro, footprint } = setup();
+    service.onModuleInit();
+    const listener = footprint.onSnapshot.mock.calls[0][0];
+    await listener(snapshot({ evictedKeysDelta: 50 }), { partial: true });
+    expect(pro.dispatchKvCacheEvictionRisk).toHaveBeenCalledWith(expect.objectContaining({ reason: 'evicting', active: false }));
+    service.onModuleDestroy();
+  });
+
   it('forgets the evicting hold when the connection is removed', async () => {
     const { service, pro, remove } = setup();
     service.onModuleInit();