diff --git a/.changes/electron-backend-zero-delay-cancellation.md b/.changes/electron-backend-zero-delay-cancellation.md new file mode 100644 index 000000000..6f5aef92e --- /dev/null +++ b/.changes/electron-backend-zero-delay-cancellation.md @@ -0,0 +1,8 @@ +--- +type: fix +area: electron-backend +--- + +Cancelling a large Xtream import now stops database work even while it is +running at full speed, instead of letting the import finish before processing +the cancel request. diff --git a/apps/electron-backend/src/app/workers/database-worker-zero-delay-cancellation.spec.ts b/apps/electron-backend/src/app/workers/database-worker-zero-delay-cancellation.spec.ts new file mode 100644 index 000000000..b32236964 --- /dev/null +++ b/apps/electron-backend/src/app/workers/database-worker-zero-delay-cancellation.spec.ts @@ -0,0 +1,180 @@ +import { MessageChannel, type MessagePort } from 'node:worker_threads'; +import type { + DbWorkerIncomingMessage, + DbWorkerMessage, + DbWorkerResponseMessage, +} from './database-worker.types'; + +const BATCH_DELAY_ENV = 'IPTVNATOR_DB_WORKER_BATCH_DELAY_MS'; +const WORKER_PROFILING_ENV = 'IPTVNATOR_PERF_WORKER_PROFILING'; +const originalBatchDelay = process.env[BATCH_DELAY_ENV]; +const originalWorkerProfiling = process.env[WORKER_PROFILING_ENV]; + +function restoreEnvironment(name: string, value: string | undefined): void { + if (value === undefined) { + delete process.env[name]; + } else { + process.env[name] = value; + } +} + +function postWorkerMessage( + port: MessagePort, + message: DbWorkerIncomingMessage +): void { + port.postMessage(message); +} + +describe('database worker zero-delay cancellation', () => { + let workerPort: MessagePort | null = null; + let clientPort: MessagePort | null = null; + + afterEach(() => { + workerPort?.close(); + clientPort?.close(); + workerPort = null; + clientPort = null; + restoreEnvironment(BATCH_DELAY_ENV, originalBatchDelay); + restoreEnvironment(WORKER_PROFILING_ENV, originalWorkerProfiling); + jest.restoreAllMocks(); + jest.resetModules(); + }); + + it('processes cancel posted from progress before the default-delay operation completes', async () => { + delete process.env[BATCH_DELAY_ENV]; + delete process.env[WORKER_PROFILING_ENV]; + + const channel = new MessageChannel(); + workerPort = channel.port1; + clientPort = channel.port2; + const messages: DbWorkerMessage[] = []; + let cancelPosted = false; + let settleResponse!: (response: DbWorkerResponseMessage) => void; + const responsePromise = new Promise( + (resolve) => { + settleResponse = resolve; + } + ); + + clientPort.on('message', (message: DbWorkerMessage) => { + messages.push(message); + + if ( + !cancelPosted && + message.type === 'event' && + message.event.status === 'progress' + ) { + cancelPosted = true; + postWorkerMessage(clientPort!, { + type: 'cancel', + operationId: 'operation-zero-delay', + }); + } + + if (message.type === 'response') { + settleResponse(message); + } + }); + + jest.doMock('worker_threads', () => ({ + ...jest.requireActual('worker_threads'), + parentPort: workerPort, + })); + jest.doMock('./database.worker-connection', () => ({ + closeWorkerDatabase: jest.fn(), + getWorkerDatabase: jest.fn().mockResolvedValue({}), + })); + jest.doMock('../database/operations/content.operations', () => ({ + saveContent: jest.fn( + async ( + _db: unknown, + _playlistId: string, + _streams: unknown[], + _type: string, + control: { + checkpoint: () => Promise; + onProgress: (progress: { + phase: string; + current: number; + total: number; + }) => Promise; + } + ) => { + for (let current = 1; current <= 3; current += 1) { + await control.checkpoint(); + await control.onProgress({ + phase: 'saving-content', + current, + total: 3, + }); + } + return { count: 3, success: true }; + } + ), + })); + jest.doMock('./worker-performance-capture', () => ({ + armWorkerPerformanceCapture: jest.fn(), + executeWithWorkerPerformanceCapture: jest.fn( + async (_capture: unknown, execute: () => Promise) => { + try { + return { + error: null, + performance: undefined, + result: await execute(), + success: true, + }; + } catch (error) { + return { + error, + performance: undefined, + result: undefined, + success: false, + }; + } + } + ), + registerDatabaseWorkerPerformanceCapture: jest.fn(), + releaseDatabaseWorkerPerformanceCapture: jest.fn(), + stampWorkerPerformanceResponsePostedEpoch: jest.fn( + (_capture: unknown, performance: unknown) => performance + ), + startWorkerPerformanceCapture: jest.fn(() => null), + })); + jest.spyOn(console, 'error').mockImplementation(() => undefined); + + await import('./database.worker'); + postWorkerMessage(clientPort, { + type: 'request', + operation: 'DB_SAVE_CONTENT', + payload: { + operationId: 'operation-zero-delay', + playlistId: 'playlist-1', + streams: [{ stream_id: 1 }, { stream_id: 2 }], + type: 'live', + }, + requestId: 'request-zero-delay', + }); + + const response = await responsePromise; + + expect(cancelPosted).toBe(true); + expect(messages).toContainEqual( + expect.objectContaining({ + event: expect.objectContaining({ + operationId: 'operation-zero-delay', + status: 'cancelled', + }), + requestId: 'request-zero-delay', + type: 'event', + }) + ); + expect(response).toEqual( + expect.objectContaining({ + error: expect.objectContaining({ name: 'AbortError' }), + requestId: 'request-zero-delay', + success: false, + type: 'response', + }) + ); + }); +}); diff --git a/apps/electron-backend/src/app/workers/database.worker.ts b/apps/electron-backend/src/app/workers/database.worker.ts index 61ca7f54d..7ac3e6501 100644 --- a/apps/electron-backend/src/app/workers/database.worker.ts +++ b/apps/electron-backend/src/app/workers/database.worker.ts @@ -192,6 +192,7 @@ function postEvent(requestId: string, event: DbOperationEvent): void { async function pauseBetweenBatches(): Promise { if (batchDelayMs <= 0) { + await new Promise((resolve) => setImmediate(resolve)); return; } diff --git a/docs/architecture/sqlite-db-worker.md b/docs/architecture/sqlite-db-worker.md index 54f8bc266..56504eb66 100644 --- a/docs/architecture/sqlite-db-worker.md +++ b/docs/architecture/sqlite-db-worker.md @@ -723,7 +723,10 @@ For deterministic E2E timing, tests may set: IPTVNATOR_DB_WORKER_BATCH_DELAY_MS=20 ``` -This delay is test-only and disabled by default. +This artificial delay is test-only and disabled by default. At the default +value of `0`, every cancellable batch checkpoint still yields one event-loop +turn without adding a timer delay. That yield lets the worker receive a queued +cancel message before starting the next batch. ### Useful verification commands