perf(stats): use compressed incremental snapshot transfers (#241)

src/core/services/stats-sync/cli-args.test.ts

@@ -54,6 +54,20 @@ test('parseSyncCliTokens handles the temp-dir protocol modes', () => {
   );
 });
 
+test('transfer cache keys are restricted to temp helpers and cannot contain paths', () => {
+  const key = 'a'.repeat(64);
+  for (const mode of [['--make-temp'], ['--remove-temp', '/tmp/subminer-sync-x']]) {
+    const parsed = parseSyncCliTokens(['sync', ...mode, '--transfer-cache', key]);
+    assert.equal(parsed.kind, 'run');
+    if (parsed.kind === 'run') assert.equal(parsed.args.syncTransferCacheKey, key);
+    assert.equal(
+      parseSyncCliTokens(['sync', ...mode, '--transfer-cache', '../../bad']).kind,
+      'error',
+    );
+  }
+  assert.equal(parseSyncCliTokens(['sync', 'host', '--transfer-cache', key]).kind, 'error');
+});
+
 test('parseSyncCliTokens owns the sync CLI validation rules', () => {
   assert.equal(parseSyncCliTokens([]).kind, 'error');
   assert.equal(parseSyncCliTokens(['sync']).kind, 'error');

src/core/services/stats-sync/cli-args.ts

@@ -1,4 +1,5 @@
 import type { SyncFlowArgs } from './sync-flow';
+import { isTransferCacheKey } from './transfer-cache';
 
 export const SYNC_CLI_FLAG = '--sync-cli';
 
@@ -43,6 +44,7 @@ export function parseSyncCliTokens(tokens: readonly string[]): ParsedSyncCli {
   let json = false;
   let makeTemp = false;
   let removeTemp = '';
+  let transferCacheKey = '';
   let remoteCmd = '';
   let dbPath = '';
   let logLevel = 'warn';
@@ -51,6 +53,7 @@ export function parseSyncCliTokens(tokens: readonly string[]): ParsedSyncCli {
     ['--snapshot', (value) => (snapshot = value.trim())],
     ['--merge', (value) => (merge = value.trim())],
     ['--remove-temp', (value) => (removeTemp = value.trim())],
+    ['--transfer-cache', (value) => (transferCacheKey = value.trim())],
     ['--remote-cmd', (value) => (remoteCmd = value.trim())],
     ['--db', (value) => (dbPath = value.trim())],
     ['--log-level', (value) => (logLevel = value.trim() || 'warn')],
@@ -93,6 +96,13 @@ export function parseSyncCliTokens(tokens: readonly string[]): ParsedSyncCli {
   }
 
   if (push && pull) return { kind: 'error', message: 'Sync --push and --pull cannot be combined.' };
+  if (transferCacheKey && (!isTransferCacheKey(transferCacheKey) || (!makeTemp && !removeTemp))) {
+    return {
+      kind: 'error',
+      message:
+        '--transfer-cache requires a 64-character lowercase hex key and --make-temp or --remove-temp.',
+    };
+  }
   if ((push || pull) && !host) {
     return { kind: 'error', message: 'Sync --push and --pull require a host.' };
   }
@@ -137,6 +147,7 @@ export function parseSyncCliTokens(tokens: readonly string[]): ParsedSyncCli {
       syncCheck: check,
       syncMakeTemp: makeTemp,
       syncRemoveTempPath: removeTemp,
+      syncTransferCacheKey: transferCacheKey,
       logLevel,
     },
   };
@@ -161,6 +172,7 @@ export function syncCliUsage(): string {
     '  --check              Test the SSH connection and remote SubMiner availability',
     '  --db <file>          Override the local stats database path',
     '  --remote-cmd <cmd>   SubMiner app or launcher command to run on the remote host',
+    '  --transfer-cache <key>  Reuse/save a received snapshot with temp helpers (internal)',
     '  -f, --force          Skip the running-app safety check',
     '  --json               Emit machine-readable NDJSON progress output',
     '  --log-level <level>  Log level',

src/core/services/stats-sync/snapshot-transfer.test.ts

@@ -0,0 +1,238 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { randomBytes } from 'node:crypto';
+import { spawnSync } from 'node:child_process';
+import { createSnapshotTransfer, runRsync } from './snapshot-transfer';
+import { createTransferCache, transferCacheKey } from './transfer-cache';
+
+type TransferDeps = NonNullable<Parameters<typeof createSnapshotTransfer>[2]>;
+
+function commandResult(status = 0, stderr = ''): ReturnType<TransferDeps['runRsync']> {
+  return { status, stderr, stdout: '', pid: 0, output: [null, '', stderr], signal: null };
+}
+
+function makeDeps(overrides: Partial<TransferDeps> = {}): TransferDeps {
+  return {
+    platform: 'linux',
+    runRsync: () => commandResult(),
+    runSsh: () => ({ status: 0, stdout: '', stderr: '' }),
+    runScp: () => assert.fail('Unexpected scp fallback'),
+    ...overrides,
+  };
+}
+
+test('snapshot transfer falls back when rsync is unavailable or an endpoint is Windows', () => {
+  for (const scenario of ['local-missing', 'remote-missing', 'local-windows', 'remote-windows']) {
+    const copies: string[][] = [];
+    const transfer = createSnapshotTransfer(
+      'macbook',
+      scenario === 'remote-windows' ? 'windows-cmd' : 'posix',
+      makeDeps({
+        platform: scenario === 'local-windows' ? 'win32' : 'linux',
+        runRsync: () => commandResult(scenario === 'local-missing' ? 1 : 0),
+        runSsh: () => ({ status: scenario === 'remote-missing' ? 127 : 0, stdout: '', stderr: '' }),
+        runScp: (from, to) => copies.push([from, to]),
+      }),
+    );
+    assert.equal(transfer.kind, 'scp', scenario);
+    transfer.copy({
+      direction: 'download',
+      localPath: '/local.sqlite',
+      remotePath: '/remote.sqlite',
+    });
+    transfer.copy({
+      direction: 'upload',
+      localPath: '/local.sqlite',
+      remotePath: '/remote.sqlite',
+    });
+    assert.deepEqual(copies, [
+      ['macbook:/remote.sqlite', '/local.sqlite'],
+      ['/local.sqlite', 'macbook:/remote.sqlite'],
+    ]);
+  }
+});
+
+test('failed rsync transfers report errors without silently retrying through scp', () => {
+  const transfer = createSnapshotTransfer(
+    'macbook',
+    'posix',
+    makeDeps({
+      runRsync: (args) => commandResult(args.includes('--version') ? 0 : 23, 'Permission denied'),
+    }),
+  );
+  assert.throws(
+    () =>
+      transfer.copy({
+        direction: 'upload',
+        localPath: '/local.sqlite',
+        remotePath: '/remote.sqlite',
+      }),
+    /rsync upload failed for macbook: Permission denied/,
+  );
+  assert.throws(() => createSnapshotTransfer('-oProxyCommand=bad', 'posix', makeDeps()), /option/);
+});
+
+const hasRsync = process.platform !== 'win32' && spawnSync('rsync', ['--version']).status === 0;
+
+test(
+  'rsync forces SSH, preserves its environment, and fails on a process timeout',
+  { skip: process.platform === 'win32' },
+  () => {
+    const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-rsync-process-test-'));
+    const previousPath = process.env.PATH;
+    const previousRsh = process.env.RSYNC_RSH;
+    try {
+      process.env.PATH = `${dir}${path.delimiter}${previousPath ?? ''}`;
+      process.env.RSYNC_RSH = 'unexpected-transport';
+      const executable = path.join(dir, 'rsync');
+      fs.writeFileSync(
+        executable,
+        '#!/bin/sh\nprintf "%s\\n" "$@" "$RSYNC_RSH" "$RSYNC_OLD_ARGS"\n',
+        { mode: 0o700 },
+      );
+      const result = runRsync(['--version']);
+      assert.equal(result.status, 0);
+      assert.deepEqual(result.stdout.trim().split('\n'), [
+        '--rsh=ssh',
+        '--version',
+        'unexpected-transport',
+        '1',
+      ]);
+
+      fs.writeFileSync(executable, '#!/bin/sh\nexec /bin/sleep 5\n');
+      const transfer = createSnapshotTransfer(
+        'macbook',
+        'posix',
+        makeDeps({
+          runRsync: (args) => (args.includes('--version') ? commandResult() : runRsync(args, 50)),
+        }),
+      );
+      assert.throws(
+        () =>
+          transfer.copy({
+            direction: 'download',
+            localPath: '/local.sqlite',
+            remotePath: '/remote.sqlite',
+          }),
+        /rsync download timed out for macbook/,
+      );
+    } finally {
+      if (previousPath === undefined) delete process.env.PATH;
+      else process.env.PATH = previousPath;
+      if (previousRsh === undefined) delete process.env.RSYNC_RSH;
+      else process.env.RSYNC_RSH = previousRsh;
+      fs.rmSync(dir, { recursive: true, force: true });
+    }
+  },
+);
+
+for (const direction of ['download', 'upload'] as const) {
+  test(
+    `rsync ${direction} reuses snapshot blocks and preserves the basis`,
+    { skip: !hasRsync },
+    () => {
+      const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-transfer-test-'));
+      try {
+        const localDir = path.join(dir, 'local');
+        const remoteDir = path.join(dir, "remote space ' $(false)");
+        fs.mkdirSync(localDir);
+        fs.mkdirSync(remoteDir);
+        // Emulate SSH's remote shell with real rsync processes, without sshd.
+        const remoteShell = path.join(dir, 'remote-shell');
+        fs.writeFileSync(remoteShell, '#!/bin/sh\nshift\nexec /bin/sh -c "$*"\n', { mode: 0o700 });
+        const localPath = path.join(
+          localDir,
+          direction === 'download' ? 'incoming' : '',
+          'snapshot.sqlite',
+        );
+        const remotePath = path.join(
+          remoteDir,
+          direction === 'upload' ? 'incoming' : '',
+          'snapshot.sqlite',
+        );
+        const source = direction === 'download' ? remotePath : localPath;
+        const destination = direction === 'download' ? localPath : remotePath;
+        const basis = path.join(path.dirname(destination), '..', 'snapshot.sqlite');
+        // Incompressible data ensures savings come from matching blocks.
+        const original = randomBytes(4 * 1024 * 1024);
+        fs.writeFileSync(basis, original);
+        fs.mkdirSync(path.dirname(destination), { recursive: true });
+        fs.writeFileSync(destination, original);
+        const updated = Buffer.from(original);
+        updated.fill(42, 65536, 69632);
+        fs.writeFileSync(source, updated);
+        let stats = '';
+        const transfer = createSnapshotTransfer(
+          'test-peer',
+          'posix',
+          makeDeps({
+            runRsync: (args) => {
+              const result = spawnSync(
+                'rsync',
+                [
+                  `--rsh=${remoteShell}`,
+                  ...args.map((arg) => (arg === '--quiet' ? '--stats' : arg)),
+                ],
+                { encoding: 'utf8', env: { ...process.env, RSYNC_OLD_ARGS: '1', LC_ALL: 'C' } },
+              );
+              stats = result.stdout;
+              return result;
+            },
+          }),
+        );
+        assert.equal(transfer.kind, 'rsync');
+        transfer.copy({ direction, localPath, remotePath });
+        assert.deepEqual(fs.readFileSync(destination), updated);
+        assert.deepEqual(fs.readFileSync(basis), original);
+        const matched = /Matched data: ([\d,]+) (?:bytes|B)/.exec(stats)?.[1];
+        assert.ok(matched, stats);
+        assert.ok(Number(matched.replaceAll(',', '')) > original.length * 0.95, stats);
+
+        const coldDir = path.join(dir, 'cold');
+        fs.mkdirSync(coldDir);
+        const coldDestination = path.join(coldDir, 'incoming', 'snapshot.sqlite');
+        transfer.copy({
+          direction,
+          localPath: direction === 'download' ? coldDestination : localPath,
+          remotePath: direction === 'upload' ? coldDestination : remotePath,
+        });
+        assert.deepEqual(fs.readFileSync(coldDestination), updated);
+
+        // A later sync starts in a new directory and reuses the prior peer's
+        // received file even when the source has grown since that transfer.
+        const cache = createTransferCache(path.join(dir, 'cache'));
+        const key = transferCacheKey('peer');
+        cache.remember(key, direction === 'download' ? localDir : remoteDir);
+        const nextDir = path.join(dir, 'next');
+        cache.seed(key, nextDir);
+        const grown = Buffer.concat([updated, randomBytes(4096)]);
+        fs.writeFileSync(source, grown);
+        const nextDestination = path.join(nextDir, 'incoming', 'snapshot.sqlite');
+        transfer.copy({
+          direction,
+          localPath: direction === 'download' ? nextDestination : localPath,
+          remotePath: direction === 'upload' ? nextDestination : remotePath,
+        });
+        assert.deepEqual(fs.readFileSync(nextDestination), grown);
+        const cachedMatches = /Matched data: ([\d,]+) (?:bytes|B)/.exec(stats)?.[1];
+        assert.ok(cachedMatches, stats);
+        assert.ok(Number(cachedMatches.replaceAll(',', '')) > updated.length * 0.95, stats);
+
+        // A retry must replace stale content even if size and mtime agree.
+        updated.fill(43, 131072, 135168);
+        fs.writeFileSync(source, updated);
+        const timestamp = new Date(1_700_000_000_000);
+        fs.utimesSync(source, timestamp, timestamp);
+        fs.utimesSync(destination, timestamp, timestamp);
+        transfer.copy({ direction, localPath, remotePath });
+        assert.deepEqual(fs.readFileSync(destination), updated);
+        assert.deepEqual(fs.readFileSync(basis), original);
+      } finally {
+        fs.rmSync(dir, { recursive: true, force: true });
+      }
+    },
+  );
+}

src/core/services/stats-sync/snapshot-transfer.ts

@@ -0,0 +1,85 @@
+import { spawnSync } from 'node:child_process';
+import path from 'node:path';
+import { assertSafeSshHost, runScp, runSsh, shellQuote, type RemoteShellFlavor } from './ssh';
+
+const RSYNC_OPTIONS = ['--compress', '--checksum'];
+
+export function runRsync(args: string[], timeoutMs = 30 * 60_000) {
+  return spawnSync('rsync', ['--rsh=ssh', ...args], {
+    encoding: 'utf8',
+    stdio: ['inherit', 'pipe', 'pipe'],
+    timeout: timeoutMs,
+    killSignal: 'SIGKILL',
+    // Quote remote paths ourselves for both modern rsync and macOS openrsync.
+    env: { ...process.env, RSYNC_OLD_ARGS: '1' },
+  });
+}
+
+interface TransferDeps {
+  platform: NodeJS.Platform;
+  runRsync: typeof runRsync;
+  runSsh: typeof runSsh;
+  runScp: typeof runScp;
+}
+
+export interface SnapshotTransfer {
+  kind: 'rsync' | 'scp';
+  copy: (request: {
+    direction: 'download' | 'upload';
+    localPath: string;
+    remotePath: string;
+  }) => void;
+}
+
+/**
+ * For rsync, both paths name snapshot.sqlite, with the destination inside an
+ * incoming/ directory seeded from the transfer cache. rsync creates incoming/
+ * when there is no cached basis and verifies the reconstructed file.
+ * Missing rsync and Windows endpoints use compressed scp with ordinary paths.
+ */
+export function createSnapshotTransfer(
+  host: string,
+  flavor: RemoteShellFlavor,
+  deps: TransferDeps = { platform: process.platform, runRsync, runSsh, runScp },
+): SnapshotTransfer {
+  assertSafeSshHost(host);
+  const canUseRsync =
+    deps.platform !== 'win32' &&
+    flavor === 'posix' &&
+    deps.runRsync([...RSYNC_OPTIONS, '--version']).status === 0 &&
+    deps.runSsh(host, `rsync ${RSYNC_OPTIONS.join(' ')} --version`, {
+      batchMode: true,
+      connectTimeoutSeconds: 10,
+      timeoutMs: 15_000,
+    }).status === 0;
+
+  return {
+    kind: canUseRsync ? 'rsync' : 'scp',
+    copy: ({ direction, localPath, remotePath }) => {
+      // A directory destination lets rsync create incoming/ on either end,
+      // including peers running older SubMiner versions.
+      const local =
+        canUseRsync && direction === 'download' ? `${path.dirname(localPath)}/` : localPath;
+      const remoteTarget =
+        canUseRsync && direction === 'upload' ? `${path.posix.dirname(remotePath)}/` : remotePath;
+      const remote = `${host}:${canUseRsync ? shellQuote(remoteTarget) : remoteTarget}`;
+      const [from, to] =
+        direction === 'download' ? ([remote, local] as const) : ([local, remote] as const);
+      if (!canUseRsync) {
+        deps.runScp(from, to);
+        return;
+      }
+      // --checksum prevents a same-size, same-mtime snapshot being skipped.
+      // Without --inplace, rsync replaces the staged basis only after the
+      // reconstructed file passes its transfer checksum.
+      const result = deps.runRsync([...RSYNC_OPTIONS, '--quiet', '--', from, to]);
+      if (result.error && 'code' in result.error && result.error.code === 'ETIMEDOUT') {
+        throw new Error(`rsync ${direction} timed out for ${host}`);
+      }
+      if (result.error) throw new Error(`Failed to run rsync: ${result.error.message}`);
+      if (result.status !== 0) {
+        throw new Error(`rsync ${direction} failed for ${host}: ${result.stderr.trim()}`);
+      }
+    },
+  };
+}

src/core/services/stats-sync/ssh.ts

@@ -74,7 +74,7 @@ function assertSafeScpEndpoint(endpoint: string): void {
 export function runScp(from: string, to: string): void {
   assertSafeScpEndpoint(from);
   assertSafeScpEndpoint(to);
-  const result = spawnSync('scp', ['-q', from, to], {
+  const result = spawnSync('scp', ['-C', '-q', from, to], {
     encoding: 'utf8',
     stdio: ['inherit', 'inherit', 'inherit'],
   });

src/core/services/stats-sync/sync-flow.test.ts

@@ -25,6 +25,7 @@ function makeContext(overrides: Partial<SyncFlowContext['args']> = {}): SyncFlow
       syncCheck: false,
       syncMakeTemp: false,
       syncRemoveTempPath: '',
+      syncTransferCacheKey: '',
       logLevel: 'warn',
       ...overrides,
     },
@@ -46,7 +47,8 @@ function makeDeps(overrides: Partial<SyncFlowDeps> = {}): SyncFlowDeps {
     assertSafeSshHost: () => {},
     detectRemoteShellFlavor: () => 'posix',
     resolveRemoteSubminerCommand: () => 'subminer',
-    runScp: () => {},
+    createSnapshotTransfer: () => ({ kind: 'scp', copy: () => {} }),
+    transferCache: { seed: () => {}, remember: () => {} },
     runSsh: () => ok(),
     canConnectUnixSocket: async () => false,
     realpathSync: (candidate) => candidate,
@@ -116,9 +118,9 @@ test('runSyncFlow dispatches snapshot, merge, host, and missing-target modes', a
       calls.push(`ssh:${command}`);
       return command.includes(' sync --make-temp') ? ok('/tmp/subminer-sync-remote\n') : ok();
     },
-    runScp: (from, to) => {
+    createSnapshotTransfer: scpTransfer((from, to) => {
       calls.push(`scp:${from}->${to}`);
-    },
+    }),
   });
 
   await runSyncFlow(
@@ -146,6 +148,19 @@ test('runSyncFlow dispatches snapshot, merge, host, and missing-target modes', a
   );
 });
 
+function scpTransfer(
+  copy: (from: string, to: string) => void,
+): SyncFlowDeps['createSnapshotTransfer'] {
+  return (host) => ({
+    kind: 'scp',
+    copy: ({ direction, localPath, remotePath }) => {
+      const remote = `${host}:${remotePath}`;
+      if (direction === 'download') copy(remote, localPath);
+      else copy(localPath, remote);
+    },
+  });
+}
+
 function makeHostDeps(calls: string[], overrides: Partial<SyncFlowDeps> = {}): SyncFlowDeps {
   return makeDeps({
     createDbSnapshot: (_dbPath, outPath) => {
@@ -164,10 +179,10 @@ function makeHostDeps(calls: string[], overrides: Partial<SyncFlowDeps> = {}): S
       if (command.includes(' sync --make-temp')) return ok('/tmp/subminer-sync-remote\n');
       return ok();
     },
-    runScp: (from, to) => {
+    createSnapshotTransfer: scpTransfer((from, to) => {
       calls.push(`scp:${from}->${to}`);
       if (!to.includes(':')) fs.writeFileSync(to, 'pulled');
-    },
+    }),
     ...overrides,
   });
 }
@@ -248,6 +263,153 @@ test('runHostSync pull only snapshots remotely and merges locally', async () =>
   assert.ok(!calls.some((call) => call.includes(' sync --merge ')));
 });
 
+for (const direction of ['push', 'pull', 'both'] as const) {
+  test(`runHostSync ${direction} snapshots sources and maintains receiver caches`, async () => {
+    const calls: string[] = [];
+    const copies: string[] = [];
+    const cacheCalls: string[] = [];
+    await runSyncFlow(
+      makeContext({
+        syncDbPath: '/tmp/local.sqlite',
+        syncHost: 'media-box',
+        syncDirection: direction,
+      }),
+      makeHostDeps(calls, {
+        transferCache: {
+          seed: (key) => cacheCalls.push(`seed:${key}`),
+          remember: (key) => cacheCalls.push(`remember:${key}`),
+        },
+        createSnapshotTransfer: () => ({
+          kind: 'rsync',
+          copy: ({ direction: copyDirection, localPath, remotePath }) => {
+            assert.equal(
+              calls.some((call) => call.startsWith('snapshot:')),
+              direction !== 'pull',
+            );
+            assert.equal(
+              calls.some((call) => call.includes(' sync --snapshot ')),
+              direction !== 'push',
+            );
+            if (copyDirection === 'upload')
+              assert.equal(fs.readFileSync(localPath, 'utf8'), 'snapshot');
+            assert.equal(path.posix.basename(remotePath), 'snapshot.sqlite');
+            assert.equal(
+              path.posix.basename(path.posix.dirname(remotePath)),
+              copyDirection === 'upload' ? 'incoming' : 'subminer-sync-remote',
+            );
+            copies.push(copyDirection);
+          },
+        }),
+      }),
+    );
+    assert.deepEqual(
+      copies,
+      direction === 'both'
+        ? ['download', 'upload']
+        : direction === 'pull'
+          ? ['download']
+          : ['upload'],
+    );
+    assert.equal(calls.includes('local-merge'), direction !== 'push');
+    assert.equal(cacheCalls.length, direction === 'push' ? 0 : 2);
+    if (cacheCalls.length) assert.equal(cacheCalls[0]?.slice(5), cacheCalls[1]?.slice(9));
+    const remoteCacheCalls = calls.filter((call) => call.includes('--transfer-cache'));
+    assert.equal(remoteCacheCalls.length, direction === 'pull' ? 0 : 2);
+    if (remoteCacheCalls.length) {
+      assert.ok(remoteCacheCalls[0]?.includes('--make-temp'));
+      assert.ok(remoteCacheCalls[1]?.includes('--remove-temp'));
+      assert.equal(
+        remoteCacheCalls[0]?.split('--transfer-cache ')[1],
+        remoteCacheCalls[1]?.split('--transfer-cache ')[1],
+      );
+    }
+    assert.equal(
+      calls.some((call) => call.includes(' sync --merge ')),
+      direction !== 'pull',
+    );
+  });
+}
+
+test('runHostSync does not merge an incomplete transfer and removes its temp files', async () => {
+  const calls: string[] = [];
+  let localTmpDir = '';
+  await assert.rejects(
+    () =>
+      runSyncFlow(
+        makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }),
+        makeHostDeps(calls, {
+          transferCache: {
+            seed: () => {},
+            remember: () => assert.fail('Failed sync must not update cache'),
+          },
+          mkdtempSync: (prefix) => {
+            localTmpDir = fs.mkdtempSync(prefix);
+            return localTmpDir;
+          },
+          createSnapshotTransfer: () => ({
+            kind: 'rsync',
+            copy: () => {
+              throw new Error('connection lost');
+            },
+          }),
+        }),
+      ),
+    /connection lost/,
+  );
+  assert.ok(!calls.includes('local-merge'));
+  assert.ok(!calls.some((call) => call.includes(' sync --merge ')));
+  assert.ok(calls.some((call) => call.includes(' sync --remove-temp ')));
+  assert.equal(fs.existsSync(localTmpDir), false);
+  assert.ok(
+    !calls.some((call) => call.includes('--remove-temp') && call.includes('--transfer-cache')),
+  );
+});
+
+for (const stderr of [
+  'Unknown sync option: --transfer-cache',
+  "error: unknown option '--transfer-cache'\n\nUsage: subminer sync [options] [host]",
+]) {
+  test(`runHostSync falls back when the peer reports ${stderr.split('\n')[0]}`, async () => {
+    const calls: string[] = [];
+    await runSyncFlow(
+      makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }),
+      makeHostDeps(calls, {
+        createSnapshotTransfer: () => ({ kind: 'rsync', copy: () => {} }),
+        runSsh: (_host, command) => {
+          calls.push(command);
+          if (command.includes('--transfer-cache')) return { status: 2, stdout: '', stderr };
+          return command.includes('--make-temp') ? ok('/tmp/subminer-sync-remote') : ok();
+        },
+      }),
+    );
+    assert.equal(calls.filter((call) => call.includes('--make-temp')).length, 2);
+    assert.ok(
+      calls.some((call) => call.includes('--remove-temp') && !call.includes('--transfer-cache')),
+    );
+  });
+}
+
+test('runHostSync does not retry unrelated remote temp failures', async () => {
+  const calls: string[] = [];
+  await assert.rejects(
+    runSyncFlow(
+      makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }),
+      makeHostDeps(calls, {
+        createSnapshotTransfer: () => ({
+          kind: 'rsync',
+          copy: () => assert.fail('Must not transfer'),
+        }),
+        runSsh: (_host, command) => {
+          calls.push(command);
+          return { status: 1, stdout: '', stderr: 'Permission denied' };
+        },
+      }),
+    ),
+    /Could not create a temporary directory on media-box.*\nPermission denied/,
+  );
+  assert.equal(calls.filter((call) => call.includes('--make-temp')).length, 1);
+});
+
 test('runSyncFlow --json emits NDJSON progress events and a final result', async () => {
   const lines: string[] = [];
   const remoteSummary = {
@@ -436,10 +598,10 @@ test('runHostSync speaks Windows shells: app command, double quotes, temp protoc
       if (command.includes(' sync --make-temp')) return ok(`${winTemp}\r\n`);
       return ok();
     },
-    runScp: (from, to) => {
+    createSnapshotTransfer: scpTransfer((from, to) => {
       scpCalls.push(`${from}->${to}`);
       if (!to.includes(':')) fs.writeFileSync(to, 'pulled');
-    },
+    }),
   });
 
   await runSyncFlow(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'win-box' }), deps);

