perf(stats): use compressed incremental snapshot transfers

- Prefer checksum-based fuzzy rsync with compressed scp fallback
- Preserve bidirectional snapshot bases during transfers
This commit is contained in:
2026-09-07 23:27:11 -07:00
parent 1e5d7747b4
commit e7e4a62504
4 changed files with 82 additions and 9 deletions
@@ -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()}`);
}
},
};
}
+1 -1
View File
@@ -74,7 +74,7 @@ function assertSafeScpEndpoint(endpoint: string): void {
export function runScp(from: string, to: string): void { export function runScp(from: string, to: string): void {
assertSafeScpEndpoint(from); assertSafeScpEndpoint(from);
assertSafeScpEndpoint(to); assertSafeScpEndpoint(to);
const result = spawnSync('scp', ['-q', from, to], { const result = spawnSync('scp', ['-C', '-q', from, to], {
encoding: 'utf8', encoding: 'utf8',
stdio: ['inherit', 'inherit', 'inherit'], stdio: ['inherit', 'inherit', 'inherit'],
}); });
+8 -6
View File
@@ -3,6 +3,7 @@ import path from 'node:path';
import { formatMergeSummary } from './merge'; import { formatMergeSummary } from './merge';
import { quoteForRemoteShell } from './ssh'; import { quoteForRemoteShell } from './ssh';
import type { RemoteRunResult, RemoteShellFlavor, RunSshOptions } from './ssh'; import type { RemoteRunResult, RemoteShellFlavor, RunSshOptions } from './ssh';
import type { createSnapshotTransfer } from './snapshot-transfer';
import { import {
parseSyncProgressLine, parseSyncProgressLine,
type SyncMergeSummary, 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 * snapshot/merge engine, filesystem, and progress/bookkeeping output. The
* app's --sync-cli mode (src/main/sync-cli.ts) provides the only production * app's --sync-cli mode (src/main/sync-cli.ts) provides the only production
* binding; pure helpers are imported directly. * binding; pure helpers are imported directly.
@@ -51,7 +52,7 @@ export interface SyncFlowDeps {
flavor: RemoteShellFlavor, flavor: RemoteShellFlavor,
runRemote?: (host: string, remoteCommand: string) => RemoteRunResult, runRemote?: (host: string, remoteCommand: string) => RemoteRunResult,
) => string; ) => string;
runScp: (from: string, to: string) => void; createSnapshotTransfer: typeof createSnapshotTransfer;
runSsh: (host: string, remoteCommand: string, options?: RunSshOptions) => RemoteRunResult; runSsh: (host: string, remoteCommand: string, options?: RunSshOptions) => RemoteRunResult;
canConnectUnixSocket: (socketPath: string) => Promise<boolean>; canConnectUnixSocket: (socketPath: string) => Promise<boolean>;
realpathSync: (candidate: string) => string; realpathSync: (candidate: string) => string;
@@ -309,6 +310,7 @@ export async function runHostSync(
const flavor = deps.detectRemoteShellFlavor(host, deps.runSsh); const flavor = deps.detectRemoteShellFlavor(host, deps.runSsh);
const remoteCmd = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null, flavor); const remoteCmd = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null, flavor);
const transfer = deps.createSnapshotTransfer(host, flavor);
const quote = (value: string) => quoteForRemoteShell(flavor, value); const quote = (value: string) => quoteForRemoteShell(flavor, value);
if (args.logLevel === 'debug') { if (args.logLevel === 'debug') {
console.error(`Remote subminer command (${flavor}): ${remoteCmd}`); console.error(`Remote subminer command (${flavor}): ${remoteCmd}`);
@@ -332,7 +334,7 @@ export async function runHostSync(
const forceFlag = args.syncForce ? ' --force' : ''; const forceFlag = args.syncForce ? ' --force' : '';
const localSnapshot = path.join(localTmpDir, 'local.sqlite'); const localSnapshot = path.join(localTmpDir, 'local.sqlite');
if (shouldPush) { if (shouldPush || transfer.kind === 'rsync') {
deps.consoleLog(`Snapshotting local database (${dbPath})...`); deps.consoleLog(`Snapshotting local database (${dbPath})...`);
deps.emitEvent({ deps.emitEvent({
type: 'stage', type: 'stage',
@@ -343,7 +345,7 @@ export async function runHostSync(
} }
const remoteSnapshot = `${remoteTmpDir}/snapshot.sqlite`; const remoteSnapshot = `${remoteTmpDir}/snapshot.sqlite`;
if (shouldPull) { if (shouldPull || transfer.kind === 'rsync') {
deps.consoleLog(`Snapshotting ${host}...`); deps.consoleLog(`Snapshotting ${host}...`);
deps.emitEvent({ type: 'stage', stage: 'snapshot-remote', message: `Snapshotting ${host}` }); deps.emitEvent({ type: 'stage', stage: 'snapshot-remote', message: `Snapshotting ${host}` });
const snapshotRun = deps.runSsh( const snapshotRun = deps.runSsh(
@@ -362,12 +364,12 @@ export async function runHostSync(
stage: 'download', stage: 'download',
message: `Copying snapshot from ${host}`, 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}/incoming.sqlite`;
if (shouldPush) { if (shouldPush) {
deps.emitEvent({ type: 'stage', stage: 'upload', message: `Copying snapshot to ${host}` }); 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) { if (shouldPull) {
+2 -2
View File
@@ -14,9 +14,9 @@ import {
assertSafeSshHost, assertSafeSshHost,
detectRemoteShellFlavor, detectRemoteShellFlavor,
resolveRemoteSubminerCommand, resolveRemoteSubminerCommand,
runScp,
runSsh, runSsh,
} from '../core/services/stats-sync/ssh'; } from '../core/services/stats-sync/ssh';
import { createSnapshotTransfer } from '../core/services/stats-sync/snapshot-transfer';
import { import {
ensureTrackerQuiescentFlow, ensureTrackerQuiescentFlow,
runSyncFlow, runSyncFlow,
@@ -63,7 +63,7 @@ function buildSyncCliDeps(): SyncFlowDeps {
assertSafeSshHost, assertSafeSshHost,
detectRemoteShellFlavor, detectRemoteShellFlavor,
resolveRemoteSubminerCommand, resolveRemoteSubminerCommand,
runScp, createSnapshotTransfer,
runSsh, runSsh,
canConnectUnixSocket: canConnectSocket, canConnectUnixSocket: canConnectSocket,
realpathSync: (candidate) => fs.realpathSync(candidate), realpathSync: (candidate) => fs.realpathSync(candidate),