test(downloads): harden series queue fixtures

apps/electron-backend-e2e/src/series-download-queue.e2e.ts

@@ -26,10 +26,38 @@ const expectedEpisodeIds = [80_000, 80_001, 80_002, 80_003];
 
 type EpisodeQueueRow = {
     episodeNumber: number | null;
+    seriesXtreamId: number | null;
     status: string;
     xtreamId: number;
 };
 
+const expectedEpisodeQueue: EpisodeQueueRow[] = [
+    {
+        episodeNumber: 1,
+        seriesXtreamId: 30_000,
+        status: 'downloading',
+        xtreamId: expectedEpisodeIds[0],
+    },
+    {
+        episodeNumber: 2,
+        seriesXtreamId: 30_000,
+        status: 'queued',
+        xtreamId: expectedEpisodeIds[1],
+    },
+    {
+        episodeNumber: 3,
+        seriesXtreamId: 30_000,
+        status: 'queued',
+        xtreamId: expectedEpisodeIds[2],
+    },
+    {
+        episodeNumber: 4,
+        seriesXtreamId: 30_000,
+        status: 'queued',
+        xtreamId: expectedEpisodeIds[3],
+    },
+];
+
 type SeriesFixture = Awaited<ReturnType<typeof fetchXtreamSeriesFixture>>;
 
 function waitForQueueStabilityInterval(): Promise<void> {
@@ -49,28 +77,18 @@ function readFirstSeriesTitle(fixture: SeriesFixture): string {
     return title;
 }
 
-async function readEpisodeQueue(
-    page: Page,
-    episodeIds: readonly number[]
-): Promise<EpisodeQueueRow[]> {
-    return page.evaluate(
-        async (expectedIds) => {
-            const downloads =
-                (await window.electron?.downloadsGetList?.()) ?? [];
-            return downloads
-                .filter(
-                    (download) =>
-                        download.contentType === 'episode' &&
-                        expectedIds.includes(download.xtreamId)
-                )
-                .map((download) => ({
-                    episodeNumber: download.episodeNumber ?? null,
-                    status: download.status,
-                    xtreamId: download.xtreamId,
-                }));
-        },
-        [...episodeIds]
-    );
+async function readEpisodeQueue(page: Page): Promise<EpisodeQueueRow[]> {
+    return page.evaluate(async () => {
+        const downloads = (await window.electron?.downloadsGetList?.()) ?? [];
+        return downloads
+            .filter((download) => download.contentType === 'episode')
+            .map((download) => ({
+                episodeNumber: download.episodeNumber ?? null,
+                seriesXtreamId: download.seriesXtreamId ?? null,
+                status: download.status,
+                xtreamId: download.xtreamId,
+            }));
+    });
 }
 
 test.describe('Electron Series Download Queue', () => {
@@ -137,22 +155,10 @@ test.describe('Electron Series Download Queue', () => {
             await secondEpisodeAction.click();
 
             await expect
-                .poll(
-                    () => readEpisodeQueue(app.mainWindow, expectedEpisodeIds),
-                    { timeout: 20_000 }
-                )
-                .toEqual([
-                    {
-                        episodeNumber: 1,
-                        status: 'downloading',
-                        xtreamId: expectedEpisodeIds[0],
-                    },
-                    {
-                        episodeNumber: 2,
-                        status: 'queued',
-                        xtreamId: expectedEpisodeIds[1],
-                    },
-                ]);
+                .poll(() => readEpisodeQueue(app.mainWindow), {
+                    timeout: 20_000,
+                })
+                .toEqual(expectedEpisodeQueue.slice(0, 2));
 
             const seasonAction = app.mainWindow.getByTestId('download-season');
             await expect(seasonAction).toBeVisible();
@@ -166,42 +172,17 @@ test.describe('Electron Series Download Queue', () => {
             ).toBeVisible({ timeout: 20_000 });
 
             await expect
-                .poll(
-                    async () =>
-                        (
-                            await readEpisodeQueue(
-                                app.mainWindow,
-                                expectedEpisodeIds
-                            )
-                        ).map((row) => row.xtreamId),
-                    { timeout: 20_000 }
-                )
-                .toEqual(expectedEpisodeIds);
+                .poll(() => readEpisodeQueue(app.mainWindow), {
+                    timeout: 20_000,
+                })
+                .toEqual(expectedEpisodeQueue);
 
-            const queuedSeason = await readEpisodeQueue(
-                app.mainWindow,
-                expectedEpisodeIds
-            );
-            expect(new Set(queuedSeason.map((row) => row.xtreamId)).size).toBe(
-                4
-            );
-            expect(queuedSeason.map((row) => row.status)).toEqual([
-                'downloading',
-                'queued',
-                'queued',
-                'queued',
-            ]);
             await expect(seasonAction).toBeDisabled();
             await expect(seasonAction).toContainText('Download season (0)');
 
             await waitForQueueStabilityInterval();
-            const stableQueue = await readEpisodeQueue(
-                app.mainWindow,
-                expectedEpisodeIds
-            );
-            expect(stableQueue).toHaveLength(4);
-            expect(new Set(stableQueue.map((row) => row.xtreamId)).size).toBe(
-                4
+            expect(await readEpisodeQueue(app.mainWindow)).toEqual(
+                expectedEpisodeQueue
             );
         } finally {
             await closeElectronApp(app);

apps/xtream-mock-server/src/app/server.spec.ts

@@ -1,9 +1,14 @@
 import { connect } from 'node:net';
+import express from 'express';
 import { resetAll } from './data-store.js';
 import {
     createXtreamMockApp,
     parseXtreamMockServerEnvironment,
 } from './server.js';
+import {
+    type SlowSeriesDownloadOptions,
+    streamSlowSeriesDownload,
+} from './slow-series-download.js';
 import { startLoopbackServer } from './testing/http-server.fixture.js';
 
 jest.mock('@faker-js/faker', () => {
@@ -224,6 +229,64 @@ describe('Xtream mock server factory', () => {
     });
 });
 
+describe('Slow series download stream', () => {
+    it('completes the configured byte count over a loopback response', async () => {
+        const options = {
+            chunkSize: 1_024,
+            intervalMs: 1,
+            totalBytes: 10 * 1_024 + 7,
+        } satisfies SlowSeriesDownloadOptions;
+        const running = await startLoopbackServer(
+            createSlowSeriesDownloadApp(options)
+        );
+        let closed = false;
+
+        try {
+            const response = await fetch(`${running.origin}/slow-series`);
+            const body = await response.arrayBuffer();
+
+            expect(response.status).toBe(200);
+            expect(response.headers.get('content-type')).toContain('video/mp4');
+            expect(Number(response.headers.get('content-length'))).toBe(
+                options.totalBytes
+            );
+            expect(body.byteLength).toBe(options.totalBytes);
+
+            await within(running.close(), 1_000);
+            closed = true;
+        } finally {
+            if (!closed) await running.close().catch(() => undefined);
+        }
+    });
+
+    it('stops a longer recursive stream after the response body is cancelled', async () => {
+        const options = {
+            chunkSize: 1_024,
+            intervalMs: 25,
+            totalBytes: 64 * 1_024,
+        } satisfies SlowSeriesDownloadOptions;
+        const running = await startLoopbackServer(
+            createSlowSeriesDownloadApp(options)
+        );
+        let closed = false;
+
+        try {
+            const response = await fetch(`${running.origin}/slow-series`);
+            const reader = response.body?.getReader();
+            if (!reader) throw new Error('Expected a streaming response body.');
+
+            const firstChunk = await within(reader.read(), 1_000);
+            expect(firstChunk.done).toBe(false);
+            expect(firstChunk.value?.byteLength).toBeGreaterThan(0);
+            await within(reader.cancel(), 1_000);
+            await within(running.close(), 1_000);
+            closed = true;
+        } finally {
+            if (!closed) await running.close().catch(() => undefined);
+        }
+    });
+});
+
 describe('Xtream mock environment parsing', () => {
     it('uses safe defaults and enables control only for the exact flag', () => {
         expect(parseXtreamMockServerEnvironment({})).toEqual({
@@ -303,3 +366,28 @@ async function rawLoopbackGet(
         });
     });
 }
+
+function createSlowSeriesDownloadApp(
+    options: SlowSeriesDownloadOptions
+): express.Express {
+    const app = express();
+    app.get('/slow-series', (request, response) => {
+        streamSlowSeriesDownload(request, response, options);
+    });
+    return app;
+}
+
+async function within<T>(promise: Promise<T>, timeoutMs: number): Promise<T> {
+    let timer: NodeJS.Timeout | undefined;
+    const timeout = new Promise<never>((_resolve, reject) => {
+        timer = setTimeout(
+            () => reject(new Error(`Operation exceeded ${timeoutMs}ms.`)),
+            timeoutMs
+        );
+    });
+    try {
+        return await Promise.race([promise, timeout]);
+    } finally {
+        if (timer) clearTimeout(timer);
+    }
+}

apps/xtream-mock-server/src/app/server.ts

@@ -15,6 +15,7 @@ import {
 } from './performance-control.js';
 import { dispatchAction } from './routes/dispatch.js';
 import { getScenario } from './scenarios.js';
+import { streamSlowSeriesDownload } from './slow-series-download.js';
 
 export { createXtreamMockServerShutdown } from './server-lifecycle.js';
 
@@ -47,9 +48,6 @@ const DEFAULT_NORMAL_HOST = '0.0.0.0';
 const DEFAULT_CONTROL_HOST = '127.0.0.1';
 const PERFORMANCE_USERNAME = 'performance';
 const PERFORMANCE_PASSWORD = 'performance';
-const DOWNLOAD_STREAM_BYTES = 8 * 1024 * 1024;
-const DOWNLOAD_STREAM_CHUNK_INTERVAL_MS = 100;
-const DOWNLOAD_STREAM_CHUNK = Buffer.alloc(32 * 1024);
 const marketingRasterAssetRoot = join(
     process.cwd(),
     'apps/xtream-mock-server/public/marketing'
@@ -282,60 +280,6 @@ function isSlowSeriesDownloadRequest(request: Request): boolean {
     );
 }
 
-function streamSlowSeriesDownload(request: Request, response: Response): void {
-    let sentBytes = 0;
-    let stopped = false;
-    let timer: NodeJS.Timeout | undefined;
-
-    const stop = () => {
-        stopped = true;
-        if (timer) {
-            clearTimeout(timer);
-            timer = undefined;
-        }
-        response.off('drain', scheduleChunk);
-    };
-    const scheduleChunk = () => {
-        if (!stopped && !timer) {
-            timer = setTimeout(writeChunk, DOWNLOAD_STREAM_CHUNK_INTERVAL_MS);
-        }
-    };
-    const writeChunk = () => {
-        timer = undefined;
-        if (stopped || response.destroyed || response.writableEnded) {
-            stop();
-            return;
-        }
-
-        const remainingBytes = DOWNLOAD_STREAM_BYTES - sentBytes;
-        const chunk =
-            remainingBytes >= DOWNLOAD_STREAM_CHUNK.length
-                ? DOWNLOAD_STREAM_CHUNK
-                : DOWNLOAD_STREAM_CHUNK.subarray(0, remainingBytes);
-        sentBytes += chunk.length;
-        const canContinue = response.write(chunk);
-
-        if (sentBytes >= DOWNLOAD_STREAM_BYTES) {
-            response.end();
-        } else if (canContinue) {
-            scheduleChunk();
-        } else {
-            response.once('drain', scheduleChunk);
-        }
-    };
-
-    request.once('close', stop);
-    response.once('close', stop);
-    response.once('finish', stop);
-    response
-        .status(200)
-        .type('video/mp4')
-        .set('Content-Length', String(DOWNLOAD_STREAM_BYTES))
-        .set('Cache-Control', 'no-store')
-        .flushHeaders();
-    scheduleChunk();
-}
-
 function isPerformanceMediaRequest(
     request: Request,
     controlEnabled: boolean

apps/xtream-mock-server/src/app/slow-series-download.ts

@@ -0,0 +1,93 @@
+import type { Request, Response } from 'express';
+
+export interface SlowSeriesDownloadOptions {
+    readonly chunkSize: number;
+    readonly intervalMs: number;
+    readonly totalBytes: number;
+}
+
+export const DEFAULT_SLOW_SERIES_DOWNLOAD_OPTIONS: SlowSeriesDownloadOptions = {
+    chunkSize: 32 * 1024,
+    intervalMs: 100,
+    totalBytes: 8 * 1024 * 1024,
+};
+
+export function streamSlowSeriesDownload(
+    request: Request,
+    response: Response,
+    options: SlowSeriesDownloadOptions = DEFAULT_SLOW_SERIES_DOWNLOAD_OPTIONS
+): void {
+    validateOptions(options);
+    const chunkBuffer = Buffer.alloc(options.chunkSize);
+    let sentBytes = 0;
+    let stopped = false;
+    let timer: NodeJS.Timeout | undefined;
+
+    function stop(): void {
+        if (stopped) return;
+        stopped = true;
+        if (timer) {
+            clearTimeout(timer);
+            timer = undefined;
+        }
+        request.off('close', stop);
+        response.off('close', stop);
+        response.off('finish', stop);
+        response.off('drain', scheduleChunk);
+    }
+
+    function scheduleChunk(): void {
+        if (!stopped && !timer) {
+            timer = setTimeout(writeChunk, options.intervalMs);
+        }
+    }
+
+    function writeChunk(): void {
+        timer = undefined;
+        if (stopped || response.destroyed || response.writableEnded) {
+            stop();
+            return;
+        }
+
+        const remainingBytes = options.totalBytes - sentBytes;
+        const chunk =
+            remainingBytes >= chunkBuffer.length
+                ? chunkBuffer
+                : chunkBuffer.subarray(0, remainingBytes);
+        sentBytes += chunk.length;
+        const canContinue = response.write(chunk);
+
+        if (sentBytes >= options.totalBytes) {
+            response.end();
+        } else if (canContinue) {
+            scheduleChunk();
+        } else {
+            response.once('drain', scheduleChunk);
+        }
+    }
+
+    request.once('close', stop);
+    response.once('close', stop);
+    response.once('finish', stop);
+    response
+        .status(200)
+        .type('video/mp4')
+        .set('Content-Length', String(options.totalBytes))
+        .set('Cache-Control', 'no-store')
+        .flushHeaders();
+    scheduleChunk();
+}
+
+function validateOptions(options: SlowSeriesDownloadOptions): void {
+    if (!Number.isSafeInteger(options.totalBytes) || options.totalBytes <= 0) {
+        throw new Error('Slow download totalBytes must be a positive integer.');
+    }
+    if (!Number.isSafeInteger(options.chunkSize) || options.chunkSize <= 0) {
+        throw new Error('Slow download chunkSize must be a positive integer.');
+    }
+    if (!Number.isSafeInteger(options.intervalMs) || options.intervalMs < 0) {
+        throw new Error(
+            'Slow download intervalMs must be a non-negative integer.'
+        );
+    }
+}