src/core/services/stats-sync/sync-flow.ts

@@ -3,6 +3,8 @@ import path from 'node:path';
 import { formatMergeSummary } from './merge';
 import { quoteForRemoteShell } from './ssh';
 import type { RemoteRunResult, RemoteShellFlavor, RunSshOptions } from './ssh';
+import type { createSnapshotTransfer } from './snapshot-transfer';
+import { transferCacheKey, type createTransferCache } from './transfer-cache';
 import {
   parseSyncProgressLine,
   type SyncMergeSummary,
@@ -22,6 +24,7 @@ export interface SyncFlowArgs {
   syncCheck: boolean;
   syncMakeTemp: boolean;
   syncRemoveTempPath: string;
+  syncTransferCacheKey: string;
   logLevel: string;
 }
 
@@ -31,7 +34,7 @@ export interface SyncFlowContext {
 }
 
 /**
- * Process/IO seams the sync flow needs stubbed in tests: SSH/scp, the DB
+ * Process/IO seams the sync flow needs stubbed in tests: SSH/transfers, the DB
  * snapshot/merge engine, filesystem, and progress/bookkeeping output. The
  * app's --sync-cli mode (src/main/sync-cli.ts) provides the only production
  * binding; pure helpers are imported directly.
@@ -51,7 +54,8 @@ export interface SyncFlowDeps {
     flavor: RemoteShellFlavor,
     runRemote?: (host: string, remoteCommand: string) => RemoteRunResult,
   ) => string;
-  runScp: (from: string, to: string) => void;
+  createSnapshotTransfer: typeof createSnapshotTransfer;
+  transferCache: ReturnType<typeof createTransferCache>;
   runSsh: (host: string, remoteCommand: string, options?: RunSshOptions) => RemoteRunResult;
   canConnectUnixSocket: (socketPath: string) => Promise<boolean>;
   realpathSync: (candidate: string) => string;
@@ -95,12 +99,17 @@ function assertRemovableSyncTempDir(target: string): string {
   return resolved;
 }
 
-function runMakeTempMode(deps: SyncFlowDeps): void {
-  deps.consoleLog(makeSyncTempDir(deps.mkdtempSync));
+function runMakeTempMode(context: SyncFlowContext, deps: SyncFlowDeps): void {
+  const dir = makeSyncTempDir(deps.mkdtempSync);
+  if (context.args.syncTransferCacheKey)
+    deps.transferCache.seed(context.args.syncTransferCacheKey, dir);
+  deps.consoleLog(dir);
 }
 
 function runRemoveTempMode(context: SyncFlowContext, deps: SyncFlowDeps): void {
   const target = assertRemovableSyncTempDir(context.args.syncRemoveTempPath);
+  if (context.args.syncTransferCacheKey)
+    deps.transferCache.remember(context.args.syncTransferCacheKey, target);
   deps.rmSync(target, { recursive: true, force: true });
 }
 
@@ -260,9 +269,10 @@ function cleanupRemote(
   remoteTmpDir: string,
   quote: (value: string) => string,
   deps: SyncFlowDeps,
+  cacheFlag = '',
 ): void {
   if (!path.posix.basename(remoteTmpDir).startsWith(SYNC_TEMP_PREFIX)) return;
-  deps.runSsh(host, `${remoteCmd} sync --remove-temp ${quote(remoteTmpDir)}`);
+  deps.runSsh(host, `${remoteCmd} sync --remove-temp ${quote(remoteTmpDir)}${cacheFlag}`);
 }
 
 /**
@@ -309,19 +319,37 @@ export async function runHostSync(
 
   const flavor = deps.detectRemoteShellFlavor(host, deps.runSsh);
   const remoteCmd = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null, flavor);
+  const transfer = deps.createSnapshotTransfer(host, flavor);
   const quote = (value: string) => quoteForRemoteShell(flavor, value);
   if (args.logLevel === 'debug') {
     console.error(`Remote subminer command (${flavor}): ${remoteCmd}`);
   }
 
   const localTmpDir = makeSyncTempDir(deps.mkdtempSync);
+  const localCacheKey = transferCacheKey(`download\0${dbPath}\0${host}`);
+  const remoteCacheKey = transferCacheKey(`upload\0${os.hostname()}\0${dbPath}`);
+  let remoteCacheFlag = transfer.kind === 'rsync' ? ` --transfer-cache ${remoteCacheKey}` : '';
+  let syncSucceeded = false;
   let remoteTmpDir = '';
   let pulledSummary: SyncMergeSummary | null = null;
   try {
     // Signal failures by throwing (not fail(), which exits synchronously and
     // would skip the finally cleanup, leaking temp dirs holding snapshot data).
     // main().catch() reports the message the same way fail() would.
-    const mktemp = deps.runSsh(host, `${remoteCmd} sync --make-temp`);
+    if (transfer.kind === 'rsync' && shouldPull)
+      deps.transferCache.seed(localCacheKey, localTmpDir);
+    let mktemp = deps.runSsh(
+      host,
+      `${remoteCmd} sync --make-temp${shouldPush ? remoteCacheFlag : ''}`,
+    );
+    if (
+      mktemp.status !== 0 &&
+      (mktemp.stderr.includes('Unknown sync option: --transfer-cache') ||
+        mktemp.stderr.includes("error: unknown option '--transfer-cache'"))
+    ) {
+      remoteCacheFlag = '';
+      mktemp = deps.runSsh(host, `${remoteCmd} sync --make-temp`);
+    }
     remoteTmpDir = mktemp.status === 0 ? parseRemoteTempDir(mktemp.stdout) : '';
     if (!remoteTmpDir) {
       throw new Error(
@@ -331,7 +359,7 @@ export async function runHostSync(
 
     const forceFlag = args.syncForce ? ' --force' : '';
 
-    const localSnapshot = path.join(localTmpDir, 'local.sqlite');
+    const localSnapshot = path.join(localTmpDir, 'snapshot.sqlite');
     if (shouldPush) {
       deps.consoleLog(`Snapshotting local database (${dbPath})...`);
       deps.emitEvent({
@@ -355,19 +383,30 @@ export async function runHostSync(
       }
     }
 
-    const pulledSnapshot = path.join(localTmpDir, 'remote.sqlite');
+    const pulledSnapshot = path.join(
+      localTmpDir,
+      transfer.kind === 'rsync' ? 'incoming/snapshot.sqlite' : 'remote.sqlite',
+    );
     if (shouldPull) {
       deps.emitEvent({
         type: 'stage',
         stage: 'download',
         message: `Copying snapshot from ${host}`,
       });
-      deps.runScp(`${host}:${remoteSnapshot}`, pulledSnapshot);
+      transfer.copy({
+        direction: 'download',
+        remotePath: remoteSnapshot,
+        localPath: pulledSnapshot,
+      });
     }
-    const incomingSnapshot = `${remoteTmpDir}/incoming.sqlite`;
+    const incomingSnapshot = `${remoteTmpDir}/${transfer.kind === 'rsync' ? 'incoming/snapshot.sqlite' : 'incoming.sqlite'}`;
     if (shouldPush) {
       deps.emitEvent({ type: 'stage', stage: 'upload', message: `Copying snapshot to ${host}` });
-      deps.runScp(localSnapshot, `${host}:${incomingSnapshot}`);
+      transfer.copy({
+        direction: 'upload',
+        localPath: localSnapshot,
+        remotePath: incomingSnapshot,
+      });
     }
 
     if (shouldPull) {
@@ -416,6 +455,7 @@ export async function runHostSync(
       }
     }
 
+    syncSucceeded = true;
     deps.consoleLog('\nSync complete.');
     deps.recordHostSyncResult(host, 'success', formatHostSyncDetail(direction, pulledSummary));
   } catch (error) {
@@ -430,10 +470,19 @@ export async function runHostSync(
     }
     throw error;
   } finally {
+    if (syncSucceeded && transfer.kind === 'rsync' && shouldPull)
+      deps.transferCache.remember(localCacheKey, localTmpDir);
     deps.rmSync(localTmpDir, { recursive: true, force: true });
     if (remoteTmpDir) {
       try {
-        cleanupRemote(host, remoteCmd, remoteTmpDir, quote, deps);
+        cleanupRemote(
+          host,
+          remoteCmd,
+          remoteTmpDir,
+          quote,
+          deps,
+          syncSucceeded && shouldPush ? remoteCacheFlag : '',
+        );
       } catch {
         // best effort
       }
@@ -451,7 +500,7 @@ export async function runSyncFlow(
 
   try {
     if (args.syncMakeTemp) {
-      runMakeTempMode(deps);
+      runMakeTempMode(context, deps);
     } else if (args.syncRemoveTempPath) {
       runRemoveTempMode(context, deps);
     } else {

src/core/services/stats-sync/transfer-cache.test.ts

@@ -0,0 +1,59 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { createTransferCache, transferCacheKey } from './transfer-cache';
+
+test('transfer cache isolates peers and active transfers while replacing previous snapshots', () => {
+  const root = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-cache-test-'));
+  try {
+    const cacheDir = path.join(root, 'cache');
+    const cache = createTransferCache(cacheDir);
+    const key = transferCacheKey('peer');
+    const first = path.join(root, 'first');
+    fs.mkdirSync(path.join(first, 'incoming'), { recursive: true });
+    const incoming = path.join(first, 'incoming', 'snapshot.sqlite');
+    fs.writeFileSync(incoming, 'first received snapshot');
+    cache.remember(key, first);
+    const second = path.join(root, 'second');
+    cache.seed(key, second);
+    fs.writeFileSync(incoming, 'next received snapshot');
+    cache.remember(key, first);
+    assert.equal(
+      fs.readFileSync(path.join(second, 'incoming', 'snapshot.sqlite'), 'utf8'),
+      'first received snapshot',
+    );
+    const third = path.join(root, 'third');
+    cache.seed(key, third);
+    assert.equal(
+      fs.readFileSync(path.join(third, 'incoming', 'snapshot.sqlite'), 'utf8'),
+      'next received snapshot',
+    );
+    assert.deepEqual(fs.readdirSync(cacheDir), [`${key}.sqlite`]);
+    const other = path.join(root, 'other');
+    cache.seed(transferCacheKey('other peer'), other);
+    assert.equal(fs.existsSync(path.join(other, 'incoming', 'snapshot.sqlite')), false);
+    assert.throws(() => cache.seed('../outside', other), /Invalid/);
+    assert.throws(() => cache.remember('../outside', first), /Invalid/);
+  } finally {
+    fs.rmSync(root, { recursive: true, force: true });
+  }
+});
+
+test('unavailable cache storage and missing incoming snapshots do not prevent sync', () => {
+  const root = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-cache-test-'));
+  try {
+    const unavailable = path.join(root, 'file');
+    fs.writeFileSync(unavailable, 'not a directory');
+    const cache = createTransferCache(unavailable);
+    const key = transferCacheKey('peer');
+    const temp = path.join(root, 'transfer');
+    assert.doesNotThrow(() => cache.seed(key, temp));
+    assert.doesNotThrow(() => cache.remember(key, temp));
+    fs.writeFileSync(path.join(temp, 'incoming', 'snapshot.sqlite'), 'received');
+    assert.doesNotThrow(() => cache.remember(key, temp));
+  } finally {
+    fs.rmSync(root, { recursive: true, force: true });
+  }
+});

src/core/services/stats-sync/transfer-cache.ts

@@ -0,0 +1,61 @@
+import fs from 'node:fs';
+import path from 'node:path';
+import { createHash } from 'node:crypto';
+import { getDefaultConfigDir } from '../../../shared/setup-state';
+
+export function transferCacheKey(peer: string): string {
+  return createHash('sha256').update(peer).digest('hex');
+}
+
+export function isTransferCacheKey(value: string): boolean {
+  return /^[a-f0-9]{64}$/.test(value);
+}
+
+/**
+ * Keep one previously received snapshot per peer as an rsync basis. Copies
+ * isolate active transfers from concurrent cache replacements. A missing or
+ * unusable cache only costs bandwidth; it must never prevent a sync.
+ */
+export function createTransferCache(
+  directory = path.join(getDefaultConfigDir(), 'sync-transfer-cache'),
+) {
+  function cachePath(key: string): string {
+    if (!isTransferCacheKey(key)) throw new Error('Invalid sync transfer cache key');
+    return path.join(directory, `${key}.sqlite`);
+  }
+
+  return {
+    seed(key: string, tempDir: string): void {
+      const source = cachePath(key);
+      const incoming = path.join(tempDir, 'incoming', 'snapshot.sqlite');
+      try {
+        fs.mkdirSync(path.dirname(incoming), { recursive: true, mode: 0o700 });
+        fs.copyFileSync(source, incoming, fs.constants.COPYFILE_FICLONE);
+      } catch {
+        // A cold transfer sends a complete compressed snapshot.
+      }
+    },
+
+    remember(key: string, tempDir: string): void {
+      const target = cachePath(key);
+      const incoming = path.join(tempDir, 'incoming', 'snapshot.sqlite');
+      let staging = '';
+      try {
+        if (!fs.existsSync(incoming)) return;
+        fs.mkdirSync(directory, { recursive: true, mode: 0o700 });
+        staging = fs.mkdtempSync(path.join(directory, '.write-'));
+        const snapshot = path.join(staging, 'snapshot.sqlite');
+        fs.copyFileSync(incoming, snapshot, fs.constants.COPYFILE_FICLONE);
+        fs.renameSync(snapshot, target);
+      } catch {
+        // An older basis is still valid. Never publish a partially copied file.
+      } finally {
+        try {
+          if (staging) fs.rmSync(staging, { recursive: true, force: true });
+        } catch {
+          // Cache cleanup is optional too.
+        }
+      }
+    },
+  };
+}