From e7e4a62504aec851cce534979fda68ecbb2f36d8 Mon Sep 17 00:00:00 2001 From: sudacode Date: Mon, 7 Sep 2026 23:27:11 -0700 Subject: [PATCH] perf(stats): use compressed incremental snapshot transfers - Prefer checksum-based fuzzy rsync with compressed scp fallback - Preserve bidirectional snapshot bases during transfers --- .../services/stats-sync/snapshot-transfer.ts | 71 +++++++++++++++++++ src/core/services/stats-sync/ssh.ts | 2 +- src/core/services/stats-sync/sync-flow.ts | 14 ++-- src/main/sync-cli.ts | 4 +- 4 files changed, 82 insertions(+), 9 deletions(-) create mode 100644 src/core/services/stats-sync/snapshot-transfer.ts diff --git a/src/core/services/stats-sync/snapshot-transfer.ts b/src/core/services/stats-sync/snapshot-transfer.ts new file mode 100644 index 00000000..b9f8f604 --- /dev/null +++ b/src/core/services/stats-sync/snapshot-transfer.ts @@ -0,0 +1,71 @@ +import { spawnSync } from 'node:child_process'; +import { assertSafeSshHost, runScp, runSsh, shellQuote, type RemoteShellFlavor } from './ssh'; + +const RSYNC_OPTIONS = ['--compress', '--checksum', '--fuzzy']; + +function runRsync(args: string[]) { + return spawnSync('rsync', args, { + encoding: 'utf8', + stdio: ['inherit', 'pipe', 'pipe'], + // 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; +} + +/** + * Reuse the receiving machine's snapshot as rsync's fuzzy basis. Transfers + * always write a separate file; the basis remains available for the other + * direction. Missing rsync and Windows endpoints use compressed scp. + */ +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 }) => { + const remote = `${host}:${canUseRsync ? shellQuote(remotePath) : remotePath}`; + const endpoints = direction === 'download' ? [remote, localPath] : [localPath, remote]; + if (!canUseRsync) { + deps.runScp(endpoints[0]!, endpoints[1]!); + return; + } + // --checksum prevents a same-size, same-mtime snapshot being skipped. + // Avoid --inplace: neither a failed transfer nor a matching basis may + // modify the snapshot we still need to send in the opposite direction. + const result = deps.runRsync([...RSYNC_OPTIONS, '--quiet', '--', ...endpoints]); + 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()}`); + } + }, + }; +} diff --git a/src/core/services/stats-sync/ssh.ts b/src/core/services/stats-sync/ssh.ts index 99deb1dd..37778d7b 100644 --- a/src/core/services/stats-sync/ssh.ts +++ b/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'], }); diff --git a/src/core/services/stats-sync/sync-flow.ts b/src/core/services/stats-sync/sync-flow.ts index 415a0cc8..b1080b9d 100644 --- a/src/core/services/stats-sync/sync-flow.ts +++ b/src/core/services/stats-sync/sync-flow.ts @@ -3,6 +3,7 @@ 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 { parseSyncProgressLine, type SyncMergeSummary, @@ -31,7 +32,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 +52,7 @@ export interface SyncFlowDeps { flavor: RemoteShellFlavor, runRemote?: (host: string, remoteCommand: string) => RemoteRunResult, ) => string; - runScp: (from: string, to: string) => void; + createSnapshotTransfer: typeof createSnapshotTransfer; runSsh: (host: string, remoteCommand: string, options?: RunSshOptions) => RemoteRunResult; canConnectUnixSocket: (socketPath: string) => Promise; realpathSync: (candidate: string) => string; @@ -309,6 +310,7 @@ 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}`); @@ -332,7 +334,7 @@ export async function runHostSync( const forceFlag = args.syncForce ? ' --force' : ''; const localSnapshot = path.join(localTmpDir, 'local.sqlite'); - if (shouldPush) { + if (shouldPush || transfer.kind === 'rsync') { deps.consoleLog(`Snapshotting local database (${dbPath})...`); deps.emitEvent({ type: 'stage', @@ -343,7 +345,7 @@ export async function runHostSync( } const remoteSnapshot = `${remoteTmpDir}/snapshot.sqlite`; - if (shouldPull) { + if (shouldPull || transfer.kind === 'rsync') { deps.consoleLog(`Snapshotting ${host}...`); deps.emitEvent({ type: 'stage', stage: 'snapshot-remote', message: `Snapshotting ${host}` }); const snapshotRun = deps.runSsh( @@ -362,12 +364,12 @@ export async function runHostSync( 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`; 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) { diff --git a/src/main/sync-cli.ts b/src/main/sync-cli.ts index 2623ad26..1e9dfad0 100644 --- a/src/main/sync-cli.ts +++ b/src/main/sync-cli.ts @@ -14,9 +14,9 @@ import { assertSafeSshHost, detectRemoteShellFlavor, resolveRemoteSubminerCommand, - runScp, runSsh, } from '../core/services/stats-sync/ssh'; +import { createSnapshotTransfer } from '../core/services/stats-sync/snapshot-transfer'; import { ensureTrackerQuiescentFlow, runSyncFlow, @@ -63,7 +63,7 @@ function buildSyncCliDeps(): SyncFlowDeps { assertSafeSshHost, detectRemoteShellFlavor, resolveRemoteSubminerCommand, - runScp, + createSnapshotTransfer, runSsh, canConnectUnixSocket: canConnectSocket, realpathSync: (candidate) => fs.realpathSync(candidate),