diff --git a/.changes/playback-reused-mpv-command-replies.md b/.changes/playback-reused-mpv-command-replies.md new file mode 100644 index 000000000..806c3a8fb --- /dev/null +++ b/.changes/playback-reused-mpv-command-replies.md @@ -0,0 +1,9 @@ +--- +type: fix +area: playback +--- + +With "Reuse player instance" on, IPTVnator now waits for MPV to confirm each +command sent to the running player. A command MPV rejects or never answers +falls back to a fresh MPV launch instead of reporting the stream as opened +while nothing changed. diff --git a/apps/electron-backend/src/app/events/external-player-session-registry.spec.ts b/apps/electron-backend/src/app/events/external-player-session-registry.spec.ts index 21ea3865d..98781dca7 100644 --- a/apps/electron-backend/src/app/events/external-player-session-registry.spec.ts +++ b/apps/electron-backend/src/app/events/external-player-session-registry.spec.ts @@ -30,6 +30,26 @@ describe('ExternalPlayerSessionRegistry', () => { expect(updates.at(-1)?.status).toBe('opened'); }); + it('numbers sessions in the order they began', () => { + const first = registry.beginSession({ + player: 'mpv', + title: 'First', + streamUrl: 'https://example.com/one.m3u8', + }); + const second = registry.beginSession({ + player: 'mpv', + title: 'Second', + streamUrl: 'https://example.com/two.m3u8', + }); + + expect(first.launchOrder).toEqual(expect.any(Number)); + expect(second.launchOrder).toBeGreaterThan(first.launchOrder ?? 0); + // Later updates keep the order the session began with. + expect(registry.markOpened(first.id)?.launchOrder).toBe( + first.launchOrder + ); + }); + it('keeps close capability and closes the session explicitly', async () => { const close = jest.fn(); const session = registry.beginSession({ @@ -144,9 +164,9 @@ describe('ExternalPlayerSessionRegistry', () => { registry.markClosed(closed.id); expect(registry.markPlaying(closed.id)?.status).toBe('closed'); - expect( - registry.markError(closed.id, 'Late process exit')?.status - ).toBe('closed'); + expect(registry.markError(closed.id, 'Late process exit')?.status).toBe( + 'closed' + ); expect(updates.at(-1)).toMatchObject({ id: closed.id, status: 'closed', diff --git a/apps/electron-backend/src/app/events/external-player-session-registry.ts b/apps/electron-backend/src/app/events/external-player-session-registry.ts index a20511330..cfe485c67 100644 --- a/apps/electron-backend/src/app/events/external-player-session-registry.ts +++ b/apps/electron-backend/src/app/events/external-player-session-registry.ts @@ -44,6 +44,7 @@ function isRestorableSession(session: ExternalPlayerSession): boolean { export class ExternalPlayerSessionRegistry { private readonly sessions = new Map(); private activeSessionId: string | null = null; + private launchCount = 0; constructor( private readonly onUpdate: (session: ExternalPlayerSession) => void @@ -63,6 +64,7 @@ export class ExternalPlayerSessionRegistry { contentInfo: options.contentInfo, startedAt, updatedAt: startedAt, + launchOrder: ++this.launchCount, canClose: false, }; diff --git a/apps/electron-backend/src/app/events/mpv-ipc-command.spec.ts b/apps/electron-backend/src/app/events/mpv-ipc-command.spec.ts new file mode 100644 index 000000000..3a77303cd --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-ipc-command.spec.ts @@ -0,0 +1,298 @@ +jest.mock('net', () => ({ + createConnection: jest.fn(), +})); + +jest.mock('./external-player-runtime', () => ({ + traceExternalPlayer: jest.fn(), +})); + +import { EventEmitter } from 'events'; +import { createConnection } from 'net'; +import { + MPV_IPC_COMMAND_TIMEOUT_MS, + MpvIpcCommandError, + sendMpvCommand, +} from './mpv-ipc-command'; + +type SocketMock = EventEmitter & { + write: jest.Mock; + destroy: jest.Mock; + destroyed: boolean; +}; + +const createConnectionMock = createConnection as unknown as jest.Mock; +const socketPath = '/tmp/mpv-ipc-command.sock'; +const streamUrl = 'https://example.com/stream.m3u8?token=secret-token'; + +function nextSocket(): SocketMock { + const socket: SocketMock = Object.assign(new EventEmitter(), { + write: jest.fn(() => true), + destroy: jest.fn(() => { + socket.destroyed = true; + }), + destroyed: false, + }); + createConnectionMock.mockReturnValueOnce(socket); + return socket; +} + +function writtenRequest(socket: SocketMock): { + command: Array; + request_id: number; +} { + return JSON.parse(String(socket.write.mock.calls[0][0]).trim()); +} + +function reply(socket: SocketMock, ...lines: Array): void { + socket.emit( + 'data', + Buffer.from( + lines + .map((line) => + typeof line === 'string' ? line : JSON.stringify(line) + ) + .join('\n') + '\n' + ) + ); +} + +async function settle(): Promise { + await new Promise((resolve) => setImmediate(resolve)); +} + +describe('sendMpvCommand', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('sends a request_id and resolves with the matching reply', async () => { + const socket = nextSocket(); + const pending = sendMpvCommand(socketPath, 'loadfile', [ + streamUrl, + 'replace', + -1, + 'start=0', + ]); + socket.emit('connect'); + + const request = writtenRequest(socket); + expect(request.command).toEqual([ + 'loadfile', + streamUrl, + 'replace', + -1, + 'start=0', + ]); + expect(Number.isInteger(request.request_id)).toBe(true); + expect(String(socket.write.mock.calls[0][0]).endsWith('\n')).toBe(true); + + reply(socket, { + data: { playlist_entry_id: 1 }, + request_id: request.request_id, + error: 'success', + }); + await expect(pending).resolves.toEqual({ + request_id: request.request_id, + error: 'success', + data: { playlist_entry_id: 1 }, + }); + expect(socket.destroy).toHaveBeenCalledTimes(1); + }); + + it('gives consecutive commands distinct request ids', async () => { + const first = nextSocket(); + const second = nextSocket(); + const firstPending = sendMpvCommand(socketPath, 'set_property', [ + 'user-agent', + 'agent', + ]); + const secondPending = sendMpvCommand(socketPath, 'set_property', [ + 'referrer', + 'https://example.com/', + ]); + first.emit('connect'); + second.emit('connect'); + const firstId = writtenRequest(first).request_id; + const secondId = writtenRequest(second).request_id; + expect(secondId).not.toBe(firstId); + + reply(first, { request_id: firstId, error: 'success' }); + reply(second, { request_id: secondId, error: 'success' }); + await expect(firstPending).resolves.toMatchObject({ + request_id: firstId, + }); + await expect(secondPending).resolves.toMatchObject({ + request_id: secondId, + }); + }); + + it('resolves false without writing when dispatch is vetoed', async () => { + const socket = nextSocket(); + const onDispatched = jest.fn(); + const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl], { + shouldDispatch: () => false, + onDispatched, + }); + socket.emit('connect'); + + await expect(pending).resolves.toBe(false); + expect(socket.write).not.toHaveBeenCalled(); + expect(onDispatched).not.toHaveBeenCalled(); + expect(socket.destroy).toHaveBeenCalledTimes(1); + }); + + it('rejects with mpv error text after the command was dispatched', async () => { + const socket = nextSocket(); + const onDispatched = jest.fn(); + const pending = sendMpvCommand(socketPath, 'seek', ['10', 'absolute'], { + onDispatched, + }); + socket.emit('connect'); + expect(onDispatched).toHaveBeenCalledTimes(1); + + reply(socket, { + request_id: writtenRequest(socket).request_id, + error: 'error running command', + }); + await expect(pending).rejects.toMatchObject({ + name: 'MpvIpcCommandError', + command: 'seek', + dispatched: true, + mpvError: 'error running command', + message: 'mpv seek: error running command', + }); + expect(socket.destroy).toHaveBeenCalledTimes(1); + }); + + it.each([ + ['a missing error', {}], + ['a non-string error', { error: 0 }], + ])( + 'rejects a matching reply with %s instead of confirming it', + async (_label, fields) => { + const socket = nextSocket(); + const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]); + socket.emit('connect'); + + reply(socket, { + request_id: writtenRequest(socket).request_id, + data: { playlist_entry_id: 1 }, + ...fields, + }); + await expect(pending).rejects.toMatchObject({ + name: 'MpvIpcCommandError', + dispatched: true, + mpvError: null, + message: 'mpv loadfile: malformed reply', + }); + } + ); + + it('rejects an undispatched command on a socket error without echoing arguments', async () => { + const socket = nextSocket(); + const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]); + socket.emit('error', new Error('connect ENOENT')); + + const error = (await pending.catch((caught) => caught)) as unknown; + expect(error).toBeInstanceOf(MpvIpcCommandError); + expect(error).toMatchObject({ + dispatched: false, + mpvError: null, + message: 'mpv loadfile: connect ENOENT', + }); + expect(socket.write).not.toHaveBeenCalled(); + expect(String((error as Error).message)).not.toContain('secret'); + }); + + it('rejects when the socket closes before a reply', async () => { + const socket = nextSocket(); + const pending = sendMpvCommand(socketPath, 'quit', []); + socket.emit('connect'); + socket.emit('close'); + + await expect(pending).rejects.toMatchObject({ + command: 'quit', + dispatched: true, + message: 'mpv quit: socket closed before reply', + }); + }); + + it('times out a reply that never comes', async () => { + jest.useFakeTimers(); + const socket = nextSocket(); + const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]); + socket.emit('connect'); + let settled = false; + void pending.catch(() => (settled = true)); + + await jest.advanceTimersByTimeAsync(MPV_IPC_COMMAND_TIMEOUT_MS - 1); + expect(settled).toBe(false); + await jest.advanceTimersByTimeAsync(1); + expect(settled).toBe(true); + await expect(pending).rejects.toMatchObject({ + dispatched: true, + message: 'mpv loadfile: reply timed out', + }); + expect(socket.destroy).toHaveBeenCalledTimes(1); + }); + + it('skips events, foreign replies and malformed lines, and joins split chunks', async () => { + const socket = nextSocket(); + const pending = sendMpvCommand(socketPath, 'loadfile', [streamUrl]); + socket.emit('connect'); + const requestId = writtenRequest(socket).request_id; + let settled = false; + void pending.then( + () => (settled = true), + () => (settled = true) + ); + + reply( + socket, + { event: 'start-file', playlist_entry_id: 1 }, + { + event: 'end-file', + reason: 'error', + file_error: 'loading failed', + }, + { request_id: requestId + 1, error: 'error running command' }, + 'not json', + '42' + ); + await settle(); + expect(settled).toBe(false); + + const ownReply = JSON.stringify({ + request_id: requestId, + error: 'success', + }); + socket.emit('data', Buffer.from(ownReply.slice(0, 10))); + await settle(); + expect(settled).toBe(false); + socket.emit('data', Buffer.from(ownReply.slice(10) + '\n')); + + await expect(pending).resolves.toMatchObject({ + request_id: requestId, + error: 'success', + }); + }); + + it('ignores data that arrives after settling', async () => { + const socket = nextSocket(); + const pending = sendMpvCommand(socketPath, 'quit', []); + socket.emit('connect'); + const requestId = writtenRequest(socket).request_id; + reply(socket, { request_id: requestId, error: 'success' }); + await expect(pending).resolves.toMatchObject({ error: 'success' }); + + expect(() => + reply(socket, { request_id: requestId, error: 'late failure' }) + ).not.toThrow(); + expect(() => socket.emit('close')).not.toThrow(); + expect(socket.destroy).toHaveBeenCalledTimes(1); + }); +}); diff --git a/apps/electron-backend/src/app/events/mpv-ipc-command.ts b/apps/electron-backend/src/app/events/mpv-ipc-command.ts new file mode 100644 index 000000000..61560455b --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-ipc-command.ts @@ -0,0 +1,158 @@ +import { createConnection } from 'net'; +import { traceExternalPlayer } from './external-player-runtime'; + +/** + * Bounds connect, write and the wait for mpv's reply. It matches the IPC + * quit bound the reuse teardown relies on before its exit windows start. + */ +export const MPV_IPC_COMMAND_TIMEOUT_MS = 2_000; + +/** One newline-delimited JSON reply of mpv's IPC protocol. */ +export interface MpvIpcReply { + request_id: number; + error: string; + data?: unknown; +} + +export interface MpvIpcCommandOptions { + /** Checked once connected; a `false` skips the write and resolves `false`. */ + shouldDispatch?: () => boolean; + /** Runs right after the command was written, before any reply. */ + onDispatched?: () => void; +} + +/** + * A command mpv answered with an error, or one whose reply never came. + * `dispatched` tells whether the command reached the socket, so callers can + * treat the player state as possibly changed. The message never carries + * command arguments, which may include stream URLs. + */ +export class MpvIpcCommandError extends Error { + constructor( + readonly command: string, + message: string, + readonly dispatched: boolean, + readonly mpvError: string | null = null + ) { + super(`mpv ${command}: ${message}`); + this.name = 'MpvIpcCommandError'; + } +} + +let nextRequestId = 1; + +function parseReplyLine(line: string): Partial | null { + try { + const parsed: unknown = JSON.parse(line); + return parsed && typeof parsed === 'object' + ? (parsed as Partial) + : null; + } catch { + return null; + } +} + +/** + * Sends one command to a reused mpv over its IPC socket and resolves with + * mpv's matching reply. Every command carries a `request_id`; event lines + * and replies to other requests on the same connection are skipped. Resolves + * `false` when `shouldDispatch` vetoed the write, and rejects with + * {@link MpvIpcCommandError} on an error reply, a missing reply, or a socket + * failure. + */ +export function sendMpvCommand( + socketPath: string, + command: string, + args: Array, + options: MpvIpcCommandOptions = {} +): Promise { + return new Promise((resolve, reject) => { + const requestId = nextRequestId++; + const client = createConnection(socketPath); + const request = + JSON.stringify({ + command: [command, ...args], + request_id: requestId, + }) + '\n'; + let settled = false; + let dispatched = false; + let buffered = ''; + let timeoutHandle: NodeJS.Timeout | null = null; + const settle = ( + outcome: { reply: MpvIpcReply | false } | { error: Error } + ) => { + if (settled) return; + settled = true; + if (timeoutHandle) clearTimeout(timeoutHandle); + if (!client.destroyed) client.destroy(); + if ('error' in outcome) { + reject(outcome.error); + } else { + resolve(outcome.reply); + } + }; + const fail = (message: string, mpvError: string | null = null) => + settle({ + error: new MpvIpcCommandError( + command, + message, + dispatched, + mpvError + ), + }); + const consumeLine = (line: string) => { + const reply = parseReplyLine(line); + if (!reply || reply.request_id !== requestId) return; + // mpv always answers with a string `error`; anything else is not + // a confirmation and must not skip the fallback. + if (typeof reply.error !== 'string') { + fail('malformed reply'); + return; + } + const error = reply.error; + traceExternalPlayer('mpv ipc reply', { command, error }); + if (error === 'success') { + settle({ + reply: { request_id: requestId, error, data: reply.data }, + }); + } else { + fail(error, error); + } + }; + + client.on('connect', () => { + if (options.shouldDispatch && !options.shouldDispatch()) { + settle({ reply: false }); + return; + } + traceExternalPlayer('mpv ipc command', { + command, + argsCount: args.length, + }); + try { + client.write(request); + dispatched = true; + options.onDispatched?.(); + } catch (error) { + fail(error instanceof Error ? error.message : String(error)); + } + }); + client.on('data', (chunk) => { + if (settled) return; + buffered += chunk.toString(); + const lines = buffered.split('\n'); + buffered = lines.pop() ?? ''; + for (const line of lines) { + if (line.trim()) consumeLine(line); + if (settled) return; + } + }); + client.on('error', (error) => fail(error.message)); + client.on('close', () => fail('socket closed before reply')); + timeoutHandle = setTimeout( + () => fail('reply timed out'), + MPV_IPC_COMMAND_TIMEOUT_MS + ); + timeoutHandle.unref(); + }); +} diff --git a/apps/electron-backend/src/app/events/mpv-ipc-socket.test-helpers.ts b/apps/electron-backend/src/app/events/mpv-ipc-socket.test-helpers.ts new file mode 100644 index 000000000..7611a68e1 --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-ipc-socket.test-helpers.ts @@ -0,0 +1,55 @@ +import { EventEmitter } from 'events'; + +export type MpvSocketMock = EventEmitter & { + write: jest.Mock; + end: jest.Mock; + destroy: jest.Mock; +}; + +export interface MpvSocketMockOptions { + /** Collects every chunk written to the socket. */ + written?: string[]; + /** Emit `connect` on the next tick; `false` leaves that to the test. */ + connect?: boolean; + /** + * mpv's `error` field for a written command, `'success'` by default. + * `null` leaves the command unanswered. + */ + reply?: (command: Array) => string | null; +} + +interface MpvIpcRequest { + command: Array; + request_id: number; +} + +/** + * A reused-instance mpv IPC socket: records writes and answers each command + * on the next tick with a reply carrying its `request_id`, as mpv does. + */ +export function createMpvSocketMock({ + written, + connect = true, + reply = () => 'success', +}: MpvSocketMockOptions = {}): MpvSocketMock { + const socket: MpvSocketMock = Object.assign(new EventEmitter(), { + write: jest.fn((chunk: string) => { + written?.push(chunk); + const request = JSON.parse(chunk.trim()) as MpvIpcRequest; + const error = reply(request.command); + if (error !== null) { + const line = + JSON.stringify({ request_id: request.request_id, error }) + + '\n'; + setImmediate(() => socket.emit('data', Buffer.from(line))); + } + return true; + }), + end: jest.fn(), + destroy: jest.fn(), + }); + if (connect) { + setImmediate(() => socket.emit('connect')); + } + return socket; +} diff --git a/apps/electron-backend/src/app/events/mpv-launch-queue.spec.ts b/apps/electron-backend/src/app/events/mpv-launch-queue.spec.ts new file mode 100644 index 000000000..756293cad --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-launch-queue.spec.ts @@ -0,0 +1,856 @@ +/** + * MPV launch queue: launches run one at a time, only the newest queued one + * runs, and Stop on a queued launch delegates to the in-flight launch's and + * the stored player's own closers. Reply handling lives in + * `mpv-session.service.ipc-reply.spec.ts`. + */ +jest.mock('electron', () => ({ + ipcMain: { + handle: jest.fn(), + }, +})); + +jest.mock('child_process', () => ({ + spawn: jest.fn(), +})); + +jest.mock('net', () => ({ + createConnection: jest.fn(), + createServer: jest.fn(), +})); + +jest.mock('../app', () => ({ + __esModule: true, + default: { + mainWindow: null, + }, +})); + +jest.mock('../services/store.service', () => ({ + MPV_PLAYER_ARGUMENTS: 'MPV_PLAYER_ARGUMENTS', + MPV_PLAYER_PATH: 'MPV_PLAYER_PATH', + MPV_REUSE_INSTANCE: 'MPV_REUSE_INSTANCE', + store: { + get: jest.fn(), + set: jest.fn(), + }, +})); + +import { + MPV_PLAYER_PATH, + MPV_REUSE_INSTANCE, + store, +} from '../services/store.service'; +import { externalPlayerSessions } from './external-player-runtime'; +import { MPV_IPC_COMMAND_TIMEOUT_MS } from './mpv-ipc-command'; +import { MpvSocketMock } from './mpv-ipc-socket.test-helpers'; +import { + openMpvPlayer, + setMpvReuseInstance, + shutdownMpvSession, +} from './mpv-session.service'; +import { + spawnMock, + createMockChildProcess, + mockMpvSockets, + nextTick, + launchReusableSession, + mockUnansweredSockets, + mockSocketsQuittingInto, + commandsOf, + writtenRequestId, +} from './mpv-session.test-helpers'; + +describe('MPV launch queue', () => { + beforeEach(() => { + jest.clearAllMocks(); + shutdownMpvSession(); + }); + + afterEach(() => { + jest.useRealTimers(); + shutdownMpvSession(); + }); + it('runs a quick second launch only after the first one settles', async () => { + const { proc } = await launchReusableSession(); + const sockets = mockUnansweredSockets(); + const commandOf = (socket: MpvSocketMock) => + ( + JSON.parse(String(socket.write.mock.calls[0][0])) as { + command: string[]; + } + ).command[0]; + const answer = (socket: MpvSocketMock, id: number) => + socket.emit( + 'data', + Buffer.from( + JSON.stringify({ request_id: id, error: 'success' }) + '\n' + ) + ); + + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const firstId = await writtenRequestId(sockets[0]); + const second = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + userAgent: 'IPTVnator test agent', + }); + await nextTick(); + await nextTick(); + // The second launch waits in the queue: nothing else reached mpv. + expect(sockets).toHaveLength(1); + + answer(sockets[0], firstId); + await expect(first).resolves.toMatchObject({ status: 'opened' }); + answer(sockets[1], await writtenRequestId(sockets[1])); + while (sockets.length < 3) await nextTick(); + answer(sockets[2], await writtenRequestId(sockets[2])); + await expect(second).resolves.toMatchObject({ status: 'opened' }); + + expect(sockets.map(commandOf)).toEqual([ + 'loadfile', + 'set_property', + 'loadfile', + ]); + expect(proc.kill).not.toHaveBeenCalled(); + expect(spawnMock).toHaveBeenCalledTimes(1); + }); + + it('lets Stop on a queued launch cancel it and stop the in-flight player', async () => { + const { proc } = await launchReusableSession(); + const sockets = mockSocketsQuittingInto(proc); + + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + await writtenRequestId(sockets[0]); + const second = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + // The queued launch is already the session the dock shows and stops. + const queuedId = externalPlayerSessions.getActiveSessionId() as string; + expect(externalPlayerSessions.getSession(queuedId)).toMatchObject({ + title: 'Channel B', + status: 'launching', + canClose: true, + }); + + // Stop goes through the in-flight launch's own closer: quit, exit. + await expect( + externalPlayerSessions.closeSession(queuedId) + ).resolves.toMatchObject({ id: queuedId, status: 'closed' }); + expect(proc.exitCode).toBe(0); + // The quit player's socket closes before any reply. + sockets[0].emit('close'); + + await expect(first).resolves.toMatchObject({ status: 'closed' }); + await expect(second).resolves.toMatchObject({ + id: queuedId, + status: 'closed', + }); + // Nothing started playing again after Stop. + expect(commandsOf(sockets)).toEqual(['loadfile', 'quit']); + expect(spawnMock).toHaveBeenCalledTimes(1); + }); + + it('lets Stop on the newest queued launch cancel every launch queued before it', async () => { + const { proc } = await launchReusableSession(); + const sockets = mockSocketsQuittingInto(proc); + + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + await writtenRequestId(sockets[0]); + const second = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const third = openMpvPlayer({ + title: 'Channel C', + url: 'https://example.com/c.m3u8', + }); + const newestId = externalPlayerSessions.getActiveSessionId() as string; + + await expect( + externalPlayerSessions.closeSession(newestId) + ).resolves.toMatchObject({ status: 'closed' }); + sockets[0].emit('close'); + + await expect(first).resolves.toMatchObject({ status: 'closed' }); + await expect(second).resolves.toMatchObject({ status: 'closed' }); + await expect(third).resolves.toMatchObject({ status: 'closed' }); + expect(proc.exitCode).toBe(0); + expect(commandsOf(sockets)).toEqual(['loadfile', 'quit']); + expect(spawnMock).toHaveBeenCalledTimes(1); + }); + + it('hands the dock back to the live player when a queued launch is blocked by its teardown', async () => { + const { proc } = await launchReusableSession(); + // The first launch's loadfile fails and its teardown never sees the + // child exit, so the queued launch finds the teardown gate closed. + (proc.kill as jest.Mock).mockImplementation(() => true); + mockMpvSockets({ + reply: (command) => + command[0] === 'loadfile' ? 'error running command' : 'success', + }); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + jest.useFakeTimers(); + + try { + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const firstId = + externalPlayerSessions.getActiveSessionId() as string; + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + const second = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const secondId = + externalPlayerSessions.getActiveSessionId() as string; + const secondRejection = expect(second).rejects.toThrow( + 'previous external player is still shutting down' + ); + + await jest.advanceTimersByTimeAsync(6_000); + await firstRejection; + await secondRejection; + + // The session owning the still-running child is active again + // and can be stopped; the blocked launch is a terminal error. + expect(externalPlayerSessions.getActiveSessionId()).toBe(firstId); + expect(externalPlayerSessions.getSession(firstId)).toMatchObject({ + status: 'error', + canClose: true, + }); + expect(externalPlayerSessions.getSession(secondId)).toMatchObject({ + status: 'error', + canClose: false, + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + }); + + it('runs only the newest queued launch once the in-flight one settles', async () => { + await launchReusableSession(); + const sockets = mockUnansweredSockets(); + const answer = (socket: MpvSocketMock, id: number) => + socket.emit( + 'data', + Buffer.from( + JSON.stringify({ request_id: id, error: 'success' }) + '\n' + ) + ); + + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const firstId = await writtenRequestId(sockets[0]); + const skipped = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const newest = openMpvPlayer({ + title: 'Channel C', + url: 'https://example.com/c.m3u8', + }); + const newestId = externalPlayerSessions.getActiveSessionId(); + + answer(sockets[0], firstId); + await expect(first).resolves.toMatchObject({ status: 'opened' }); + // The skipped launch never reached mpv nor took the dock. + await expect(skipped).resolves.toMatchObject({ status: 'closed' }); + expect(externalPlayerSessions.getActiveSessionId()).toBe(newestId); + answer(sockets[1], await writtenRequestId(sockets[1])); + await expect(newest).resolves.toMatchObject({ + id: newestId, + status: 'opened', + }); + expect(sockets).toHaveLength(2); + expect(spawnMock).toHaveBeenCalledTimes(1); + }); + + it('keeps a queued launch stoppable when its Stop cannot confirm the exit', async () => { + const { proc } = await launchReusableSession(); + (proc.kill as jest.Mock).mockImplementation(() => true); + mockUnansweredSockets(); + jest.useFakeTimers(); + + try { + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const queued = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const queuedId = + externalPlayerSessions.getActiveSessionId() as string; + await jest.advanceTimersByTimeAsync(0); + + // Stop goes through the in-flight launch's closer: its quit gets + // no reply, and the bounded teardown cannot confirm the exit. + const closing = externalPlayerSessions.closeSession(queuedId); + const closeRejection = expect(closing).rejects.toThrow( + 'External player process did not exit' + ); + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + await jest.advanceTimersByTimeAsync(10_000); + await closeRejection; + await firstRejection; + await expect(queued).resolves.toMatchObject({ + id: queuedId, + status: 'error', + canClose: true, + }); + + // A retried Stop reaches the still-running child again. + const killsBeforeRetry = (proc.kill as jest.Mock).mock.calls.length; + const retry = externalPlayerSessions.closeSession(queuedId); + await jest.advanceTimersByTimeAsync( + MPV_IPC_COMMAND_TIMEOUT_MS + 100 + ); + expect((proc.kill as jest.Mock).mock.calls.length).toBeGreaterThan( + killsBeforeRetry + ); + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + await expect(retry).resolves.toMatchObject({ + id: queuedId, + status: 'closed', + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + if (proc.exitCode === null) { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + } + }); + + it('lets a late Stop for an older queued launch cancel only that launch', async () => { + const { proc } = await launchReusableSession(); + const sockets = mockUnansweredSockets(); + const answer = (socket: MpvSocketMock, id: number) => + socket.emit( + 'data', + Buffer.from( + JSON.stringify({ request_id: id, error: 'success' }) + '\n' + ) + ); + + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const firstId = await writtenRequestId(sockets[0]); + const older = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const olderId = externalPlayerSessions.getActiveSessionId() as string; + const newest = openMpvPlayer({ + title: 'Channel C', + url: 'https://example.com/c.m3u8', + }); + + await expect( + externalPlayerSessions.closeSession(olderId) + ).resolves.toMatchObject({ id: olderId, status: 'closed' }); + expect(proc.kill).not.toHaveBeenCalled(); + + answer(sockets[0], firstId); + await expect(first).resolves.toMatchObject({ status: 'opened' }); + await expect(older).resolves.toMatchObject({ status: 'closed' }); + answer(sockets[1], await writtenRequestId(sockets[1])); + await expect(newest).resolves.toMatchObject({ status: 'opened' }); + expect(commandsOf(sockets)).toEqual(['loadfile', 'loadfile']); + expect(proc.kill).not.toHaveBeenCalled(); + }); + + it('settles a fresh launch and its queued Stop when the exit cannot be confirmed', async () => { + const proc = createMockChildProcess(); + (proc.kill as jest.Mock).mockImplementation(() => { + Object.defineProperty(proc, 'killed', { + value: true, + configurable: true, + }); + return true; + }); + spawnMock.mockReturnValueOnce(proc); + (store.get as unknown as jest.Mock).mockImplementation( + (key: string, fallback?: unknown) => + ({ + [MPV_PLAYER_PATH]: '/usr/bin/mpv', + [MPV_REUSE_INSTANCE]: true, + })[key] ?? fallback + ); + jest.useFakeTimers(); + + try { + // The first launch spawns fresh; the second queues during its + // start-up confirmation window. + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const queued = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const queuedId = + externalPlayerSessions.getActiveSessionId() as string; + // The Stop arrives over IPC, after the first launch spawned. + await jest.advanceTimersByTimeAsync(0); + let firstSettled = false; + void first.then( + () => (firstSettled = true), + () => (firstSettled = true) + ); + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + const closeRejection = expect( + externalPlayerSessions.closeSession(queuedId) + ).rejects.toThrow('External player process did not exit'); + + await jest.advanceTimersByTimeAsync(6_000); + expect(firstSettled).toBe(true); + await firstRejection; + await closeRejection; + await expect(queued).resolves.toMatchObject({ + id: queuedId, + status: 'error', + canClose: true, + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + }); + + it('never lets a retried queued Stop reach a player started later', async () => { + const { proc } = await launchReusableSession(); + (proc.kill as jest.Mock).mockImplementation(() => true); + const sockets = mockUnansweredSockets(); + jest.useFakeTimers(); + const later = createMockChildProcess(); + + try { + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const queued = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const queuedId = + externalPlayerSessions.getActiveSessionId() as string; + // The Stop arrives after the first launch wrote its loadfile. + while (!sockets[0]?.write.mock.calls.length) { + await jest.advanceTimersByTimeAsync(0); + } + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + const closeRejection = expect( + externalPlayerSessions.closeSession(queuedId) + ).rejects.toThrow('External player process did not exit'); + await jest.advanceTimersByTimeAsync(10_000); + await firstRejection; + await closeRejection; + await expect(queued).resolves.toMatchObject({ status: 'error' }); + + // The stuck child exits on its own and a new stream starts. + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + spawnMock.mockReturnValueOnce(later); + const next = openMpvPlayer({ + title: 'Channel C', + url: 'https://example.com/c.m3u8', + }); + await jest.advanceTimersByTimeAsync(100); + await expect(next).resolves.toMatchObject({ status: 'opened' }); + + // A retried Stop for the old queued launch stays on its own + // targets and leaves the new player alone. + const retry = externalPlayerSessions.closeSession(queuedId); + await jest.advanceTimersByTimeAsync( + MPV_IPC_COMMAND_TIMEOUT_MS + 100 + ); + await expect(retry).resolves.toMatchObject({ status: 'closed' }); + expect(later.kill).not.toHaveBeenCalled(); + } finally { + if (proc.exitCode === null) { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + } + }); + + it('restores the last started launch when a queued one is blocked by its teardown', async () => { + const proc = createMockChildProcess(); + (proc.kill as jest.Mock).mockImplementation(() => { + Object.defineProperty(proc, 'killed', { + value: true, + configurable: true, + }); + return true; + }); + spawnMock.mockReturnValueOnce(proc); + // Reuse off: the detached child is never stored for reuse. + (store.get as unknown as jest.Mock).mockImplementation( + (key: string, fallback?: unknown) => + ({ + [MPV_PLAYER_PATH]: '/usr/bin/mpv', + [MPV_REUSE_INSTANCE]: false, + })[key] ?? fallback + ); + jest.useFakeTimers(); + + try { + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const firstId = + externalPlayerSessions.getActiveSessionId() as string; + const skipped = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const newest = openMpvPlayer({ + title: 'Channel C', + url: 'https://example.com/c.m3u8', + }); + const newestId = + externalPlayerSessions.getActiveSessionId() as string; + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + const newestRejection = expect(newest).rejects.toThrow( + 'previous external player is still shutting down' + ); + + // A delayed Stop for the first launch cannot confirm its exit. + const closeRejection = expect( + externalPlayerSessions.closeSession(firstId) + ).rejects.toThrow('External player process did not exit'); + await jest.advanceTimersByTimeAsync(6_000); + await closeRejection; + await firstRejection; + await expect(skipped).resolves.toMatchObject({ status: 'closed' }); + await newestRejection; + + // The dock goes back to the session owning the stuck child. + expect(externalPlayerSessions.getActiveSessionId()).toBe(firstId); + expect(externalPlayerSessions.getSession(firstId)).toMatchObject({ + status: 'error', + canClose: true, + }); + expect(externalPlayerSessions.getSession(newestId)).toMatchObject({ + status: 'error', + canClose: false, + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + }); + + it('lets a queued Stop reach the child a failed launch is tearing down', async () => { + const { proc } = await launchReusableSession(); + (proc.kill as jest.Mock).mockImplementation(() => true); + // The first launch's loadfile fails; its teardown then waits for an + // exit that never comes. + mockMpvSockets({ + reply: (command) => + command[0] === 'loadfile' ? 'error running command' : null, + }); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + jest.useFakeTimers(); + + try { + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + const queued = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const queuedId = + externalPlayerSessions.getActiveSessionId() as string; + for ( + let tick = 0; + tick < 50 && !(proc.kill as jest.Mock).mock.calls.length; + tick += 1 + ) { + await jest.advanceTimersByTimeAsync(10); + } + expect(proc.kill).toHaveBeenCalled(); + + // Stop during that teardown must not report success while the + // child is still alive. + const closeRejection = expect( + externalPlayerSessions.closeSession(queuedId) + ).rejects.toThrow('External player process did not exit'); + await jest.advanceTimersByTimeAsync(10_000); + await closeRejection; + await firstRejection; + await expect(queued).resolves.toMatchObject({ + id: queuedId, + status: 'error', + canClose: true, + }); + + // A retry reaches the same child and settles once it exits; the + // retried quit gets no reply within its bound. + const retry = externalPlayerSessions.closeSession(queuedId); + await jest.advanceTimersByTimeAsync(0); + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + await jest.advanceTimersByTimeAsync( + MPV_IPC_COMMAND_TIMEOUT_MS + 100 + ); + await expect(retry).resolves.toMatchObject({ + id: queuedId, + status: 'closed', + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + if (proc.exitCode === null) { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + } + }); + + it('keeps a queued Stop working after the previous owner closed during a failed handback', async () => { + const { proc, session: previous } = await launchReusableSession(); + (proc.kill as jest.Mock).mockImplementation(() => true); + const sockets = mockUnansweredSockets(); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + jest.useFakeTimers(); + + try { + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + const loadfileId = await (async () => { + while (!sockets[0]?.write.mock.calls.length) { + await jest.advanceTimersByTimeAsync(0); + } + return ( + JSON.parse(String(sockets[0].write.mock.calls[0][0])) as { + request_id: number; + } + ).request_id; + })(); + // A late Stop for the previous session closes it; the child now + // belongs to the first launch. + await expect( + externalPlayerSessions.closeSession(previous.id) + ).resolves.toMatchObject({ status: 'closed' }); + const queued = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const queuedId = + externalPlayerSessions.getActiveSessionId() as string; + + // mpv rejects the load: the child is handed back to the closed + // previous session while the first launch tears it down. + sockets[0].emit( + 'data', + Buffer.from( + JSON.stringify({ + request_id: loadfileId, + error: 'error running command', + }) + '\n' + ) + ); + await jest.advanceTimersByTimeAsync(10); + + const closeRejection = expect( + externalPlayerSessions.closeSession(queuedId) + ).rejects.toThrow('External player process did not exit'); + await jest.advanceTimersByTimeAsync(10_000); + await closeRejection; + await firstRejection; + await expect(queued).resolves.toMatchObject({ + id: queuedId, + status: 'error', + canClose: true, + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + if (proc.exitCode === null) { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + } + }); + + it('keeps a queued Stop working when reuse is disabled before the in-flight load', async () => { + const { proc } = await launchReusableSession(); + (proc.kill as jest.Mock).mockImplementation(() => true); + const sockets = mockUnansweredSockets(); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + jest.useFakeTimers(); + + try { + // The first launch still waits on its user-agent reply, so the + // child still belongs to the session before it. + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + userAgent: 'IPTVnator test agent', + }); + const firstRejection = expect(first).rejects.toThrow( + 'External player process did not exit' + ); + while (!sockets[0]?.write.mock.calls.length) { + await jest.advanceTimersByTimeAsync(0); + } + const queued = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const queuedId = + externalPlayerSessions.getActiveSessionId() as string; + + // Disabling reuse clears the stored child before it has exited. + setMpvReuseInstance(false); + const closeRejection = expect( + externalPlayerSessions.closeSession(queuedId) + ).rejects.toThrow('External player process did not exit'); + await jest.advanceTimersByTimeAsync(10_000); + await closeRejection; + await firstRejection; + await expect(queued).resolves.toMatchObject({ + id: queuedId, + status: 'error', + canClose: true, + }); + + // The first launch has settled; a retried Stop must still reach + // the live child and settle only once it exits. + const killsBeforeRetry = (proc.kill as jest.Mock).mock.calls.length; + const retry = externalPlayerSessions.closeSession(queuedId); + let retried = false; + void retry.then( + () => (retried = true), + () => (retried = true) + ); + await jest.advanceTimersByTimeAsync( + MPV_IPC_COMMAND_TIMEOUT_MS + 100 + ); + expect(retried).toBe(false); + expect((proc.kill as jest.Mock).mock.calls.length).toBeGreaterThan( + killsBeforeRetry + ); + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + await expect(retry).resolves.toMatchObject({ + id: queuedId, + status: 'closed', + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + if (proc.exitCode === null) { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + } + }); + + it('starts a queued launch on the fresh player a failed launch spawned', async () => { + const { proc } = await launchReusableSession(); + const freshProc = createMockChildProcess(); + spawnMock.mockReturnValueOnce(freshProc); + // The first launch's loadfile fails; every later command succeeds. + // Each loadfile records how many players had been spawned by then. + let rejectedLoadfile = false; + const spawnsAtLoadfile: number[] = []; + mockMpvSockets({ + reply: (command) => { + if (command[0] === 'loadfile') { + spawnsAtLoadfile.push(spawnMock.mock.calls.length); + } + if (command[0] === 'loadfile' && !rejectedLoadfile) { + rejectedLoadfile = true; + return 'error running command'; + } + return 'success'; + }, + }); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + + try { + const first = openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + const second = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + + await expect(first).resolves.toMatchObject({ status: 'opened' }); + const secondSession = await second; + // The second launch loaded into the replacement the first one's + // fallback spawned, not into the child that fallback tore down. + expect(spawnsAtLoadfile).toEqual([1, 2]); + expect( + externalPlayerSessions.getSession(secondSession.id)?.status + ).toBe('opened'); + expect(proc.kill).toHaveBeenCalledTimes(1); + expect(spawnMock).toHaveBeenCalledTimes(2); + expect(freshProc.kill).not.toHaveBeenCalled(); + } finally { + errorSpy.mockRestore(); + } + }); +}); diff --git a/apps/electron-backend/src/app/events/mpv-launch-queue.ts b/apps/electron-backend/src/app/events/mpv-launch-queue.ts new file mode 100644 index 000000000..898c2d8ab --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-launch-queue.ts @@ -0,0 +1,182 @@ +import { + ExternalPlayerSession, + readExternalPlayerErrorCode, +} from '@iptvnator/shared/interfaces'; +import { externalPlayerProcessTeardownGate } from './external-player-process'; +import { externalPlayerSessions } from './external-player-runtime'; + +/** What a launch knows about its place in the queue when it starts. */ +export interface QueuedMpvLaunch { + /** Whether this launch waited behind another one. */ + queued: boolean; + /** + * Whether Stop or shutdown cancelled this launch while it waited, or a + * newer launch is queued behind it. + */ + cancelled: () => boolean; + /** This launch's own queued Stop, if one was requested. */ + pendingStop: () => Promise | null; + /** Called once the launch really starts acting on a player. */ + markInFlight: () => void; + /** The last launch that started before this one runs. */ + lastStartedSessionId: () => string | null; +} + +/** + * Runs MPV launches one at a time. A reused launch waits for mpv's replies + * (bounded per command) and a failed one for its teardown and fallback + * spawn, so overlapping launches would act on the same child. Queued, each + * launch starts from the player the previous one left behind, and only the + * newest queued launch runs. An idle queue starts a launch synchronously. + * + * Until a queued launch starts, its Stop cancels it and every launch queued + * before it. Stop on the newest one also closes, through their own closers, + * the in-flight launch and the session owning the stored player, so the + * exact-child, retry and settle rules of those closers apply. + */ +export class MpvLaunchQueue { + private pending: Promise | null = null; + private latest = 0; + private cancelledThrough = 0; + private inFlightSessionId: string | null = null; + // The last launch that actually started. With reuse off its detached + // child is not stored, so it is the only handle on that player once the + // launch settled with an unconfirmed teardown. + private lastStartedSessionId: string | null = null; + + constructor(private readonly storedOwnerSessionId: () => string | null) {} + + /** Cancels every launch queued so far, for app shutdown. */ + cancelAll(): void { + this.cancelledThrough = this.latest; + } + + enqueue( + session: ExternalPlayerSession, + launch: (queue: QueuedMpvLaunch) => Promise + ): Promise { + const previous = this.pending; + const number = ++this.latest; + let stop: Promise | null = null; + // Fixed at the first Stop, so a retry never reaches a newer player. + let stopTargets: string[] | null = null; + if (previous) { + externalPlayerSessions.attachCloser(session.id, () => { + this.cancelledThrough = Math.max(this.cancelledThrough, number); + // A late Stop for an older queued launch only cancels it. + stopTargets ??= + number === this.latest ? this.stopTargets(session.id) : []; + stop = Promise.all( + stopTargets.map((id) => + externalPlayerSessions.closeSession(id) + ) + ).then(() => undefined); + return stop; + }); + } + const queue: QueuedMpvLaunch = { + queued: previous !== null, + cancelled: () => + number <= this.cancelledThrough || number !== this.latest, + pendingStop: () => stop, + markInFlight: () => { + this.inFlightSessionId = session.id; + this.lastStartedSessionId = session.id; + }, + lastStartedSessionId: () => this.lastStartedSessionId, + }; + const run = () => launch(queue); + // The queue is released before the caller sees the result, so a + // launch made right after this one settles starts synchronously. + let settled: Promise | null = null; + const result = (previous ? previous.then(run) : run()).finally(() => { + if (this.inFlightSessionId === session.id) { + this.inFlightSessionId = null; + } + if (this.pending === settled) this.pending = null; + }); + settled = result.catch(() => undefined); + this.pending = settled; + return result; + } + + private stopTargets(sessionId: string): string[] { + const targets: string[] = []; + for (const id of [ + this.inFlightSessionId, + this.storedOwnerSessionId(), + this.lastStartedSessionId, + ]) { + // Only a session with its own closer can confirm its teardown; + // closing one without would just mark it closed. + if ( + id && + id !== sessionId && + !targets.includes(id) && + externalPlayerSessions.getSession(id)?.canClose + ) { + targets.push(id); + } + } + return targets; + } +} + +/** + * Settles a queued launch that will not start. When its own Stop is still + * closing the in-flight player, the outcome waits for it: a Stop that + * cannot confirm the exit leaves a closable error, so Stop can be retried. + */ +export async function settleCancelledQueuedLaunch( + session: ExternalPlayerSession, + pendingStop: Promise | null +): Promise { + if (pendingStop) { + try { + await pendingStop; + } catch (error) { + return ( + externalPlayerSessions.markError( + session.id, + error instanceof Error ? error.message : String(error), + { canClose: true } + ) ?? session + ); + } + } + return externalPlayerSessions.markClosed(session.id) ?? session; +} + +/** + * A queued launch starts after an earlier one may have left a child whose + * exit is unconfirmed. Then the dock goes back to the session that still + * owns that child; failing that, this session keeps its queued closer so + * Stop can still reach the child. + */ +export function assertQueuedLaunchAllowed( + session: ExternalPlayerSession, + restoreCandidates: Array +): void { + try { + externalPlayerProcessTeardownGate.assertLaunchAllowed(); + } catch (error) { + let restored: ExternalPlayerSession | null = null; + for (const id of restoreCandidates) { + if (!id || id === session.id) continue; + restored = externalPlayerSessions.restoreActiveSession( + id, + session.id + ); + if (restored) break; + } + externalPlayerSessions.markError( + session.id, + error instanceof Error ? error.message : String(error), + { + code: readExternalPlayerErrorCode(error) ?? undefined, + canClose: !restored, + } + ); + throw error; + } +} diff --git a/apps/electron-backend/src/app/events/mpv-output-errors.ts b/apps/electron-backend/src/app/events/mpv-output-errors.ts new file mode 100644 index 000000000..7d2d37b9a --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-output-errors.ts @@ -0,0 +1,69 @@ +import type { ChildProcess } from 'child_process'; +import { + sendPlayerErrorNotification, + traceExternalPlayer, +} from './external-player-runtime'; + +function shouldIgnoreMpvStdoutLine(line: string): boolean { + return /^(\(Paused\)\s*)?AV:\s/.test(line); +} + +/** + * Reports what a reusable MPV child prints: stream errors on stdout or stderr + * reach the renderer as player errors; the periodic status line is ignored. + */ +export function attachMpvOutputHandlers(proc: ChildProcess): void { + if (proc.stdout) { + proc.stdout.on('data', (data) => { + const lines = data + .toString() + .split('\n') + .map((line) => line.trim()) + .filter(Boolean); + + for (const output of lines) { + if (shouldIgnoreMpvStdoutLine(output)) { + continue; + } + + traceExternalPlayer('mpv stdout', { output }); + + if ( + output.includes('Failed to open') || + output.includes('Error opening') || + output.includes('Protocol not found') || + output.includes('Connection refused') || + output.includes('error') || + output.includes('403') || + output.includes('404') + ) { + console.error('[MPV ERROR from stdout]:', output); + sendPlayerErrorNotification('MPV', output); + } + } + }); + } + + if (proc.stderr) { + proc.stderr.on('data', (data) => { + const output = data.toString().trim(); + if (output) { + console.error('[MPV stderr]:', output); + + if ( + output.includes('Failed to open') || + output.includes('Exiting... (Errors when loading file)') || + output.includes('Error opening') || + output.includes('Protocol not found') || + output.includes('Connection refused') || + output.includes('error') || + output.includes('403') || + output.includes('404') + ) { + console.error('[MPV ERROR]:', output); + sendPlayerErrorNotification('MPV', output); + } + } + }); + } +} diff --git a/apps/electron-backend/src/app/events/mpv-position-polling.ts b/apps/electron-backend/src/app/events/mpv-position-polling.ts new file mode 100644 index 000000000..d2f049e93 --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-position-polling.ts @@ -0,0 +1,121 @@ +import { createConnection } from 'net'; +import { PlayerContentInfo } from '@iptvnator/shared/interfaces'; +import { sendPlaybackPositionUpdate } from './external-player-runtime'; + +/** + * Playback-position polling of an external MPV over its IPC socket. One poll + * runs at a time across launches: starting a new one stops the previous one. + */ +let positionPollingInterval: NodeJS.Timeout | null = null; +// The first poll waits for MPV to open its IPC socket. The handle is kept so +// a player that exits during that wait does not start an orphaned interval. +let positionPollingDelay: NodeJS.Timeout | null = null; +// Bumped on every stop. A poll that was already awaiting an IPC reply when +// polling stopped (for example when a reused player loads the next stream) +// must not report what it reads under the previous content. +let positionPollingGeneration = 0; + +async function getMpvProperty( + socketPath: string, + property: string +): Promise { + return new Promise((resolve) => { + const client = createConnection(socketPath); + const request = + JSON.stringify({ + command: ['get_property', property], + }) + '\n'; + + let data = ''; + + client.on('connect', () => { + client.write(request); + }); + + const timeoutHandle = setTimeout(() => { + client.destroy(); + resolve(null); + }, 2000); + + client.on('data', (chunk) => { + data += chunk.toString(); + if (data.includes('\n')) { + clearTimeout(timeoutHandle); + try { + const lines = data.split('\n'); + for (const line of lines) { + if (!line.trim()) continue; + const response = JSON.parse(line); + if (response.data !== undefined) { + client.destroy(); + resolve(response.data); + return; + } + } + } catch { + // Ignore partial JSON while MPV is still writing. + } + } + }); + + client.on('end', () => { + clearTimeout(timeoutHandle); + if (data && !data.includes('\n')) { + try { + const response = JSON.parse(data); + resolve(response.data ?? null); + } catch { + resolve(null); + } + } + }); + + client.on('error', () => { + clearTimeout(timeoutHandle); + resolve(null); + }); + }); +} + +export function stopPositionPolling(): void { + positionPollingGeneration += 1; + if (positionPollingDelay) { + clearTimeout(positionPollingDelay); + positionPollingDelay = null; + } + if (positionPollingInterval) { + clearInterval(positionPollingInterval); + positionPollingInterval = null; + } +} + +export function startPositionPolling( + socketPath: string, + contentInfo: PlayerContentInfo, + sessionId: string +): void { + stopPositionPolling(); + const generation = positionPollingGeneration; + const isCurrent = () => generation === positionPollingGeneration; + + positionPollingDelay = setTimeout(() => { + positionPollingDelay = null; + positionPollingInterval = setInterval(async () => { + try { + const position = await getMpvProperty(socketPath, 'time-pos'); + if (!isCurrent()) return; + const duration = await getMpvProperty(socketPath, 'duration'); + if (!isCurrent()) return; + + if (position !== null) { + sendPlaybackPositionUpdate(sessionId, contentInfo, { + positionSeconds: Math.floor(position), + durationSeconds: duration ? Math.floor(duration) : null, + }); + } + } catch { + stopPositionPolling(); + } + }, 5000); + }, 2000); +} diff --git a/apps/electron-backend/src/app/events/mpv-reusable-process.ts b/apps/electron-backend/src/app/events/mpv-reusable-process.ts index db5234cdb..1c397fb21 100644 --- a/apps/electron-backend/src/app/events/mpv-reusable-process.ts +++ b/apps/electron-backend/src/app/events/mpv-reusable-process.ts @@ -1,20 +1,22 @@ import { ChildProcess } from 'child_process'; -import { createConnection } from 'net'; import { ExternalPlayerSession, PlayerContentInfo, } from '@iptvnator/shared/interfaces'; +import { redactSensitiveData } from '@iptvnator/shared/logging'; import { joinMpvHeaderFields } from '../util/mpv-string-list.util'; import { externalPlayerProcessTeardownGate } from './external-player-process'; import { externalPlayerSessions, traceExternalPlayer, } from './external-player-runtime'; +import { sendMpvCommand } from './mpv-ipc-command'; /** * The per-file start of a reuse load: the requested offset, else zero. A - * `seek` right after `loadfile` runs before the file is loaded and fails, - * so the offset travels with the load itself. + * `seek` right after `loadfile` runs before the file is loaded and mpv + * rejects it, so the offset travels with the load itself and no seek is + * sent on the reuse path. */ function reuseStartSeconds(startTime: number | undefined): number { return typeof startTime === 'number' && @@ -24,7 +26,9 @@ function reuseStartSeconds(startTime: number | undefined): number { : 0; } -const MPV_IPC_COMMAND_TIMEOUT_MS = 2_000; +function hasExited(child: ChildProcess): boolean { + return child.exitCode !== null || child.signalCode !== null; +} export interface MpvReuseAttemptState { contentMutated: boolean; @@ -50,68 +54,33 @@ interface MpvReuseOptions { stopPositionPolling: () => void; } -function sendMpvCommand( - socketPath: string, - command: string, - args: Array, - shouldDispatch?: () => boolean -): Promise { - return new Promise((resolve, reject) => { - const client = createConnection(socketPath); - const request = JSON.stringify({ command: [command, ...args] }) + '\n'; - let settled = false; - let timeoutHandle: NodeJS.Timeout | null = null; - const complete = (error?: Error, dispatched = true) => { - if (settled) return; - settled = true; - if (timeoutHandle) clearTimeout(timeoutHandle); - if (!dispatched && !client.destroyed) client.destroy(); - if (error) { - reject(error); - } else { - resolve(dispatched); - } - }; - - client.on('connect', () => { - if (shouldDispatch && !shouldDispatch()) { - complete(undefined, false); - return; - } - traceExternalPlayer('mpv ipc command', { - command, - argsCount: args.length, - }); - try { - client.write(request); - client.end(); - complete(); - } catch (error) { - complete( - error instanceof Error ? error : new Error(String(error)) - ); - } - }); - client.on('error', (error) => complete(error)); - timeoutHandle = setTimeout(() => { - complete(new Error('MPV IPC command timed out')); - client.destroy(); - }, MPV_IPC_COMMAND_TIMEOUT_MS); - timeoutHandle.unref(); - }); -} - /** Owns the one MPV child/socket retained when instance reuse is enabled. */ export class MpvReusableProcess { private process: ChildProcess | null = null; private socketPath: string | null = null; private processSessionId: string | null = null; private readonly processSessionIds = new WeakMap(); + // Bumped whenever the stored player is stopped (reuse disabled or app + // quitting). Launches run one at a time, so the only thing that can + // overtake an in-flight reuse attempt is that stop, which cancels it. + private stopGeneration = 0; currentSessionId(): string | null { return this.processSessionId; } + /** + * The session that owns the stored child right now. A failed reuse + * attempt hands the child back to its previous owner while it tears it + * down, so this reads the per-child mapping, not the last recorded id. + */ + storedOwnerSessionId(): string | null { + if (!this.process) return null; + return ( + this.processSessionIds.get(this.process) ?? this.processSessionId + ); + } + sessionIdFor(process: ChildProcess, fallback: string): string { return this.processSessionIds.get(process) ?? fallback; } @@ -140,6 +109,10 @@ export class MpvReusableProcess { stopPositionPolling: () => void, guardFutureLaunches = false ): void { + // A launch still waiting for mpv's reply must not fall back to a + // fresh, again-reusable player once reuse is disabled or the app + // quits: cancel it so it settles its own session instead. + this.stopGeneration += 1; const process = this.process; if (!process || process.killed) return; traceExternalPlayer(reason); @@ -163,6 +136,29 @@ export class MpvReusableProcess { traceExternalPlayer('reuse existing mpv instance'); const { session, state } = options; + const generation = this.stopGeneration; + const cancelled = () => generation !== this.stopGeneration; + // The stored-player stop is already tearing the child down; settle + // only once its exit is confirmed, else stay a closable error so Stop + // can retry while the teardown gate blocks further launches. + const settleCancelled = async () => { + traceExternalPlayer('mpv reuse cancelled by a stored-player stop'); + try { + await externalPlayerProcessTeardownGate.terminate( + reusedProcess + ); + } catch (error) { + state.teardownUnconfirmed = true; + return ( + externalPlayerSessions.markError( + session.id, + error instanceof Error ? error.message : String(error), + { canClose: true } + ) ?? session + ); + } + return externalPlayerSessions.markClosed(session.id) ?? session; + }; const reusedProcessSessionId = this.processSessionIds.get(reusedProcess) ?? options.previousProcessSessionId; @@ -170,7 +166,18 @@ export class MpvReusableProcess { let retryableClose: Promise | null = null; let launchClose: Promise | null = null; + // Launches run one at a time, so nothing else acts on this child while + // this attempt is in flight: its Stop always reaches the child then, + // whoever the per-child mapping names (the previous owner before + // loadfile or during a failed attempt's teardown, possibly closed + // already). Once the attempt settles, ownership decides as before. + let attemptInFlight = true; + // A Stop of this session that could not confirm the exit keeps the + // right to retry that exact child after the attempt settles. + let ownsFailedTeardown = false; const ownsReusedProcess = () => + attemptInFlight || + ownsFailedTeardown || this.processSessionIds.get(reusedProcess) === session.id || externalPlayerSessions.getActiveSessionId() === session.id; const closeReusedProcess = async () => { @@ -208,6 +215,7 @@ export class MpvReusableProcess { launchClose ??= closeAttempt; void closeAttempt.catch((error) => { if (retryableClose === closeAttempt) retryableClose = null; + ownsFailedTeardown = true; state.teardownUnconfirmed = true; externalPlayerSessions.markError( session.id, @@ -219,28 +227,34 @@ export class MpvReusableProcess { return retryableClose; }); - try { - await this.applyReuseCommands( - options, - reusedSocketPath, - () => !closeRequested - ); - if (closeRequested) return await finishRequestedClose(); - + // Once `loadfile` reaches the socket the player content may have + // changed, whatever mpv replies, so the attempted session owns the + // child from the write onwards: an exit or a stale Stop for the + // previous session during the reply wait then lands on the right + // session instead of terminating the new playback, and the previous + // session's position poll can no longer read the new stream's + // position under the old content. + const onContentDispatched = () => { state.contentMutated = true; this.processSessionId = session.id; this.processSessionIds.set(reusedProcess, session.id); options.stopPositionPolling(); + }; - if (options.startTime) { - await sendMpvCommand( - reusedSocketPath, - 'seek', - [String(options.startTime), 'absolute'], - () => !closeRequested - ); - } + try { + await this.applyReuseCommands( + options, + reusedSocketPath, + () => !closeRequested && !cancelled(), + onContentDispatched + ); if (closeRequested) return await finishRequestedClose(); + if (cancelled()) return await settleCancelled(); + if (hasExited(reusedProcess)) { + // The child exited while its reply was in flight; its exit + // handler has already settled the session that owned it. + return externalPlayerSessions.getSession(session.id) ?? session; + } if (options.contentInfo) { options.startPositionPolling( @@ -256,7 +270,17 @@ export class MpvReusableProcess { const current = externalPlayerSessions.getSession(session.id); if (current?.status === 'closed') return current; if (closeRequested) return await finishRequestedClose(); - console.error('Failed to send command to existing MPV:', error); + if (cancelled()) return await settleCancelled(); + if (current?.status === 'error' && hasExited(reusedProcess)) { + // The child died while its reply was in flight and its exit + // handler already reported the failure; a fresh launch under + // this terminal session could never be shown as opened. + return current; + } + console.error( + 'Failed to send command to existing MPV:', + redactSensitiveData(error) + ); if (state.contentMutated) { if (reusedProcessSessionId) { @@ -282,21 +306,27 @@ export class MpvReusableProcess { this.clear(reusedProcess); options.stopPositionPolling(); if (closeRequested) return await finishRequestedClose(); + // Reuse may have been disabled, or the app may be quitting, + // while the teardown above waited for the child to exit. + if (cancelled()) return await settleCancelled(); return null; + } finally { + attemptInFlight = false; } } private async applyReuseCommands( options: MpvReuseOptions, socketPath: string, - shouldDispatch: () => boolean + shouldDispatch: () => boolean, + onContentDispatched: () => void ): Promise { if (options.effectiveUserAgent) { const dispatched = await sendMpvCommand( socketPath, 'set_property', ['user-agent', options.effectiveUserAgent], - shouldDispatch + { shouldDispatch } ); if (!dispatched) return; } @@ -305,7 +335,7 @@ export class MpvReusableProcess { socketPath, 'set_property', ['referrer', options.effectiveReferer], - shouldDispatch + { shouldDispatch } ); if (!dispatched) return; } @@ -317,7 +347,7 @@ export class MpvReusableProcess { 'http-header-fields', joinMpvHeaderFields(options.headerFields), ], - shouldDispatch + { shouldDispatch } ); if (!dispatched) return; } @@ -335,13 +365,13 @@ export class MpvReusableProcess { -1, fileOptions.join(','), ]; - const dispatched = await sendMpvCommand( + const reply = await sendMpvCommand( socketPath, 'loadfile', loadFileArgs, - shouldDispatch + { shouldDispatch, onDispatched: onContentDispatched } ); - if (!dispatched) return; + if (!reply) return; traceExternalPlayer('loaded new url in existing mpv instance'); } } diff --git a/apps/electron-backend/src/app/events/mpv-session.service.ipc-reply.spec.ts b/apps/electron-backend/src/app/events/mpv-session.service.ipc-reply.spec.ts new file mode 100644 index 000000000..6f54c4dce --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-session.service.ipc-reply.spec.ts @@ -0,0 +1,663 @@ +/** + * Reply handling of the reused-instance MPV IPC path. Every command carries a + * `request_id` and the launch waits, bounded, for mpv's matching reply; an + * error reply or a missing one falls back to a fresh launch exactly like a + * socket failure. Teardown, ownership and Stop-interleaving coverage lives in + * `mpv-session.service.spec.ts`. + */ +jest.mock('electron', () => ({ + ipcMain: { + handle: jest.fn(), + }, +})); + +jest.mock('child_process', () => ({ + spawn: jest.fn(), +})); + +jest.mock('net', () => ({ + createConnection: jest.fn(), + createServer: jest.fn(), +})); + +jest.mock('../app', () => ({ + __esModule: true, + default: { + mainWindow: null, + }, +})); + +jest.mock('../services/store.service', () => ({ + MPV_PLAYER_ARGUMENTS: 'MPV_PLAYER_ARGUMENTS', + MPV_PLAYER_PATH: 'MPV_PLAYER_PATH', + MPV_REUSE_INSTANCE: 'MPV_REUSE_INSTANCE', + store: { + get: jest.fn(), + set: jest.fn(), + }, +})); + +import { + MPV_PLAYER_PATH, + MPV_REUSE_INSTANCE, + store, +} from '../services/store.service'; +import * as externalPlayerRuntime from './external-player-runtime'; +import { externalPlayerSessions } from './external-player-runtime'; +import { MPV_IPC_COMMAND_TIMEOUT_MS } from './mpv-ipc-command'; +import { + createMpvSocketMock, + MpvSocketMock, +} from './mpv-ipc-socket.test-helpers'; +import { + openMpvPlayer, + setMpvReuseInstance, + shutdownMpvSession, +} from './mpv-session.service'; +import { + spawnMock, + createConnectionMock, + createMockChildProcess, + mockMpvSockets, + nextTick, + launchReusableInstance, + launchReusableSession, + mockUnansweredSockets, + writtenRequestId, +} from './mpv-session.test-helpers'; + +describe('reused MPV IPC replies', () => { + beforeEach(() => { + jest.clearAllMocks(); + shutdownMpvSession(); + }); + + afterEach(() => { + jest.useRealTimers(); + shutdownMpvSession(); + }); + it('opens the session once mpv acknowledges the content command', async () => { + const reusedProc = await launchReusableInstance(); + const written: string[] = []; + mockMpvSockets({ written }); + + const session = await openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + startTime: 30, + }); + + expect(session.status).toBe('opened'); + expect(spawnMock).toHaveBeenCalledTimes(1); + expect(reusedProc.kill).not.toHaveBeenCalled(); + const requests = written.map( + (chunk) => + JSON.parse(chunk.trim()) as { + command: Array; + request_id: number; + } + ); + // The offset travels with loadfile; no seek races the load. + expect(requests.map((request) => request.command[0])).toEqual([ + 'loadfile', + ]); + expect(Number.isInteger(requests[0].request_id)).toBe(true); + }); + + it('falls back to a fresh launch when mpv rejects the content command', async () => { + const reusedProc = await launchReusableInstance(); + const freshProc = createMockChildProcess(); + spawnMock.mockReturnValueOnce(freshProc); + mockMpvSockets({ + reply: (command) => + command[0] === 'loadfile' ? 'error running command' : 'success', + }); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + + try { + const session = await openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8?token=secret-token', + }); + + expect(reusedProc.kill).toHaveBeenCalledTimes(1); + expect(spawnMock).toHaveBeenCalledTimes(2); + expect(session.status).toBe('opened'); + expect(externalPlayerSessions.getSession(session.id)?.status).toBe( + 'opened' + ); + const logged = JSON.stringify(errorSpy.mock.calls); + expect(logged).toContain('mpv loadfile: error running command'); + expect(logged).not.toContain('secret-token'); + } finally { + errorSpy.mockRestore(); + } + }); + + it('falls back to a fresh launch when mpv never replies', async () => { + const reusedProc = await launchReusableInstance(); + const freshProc = createMockChildProcess(); + spawnMock.mockReturnValueOnce(freshProc); + mockMpvSockets({ reply: () => null }); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + jest.useFakeTimers(); + + try { + const opening = openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + }); + await jest.advanceTimersByTimeAsync(MPV_IPC_COMMAND_TIMEOUT_MS - 1); + expect(reusedProc.kill).not.toHaveBeenCalled(); + expect(spawnMock).toHaveBeenCalledTimes(1); + + await jest.advanceTimersByTimeAsync(1); + expect(reusedProc.kill).toHaveBeenCalledTimes(1); + // Exit of the reused child, the fresh spawn and its start + // confirmation all run on timers. + await jest.advanceTimersByTimeAsync(500); + + await expect(opening).resolves.toMatchObject({ + status: 'opened', + }); + expect(spawnMock).toHaveBeenCalledTimes(2); + expect(JSON.stringify(errorSpy.mock.calls)).toContain( + 'mpv loadfile: reply timed out' + ); + } finally { + errorSpy.mockRestore(); + } + }); + + it('ignores events and other replies while waiting for its own', async () => { + await launchReusableInstance(); + const sockets: MpvSocketMock[] = []; + createConnectionMock.mockImplementation(() => { + const socket = createMpvSocketMock({ reply: () => null }); + sockets.push(socket); + return socket; + }); + + const opening = openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + }); + let settled = false; + void opening.then( + () => (settled = true), + () => (settled = true) + ); + while (!sockets[0]?.write.mock.calls.length) { + await nextTick(); + } + const { request_id: requestId } = JSON.parse( + sockets[0].write.mock.calls[0][0] as string + ) as { request_id: number }; + const push = (...lines: object[]) => + sockets[0].emit( + 'data', + Buffer.from( + lines.map((line) => JSON.stringify(line)).join('\n') + '\n' + ) + ); + + // mpv broadcasts events on every connection and may still be + // answering an earlier request; none of these is this reply. + push( + { event: 'start-file', playlist_entry_id: 1 }, + { + event: 'end-file', + reason: 'error', + file_error: 'loading failed', + }, + { request_id: requestId + 1000, error: 'error running command' } + ); + await nextTick(); + await nextTick(); + expect(settled).toBe(false); + expect(spawnMock).toHaveBeenCalledTimes(1); + + push({ request_id: requestId, error: 'success' }); + await expect(opening).resolves.toMatchObject({ status: 'opened' }); + expect(spawnMock).toHaveBeenCalledTimes(1); + expect(sockets).toHaveLength(1); + }); + + it('lets a stale Stop for the previous session pass while the loadfile reply is pending', async () => { + const { proc, session: previous } = await launchReusableSession(); + const sockets = mockUnansweredSockets(); + + const opening = openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + }); + const requestId = await writtenRequestId(sockets[0]); + + // The child already belongs to the new session: the old closer + // must neither kill it nor send quit. + await expect( + externalPlayerSessions.closeSession(previous.id) + ).resolves.toMatchObject({ id: previous.id, status: 'closed' }); + expect(proc.kill).not.toHaveBeenCalled(); + expect(sockets).toHaveLength(1); + + sockets[0].emit( + 'data', + Buffer.from( + JSON.stringify({ request_id: requestId, error: 'success' }) + + '\n' + ) + ); + await expect(opening).resolves.toMatchObject({ status: 'opened' }); + expect(proc.kill).not.toHaveBeenCalled(); + expect(spawnMock).toHaveBeenCalledTimes(1); + }); + + it('settles the new session as closed when the child exits during the reply wait', async () => { + const { proc, session: previous } = await launchReusableSession(); + const sockets = mockUnansweredSockets(); + + const opening = openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + contentInfo: { + playlistId: 'playlist-1', + contentXtreamId: 7, + contentType: 'vod', + }, + }); + const requestId = await writtenRequestId(sockets[0]); + const replacementId = + externalPlayerSessions.getActiveSessionId() as string; + + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + sockets[0].emit( + 'data', + Buffer.from( + JSON.stringify({ request_id: requestId, error: 'success' }) + + '\n' + ) + ); + + // The exit landed on the session that owned the child at that + // moment, and a dead player is never reported as opened. + await expect(opening).resolves.toMatchObject({ + id: replacementId, + status: 'closed', + }); + expect(externalPlayerSessions.getSession(previous.id)?.status).toBe( + 'opened' + ); + expect(spawnMock).toHaveBeenCalledTimes(1); + }); + + it('stops the previous position poll as soon as loadfile is written', async () => { + jest.useFakeTimers(); + const contentInfo = { + playlistId: 'playlist-1', + contentXtreamId: 7, + contentType: 'vod' as const, + }; + const proc = createMockChildProcess(); + spawnMock.mockReturnValueOnce(proc); + (store.get as unknown as jest.Mock).mockImplementation( + (key: string, fallback?: unknown) => + ({ + [MPV_PLAYER_PATH]: '/usr/bin/mpv', + [MPV_REUSE_INSTANCE]: true, + })[key] ?? fallback + ); + const sockets = mockUnansweredSockets(); + const first = openMpvPlayer({ + title: 'First stream', + url: 'https://example.com/one.m3u8', + contentInfo, + }); + await jest.advanceTimersByTimeAsync(100); + await first; + // The first session polls 2 s after launch, then every 5 s: the + // next poll is due at 7 s, inside the reply window of a loadfile + // written at 6.5 s. + await jest.advanceTimersByTimeAsync(6_500); + const pollsBefore = sockets.length; + + const opening = openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + contentInfo: { ...contentInfo, contentXtreamId: 8 }, + }); + await jest.advanceTimersByTimeAsync(0); + const loadfileSocket = sockets[pollsBefore]; + const requestId = await writtenRequestId(loadfileSocket); + await jest.advanceTimersByTimeAsync(1_000); + + const commandsAfterLoadfile = sockets + .slice(pollsBefore + 1) + .flatMap((socket) => + socket.write.mock.calls.map( + ([request]) => + (JSON.parse(String(request)) as { command: string[] }) + .command[0] + ) + ); + expect(commandsAfterLoadfile).not.toContain('get_property'); + + loadfileSocket.emit( + 'data', + Buffer.from( + JSON.stringify({ request_id: requestId, error: 'success' }) + + '\n' + ) + ); + await jest.advanceTimersByTimeAsync(0); + await expect(opening).resolves.toMatchObject({ status: 'opened' }); + }); + + it('drops a position read that was in flight when loadfile was written', async () => { + // The previous test's player exits on a real tick, and any MPV exit + // stops polling; let it land before this test starts its poll. + await nextTick(); + await nextTick(); + jest.useFakeTimers(); + const sendPosition = jest.spyOn( + externalPlayerRuntime, + 'sendPlaybackPositionUpdate' + ); + const proc = createMockChildProcess(); + spawnMock.mockReturnValueOnce(proc); + (store.get as unknown as jest.Mock).mockImplementation( + (key: string, fallback?: unknown) => + ({ + [MPV_PLAYER_PATH]: '/usr/bin/mpv', + [MPV_REUSE_INSTANCE]: true, + })[key] ?? fallback + ); + const sockets = mockUnansweredSockets(); + const commandOf = (socket: MpvSocketMock) => + socket.write.mock.calls.length + ? ( + JSON.parse(String(socket.write.mock.calls[0][0])) as { + command: string[]; + } + ).command + : []; + const first = openMpvPlayer({ + title: 'First stream', + url: 'https://example.com/one.m3u8', + contentInfo: { + playlistId: 'playlist-1', + contentXtreamId: 7, + contentType: 'vod', + }, + }); + await jest.advanceTimersByTimeAsync(100); + const previous = await first; + + // The first poll fires 2 s + 5 s after launch and waits for time-pos. + await jest.advanceTimersByTimeAsync(6_950); + const timePosSocket = sockets.find( + (socket) => commandOf(socket)[1] === 'time-pos' + ); + expect(timePosSocket).toBeDefined(); + + const opening = openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + }); + await jest.advanceTimersByTimeAsync(0); + const loadfileSocket = sockets.find( + (socket) => commandOf(socket)[0] === 'loadfile' + ) as MpvSocketMock; + const requestId = await writtenRequestId(loadfileSocket); + + // The pending read now returns the new stream's position. + timePosSocket?.emit( + 'data', + Buffer.from(JSON.stringify({ data: 12.5, error: 'success' }) + '\n') + ); + await jest.advanceTimersByTimeAsync(10); + loadfileSocket.emit( + 'data', + Buffer.from( + JSON.stringify({ request_id: requestId, error: 'success' }) + + '\n' + ) + ); + await jest.advanceTimersByTimeAsync(0); + await expect(opening).resolves.toMatchObject({ status: 'opened' }); + + // An unguarded poll would now ask for the duration and, once that + // read times out, report the new position under the old content. + await jest.advanceTimersByTimeAsync(MPV_IPC_COMMAND_TIMEOUT_MS + 100); + expect( + sockets.some((socket) => commandOf(socket)[1] === 'duration') + ).toBe(false); + expect(sendPosition).not.toHaveBeenCalledWith( + previous.id, + expect.anything(), + expect.anything() + ); + sendPosition.mockRestore(); + }); + + it('keeps a reuse cancelled by disabling reuse stoppable until its child exits', async () => { + const { proc } = await launchReusableSession(); + (proc.kill as jest.Mock).mockImplementation(() => true); + const sockets = mockUnansweredSockets(); + jest.useFakeTimers(); + + try { + const opening = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + while (!sockets[0]?.write.mock.calls.length) { + await jest.advanceTimersByTimeAsync(0); + } + + setMpvReuseInstance(false); + sockets[0].emit('close'); + await jest.advanceTimersByTimeAsync(6_000); + + // The child never confirmed its exit: the session stays a + // closable error instead of a closed one Stop cannot retry. + await expect(opening).resolves.toMatchObject({ + status: 'error', + canClose: true, + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + }); + + it('keeps Stop working for a reused session restored after a later attempt fails', async () => { + const { proc } = await launchReusableSession(); + // Exit is never confirmed, so a failing later attempt cannot finish + // its teardown and the previous owner is restored. + (proc.kill as jest.Mock).mockImplementation(() => true); + mockMpvSockets(); + const restored = await openMpvPlayer({ + title: 'Channel A', + url: 'https://example.com/a.m3u8', + }); + expect(restored.status).toBe('opened'); + createConnectionMock.mockImplementation(() => { + const socket = createMpvSocketMock({ connect: false }); + setImmediate(() => + socket.emit('error', new Error('reuse command failed')) + ); + return socket; + }); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + jest.useFakeTimers(); + + try { + const failing = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + const rejection = expect(failing).rejects.toThrow( + 'External player process did not exit' + ); + await jest.advanceTimersByTimeAsync(6_000); + await rejection; + expect(externalPlayerSessions.getActiveSessionId()).toBe( + restored.id + ); + + const written: string[] = []; + mockMpvSockets({ written }); + let closed = false; + const closing = externalPlayerSessions + .closeSession(restored.id) + .then((session) => { + closed = true; + return session; + }); + await jest.advanceTimersByTimeAsync(10); + + // Stop reaches the child and waits for its exit. + expect( + written.map( + (chunk) => + (JSON.parse(chunk.trim()) as { command: string[] }) + .command[0] + ) + ).toContain('quit'); + expect(closed).toBe(false); + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + await expect(closing).resolves.toMatchObject({ + id: restored.id, + status: 'closed', + }); + } finally { + errorSpy.mockRestore(); + if (proc.exitCode === null) { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + } + } + }); + + it.each([ + ['reuse is disabled', () => setMpvReuseInstance(false)], + ['the app shuts down', () => shutdownMpvSession()], + ])( + 'does not fall back to a fresh player when %s during the reply wait', + async (_label, stopStored) => { + const { proc } = await launchReusableSession(); + // The kill is sent but its exit is reported only later, after + // the socket close already rejected the pending loadfile. + (proc.kill as jest.Mock).mockImplementation(() => true); + const sockets = mockUnansweredSockets(); + + const opening = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + await writtenRequestId(sockets[0]); + + stopStored(); + sockets[0].emit('close'); + await nextTick(); + await nextTick(); + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + + await expect(opening).resolves.toMatchObject({ + status: 'closed', + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } + ); + + it.each([ + ['reuse is disabled', () => setMpvReuseInstance(false)], + ['the app shuts down', () => shutdownMpvSession()], + ])( + 'does not fall back to a fresh player when %s during the failed-reuse teardown', + async (_label, stopStored) => { + const { proc } = await launchReusableSession(); + // mpv rejects the content command, and the teardown that + // follows waits for an exit that is reported only later. + (proc.kill as jest.Mock).mockImplementation(() => true); + mockMpvSockets({ + reply: (command) => + command[0] === 'loadfile' + ? 'error running command' + : 'success', + }); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + + try { + const opening = openMpvPlayer({ + title: 'Channel B', + url: 'https://example.com/b.m3u8', + }); + while (!(proc.kill as jest.Mock).mock.calls.length) { + await nextTick(); + } + + stopStored(); + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + + await expect(opening).resolves.toMatchObject({ + status: 'closed', + }); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + } + } + ); + + it('keeps the errored session when the child crashes during the reply wait', async () => { + const { proc, session: previous } = await launchReusableSession(); + const sockets = mockUnansweredSockets(); + const errorSpy = jest + .spyOn(console, 'error') + .mockImplementation(() => undefined); + + try { + const opening = openMpvPlayer({ + title: 'Second stream', + url: 'https://example.com/two.m3u8', + }); + await writtenRequestId(sockets[0]); + const replacementId = + externalPlayerSessions.getActiveSessionId() as string; + + // A crash closes the IPC socket before any reply: the exit + // handler reports it, and no fresh player is launched under a + // session that could never be shown as opened. + Object.defineProperty(proc, 'exitCode', { value: 1 }); + proc.emit('exit', 1); + sockets[0].emit('close'); + + await expect(opening).resolves.toMatchObject({ + id: replacementId, + status: 'error', + errorCode: 'closed-unexpectedly', + }); + expect(externalPlayerSessions.getSession(previous.id)?.status).toBe( + 'opened' + ); + expect(spawnMock).toHaveBeenCalledTimes(1); + } finally { + errorSpy.mockRestore(); + } + }); +}); diff --git a/apps/electron-backend/src/app/events/mpv-session.service.spec.ts b/apps/electron-backend/src/app/events/mpv-session.service.spec.ts index 0adfc02d2..a93ed2160 100644 --- a/apps/electron-backend/src/app/events/mpv-session.service.spec.ts +++ b/apps/electron-backend/src/app/events/mpv-session.service.spec.ts @@ -61,6 +61,11 @@ import { setVlcReuseInstance, shutdownVlcSession, } from './vlc-session.service'; +import { + createMpvSocketMock, + MpvSocketMock, + MpvSocketMockOptions, +} from './mpv-ipc-socket.test-helpers'; function createMockChildProcess(): ChildProcess { return Object.assign(new EventEmitter(), { @@ -95,6 +100,13 @@ function mockStoreValues(values: Record): void { ); } +/** Every reused-instance IPC connection answers like a healthy mpv. */ +function mockMpvSockets(options: MpvSocketMockOptions = {}): void { + (createConnection as unknown as jest.Mock).mockImplementation(() => + createMpvSocketMock(options) + ); +} + describe('external player shutdown on app quit', () => { beforeEach(() => { jest.clearAllMocks(); @@ -263,15 +275,7 @@ describe('external player shutdown on app quit', () => { title: 'First stream', url: 'https://example.com/one.m3u8', }); - (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => socket.emit('connect')); - return socket; - }); + mockMpvSockets(); const current = await openMpvPlayer({ title: 'Second stream', url: 'https://example.com/two.m3u8', @@ -305,15 +309,7 @@ describe('external player shutdown on app quit', () => { url: 'https://example.com/one.m3u8', }); const written: string[] = []; - (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn((chunk: string) => written.push(chunk)), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => socket.emit('connect')); - return socket; - }); + mockMpvSockets({ written }); const reused = await openMpvPlayer({ title: 'Second stream', url: 'https://example.com/two.m3u8', @@ -394,15 +390,7 @@ describe('external player shutdown on app quit', () => { expect(launchArgs).toContain('--start=42'); const written: string[] = []; - (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn((chunk: string) => written.push(chunk)), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => socket.emit('connect')); - return socket; - }); + mockMpvSockets({ written }); const loadfileOptionsOf = async (startTime: number | undefined) => { written.length = 0; await openMpvPlayer({ @@ -456,15 +444,7 @@ describe('external player shutdown on app quit', () => { // Capture the JSON IPC traffic of the second, reused launch. const written: string[] = []; - (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn((chunk: string) => written.push(chunk)), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => socket.emit('connect')); - return socket; - }); + mockMpvSockets({ written }); await openMpvPlayer({ title: 'Stalker live stream', @@ -509,15 +489,7 @@ describe('external player shutdown on app quit', () => { url: 'https://example.com/one.m3u8', }); - (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => socket.emit('connect')); - return socket; - }); + mockMpvSockets(); const reused = await openMpvPlayer({ title: 'Second stream', url: 'https://example.com/two.m3u8', @@ -560,15 +532,7 @@ describe('external player shutdown on app quit', () => { title: 'First stream', url: 'https://example.com/one.m3u8', }); - (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => socket.emit('connect')); - return socket; - }); + mockMpvSockets(); const reused = await openMpvPlayer({ title: 'Second stream', url: 'https://example.com/two.m3u8', @@ -622,27 +586,15 @@ describe('external player shutdown on app quit', () => { url: 'https://example.com/one.m3u8', }); - (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => socket.emit('connect')); - return socket; - }); + mockMpvSockets(); const session = await openMpvPlayer({ title: 'Second stream', url: 'https://example.com/two.m3u8', }); - const sockets: EventEmitter[] = []; + const sockets: MpvSocketMock[] = []; (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), - }); + const socket = createMpvSocketMock({ connect: false }); sockets.push(socket); return socket; }); @@ -676,19 +628,12 @@ describe('external player shutdown on app quit', () => { url: 'https://example.com/one.m3u8', }); - const sockets: Array< - EventEmitter & { write: jest.Mock; end: jest.Mock } - > = []; + const sockets: MpvSocketMock[] = []; (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), + const socket = createMpvSocketMock({ + connect: sockets.length >= 2, }); sockets.push(socket); - if (sockets.length > 2) { - setImmediate(() => socket.emit('connect')); - } return socket; }); @@ -720,7 +665,7 @@ describe('external player shutdown on app quit', () => { expect(spawn).toHaveBeenCalledTimes(1); }); - it('does not spawn a replacement when Stop interrupts reused MPV seek', async () => { + it('does not spawn a replacement when Stop interrupts a reused MPV loadfile reply', async () => { shutdownMpvSession(); const proc = createMockChildProcess(); (spawn as unknown as jest.Mock).mockReturnValue(proc); @@ -733,17 +678,14 @@ describe('external player shutdown on app quit', () => { url: 'https://example.com/one.m3u8', }); - const sockets: EventEmitter[] = []; + // The content command is written but mpv never answers it; the quit + // connection that Stop opens answers normally. + const sockets: MpvSocketMock[] = []; (createConnection as unknown as jest.Mock).mockImplementation(() => { - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), + const socket = createMpvSocketMock({ + reply: sockets.length === 0 ? () => null : undefined, }); sockets.push(socket); - if (sockets.length === 1 || sockets.length === 3) { - setImmediate(() => socket.emit('connect')); - } return socket; }); @@ -752,19 +694,19 @@ describe('external player shutdown on app quit', () => { url: 'https://example.com/two.m3u8', startTime: 120, }); - while (sockets.length < 2) { + while (!sockets[0]?.write.mock.calls.length) { await new Promise((resolve) => setImmediate(resolve)); } const replacementId = externalPlayerSessions.getActiveSessionId() as string; const closing = externalPlayerSessions.closeSession(replacementId); - while (sockets.length < 3) { + while (sockets.length < 2) { await new Promise((resolve) => setImmediate(resolve)); } Object.defineProperty(proc, 'exitCode', { value: 0 }); proc.emit('exit', 0); - sockets[1].emit('error', new Error('connection closed')); + sockets[0].emit('error', new Error('connection closed')); await expect(closing).resolves.toMatchObject({ id: replacementId, @@ -972,22 +914,11 @@ describe('external player shutdown on app quit', () => { title: 'First stream', url: 'https://example.com/one.m3u8', }); - let connectionCount = 0; - (createConnection as unknown as jest.Mock).mockImplementation(() => { - connectionCount += 1; - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => { - if (connectionCount === 1) { - socket.emit('connect'); - } else { - socket.emit('error', new Error('seek failed')); - } - }); - return socket; + // mpv rejects the content command after it was written, so the + // player may already have changed. + mockMpvSockets({ + reply: (command) => + command[0] === 'loadfile' ? 'error running command' : 'success', }); jest.useFakeTimers(); @@ -1153,22 +1084,11 @@ describe('external player shutdown on app quit', () => { title: 'First stream', url: 'https://example.com/one.m3u8', }); - let connectionCount = 0; - (createConnection as unknown as jest.Mock).mockImplementation(() => { - connectionCount += 1; - const socket = Object.assign(new EventEmitter(), { - write: jest.fn(), - end: jest.fn(), - destroy: jest.fn(), - }); - setImmediate(() => { - if (connectionCount === 1) { - socket.emit('connect'); - } else { - socket.emit('error', new Error('seek failed')); - } - }); - return socket; + // mpv rejects the content command after it was written, so the + // player may already have changed. + mockMpvSockets({ + reply: (command) => + command[0] === 'loadfile' ? 'error running command' : 'success', }); const session = await openMpvPlayer({ diff --git a/apps/electron-backend/src/app/events/mpv-session.service.ts b/apps/electron-backend/src/app/events/mpv-session.service.ts index f70ff2a2f..92e756419 100644 --- a/apps/electron-backend/src/app/events/mpv-session.service.ts +++ b/apps/electron-backend/src/app/events/mpv-session.service.ts @@ -1,6 +1,6 @@ import { spawn } from 'child_process'; -import { createConnection } from 'net'; import { + ExternalPlayerSession, PlayerContentInfo, readExternalPlayerErrorCode, } from '@iptvnator/shared/interfaces'; @@ -28,15 +28,25 @@ import { buildPlayerStartError, externalPlayerSessions, maskUrlForLogs, - sendPlaybackPositionUpdate, sendPlayerErrorNotification, traceExternalPlayer, } from './external-player-runtime'; import { externalPlayerProcessTeardownGate } from './external-player-process'; +import { + assertQueuedLaunchAllowed, + MpvLaunchQueue, + QueuedMpvLaunch, + settleCancelledQueuedLaunch, +} from './mpv-launch-queue'; import { MpvReusableProcess, MpvReuseAttemptState, } from './mpv-reusable-process'; +import { attachMpvOutputHandlers } from './mpv-output-errors'; +import { + startPositionPolling, + stopPositionPolling, +} from './mpv-position-polling'; export interface OpenExternalPlayerRequest { url: string; @@ -51,11 +61,6 @@ export interface OpenExternalPlayerRequest { } const reusableMpvProcess = new MpvReusableProcess(); -let positionPollingInterval: NodeJS.Timeout | null = null; -// The first poll waits for MPV to open its IPC socket. The handle is kept so -// a player that exits during that wait does not start an orphaned interval. -let positionPollingDelay: NodeJS.Timeout | null = null; - function getMpvPath(options: PlayerPathOptions = {}): string { return ( normalizeCustomPlayerPath(store.get(MPV_PLAYER_PATH)) ?? @@ -63,110 +68,6 @@ function getMpvPath(options: PlayerPathOptions = {}): string { ); } -async function getMpvProperty( - socketPath: string, - property: string -): Promise { - return new Promise((resolve) => { - const client = createConnection(socketPath); - const request = - JSON.stringify({ - command: ['get_property', property], - }) + '\n'; - - let data = ''; - - client.on('connect', () => { - client.write(request); - }); - - const timeoutHandle = setTimeout(() => { - client.destroy(); - resolve(null); - }, 2000); - - client.on('data', (chunk) => { - data += chunk.toString(); - if (data.includes('\n')) { - clearTimeout(timeoutHandle); - try { - const lines = data.split('\n'); - for (const line of lines) { - if (!line.trim()) continue; - const response = JSON.parse(line); - if (response.data !== undefined) { - client.destroy(); - resolve(response.data); - return; - } - } - } catch { - // Ignore partial JSON while MPV is still writing. - } - } - }); - - client.on('end', () => { - clearTimeout(timeoutHandle); - if (data && !data.includes('\n')) { - try { - const response = JSON.parse(data); - resolve(response.data ?? null); - } catch { - resolve(null); - } - } - }); - - client.on('error', () => { - clearTimeout(timeoutHandle); - resolve(null); - }); - }); -} - -function stopPositionPolling(): void { - if (positionPollingDelay) { - clearTimeout(positionPollingDelay); - positionPollingDelay = null; - } - if (positionPollingInterval) { - clearInterval(positionPollingInterval); - positionPollingInterval = null; - } -} - -function shouldIgnoreMpvStdoutLine(line: string): boolean { - return /^(\(Paused\)\s*)?AV:\s/.test(line); -} - -function startPositionPolling( - socketPath: string, - contentInfo: PlayerContentInfo, - sessionId: string -): void { - stopPositionPolling(); - - positionPollingDelay = setTimeout(() => { - positionPollingDelay = null; - positionPollingInterval = setInterval(async () => { - try { - const position = await getMpvProperty(socketPath, 'time-pos'); - const duration = await getMpvProperty(socketPath, 'duration'); - - if (position !== null) { - sendPlaybackPositionUpdate(sessionId, contentInfo, { - positionSeconds: Math.floor(position), - durationSeconds: duration ? Math.floor(duration) : null, - }); - } - } catch { - stopPositionPolling(); - } - }, 5000); - }, 2000); -} - export function setMpvReuseInstance(reuseInstance: boolean): void { traceExternalPlayer('set mpv reuse instance', { reuseInstance }); store.set(MPV_REUSE_INSTANCE, reuseInstance); @@ -186,33 +87,71 @@ export function setMpvReuseInstance(reuseInstance: boolean): void { * app and keeps playing after quit. */ export function shutdownMpvSession(): void { + mpvLaunchQueue.cancelAll(); reusableMpvProcess.stopStored( 'kill reused mpv process on app shutdown', stopPositionPolling ); } -export async function openMpvPlayer({ - url, - title, - thumbnail, - userAgent, - referer, - origin, - contentInfo, - startTime, - headers, -}: OpenExternalPlayerRequest) { - externalPlayerProcessTeardownGate.assertLaunchAllowed(); +// MPV launches run one at a time; see MpvLaunchQueue. Each launch begins its +// session at once, so the newest launch is the active one Stop targets. +const mpvLaunchQueue = new MpvLaunchQueue(() => + reusableMpvProcess.storedOwnerSessionId() +); + +export function openMpvPlayer( + request: OpenExternalPlayerRequest +): Promise { + try { + externalPlayerProcessTeardownGate.assertLaunchAllowed(); + } catch (error) { + return Promise.reject(error); + } const displacedSessionId = externalPlayerSessions.getActiveSessionId(); - const previousProcessSessionId = reusableMpvProcess.currentSessionId(); const session = externalPlayerSessions.beginSession({ player: 'mpv', - title, - thumbnail, - streamUrl: url, - contentInfo, + title: request.title, + thumbnail: request.thumbnail, + streamUrl: request.url, + contentInfo: request.contentInfo, }); + return mpvLaunchQueue.enqueue(session, (queue) => + launchMpvPlayer(request, session, displacedSessionId, queue) + ); +} + +async function launchMpvPlayer( + { + url, + title, + userAgent, + referer, + origin, + contentInfo, + startTime, + headers, + }: OpenExternalPlayerRequest, + session: ExternalPlayerSession, + displacedSessionId: string | null, + queue: QueuedMpvLaunch +): Promise { + const current = externalPlayerSessions.getSession(session.id); + // Stopped while it waited in the queue, directly or by a later Stop. + if (current?.status === 'closed') return current; + if (queue.queued && queue.cancelled()) { + return settleCancelledQueuedLaunch(session, queue.pendingStop()); + } + if (queue.queued) { + // The last started launch owns the child the gate waits for; the + // session displaced at enqueue time is the fallback. + assertQueuedLaunchAllowed(session, [ + queue.lastStartedSessionId(), + displacedSessionId, + ]); + } + queue.markInFlight(); + const previousProcessSessionId = reusableMpvProcess.currentSessionId(); const reuseState: MpvReuseAttemptState = { teardownUnconfirmed: false, contentMutated: false, @@ -356,61 +295,7 @@ export async function openMpvPlayer({ stdio: reuseInstance ? ['ignore', 'pipe', 'pipe'] : 'ignore', }); - if (proc.stdout) { - proc.stdout.on('data', (data) => { - const lines = data - .toString() - .split('\n') - .map((line) => line.trim()) - .filter(Boolean); - - for (const output of lines) { - if (shouldIgnoreMpvStdoutLine(output)) { - continue; - } - - traceExternalPlayer('mpv stdout', { output }); - - if ( - output.includes('Failed to open') || - output.includes('Error opening') || - output.includes('Protocol not found') || - output.includes('Connection refused') || - output.includes('error') || - output.includes('403') || - output.includes('404') - ) { - console.error('[MPV ERROR from stdout]:', output); - sendPlayerErrorNotification('MPV', output); - } - } - }); - } - - if (proc.stderr) { - proc.stderr.on('data', (data) => { - const output = data.toString().trim(); - if (output) { - console.error('[MPV stderr]:', output); - - if ( - output.includes('Failed to open') || - output.includes( - 'Exiting... (Errors when loading file)' - ) || - output.includes('Error opening') || - output.includes('Protocol not found') || - output.includes('Connection refused') || - output.includes('error') || - output.includes('403') || - output.includes('404') - ) { - console.error('[MPV ERROR]:', output); - sendPlayerErrorNotification('MPV', output); - } - } - }); - } + attachMpvOutputHandlers(proc); proc.on('error', (err) => { console.error('Failed to start MPV player:', err); diff --git a/apps/electron-backend/src/app/events/mpv-session.test-helpers.ts b/apps/electron-backend/src/app/events/mpv-session.test-helpers.ts new file mode 100644 index 000000000..ab109bc22 --- /dev/null +++ b/apps/electron-backend/src/app/events/mpv-session.test-helpers.ts @@ -0,0 +1,132 @@ +/** + * Shared fixtures for the reused-MPV specs: a mock child, a reusable + * instance to hand launches over to, and IPC sockets that answer, stay + * silent, or let `quit` stop the child. + */ +import { spawn, type ChildProcess } from 'child_process'; +import { EventEmitter } from 'events'; +import { createConnection } from 'net'; +import type { ExternalPlayerSession } from '@iptvnator/shared/interfaces'; +import { + MPV_PLAYER_PATH, + MPV_REUSE_INSTANCE, + store, +} from '../services/store.service'; +import { + createMpvSocketMock, + MpvSocketMock, + MpvSocketMockOptions, +} from './mpv-ipc-socket.test-helpers'; +import { openMpvPlayer } from './mpv-session.service'; + +export const spawnMock = spawn as unknown as jest.Mock; +export const createConnectionMock = createConnection as unknown as jest.Mock; + +export function createMockChildProcess(): ChildProcess { + const proc = Object.assign(new EventEmitter(), { + exitCode: null, + killed: false, + kill: jest.fn(), + signalCode: null, + stderr: null, + stdout: null, + unref: jest.fn(), + }) as unknown as ChildProcess; + (proc.kill as jest.Mock).mockImplementation(() => { + setImmediate(() => { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + }); + return true; + }); + return proc; +} + +export function mockMpvSockets(options: MpvSocketMockOptions = {}): void { + createConnectionMock.mockImplementation(() => createMpvSocketMock(options)); +} + +export function nextTick(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + +/** Spawns the reusable instance a later launch is handed over to. */ +export async function launchReusableInstance(): Promise { + return (await launchReusableSession()).proc; +} + +export async function launchReusableSession(): Promise<{ + proc: ChildProcess; + session: ExternalPlayerSession; +}> { + const proc = createMockChildProcess(); + spawnMock.mockReturnValueOnce(proc); + (store.get as unknown as jest.Mock).mockImplementation( + (key: string, fallback?: unknown) => + ({ + [MPV_PLAYER_PATH]: '/usr/bin/mpv', + [MPV_REUSE_INSTANCE]: true, + })[key] ?? fallback + ); + const session = await openMpvPlayer({ + title: 'First stream', + url: 'https://example.com/one.m3u8', + }); + return { proc, session }; +} + +/** A socket whose written command mpv never answers; the test replies. */ +export function mockUnansweredSockets(): MpvSocketMock[] { + const sockets: MpvSocketMock[] = []; + createConnectionMock.mockImplementation(() => { + const socket = createMpvSocketMock({ reply: () => null }); + sockets.push(socket); + return socket; + }); + return sockets; +} + +/** + * Like mockUnansweredSockets, but a `quit` succeeds and the child then exits, + * as a real mpv does. + */ +export function mockSocketsQuittingInto(proc: ChildProcess): MpvSocketMock[] { + const sockets: MpvSocketMock[] = []; + createConnectionMock.mockImplementation(() => { + const socket = createMpvSocketMock({ + reply: (command) => { + if (command[0] !== 'quit') return null; + setImmediate(() => { + Object.defineProperty(proc, 'exitCode', { value: 0 }); + proc.emit('exit', 0); + }); + return 'success'; + }, + }); + sockets.push(socket); + return socket; + }); + return sockets; +} + +export function commandsOf(sockets: MpvSocketMock[]): string[] { + return sockets.map( + (socket) => + ( + JSON.parse(String(socket.write.mock.calls[0][0])) as { + command: string[]; + } + ).command[0] + ); +} + +export async function writtenRequestId(socket: MpvSocketMock): Promise { + while (!socket?.write.mock.calls.length) { + await nextTick(); + } + return ( + JSON.parse(socket.write.mock.calls[0][0] as string) as { + request_id: number; + } + ).request_id; +} diff --git a/apps/web/src/app/services/external-playback.service.spec.ts b/apps/web/src/app/services/external-playback.service.spec.ts index ab7fee381..510b6b623 100644 --- a/apps/web/src/app/services/external-playback.service.spec.ts +++ b/apps/web/src/app/services/external-playback.service.spec.ts @@ -62,6 +62,49 @@ describe('ExternalPlaybackService', () => { expect(service.visibleSession()).toEqual(next); }); + it('keeps the newest launch when an older launch republishes as launching', () => { + const older = createSession({ + id: 'session-older', + startedAt: '2026-03-07T10:00:00.000Z', + }); + const newer = createSession({ + id: 'session-newer', + title: 'Newer Example', + startedAt: '2026-03-07T10:00:01.000Z', + }); + listener?.(older); + listener?.(newer); + + // The older launch is still settling and republishes itself. + listener?.({ + ...older, + canClose: true, + updatedAt: '2026-03-07T10:00:02.000Z', + }); + + expect(service.activeSession()?.id).toBe('session-newer'); + }); + + it('orders launches made in the same millisecond by launch order', () => { + const older = createSession({ id: 'session-older', launchOrder: 1 }); + const newer = createSession({ + id: 'session-newer', + title: 'Newer Example', + launchOrder: 2, + }); + listener?.(older); + listener?.(newer); + + // Same startedAt: only the launch order tells the two apart. + listener?.({ + ...older, + canClose: true, + updatedAt: '2026-03-07T10:00:02.000Z', + }); + + expect(service.activeSession()?.id).toBe('session-newer'); + }); + it('matches only active non-error sessions to content info', () => { const playing = createSession({ status: 'playing', diff --git a/apps/web/src/app/services/external-playback.service.ts b/apps/web/src/app/services/external-playback.service.ts index 161848f00..745032660 100644 --- a/apps/web/src/app/services/external-playback.service.ts +++ b/apps/web/src/app/services/external-playback.service.ts @@ -4,6 +4,17 @@ import { PlayerContentInfo, } from '@iptvnator/shared/interfaces'; +/** Launch order when both carry it, else the start time. */ +function startedBefore( + session: ExternalPlayerSession, + other: ExternalPlayerSession +): boolean { + if (session.launchOrder !== undefined && other.launchOrder !== undefined) { + return session.launchOrder < other.launchOrder; + } + return Date.parse(session.startedAt) < Date.parse(other.startedAt); +} + @Injectable({ providedIn: 'root', }) @@ -105,17 +116,25 @@ export class ExternalPlaybackService { const current = this.activeSession(); const restoresCurrentReplacement = current?.id === session.restoredFromSessionId; + // A launching update means a new launch only when it did not start + // before the current one: an older launch still settling (MPV runs + // launches one at a time) must not take the dock back from it. + const isLatestLaunch = + session.status === 'launching' && + (!current || + current.id === session.id || + !startedBefore(session, current)); if ( !current || current.id === session.id || - session.status === 'launching' || + isLatestLaunch || restoresCurrentReplacement ) { this.activeSession.set(session); } - if (session.status === 'launching' || restoresCurrentReplacement) { + if (isLatestLaunch || restoresCurrentReplacement) { this.dismissedSessionId.set(null); } } diff --git a/docs/architecture/embedded-inline-playback.md b/docs/architecture/embedded-inline-playback.md index abf41b729..628a4ad89 100644 --- a/docs/architecture/embedded-inline-playback.md +++ b/docs/architecture/embedded-inline-playback.md @@ -1029,6 +1029,61 @@ is also idempotent: it returns the closed snapshot without invoking the saved closer, and a later process error cannot revive it as a visible failure. Reused MPV commands use the socket captured for their exact child, so a delayed close cannot send `quit` to a replacement process through a newer global socket. +Every reused MPV command carries a `request_id` and waits, within the same 2 s +IPC bound, for mpv's matching newline-delimited reply; event lines and replies +to other requests on that connection are skipped. An error reply or a missing +reply fails the handoff exactly like a socket error: the reused child is torn +down and a fresh launch follows. `loadfile` marks the content as possibly +changed and moves ownership of the child to the attempted session from the +moment it is written, whatever mpv replies, and stops the previous session's +position poll there, including a read already awaiting mpv, so an exit or a +stale Stop for the previous session during +the reply wait settles the right session and no poll can save the new stream's +position under the old content; a child that exited while its reply was in +flight is returned as the session its exit handler settled, never reported as +opened and never replaced by a fresh launch under that terminal session. +MPV launches run one at a time. A reused launch waits for mpv's replies and a +failed one for its teardown and fallback spawn, so a launch that arrives +meanwhile, such as a quick channel switch, starts only once the previous one +has settled and acts on the player that one left behind; two launches never +act on the same child. Each launch begins its session at once, so the newest +launch is the active session; the renderer lets a `launching` update take the +dock only when that launch did not start before the current one, by the +session's `launchOrder` (which separates launches in the same millisecond) or +else its `startedAt`, so an older launch still settling cannot pull the dock +back. Only the newest queued launch +runs: an older +one that comes up while a newer one waits behind it settles closed without +reaching mpv or taking the dock. Stop on a launch still waiting in the queue +cancels it and every launch queued before it; on the newest queued launch it +also closes, through their own closers, the in-flight launch and the session +owning the stored player at that moment (read from the per-child mapping, which +a failed attempt hands back to the previous owner during its teardown), so +their exact-child, retry and settle rules apply. Because launches run one at a +time, nothing else acts on a reused child while an attempt is in flight, so +that attempt's own Stop always reaches its child until the attempt settles, +whatever the per-child mapping names (the previous owner before `loadfile` or +during a failed attempt's teardown, possibly closed already); after it settles, +ownership decides as before, except that a Stop of that attempt which could +not confirm the exit keeps the right to retry that exact child. These rules apply +and nothing starts playing after Stop. Those targets are fixed at the first +Stop, so a retried Stop never reaches a player started later, and a late Stop +for an older queued launch cancels only that launch. The cancelled launch +settles only once its own Stop has: if the exit cannot be confirmed it stays a +closable error, so Stop can be retried. App shutdown cancels the whole queue. +A queued launch that finds the teardown gate closed, because the launch before +it could not confirm its child's exit, restores the session that still owns +that child, preferring the last launch that started (with reuse off its +detached child is not stored anywhere else); if that is not possible, it stays +a closable error so Stop can still tear the child down. Stopping the stored player, when reuse is disabled or +the app quits, cancels an in-flight reuse attempt: it dispatches no further +commands, never falls back to a fresh reusable player, also after waiting for +a failed child's exit, and settles its own session as closed once the child's +exit is confirmed, or as a closable error when it is not. No `seek` follows +`loadfile`: mpv rejects a seek before the file is loaded, so the resume offset +travels as the per-file `start` option. The `loadfile` reply only acknowledges +the command; a stream that later fails to open is reported through the reused +child's output and exit, not through the reply. If Stop is observed before a pending MPV content command or VLC enqueue command is dispatched, that command is skipped. A source handoff also fails closed while a live session has no closer (`canClose: false`); renderer Dismiss is not @@ -1142,8 +1197,13 @@ External MPV/VLC integration is split across focused main-process modules: session tracking, trace logging, renderer notifications, playback-position forwarding, and user-facing start errors. - `apps/electron-backend/src/app/events/mpv-session.service.ts` owns fresh MPV - launches and progress polling; `mpv-reusable-process.ts` owns the tracked - child, captured socket, remapping, retryable close, and reuse handoff. + launches; `mpv-position-polling.ts` owns progress polling over the IPC socket + and `mpv-output-errors.ts` the child's stdout/stderr error reporting; + `mpv-reusable-process.ts` owns the tracked + child, captured socket, remapping, retryable close, and reuse handoff; + `mpv-ipc-command.ts` owns the bounded request/reply IPC command the reuse + handoff and its quit use; `mpv-launch-queue.ts` runs MPV launches one at a + time and owns queued-launch cancellation and Stop. - `apps/electron-backend/src/app/events/vlc-session.service.ts` owns fresh VLC launches and progress polling; `vlc-reusable-process.ts` owns the tracked child, RC-port remapping, retryable close, and reuse handoff, while diff --git a/docs/architecture/idle-work-audit-2026-09.md b/docs/architecture/idle-work-audit-2026-09.md index 8552a9a7b..569c67331 100644 --- a/docs/architecture/idle-work-audit-2026-09.md +++ b/docs/architecture/idle-work-audit-2026-09.md @@ -210,7 +210,7 @@ user who leaves that screen open pays them indefinitely, minimized included. | Source | What it does | Period | Cost per firing | Justified | Evidence | Follow-up | | --- | --- | --- | --- | --- | --- | --- | -| [mpv-session.service.ts:140–141](../../apps/electron-backend/src/app/events/mpv-session.service.ts) | External MPV position poll | 2 s delay, then 5 s | Two IPC-socket round-trips and one `playback-position-update` | Yes while playing. The early-exit leak found here (an unstored start-delay handle leaving an orphaned 5 s poll) is **resolved by #1720**. Still open: in reuse mode the poll keeps querying an idle MPV (no IPC while `time-pos` is null). | Static | Early-exit leak: done (#1720). Reuse mode: needs an idle signal from MPV before the poll can stop; not worth its own thread at two local socket calls per 5 s. | +| [mpv-position-polling.ts](../../apps/electron-backend/src/app/events/mpv-position-polling.ts) | External MPV position poll | 2 s delay, then 5 s | Two IPC-socket round-trips and one `playback-position-update` | Yes while playing. The early-exit leak found here (an unstored start-delay handle leaving an orphaned 5 s poll) is **resolved by #1720**. Still open: in reuse mode the poll keeps querying an idle MPV (no IPC while `time-pos` is null). | Static | Early-exit leak: done (#1720). Reuse mode: needs an idle signal from MPV before the poll can stop; not worth its own thread at two local socket calls per 5 s. | | [vlc-session.service.ts:81–82](../../apps/electron-backend/src/app/events/vlc-session.service.ts) | External VLC position poll | 1.5 s delay, then 2 s | Up to three localhost TCP connections, then IPC | Yes while playing. The early-exit leak (orphaned 2 s poll) is **resolved by #1720**. | Static | Done (#1720). | | [embedded-mpv-native.service.ts:1068](../../apps/electron-backend/src/app/services/embedded-mpv-native.service.ts) | Embedded MPV session snapshot poll | 500 ms while a session exists | Native snapshot, diff, and IPC only on change | Yes during playback. It is cleared when the last session closes. | Static | None | | [embedded-mpv-session-controller.ts:197](../../libs/ui/playback/src/lib/embedded-mpv-player/embedded-mpv-session-controller.ts) | Renderer-side bounds poll for a native-view embedded MPV session: compares `getBoundingClientRect()` with the last synced bounds | 500 ms while a native-view session is open (skipped for frame-copy) | One layout read outside the Angular zone; on drift a rAF and a `setEmbeddedMpvBounds` IPC | Yes during playback: a position-only layout shift would otherwise leave the native view misplaced | Static | Cover with the throttling thread: it kept its rate while minimized under `backgroundThrottling: false`. | diff --git a/libs/shared/interfaces/src/lib/external-player-session.interface.ts b/libs/shared/interfaces/src/lib/external-player-session.interface.ts index d7c17c436..2df17762a 100644 --- a/libs/shared/interfaces/src/lib/external-player-session.interface.ts +++ b/libs/shared/interfaces/src/lib/external-player-session.interface.ts @@ -92,6 +92,11 @@ export interface ExternalPlayerSession { contentInfo?: PlayerContentInfo; startedAt: string; updatedAt: string; + /** + * Order in which sessions began in this app run. Unlike `startedAt` it + * tells apart launches made in the same millisecond. + */ + launchOrder?: number; /** Raw failure detail; the dock translates `errorCode` instead. */ error?: string; errorCode?: ExternalPlayerErrorCode; diff --git a/tools/eslint/max-lines-baseline.mjs b/tools/eslint/max-lines-baseline.mjs index 1afe8ffae..ea5ee98de 100644 --- a/tools/eslint/max-lines-baseline.mjs +++ b/tools/eslint/max-lines-baseline.mjs @@ -13,7 +13,6 @@ export const maxLinesBaseline = [ 'apps/electron-backend/src/app/database/operations/content.operations.ts', 'apps/electron-backend/src/app/database/operations/playlist.operations.ts', 'apps/electron-backend/src/app/events/epg-query.service.ts', - 'apps/electron-backend/src/app/events/mpv-session.service.ts', 'apps/electron-backend/src/app/events/vlc-session.service.ts', 'apps/electron-backend/src/app/services/app-update.service.ts', 'apps/electron-backend/src/app/services/embedded-mpv-native.service.ts',