fix(mpv): recover from stalled IPC connects (#204)

src/core/services/mpv-transport.test.ts

@@ -38,7 +38,15 @@ class ManualCloseSocket extends FakeSocket {
   }
 }
 
-const wait = () => new Promise((resolve) => setTimeout(resolve, 0));
+class HangingSocket extends FakeSocket {
+  override connect(path: string): void {
+    this.connectedPaths.push(path);
+    // Never emits 'connect', 'error', or 'close' on its own: models a named
+    // pipe dial that stalls indefinitely.
+  }
+}
+
+const wait = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms));
 
 test('getMpvReconnectDelay follows existing reconnect ramp', () => {
   assert.equal(getMpvReconnectDelay(0, true), 1000);
@@ -232,6 +240,75 @@ test('MpvSocketTransport.shutdown clears socket and lifecycle flags', async () =
   assert.deepEqual(events, []);
 });
 
+test('MpvSocketTransport aborts a hung connect after the timeout and allows a fresh dial', async () => {
+  const events: string[] = [];
+  const errors: Error[] = [];
+  const sockets: HangingSocket[] = [];
+  const transport = new MpvSocketTransport({
+    socketPath: '/tmp/mpv.sock',
+    connectTimeoutMs: 5,
+    onConnect: () => {
+      events.push('connect');
+    },
+    onData: () => {},
+    onError: (error) => {
+      events.push('error');
+      errors.push(error);
+    },
+    onClose: () => {
+      events.push('close');
+    },
+    socketFactory: () => {
+      const socket = new HangingSocket();
+      sockets.push(socket);
+      return socket as unknown as net.Socket;
+    },
+  });
+
+  transport.connect();
+  assert.equal(transport.isConnecting, true);
+
+  await wait(20);
+
+  assert.deepEqual(events, ['error', 'close']);
+  assert.match(errors[0]!.message, /connect timed out/);
+  assert.equal(sockets[0]!.destroyed, true);
+  assert.equal(transport.isConnecting, false);
+  assert.equal(transport.isConnected, false);
+
+  transport.connect();
+  assert.equal(transport.isConnecting, true);
+  assert.equal(sockets.length, 2);
+  assert.equal(sockets[1]!.connectedPaths.at(0), '/tmp/mpv.sock');
+
+  transport.shutdown();
+});
+
+test('MpvSocketTransport does not fire the connect timeout after a successful connect', async () => {
+  const events: string[] = [];
+  const transport = new MpvSocketTransport({
+    socketPath: '/tmp/mpv.sock',
+    connectTimeoutMs: 5,
+    onConnect: () => {
+      events.push('connect');
+    },
+    onData: () => {},
+    onError: () => {
+      events.push('error');
+    },
+    onClose: () => {
+      events.push('close');
+    },
+    socketFactory: () => new FakeSocket() as unknown as net.Socket,
+  });
+
+  transport.connect();
+  await wait(20);
+
+  assert.deepEqual(events, ['connect']);
+  assert.equal(transport.isConnected, true);
+});
+
 test('MpvSocketTransport ignores stale socket events after shutdown and reconnect', async () => {
   const events: string[] = [];
   const sockets: ManualCloseSocket[] = [];

src/core/services/mpv-transport.ts

@@ -62,6 +62,8 @@ interface MpvSocketTransportEvents {
   onClose: () => void;
 }
 
+export const MPV_CONNECT_TIMEOUT_MS = 5000;
+
 export interface MpvSocketTransportOptions {
   socketPath: string;
   onConnect: () => void;
@@ -69,13 +71,16 @@ export interface MpvSocketTransportOptions {
   onError: (error: Error) => void;
   onClose: () => void;
   socketFactory?: () => net.Socket;
+  connectTimeoutMs?: number;
 }
 
 export class MpvSocketTransport {
   private socketPath: string;
   private readonly callbacks: MpvSocketTransportEvents;
   private readonly socketFactory: () => net.Socket;
+  private readonly connectTimeoutMs: number;
   private socketRef: net.Socket | null = null;
+  private connectTimer: ReturnType<typeof setTimeout> | null = null;
   public socket: net.Socket | null = null;
   public connected = false;
   public connecting = false;
@@ -83,6 +88,7 @@ export class MpvSocketTransport {
   constructor(options: MpvSocketTransportOptions) {
     this.socketPath = options.socketPath;
     this.socketFactory = options.socketFactory ?? (() => new net.Socket());
+    this.connectTimeoutMs = options.connectTimeoutMs ?? MPV_CONNECT_TIMEOUT_MS;
     this.callbacks = {
       onConnect: options.onConnect,
       onData: options.onData,
@@ -91,6 +97,31 @@ export class MpvSocketTransport {
     };
   }
 
+  private clearConnectTimeout(): void {
+    if (this.connectTimer) {
+      clearTimeout(this.connectTimer);
+      this.connectTimer = null;
+    }
+  }
+
+  // A named-pipe/socket dial that neither connects nor errors would otherwise
+  // latch `connecting` forever and silently block every future connect().
+  private armConnectTimeout(socket: net.Socket): void {
+    this.clearConnectTimeout();
+    this.connectTimer = setTimeout(() => {
+      this.connectTimer = null;
+      if (this.socketRef !== socket || this.connected) return;
+      this.connecting = false;
+      this.callbacks.onError(
+        new Error(`MPV IPC connect timed out after ${this.connectTimeoutMs}ms: ${this.socketPath}`),
+      );
+      // Destroying the socket emits 'close', which drives the normal
+      // disconnect path (including reconnect scheduling) upstream.
+      socket.destroy();
+    }, this.connectTimeoutMs);
+    this.connectTimer.unref?.();
+  }
+
   setSocketPath(socketPath: string): void {
     this.socketPath = socketPath;
   }
@@ -111,6 +142,7 @@ export class MpvSocketTransport {
 
     socket.on('connect', () => {
       if (this.socketRef !== socket) return;
+      this.clearConnectTimeout();
       this.connected = true;
       this.connecting = false;
       this.callbacks.onConnect();
@@ -123,6 +155,7 @@ export class MpvSocketTransport {
 
     socket.on('error', (error: Error) => {
       if (this.socketRef !== socket) return;
+      this.clearConnectTimeout();
       this.connected = false;
       this.connecting = false;
       this.callbacks.onError(error);
@@ -130,12 +163,14 @@ export class MpvSocketTransport {
 
     socket.on('close', () => {
       if (this.socketRef !== socket) return;
+      this.clearConnectTimeout();
       this.connected = false;
       this.connecting = false;
       this.callbacks.onClose();
     });
 
     socket.connect(this.socketPath);
+    this.armConnectTimeout(socket);
   }
 
   send(payload: MpvSocketMessagePayload): boolean {
@@ -149,6 +184,7 @@ export class MpvSocketTransport {
   }
 
   shutdown(): void {
+    this.clearConnectTimeout();
     const socket = this.socketRef;
     this.socketRef = null;
     this.socket = null;

src/core/services/mpv.test.ts

@@ -1,5 +1,6 @@
 import test from 'node:test';
 import assert from 'node:assert/strict';
+import { EventEmitter } from 'node:events';
 import {
   MpvIpcClient,
   MpvIpcClientDeps,
@@ -23,6 +24,18 @@ function makeDeps(overrides: Partial<MpvIpcClientProtocolDeps> = {}): MpvIpcClie
   };
 }
 
+const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (!predicate()) {
+    if (Date.now() >= deadline) {
+      throw new Error('Timed out waiting for MPV retry connection');
+    }
+    await wait(10);
+  }
+}
+
 function captureWarnLogs(run: () => void): string[] {
   const originalWarn = console.warn;
   const originalLogLevel = process.env.SUBMINER_LOG_LEVEL;
@@ -756,3 +769,117 @@ test('MpvIpcClient playNextSubtitle still auto-pauses at end while already playi
   assert.equal((client as any).pendingPauseAtSubEnd, true);
   assert.deepEqual(commands, [{ command: ['sub-seek', 1] }]);
 });
+
+class HangingTestSocket extends EventEmitter {
+  public connectedPaths: string[] = [];
+  public destroyed = false;
+
+  connect(path: string): void {
+    this.connectedPaths.push(path);
+    // Never resolves: models a stalled named-pipe dial.
+  }
+
+  write(): boolean {
+    return true;
+  }
+
+  destroy(): void {
+    this.destroyed = true;
+  }
+}
+
+class RetryTestSocket extends EventEmitter {
+  public connectedPaths: string[] = [];
+  public destroyed = false;
+
+  constructor(private readonly shouldConnect: boolean) {
+    super();
+  }
+
+  connect(path: string): void {
+    this.connectedPaths.push(path);
+    if (this.shouldConnect) {
+      setTimeout(() => this.emit('connect'), 0);
+    }
+  }
+
+  write(): boolean {
+    return true;
+  }
+
+  destroy(): void {
+    if (this.destroyed) return;
+    this.destroyed = true;
+    this.emit('close');
+  }
+}
+
+test('MpvIpcClient automatically retries the same socket path after a connect timeout', async () => {
+  const sockets: RetryTestSocket[] = [];
+  let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  const originalLogLevel = process.env.SUBMINER_LOG_LEVEL;
+  const client = new MpvIpcClient(
+    '/tmp/mpv.sock',
+    makeDeps({
+      connectTimeoutMs: 5,
+      getReconnectTimer: () => reconnectTimer,
+      setReconnectTimer: (timer) => {
+        reconnectTimer = timer;
+      },
+      socketFactory: () => {
+        const socket = new RetryTestSocket(sockets.length > 0);
+        sockets.push(socket);
+        return socket as unknown as import('node:net').Socket;
+      },
+    }),
+  );
+
+  process.env.SUBMINER_LOG_LEVEL = 'error';
+  try {
+    client.connect();
+    await waitFor(() => client.connected);
+
+    assert.equal(sockets.length, 2);
+    assert.equal(sockets[0]!.destroyed, true);
+    assert.equal(sockets[0]!.connectedPaths.at(0), '/tmp/mpv.sock');
+    assert.equal(sockets[1]!.connectedPaths.at(0), '/tmp/mpv.sock');
+    assert.equal(client.connected, true);
+  } finally {
+    if (originalLogLevel === undefined) {
+      delete process.env.SUBMINER_LOG_LEVEL;
+    } else {
+      process.env.SUBMINER_LOG_LEVEL = originalLogLevel;
+    }
+    if (reconnectTimer) clearTimeout(reconnectTimer);
+    (client as any).transport.shutdown();
+  }
+});
+
+test('MpvIpcClient.setSocketPath aborts an in-flight connect so the next dial targets the new path', () => {
+  const sockets: HangingTestSocket[] = [];
+  const client = new MpvIpcClient(
+    '/tmp/mpv-old.sock',
+    makeDeps({
+      socketFactory: () => {
+        const socket = new HangingTestSocket();
+        sockets.push(socket);
+        return socket as unknown as import('node:net').Socket;
+      },
+    }),
+  );
+
+  client.connect();
+  assert.equal(sockets.length, 1);
+  assert.equal(sockets[0]!.connectedPaths.at(0), '/tmp/mpv-old.sock');
+  assert.equal((client as any).connecting, true);
+
+  client.setSocketPath('/tmp/mpv-new.sock');
+  assert.equal((client as any).connecting, false);
+  assert.equal(sockets[0]!.destroyed, true);
+
+  client.connect();
+  assert.equal(sockets.length, 2);
+  assert.equal(sockets[1]!.connectedPaths.at(0), '/tmp/mpv-new.sock');
+
+  (client as any).transport.shutdown();
+});

src/core/services/mpv.ts

@@ -9,7 +9,11 @@ import {
   splitMpvMessagesFromBuffer,
 } from './mpv-protocol';
 import { requestMpvInitialState, subscribeToMpvProperties } from './mpv-properties';
-import { scheduleMpvReconnect, MpvSocketTransport } from './mpv-transport';
+import {
+  scheduleMpvReconnect,
+  MpvSocketTransport,
+  MpvSocketTransportOptions,
+} from './mpv-transport';
 import { createLogger } from '../../logger';
 
 const logger = createLogger('main:mpv');
@@ -110,6 +114,8 @@ export interface MpvIpcClientProtocolDeps {
   shouldAutoLoadSecondarySubTrack?: (path: string) => boolean;
   shouldQuitOnMpvShutdown?: () => boolean;
   requestAppQuit?: () => void;
+  socketFactory?: MpvSocketTransportOptions['socketFactory'];
+  connectTimeoutMs?: number;
 }
 
 export interface MpvIpcClientDeps extends MpvIpcClientProtocolDeps {}
@@ -188,6 +194,8 @@ export class MpvIpcClient implements MpvClient {
 
     this.transport = new MpvSocketTransport({
       socketPath,
+      socketFactory: deps.socketFactory,
+      connectTimeoutMs: deps.connectTimeoutMs,
       onConnect: () => {
         this.connected = true;
         this.connecting = false;
@@ -289,6 +297,14 @@ export class MpvIpcClient implements MpvClient {
         previousSocketPath: this.socketPath,
         socketPath,
       });
+      if (this.connecting && !this.connected) {
+        // Abort the in-flight dial to the old path; otherwise the connecting
+        // latch turns every later connect() into a no-op while we hang on a
+        // stale socket.
+        logger.debug('Aborting in-flight MPV IPC connect for socket path change.');
+        this.transport.shutdown();
+        this.connecting = false;
+      }
     }
     this.socketPath = socketPath;
     this.transport.setSocketPath(socketPath);