fix(downloads): resume interrupted Xtream VOD transfers (#1329)

* fix(downloads): resume interrupted Xtream VOD transfers

* fix(downloads): validate partials before resuming

* fix(downloads): propagate headers to episode transfers
This commit is contained in:
4gray authored and GitHub committed 2026-08-01 22:01:10 +02:00
1 parent d44045de31
commit aba89d64cf
29 files changed
+894 -92

No files matched your search

@@ -0,0 +1,117 @@
import { mkdirSync, readFileSync, statSync } from 'fs';
import { join } from 'path';
import type { Page } from '@playwright/test';
import {
addXtreamPortal,
closeElectronApp,
expect,
launchElectronApp,
resetMockServers,
test,
waitForXtreamWorkspaceReady,
} from './electron-test-fixtures';
import {
createInterruptedRangeServer,
INTERRUPTED_RANGE_SERVER_ETAG,
startDownload,
} from './downloads.e2e-support';
async function openDownloadsPage(page: Page): Promise<void> {
await page.getByRole('button', { name: 'Open downloads' }).click();
await page.waitForURL(/\/workspace\/downloads(?:\?.*)?$/);
}
async function getPlaylistId(page: Page, title: string): Promise<string> {
const playlists = await page.evaluate(
async () => (await window.electron?.dbGetAppPlaylistMetas?.()) ?? []
);
const playlist = playlists.find((entry) => entry.title === title);
expect(playlist, `playlist "${title}" should exist`).toBeDefined();
return playlist?._id ?? '';
}
test.describe('Electron download reliability', () => {
test('@downloads @electron retains a network-interrupted partial and retries it with HTTP Range', async ({
dataDir,
request,
}) => {
await resetMockServers(request, ['xtream']);
const rangeServer = await createInterruptedRangeServer();
const app = await launchElectronApp(dataDir);
try {
await addXtreamPortal(app.mainWindow, {
name: 'Reset Portal',
username: 'user1',
password: 'pass1',
});
await waitForXtreamWorkspaceReady(app.mainWindow);
await openDownloadsPage(app.mainWindow);
const downloadsDir = join(dataDir, 'e2e-reset-downloads');
mkdirSync(downloadsDir, { recursive: true });
await app.electronApp.evaluate(({ dialog }, folder) => {
dialog.showOpenDialog = async () =>
({
canceled: false,
filePaths: [folder],
}) as Awaited<ReturnType<typeof dialog.showOpenDialog>>;
}, downloadsDir);
await app.mainWindow
.getByRole('button', { name: 'Change Folder' })
.click();
const playlistId = await getPlaylistId(
app.mainWindow,
'Reset Portal'
);
const downloadId = await startDownload(app.mainWindow, {
playlistId,
xtreamId: 9801,
contentType: 'vod',
title: 'E2E Reset Movie',
url: rangeServer.url,
downloadFolder: downloadsDir,
});
const item = app.mainWindow.getByTestId(
`download-queue-item-${downloadId}`
);
await expect(item.locator('.download-queue__status')).toContainText(
'Failed',
{ timeout: 30000 }
);
await expect(item.locator('.download-queue__error')).toContainText(
'DOWNLOAD_NETWORK_INTERRUPTED (ECONNRESET)'
);
const partialPath = join(downloadsDir, 'E2E Reset Movie.mp4.part');
expect(statSync(partialPath).size).toBe(
rangeServer.interruptedBytes
);
await item
.getByRole('button', { name: 'Retry E2E Reset Movie' })
.click();
await expect(
app.mainWindow.getByTestId(
`download-library-movie-${downloadId}`
)
).toBeVisible({ timeout: 30000 });
const resumeRequest = rangeServer.requests.find(
(entry) => entry.range
);
expect(resumeRequest?.range).toBe(
`bytes=${rangeServer.interruptedBytes}-`
);
expect(resumeRequest?.ifRange).toBe(INTERRUPTED_RANGE_SERVER_ETAG);
const finalFile = readFileSync(
join(downloadsDir, 'E2E Reset Movie.mp4')
);
expect(finalFile.equals(rangeServer.payload)).toBe(true);
} finally {
await closeElectronApp(app);
await rangeServer.close();
}
});
});
@@ -31,6 +31,10 @@ interface ThrottledRangeServer {
url: string;
}
interface InterruptedRangeServer extends ThrottledRangeServer {
interruptedBytes: number;
}
interface TruncatedDownloadServer {
close: () => Promise<void>;
payload: Buffer;
@@ -39,6 +43,7 @@ interface TruncatedDownloadServer {
}
export const RANGE_SERVER_ETAG = '"e2e-range-etag"';
export const INTERRUPTED_RANGE_SERVER_ETAG = '"e2e-reset-etag"';
const downloadPlayCaptureKey = '__iptvnatorE2eDownloadPlayPaths';
export function getStalkerSeriesDownloadTarget(
@@ -172,6 +177,63 @@ export async function createThrottledRangeServer(
};
}
/**
* Resets the first full response after writing a valid prefix, then serves the
* remainder to a Range retry. This matches a provider/proxy connection drop
* without manufacturing a clean EOF.
*/
export async function createInterruptedRangeServer(): Promise<InterruptedRangeServer> {
const payload = Buffer.alloc(64 * 1024, 9);
const interruptedBytes = 16 * 1024;
const requests: RangeServerRequest[] = [];
const server = createServer((req, res) => {
const range = req.headers.range;
const ifRange = req.headers['if-range'];
requests.push({
ifRange: typeof ifRange === 'string' ? ifRange : undefined,
range: typeof range === 'string' ? range : undefined,
});
const offset = range
? Number(/^bytes=(\d+)-$/.exec(range)?.[1] ?? Number.NaN)
: 0;
if (range && Number.isFinite(offset)) {
res.writeHead(206, {
'Content-Length': payload.length - offset,
'Content-Range': `bytes ${offset}-${payload.length - 1}/${payload.length}`,
'Content-Type': 'video/mp4',
ETag: INTERRUPTED_RANGE_SERVER_ETAG,
});
res.end(payload.subarray(offset));
return;
}
res.writeHead(200, {
'Content-Length': payload.length,
'Content-Type': 'video/mp4',
ETag: INTERRUPTED_RANGE_SERVER_ETAG,
});
res.write(payload.subarray(0, interruptedBytes), () => {
setTimeout(() => res.socket?.destroy(), 20);
});
});
await new Promise<void>((resolve) =>
server.listen(0, '127.0.0.1', resolve)
);
const { port } = server.address() as AddressInfo;
return {
close: () =>
new Promise<void>((resolve) => server.close(() => resolve())),
interruptedBytes,
payload,
requests,
url: `http://127.0.0.1:${port}/media/e2e-reset-movie.mp4`,
};
}
/**
* Ends a chunked response cleanly while Content-Range advertises a larger
* representation. The runtime therefore retains the valid .part for a Range
@@ -14,12 +14,16 @@ import type {
DownloadTask,
TransferProgress,
} from './download-task';
import { describeError, TruncatedTransferError } from './download-transfer';
import {
describeError,
InterruptedTransferError,
TruncatedTransferError,
} from './download-transfer';
/**
* Persistence for a failed startDownload() attempt, after its cancel/pause
* checkpoints have been ruled out. Chooses between: retaining a truncated
* transfer for a Range retry, committing an already-finalized file,
* checkpoints have been ruled out. Chooses between: retaining a recoverable
* partial for a Range retry, committing an already-finalized file,
* retaining a completed partial, or the generic delete-partial failure.
*/
export async function handleDownloadFailure(
@@ -28,8 +32,12 @@ export async function handleDownloadFailure(
reservation: ReservedPartialDownloadFile | undefined,
error: unknown
): Promise<void> {
if (error instanceof TruncatedTransferError && reservation) {
// The short response is retained so a retry can continue the
if (
(error instanceof TruncatedTransferError ||
error instanceof InterruptedTransferError) &&
reservation
) {
// The recoverable partial is retained so a retry can continue the
// transfer via Range instead of starting over.
await persistCompletedPartialFailure(
db,
@@ -338,7 +346,9 @@ function canCopyCompletedPartialAfterLinkFailure(error: unknown): boolean {
}
/** @returns false when a .part exists but could not be deleted. */
export function removePartialFile(filePath: string | null | undefined): boolean {
export function removePartialFile(
filePath: string | null | undefined
): boolean {
try {
removePartialDownloadFile(filePath);
return true;
@@ -3,6 +3,7 @@ interface SetupOptions {
cleanupError?: Error;
claim?: { changes: number };
fileReappeared?: boolean;
playlistType?: 'xtream' | 'stalker';
row?: Record<string, unknown> | null;
urlError?: Error;
}
@@ -25,13 +26,18 @@ async function setup(options: SetupOptions = {}) {
url: 'https://example.test/movie.mp4',
...options.row,
};
const limit = jest.fn().mockResolvedValue(row ? [row] : []);
const downloadLimit = jest.fn().mockResolvedValue(row ? [row] : []);
const playlistLimit = jest
.fn()
.mockResolvedValue([{ type: options.playlistType ?? 'stalker' }]);
const where = jest.fn().mockResolvedValue(options.claim ?? { changes: 1 });
const set = jest.fn(() => ({ where }));
const db = {
select: jest.fn(() => ({
select: jest.fn((selection?: unknown) => ({
from: jest.fn(() => ({
where: jest.fn(() => ({ limit })),
where: jest.fn(() => ({
limit: selection ? playlistLimit : downloadLimit,
})),
})),
})),
update: jest.fn(() => ({ set })),
@@ -118,6 +124,27 @@ describe('redownload missing completed file', () => {
});
});
it('adds the Xtream fallback User-Agent to a legacy row', async () => {
const harness = await setup({
playlistType: 'xtream',
row: {
playlistId: 'playlist-1',
requestHeaders: null,
},
});
await expect(harness.redownloadMissingRequest(42)).resolves.toEqual({
success: true,
});
expect(harness.enqueueDownload).toHaveBeenCalledWith(
expect.objectContaining({
headers: {
'User-Agent': 'VLC/3.0.18 LibVLC/3.0.18',
},
})
);
});
it('recovers a file that reappeared without updating or enqueueing', async () => {
const harness = await setup({ fileReappeared: true });
@@ -6,7 +6,7 @@ import * as schema from '../../database/schema';
import { assertRemoteUrlAllowed } from '../url-safety';
import { isAvailableDownloadFile } from './download-file-availability';
import { removePartialDownloadFile } from './download-file-path';
import { parseStoredHeaders } from './download-requests';
import { resolveStoredDownloadHeaders } from './download-request-headers';
import { enqueueDownload } from './download-runtime';
export interface RedownloadMissingResult {
@@ -61,6 +61,7 @@ export async function redownloadMissingRequest(
}
await assertRemoteUrlAllowed(item.url, { allowPrivateNetworks: true });
const headers = await resolveStoredDownloadHeaders(db, item);
try {
removePartialDownloadFile(item.filePath);
@@ -102,7 +103,7 @@ export async function redownloadMissingRequest(
directory: dirname(item.filePath),
fileName: basename(item.filePath),
filePath: item.filePath,
headers: parseStoredHeaders(item.requestHeaders),
headers,
id: item.id,
resumeValidator: null,
totalBytes: null,
@@ -0,0 +1,80 @@
import type { DownloadsDatabase } from './download-task';
function createDatabase(playlistType?: 'xtream' | 'stalker') {
const limit = jest
.fn()
.mockResolvedValue(playlistType ? [{ type: playlistType }] : []);
const db = {
select: jest.fn(() => ({
from: jest.fn(() => ({
where: jest.fn(() => ({ limit })),
})),
})),
} as unknown as DownloadsDatabase;
return { db, select: db.select };
}
describe('stored download request headers', () => {
it('adds the provider-compatible fallback for a legacy Xtream row', async () => {
const { db } = createDatabase('xtream');
const { resolveStoredDownloadHeaders } =
await import('./download-request-headers');
await expect(
resolveStoredDownloadHeaders(db, {
playlistId: 'playlist-1',
requestHeaders: JSON.stringify({
Authorization: 'not-allowed',
Referer: 'https://example.test/',
}),
})
).resolves.toEqual({
Referer: 'https://example.test/',
'User-Agent': 'VLC/3.0.18 LibVLC/3.0.18',
});
});
it('does not add the Xtream fallback to a Stalker row', async () => {
const { db } = createDatabase('stalker');
const { resolveStoredDownloadHeaders } =
await import('./download-request-headers');
await expect(
resolveStoredDownloadHeaders(db, {
playlistId: 'playlist-1',
requestHeaders: null,
})
).resolves.toBeUndefined();
});
it('adds the player fallback when a legacy row outlives its deleted source', async () => {
const { db } = createDatabase();
const { resolveStoredDownloadHeaders } =
await import('./download-request-headers');
await expect(
resolveStoredDownloadHeaders(db, {
playlistId: 'deleted-playlist',
requestHeaders: null,
})
).resolves.toEqual({
'User-Agent': 'VLC/3.0.18 LibVLC/3.0.18',
});
});
it('preserves an explicit stored User-Agent without querying the playlist', async () => {
const { db, select } = createDatabase('xtream');
const { resolveStoredDownloadHeaders } =
await import('./download-request-headers');
await expect(
resolveStoredDownloadHeaders(db, {
playlistId: 'playlist-1',
requestHeaders: JSON.stringify({
'User-Agent': 'Custom/1.0',
}),
})
).resolves.toEqual({ 'User-Agent': 'Custom/1.0' });
expect(select).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,66 @@
import { XTREAM_CLIENT_USER_AGENT } from '@iptvnator/shared/interfaces';
import { eq } from 'drizzle-orm';
import * as schema from '../../database/schema';
import type { DownloadsDatabase } from './download-task';
const STORED_HEADER_ALLOWLIST = ['User-Agent', 'Origin', 'Referer'] as const;
export function parseStoredHeaders(
value: string | null | undefined
): Record<string, string> | undefined {
if (!value) {
return undefined;
}
try {
const parsed = JSON.parse(value) as unknown;
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
return undefined;
}
// Re-apply the write-time allowlist so a tampered or imported
// database row cannot smuggle arbitrary headers into requests.
const entries = parsed as Record<string, unknown>;
const headers = STORED_HEADER_ALLOWLIST.reduce<Record<string, string>>(
(acc, key) => {
const headerValue = entries[key];
if (typeof headerValue === 'string') {
acc[key] = headerValue;
}
return acc;
},
{}
);
return Object.keys(headers).length > 0 ? headers : undefined;
} catch {
return undefined;
}
}
export async function resolveStoredDownloadHeaders(
db: DownloadsDatabase,
item: { playlistId: string; requestHeaders: string | null | undefined }
): Promise<Record<string, string> | undefined> {
const headers = parseStoredHeaders(item.requestHeaders);
if (headers?.['User-Agent']?.trim()) {
return headers;
}
const playlists = await db
.select({ type: schema.playlists.type })
.from(schema.playlists)
.where(eq(schema.playlists.id, item.playlistId))
.limit(1);
const playlistType = playlists[0]?.type;
if (playlistType !== undefined && playlistType !== 'xtream') {
return headers;
}
// Download rows intentionally survive individual playlist deletion. Older
// rows have no stored provider marker, so a missing source must use the
// IPTV-player fallback as the only recoverable identity-compatible default.
return {
...headers,
'User-Agent': XTREAM_CLIENT_USER_AGENT,
};
}
@@ -384,6 +384,66 @@ describe('download request metadata snapshots', () => {
});
describe('download requests resume', () => {
it('adds the Xtream fallback User-Agent to a legacy paused row', async () => {
jest.resetModules();
const schema = await import('../../database/schema');
const row = {
filePath: '/downloads/movie.mp4',
id: 42,
playlistId: 'playlist-1',
requestHeaders: null,
resumeValidator: '"etag-9"',
status: 'paused',
title: 'Movie',
totalBytes: 100,
url: 'https://example.test/movie.mp4',
};
const downloadLimit = jest.fn().mockResolvedValue([row]);
const playlistLimit = jest.fn().mockResolvedValue([{ type: 'xtream' }]);
const db = {
select: jest.fn(() => ({
from: jest.fn((table: unknown) => ({
where: jest.fn(() => ({
limit:
table === schema.playlists
? playlistLimit
: downloadLimit,
})),
})),
})),
update: jest.fn(() => ({
set: jest.fn(() => ({
where: jest.fn().mockResolvedValue({ changes: 1 }),
})),
})),
};
const enqueueDownload = jest.fn();
const authorizer = {
requireAuthorized: jest.fn(async (directory: string) => directory),
} as unknown as DownloadDirectoryAuthorizer;
jest.doMock('../../database/connection', () => ({
getDatabase: jest.fn().mockResolvedValue(db),
}));
jest.doMock('../url-safety', () => ({
assertRemoteUrlAllowed: jest.fn().mockResolvedValue(undefined),
}));
jest.doMock('./download-runtime', () => ({ enqueueDownload }));
const { resumeDownloadRequest } = await import('./download-requests');
await expect(
resumeDownloadRequest(42, '/unused', authorizer)
).resolves.toEqual({ success: true });
expect(enqueueDownload).toHaveBeenCalledWith(
expect.objectContaining({
headers: {
'User-Agent': 'VLC/3.0.18 LibVLC/3.0.18',
},
})
);
});
it('enqueues a paused download with stored headers and original target path', async () => {
jest.resetModules();
@@ -6,6 +6,7 @@ import * as schema from '../../database/schema';
import { assertRemoteUrlAllowed } from '../url-safety';
import { DownloadDirectoryAuthorizer } from './download-directory-authorization';
import { removePartialDownloadFile } from './download-file-path';
import { resolveStoredDownloadHeaders } from './download-request-headers';
import {
assertDownloadMetadataArtworkDiffersFromStream,
assertDownloadMetadataMatchesContentType,
@@ -80,40 +81,6 @@ function serializeHeaders(
return headers ? JSON.stringify(headers) : null;
}
const STORED_HEADER_ALLOWLIST = ['User-Agent', 'Origin', 'Referer'] as const;
export function parseStoredHeaders(
value: string | null
): Record<string, string> | undefined {
if (!value) {
return undefined;
}
try {
const parsed = JSON.parse(value) as unknown;
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
return undefined;
}
// Re-apply the write-time allowlist so a tampered or imported
// database row cannot smuggle arbitrary headers into requests.
const entries = parsed as Record<string, unknown>;
const headers = STORED_HEADER_ALLOWLIST.reduce<Record<string, string>>(
(acc, key) => {
const headerValue = entries[key];
if (typeof headerValue === 'string') {
acc[key] = headerValue;
}
return acc;
},
{}
);
return Object.keys(headers).length > 0 ? headers : undefined;
} catch {
return undefined;
}
}
export async function startDownloadRequest(
data: StartDownloadRequest,
authorizer: DownloadDirectoryAuthorizer
@@ -320,6 +287,7 @@ export async function retryDownloadRequest(
const fileName = retainedFilePath
? basename(retainedFilePath)
: createFileName(item.title, item.url);
const headers = await resolveStoredDownloadHeaders(db, item);
const queuedUpdate = retainedFilePath
? {
errorMessage: null,
@@ -345,7 +313,7 @@ export async function retryDownloadRequest(
directory,
fileName,
filePath: retainedFilePath,
headers: parseStoredHeaders(item.requestHeaders),
headers,
id: item.id,
resumeValidator: retainedFilePath ? item.resumeValidator : null,
totalBytes: retainedFilePath ? item.totalBytes : null,
@@ -388,6 +356,7 @@ export async function resumeDownloadRequest(
const fileName = item.filePath
? basename(item.filePath)
: createFileName(item.title, item.url);
const headers = await resolveStoredDownloadHeaders(db, item);
// Claim the row atomically: a concurrent resume for the same id loses
// this conditional update and must not enqueue a second task.
@@ -416,7 +385,7 @@ export async function resumeDownloadRequest(
directory,
fileName,
filePath: item.filePath,
headers: parseStoredHeaders(item.requestHeaders),
headers,
id: item.id,
resumeValidator: item.resumeValidator,
totalBytes: item.totalBytes,
@@ -62,6 +62,7 @@ describe('destination collision handling', () => {
runtime.enqueueDownload({
...createTask(),
filePath: '/downloads/movie.mp4',
resumeValidator: '"etag-1"',
totalBytes: 100,
});
await waitForStatus(set, 'completed');
@@ -11,6 +11,7 @@ interface ResumeHarness {
interface ResumeHarnessOptions {
partialSize: number;
partialSizeAfterTransferError?: number;
response: {
data: Readable;
headers: Record<string, string>;
@@ -82,7 +83,12 @@ async function setupResumeHarness(
}));
jest.doMock('./download-file-path', () => ({
getPartialDownloadPath: (filePath: string) => `${filePath}.part`,
getPartialDownloadSize: jest.fn(() => options.partialSize),
getPartialDownloadSize: jest
.fn()
.mockReturnValueOnce(options.partialSize)
.mockReturnValue(
options.partialSizeAfterTransferError ?? options.partialSize
),
removePartialDownloadFile,
reserveAvailablePartialDownloadFile: jest.fn(
(directory: string, filename: string) => ({
@@ -145,6 +151,34 @@ describe('download resume validation', () => {
);
});
it('restarts without Range when a retained partial has no validator', async () => {
const harness = await setupResumeHarness({
finalSize: 4,
partialSize: 50,
response: {
data: Readable.from([Buffer.from('full')]),
headers: { 'content-length': '4', etag: '"etag-new"' },
status: 200,
},
});
harness.runtime.enqueueDownload(
createTask({
filePath: '/downloads/movie.mp4',
totalBytes: 54,
})
);
await waitForStatus(harness.set, 'completed');
const requestOptions =
harness.requestWithValidatedRedirects.mock.calls[0][1];
expect(requestOptions.headers).toEqual({});
expect(harness.createWriteStream).toHaveBeenCalledWith(
'/downloads/movie.mp4.part',
{ flags: 'w' }
);
});
it('restarts from byte zero when a resume request is answered with 200', async () => {
const harness = await setupResumeHarness({
finalSize: 4,
@@ -209,6 +243,7 @@ describe('download resume validation', () => {
harness.runtime.enqueueDownload(
createTask({
filePath: '/downloads/movie.mp4',
resumeValidator: '"etag-1"',
totalBytes: 54,
})
);
@@ -245,6 +280,7 @@ describe('download resume validation', () => {
harness.runtime.enqueueDownload(
createTask({
filePath: '/downloads/movie.mp4',
resumeValidator: '"etag-1"',
totalBytes: 100,
})
);
@@ -264,6 +300,191 @@ describe('download resume validation', () => {
}
});
it('retains received bytes when the connection resets mid-transfer', async () => {
const body = new PassThrough();
const harness = await setupResumeHarness({
finalSize: 'enoent',
partialSize: 0,
partialSizeAfterTransferError: 20,
response: {
data: body,
headers: {
'content-length': '100',
etag: '"etag-reset"',
},
status: 200,
},
});
const consoleError = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
try {
harness.runtime.enqueueDownload(createTask());
while (
harness.requestWithValidatedRedirects.mock.calls.length < 1
) {
await new Promise<void>((resolve) => setImmediate(resolve));
}
body.write(Buffer.alloc(20, 'r'));
const resetError = new Error(
'socket hang up'
) as NodeJS.ErrnoException;
resetError.code = 'ECONNRESET';
body.destroy(resetError);
await waitForStatus(harness.set, 'failed');
expect(harness.set).toHaveBeenCalledWith(
expect.objectContaining({
bytesDownloaded: 20,
errorMessage:
'DOWNLOAD_NETWORK_INTERRUPTED (ECONNRESET): Retry to continue from the saved partial file',
filePath: '/downloads/movie.mp4',
status: 'failed',
totalBytes: 100,
})
);
expect(harness.set).toHaveBeenCalledWith(
expect.objectContaining({
resumeValidator: '"etag-reset"',
})
);
expect(harness.removePartialDownloadFile).not.toHaveBeenCalled();
} finally {
consoleError.mockRestore();
}
});
it('retains an existing partial when a resumed response resets before another byte', async () => {
const body = new PassThrough();
const harness = await setupResumeHarness({
finalSize: 'enoent',
partialSize: 40,
partialSizeAfterTransferError: 40,
response: {
data: body,
headers: { 'content-range': 'bytes 40-99/100' },
status: 206,
},
});
const consoleError = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
try {
harness.runtime.enqueueDownload(
createTask({
filePath: '/downloads/movie.mp4',
resumeValidator: '"etag-reset"',
totalBytes: 100,
})
);
while (
harness.requestWithValidatedRedirects.mock.calls.length < 1
) {
await new Promise<void>((resolve) => setImmediate(resolve));
}
const resetError = new Error(
'socket hang up'
) as NodeJS.ErrnoException;
resetError.code = 'ECONNRESET';
body.destroy(resetError);
await waitForStatus(harness.set, 'failed');
expect(harness.set).toHaveBeenCalledWith(
expect.objectContaining({
bytesDownloaded: 40,
errorMessage:
'DOWNLOAD_NETWORK_INTERRUPTED (ECONNRESET): Retry to continue from the saved partial file',
filePath: '/downloads/movie.mp4',
status: 'failed',
totalBytes: 100,
})
);
expect(harness.removePartialDownloadFile).not.toHaveBeenCalled();
} finally {
consoleError.mockRestore();
}
});
it.each([
{
code: 'EUNKNOWN',
headers: { 'content-length': '100' },
label: 'unknown stream error',
partialSizeAfterTransferError: 20,
},
{
code: 'ECONNRESET',
headers: {},
label: 'response without an advertised total',
partialSizeAfterTransferError: 20,
},
{
code: 'ECONNRESET',
headers: { 'content-length': '100' },
label: 'response without a representation validator',
partialSizeAfterTransferError: 20,
},
{
code: 'ECONNRESET',
headers: { 'content-length': '100' },
label: 'fresh zero-byte failure',
partialSizeAfterTransferError: 0,
},
{
code: 'ECONNRESET',
headers: { 'content-length': '100' },
label: 'partial larger than the advertised total',
partialSizeAfterTransferError: 101,
},
])(
'uses generic cleanup for $label',
async ({ code, headers, partialSizeAfterTransferError }) => {
const body = new PassThrough();
const harness = await setupResumeHarness({
finalSize: 'enoent',
partialSize: 0,
partialSizeAfterTransferError,
response: { data: body, headers, status: 200 },
});
const consoleError = jest
.spyOn(console, 'error')
.mockImplementation(() => undefined);
try {
harness.runtime.enqueueDownload(createTask());
while (
harness.requestWithValidatedRedirects.mock.calls.length < 1
) {
await new Promise<void>((resolve) => setImmediate(resolve));
}
const streamError = new Error(
'socket hang up'
) as NodeJS.ErrnoException;
streamError.code = code;
body.destroy(streamError);
await waitForStatus(harness.set, 'failed');
expect(harness.removePartialDownloadFile).toHaveBeenCalledWith(
'/downloads/movie.mp4'
);
expect(harness.set).toHaveBeenCalledWith(
expect.objectContaining({
errorMessage: 'socket hang up',
filePath: null,
status: 'failed',
})
);
} finally {
consoleError.mockRestore();
}
}
);
it('captures a strong ETag from the first response for later resumes', async () => {
const harness = await setupResumeHarness({
finalSize: 4,
@@ -42,17 +42,16 @@ describe('download runtime pause and resume', () => {
it('persists active pause without deleting the partial file', async () => {
jest.resetModules();
const set = jest.fn(() => ({ where: jest.fn().mockResolvedValue(undefined) }));
const set = jest.fn(() => ({
where: jest.fn().mockResolvedValue(undefined),
}));
const db = {
update: jest.fn(() => ({ set })),
};
const removePartialDownloadFile = jest.fn();
const stream = new PassThrough();
const requestWithValidatedRedirects = jest.fn(
async (
_url: string,
options: { signal?: AbortSignal }
) => {
async (_url: string, options: { signal?: AbortSignal }) => {
options.signal?.addEventListener('abort', () => {
stream.destroy(new Error('aborted'));
});
@@ -119,10 +118,12 @@ describe('download runtime pause and resume', () => {
}
});
it('uses an HTTP Range header when a partial file already exists', async () => {
it('uses an HTTP Range header when a validated partial file already exists', async () => {
jest.resetModules();
const set = jest.fn(() => ({ where: jest.fn().mockResolvedValue(undefined) }));
const set = jest.fn(() => ({
where: jest.fn().mockResolvedValue(undefined),
}));
const db = {
update: jest.fn(() => ({ set })),
};
@@ -168,13 +169,17 @@ describe('download runtime pause and resume', () => {
runtime.enqueueDownload({
...createTask(),
filePath: '/downloads/movie.mp4',
resumeValidator: '"etag-1"',
});
await waitForCallCount(requestWithValidatedRedirects, 1);
expect(requestWithValidatedRedirects).toHaveBeenCalledWith(
'https://example.test/movie.mp4',
expect.objectContaining({
headers: expect.objectContaining({ Range: 'bytes=50-' }),
headers: expect.objectContaining({
'If-Range': '"etag-1"',
Range: 'bytes=50-',
}),
}),
{ allowPrivateNetworks: true }
);
@@ -184,7 +189,9 @@ describe('download runtime pause and resume', () => {
it('deletes a queued resumed partial file when the queued task is canceled', async () => {
jest.resetModules();
const set = jest.fn(() => ({ where: jest.fn().mockResolvedValue(undefined) }));
const set = jest.fn(() => ({
where: jest.fn().mockResolvedValue(undefined),
}));
const db = {
update: jest.fn(() => ({ set })),
};
@@ -271,7 +278,9 @@ describe('download runtime pause and resume', () => {
it('retains the partial path when canceling a queued task whose partial cannot be deleted', async () => {
jest.resetModules();
const set = jest.fn(() => ({ where: jest.fn().mockResolvedValue(undefined) }));
const set = jest.fn(() => ({
where: jest.fn().mockResolvedValue(undefined),
}));
const db = { update: jest.fn(() => ({ set })) };
const removePartialDownloadFile = jest.fn(() => {
throw new Error('EPERM: locked');
@@ -338,7 +347,9 @@ describe('download runtime pause and resume', () => {
it('retains the partial path when canceling a paused row whose partial cannot be deleted', async () => {
jest.resetModules();
const set = jest.fn(() => ({ where: jest.fn().mockResolvedValue(undefined) }));
const set = jest.fn(() => ({
where: jest.fn().mockResolvedValue(undefined),
}));
const db = {
select: jest.fn(() => ({
from: jest.fn(() => ({
@@ -34,12 +34,39 @@ export class TruncatedTransferError extends Error {
}
}
const RETAINABLE_NETWORK_ERROR_CODES = new Set([
'ECONNABORTED',
'ECONNRESET',
'EPIPE',
'ETIMEDOUT',
'ERR_HTTP2_STREAM_CANCEL',
'ERR_HTTP2_STREAM_ERROR',
'ERR_STREAM_PREMATURE_CLOSE',
]);
export class InterruptedTransferError extends Error {
constructor(
readonly progress: TransferProgress,
networkCode: string
) {
super(
`DOWNLOAD_NETWORK_INTERRUPTED (${networkCode}): Retry to continue from the saved partial file`
);
}
}
export async function transferToPartialFile(
db: DownloadsDatabase,
task: DownloadTask,
reservation: ReservedPartialDownloadFile
): Promise<TransferProgress> {
const resumeOffset = getResumeOffset(task, reservation);
const retainedOffset = getResumeOffset(task, reservation);
const resumeOffset = task.resumeValidator ? retainedOffset : 0;
if (retainedOffset > 0 && resumeOffset === 0) {
console.warn(
`[Downloads] Restarting ${reservation.filename} from the beginning (saved partial has no ETag or Last-Modified validator)`
);
}
const headers = {
...(task.headers ?? {}),
@@ -130,6 +157,22 @@ export async function transferToPartialFile(
try {
await pipeline(readable, output);
} catch (error) {
const interruptedProgress = getInterruptedTransferProgress(
error,
reservation,
effectiveOffset,
totalBytes,
task.resumeValidator
);
if (interruptedProgress) {
await persistProgress(db, task, interruptedProgress.progress);
throw new InterruptedTransferError(
interruptedProgress.progress,
interruptedProgress.networkCode
);
}
throw error;
} finally {
abortController.signal.removeEventListener('abort', abortStream);
}
@@ -141,6 +184,40 @@ export async function transferToPartialFile(
return { bytesDownloaded, totalBytes };
}
function getInterruptedTransferProgress(
error: unknown,
reservation: ReservedPartialDownloadFile,
initialBytes: number,
totalBytes: number | null,
resumeValidator: string | null | undefined
): { networkCode: string; progress: TransferProgress } | null {
const networkCode =
error && typeof error === 'object' && 'code' in error
? String(error.code)
: '';
if (
!RETAINABLE_NETWORK_ERROR_CODES.has(networkCode) ||
totalBytes === null ||
!resumeValidator
) {
return null;
}
const bytesDownloaded = getPartialDownloadSize(reservation.path);
if (
bytesDownloaded === 0 ||
bytesDownloaded < initialBytes ||
bytesDownloaded >= totalBytes
) {
return null;
}
return {
networkCode,
progress: { bytesDownloaded, totalBytes },
};
}
function getResumeOffset(
task: DownloadTask,
reservation: ReservedPartialDownloadFile
@@ -13,13 +13,10 @@
import axios, { AxiosRequestConfig } from 'axios';
import { ipcMain } from 'electron';
import { XTREAM_CLIENT_USER_AGENT } from '@iptvnator/shared/interfaces';
import { UnsafeUrlError } from './url-safety';
import { requestWithValidatedRedirects } from '../util/validated-axios';
// Some Xtream panels sit behind Cloudflare (or similar WAFs) configured to
// challenge generic browser-looking User-Agents while allowlisting known
// IPTV player clients. A VLC-style User-Agent reliably passes those checks.
const PROBE_CLIENT_USER_AGENT = 'VLC/3.0.18 LibVLC/3.0.18';
const PROBE_TIMEOUT_MS = 10000;
export interface StreamProbePayload {
@@ -64,7 +61,7 @@ export async function runStreamProbe(
// The playlist's own User-Agent wins when it has one: the default
// below is a guess that merely gets past most WAFs, while that
// one is what the server was configured to expect.
'User-Agent': payload.userAgent?.trim() || PROBE_CLIENT_USER_AGENT,
'User-Agent': payload.userAgent?.trim() || XTREAM_CLIENT_USER_AGENT,
...(payload.referer?.trim()
? { Referer: payload.referer.trim() }
: {}),
@@ -95,8 +92,7 @@ export async function runStreamProbe(
);
// A ranged GET opens a stream we never read — release it immediately.
const responseBody = response.data as
| { destroy?: () => void }
| undefined;
{ destroy?: () => void } | undefined;
responseBody?.destroy?.();
return {
status: response.status,
@@ -8,6 +8,7 @@ import { ipcMain } from 'electron';
import {
PortalDebugEvent,
XTREAM_CANCEL_SESSION,
XTREAM_CLIENT_USER_AGENT,
XTREAM_MAIN_PERFORMANCE_PHASE,
normalizeXtreamServerUrl,
} from '@iptvnator/shared/interfaces';
@@ -20,11 +21,6 @@ import {
} from './xtream-performance';
import { cancelXtreamSessionRequests } from './xtream-session-cancellation';
// Some Xtream panels sit behind Cloudflare (or similar WAFs) configured to
// challenge generic browser-looking User-Agents while allowlisting known
// IPTV player clients. A VLC-style User-Agent reliably passes those checks.
const XTREAM_CLIENT_USER_AGENT = 'VLC/3.0.18 LibVLC/3.0.18';
export default class XtreamEvents {
static bootstrapXtreamEvents(): Electron.IpcMain {
return ipcMain;