From 32c62b076b3fdc7d57b03c9edb905235a3bbf37f Mon Sep 17 00:00:00 2001 From: sudacode Date: Thu, 9 Jul 2026 18:35:36 -0700 Subject: [PATCH] fix(launcher): ignore stale sync sockets --- changes/launcher-sync-command.md | 2 +- docs-site/launcher-script.md | 2 +- launcher/commands/sync-command.test.ts | 51 +++++++++++++++++++------- launcher/commands/sync-command.ts | 39 ++++++++++---------- launcher/main.ts | 2 +- launcher/mpv.ts | 2 +- 6 files changed, 62 insertions(+), 36 deletions(-) diff --git a/changes/launcher-sync-command.md b/changes/launcher-sync-command.md index a39d8da1..5d5dfbf0 100644 --- a/changes/launcher-sync-command.md +++ b/changes/launcher-sync-command.md @@ -1,4 +1,4 @@ type: added area: launcher -- Added `subminer sync ` to merge immersion stats and watch history between two machines over SSH. Each side snapshots its database (`VACUUM INTO`), snapshots are exchanged with `scp`, and each machine merges the other's data as an insert-only union keyed on session UUIDs / video keys / series title keys, so re-syncing is idempotent and nothing is double-counted. Lifetime totals and daily/monthly rollups are updated incrementally (history older than the session retention window is preserved); remote-only historical rollups are copied only when they do not conflict with retained local session history. `subminer sync --snapshot ` and `subminer sync --merge ` expose the underlying steps for manual transfers. The command refuses to run while the stats daemon or an mpv session is active (`--force` overrides), keeps that guard in place through local/remote merges, supplies standard SubMiner and Bun paths to non-interactive SSH commands, verifies the remote launcher starts, reports remote stderr on failures, and aborts on stats schema version mismatches. +- Added `subminer sync ` to merge immersion stats and watch history between two machines over SSH. Each side snapshots its database (`VACUUM INTO`), snapshots are exchanged with `scp`, and each machine merges the other's data as an insert-only union keyed on session UUIDs / video keys / series title keys, so re-syncing is idempotent and nothing is double-counted. Lifetime totals and daily/monthly rollups are updated incrementally (history older than the session retention window is preserved); remote-only historical rollups are copied only when they do not conflict with retained local session history. `subminer sync --snapshot ` and `subminer sync --merge ` expose the underlying steps for manual transfers. The command refuses to run while the stats daemon or a live mpv session is active (`--force` overrides), ignores stale mpv socket files, keeps the guard in place through local/remote merges, supplies standard SubMiner and Bun paths to non-interactive SSH commands, verifies the remote launcher starts, reports remote stderr on failures, and aborts on stats schema version mismatches. diff --git a/docs-site/launcher-script.md b/docs-site/launcher-script.md index 573987f1..f2557a86 100644 --- a/docs-site/launcher-script.md +++ b/docs-site/launcher-script.md @@ -90,7 +90,7 @@ subminer sync macbook --remote-cmd ~/bin/subminer # custom remote launcher path How it works: each side takes a consistent snapshot of its database (`VACUUM INTO`), the snapshots are exchanged over `scp`, and each machine merges the other's snapshot into its own database. The merge is an insert-only union keyed on stable identifiers (session UUIDs, video keys, series title keys, word/kanji identity), so it is safe to re-run at any time — syncing twice changes nothing, and nothing is ever overwritten or summed twice. Lifetime totals and rollup charts are updated incrementally, so history older than the session retention window is preserved on both sides. -Close SubMiner (and stop the background stats daemon, `subminer stats -s`) on both machines before syncing; the command refuses to run while a SubMiner process may be writing the database (`--force` overrides). Both machines must be on the same SubMiner version — the sync aborts on a stats schema mismatch. Remote sync checks standard SubMiner and Bun locations (`~/.local/bin`, `~/.bun/bin`, Homebrew, `/usr/local/bin`, `/usr/bin`, and `/bin`) even when the non-interactive SSH shell omits them from `PATH`. +Close SubMiner (and stop the background stats daemon, `subminer stats -s`) on both machines before syncing; the command refuses to run while a SubMiner process may be writing the database (`--force` overrides). The mpv safety check requires a live socket connection, so a stale socket file left after mpv exits does not block sync. Both machines must be on the same SubMiner version — the sync aborts on a stats schema mismatch. Remote sync checks standard SubMiner and Bun locations (`~/.local/bin`, `~/.bun/bin`, Homebrew, `/usr/local/bin`, `/usr/bin`, and `/bin`) even when the non-interactive SSH shell omits them from `PATH`. Two lower-level modes are used internally over SSH and also work standalone for manual transfers (e.g. via a USB drive): diff --git a/launcher/commands/sync-command.test.ts b/launcher/commands/sync-command.test.ts index 2f5e4c6d..f0dafc4b 100644 --- a/launcher/commands/sync-command.test.ts +++ b/launcher/commands/sync-command.test.ts @@ -6,7 +6,7 @@ import path from 'node:path'; import type { Args } from '../types.js'; import { createEmptyMergeSummary } from '../sync/sync-shared.js'; import type { LauncherCommandContext } from './context.js'; -import { runSyncCommand, type SyncCommandDeps } from './sync-command.js'; +import { ensureTrackerQuiescent, runSyncCommand, type SyncCommandDeps } from './sync-command.js'; function makeContext(overrides: Partial): LauncherCommandContext { return { @@ -35,7 +35,29 @@ function ok(stdout = ''): { status: number; stdout: string; stderr: string } { return { status: 0, stdout, stderr: '' }; } -test('runSyncCommand dispatches snapshot, merge, host, and missing-target modes', () => { +test('ensureTrackerQuiescent ignores stale sockets but rejects live sockets', async () => { + const context = makeContext({ syncDbPath: '/tmp/local.sqlite' }); + context.mpvSocketPath = '/tmp/subminer-socket'; + let socketConnectable = false; + const deps: Partial = { + realpathSync: (() => '/tracker.sqlite') as unknown as typeof fs.realpathSync, + findLiveStatsDaemonPid: () => null, + canConnectUnixSocket: async () => socketConnectable, + fail: (message: string): never => { + throw new Error(message); + }, + }; + + await ensureTrackerQuiescent(context, '/tmp/local.sqlite', deps); + + socketConnectable = true; + await assert.rejects( + async () => ensureTrackerQuiescent(context, '/tmp/local.sqlite', deps), + /mpv\/SubMiner session appears to be running/, + ); +}); + +test('runSyncCommand dispatches snapshot, merge, host, and missing-target modes', async () => { const calls: string[] = []; const deps: Partial = { createDbSnapshot: (dbPath: string, outPath: string) => { @@ -46,7 +68,7 @@ test('runSyncCommand dispatches snapshot, merge, host, and missing-target modes' return createEmptyMergeSummary(); }, formatMergeSummary: () => 'summary', - ensureTrackerQuiescent: () => { + ensureTrackerQuiescent: async () => { calls.push('quiescent'); }, assertSafeSshHost: (host: string) => { @@ -66,7 +88,7 @@ test('runSyncCommand dispatches snapshot, merge, host, and missing-target modes' }; assert.equal( - runSyncCommand( + await runSyncCommand( makeContext({ syncDbPath: '/tmp/local.sqlite', syncSnapshotPath: '/tmp/out.sqlite' }), deps, ), @@ -74,23 +96,26 @@ test('runSyncCommand dispatches snapshot, merge, host, and missing-target modes' ); assert.ok(calls.includes('snapshot:/tmp/local.sqlite->/tmp/out.sqlite')); - runSyncCommand( + await runSyncCommand( makeContext({ syncDbPath: '/tmp/local.sqlite', syncMergePath: '/tmp/in.sqlite' }), deps, ); assert.ok(calls.includes('quiescent')); assert.ok(calls.includes('merge:/tmp/local.sqlite<-/tmp/in.sqlite')); - runSyncCommand(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }), deps); + await runSyncCommand( + makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }), + deps, + ); assert.ok(calls.includes('host:media-box')); - assert.throws( + await assert.rejects( () => runSyncCommand(makeContext({ syncDbPath: '/tmp/local.sqlite' }), deps), /sync requires a host, --snapshot , or --merge /, ); }); -test('runHostSync keeps tracker quiescent through local and remote merge and cleans up after failure', () => { +test('runHostSync keeps tracker quiescent through local and remote merge and cleans up after failure', async () => { const calls: string[] = []; let localTmpDir = ''; const deps: Partial = { @@ -103,7 +128,7 @@ test('runHostSync keeps tracker quiescent through local and remote merge and cle return createEmptyMergeSummary(); }, formatMergeSummary: () => 'summary', - ensureTrackerQuiescent: () => { + ensureTrackerQuiescent: async () => { calls.push('quiescent'); }, assertSafeSshHost: () => {}, @@ -127,7 +152,7 @@ test('runHostSync keeps tracker quiescent through local and remote merge and cle }, }; - assert.throws( + await assert.rejects( () => runSyncCommand(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }), deps), /Remote merge failed on media-box[\s\S]*remote merge exploded/, @@ -139,12 +164,12 @@ test('runHostSync keeps tracker quiescent through local and remote merge and cle assert.equal(fs.existsSync(localTmpDir), false); }); -test('runHostSync includes remote snapshot stderr in failures', () => { +test('runHostSync includes remote snapshot stderr in failures', async () => { const deps: Partial = { createDbSnapshot: (_dbPath: string, outPath: string) => { fs.writeFileSync(outPath, 'snapshot'); }, - ensureTrackerQuiescent: () => {}, + ensureTrackerQuiescent: async () => {}, assertSafeSshHost: () => {}, resolveRemoteSubminerCommand: () => 'subminer', runSsh: (_host: string, command: string) => { @@ -157,7 +182,7 @@ test('runHostSync includes remote snapshot stderr in failures', () => { runScp: () => {}, }; - assert.throws( + await assert.rejects( () => runSyncCommand(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }), deps), /Remote snapshot failed on media-box[\s\S]*snapshot permission denied/, diff --git a/launcher/commands/sync-command.ts b/launcher/commands/sync-command.ts index 2a584bde..3c9ece7e 100644 --- a/launcher/commands/sync-command.ts +++ b/launcher/commands/sync-command.ts @@ -17,6 +17,7 @@ import { shellQuote, } from '../sync/ssh.js'; import { resolvePathMaybe } from '../util.js'; +import { canConnectUnixSocket } from '../mpv.js'; import type { LauncherCommandContext } from './context.js'; import type { RemoteRunResult } from '../sync/ssh.js'; @@ -31,13 +32,13 @@ export interface SyncCommandDeps { runSsh: typeof runSsh; fail: typeof fail; log: typeof log; - existsSync: typeof fs.existsSync; + canConnectUnixSocket: typeof canConnectUnixSocket; realpathSync: typeof fs.realpathSync; mkdtempSync: typeof fs.mkdtempSync; rmSync: typeof fs.rmSync; consoleLog: typeof console.log; writeStdout: typeof process.stdout.write; - ensureTrackerQuiescent: (context: LauncherCommandContext, dbPath: string) => void; + ensureTrackerQuiescent: (context: LauncherCommandContext, dbPath: string) => Promise; } function resolveDbPath(context: LauncherCommandContext): string { @@ -54,11 +55,11 @@ function isTrackerDb(dbPath: string, deps: SyncCommandDeps): boolean { } } -export function ensureTrackerQuiescent( +export async function ensureTrackerQuiescent( context: LauncherCommandContext, dbPath: string, inputDeps: Partial = {}, -): void { +): Promise { const deps = resolveSyncCommandDeps(inputDeps); if (context.args.syncForce) return; // A running SubMiner only holds the tracker's own database; --db pointed @@ -70,7 +71,7 @@ export function ensureTrackerQuiescent( `The SubMiner stats server is running (pid ${daemonPid}). Stop it with "subminer stats -s" (or close SubMiner) before syncing, or pass --force.`, ); } - if (context.mpvSocketPath && deps.existsSync(context.mpvSocketPath)) { + if (context.mpvSocketPath && (await deps.canConnectUnixSocket(context.mpvSocketPath))) { deps.fail( `An mpv/SubMiner session appears to be running (socket ${context.mpvSocketPath}). Close it before syncing, or pass --force.`, ); @@ -88,13 +89,13 @@ const defaultSyncCommandDeps: SyncCommandDeps = { runSsh, fail, log, - existsSync: fs.existsSync, + canConnectUnixSocket, realpathSync: fs.realpathSync, mkdtempSync: fs.mkdtempSync, rmSync: fs.rmSync, consoleLog: console.log, writeStdout: process.stdout.write.bind(process.stdout), - ensureTrackerQuiescent: (context, dbPath) => ensureTrackerQuiescent(context, dbPath), + ensureTrackerQuiescent: async (context, dbPath) => ensureTrackerQuiescent(context, dbPath), }; function resolveSyncCommandDeps(inputDeps: Partial = {}): SyncCommandDeps { @@ -112,13 +113,13 @@ export function runSnapshotMode( deps.consoleLog(outPath); } -export function runMergeMode( +export async function runMergeMode( context: LauncherCommandContext, dbPath: string, inputDeps: Partial = {}, -): void { +): Promise { const deps = resolveSyncCommandDeps(inputDeps); - deps.ensureTrackerQuiescent(context, dbPath); + await deps.ensureTrackerQuiescent(context, dbPath); const snapshotPath = resolvePathMaybe(context.args.syncMergePath); const summary = deps.mergeSnapshotIntoDb(dbPath, snapshotPath); deps.consoleLog(deps.formatMergeSummary(summary)); @@ -134,17 +135,17 @@ function formatRemoteRunError(message: string, run: RemoteRunResult): string { return stderr ? `${message}\n${stderr}` : message; } -export function runHostSync( +export async function runHostSync( context: LauncherCommandContext, dbPath: string, inputDeps: Partial = {}, -): void { +): Promise { const deps = resolveSyncCommandDeps(inputDeps); const { args } = context; const host = args.syncHost; deps.assertSafeSshHost(host); - deps.ensureTrackerQuiescent(context, dbPath); + await deps.ensureTrackerQuiescent(context, dbPath); const remoteCmd = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null); deps.log('debug', args.logLevel, `Remote subminer command: ${remoteCmd}`); @@ -183,12 +184,12 @@ export function runHostSync( deps.runScp(localSnapshot, `${host}:${incomingSnapshot}`); deps.consoleLog(`\nMerging ${host} -> local:`); - deps.ensureTrackerQuiescent(context, dbPath); + await deps.ensureTrackerQuiescent(context, dbPath); const summary = deps.mergeSnapshotIntoDb(dbPath, pulledSnapshot); deps.consoleLog(deps.formatMergeSummary(summary)); deps.consoleLog(`\nMerging local -> ${host}:`); - deps.ensureTrackerQuiescent(context, dbPath); + await deps.ensureTrackerQuiescent(context, dbPath); const mergeRun = deps.runSsh( host, `${remoteCmd} sync --merge ${shellQuote(incomingSnapshot)}${forceFlag}`, @@ -216,10 +217,10 @@ export function runHostSync( } } -export function runSyncCommand( +export async function runSyncCommand( context: LauncherCommandContext, inputDeps: Partial = {}, -): boolean { +): Promise { const deps = resolveSyncCommandDeps(inputDeps); const { args } = context; if (!args.sync) return false; @@ -228,9 +229,9 @@ export function runSyncCommand( if (args.syncSnapshotPath) { runSnapshotMode(context, dbPath, deps); } else if (args.syncMergePath) { - runMergeMode(context, dbPath, deps); + await runMergeMode(context, dbPath, deps); } else if (args.syncHost) { - runHostSync(context, dbPath, deps); + await runHostSync(context, dbPath, deps); } else { deps.fail('sync requires a host, --snapshot , or --merge .'); } diff --git a/launcher/main.ts b/launcher/main.ts index a3b200fc..7dce9bc4 100644 --- a/launcher/main.ts +++ b/launcher/main.ts @@ -108,7 +108,7 @@ async function main(): Promise { return; } - if (runSyncCommand(context)) { + if (await runSyncCommand(context)) { return; } diff --git a/launcher/mpv.ts b/launcher/mpv.ts index 5dd8f42a..d249b9ff 100644 --- a/launcher/mpv.ts +++ b/launcher/mpv.ts @@ -1683,7 +1683,7 @@ async function sleepMs(ms: number): Promise { await new Promise((resolve) => setTimeout(resolve, ms)); } -async function canConnectUnixSocket(socketPath: string): Promise { +export async function canConnectUnixSocket(socketPath: string): Promise { return await new Promise((resolve) => { const socket = net.createConnection(socketPath); let settled = false;