fix(migration): count keys on every cluster master during validation

apps/api/src/migration/__tests__/key-count-comparator.spec.ts

@@ -0,0 +1,52 @@
+import Valkey, { Cluster } from 'iovalkey';
+import { compareKeyCounts, countKeys } from '../validation/key-count-comparator';
+
+function standalone(keys: number): Valkey {
+  return { dbsize: jest.fn().mockResolvedValue(keys) } as unknown as Valkey;
+}
+
+function cluster(perMaster: number[]): Valkey {
+  const client = Object.create(Cluster.prototype) as Cluster & { dbsize: jest.Mock };
+  // A bare DBSIZE on a Cluster client hits one random node; make that visible.
+  client.dbsize = jest.fn().mockResolvedValue(perMaster[0]);
+  client.nodes = jest
+    .fn()
+    .mockReturnValue(
+      perMaster.map((keys) => ({ dbsize: jest.fn().mockResolvedValue(keys) })),
+    ) as unknown as Cluster['nodes'];
+  return client as unknown as Valkey;
+}
+
+describe('key-count-comparator', () => {
+  it('sums DBSIZE over every master of a cluster client', async () => {
+    const client = cluster([1008, 988, 1004]);
+
+    expect(await countKeys(client)).toBe(3000);
+    expect((client as unknown as Cluster).nodes).toHaveBeenCalledWith('master');
+  });
+
+  it('uses DBSIZE directly for a standalone client', async () => {
+    expect(await countKeys(standalone(42))).toBe(42);
+  });
+
+  it('fails loudly when a cluster client has no masters', async () => {
+    await expect(countKeys(cluster([]))).rejects.toThrow('No cluster masters');
+  });
+
+  it('reports no discrepancy for a complete standalone -> cluster migration', async () => {
+    const result = await compareKeyCounts(standalone(3000), cluster([1008, 988, 1004]));
+
+    expect(result).toMatchObject({
+      sourceKeys: 3000,
+      targetKeys: 3000,
+      discrepancy: 0,
+      discrepancyPercent: 0,
+    });
+  });
+
+  it('reports no discrepancy for a complete cluster -> standalone migration', async () => {
+    const result = await compareKeyCounts(cluster([1008, 988, 1004]), standalone(3000));
+
+    expect(result.discrepancyPercent).toBe(0);
+  });
+});

apps/api/src/migration/__tests__/migration-validation.service.spec.ts

@@ -162,6 +162,31 @@ describe('MigrationValidationService', () => {
       expect(clusterRegistry.mockAdapter.getInfo).toHaveBeenCalledWith(['cluster']);
     });
 
