diff --git a/CLAUDE.md b/CLAUDE.md index 3988dffb8..beab81573 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -747,7 +747,7 @@ This project uses modern Angular signal-based APIs and patterns. **ALWAYS** use - `database.events.ts` - Database CRUD operations - `playlist.events.ts` - Playlist import/update - `playlist-open.events.ts` - Playlist files handed over by the OS (argv, file association, macOS `open-file`); the queue itself lives in `services/playlist-open-request.ts` - - `epg.events.ts` - EPG IPC registration; freshness/fetch orchestration lives in `epg-fetch.service.ts`, manual channel-mapping resolution and CRUD in `epg-mapping.service.ts`, worker lifecycle in `epg-worker.service.ts`, DB lookups in `epg-query.service.ts` + - `epg.events.ts` - EPG IPC registration; freshness/fetch orchestration lives in `epg-fetch.service.ts`, manual channel-mapping resolution and CRUD in `epg-mapping.service.ts`, source orchestration in `epg-worker.service.ts`, per-import lifecycle in `epg-fetch-operation.ts`, worker bootstrap/shutdown and clear protocol in `epg-worker-runtime.ts`, DB lookups in `epg-query.service.ts` - `xtream.events.ts` - Xtream Codes API - `stalker.events.ts` - Stalker portal API - `connectivity-guard.events.ts` - `CONNECTIVITY_GUARD_RESET`: forgets the connection failures recorded for a portal host. Both portal handlers above run every request through the per-host circuit breaker (rules in `@iptvnator/shared/host-health`, process-wide instance in `util/host-connectivity-guard.ts`; the web backend runs the same breaker over its proxy routes) — after 2 consecutive connection-level failures (no HTTP response; `ETIMEDOUT`/`ENOTFOUND`/`ECONNREFUSED`/… but never `ECONNRESET`) requests to that endpoint fail immediately for 30 s. The key is `URL.origin`, not `URL.host`, which would give `http://panel` and `https://panel` one shared record and let a dead TLS listener fast-fail the working HTTP one instead of hanging the full 30 s/15 s axios timeout again, with one half-open trial request afterwards. Any HTTP response (4xx and 5xx included) clears the record. The refusal is a real `Error` whose wording is a renderer contract (`buildHostConnectivityFastFailMessage` in `libs/shared/interfaces`): it must carry no `HTTP Error `, no timeout wording and none of the auth phrases, or Stalker endpoint discovery misclassifies it and lazy portal repair fires against a host just declared dead. Discovery probes are exempt via the `skipConnectionGuard` payload flag (bypass + no failure counting, but successes still clear the record). Every user-driven retry/refresh that issues portal requests must reset BEFORE its first request, or the affordance fast-fails and looks broken; automatic and first-load paths deliberately do not reset. Current senders: Xtream content-gate Retry, Stalker catalog append retry (`retryContentPage`), Stalker search-page retry, `StalkerItvCacheService.refresh()` (Live TV refresh), both account-info dialogs' Retry, the destructive Xtream refresh (`XtreamRefreshFlowService`, before it deletes the cached catalog — one flow shared by both entry points, `PlaylistRefreshActionService.refreshXtream()` and `RecentPlaylistsComponent.refreshXtreamPlaylist()`, which supply only a progress reporter), `StalkerPortalDiscoveryService.discover()`, and `PortalStatusService` on `skipCache`. Kill switch: `IPTVNATOR_DISABLE_CONNECTIVITY_GUARD=1`. Contract: `docs/architecture/host-connectivity-guard.md` diff --git a/apps/electron-backend/src/app/events/epg-fetch-operation.ts b/apps/electron-backend/src/app/events/epg-fetch-operation.ts new file mode 100644 index 000000000..a9809dc7a --- /dev/null +++ b/apps/electron-backend/src/app/events/epg-fetch-operation.ts @@ -0,0 +1,280 @@ +import type { Worker } from 'worker_threads'; +import type { + ElectronBridgeSecurityErrorCode, + ElectronBridgeTrustOptions, +} from '@iptvnator/shared/interfaces'; +import { epgSourceGeneration } from './epg-source-generation'; +import type { EpgWorkerRuntime } from './epg-worker-runtime'; +import type { EpgProgressStats, EpgWorkerService } from './epg-worker.service'; + +interface EpgWorkerMessage { + type: string; + error?: string; + errorCode?: ElectronBridgeSecurityErrorCode; + errorHost?: string; + url?: string; + stats?: EpgProgressStats; +} + +interface FetchContext { + runtime: EpgWorkerRuntime; + workers: Map; + fetchedUrls: Set; + loggerLabel: string; + fetchTimeoutMs: number; + sendProgressToRenderer: EpgWorkerService['sendProgressToRenderer']; +} + +/** One import's message, cancellation, timeout and exit lifecycle. */ +export function runEpgFetch( + url: string, + options: ElectronBridgeTrustOptions, + context: FetchContext +): Promise { + const generation = epgSourceGeneration(url); + return new Promise((resolve, reject) => { + let worker: Worker; + try { + worker = context.runtime.createEpgWorker(); + } catch (error) { + console.error( + context.loggerLabel, + 'Failed to create worker:', + error + ); + reject(error); + return; + } + + context.workers.set(url, worker); + + // Guards against double-settling and keeps the outer loop moving + // when the worker dies or hangs without sending EPG_COMPLETE/EPG_ERROR. + let settled = false; + let timeoutId: ReturnType | undefined; + let lastProgressStats: EpgProgressStats = { + totalChannels: 0, + totalPrograms: 0, + }; + + const clearFetchTimeout = () => { + if (timeoutId) { + clearTimeout(timeoutId); + timeoutId = undefined; + } + }; + + const settle = (fn: () => void) => { + if (settled) return; + settled = true; + clearFetchTimeout(); + fn(); + }; + + const cancelIfRetired = (exited = false): boolean => { + if (generation === epgSourceGeneration(url)) return false; + if (context.workers.get(url) === worker) + context.workers.delete(url); + settle(() => { + context.sendProgressToRenderer( + url, + 'cancelled', + undefined, + undefined, + undefined, + undefined, + undefined, + generation + ); + if (exited) resolve(); + else + void context.runtime + .terminateWorker(worker, 'retired fetch') + .then(() => resolve()); + }); + return true; + }; + + const scheduleFetchTimeout = () => { + clearFetchTimeout(); + timeoutId = setTimeout(() => { + handleFetchTimeout(); + }, context.fetchTimeoutMs); + }; + + const hasProgressMoved = (stats: EpgProgressStats): boolean => + stats.totalChannels > lastProgressStats.totalChannels || + stats.totalPrograms > lastProgressStats.totalPrograms; + + const recordProgress = (stats: EpgProgressStats): void => { + lastProgressStats = { + totalChannels: Math.max( + lastProgressStats.totalChannels, + stats.totalChannels + ), + totalPrograms: Math.max( + lastProgressStats.totalPrograms, + stats.totalPrograms + ), + }; + }; + + const handleFetchTimeout = () => { + if (settled || cancelIfRetired()) return; + const errorMessage = `EPG fetch timed out after ${ + context.fetchTimeoutMs / 1000 + }s without progress`; + console.error(context.loggerLabel, `${errorMessage}: ${url}`); + context.sendProgressToRenderer( + url, + 'error', + undefined, + errorMessage + ); + context.workers.delete(url); + // Settle only after the worker thread is really gone: a + // terminated-but-still-running worker can keep holding the + // SQLite lock and block the next EPG fetch. + settle(() => { + void context.runtime + .terminateWorker(worker, 'timed out fetch') + .then(() => reject(new Error(errorMessage))); + }); + }; + + scheduleFetchTimeout(); + + worker.on('message', async (message: EpgWorkerMessage) => { + if (settled || generation !== epgSourceGeneration(url)) return; + try { + switch (message.type) { + case 'READY': + scheduleFetchTimeout(); + context.sendProgressToRenderer(url, 'loading', { + totalChannels: 0, + totalPrograms: 0, + }); + worker.postMessage({ + type: 'FETCH_EPG', + url, + options, + }); + break; + + case 'EPG_PROGRESS': + if (message.stats) { + if (hasProgressMoved(message.stats)) { + recordProgress(message.stats); + scheduleFetchTimeout(); + } + context.sendProgressToRenderer( + url, + 'loading', + message.stats + ); + } + break; + + case 'EPG_COMPLETE': + console.log( + context.loggerLabel, + `EPG parsing complete for ${url}:`, + message.stats + ); + context.sendProgressToRenderer( + url, + 'complete', + message.stats + ); + context.fetchedUrls.add(url); + context.workers.delete(url); + settle(() => { + void context.runtime + .terminateWorker(worker, 'completed fetch') + .then(() => resolve()); + }); + break; + + case 'EPG_ERROR': + console.error( + context.loggerLabel, + 'Worker error:', + message.error + ); + context.sendProgressToRenderer( + url, + 'error', + undefined, + message.error, + undefined, + message.errorCode, + message.errorHost + ); + context.workers.delete(url); + settle(() => { + void context.runtime + .terminateWorker(worker, 'failed fetch') + .then(() => + reject( + new Error( + message.error || 'Unknown error' + ) + ) + ); + }); + break; + } + } catch (err) { + if (settled || cancelIfRetired()) return; + console.error( + context.loggerLabel, + 'Error handling message:', + err + ); + context.sendProgressToRenderer( + url, + 'error', + undefined, + err instanceof Error ? err.message : String(err) + ); + context.workers.delete(url); + settle(() => { + void context.runtime + .terminateWorker(worker, 'failed message handling') + .then(() => reject(err)); + }); + } + }); + + worker.on('error', (error) => { + if (settled || cancelIfRetired()) return; + console.error(context.loggerLabel, 'Worker error event:', error); + context.sendProgressToRenderer( + url, + 'error', + undefined, + error.message + ); + context.workers.delete(url); + settle(() => { + void context.runtime + .terminateWorker(worker, 'errored fetch') + .then(() => reject(error)); + }); + }); + + worker.on('exit', (code) => { + if (settled || cancelIfRetired(true)) return; + const errorMessage = `Worker exited unexpectedly (code ${code})`; + console.error(context.loggerLabel, `${errorMessage}: ${url}`); + context.sendProgressToRenderer( + url, + 'error', + undefined, + errorMessage + ); + context.workers.delete(url); + settle(() => reject(new Error(errorMessage))); + }); + }); +} diff --git a/apps/electron-backend/src/app/events/epg-worker-runtime.ts b/apps/electron-backend/src/app/events/epg-worker-runtime.ts new file mode 100644 index 000000000..45c444af2 --- /dev/null +++ b/apps/electron-backend/src/app/events/epg-worker-runtime.ts @@ -0,0 +1,156 @@ +import { app } from 'electron'; +import * as path from 'path'; +import { pathToFileURL } from 'url'; +import { Worker } from 'worker_threads'; +import { resolveWorkerRuntimeBootstrap } from '../workers/worker-runtime-paths'; + +interface ClearWorkerMessage { + type: 'CLEAR_EPG' | 'CLEAR_EPG_SOURCE'; + sourceUrl?: string; +} + +/** Worker bootstrap, shutdown and one-shot cache-clear protocol. */ +export class EpgWorkerRuntime { + constructor( + private readonly loggerLabel: string, + private readonly fetchTimeoutMs: number + ) {} + + runClearWorker(options: { + timeoutLabel: string; + exitLabel: string; + readyMessage: ClearWorkerMessage; + completeWorkerLabel: string; + failedWorkerLabel: string; + erroredWorkerLabel: string; + onComplete: (worker: Worker) => Promise; + }): Promise { + return new Promise((resolve, reject) => { + let worker: Worker; + try { + worker = this.createEpgWorker(); + } catch (error) { + console.error( + this.loggerLabel, + 'Failed to create worker for clear:', + error + ); + reject(error); + return; + } + + let settled = false; + const settle = (fn: () => void) => { + if (settled) return; + settled = true; + clearTimeout(timeoutId); + fn(); + }; + + const timeoutId = setTimeout(() => { + const errorMessage = `${options.timeoutLabel} timed out after ${ + this.fetchTimeoutMs / 1000 + }s`; + console.error(this.loggerLabel, errorMessage); + settle(() => { + void this.terminateWorker( + worker, + `timed out ${options.timeoutLabel}` + ).then(() => reject(new Error(errorMessage))); + }); + }, this.fetchTimeoutMs); + + worker.on( + 'message', + (message: { type: string; error?: string }) => { + if (message.type === 'READY') { + worker.postMessage(options.readyMessage); + } else if (message.type === 'CLEAR_COMPLETE') { + settle(() => { + void options + .onComplete(worker) + .then(() => resolve(), reject); + }); + } else if (message.type === 'EPG_ERROR') { + console.error( + this.loggerLabel, + 'Worker clear error:', + message.error + ); + settle(() => { + void this.terminateWorker( + worker, + options.failedWorkerLabel + ).then(() => + reject( + new Error(message.error || 'Clear failed') + ) + ); + }); + } + } + ); + + worker.on('error', (error) => { + console.error( + this.loggerLabel, + 'Worker error during clear:', + error + ); + settle(() => { + void this.terminateWorker( + worker, + options.erroredWorkerLabel + ).then(() => reject(error)); + }); + }); + + worker.on('exit', (code) => { + if (settled) return; + const errorMessage = `${options.exitLabel} exited unexpectedly (code ${code})`; + console.error(this.loggerLabel, errorMessage); + settle(() => reject(new Error(errorMessage))); + }); + }); + } + + /** + * Awaits worker shutdown so callers can sequence work (e.g. the next DB + * access) after the thread has really exited. Termination failures are + * logged and swallowed — there is nothing actionable left to do. + */ + async terminateWorker(worker: Worker, context: string): Promise { + try { + await worker.terminate(); + } catch (error) { + console.error( + this.loggerLabel, + `Failed to terminate ${context} worker:`, + error + ); + } + } + + createEpgWorker(): Worker { + const bootstrap = resolveWorkerRuntimeBootstrap({ + isPackaged: app.isPackaged, + workerFilename: 'epg-parser.worker.js', + developmentWorkerDir: path.join(__dirname, 'workers'), + resourcesPath: ( + process as NodeJS.Process & { resourcesPath?: string } + ).resourcesPath, + appPath: app.getAppPath(), + }); + + const workerURL = pathToFileURL(bootstrap.workerPath); + return new Worker(workerURL, { + resourceLimits: { + maxOldGenerationSizeMb: 4096, + maxYoungGenerationSizeMb: 512, + }, + workerData: { + nativeModuleSearchPaths: bootstrap.nativeModuleSearchPaths, + }, + }); + } +} diff --git a/apps/electron-backend/src/app/events/epg-worker.service.ts b/apps/electron-backend/src/app/events/epg-worker.service.ts index 64f00726f..647c592be 100644 --- a/apps/electron-backend/src/app/events/epg-worker.service.ts +++ b/apps/electron-backend/src/app/events/epg-worker.service.ts @@ -3,15 +3,14 @@ import { requestEpgSource, retireEpgSource, } from './epg-source-generation'; -import { app, BrowserWindow } from 'electron'; -import * as path from 'path'; -import { pathToFileURL } from 'url'; +import { BrowserWindow } from 'electron'; +import { EpgWorkerRuntime } from './epg-worker-runtime'; +import { runEpgFetch } from './epg-fetch-operation'; import { Worker } from 'worker_threads'; import { ElectronBridgeSecurityErrorCode, ElectronBridgeTrustOptions, } from '@iptvnator/shared/interfaces'; -import { resolveWorkerRuntimeBootstrap } from '../workers/worker-runtime-paths'; export type EpgProgressStatus = 'queued' | 'loading' | 'complete' | 'error' | 'cancelled'; @@ -21,30 +20,20 @@ export interface EpgProgressStats { totalPrograms: number; } -interface EpgWorkerMessage { - type: string; - error?: string; - errorCode?: ElectronBridgeSecurityErrorCode; - errorHost?: string; - url?: string; - stats?: EpgProgressStats; -} - -interface ClearWorkerMessage { - type: 'CLEAR_EPG' | 'CLEAR_EPG_SOURCE'; - sourceUrl?: string; -} - export class EpgWorkerService { private readonly fetchedUrls = new Set(); private readonly workers = new Map(); private readonly inFlightFetches = new Map>(); private readonly inFlightSourceClears = new Map>(); + private readonly runtime: EpgWorkerRuntime; + constructor( private readonly loggerLabel = '[EPG Events]', private readonly fetchTimeoutMs = 5 * 60 * 1000 - ) {} + ) { + this.runtime = new EpgWorkerRuntime(loggerLabel, fetchTimeoutMs); + } hasFetchedUrl(url: string): boolean { return this.fetchedUrls.has(url); @@ -142,258 +131,18 @@ export class EpgWorkerService { url: string, options: ElectronBridgeTrustOptions ): Promise { - const generation = epgSourceGeneration(url); - return new Promise((resolve, reject) => { - let worker: Worker; - try { - worker = this.createEpgWorker(); - } catch (error) { - console.error( - this.loggerLabel, - 'Failed to create worker:', - error - ); - reject(error); - return; - } - - this.workers.set(url, worker); - - // Guards against double-settling and keeps the outer loop moving - // when the worker dies or hangs without sending EPG_COMPLETE/EPG_ERROR. - let settled = false; - let timeoutId: ReturnType | undefined; - let lastProgressStats: EpgProgressStats = { - totalChannels: 0, - totalPrograms: 0, - }; - - const clearFetchTimeout = () => { - if (timeoutId) { - clearTimeout(timeoutId); - timeoutId = undefined; - } - }; - - const settle = (fn: () => void) => { - if (settled) return; - settled = true; - clearFetchTimeout(); - fn(); - }; - - const cancelIfRetired = (exited = false): boolean => { - if (generation === epgSourceGeneration(url)) return false; - if (this.workers.get(url) === worker) this.workers.delete(url); - settle(() => { - this.sendProgressToRenderer( - url, - 'cancelled', - undefined, - undefined, - undefined, - undefined, - undefined, - generation - ); - if (exited) resolve(); - else - void this.terminateWorker(worker, 'retired fetch').then( - () => resolve() - ); - }); - return true; - }; - - const scheduleFetchTimeout = () => { - clearFetchTimeout(); - timeoutId = setTimeout(() => { - handleFetchTimeout(); - }, this.fetchTimeoutMs); - }; - - const hasProgressMoved = (stats: EpgProgressStats): boolean => - stats.totalChannels > lastProgressStats.totalChannels || - stats.totalPrograms > lastProgressStats.totalPrograms; - - const recordProgress = (stats: EpgProgressStats): void => { - lastProgressStats = { - totalChannels: Math.max( - lastProgressStats.totalChannels, - stats.totalChannels - ), - totalPrograms: Math.max( - lastProgressStats.totalPrograms, - stats.totalPrograms - ), - }; - }; - - const handleFetchTimeout = () => { - if (settled || cancelIfRetired()) return; - const errorMessage = `EPG fetch timed out after ${ - this.fetchTimeoutMs / 1000 - }s without progress`; - console.error(this.loggerLabel, `${errorMessage}: ${url}`); - this.sendProgressToRenderer( - url, - 'error', - undefined, - errorMessage - ); - this.workers.delete(url); - // Settle only after the worker thread is really gone: a - // terminated-but-still-running worker can keep holding the - // SQLite lock and block the next EPG fetch. - settle(() => { - void this.terminateWorker(worker, 'timed out fetch').then( - () => reject(new Error(errorMessage)) - ); - }); - }; - - scheduleFetchTimeout(); - - worker.on('message', async (message: EpgWorkerMessage) => { - if (settled || generation !== epgSourceGeneration(url)) return; - try { - switch (message.type) { - case 'READY': - scheduleFetchTimeout(); - this.sendProgressToRenderer(url, 'loading', { - totalChannels: 0, - totalPrograms: 0, - }); - worker.postMessage({ - type: 'FETCH_EPG', - url, - options, - }); - break; - - case 'EPG_PROGRESS': - if (message.stats) { - if (hasProgressMoved(message.stats)) { - recordProgress(message.stats); - scheduleFetchTimeout(); - } - this.sendProgressToRenderer( - url, - 'loading', - message.stats - ); - } - break; - - case 'EPG_COMPLETE': - console.log( - this.loggerLabel, - `EPG parsing complete for ${url}:`, - message.stats - ); - this.sendProgressToRenderer( - url, - 'complete', - message.stats - ); - this.fetchedUrls.add(url); - this.workers.delete(url); - settle(() => { - void this.terminateWorker( - worker, - 'completed fetch' - ).then(() => resolve()); - }); - break; - - case 'EPG_ERROR': - console.error( - this.loggerLabel, - 'Worker error:', - message.error - ); - this.sendProgressToRenderer( - url, - 'error', - undefined, - message.error, - undefined, - message.errorCode, - message.errorHost - ); - this.workers.delete(url); - settle(() => { - void this.terminateWorker( - worker, - 'failed fetch' - ).then(() => - reject( - new Error( - message.error || 'Unknown error' - ) - ) - ); - }); - break; - } - } catch (err) { - if (settled || cancelIfRetired()) return; - console.error( - this.loggerLabel, - 'Error handling message:', - err - ); - this.sendProgressToRenderer( - url, - 'error', - undefined, - err instanceof Error ? err.message : String(err) - ); - this.workers.delete(url); - settle(() => { - void this.terminateWorker( - worker, - 'failed message handling' - ).then(() => reject(err)); - }); - } - }); - - worker.on('error', (error) => { - if (settled || cancelIfRetired()) return; - console.error(this.loggerLabel, 'Worker error event:', error); - this.sendProgressToRenderer( - url, - 'error', - undefined, - error.message - ); - this.workers.delete(url); - settle(() => { - void this.terminateWorker(worker, 'errored fetch').then( - () => reject(error) - ); - }); - }); - - worker.on('exit', (code) => { - if (settled || cancelIfRetired(true)) return; - const errorMessage = `Worker exited unexpectedly (code ${code})`; - console.error(this.loggerLabel, `${errorMessage}: ${url}`); - this.sendProgressToRenderer( - url, - 'error', - undefined, - errorMessage - ); - this.workers.delete(url); - settle(() => reject(new Error(errorMessage))); - }); + return runEpgFetch(url, options, { + runtime: this.runtime, + workers: this.workers, + fetchedUrls: this.fetchedUrls, + loggerLabel: this.loggerLabel, + fetchTimeoutMs: this.fetchTimeoutMs, + sendProgressToRenderer: this.sendProgressToRenderer.bind(this), }); } async clearEpgData(): Promise { - return this.runClearWorker({ + return this.runtime.runClearWorker({ timeoutLabel: 'EPG clear', exitLabel: 'Clear worker', readyMessage: { type: 'CLEAR_EPG' }, @@ -408,14 +157,14 @@ export class EpgWorkerService { // to be free. const terminations = [...this.workers.values()].map( (runningWorker) => - this.terminateWorker( + this.runtime.terminateWorker( runningWorker, 'fetch during clear' ) ); this.workers.clear(); terminations.push( - this.terminateWorker(worker, 'completed clear') + this.runtime.terminateWorker(worker, 'completed clear') ); await Promise.all(terminations); }, @@ -449,13 +198,13 @@ export class EpgWorkerService { const runningWorker = this.workers.get(normalizedSourceUrl); if (runningWorker) { this.workers.delete(normalizedSourceUrl); - await this.terminateWorker(runningWorker, 'source clear'); + await this.runtime.terminateWorker(runningWorker, 'source clear'); } // Error/timeout handlers may already have removed the worker from the // lookup, but its fetch promise still owns asynchronous termination. if (inFlight) await inFlight.catch(() => undefined); - return this.runClearWorker({ + return this.runtime.runClearWorker({ timeoutLabel: 'EPG source clear', exitLabel: 'Source clear worker', readyMessage: { @@ -471,148 +220,10 @@ export class EpgWorkerService { `EPG data cleared for source via worker: ${normalizedSourceUrl}` ); this.fetchedUrls.delete(normalizedSourceUrl); - await this.terminateWorker(worker, 'completed source clear'); - }, - }); - } - - private runClearWorker(options: { - timeoutLabel: string; - exitLabel: string; - readyMessage: ClearWorkerMessage; - completeWorkerLabel: string; - failedWorkerLabel: string; - erroredWorkerLabel: string; - onComplete: (worker: Worker) => Promise; - }): Promise { - return new Promise((resolve, reject) => { - let worker: Worker; - try { - worker = this.createEpgWorker(); - } catch (error) { - console.error( - this.loggerLabel, - 'Failed to create worker for clear:', - error + await this.runtime.terminateWorker( + worker, + 'completed source clear' ); - reject(error); - return; - } - - let settled = false; - const settle = (fn: () => void) => { - if (settled) return; - settled = true; - clearTimeout(timeoutId); - fn(); - }; - - const timeoutId = setTimeout(() => { - const errorMessage = `${options.timeoutLabel} timed out after ${ - this.fetchTimeoutMs / 1000 - }s`; - console.error(this.loggerLabel, errorMessage); - settle(() => { - void this.terminateWorker( - worker, - `timed out ${options.timeoutLabel}` - ).then(() => reject(new Error(errorMessage))); - }); - }, this.fetchTimeoutMs); - - worker.on( - 'message', - (message: { type: string; error?: string }) => { - if (message.type === 'READY') { - worker.postMessage(options.readyMessage); - } else if (message.type === 'CLEAR_COMPLETE') { - settle(() => { - void options - .onComplete(worker) - .then(() => resolve(), reject); - }); - } else if (message.type === 'EPG_ERROR') { - console.error( - this.loggerLabel, - 'Worker clear error:', - message.error - ); - settle(() => { - void this.terminateWorker( - worker, - options.failedWorkerLabel - ).then(() => - reject( - new Error(message.error || 'Clear failed') - ) - ); - }); - } - } - ); - - worker.on('error', (error) => { - console.error( - this.loggerLabel, - 'Worker error during clear:', - error - ); - settle(() => { - void this.terminateWorker( - worker, - options.erroredWorkerLabel - ).then(() => reject(error)); - }); - }); - - worker.on('exit', (code) => { - if (settled) return; - const errorMessage = `${options.exitLabel} exited unexpectedly (code ${code})`; - console.error(this.loggerLabel, errorMessage); - settle(() => reject(new Error(errorMessage))); - }); - }); - } - - /** - * Awaits worker shutdown so callers can sequence work (e.g. the next DB - * access) after the thread has really exited. Termination failures are - * logged and swallowed — there is nothing actionable left to do. - */ - private async terminateWorker( - worker: Worker, - context: string - ): Promise { - try { - await worker.terminate(); - } catch (error) { - console.error( - this.loggerLabel, - `Failed to terminate ${context} worker:`, - error - ); - } - } - - private createEpgWorker(): Worker { - const bootstrap = resolveWorkerRuntimeBootstrap({ - isPackaged: app.isPackaged, - workerFilename: 'epg-parser.worker.js', - developmentWorkerDir: path.join(__dirname, 'workers'), - resourcesPath: ( - process as NodeJS.Process & { resourcesPath?: string } - ).resourcesPath, - appPath: app.getAppPath(), - }); - - const workerURL = pathToFileURL(bootstrap.workerPath); - return new Worker(workerURL, { - resourceLimits: { - maxOldGenerationSizeMb: 4096, - maxYoungGenerationSizeMb: 512, - }, - workerData: { - nativeModuleSearchPaths: bootstrap.nativeModuleSearchPaths, }, }); } diff --git a/docs/architecture/m3u-playlist-module.md b/docs/architecture/m3u-playlist-module.md index 9482c4772..c365e2e2d 100644 --- a/docs/architecture/m3u-playlist-module.md +++ b/docs/architecture/m3u-playlist-module.md @@ -1332,7 +1332,9 @@ not global XMLTV owners. No source-discovery or provider matching policy changes Reconciliation finds old sources in XMLTV channel, programme and per-source metadata tables, plus queued imports. It retires their generations before waiting for workers to exit, then uses the -existing source-clear worker. Successfully cleared request candidates are forgotten +`EpgWorkerRuntime` source-clear protocol. The runtime owns worker bootstrap and +shutdown; `EpgWorkerService` owns source generations and serialization, while +`runEpgFetch` owns each import’s message/timeout/exit lifecycle. Successfully cleared request candidates are forgotten without resetting their generation fences; failed cleanups remain retryable. Same-URL clears are serialized and replacement imports await the outstanding clear, so an older cleanup cannot erase a newly re-added source. Source cleanup diff --git a/tools/eslint/max-lines-baseline.mjs b/tools/eslint/max-lines-baseline.mjs index 416cdcbfa..c609d81dc 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/epg-worker.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',