diff --git a/electron/lib/localInference.js b/electron/lib/localInference.js index d26ea6f6d..e45f9878b 100644 --- a/electron/lib/localInference.js +++ b/electron/lib/localInference.js @@ -2,8 +2,8 @@ const { ipcMain, app, BrowserWindow } = require('electron'); const path = require('path'); const fs = require('fs'); const https = require('https'); -const http = require('http'); const { spawn, execFile } = require('child_process'); +const { downloadFile } = require('./localInferenceDownload'); const { getBundledBinaryResourceDir, pickBinaryAssetForPlatform, @@ -67,84 +67,6 @@ function fetchJson(url) { }); } -// ─── Robust HTTPS download with redirect-following, range-resume, and retry ─── -function downloadFile(url, destPath, onProgress) { - const tmp = destPath + '.part'; - - // Outer total so progress never goes backwards across retries/redirects - let knownTotal = 0; - - const attempt = (requestUrl, redirectsLeft, retriesLeft) => new Promise((resolve, reject) => { - // Resume from however many bytes are already on disk - const alreadyDownloaded = fs.existsSync(tmp) ? fs.statSync(tmp).size : 0; - - const parsed = new URL(requestUrl); - const mod = parsed.protocol === 'https:' ? https : http; - - const reqHeaders = { - 'User-Agent': 'Mozilla/5.0 (compatible; open-generative-ai/1.0)', - 'Accept': '*/*', - 'Connection': 'keep-alive', - }; - if (alreadyDownloaded > 0) reqHeaders['Range'] = `bytes=${alreadyDownloaded}-`; - - const req = mod.get({ hostname: parsed.hostname, path: parsed.pathname + parsed.search, headers: reqHeaders }, (res) => { - const { statusCode, headers } = res; - - // Follow redirects - if ([301, 302, 303, 307, 308].includes(statusCode)) { - res.resume(); - if (redirectsLeft <= 0) { reject(new Error('Too many redirects')); return; } - resolve(attempt(headers.location, redirectsLeft - 1, retriesLeft)); - return; - } - - // 206 Partial Content (range accepted) or 200 OK (server ignored Range) - if (statusCode !== 200 && statusCode !== 206) { - res.resume(); - reject(new Error(`HTTP ${statusCode} from ${parsed.hostname}${parsed.pathname}`)); - return; - } - - // content-length on a 206 is the remaining bytes; on 200 it's the full file - const chunkSize = parseInt(headers['content-length'] || '0', 10); - if (statusCode === 200) { - // Server ignored our Range header — restart the file - if (fs.existsSync(tmp)) fs.unlinkSync(tmp); - knownTotal = chunkSize; - } else { - // 206: total = already downloaded + remaining - knownTotal = alreadyDownloaded + chunkSize; - } - - let received = alreadyDownloaded; - const out = fs.createWriteStream(tmp, { flags: statusCode === 206 ? 'a' : 'w' }); - - res.on('data', (chunk) => { - received += chunk.length; - if (knownTotal && onProgress) onProgress(received / knownTotal); - }); - res.pipe(out); - out.on('finish', () => { fs.renameSync(tmp, destPath); resolve(); }); - out.on('error', reject); - res.on('error', reject); - }); - - req.on('error', (err) => { - if (retriesLeft > 0) { - console.warn(`[download] ${err.message} — retrying in 3s (${retriesLeft} left)`); - setTimeout(() => resolve(attempt(requestUrl, redirectsLeft, retriesLeft - 1)), 3000); - } else { - reject(err); - } - }); - - req.setTimeout(60000, () => req.destroy(new Error('Request timed out'))); - }); - - return attempt(url, 10, 5); -} - // ─── Extract zip on each platform ──────────────────────────────────────────── function extractZip(zipPath, destDir) { return new Promise((resolve, reject) => { @@ -405,8 +327,10 @@ async function deleteModel(modelId) { const filePath = path.join(MODELS_DIR, model.filename); if (fs.existsSync(filePath)) fs.unlinkSync(filePath); - const partPath = filePath + '.part'; - if (fs.existsSync(partPath)) fs.unlinkSync(partPath); + for (const suffix of ['.part', '.part.meta.json', '.part.fresh']) { + const partialArtifact = filePath + suffix; + if (fs.existsSync(partialArtifact)) fs.unlinkSync(partialArtifact); + } return { ok: true }; } diff --git a/electron/lib/localInferenceDownload.js b/electron/lib/localInferenceDownload.js new file mode 100644 index 000000000..33036f789 --- /dev/null +++ b/electron/lib/localInferenceDownload.js @@ -0,0 +1,405 @@ +const fs = require('fs'); +const { Readable, Transform } = require('stream'); +const { pipeline } = require('stream/promises'); + +const REQUEST_HEADERS = { + 'User-Agent': 'Mozilla/5.0 (compatible; open-generative-ai/1.0)', + 'Accept': '*/*', + 'Accept-Encoding': 'identity', +}; + +function parseContentLength(value) { + if (value === null) return null; + if (typeof value !== 'string' || !/^\d+$/.test(value)) return null; + const parsed = Number(value); + return Number.isSafeInteger(parsed) ? parsed : null; +} + +function parseContentRange(value) { + const match = /^bytes\s+(\d+)-(\d+)\/(\d+)$/i.exec(value || ''); + if (!match) return null; + const start = Number(match[1]); + const end = Number(match[2]); + const total = Number(match[3]); + if (![start, end, total].every(Number.isSafeInteger)) return null; + if (start > end || end >= total) return null; + return { start, end, total }; +} + +function parseStrongEtag(value) { + if (typeof value !== 'string') return null; + const etag = value.trim(); + if (/^W\//i.test(etag)) return null; + return /^"[\x21\x23-\x7E\u0080-\u00FF]*"$/.test(etag) ? etag : null; +} + +function normalizeSourceUrl(url) { + const parsed = new URL(url); + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') { + throw new Error(`Unsupported download protocol: ${parsed.protocol}`); + } + parsed.hash = ''; + return parsed.toString(); +} + +function getPaths(destPath) { + const partPath = destPath + '.part'; + return { + partPath, + metadataPath: partPath + '.meta.json', + freshPath: partPath + '.fresh', + }; +} + +function removeIfExists(filePath) { + if (fs.existsSync(filePath)) fs.unlinkSync(filePath); +} + +function writeMetadata(metadataPath, sourceUrl, etag) { + const temporaryPath = `${metadataPath}.tmp-${process.pid}-${Date.now()}`; + try { + fs.writeFileSync(temporaryPath, JSON.stringify({ version: 1, sourceUrl, etag }), 'utf8'); + removeIfExists(metadataPath); + fs.renameSync(temporaryPath, metadataPath); + } finally { + removeIfExists(temporaryPath); + } +} + +function readResumeState(paths, sourceUrl) { + try { + const stat = fs.statSync(paths.partPath); + const metadata = JSON.parse(fs.readFileSync(paths.metadataPath, 'utf8')); + const etag = parseStrongEtag(metadata.etag); + if (!stat.isFile() || stat.size <= 0) return null; + if (metadata.version !== 1 || metadata.sourceUrl !== sourceUrl || !etag) return null; + return { size: stat.size, etag }; + } catch { + return null; + } +} + +function getFetch(options) { + const fetchImpl = options.fetchImpl || globalThis.fetch; + if (typeof fetchImpl !== 'function') throw new Error('This runtime does not provide fetch'); + return fetchImpl; +} + +function createInactivityTimeout(controller, timeoutMs, label) { + let timer; + const arm = () => { + clearTimeout(timer); + timer = setTimeout(() => controller.abort(new Error(`${label} timed out`)), timeoutMs); + }; + const clear = () => clearTimeout(timer); + arm(); + return { arm, clear }; +} + +async function openResponse( + url, + method, + headers, + options, + redirectsLeft = options.maxRedirects ?? 10, + retryState = { remaining: options.maxRetries ?? 5 } +) { + const fetchImpl = getFetch(options); + let response; + let controller; + let timeout; + + while (!response) { + controller = new AbortController(); + timeout = createInactivityTimeout( + controller, + options.timeoutMs ?? 60000, + `${method} request` + ); + try { + response = await fetchImpl(url, { + method, + redirect: method === 'HEAD' ? 'follow' : 'manual', + signal: controller.signal, + headers: { ...REQUEST_HEADERS, ...headers }, + }); + } catch (err) { + timeout.clear(); + if (retryState.remaining <= 0) { + throw new Error(`${method} ${url} failed: ${err.message}`, { cause: err }); + } + retryState.remaining -= 1; + const delayMs = options.retryDelayMs ?? 3000; + if (delayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, delayMs)); + } + } + } + + try { + timeout.arm(); + const session = { response, controller, timeout }; + if (method === 'GET' && [301, 302, 303, 307, 308].includes(response.status)) { + if (redirectsLeft <= 0) { + await discardResponse(session); + throw new Error('Too many redirects'); + } + let redirectUrl; + try { + const location = response.headers.get('location'); + if (!location) throw new Error('missing Location header'); + redirectUrl = normalizeSourceUrl( + new URL(location, url).toString() + ); + } catch (err) { + await discardResponse(session); + throw new Error(`Invalid redirect Location: ${err.message}`, { cause: err }); + } + await discardResponse(session); + return openResponse( + redirectUrl, + method, + headers, + options, + redirectsLeft - 1, + retryState + ); + } + return session; + } catch (err) { + timeout.clear(); + throw new Error(`${method} ${url} failed: ${err.message}`, { cause: err }); + } +} + +function waitForClose(stream) { + if (stream.closed) return Promise.resolve(); + return new Promise((resolve) => stream.once('close', resolve)); +} + +async function discardResponse(session) { + session.timeout.clear(); + session.controller.abort(); + try { + await session.response.body?.cancel(); + } catch { + // Abort already closed the response transport. + } +} + +async function revalidateCompletePartial(sourceUrl, resumeState, options) { + let session; + try { + session = await openResponse(sourceUrl, 'HEAD', {}, options); + const { response } = session; + return response.status === 200 + && parseStrongEtag(response.headers.get('etag')) === resumeState.etag + && parseContentLength(response.headers.get('content-length')) === resumeState.size; + } catch { + return false; + } finally { + if (session) await discardResponse(session); + } +} + +function promoteValidatedPartial(paths, destPath, sourceUrl, expectedState, onProgress) { + const currentState = readResumeState(paths, sourceUrl); + if ( + !currentState + || currentState.etag !== expectedState.etag + || currentState.size !== expectedState.size + ) { + return false; + } + const finalStat = fs.statSync(paths.partPath); + if (!finalStat.isFile() || finalStat.size !== expectedState.size) return false; + fs.renameSync(paths.partPath, destPath); + removeIfExists(paths.metadataPath); + onProgress(1); + return true; +} + +async function saveResponse(session, outputPath, flags, initialSize, expectedBytes, expectedSize, onProgress) { + let responseBytes = 0; + let output; + let outputClosed; + const counter = new Transform({ + transform(chunk, _encoding, callback) { + session.timeout.arm(); + responseBytes += chunk.length; + if (expectedSize) { + onProgress(Math.min((initialSize + responseBytes) / expectedSize, 1)); + } + callback(null, chunk); + }, + }); + + try { + if (session.response.body) { + output = fs.createWriteStream(outputPath, { flags }); + outputClosed = waitForClose(output); + await pipeline( + Readable.fromWeb(session.response.body), + counter, + output + ); + await outputClosed; + } else { + fs.writeFileSync(outputPath, Buffer.alloc(0), { flag: flags }); + } + } catch (err) { + session.controller.abort(); + if (output) { + output.destroy(); + await outputClosed; + } + throw err; + } finally { + session.timeout.clear(); + } + + if (expectedBytes !== null && responseBytes !== expectedBytes) { + throw new Error(`Download received ${responseBytes} bytes, expected ${expectedBytes}`); + } + const finalSize = fs.statSync(outputPath).size; + if (expectedSize !== null && finalSize !== expectedSize) { + throw new Error(`Partial download is ${finalSize} bytes, expected ${expectedSize}`); + } +} + +async function rejectResponse(session, message) { + await discardResponse(session); + throw new Error(message); +} + +async function validatedContentLength(session) { + const raw = session.response.headers.get('content-length'); + const parsed = parseContentLength(raw); + if (raw !== null && parsed === null) { + return rejectResponse(session, 'Invalid Content-Length'); + } + return parsed; +} + +async function finishFreshDownload(sourceUrl, destPath, paths, onProgress, options, existingSession) { + removeIfExists(paths.freshPath); + let session = existingSession; + try { + if (!session) session = await openResponse(sourceUrl, 'GET', {}, options); + if (session.response.status !== 200) { + return await rejectResponse(session, `HTTP ${session.response.status} during fresh download`); + } + const contentLength = await validatedContentLength(session); + await saveResponse(session, paths.freshPath, 'w', 0, contentLength, contentLength, onProgress); + fs.renameSync(paths.freshPath, destPath); + removeIfExists(paths.partPath); + removeIfExists(paths.metadataPath); + } catch (err) { + if (session) await discardResponse(session); + removeIfExists(paths.freshPath); + throw err; + } +} + +async function downloadFileInternal(url, destPath, onProgress, options) { + const sourceUrl = normalizeSourceUrl(url); + const paths = getPaths(destPath); + removeIfExists(paths.freshPath); + + const resumeState = readResumeState(paths, sourceUrl); + const hasPartialArtifacts = fs.existsSync(paths.partPath) || fs.existsSync(paths.metadataPath); + const headers = resumeState + ? { Range: `bytes=${resumeState.size}-`, 'If-Range': resumeState.etag } + : {}; + const session = await openResponse(sourceUrl, 'GET', headers, options); + const { response } = session; + + if (response.status === 416 && resumeState) { + await discardResponse(session); + if ( + await revalidateCompletePartial(sourceUrl, resumeState, options) + && promoteValidatedPartial(paths, destPath, sourceUrl, resumeState, onProgress) + ) { + return; + } + return finishFreshDownload(sourceUrl, destPath, paths, onProgress, options); + } + + if (response.status === 206) { + if (!resumeState) { + return rejectResponse(session, 'Unexpected HTTP 206 without a validated partial'); + } + const responseEtag = parseStrongEtag(response.headers.get('etag')); + const contentRange = parseContentRange(response.headers.get('content-range')); + if (responseEtag !== resumeState.etag || !contentRange) { + await discardResponse(session); + return finishFreshDownload(sourceUrl, destPath, paths, onProgress, options); + } + if (contentRange.start !== resumeState.size) { + return rejectResponse(session, `HTTP 206 started at ${contentRange.start}, expected ${resumeState.size}`); + } + const rangeLength = contentRange.end - contentRange.start + 1; + const contentLength = await validatedContentLength(session); + if (contentLength !== null && contentLength !== rangeLength) { + return rejectResponse(session, 'HTTP 206 Content-Length did not match Content-Range'); + } + await saveResponse( + session, + paths.partPath, + 'a', + resumeState.size, + rangeLength, + contentRange.total, + onProgress + ); + fs.renameSync(paths.partPath, destPath); + removeIfExists(paths.metadataPath); + return; + } + + if (response.status !== 200) { + return rejectResponse(session, `HTTP ${response.status} from ${new URL(response.url).hostname}`); + } + + if (hasPartialArtifacts) { + return finishFreshDownload(sourceUrl, destPath, paths, onProgress, options, session); + } + + try { + const contentLength = await validatedContentLength(session); + const etag = parseStrongEtag(response.headers.get('etag')); + removeIfExists(paths.partPath); + removeIfExists(paths.metadataPath); + if (etag) writeMetadata(paths.metadataPath, sourceUrl, etag); + await saveResponse(session, paths.partPath, 'w', 0, contentLength, contentLength, onProgress); + fs.renameSync(paths.partPath, destPath); + removeIfExists(paths.metadataPath); + } catch (err) { + await discardResponse(session); + const etag = parseStrongEtag(response.headers.get('etag')); + if (!etag) { + removeIfExists(paths.partPath); + removeIfExists(paths.metadataPath); + } + throw err; + } +} + +function downloadFile(url, destPath, onProgress, options = {}) { + const reportProgress = (progress) => { + if (!onProgress) return; + try { + onProgress(progress); + } catch (err) { + console.warn(`[download] Progress listener failed: ${err.message}`); + } + }; + return Promise.resolve().then(() => downloadFileInternal( + url, + destPath, + reportProgress, + options + )); +} + +module.exports = { downloadFile }; diff --git a/tests/localInferenceDownload.test.js b/tests/localInferenceDownload.test.js new file mode 100644 index 000000000..f478e559f --- /dev/null +++ b/tests/localInferenceDownload.test.js @@ -0,0 +1,375 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const http = require('node:http'); +const os = require('node:os'); +const path = require('node:path'); + +const { downloadFile } = require('../electron/lib/localInferenceDownload'); + +async function withServer(handler, run) { + const server = http.createServer(handler); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + try { + return await run(`http://127.0.0.1:${server.address().port}/model.bin`); + } finally { + server.closeAllConnections(); + await new Promise((resolve) => server.close(resolve)); + } +} + +async function withTempDir(run) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'oga-download-')); + try { + return await run(dir); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } +} + +function writeResumeState(destination, sourceUrl, contents, etag = '"v1"') { + fs.writeFileSync(destination + '.part', contents); + fs.writeFileSync( + destination + '.part.meta.json', + JSON.stringify({ version: 1, sourceUrl, etag }) + ); +} + +async function expectSocketClosed(socketClosed, getSocket, message) { + let timeout; + try { + await Promise.race([ + socketClosed, + new Promise((_, reject) => { + timeout = setTimeout(() => { + getSocket()?.destroy(); + reject(new Error(message)); + }, 1000); + }), + ]); + } finally { + clearTimeout(timeout); + } +} + +test('resumes only with a matching strong ETag and strict HTTP 206 range', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + const progress = []; + let requestHeaders; + await withServer((req, res) => { + requestHeaders = req.headers; + res.writeHead(206, { + 'Content-Range': 'bytes 6-10/11', + 'Content-Length': '5', + ETag: '"v1"', + }); + res.end('world'); + }, async (url) => { + writeResumeState(destination, url, 'hello '); + await downloadFile(url, destination, (value) => progress.push(value)); + }); + assert.equal(requestHeaders.range, 'bytes=6-'); + assert.equal(requestHeaders['if-range'], '"v1"'); + assert.equal(fs.readFileSync(destination, 'utf8'), 'hello world'); + assert.equal(progress.at(-1), 1); + }); +}); + +test('revalidates an Xet-style headerless HTTP 416 with canonical HEAD', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + const requests = []; + await withServer((req, res) => { + requests.push({ + method: req.method, + path: req.url, + range: req.headers.range, + ifRange: req.headers['if-range'], + }); + if (req.url === '/model.bin') { + res.writeHead(302, { Location: '/xet-asset' }); + res.end(); + } else if (req.method === 'HEAD') { + res.writeHead(200, { 'Content-Length': '8', ETag: '"v1"' }); + res.end(); + } else { + res.writeHead(416, { 'Content-Length': '0' }); + res.end(); + } + }, async (url) => { + writeResumeState(destination, url, 'complete'); + await downloadFile(url, destination); + }); + assert.deepEqual(requests, [ + { method: 'GET', path: '/model.bin', range: 'bytes=8-', ifRange: '"v1"' }, + { method: 'GET', path: '/xet-asset', range: 'bytes=8-', ifRange: '"v1"' }, + { method: 'HEAD', path: '/model.bin', range: undefined, ifRange: undefined }, + { method: 'HEAD', path: '/xet-asset', range: undefined, ifRange: undefined }, + ]); + assert.equal(fs.readFileSync(destination, 'utf8'), 'complete'); + assert.equal(fs.existsSync(destination + '.part.meta.json'), false); + }); +}); + +test('retries one pre-response socket failure after a redirect', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + const requests = []; + let assetRequests = 0; + await withServer((req, res) => { + requests.push(req.url); + if (req.url === '/model.bin') { + res.writeHead(302, { Location: '/asset' }); + res.end(); + return; + } + assetRequests += 1; + if (assetRequests === 1) { + req.socket.destroy(); + return; + } + res.writeHead(200, { 'Content-Length': '5' }); + res.end('model'); + }, async (url) => { + await downloadFile(url, destination, undefined, { + maxRetries: 1, + retryDelayMs: 0, + }); + }); + assert.deepEqual(requests, ['/model.bin', '/asset', '/asset']); + assert.equal(fs.readFileSync(destination, 'utf8'), 'model'); + }); +}); + +test('carries one retry budget across redirects', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + const requests = []; + let rootRequests = 0; + let assetRequests = 0; + await withServer((req, res) => { + requests.push(req.url); + if (req.url === '/model.bin') { + rootRequests += 1; + if (rootRequests === 1) { + req.socket.destroy(); + return; + } + res.writeHead(302, { Location: '/asset' }); + res.end(); + return; + } + assetRequests += 1; + if (assetRequests === 1) { + req.socket.destroy(); + return; + } + res.writeHead(200, { 'Content-Length': '5' }); + res.end('model'); + }, async (url) => { + await assert.rejects(downloadFile(url, destination, undefined, { + maxRetries: 1, + retryDelayMs: 0, + })); + }); + assert.deepEqual(requests, ['/model.bin', '/model.bin', '/asset']); + assert.equal(fs.existsSync(destination), false); + }); +}); + +test('does not promote after HEAD ETag mismatch and preserves the old partial if fresh GET fails', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + let requestCount = 0; + await withServer((req, res) => { + requestCount += 1; + if (requestCount === 1) { + res.writeHead(416, { 'Content-Length': '0' }); + res.end(); + } else if (req.method === 'HEAD') { + res.writeHead(200, { 'Content-Length': '3', ETag: '"v2"' }); + res.end(); + } else { + res.writeHead(200, { 'Content-Length': '5', ETag: '"v2"' }); + res.flushHeaders(); + res.write('fr'); + setImmediate(() => res.socket.destroy()); + } + }, async (url) => { + writeResumeState(destination, url, 'old'); + await assert.rejects(downloadFile(url, destination)); + assert.equal(fs.readFileSync(destination + '.part', 'utf8'), 'old'); + assert.equal(fs.existsSync(destination + '.part.meta.json'), true); + assert.equal(fs.existsSync(destination + '.part.fresh'), false); + assert.equal(fs.existsSync(destination), false); + }); + assert.equal(requestCount, 3); + }); +}); + +test('cleans stale fresh output and transactionally replaces a legacy partial', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + let staleFreshWasRemoved; + fs.writeFileSync(destination + '.part', 'legacy'); + fs.writeFileSync(destination + '.part.fresh', 'stale'); + await withServer((_req, res) => { + staleFreshWasRemoved = !fs.existsSync(destination + '.part.fresh'); + res.writeHead(200, { 'Content-Length': '5', ETag: '"v2"' }); + res.end('fresh'); + }, (url) => downloadFile(url, destination)); + assert.equal(staleFreshWasRemoved, true); + assert.equal(fs.readFileSync(destination, 'utf8'), 'fresh'); + assert.equal(fs.existsSync(destination + '.part'), false); + assert.equal(fs.existsSync(destination + '.part.fresh'), false); + }); +}); + +test('rejects a malformed streaming redirect without crashing and closes its socket', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + let socket; + let interval; + let resolveClosed; + const closed = new Promise((resolve) => { resolveClosed = resolve; }); + await withServer((req, res) => { + socket = req.socket; + socket.once('close', () => { + clearInterval(interval); + resolveClosed(); + }); + res.on('error', () => {}); + res.writeHead(302, { Location: 'http://[' }); + res.flushHeaders(); + interval = setInterval(() => res.write(Buffer.alloc(1024)), 2); + }, async (url) => { + await assert.rejects(downloadFile(url, destination), /fetch failed|Invalid URL/); + await expectSocketClosed(closed, () => socket, 'malformed redirect socket stayed open'); + }); + }); +}); + +test('closes a streaming HTTP error response after rejection', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + let socket; + let interval; + let resolveClosed; + const closed = new Promise((resolve) => { resolveClosed = resolve; }); + await withServer((req, res) => { + socket = req.socket; + socket.once('close', () => { + clearInterval(interval); + resolveClosed(); + }); + res.on('error', () => {}); + res.writeHead(500); + res.flushHeaders(); + interval = setInterval(() => res.write(Buffer.alloc(1024)), 2); + }, async (url) => { + await assert.rejects(downloadFile(url, destination), /HTTP 500/); + await expectSocketClosed(closed, () => socket, 'HTTP 500 socket stayed open'); + }); + }); +}); + +test('rejects a mismatched HTTP 206 range without appending to the partial', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + await withServer((_req, res) => { + res.writeHead(206, { + 'Content-Range': 'bytes 5-9/10', + 'Content-Length': '5', + ETag: '"v1"', + }); + res.end('world'); + }, async (url) => { + writeResumeState(destination, url, 'hello '); + await assert.rejects(downloadFile(url, destination), /started at 5, expected 6/); + assert.equal(fs.readFileSync(destination + '.part', 'utf8'), 'hello '); + }); + }); +}); + +test('cancels the response when the output stream fails', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + const originalCreateWriteStream = fs.createWriteStream; + let socket; + let interval; + let resolveClosed; + const closed = new Promise((resolve) => { resolveClosed = resolve; }); + await withServer((req, res) => { + socket = req.socket; + socket.once('close', () => { + clearInterval(interval); + resolveClosed(); + }); + res.on('error', () => {}); + res.writeHead(200, { 'Content-Length': '1000000', ETag: '"v1"' }); + res.flushHeaders(); + interval = setInterval(() => res.write(Buffer.alloc(1024)), 2); + }, async (url) => { + fs.createWriteStream = (...args) => { + const stream = originalCreateWriteStream(...args); + process.nextTick(() => stream.destroy(new Error('simulated write failure'))); + return stream; + }; + try { + await assert.rejects(downloadFile(url, destination), /simulated write failure/); + await expectSocketClosed(closed, () => socket, 'write failure socket stayed open'); + } finally { + fs.createWriteStream = originalCreateWriteStream; + } + }); + }); +}); + +test('closes fresh output before cleanup after an aborted response', async () => { + await withTempDir(async (dir) => { + const destination = path.join(dir, 'model.bin'); + const freshPath = destination + '.part.fresh'; + const originalCreateWriteStream = fs.createWriteStream; + const originalUnlinkSync = fs.unlinkSync; + let freshStream; + let freshWasCleaned = false; + let requestCount = 0; + + fs.writeFileSync(destination + '.part', 'legacy'); + fs.createWriteStream = (...args) => { + const stream = originalCreateWriteStream(...args); + if (args[0] === freshPath) freshStream = stream; + return stream; + }; + fs.unlinkSync = (filePath) => { + if (filePath === freshPath) { + assert.equal(freshStream?.closed, true); + freshWasCleaned = true; + } + return originalUnlinkSync(filePath); + }; + + try { + await withServer((_req, res) => { + requestCount += 1; + res.writeHead(200, { 'Content-Length': '5' }); + res.flushHeaders(); + res.write('fr'); + setImmediate(() => res.socket.destroy()); + }, async (url) => { + await assert.rejects(downloadFile(url, destination)); + }); + } finally { + fs.createWriteStream = originalCreateWriteStream; + fs.unlinkSync = originalUnlinkSync; + } + + assert.equal(requestCount, 1); + assert.equal(freshWasCleaned, true); + assert.equal(fs.existsSync(freshPath), false); + }); +});