+    it('counts a cluster source through a cluster client, not the seed node alone', async () => {
+      const { compareKeyCounts } = jest.requireMock('../validation/key-count-comparator');
+      const { Cluster } = jest.requireMock('iovalkey');
+      compareKeyCounts.mockClear();
+      Cluster.mockClear();
+
+      const sourceAdapter = {
+        ...registry.mockAdapter,
+        getInfo: jest.fn().mockResolvedValue({ cluster: { cluster_enabled: '1' } }),
+      };
+      registry.get.mockImplementation((id: string) => (id === 'conn-1' ? sourceAdapter : registry.mockAdapter));
+
+      const { id } = await service.startValidation({
+        sourceConnectionId: 'conn-1',
+        targetConnectionId: 'conn-2',
+      });
+      await new Promise(r => setTimeout(r, 100));
+
+      expect(service.getValidation(id)!.status).toBe('completed');
+      expect(Cluster).toHaveBeenCalledTimes(1);
+      const sourceCountClient = Cluster.mock.results[0].value;
+      expect(compareKeyCounts).toHaveBeenCalledWith(sourceCountClient, expect.anything(), undefined);
+      expect(sourceCountClient.quit).toHaveBeenCalled();
+    });
+
     it('should use Phase 1 analysis result when analysisId provided', async () => {
       migrationService.getJob.mockReturnValue({
         status: 'completed',

apps/api/src/migration/migration-validation.service.ts

@@ -83,8 +83,9 @@ export class MigrationValidationService {
     this.jobs.set(id, job);
 
     // 6. Fire and forget
+    const sourceAdapter = this.connectionRegistry.get(req.sourceConnectionId);
     const targetAdapter = this.connectionRegistry.get(req.targetConnectionId);
-    this.runValidation(job, sourceConfig, targetConfig, targetAdapter, req.migrationStartedAt, analysisResult).catch(err => {
+    this.runValidation(job, sourceConfig, targetConfig, sourceAdapter, targetAdapter, req.migrationStartedAt, analysisResult).catch(err => {
       this.logger.error(`Validation ${id} failed: ${err.message}`);
     });
 
@@ -95,20 +96,25 @@ export class MigrationValidationService {
     job: ValidationJob,
     sourceConfig: DatabaseConnectionConfig,
     targetConfig: DatabaseConnectionConfig,
+    sourceAdapter: import('../common/interfaces/database-port.interface').DatabasePort,
     targetAdapter: import('../common/interfaces/database-port.interface').DatabasePort,
     migrationStartedAt?: number,
     analysisResult?: MigrationAnalysisResult,
   ): Promise<void> {
     let sourceClient: Valkey | null = null;
+    let sourceCountClient: Valkey | null = null;
     let targetClient: Valkey | null = null;
 
     try {
       job.status = 'running';
 
-      // Detect if target is a cluster
-      const targetInfo = await targetAdapter.getInfo(['cluster']);
-      const targetClusterSection = (targetInfo as Record<string, Record<string, string>>).cluster ?? {};
-      const targetIsCluster = String(targetClusterSection['cluster_enabled'] ?? '0') === '1';
+      // Detect if source / target are clusters
+      const isClusterEnabled = (info: unknown): boolean => {
+        const section = (info as Record<string, Record<string, string>>).cluster ?? {};
+        return String(section['cluster_enabled'] ?? '0') === '1';
+      };
+      const sourceIsCluster = isClusterEnabled(await sourceAdapter.getInfo(['cluster']));
+      const targetIsCluster = isClusterEnabled(await targetAdapter.getInfo(['cluster']));
 
       // Create temporary iovalkey clients — same pattern as command-migration-worker.ts
       sourceClient = createClient(sourceConfig, 'BetterDB-Validation-Source');
@@ -149,7 +155,13 @@ export class MigrationValidationService {
       if (job.cancelled) return;
 
       // Step 2: Key count comparison (20%)
-      const keyCount = await compareKeyCounts(sourceClient, targetClient, analysisResult);
+      // The standalone source client only sees the seed node of a cluster, so
+      // count a cluster source through a Cluster client that reaches every master.
+      if (sourceIsCluster) {
+        sourceCountClient = createTargetClient(sourceConfig, 'BetterDB-Validation-Source-Count', true);
+        await sourceCountClient.connect();
+      }
+      const keyCount = await compareKeyCounts(sourceCountClient ?? sourceClient, targetClient, analysisResult);
       job.result.keyCount = keyCount;
       job.progress = 20;
 
@@ -215,7 +227,7 @@ export class MigrationValidationService {
       }
       job.completedAt = Date.now();
       // Graceful cleanup — never Promise.all, never disconnect()
-      const clients = [sourceClient, targetClient].filter((c): c is Valkey => c !== null);
+      const clients = [sourceClient, sourceCountClient, targetClient].filter((c): c is Valkey => c !== null);
       await Promise.allSettled(clients.map(c => c.quit()));
     }
   }

apps/api/src/migration/validation/key-count-comparator.ts

@@ -1,18 +1,35 @@
-import type Valkey from 'iovalkey';
+import Valkey, { Cluster } from 'iovalkey';
 import type { KeyCountComparison, MigrationAnalysisResult } from '@betterdb/shared';
 
 /**
- * Compare key counts between source and target using DBSIZE.
- * Optionally enrich with per-type breakdown from Phase 1 analysis.
+ * DBSIZE that covers the whole dataset. On a Cluster client a bare DBSIZE goes
+ * to one random node, so sum it over every master instead.
+ */
+export async function countKeys(client: Valkey): Promise<number> {
+  // createTargetClient() casts Cluster to Valkey, so narrow from unknown.
+  const maybeCluster: unknown = client;
+  if (maybeCluster instanceof Cluster) {
+    const masters = maybeCluster.nodes('master');
+    if (masters.length === 0) throw new Error('No cluster masters available to count keys');
+    const counts = await Promise.all(masters.map((node) => node.dbsize()));
+    return counts.reduce((sum, count) => sum + count, 0);
+  }
+  return client.dbsize();
+}
+
+/**
+ * Compare key counts between source and target using DBSIZE (summed over all
+ * masters for cluster clients). Optionally enrich with per-type breakdown from
+ * Phase 1 analysis.
  */
 export async function compareKeyCounts(
   sourceClient: Valkey,
   targetClient: Valkey,
   analysisResult?: Partial<MigrationAnalysisResult>,
 ): Promise<KeyCountComparison> {
   const [sourceKeys, targetKeys] = await Promise.all([
-    sourceClient.dbsize(),
-    targetClient.dbsize(),
+    countKeys(sourceClient),
+    countKeys(targetClient),
   ]);
 
   const discrepancy = targetKeys - sourceKeys;

apps/api/test/migration-topology.e2e-spec.ts

@@ -3,6 +3,7 @@ import request from 'supertest';
 import Valkey, { Cluster } from 'iovalkey';
 import { execSync } from 'child_process';
 import { join } from 'path';
+import type { MigrationValidationResult } from '@betterdb/shared';
 import { createTestApp } from './test-utils';
 
 /**
@@ -227,6 +228,33 @@ async function runMigration(
   return result;
 }
 
+// ── Validation runner ───────────────────────────────────────────────
+
+async function runValidation(
+  app: NestFastifyApplication,
+  sourceId: string,
+  targetId: string,
+): Promise<MigrationValidationResult> {
+  const startRes = await request(app.getHttpServer())
+    .post('/migration/validation')
+    .send({ sourceConnectionId: sourceId, targetConnectionId: targetId });
+
+  expect([200, 201]).toContain(startRes.status);
+
+  const validationId = startRes.body.id;
+
+  let result: MigrationValidationResult | undefined;
+  for (let i = 0; i < 60; i++) {
+    const poll = await request(app.getHttpServer()).get(`/migration/validation/${validationId}`);
+    expect(poll.status).toBe(200);
+    result = poll.body as MigrationValidationResult;
+    if (result.status === 'completed' || result.status === 'failed') break;
+    await sleep(500);
+  }
+  if (!result) throw new Error(`Validation ${validationId} returned no result`);
+  return result;
+}
+
 // ── Tests ───────────────────────────────────────────────────────────
 
 (RUN ? describe : describe.skip)('Migration Topology E2E', () => {
@@ -353,6 +381,14 @@ async function runMigration(
     expect(result.status).toBe('completed');
     expect(result.keysTransferred).toBe(11);
 
+    // Validation must see the whole dataset on both sides — every master of a
+    // cluster, not one node — and pass a complete migration. Runs before
+    // verifyKeys, which reads from the stream's group on the target.
+    const validation = await runValidation(app, srcId, tgtId);
+    expect(validation.status).toBe('completed');
+    expect(validation.keyCount).toMatchObject({ sourceKeys: 11, targetKeys: 11, discrepancy: 0 });
+    expect(validation.passed).toBe(true);
+
     // Verify all 11 keys arrived on the target
     const target = await openClient(targetPort, targetIsCluster);
     try {

docs/migration.md

@@ -273,11 +273,12 @@ reported. If the source has custom ACL users missing from the target, analysis
 issues a warning. If the source uses modules not loaded on the target, analysis
 flags a blocking incompatibility.
 
-### DBSIZE accuracy in cluster mode
+### Key counts in cluster mode
 
-`DBSIZE` on a cluster client is sent to a single random node, returning a
-partial count. This means the key count comparison in validation may be
-inaccurate for cluster targets. This is a known limitation.
+Validation sums `DBSIZE` over every master of a cluster source or target, so
+the key count comparison covers the whole dataset. The sample spot-check still
+reads keys from the source connection's node; on a cluster source that is one
+shard's keys.
 
 ### Concurrent writes on the source