From a76050a491a8c609ba13f2c95f6f0c07602e400b Mon Sep 17 00:00:00 2001 From: 4gray Date: Sun, 6 Sep 2026 00:15:17 +0200 Subject: [PATCH] fix(epg): cancel retired error rows and pending retries --- AGENTS.md | 7 +- CLAUDE.md | 7 +- apps/electron-backend-e2e/src/epg.e2e.ts | 72 ++++++++++++ .../epg-source-settings.service.spec.ts | 47 ++++++++ .../app/events/epg-source-settings.service.ts | 17 ++- docs/architecture/m3u-playlist-module.md | 7 +- .../src/lib/epg-progress.service.spec.ts | 104 +++++++++++++++++- .../src/lib/epg-progress.service.ts | 35 ++++-- 8 files changed, 270 insertions(+), 26 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index b75ef8ea8..fbcfa535b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -180,9 +180,10 @@ Saving Settings → EPG reconciles cached XMLTV with committed global URLs and all enabled M3U playlist sources. Startup runs the same reconciliation after settings load and playlist migration. Ordinary saves skip unchanged normalized source sets; an explicitly edited EPG field can retry a failed cleanup. -Failed settings reads and incomplete -playlist migration never authorize pruning. Removed sources retire queued and -running imports before worker-owned deletion. Shared channel IDs survive while +Failed settings reads and incomplete playlist migration never authorize pruning. +Removed sources retire queued/running imports and dismiss retained error rows +before worker-owned deletion. Retry waits for reconciliation and rechecks its +error row, including after trust-setting writes. Shared channel IDs survive while another source has programmes or per-source channel metadata. The additive `epg_channel_sources` table preserves each imported source's name, logo, URL and timestamp so removal restores a surviving snapshot; ambiguous legacy metadata diff --git a/CLAUDE.md b/CLAUDE.md index 1b88a497b..fc9f1545f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1665,9 +1665,10 @@ Saving Settings → EPG reconciles cached XMLTV with committed global URLs and all enabled M3U playlist sources. Startup runs the same reconciliation after settings load and playlist migration. Ordinary saves skip unchanged normalized source sets; an explicitly edited EPG field can retry a failed cleanup. -Failed settings reads and incomplete -playlist migration never authorize pruning. Removed sources retire queued and -running imports before worker-owned deletion. Shared channel IDs survive while +Failed settings reads and incomplete playlist migration never authorize pruning. +Removed sources retire queued/running imports and dismiss retained error rows +before worker-owned deletion. Retry waits for reconciliation and rechecks its +error row, including after trust-setting writes. Shared channel IDs survive while another source has programmes or per-source channel metadata. The additive `epg_channel_sources` table preserves each imported source's name, logo, URL and timestamp so removal restores a surviving snapshot; ambiguous legacy metadata diff --git a/apps/electron-backend-e2e/src/epg.e2e.ts b/apps/electron-backend-e2e/src/epg.e2e.ts index fa354160a..3cf4eaa4f 100644 --- a/apps/electron-backend-e2e/src/epg.e2e.ts +++ b/apps/electron-backend-e2e/src/epg.e2e.ts @@ -166,6 +166,78 @@ test.describe('Electron EPG', () => { } }); + test('@epg @electron removes a retained error row after its import worker has finished', async ({ + dataDir, + }) => { + const server = await createMutableTextServer(epgFixtureXml, { + contentType: 'application/xml', + }); + const app = await launchElectronApp(dataDir); + try { + await openSettings(app.mainWindow); + await openSettingsSection(app.mainWindow, 'epg'); + await app.mainWindow + .getByRole('button', { name: 'Add EPG source' }) + .click(); + await app.mainWindow + .locator('.epg-source-row input') + .fill(server.resourceUrl); + await saveSettings(app.mainWindow); + await expect( + app.mainWindow.locator( + '.epg-progress-panel .import-item.status-complete' + ) + ).toHaveCount(1, { timeout: 30000 }); + // Capture the actual generation; a force-fetch also awaits termination. + const generation = await app.mainWindow.evaluate(async (url) => { + await window.electron.clearEpgDataForSource(url); + const completed = new Promise((resolve) => { + window.electron.onEpgProgress?.((progress) => { + if ( + progress.url === url && + progress.status === 'complete' + ) + resolve(progress.generation ?? 0); + }); + }); + await window.electron.forceFetchEpg(url); + return completed; + }, server.resourceUrl); + await app.electronApp.evaluate( + ({ BrowserWindow }, { url, generation }) => { + BrowserWindow.getAllWindows()[0].webContents.send( + 'EPG_PROGRESS_UPDATE', + { + url, + status: 'error', + generation, + errorCode: 'invalid-tls-certificate', + error: 'Synthetic retained TLS error', + } + ); + }, + { url: server.resourceUrl, generation } + ); + await expect( + app.mainWindow.locator( + '.epg-progress-panel .import-item.status-error' + ) + ).toHaveCount(1); + await app.mainWindow + .locator('.epg-source-row button') + .nth(1) + .click(); + await saveSettings(app.mainWindow); + await expect( + app.mainWindow.locator('.epg-progress-panel .import-item') + ).toHaveCount(0); + await expect.poll(() => getEpgChannelCount(app.mainWindow)).toBe(0); + } finally { + await closeElectronApp(app); + await server.close(); + } + }); + test('@epg @electron removes only the saved source, preserves shared IDs and mappings, and repairs old orphan data on restart', async ({ dataDir, }) => { diff --git a/apps/electron-backend/src/app/events/epg-source-settings.service.spec.ts b/apps/electron-backend/src/app/events/epg-source-settings.service.spec.ts index c38e55825..23349ef07 100644 --- a/apps/electron-backend/src/app/events/epg-source-settings.service.spec.ts +++ b/apps/electron-backend/src/app/events/epg-source-settings.service.spec.ts @@ -14,6 +14,7 @@ jest.mock('../database/connection', () => ({ getDatabase: jest.fn() })); jest.mock('./epg-worker.service', () => ({ epgWorkerService: { clearEpgDataForSource: jest.fn().mockResolvedValue(undefined), + sendProgressToRenderer: jest.fn(), }, })); @@ -82,6 +83,52 @@ describe('committed EPG source reconciliation', () => { ); }); + it('cancels completed error rows even without cached data or a running worker', async () => { + rows.set(epgChannels, []); + rows.set(epgPrograms, []); + const generation = requestEpgSource('failed-import'); + await reconcileEpgSources([]); + expect(epgWorkerService.sendProgressToRenderer).toHaveBeenCalledWith( + 'failed-import', + 'cancelled', + undefined, + undefined, + undefined, + undefined, + undefined, + generation + ); + }); + + it('cancels every retired row before the first source clear can fail', async () => { + const generation = epgSourceGeneration('second'); + ( + epgWorkerService.clearEpgDataForSource as jest.Mock + ).mockImplementationOnce(async () => { + expect( + epgWorkerService.sendProgressToRenderer + ).toHaveBeenCalledWith( + 'second', + 'cancelled', + undefined, + undefined, + undefined, + undefined, + undefined, + generation + ); + throw new Error('clear failed'); + }); + await expect(reconcileEpgSources(['shared'])).rejects.toThrow( + 'clear failed' + ); + expect( + ( + epgWorkerService.sendProgressToRenderer as jest.Mock + ).mock.calls.map(([url]) => url) + ).not.toContain('shared'); + }); + it('does not clear historical request keys again after successful cleanup', async () => { requestEpgSource('removed'); await reconcileEpgSources([]); diff --git a/apps/electron-backend/src/app/events/epg-source-settings.service.ts b/apps/electron-backend/src/app/events/epg-source-settings.service.ts index bb0544d96..9e52ddb9e 100644 --- a/apps/electron-backend/src/app/events/epg-source-settings.service.ts +++ b/apps/electron-backend/src/app/events/epg-source-settings.service.ts @@ -9,6 +9,7 @@ import { } from '../database/schema'; import { epgWorkerService } from './epg-worker.service'; import { + epgSourceGeneration, forgetEpgSourceRequest, requestedEpgSources, retireEpgSource, @@ -76,7 +77,21 @@ export function reconcileEpgSources(globalUrls: string[]): Promise { ), ]; // Fence the whole obsolete set before awaiting the first worker exit. - removed.forEach(retireEpgSource); + removed.forEach((url) => { + const generation = epgSourceGeneration(url); + retireEpgSource(url); + // Finished imports can leave actionable error rows with Retry. + epgWorkerService.sendProgressToRenderer( + url, + 'cancelled', + undefined, + undefined, + undefined, + undefined, + undefined, + generation + ); + }); for (const url of removed) { await epgWorkerService.clearEpgDataForSource(url); forgetEpgSourceRequest(url, requested.get(url.trim())); diff --git a/docs/architecture/m3u-playlist-module.md b/docs/architecture/m3u-playlist-module.md index 8ec33541e..e5f6259c7 100644 --- a/docs/architecture/m3u-playlist-module.md +++ b/docs/architecture/m3u-playlist-module.md @@ -1345,8 +1345,11 @@ 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 also awaits the in-flight fetch promise when an error/timeout has already removed the worker from the lookup but its termination is still pending. -Retired queued and running imports emit generation-scoped cancellation so progress -rows disappear without reporting routine worker termination as an import failure. Programmes are deleted by source; a globally keyed +Every removed source emits generation-scoped cancellation before cleanup starts, +including retained actionable errors whose workers have already finished. Retired +queued/running imports also cancel their own generation. Retry waits for source +reconciliation and rechecks the same error row; trust-setting continuations and +old dismissal timers cannot affect a removed or replacement row. Progress rows disappear without reporting routine worker termination as an import failure. Programmes are deleted by source; a globally keyed channel is retained while another source still has programmes or channel metadata. The additive `epg_channel_sources` table is created during database initialization and records each imported source's name, logo, URL and timestamp. Before removing diff --git a/libs/epg/data-access/src/lib/epg-progress.service.spec.ts b/libs/epg/data-access/src/lib/epg-progress.service.spec.ts index 86915668c..e48f1a98a 100644 --- a/libs/epg/data-access/src/lib/epg-progress.service.spec.ts +++ b/libs/epg/data-access/src/lib/epg-progress.service.spec.ts @@ -3,11 +3,13 @@ import { EpgImportProgress, EpgRuntimeBridgeService, } from './epg-runtime-bridge.service'; -import { SettingsStore } from '@iptvnator/services'; +import { SettingsStore, EpgSourceSettingsService } from '@iptvnator/services'; +import { ELECTRON_BRIDGE_SECURITY_ERROR_CODES } from '@iptvnator/shared/interfaces'; import { EpgProgressService } from './epg-progress.service'; describe('EpgProgressService', () => { let epgBridge: Partial; + let sources: { waitForReconciliation: jest.Mock }; let settingsStore: { getSettings: jest.Mock; getTrustOptions: jest.Mock; @@ -15,11 +17,14 @@ describe('EpgProgressService', () => { }; beforeEach(() => { + sources = { + waitForReconciliation: jest.fn().mockResolvedValue(undefined), + }; epgBridge = { forceFetchEpg: jest.fn().mockResolvedValue({ success: true }), onProgress: jest.fn(), supportsDataManagement: false, - supportsProgress: false, + supportsProgress: true, }; settingsStore = { getSettings: jest.fn(() => ({ @@ -35,6 +40,7 @@ describe('EpgProgressService', () => { }); afterEach(() => { + jest.useRealTimers(); TestBed.resetTestingModule(); jest.restoreAllMocks(); }); @@ -43,6 +49,7 @@ describe('EpgProgressService', () => { TestBed.configureTestingModule({ providers: [ EpgProgressService, + { provide: EpgSourceSettingsService, useValue: sources }, { provide: EpgRuntimeBridgeService, useValue: epgBridge, @@ -57,7 +64,89 @@ describe('EpgProgressService', () => { return TestBed.inject(EpgProgressService); } + const url = 'https://example.com/epg.xml'; + const emit = (progress: EpgImportProgress) => + (epgBridge.onProgress as jest.Mock).mock.calls[0][0](progress); + const errorRow = (source = url) => + emit({ + url: source, + status: 'error', + generation: 0, + errorCode: + ELECTRON_BRIDGE_SECURITY_ERROR_CODES.InvalidTlsCertificate, + }); + + it('removes retained actionable errors when their source is cancelled', async () => { + epgBridge.supportsDataManagement = true; + const service = configureService(); + errorRow(); + emit({ url, status: 'cancelled', generation: 0 }); + expect(service.isVisible()).toBe(false); + await service.retry(url); + expect(epgBridge.forceFetchEpg).not.toHaveBeenCalled(); + }); + + it.each([true, false])( + 'waits for source reconciliation before retrying (removed=%s)', + async (removed) => { + epgBridge.supportsDataManagement = true; + let finish!: () => void; + sources.waitForReconciliation.mockReturnValue( + new Promise((resolve) => { + finish = resolve; + }) + ); + const service = configureService(); + errorRow(); + const retry = service.retry(url); + expect(epgBridge.forceFetchEpg).not.toHaveBeenCalled(); + if (removed) emit({ url, status: 'cancelled', generation: 0 }); + finish(); + await retry; + expect(epgBridge.forceFetchEpg).toHaveBeenCalledTimes( + removed ? 0 : 1 + ); + } + ); + + it.each(['private', 'tls'])( + 'does not retry a removed/replaced row after a pending %s trust write', + async (kind) => { + epgBridge.supportsDataManagement = true; + let finish!: () => void; + settingsStore.updateSettings.mockReturnValue( + new Promise((resolve) => { + finish = resolve; + }) + ); + const service = configureService(); + errorRow(); + const retry = + kind === 'private' + ? service.trustPrivateNetworkSourceAndRetry(url) + : service.trustInsecureTlsHostAndRetry(url); + emit({ url, status: 'cancelled', generation: 0 }); + emit({ url, status: 'loading', generation: 2 }); + finish(); + await retry; + expect(epgBridge.forceFetchEpg).not.toHaveBeenCalled(); + expect(service.activeCount()).toBe(1); + } + ); + + it('does not let an old dismissal timer remove a replacement import', () => { + jest.useFakeTimers(); + const service = configureService(); + emit({ url, status: 'complete', generation: 0 }); + emit({ url, status: 'cancelled', generation: 0 }); + emit({ url, status: 'loading', generation: 2 }); + jest.advanceTimersByTime(5000); + expect(service.activeCount()).toBe(1); + jest.useRealTimers(); + }); + it('does not subscribe to progress events when EPG progress support is disabled', () => { + epgBridge.supportsProgress = false; configureService(); expect(epgBridge.onProgress).not.toHaveBeenCalled(); @@ -86,19 +175,21 @@ describe('EpgProgressService', () => { expect(service.activeCount()).toBe(1); }); - it('does not force retry when EPG data management is disabled', () => { + it('does not force retry when EPG data management is disabled', async () => { const service = configureService(); - service.retry('https://example.com/epg.xml'); + errorRow(); + await service.retry('https://example.com/epg.xml'); expect(epgBridge.forceFetchEpg).not.toHaveBeenCalled(); }); - it('forces retry through the EPG runtime bridge when data management is enabled', () => { + it('forces retry through the EPG runtime bridge when data management is enabled', async () => { epgBridge.supportsDataManagement = true; const service = configureService(); - service.retry('https://example.com/epg.xml'); + errorRow(); + await service.retry('https://example.com/epg.xml'); expect(epgBridge.forceFetchEpg).toHaveBeenCalledWith( 'https://example.com/epg.xml', @@ -113,6 +204,7 @@ describe('EpgProgressService', () => { epgBridge.supportsDataManagement = true; const service = configureService(); + errorRow('http://192.168.1.30/guide.xml'); await service.trustPrivateNetworkSourceAndRetry( 'http://192.168.1.30/guide.xml' ); diff --git a/libs/epg/data-access/src/lib/epg-progress.service.ts b/libs/epg/data-access/src/lib/epg-progress.service.ts index eb71abe5c..73ca330ab 100644 --- a/libs/epg/data-access/src/lib/epg-progress.service.ts +++ b/libs/epg/data-access/src/lib/epg-progress.service.ts @@ -3,7 +3,7 @@ import { ELECTRON_BRIDGE_SECURITY_ERROR_CODES, normalizeHost, } from '@iptvnator/shared/interfaces'; -import { SettingsStore } from '@iptvnator/services'; +import { SettingsStore, EpgSourceSettingsService } from '@iptvnator/services'; import { EpgImportProgress, EpgRuntimeBridgeService, @@ -13,6 +13,7 @@ import { export class EpgProgressService { private readonly epgBridge = inject(EpgRuntimeBridgeService); private readonly settingsStore = inject(SettingsStore); + private readonly sources = inject(EpgSourceSettingsService); private readonly importsMap = signal>( new Map() ); @@ -45,20 +46,27 @@ export class EpgProgressService { this.importsMap.set(new Map()); } - retry(url: string): void { - // Clear the errored row so the backend's subsequent 'queued' event - // reappears cleanly rather than updating an existing error row. - this.removeImport(url); - if (!this.epgBridge.supportsDataManagement) { + async retry(url: string): Promise { + const progress = this.importsMap().get(url); + if ( + progress?.status !== 'error' || + !this.epgBridge.supportsDataManagement + ) return; - } - void this.epgBridge.forceFetchEpg( + // A settings save may already be retiring this row. Never bypass its + // ownership reconciliation by starting a new force-fetch generation. + await this.sources.waitForReconciliation(); + if (this.importsMap().get(url) !== progress) return; + this.removeImport(url); + await this.epgBridge.forceFetchEpg( url, this.settingsStore.getTrustOptions() ); } async trustPrivateNetworkSourceAndRetry(url: string): Promise { + const progress = this.importsMap().get(url); + if (progress?.status !== 'error') return; const settings = this.settingsStore.getSettings(); const trustedUrls = new Set( settings.trustedPrivateNetworkEpgUrls ?? [] @@ -68,13 +76,15 @@ export class EpgProgressService { await this.settingsStore.updateSettings({ trustedPrivateNetworkEpgUrls: Array.from(trustedUrls), }); - this.retry(url); + if (this.importsMap().get(url) === progress) await this.retry(url); } async trustInsecureTlsHostAndRetry( url: string, host?: string ): Promise { + const progress = this.importsMap().get(url); + if (progress?.status !== 'error') return; const trustedHost = host ?? this.getHostname(url); if (!trustedHost) { return; @@ -91,7 +101,7 @@ export class EpgProgressService { await this.settingsStore.updateSettings({ trustedInsecureTlsHosts: Array.from(trustedHosts), }); - this.retry(url); + if (this.importsMap().get(url) === progress) await this.retry(url); } private initializeListener(): void { @@ -124,7 +134,10 @@ export class EpgProgressService { progress.status === 'complete' || (progress.status === 'error' && !this.isActionableError(progress)) ) { - setTimeout(() => this.removeImport(progress.url), 5000); + setTimeout(() => { + if (this.importsMap().get(progress.url) === progress) + this.removeImport(progress.url); + }, 5000); } }