From 6deb6cf8a7dc5ad1a7531aeb9dc1c64e0ab49f3a Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Fri, 3 Jul 2026 10:18:52 +0200 Subject: [PATCH 01/18] Add Resource content stream model and stream/fs utils (#386) Co-Authored-By: Claude Fable 5 --- lib/resource.js | 26 +++++++++++++++++ lib/utils/fs.js | 12 +++++++- lib/utils/stream.js | 11 ++++++++ test/unit/resource-test.js | 51 +++++++++++++++++++++++++++++++++- test/unit/utils/stream-test.js | 34 +++++++++++++++++++++++ 5 files changed, 132 insertions(+), 2 deletions(-) create mode 100644 lib/utils/stream.js create mode 100644 test/unit/utils/stream-test.js diff --git a/lib/resource.js b/lib/resource.js index cb94435f..b6a94aa9 100644 --- a/lib/resource.js +++ b/lib/resource.js @@ -1,3 +1,4 @@ +import { Readable } from 'stream'; import types from './config/resource-types.js'; class Resource { @@ -57,6 +58,31 @@ class Resource { this.text = text; } + setContentStream (stream) { + this.contentStream = stream; + } + + /** + * Returns resource content as a Readable stream. + * For buffered (html/css) resources a fresh stream is created on each call. + * For streamed resources the live response stream is returned - it can be consumed only once. + * Returns null when content was already cleared (see clearContent). + */ + getContentStream () { + if (this.contentStream) { + return this.contentStream; + } + if (typeof this.text === 'string') { + return Readable.from(Buffer.from(this.text, this.encoding)); + } + return null; + } + + clearContent () { + this.text = null; + this.contentStream = null; + } + getDepth () { return this.depth; } diff --git a/lib/utils/fs.js b/lib/utils/fs.js index 38618024..d03f97e0 100644 --- a/lib/utils/fs.js +++ b/lib/utils/fs.js @@ -1,5 +1,7 @@ import path from 'path'; import fs from 'fs/promises'; +import { createWriteStream } from 'fs'; +import { pipeline } from 'stream/promises'; async function outputFile (file, data, encoding) { const dir = path.dirname(file); @@ -8,6 +10,14 @@ async function outputFile (file, data, encoding) { return fs.writeFile(file, data, { encoding: encoding }); } +async function outputFileStream (file, stream) { + const dir = path.dirname(file); + await fs.mkdir(dir, { recursive: true}); + + return pipeline(stream, createWriteStream(file)); +} + export { - outputFile + outputFile, + outputFileStream }; diff --git a/lib/utils/stream.js b/lib/utils/stream.js new file mode 100644 index 00000000..07880483 --- /dev/null +++ b/lib/utils/stream.js @@ -0,0 +1,11 @@ +async function readAll (stream, encoding) { + const chunks = []; + for await (const chunk of stream) { + chunks.push(chunk); + } + return Buffer.concat(chunks).toString(encoding); +} + +export { + readAll +}; diff --git a/test/unit/resource-test.js b/test/unit/resource-test.js index 545b0f71..d7875b48 100644 --- a/test/unit/resource-test.js +++ b/test/unit/resource-test.js @@ -1,5 +1,7 @@ import * as chai from 'chai'; -chai.should(); +import { Readable } from 'stream'; +import { text as readStreamText, buffer as readStreamBuffer } from 'stream/consumers'; +const should = chai.should(); import '../../test/utils/assertions.js'; import Resource from '../../lib/resource.js'; @@ -33,4 +35,51 @@ describe('Resource', function() { childOfChild.depth.should.eql(2); }); }); + + describe('#getContentStream', function () { + it('should return null when resource has no content', function () { + const resource = new Resource('http://example.com'); + should.not.exist(resource.getContentStream()); + }); + + it('should return stream set with setContentStream', function () { + const resource = new Resource('http://example.com'); + const stream = Readable.from(Buffer.from('binary data')); + resource.setContentStream(stream); + resource.getContentStream().should.be.equal(stream); + }); + + it('should return fresh stream with text content on each call', async function () { + const resource = new Resource('http://example.com'); + resource.setEncoding('utf8'); + resource.setText('some text'); + + const firstText = await readStreamText(resource.getContentStream()); + const secondText = await readStreamText(resource.getContentStream()); + firstText.should.be.eql('some text'); + secondText.should.be.eql('some text'); + }); + + it('should encode text with resource encoding', async function () { + const resource = new Resource('http://example.com'); + const originalBytes = Buffer.from([0xff, 0x00, 0xab, 0x10]); + resource.setText(originalBytes.toString('binary')); // default encoding is binary + + const streamedBytes = await readStreamBuffer(resource.getContentStream()); + streamedBytes.should.be.eql(originalBytes); + }); + }); + + describe('#clearContent', function () { + it('should clear text and content stream', function () { + const resource = new Resource('http://example.com'); + resource.setText('some text'); + resource.setContentStream(Readable.from('stream')); + + resource.clearContent(); + + should.not.exist(resource.getText()); + should.not.exist(resource.getContentStream()); + }); + }); }); diff --git a/test/unit/utils/stream-test.js b/test/unit/utils/stream-test.js new file mode 100644 index 00000000..c2c84111 --- /dev/null +++ b/test/unit/utils/stream-test.js @@ -0,0 +1,34 @@ +import * as chai from 'chai'; +import { Readable } from 'stream'; +import { readAll } from '../../../lib/utils/stream.js'; +chai.should(); + +describe('utils/stream', function () { + describe('#readAll', function () { + it('should read whole stream to string with given encoding', async function () { + const stream = Readable.from([Buffer.from('hello '), Buffer.from('world')]); + const text = await readAll(stream, 'utf8'); + text.should.be.eql('hello world'); + }); + + it('should keep bytes intact with binary encoding', async function () { + const bytes = Buffer.from([0xff, 0x00, 0xab, 0x10]); + const text = await readAll(Readable.from(bytes), 'binary'); + Buffer.from(text, 'binary').should.be.eql(bytes); + }); + + it('should reject when stream emits error', async function () { + const stream = new Readable({ + read () { + this.destroy(new Error('read error')); + } + }); + try { + await readAll(stream, 'utf8'); + throw new Error('expected readAll to reject'); + } catch (err) { + err.message.should.be.eql('read error'); + } + }); + }); +}); From 958599232b72ff75b1387c066e308450f420bb46 Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Fri, 3 Jul 2026 10:23:14 +0200 Subject: [PATCH 02/18] Rewrite request layer around got.stream with retry support (#386) Resolves at headers time with the body as a Readable stream. New afterResponse contract: {url, statusCode, headers, getBody()}. Co-Authored-By: Claude Fable 5 --- lib/config/defaults.js | 1 - lib/request.js | 175 ++++++++++++------- test/unit/request-test.js | 344 ++++++++++++++++---------------------- 3 files changed, 264 insertions(+), 256 deletions(-) diff --git a/lib/config/defaults.js b/lib/config/defaults.js index 8c731519..32bd9a3b 100644 --- a/lib/config/defaults.js +++ b/lib/config/defaults.js @@ -46,7 +46,6 @@ const config = { ], request: { throwHttpErrors: false, - responseType: 'buffer', //cookieJar: true, decompress: true, headers: { diff --git a/lib/request.js b/lib/request.js index f66ce29c..9002c00e 100644 --- a/lib/request.js +++ b/lib/request.js @@ -1,4 +1,6 @@ import got from 'got'; +import { Readable } from 'stream'; +import { buffer as readStreamToBuffer } from 'stream/consumers'; import logger from './logger.js'; import { extend } from './utils/index.js'; @@ -6,29 +8,13 @@ function getMimeType (contentType) { return contentType ? contentType.split(';')[0] : null; } -function defaultResponseHandler ({response}) { - return Promise.resolve(response); -} - function extractEncodingFromHeader (headers) { const contentTypeHeader = headers['content-type']; return contentTypeHeader && contentTypeHeader.includes('utf-8') ? 'utf8' : 'binary'; } -function getEncoding (response) { - if (response && typeof response === 'object') { - if (response.headers && typeof response.headers === 'object') { - return extractEncodingFromHeader(response.headers); - } else if (response.encoding) { - return response.encoding; - } - } - - return 'binary'; -} - -function throwTypeError (result) { +function throwAfterResponseTypeError (result) { let type = typeof result; if (result instanceof Error) { @@ -37,72 +23,143 @@ function throwTypeError (result) { type = 'array'; } - throw new Error(`Wrong response handler result. Expected string or object, but received ${type}`); + throw new Error(`Wrong afterResponse result. Expected null or object with { body?, encoding?, metadata? }, but received ${type}`); } -function getData (result) { - let data = result; - if (result && typeof result === 'object' && 'body' in result) { - data = result.body; - } - - return data; +/* + * Requests url with got.stream and resolves with { response, stream } when response headers arrive, + * before the body is consumed. Attaching the 'retry' listener is what enables got retries for + * streams - transient errors before/at response time are retried with a fresh stream. + * Mid-body errors after the stream was handed over to the consumer cannot be retried. + */ +function requestWithRetry (requestOptions) { + return new Promise((resolve, reject) => { + const attempt = (stream) => { + const onRetry = (retryCount, error, createRetryStream) => { + stream.off('response', onResponse); + stream.off('error', onError); + logger.debug(`[request] retrying request for ${requestOptions.url}, attempt ${retryCount}, reason: ${error.message}`); + attempt(createRetryStream()); + }; + const onResponse = (response) => { + stream.off('retry', onRetry); + stream.off('error', onError); + resolve({ response, stream }); + }; + const onError = (error) => { + stream.off('retry', onRetry); + stream.off('response', onResponse); + reject(error); + }; + stream.once('retry', onRetry); + stream.once('response', onResponse); + stream.once('error', onError); + }; + attempt(got.stream(requestOptions)); + }); } -function transformResult (result) { - const encoding = getEncoding(result); - const data = getData(result); +async function applyAfterResponse ({ responseData, headers, afterResponse }) { + const originalStream = responseData.stream; + let bufferedBody = null; + + const getBody = async () => { + if (bufferedBody === null) { + bufferedBody = await readStreamToBuffer(originalStream); + } + return bufferedBody; + }; + + let result; + try { + result = await afterResponse({ + response: { + url: responseData.url, + statusCode: responseData.statusCode, + headers, + getBody + } + }); + } catch (err) { + originalStream.destroy(); + throw err; + } - // Check for no data - if (data === null || data === undefined) { + if (result === null || result === undefined) { + originalStream.destroy(); return null; } - // Then stringify it. - let body = null; - if (data instanceof Buffer) { - body = data.toString(encoding); - } else if (typeof data === 'string') { - body = data; - } else { - throwTypeError(result); + if (typeof result !== 'object' || Array.isArray(result) || result instanceof Error) { + originalStream.destroy(); + throwAfterResponseTypeError(result); } - return { - body, - encoding, - metadata: result.metadata || data.metadata || null - }; + const encoding = result.encoding || responseData.encoding; + const metadata = result.metadata || null; + + if ('body' in result) { + let bodyBuffer; + if (Buffer.isBuffer(result.body)) { + bodyBuffer = result.body; + } else if (typeof result.body === 'string') { + bodyBuffer = Buffer.from(result.body, encoding); + } else { + originalStream.destroy(); + throwAfterResponseTypeError(result.body); + } + if (bufferedBody === null) { + originalStream.destroy(); // original body replaced without being read + } + return extend(responseData, { stream: Readable.from(bodyBuffer), encoding, metadata }); + } + + if (bufferedBody !== null) { + // body was read by the action - original stream is consumed, serve the buffered copy + return extend(responseData, { stream: Readable.from(bufferedBody), encoding, metadata }); + } + + return extend(responseData, { encoding, metadata }); } -async function getRequest ({url, referer, options = {}, afterResponse = defaultResponseHandler}) { - const requestOptions = extend(options, {url}); +/* + * Resolves when response headers arrive with + * { url, statusCode, mimeType, encoding, metadata, stream } + * The body is NOT buffered - it is available as a Readable in stream. + * Returns null if the resource should be skipped (afterResponse returned null). + */ +async function getRequest ({ url, referer, options = {}, afterResponse }) { + const requestOptions = extend(options, { url }); + delete requestOptions.responseType; // meaningless for streams if (referer) { - requestOptions.headers = requestOptions.headers || {}; - requestOptions.headers.referer = referer; + requestOptions.headers = extend(requestOptions.headers, { referer }); } logger.debug(`[request] sending request for url ${url}, referer ${referer}`); - const response = await got(requestOptions); + const { response, stream } = await requestWithRetry(requestOptions); logger.debug(`[request] received response for ${response.url}, statusCode ${response.statusCode}`); - const responseHandlerResult = transformResult(await afterResponse({response})); - if (!responseHandlerResult) { - return null; - } - return { + // a body error emitted before the consumer attaches its own handlers must not crash the process + stream.on('error', (err) => logger.debug(`[request] response stream error for ${response.url}: ${err.message}`)); + + const responseData = { url: response.url, + statusCode: response.statusCode, mimeType: getMimeType(response.headers['content-type']), - body: responseHandlerResult.body, - metadata: responseHandlerResult.metadata, - encoding: responseHandlerResult.encoding + encoding: extractEncodingFromHeader(response.headers), + metadata: null, + stream }; + + if (!afterResponse) { + return responseData; + } + + return applyAfterResponse({ responseData, headers: response.headers, afterResponse }); } export default { - get: getRequest, - getEncoding, - transformResult + get: getRequest }; diff --git a/test/unit/request-test.js b/test/unit/request-test.js index b2ee7256..071f023b 100644 --- a/test/unit/request-test.js +++ b/test/unit/request-test.js @@ -1,7 +1,8 @@ import * as chai from 'chai'; -chai.should(); +const should = chai.should(); import sinon from 'sinon'; import nock from 'nock'; +import { text as readStreamText, buffer as readStreamBuffer } from 'stream/consumers'; import request from '../../lib/request.js'; describe('request', () => { @@ -47,244 +48,195 @@ describe('request', () => { }); }); - it('should call afterResponse with correct params', () => { - const url = 'http://example.com'; - const scope = nock(url).get('/').reply(200, 'TEST BODY'); - let handlerStub = sinon.stub().resolves(''); - - return request.get({url, afterResponse: handlerStub}).then(() => { - scope.isDone().should.eql(true); - handlerStub.calledOnce.should.eql(true); - const afterResponseArgs = handlerStub.getCall(0).args[0]; - afterResponseArgs.response.body.should.eql('TEST BODY'); - afterResponseArgs.response.headers.should.eql({}); - }); - }); - - describe('transformResult from afterResponse', () => { - it('should return object with body and metadata properties', () => { - const url = 'http://example.com'; - nock(url).get('/').reply(200, 'TEST BODY'); - const handlerStub = sinon.stub().resolves({ - body: 'a', - metadata: 'b', - encoding: 'utf8' - }); - - return request.get({url, afterResponse: handlerStub}).then((data) => { - data.body.should.eql('a'); - data.metadata.should.eql('b'); - data.encoding.should.eql('utf8'); - }); - }); - - it('should return with metadata == null if metadata is not defined', () => { - const url = 'http://example.com'; - nock(url).get('/').reply(200, 'TEST BODY'); - const handlerStub = sinon.stub().resolves({ - body: 'a' - }); - - return request.get({url, afterResponse: handlerStub}).then((data) => { - data.body.should.eql('a'); - (data.metadata === null).should.be.true; - data.encoding.should.eql('binary'); - }); - }); - - it('should transform string result', () => { - const url = 'http://example.com'; - nock(url).get('/').reply(200, 'TEST BODY'); - const handlerStub = sinon.stub().resolves('test body'); - - return request.get({url, afterResponse: handlerStub}).then((data) => { - data.body.should.eql('test body'); - (data.metadata === null).should.be.true; - }); - }); - - it('should be rejected if wrong result (no string nor object) returned', () => { - const url = 'http://example.com'; - nock(url).get('/').reply(200, 'TEST BODY'); - const handlerStub = sinon.stub().resolves(['1', '2']); - - return request.get({url, afterResponse: handlerStub}).then(() => { - true.should.eql(false); - }).catch((e) => { - e.should.be.instanceOf(Error); - e.message.should.match(/Wrong response handler result. Expected string or object, but received/); - }); - }); - }); - - it('should return object with url, body, mimeType properties', () => { + it('should return object with url, statusCode, mimeType, encoding and stream with the body', async () => { const url = 'http://www.google.com'; nock(url).get('/').reply(200, 'Hello from Google!', { 'content-type': 'text/html; charset=utf-8' }); - return request.get({url}).then((data) => { - data.should.have.property('url'); - data.should.have.property('body'); - data.should.have.property('mimeType'); - data.url.should.eql('http://www.google.com/'); - data.body.should.eql('Hello from Google!'); - data.mimeType.should.eql('text/html'); - data.encoding.should.eql('utf8'); - }); + const data = await request.get({url}); + data.should.have.property('url'); + data.should.have.property('statusCode'); + data.should.have.property('mimeType'); + data.should.have.property('stream'); + data.url.should.eql('http://www.google.com/'); + data.statusCode.should.eql(200); + data.mimeType.should.eql('text/html'); + data.encoding.should.eql('utf8'); + (data.metadata === null).should.be.true; + + const body = await readStreamText(data.stream); + body.should.eql('Hello from Google!'); }); - it('should return mimeType = null if content-type header was not found in response', () => { - let url = 'http://www.google.com'; + it('should return mimeType = null and binary encoding if content-type header was not found in response', async () => { + const url = 'http://www.google.com'; nock(url).get('/').reply(200, 'Hello from Google!', {}); - return request.get({url}).then((data) => { - data.should.include.all.keys(['url', 'body', 'mimeType', 'encoding']); - data.url.should.eql('http://www.google.com/'); - data.body.should.eql('Hello from Google!'); - data.encoding.should.eql('binary'); - data.should.have.property('mimeType', null); - }); + const data = await request.get({url}); + data.url.should.eql('http://www.google.com/'); + data.encoding.should.eql('binary'); + data.should.have.property('mimeType', null); }); -}); -describe('get encoding', () => { - it('should return binary by default', () => { - const result = request.getEncoding(null); + it('should retry transient network errors before response', async () => { + const url = 'http://retry.example.com'; + nock(url).get('/').replyWithError(Object.assign(new Error('connection reset'), {code: 'ECONNRESET'})); + nock(url).get('/').reply(200, 'recovered'); - result.should.eql('binary'); + const data = await request.get({url, options: {retry: {limit: 1}}}); + data.statusCode.should.eql(200); + const body = await readStreamText(data.stream); + body.should.eql('recovered'); }); - it('should return binary when no content-type header supplies', () => { - const result = request.getEncoding({ - headers: {} - }); + it('should reject when request fails without retries left', async () => { + const url = 'http://fail.example.com'; + nock(url).get('/').replyWithError(Object.assign(new Error('connection reset'), {code: 'ECONNRESET'})); - result.should.eql('binary'); + try { + await request.get({url, options: {retry: {limit: 0}}}); + throw new Error('expected request.get to reject'); + } catch (err) { + err.code.should.eql('ECONNRESET'); + } }); - it('should return binary when content type header doesn\'t include utf-8', () => { - const result = request.getEncoding({ - headers: {} - }); + describe('afterResponse', () => { + it('should call afterResponse with url, statusCode, headers and getBody', async () => { + const url = 'http://example.com'; + const scope = nock(url).get('/').reply(200, 'TEST BODY', {'content-type': 'text/html'}); + const handlerStub = sinon.stub().resolves({}); - result.should.eql('binary'); - }); + await request.get({url, afterResponse: handlerStub}); + scope.isDone().should.eql(true); + handlerStub.calledOnce.should.eql(true); - it('should return binary when content type header doesn\'t include utf-8', () => { - const result = request.getEncoding({ - headers: { - 'content-type': 'text/html' - } + const {response} = handlerStub.getCall(0).args[0]; + response.url.should.eql('http://example.com/'); + response.statusCode.should.eql(200); + response.headers.should.have.property('content-type', 'text/html'); + response.getBody.should.be.a('function'); }); - result.should.eql('binary'); - }); + it('should provide body via getBody', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); - it('should return utf8 when content type includes utf-8', () => { - const result = request.getEncoding({ - headers: { - 'content-type': 'text/html; charset=utf-8' - } - }); + let receivedBody; + const afterResponse = async ({response}) => { + receivedBody = await response.getBody(); + return {}; + }; - result.should.eql('utf8'); - }); + const data = await request.get({url, afterResponse}); + receivedBody.should.be.instanceOf(Buffer); + receivedBody.toString().should.eql('TEST BODY'); - it('should return utf8 response object includes it', () => { - const result = request.getEncoding({ - encoding: 'utf8' + // original stream was consumed by getBody - body must still be readable from result + const body = await readStreamText(data.stream); + body.should.eql('TEST BODY'); }); - result.should.eql('utf8'); - }); -}); - -describe('transformResult', () => { - it('should throw with weird shaped response', () => { - try { - request.transformResult([1,2,3]); + it('should memoize getBody', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); - // We shouldn't get here. - true.should.eql(false); - } catch (e) { - e.should.be.instanceOf(Error); - e.message.should.eql('Wrong response handler result. Expected string or object, but received array'); - } - }); + const afterResponse = async ({response}) => { + const first = await response.getBody(); + const second = await response.getBody(); + first.should.be.equal(second); + return {}; + }; - it('should pass through error', () => { - try { - request.transformResult(new Error('Oh no')); + await request.get({url, afterResponse}); + }); - // We shouldn't get here. - true.should.eql(false); - } catch (e) { - e.should.be.instanceOf(Error); - e.message.should.eql('Oh no'); - } - }); + it('should return null and skip resource when action returns null', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const handlerStub = sinon.stub().resolves(null); - it('should throw with boolean input', () => { - try { - request.transformResult(true); + const data = await request.get({url, afterResponse: handlerStub}); + should.not.exist(data); + }); - // We shouldn't get here. - true.should.eql(false); - } catch (e) { - e.should.be.instanceOf(Error); - e.message.should.eql('Wrong response handler result. Expected string or object, but received boolean'); - } - }); + it('should replace body when action returns object with body', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const handlerStub = sinon.stub().resolves({ + body: 'a', + metadata: 'b', + encoding: 'utf8' + }); - it('should handle object', () => { - const result = request.transformResult({ - body: 'SOME BODY', - encoding: 'utf8', - metadata: { foo: 'bar' } + const data = await request.get({url, afterResponse: handlerStub}); + data.metadata.should.eql('b'); + data.encoding.should.eql('utf8'); + const body = await readStreamText(data.stream); + body.should.eql('a'); }); - result.should.have.property('body', 'SOME BODY'); - result.should.have.property('encoding', 'utf8'); - result.should.have.property('metadata').that.eql({ foo: 'bar' }); - }); + it('should support Buffer body', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const bodyBuffer = Buffer.from([0xff, 0x00, 0xab]); + const handlerStub = sinon.stub().resolves({body: bodyBuffer}); - it('should handle object with empty body string', () => { - const result = request.transformResult({ - body: '', - encoding: 'utf8', + const data = await request.get({url, afterResponse: handlerStub}); + (data.metadata === null).should.be.true; + const body = await readStreamBuffer(data.stream); + body.should.eql(bodyBuffer); }); - result.should.have.property('body', ''); - result.should.have.property('encoding', 'utf8'); - result.should.have.property('metadata', null); - }); + it('should keep original body streaming when action returns metadata only', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const handlerStub = sinon.stub().resolves({metadata: {foo: 'bar'}}); - it('should handle object with defaults and buffer body', () => { - const result = request.transformResult({ - body: Buffer.from('SOME BODY'), + const data = await request.get({url, afterResponse: handlerStub}); + data.metadata.should.eql({foo: 'bar'}); + const body = await readStreamText(data.stream); + body.should.eql('TEST BODY'); }); - result.should.have.property('body', 'SOME BODY'); - result.should.have.property('encoding', 'binary'); - result.should.have.property('metadata', null); - }); + it('should be rejected if wrong result (no null nor object) returned', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const handlerStub = sinon.stub().resolves(['1', '2']); - it('should handle raw string input', () => { - const result = request.transformResult('SOME BODY'); + try { + await request.get({url, afterResponse: handlerStub}); + throw new Error('expected request.get to reject'); + } catch (e) { + e.message.should.match(/Wrong afterResponse result. Expected null or object.*but received array/); + } + }); - result.should.have.property('body', 'SOME BODY'); - result.should.have.property('encoding', 'binary'); - result.should.have.property('metadata', null); - }); + it('should be rejected if string returned (removed in v7)', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const handlerStub = sinon.stub().resolves('test body'); - it('should handle null input', () => { - const result = request.transformResult(null); - (result === null).should.be.true; - }); + try { + await request.get({url, afterResponse: handlerStub}); + throw new Error('expected request.get to reject'); + } catch (e) { + e.message.should.match(/Wrong afterResponse result.*but received string/); + } + }); - it('should handle undefined input', () => { - const result = request.transformResult(undefined); - (result === null).should.be.true; + it('should be rejected with error thrown by action', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const handlerStub = sinon.stub().rejects(new Error('action failed')); + + try { + await request.get({url, afterResponse: handlerStub}); + throw new Error('expected request.get to reject'); + } catch (e) { + e.message.should.eql('action failed'); + } + }); }); }); From 7395e6185e1e5d4c0f7c1c4f9d4f19ad8653c7fc Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Fri, 3 Jul 2026 10:29:01 +0200 Subject: [PATCH 03/18] Rework scraper lifecycle to stream resources to storage (#386) Binary resources are saved immediately at response time while holding the request queue slot; html/css are buffered for link rewriting and their content is freed after save. Default FS plugin pipes streams to disk. Co-Authored-By: Claude Fable 5 --- lib/plugins/save-resource-to-fs-plugin.js | 10 +- lib/scraper.js | 117 ++++++++++++------ test/functional/base/check-it-works.js | 4 +- .../after-response-action.test.js | 2 +- test/unit/scraper-init-test.js | 2 - test/unit/scraper-test.js | 18 ++- 6 files changed, 104 insertions(+), 49 deletions(-) diff --git a/lib/plugins/save-resource-to-fs-plugin.js b/lib/plugins/save-resource-to-fs-plugin.js index 19b2d1d4..894c8d26 100644 --- a/lib/plugins/save-resource-to-fs-plugin.js +++ b/lib/plugins/save-resource-to-fs-plugin.js @@ -1,6 +1,6 @@ import path from 'path'; import fs from 'fs'; -import { outputFile } from '../utils/fs.js'; +import { outputFileStream } from '../utils/fs.js'; class SaveResourceToFileSystemPlugin { apply (registerAction) { @@ -20,8 +20,12 @@ class SaveResourceToFileSystemPlugin { registerAction('saveResource', async ({resource}) => { const filename = path.join(absoluteDirectoryPath, resource.getFilename()); - const text = resource.getText(); - await outputFile(filename, text, resource.getEncoding()); + try { + await outputFileStream(filename, resource.getContentStream()); + } catch (err) { + await fs.promises.rm(filename, {force: true}); // don't leave partially written files + throw err; + } loadedResources.push(resource); }); diff --git a/lib/scraper.js b/lib/scraper.js index ff2d810d..591f09c8 100644 --- a/lib/scraper.js +++ b/lib/scraper.js @@ -14,6 +14,7 @@ import { import * as utils from './utils/index.js'; const { extend, union, urlsEqual, getTypeByMime, getTypeByFilename, series } = utils; +import { readAll } from './utils/stream.js'; import NormalizedUrlMap from './utils/normalized-url-map.js'; const actionNames = [ @@ -122,15 +123,38 @@ class Scraper { } } - async saveResource (resource) { + async persistResource (resource) { resource.setSaved(); try { await this.resourceHandler.handleResource(resource); + + if (resource.contentStream && this.actions.saveResource.length > 1) { + // live stream can be consumed only once - buffer content + // so each saveResource action can read it + resource.setText(await readAll(resource.contentStream, 'binary')); + resource.setEncoding('binary'); + resource.setContentStream(null); + } + logger.info('saving resource ' + resource + ' to fs'); await this.runActions('saveResource', {resource}); - // ignore promise here, just notifying external code about resource saved - this.runActions('onResourceSaved', {resource}); + } finally { + const stream = resource.contentStream; + if (stream && !stream.readableEnded) { + // action resolved without consuming the stream - release the socket + stream.destroy(); + } + resource.clearContent(); + } + + // ignore promise here, just notifying external code about resource saved + this.runActions('onResourceSaved', {resource}); + } + + async saveResource (resource) { + try { + await this.persistResource(resource); } catch (err) { logger.warn('failed to save resource ' + resource); await this.handleError(err, resource); @@ -141,54 +165,75 @@ class Scraper { const self = this; const url = resource.getUrl(); + // the queue task encompasses response headers and full body consumption + // (buffering for html/css, streaming to storage for other types) + // so that requestConcurrency bounds concurrent transfers const requestPromise = Promise.resolve() - .then(async () => { + .then(() => { const referer = resource.parent ? resource.parent.getUrl() : null; - return this.requestQueue.add(async () => { - const {requestOptions} = await this.runActions('beforeRequest', {resource, requestOptions: this.options.request}); - return request.get({ + return this.requestQueue.add(async function requestCompleted () { + const {requestOptions} = await self.runActions('beforeRequest', {resource, requestOptions: self.options.request}); + const responseData = await request.get({ url, referer, options: requestOptions, - afterResponse: this.actions.afterResponse.length ? this.runActions.bind(this, 'afterResponse') : undefined + afterResponse: self.actions.afterResponse.length ? self.runActions.bind(self, 'afterResponse') : undefined }); - }); - }).then(async function requestCompleted (responseData) { - if (!responseData) { - logger.debug('no response returned for url ' + url); - return null; - } - - if (!urlsEqual(responseData.url, url)) { // Url may be changed in redirects - logger.debug('url changed. old url = ' + url + ', new url = ' + responseData.url); - if (self.requestedResourcePromises.has(responseData.url)) { - return self.requestedResourcePromises.get(responseData.url); + if (!responseData) { + logger.debug('no response returned for url ' + url); + return null; } - resource.setUrl(responseData.url); - self.requestedResourcePromises.set(responseData.url, requestPromise); - } + if (!urlsEqual(responseData.url, url)) { // Url may be changed in redirects + logger.debug('url changed. old url = ' + url + ', new url = ' + responseData.url); - resource.setType(getTypeByMime(responseData.mimeType)); + if (self.requestedResourcePromises.has(responseData.url)) { + responseData.stream.destroy(); // duplicate of an already requested resource - release the socket + // awaiting the other resource's promise here would deadlock at low concurrency + // because it may still be waiting for this queue slot - resolve it outside the queue task + return {redirectedTo: responseData.url}; + } - const { filename } = await self.runActions('generateFilename', { resource, responseData }); - resource.setFilename(filename); + resource.setUrl(responseData.url); + self.requestedResourcePromises.set(responseData.url, requestPromise); + } - // if type was not determined by mime we can try to get it from filename after it was generated - if (!resource.getType()) { - resource.setType(getTypeByFilename(filename)); - } + resource.setType(getTypeByMime(responseData.mimeType)); - if (responseData.metadata) { - resource.setMetadata(responseData.metadata); - } + const { stream, ...responseInfo } = responseData; + const { filename } = await self.runActions('generateFilename', { resource, responseData: responseInfo }); + resource.setFilename(filename); - resource.setEncoding(responseData.encoding); - resource.setText(responseData.body); + // if type was not determined by mime we can try to get it from filename after it was generated + if (!resource.getType()) { + resource.setType(getTypeByFilename(filename)); + } + + if (responseData.metadata) { + resource.setMetadata(responseData.metadata); + } - self.loadResource(resource); // Add resource to list for future downloading, see Scraper.waitForLoad - return resource; + resource.setEncoding(responseData.encoding); + + if (resource.isHtml() || resource.isCss()) { + // html and css need full text to find and update links to child resources, + // they are saved after their children are resolved - see Scraper.waitForLoad + resource.setText(await readAll(stream, responseData.encoding)); + self.loadResource(resource); + } else { + // other types need no modification - stream them to storage immediately + resource.setContentStream(stream); + self.loadResource(resource); + await self.persistResource(resource); + } + return resource; + }); + }).then(function resolveRedirectDuplicate (result) { + if (result && result.redirectedTo) { + return self.requestedResourcePromises.get(result.redirectedTo); + } + return result; }).catch(function handleError (err) { logger.error('failed to request resource ' + resource); return self.handleError(err, resource); diff --git a/test/functional/base/check-it-works.js b/test/functional/base/check-it-works.js index 758b8af8..8ff3b11b 100644 --- a/test/functional/base/check-it-works.js +++ b/test/functional/base/check-it-works.js @@ -31,7 +31,9 @@ describe('Functional: check it works', function() { return scrape(options).then((result) => { result[0].url.should.be.eql('http://example.com/'); result[0].filename.should.be.eql('index.html'); - result[0].text.should.be.eql('TEST PROMISES'); + // content is not retained in memory after save - read the saved file instead + (result[0].text === null).should.be.true; + fs.readFileSync(testDirname + '/index.html').toString().should.be.eql('TEST PROMISES'); }); }); }); diff --git a/test/functional/request-response-customizations/after-response-action.test.js b/test/functional/request-response-customizations/after-response-action.test.js index 55a33cb5..be4036fe 100644 --- a/test/functional/request-response-customizations/after-response-action.test.js +++ b/test/functional/request-response-customizations/after-response-action.test.js @@ -32,8 +32,8 @@ describe('Functional: afterResponse action in plugin', function() { if (response.statusCode === 404) { return null; } else { + // no body returned - the resource keeps streaming return { - body: response.body, metadata: { headers: response.headers, someOtherData: [ 1, 2, 3 ] diff --git a/test/unit/scraper-init-test.js b/test/unit/scraper-init-test.js index d74605d8..1efc89b9 100644 --- a/test/unit/scraper-init-test.js +++ b/test/unit/scraper-init-test.js @@ -123,7 +123,6 @@ describe('Scraper initialization', function () { s.options.request.should.deep.include({ throwHttpErrors: false, - responseType: 'buffer', decompress: true, https: { rejectUnauthorized: false @@ -145,7 +144,6 @@ describe('Scraper initialization', function () { s.options.request.should.deep.include({ throwHttpErrors: true, - responseType: 'buffer', decompress: true, https: { rejectUnauthorized: false diff --git a/test/unit/scraper-test.js b/test/unit/scraper-test.js index 2b9477b0..62a568d0 100644 --- a/test/unit/scraper-test.js +++ b/test/unit/scraper-test.js @@ -1,5 +1,5 @@ import * as chai from 'chai'; -chai.should(); +const should = chai.should(); import sinon from 'sinon'; import nock from 'nock'; import fs from 'fs'; @@ -96,6 +96,7 @@ describe('Scraper', () => { urlFilter: () => { return true; }, plugins: [ new GenerateFilenamePlugin() ] }); + sinon.stub(s, 'persistResource').resolves(); const r = new Resource('http://example.com/a.png'); r.getDepth = sinon.stub().returns(2); @@ -105,7 +106,8 @@ describe('Scraper', () => { rr.should.be.eql(r); rr.getUrl().should.be.eql('http://example.com/a.png'); rr.getFilename().should.not.be.empty; - rr.getText().should.not.be.empty; + should.exist(rr.getContentStream()); + s.persistResource.calledOnce.should.be.eql(true); }); it('should return null if the urlFilter returns false', async () =>{ @@ -131,6 +133,7 @@ describe('Scraper', () => { urlFilter: () => false, plugins: [ new GenerateFilenamePlugin() ] }); + sinon.stub(s, 'persistResource').resolves(); const r = new Resource('http://example.com'); r.getDepth = sinon.stub().returns(0); @@ -140,7 +143,7 @@ describe('Scraper', () => { rr.should.be.eql(r); rr.getUrl().should.be.eql('http://example.com'); rr.getFilename().should.not.be.empty; - rr.getText().should.not.be.empty; + should.exist(rr.getContentStream()); }); }); @@ -153,6 +156,7 @@ describe('Scraper', () => { directory: testDirname, plugins: [ new GenerateFilenamePlugin() ] }); + sinon.stub(s, 'persistResource').resolves(); const r = new Resource('http://example.com/a.png'); r.getDepth = sinon.stub().returns(212); @@ -162,7 +166,7 @@ describe('Scraper', () => { rr.should.be.eql(r); rr.getUrl().should.be.eql('http://example.com/a.png'); rr.getFilename().should.not.be.empty; - rr.getText().should.not.be.empty; + should.exist(rr.getContentStream()); }); it('should request the resource if maxDepth is set and resource depth is less than maxDept', async () =>{ @@ -174,6 +178,7 @@ describe('Scraper', () => { maxDepth: 3, plugins: [ new GenerateFilenamePlugin() ] }); + sinon.stub(s, 'persistResource').resolves(); const r = new Resource('http://example.com/a.png'); r.getDepth = sinon.stub().returns(2); @@ -183,7 +188,7 @@ describe('Scraper', () => { rr.should.be.eql(r); rr.getUrl().should.be.eql('http://example.com/a.png'); rr.getFilename().should.not.be.empty; - rr.getText().should.not.be.empty; + should.exist(rr.getContentStream()); }); it('should request the resource if maxDepth is set and resource depth is equal to maxDept', async () =>{ @@ -195,6 +200,7 @@ describe('Scraper', () => { maxDepth: 3, plugins: [ new GenerateFilenamePlugin() ] }); + sinon.stub(s, 'persistResource').resolves(); const r = new Resource('http://example.com/a.png'); r.getDepth = sinon.stub().returns(3); @@ -203,7 +209,7 @@ describe('Scraper', () => { rr.should.be.eql(r); rr.getUrl().should.be.eql('http://example.com/a.png'); rr.getFilename().should.not.be.empty; - rr.getText().should.not.be.empty; + should.exist(rr.getContentStream()); }); it('should return null if maxDepth is set and resource depth is greater than maxDepth', async () =>{ From 74bc67443735771352a612b239f6670d20eae1bc Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Fri, 3 Jul 2026 10:33:12 +0200 Subject: [PATCH 04/18] Add streaming test suites and FS plugin unit tests (#386) Co-Authored-By: Claude Fable 5 --- test/functional/streaming/streaming.test.js | 243 ++++++++++++++++++++ test/unit/plugins.test.js | 108 +++++++++ 2 files changed, 351 insertions(+) create mode 100644 test/functional/streaming/streaming.test.js diff --git a/test/functional/streaming/streaming.test.js b/test/functional/streaming/streaming.test.js new file mode 100644 index 00000000..ad4498c9 --- /dev/null +++ b/test/functional/streaming/streaming.test.js @@ -0,0 +1,243 @@ +import * as chai from 'chai'; +const should = chai.should(); +import '../../utils/assertions.js'; +import nock from 'nock'; +import fs from 'fs'; +import http from 'http'; +import { buffer as readStreamBuffer } from 'stream/consumers'; +import scrape from 'website-scraper'; + +const testDirname = './test/functional/streaming/.tmp'; + +// deterministic multi-MB binary payload +function generateBigBinary (sizeInBytes) { + const chunk = Buffer.alloc(64 * 1024); + for (let i = 0; i < chunk.length; i++) { + chunk[i] = (i * 7 + 13) % 256; + } + return Buffer.concat(Array.from({length: Math.ceil(sizeInBytes / chunk.length)}, () => chunk), sizeInBytes); +} + +describe('Functional: streaming resources to storage', function () { + + beforeEach(function () { + nock.cleanAll(); + nock.disableNetConnect(); + }); + + afterEach(function () { + nock.cleanAll(); + nock.enableNetConnect(); + fs.rmSync(testDirname, {recursive: true, force: true}); + }); + + it('should save a large binary resource byte-identical', async function () { + const pngBody = generateBigBinary(5 * 1024 * 1024); + nock('http://example.com/').get('/').reply(200, '', {'content-type': 'text/html'}); + nock('http://example.com/').get('/a.png').reply(200, pngBody, {'content-type': 'image/png'}); + + await scrape({ + urls: ['http://example.com/'], + directory: testDirname + }); + + const savedBytes = fs.readFileSync(testDirname + '/images/a.png'); + savedBytes.equals(pngBody).should.be.eql(true); + }); + + it('should free resource content from memory after save', async function () { + nock('http://example.com/').get('/').reply(200, '', {'content-type': 'text/html'}); + nock('http://example.com/').get('/a.png').reply(200, 'image content', {'content-type': 'image/png'}); + + const result = await scrape({ + urls: ['http://example.com/'], + directory: testDirname + }); + + (result[0].getText() === null).should.be.true; + should.not.exist(result[0].getContentStream()); + + const child = result[0].children.find((c) => c.url === 'http://example.com/a.png'); + (child.getText() === null).should.be.true; + should.not.exist(child.getContentStream()); + + // content is on disk instead + fs.readFileSync(testDirname + '/images/a.png').toString().should.be.eql('image content'); + }); + + it('should save body returned from afterResponse (buffered path)', async function () { + nock('http://example.com/').get('/a.png').reply(200, 'original content', {'content-type': 'image/png'}); + + class ReplaceBodyPlugin { + apply (add) { + add('afterResponse', async ({response}) => { + const body = await response.getBody(); + return { body: Buffer.concat([body, Buffer.from(' + replaced')]) }; + }); + } + } + + await scrape({ + urls: [{url: 'http://example.com/a.png', filename: 'a.png'}], + directory: testDirname, + plugins: [new ReplaceBodyPlugin()] + }); + + fs.readFileSync(testDirname + '/images/a.png').toString().should.be.eql('original content + replaced'); + }); + + it('should keep resource streaming and attach metadata when afterResponse returns no body', async function () { + const pngBody = generateBigBinary(1024 * 1024); + nock('http://example.com/').get('/a.png').reply(200, pngBody, {'content-type': 'image/png'}); + + class MetadataPlugin { + apply (add) { + add('afterResponse', ({response}) => { + return { metadata: { statusCode: response.statusCode } }; + }); + } + } + + const result = await scrape({ + urls: [{url: 'http://example.com/a.png', filename: 'a.png'}], + directory: testDirname, + plugins: [new MetadataPlugin()] + }); + + result[0].metadata.should.be.eql({statusCode: 200}); + fs.readFileSync(testDirname + '/images/a.png').equals(pngBody).should.be.eql(true); + }); + + it('should provide full content to multiple saveResource actions', async function () { + const pngBody = generateBigBinary(1024 * 1024); + nock('http://example.com/').get('/a.png').reply(200, pngBody, {'content-type': 'image/png'}); + + const saved = []; + class MultiStoragePlugin { + apply (add) { + add('saveResource', async ({resource}) => { + saved.push(await readStreamBuffer(resource.getContentStream())); + }); + add('saveResource', async ({resource}) => { + saved.push(await readStreamBuffer(resource.getContentStream())); + }); + } + } + + await scrape({ + urls: [{url: 'http://example.com/a.png', filename: 'a.png'}], + directory: testDirname, + plugins: [new MultiStoragePlugin()] + }); + + saved.should.have.length(2); + saved[0].equals(pngBody).should.be.eql(true); + saved[1].equals(pngBody).should.be.eql(true); + }); + + it('should complete scraping when saveResource action does not consume the stream', async function () { + nock('http://example.com/').get('/a.png').reply(200, generateBigBinary(1024 * 1024), {'content-type': 'image/png'}); + + const seenResources = []; + class IgnoreContentPlugin { + apply (add) { + add('saveResource', async ({resource}) => { + seenResources.push(resource.url); + }); + } + } + + await scrape({ + urls: [{url: 'http://example.com/a.png', filename: 'a.png'}], + directory: testDirname, + plugins: [new IgnoreContentPlugin()] + }); + + seenResources.should.be.eql(['http://example.com/a.png']); + }); + + // nock cannot simulate a connection dying mid-body, so these tests use a real local server + describe('mid-stream failures', function () { + let server, baseUrl; + + beforeEach(function (done) { + nock.enableNetConnect(/127\.0\.0\.1|localhost/); + server = http.createServer((req, res) => { + switch (req.url) { + case '/': + res.writeHead(200, {'content-type': 'text/html'}); + res.end(''); + break; + case '/fail.png': + res.writeHead(200, {'content-type': 'image/png'}); + res.write('partial data'); + setTimeout(() => res.destroy(), 10); + break; + default: + res.writeHead(200, {'content-type': 'image/png'}); + res.end('fine content'); + } + }); + server.listen(0, '127.0.0.1', () => { + baseUrl = `http://127.0.0.1:${server.address().port}`; + done(); + }); + }); + + afterEach(function (done) { + server.close(done); + }); + + it('should continue scraping and leave no partial file when child fails mid-stream with ignoreErrors', async function () { + await scrape({ + urls: [baseUrl + '/'], + directory: testDirname, + request: {retry: {limit: 0}}, + ignoreErrors: true + }); + + fs.existsSync(testDirname + '/index.html').should.be.eql(true); + fs.existsSync(testDirname + '/images/fail.png').should.be.eql(false); // failed, partial file removed + fs.readFileSync(testDirname + '/images/fine.png').toString().should.be.eql('fine content'); + }); + + it('should reject and remove directory when a root resource fails mid-stream without ignoreErrors', async function () { + try { + await scrape({ + urls: [ + {url: baseUrl + '/fine.png', filename: 'fine.png'}, + {url: baseUrl + '/fail.png', filename: 'fail.png'} + ], + directory: testDirname, + request: {retry: {limit: 0}}, + ignoreErrors: false + }); + throw new Error('expected scrape to reject'); + } catch (err) { + err.message.should.not.be.eql('expected scrape to reject'); + } + + fs.existsSync(testDirname).should.be.eql(false); + }); + }); + + it('should handle redirect to an already requested resource with requestConcurrency 1', async function () { + this.timeout(3000); // a deadlock in the request queue would exceed this + + nock('http://example.com/').get('/').reply(200, + '', {'content-type': 'text/html'}); + nock('http://example.com/').get('/a.png').reply(200, 'a content', {'content-type': 'image/png'}); + nock('http://example.com/').get('/b.png').reply(301, undefined, {location: 'http://example.com/a.png'}); + nock('http://example.com/').get('/a.png').reply(200, 'a content', {'content-type': 'image/png'}); + + await scrape({ + urls: ['http://example.com/'], + directory: testDirname, + requestConcurrency: 1 + }); + + fs.readFileSync(testDirname + '/images/a.png').toString().should.be.eql('a content'); + const html = fs.readFileSync(testDirname + '/index.html').toString(); + html.should.contain('src="images/a.png"'); + }); +}); diff --git a/test/unit/plugins.test.js b/test/unit/plugins.test.js index e69de29b..60540ee7 100644 --- a/test/unit/plugins.test.js +++ b/test/unit/plugins.test.js @@ -0,0 +1,108 @@ +import * as chai from 'chai'; +chai.should(); +import fs from 'fs'; +import { Readable } from 'stream'; +import SaveResourceToFileSystemPlugin from '../../lib/plugins/save-resource-to-fs-plugin.js'; +import Resource from '../../lib/resource.js'; + +const testDirname = './test/unit/.plugins-test'; + +function applyPlugin (plugin) { + const actions = {}; + plugin.apply((name, handler) => actions[name] = handler); + return actions; +} + +describe('SaveResourceToFileSystemPlugin', () => { + afterEach(() => { + fs.rmSync(testDirname, {recursive: true, force: true}); + }); + + describe('beforeStart', () => { + it('should throw if directory is not a string', () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + (() => actions.beforeStart({options: {directory: null}})).should.throw('Incorrect directory'); + (() => actions.beforeStart({options: {directory: 42}})).should.throw('Incorrect directory'); + }); + + it('should throw if directory already exists', () => { + fs.mkdirSync(testDirname, {recursive: true}); + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + (() => actions.beforeStart({options: {directory: testDirname}})).should.throw(/Directory (.*?) exists/); + }); + }); + + describe('saveResource', () => { + it('should save content stream to file creating nested directories', async () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + actions.beforeStart({options: {directory: testDirname}}); + + const resource = new Resource('http://example.com/a.png', 'images/nested/a.png'); + const content = Buffer.from([0xff, 0x00, 0xab, 0x10]); + resource.setContentStream(Readable.from(content)); + + await actions.saveResource({resource}); + + fs.readFileSync(testDirname + '/images/nested/a.png').should.be.eql(content); + }); + + it('should save buffered text resources using resource encoding', async () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + actions.beforeStart({options: {directory: testDirname}}); + + const resource = new Resource('http://example.com/', 'index.html'); + resource.setEncoding('utf8'); + resource.setText('тест'); + + await actions.saveResource({resource}); + + fs.readFileSync(testDirname + '/index.html').toString('utf8').should.be.eql('тест'); + }); + + it('should remove partially written file and rethrow when stream errors', async () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + actions.beforeStart({options: {directory: testDirname}}); + + const resource = new Resource('http://example.com/a.png', 'a.png'); + resource.setContentStream(new Readable({ + read () { + this.push('partial data'); + this.destroy(new Error('mid-stream failure')); + } + })); + + try { + await actions.saveResource({resource}); + throw new Error('expected saveResource to reject'); + } catch (err) { + err.message.should.be.eql('mid-stream failure'); + } + fs.existsSync(testDirname + '/a.png').should.be.eql(false); + }); + }); + + describe('error', () => { + it('should remove directory if resources were saved', async () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + actions.beforeStart({options: {directory: testDirname}}); + + const resource = new Resource('http://example.com/', 'index.html'); + resource.setContentStream(Readable.from('content')); + await actions.saveResource({resource}); + fs.existsSync(testDirname).should.be.eql(true); + + await actions.error(); + + fs.existsSync(testDirname).should.be.eql(false); + }); + + it('should not throw if nothing was saved', async () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + actions.beforeStart({options: {directory: testDirname}}); + + await actions.error(); + + fs.existsSync(testDirname).should.be.eql(false); + }); + }); +}); From a3c6754fea7cf423caf34f2dcddbb2d32e33b2fa Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Fri, 3 Jul 2026 10:38:53 +0200 Subject: [PATCH 05/18] Add memory benchmark, document v7 streaming changes, bump to 7.0.0 (#386) Benchmark (100 files x 20 MB, concurrency 8): - v6.0.0: peak RSS 2274 MB - v7.0.0: peak RSS 160 MB Co-Authored-By: Claude Fable 5 --- MIGRATION.md | 55 ++++++++++++++++++++++++++ README.md | 50 +++++++++++++++++------- benchmark/memory.mjs | 91 ++++++++++++++++++++++++++++++++++++++++++++ package.json | 3 +- 4 files changed, 185 insertions(+), 14 deletions(-) create mode 100644 benchmark/memory.mjs diff --git a/MIGRATION.md b/MIGRATION.md index e7697d4d..ba971aef 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -1,5 +1,60 @@ # Migration Guide +## From version 6 to 7 + +Version 7 stops storing resource content in memory ([#386](https://github.com/website-scraper/node-website-scraper/issues/386)). Resources which need no modification (images, fonts, media, scripts - everything except html and css) are streamed directly from the network to storage, and html/css content is freed as soon as the resource is saved. Memory usage no longer grows with the size of the scraped website. + +#### saveResource action + +Resource content is now exposed as a Readable stream via `resource.getContentStream()` instead of `resource.getText()`. For streamed resources the stream can be consumed only once - consume or destroy it within the action. + +```javascript +// before +registerAction('saveResource', async ({resource}) => { + await saveItSomewhere(resource.getFilename(), resource.getText()); +}); + +// after +import { buffer } from 'stream/consumers'; +registerAction('saveResource', async ({resource}) => { + await saveItSomewhere(resource.getFilename(), await buffer(resource.getContentStream())); +}); +// or pipe resource.getContentStream() to your storage to avoid buffering +``` + +#### afterResponse action + +The action is called when response headers arrive, before the body is downloaded. It receives `{response}` where response is `{url, statusCode, headers, getBody()}` instead of the full got response. Returning a string is not supported anymore - return an object or null. Return an object without `body` to keep the resource streaming (no buffering). + +```javascript +// before +registerAction('afterResponse', ({response}) => { + if (response.statusCode === 404) return null; + return { body: response.body, metadata: { headers: response.headers } }; +}); + +// after +registerAction('afterResponse', ({response}) => { + if (response.statusCode === 404) return null; + return { metadata: { headers: response.headers } }; +}); +// call `await response.getBody()` only if you need to inspect or replace the body +``` + +#### scrape() result + +Resource content is not retained after save: `resource.getText()` returns `null` in the result and in `onResourceSaved`. Read saved files from your storage instead. + +#### generateFilename action + +`responseData` no longer contains `body` - filenames are generated from headers (`url`, `statusCode`, `mimeType`, `encoding`, `metadata`) before the body is consumed. + +#### Other behavior changes + +* Binary files are written byte-for-byte from the network (previously the content was round-tripped through a latin1 string). Output for correctly-served files is unchanged; edge-case encodings may differ. +* Transient request errors are retried as before, but only until response headers are received. A connection which dies mid-body is not retried. +* `responseType` in the `request` option is ignored (responses are streamed). + ## From version 4 to 5 #### ESM module diff --git a/README.md b/README.md index 9a01a5a3..a2d7befc 100644 --- a/README.md +++ b/README.md @@ -119,7 +119,7 @@ The difference between [maxRecursiveDepth](#maxRecursiveDepth) and [maxDepth](#m only html resources with depth 2 will be filtered out, last image will be downloaded #### request -Object, custom options for http module [got](https://github.com/sindresorhus/got#options) which is used inside website-scraper. Allows to set retries, cookies, userAgent, encoding, etc. +Object, custom options for http module [got](https://github.com/sindresorhus/got#options) which is used inside website-scraper. Allows to set retries, cookies, userAgent, encoding, etc. Note that `responseType` is controlled by the scraper (responses are streamed) and is ignored if passed here. ```javascript // use same request options for all resources scrape({ @@ -244,7 +244,7 @@ class MyPlugin { registerAction('afterFinish', async () => {}); registerAction('error', async ({error}) => {console.error(error)}); registerAction('beforeRequest', async ({resource, requestOptions}) => ({requestOptions})); - registerAction('afterResponse', async ({response}) => response.body); + registerAction('afterResponse', async ({response}) => ({metadata: {headers: response.headers}})); registerAction('onResourceSaved', ({resource}) => {}); registerAction('onResourceError', ({resource, error}) => {}); registerAction('saveResource', async ({resource}) => {}); @@ -315,15 +315,19 @@ registerAction('beforeRequest', async ({resource, requestOptions}) => { ``` ##### afterResponse -Action afterResponse is called after each response, it allows to customize resource or reject its saving. +Action afterResponse is called after response headers were received, before the body is consumed. It allows to customize resource or reject its saving. Parameters - object which includes: -* response - response object from http module [got](https://github.com/sindresorhus/got#response) +* response - object which includes: + * `url` (string) - final url of the response (after redirects) + * `statusCode` (number) + * `headers` (object) + * `getBody()` - async function which buffers the response body and resolves with a `Buffer`. Note that calling it loads the whole body into memory - don't call it if you only need headers or status code, so the resource keeps streaming to storage Return resolved `Promise` with: - * object if the resource should be saved, object should contain next properties: - * `body` (string, required) - * `encoding` (`binary` or `utf8`) is used to save the file, binary is used by default. + * object if the resource should be saved, object may contain next properties: + * `body` (string or Buffer, optional) - replaces resource content; if omitted, the original body is used without buffering it in memory + * `encoding` (`binary` or `utf8`) is used to decode a string `body`, binary is used by default. * `metadata` (object) - everything you want to save for this resource (like headers, original text, timestamps, etc.), scraper will not use this field at all, it is only for the result * or null if the resource should be skipped @@ -335,8 +339,6 @@ registerAction('afterResponse', ({response}) => { return null; } else { return { - body: response.body, - encoding: 'utf8', metadata: { headers: response.headers, someOtherData: [ 1, 2, 3 ] @@ -344,6 +346,15 @@ registerAction('afterResponse', ({response}) => { } } }); + +// Modify html body before it is processed +registerAction('afterResponse', async ({response}) => { + if (response.headers['content-type']?.includes('text/html')) { + const body = await response.getBody(); + return { body: transformHtml(body.toString()), encoding: 'utf8' }; + } + return {}; // everything else keeps streaming to storage +}); ``` ##### onResourceSaved @@ -374,7 +385,7 @@ Action generateFilename is called to determine path in file system where the res Parameters - object which includes: * resource - [Resource](https://github.com/website-scraper/node-website-scraper/blob/master/lib/resource.js) object -* responseData - object returned from afterResponse action, contains `url`, `mimeType`, `body`, `metadata` properties +* responseData - object which contains `url`, `statusCode`, `mimeType`, `encoding`, `metadata` properties. Note that it does not contain the response body - filenames are generated before the body is consumed Should return object which includes: * filename - String, relative to `directory` path for specified resource @@ -419,12 +430,23 @@ Action saveResource is called to save file to some storage. Use it to save files Parameters - object which includes: * resource - [Resource](https://github.com/website-scraper/node-website-scraper/blob/master/lib/resource.js) object -If multiple actions `saveResource` added - resource will be saved to multiple storages. +Resource content is exposed as a Readable stream returned by `resource.getContentStream()`. For resources which are streamed directly from the network (all types except html and css) the stream can be consumed only once - consume it (or destroy it) within the action. + +If multiple actions `saveResource` added - resource will be saved to multiple storages (the scraper buffers streamed content in that case so every action can read it). ```javascript +import { pipeline } from 'stream/promises'; + registerAction('saveResource', async ({resource}) => { const filename = resource.getFilename(); - const text = resource.getText(); - await saveItSomewhere(filename, text); + await pipeline(resource.getContentStream(), createMyStorageWriteStream(filename)); +}); + +// or, if your storage needs the whole content at once +import { buffer } from 'stream/consumers'; + +registerAction('saveResource', async ({resource}) => { + const content = await buffer(resource.getContentStream()); + await saveItSomewhere(resource.getFilename(), content); }); ``` @@ -434,6 +456,8 @@ Array of [Resource](https://github.com/website-scraper/node-website-scraper/blob - `filename`: filename where page was saved (relative to `directory`) - `children`: array of children Resources +Note that resource content is not retained in memory after the resource was saved - read the saved files from your storage if you need the content. + ## Log and debug This module uses [debug](https://github.com/visionmedia/debug) to log events. To enable logs you should use environment variable `DEBUG`. The next command will log everything from website-scraper diff --git a/benchmark/memory.mjs b/benchmark/memory.mjs new file mode 100644 index 00000000..2c7616e3 --- /dev/null +++ b/benchmark/memory.mjs @@ -0,0 +1,91 @@ +/* + * Memory benchmark for issue #386. + * + * Starts a local http server serving an html page referencing FILES_COUNT + * binary files of FILE_SIZE_MB each (generated chunk-wise on the fly, so the + * server itself stays flat), scrapes it and reports peak memory usage. + * + * Usage: + * npm run benchmark:memory + * FILES_COUNT=50 FILE_SIZE_MB=10 REQUEST_CONCURRENCY=8 npm run benchmark:memory + */ + +import http from 'http'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; +import scrape from 'website-scraper'; + +const FILES_COUNT = parseInt(process.env.FILES_COUNT || '100', 10); +const FILE_SIZE_MB = parseInt(process.env.FILE_SIZE_MB || '20', 10); +const REQUEST_CONCURRENCY = parseInt(process.env.REQUEST_CONCURRENCY || '8', 10); + +const CHUNK = Buffer.alloc(64 * 1024, 0xab); +const CHUNKS_PER_FILE = Math.ceil(FILE_SIZE_MB * 1024 * 1024 / CHUNK.length); + +function startServer () { + const server = http.createServer((req, res) => { + if (req.url === '/') { + const imgs = Array.from({length: FILES_COUNT}, (_, i) => ``).join('\n'); + res.writeHead(200, {'content-type': 'text/html'}); + res.end(`${imgs}`); + return; + } + + res.writeHead(200, {'content-type': 'application/octet-stream'}); + let sent = 0; + const writeChunk = () => { + while (sent < CHUNKS_PER_FILE) { + sent++; + if (!res.write(CHUNK)) { + res.once('drain', writeChunk); + return; + } + } + res.end(); + }; + writeChunk(); + }); + + return new Promise((resolve) => { + server.listen(0, '127.0.0.1', () => resolve(server)); + }); +} + +function formatMB (bytes) { + return `${Math.round(bytes / 1024 / 1024)} MB`; +} + +const server = await startServer(); +const baseUrl = `http://127.0.0.1:${server.address().port}/`; +const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'website-scraper-benchmark-')); +fs.rmSync(directory, {recursive: true, force: true}); // scraper requires the directory to not exist + +const peak = {rss: 0, heapUsed: 0, external: 0, arrayBuffers: 0}; +const sampler = setInterval(() => { + const usage = process.memoryUsage(); + for (const key of Object.keys(peak)) { + peak[key] = Math.max(peak[key], usage[key]); + } +}, 200); + +console.log(`scraping ${FILES_COUNT} files x ${FILE_SIZE_MB} MB (${FILES_COUNT * FILE_SIZE_MB} MB total), concurrency ${REQUEST_CONCURRENCY}`); +const startedAt = Date.now(); + +try { + await scrape({ + urls: [baseUrl], + directory, + requestConcurrency: REQUEST_CONCURRENCY + }); +} finally { + clearInterval(sampler); + server.close(); + fs.rmSync(directory, {recursive: true, force: true}); +} + +console.log(`done in ${((Date.now() - startedAt) / 1000).toFixed(1)}s`); +console.log(`peak rss: ${formatMB(peak.rss)}`); +console.log(`peak heapUsed: ${formatMB(peak.heapUsed)}`); +console.log(`peak external: ${formatMB(peak.external)}`); +console.log(`peak arrayBuffers: ${formatMB(peak.arrayBuffers)}`); diff --git a/package.json b/package.json index 9652be77..30e612a8 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "website-scraper", - "version": "6.0.0", + "version": "7.0.0", "description": "Download website to a local directory (including all css, images, js, etc.)", "readmeFilename": "README.md", "type": "module", @@ -11,6 +11,7 @@ "scripts": { "test": "c8 --all --reporter=text --reporter=lcov mocha --recursive --timeout 7000 ./test/unit/ ./test/functional", "test-e2e": "mocha --timeout 300000 ./test/e2e/*-test.js", + "benchmark:memory": "node benchmark/memory.mjs", "eslint": "eslint lib index.mjs" }, "repository": { From 420a1465002787163618a8f05ced0ac8e071ef81 Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sat, 4 Jul 2026 23:00:25 +0200 Subject: [PATCH 06/18] Ignore benchmark directory in qlty code coverage Co-Authored-By: Claude Opus 4.8 (1M context) --- .qlty/qlty.toml | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .qlty/qlty.toml diff --git a/.qlty/qlty.toml b/.qlty/qlty.toml new file mode 100644 index 00000000..6ed42be8 --- /dev/null +++ b/.qlty/qlty.toml @@ -0,0 +1,6 @@ +config_version = "0" + +[coverage] +ignores = [ + "benchmark/**", +] From 17c58a5ff81eb6bfec6cac46eeba165588aee25f Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sun, 5 Jul 2026 11:14:08 +0200 Subject: [PATCH 07/18] Ignore qlty for test directory --- .qlty/qlty.toml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.qlty/qlty.toml b/.qlty/qlty.toml index 6ed42be8..116b11d3 100644 --- a/.qlty/qlty.toml +++ b/.qlty/qlty.toml @@ -1,5 +1,9 @@ config_version = "0" +exclude_patterns = [ + "test/**", +] + [coverage] ignores = [ "benchmark/**", From fa91c5fbab36c4373ad0b902ce17467d94f5bdf6 Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Mon, 19 Oct 2026 20:16:23 +0200 Subject: [PATCH 08/18] Address review feedback on streaming rework (#386) - Surface save failures of streamed child resources via fatalError - Destroy response stream when generateFilename or later steps throw - Buffer multi-saveResource content as Buffer instead of latin1 string - Memoize getBody promise to make concurrent calls safe - Consolidate stream cleanup in applyAfterResponse - FS plugin: remove output directory on error, don't mask save errors - Drop readAll and outputFile in favor of stream/consumers - Document concurrent saveResource calls in MIGRATION.md Co-Authored-By: Claude Opus 5.5 --- MIGRATION.md | 4 + lib/plugins/save-resource-to-fs-plugin.js | 14 ++- lib/request.js | 90 +++++++++++--------- lib/resource.js | 10 ++- lib/scraper.js | 73 ++++++++++------ lib/utils/fs.js | 8 -- lib/utils/stream.js | 11 --- test/functional/streaming/streaming.test.js | 94 +++++++++++++++++++++ test/unit/plugins.test.js | 50 +++++++++++ test/unit/request-test.js | 36 ++++++++ test/unit/resource-test.js | 14 ++- test/unit/utils/stream-test.js | 34 -------- 12 files changed, 313 insertions(+), 125 deletions(-) delete mode 100644 lib/utils/stream.js delete mode 100644 test/unit/utils/stream-test.js diff --git a/MIGRATION.md b/MIGRATION.md index ba971aef..ce6886f2 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -22,6 +22,10 @@ registerAction('saveResource', async ({resource}) => { // or pipe resource.getContentStream() to your storage to avoid buffering ``` +`saveResource` actions are no longer called one resource at a time: streamed resources are saved as their responses arrive, concurrently. If your action writes to a shared sink which is not concurrency-safe (an archive, a single file, a serial connection) - make it concurrency-safe or set `requestConcurrency: 1`. + +Note: when more than one `saveResource` action is registered, resource content is buffered in memory once so that every action can read it - memory usage is then again bounded by resource size for those resources. + #### afterResponse action The action is called when response headers arrive, before the body is downloaded. It receives `{response}` where response is `{url, statusCode, headers, getBody()}` instead of the full got response. Returning a string is not supported anymore - return an object or null. Return an object without `body` to keep the resource streaming (no buffering). diff --git a/lib/plugins/save-resource-to-fs-plugin.js b/lib/plugins/save-resource-to-fs-plugin.js index 894c8d26..8ad2b304 100644 --- a/lib/plugins/save-resource-to-fs-plugin.js +++ b/lib/plugins/save-resource-to-fs-plugin.js @@ -4,7 +4,7 @@ import { outputFileStream } from '../utils/fs.js'; class SaveResourceToFileSystemPlugin { apply (registerAction) { - let absoluteDirectoryPath, loadedResources = []; + let absoluteDirectoryPath, directoryWasValidated = false; registerAction('beforeStart', ({options}) => { if (!options.directory || typeof options.directory !== 'string') { @@ -16,6 +16,7 @@ class SaveResourceToFileSystemPlugin { if (fs.existsSync(absoluteDirectoryPath)) { throw new Error(`Directory ${absoluteDirectoryPath} exists`); } + directoryWasValidated = true; }); registerAction('saveResource', async ({resource}) => { @@ -23,14 +24,19 @@ class SaveResourceToFileSystemPlugin { try { await outputFileStream(filename, resource.getContentStream()); } catch (err) { - await fs.promises.rm(filename, {force: true}); // don't leave partially written files + try { + await fs.promises.rm(filename, {force: true}); // don't leave partially written files + } catch { + // cleanup failure must not mask the original save error + } throw err; } - loadedResources.push(resource); }); registerAction('error', async () => { - if (loadedResources.length > 0) { + // a failed save may leave directories behind even when no resource was saved - + // remove them unless the directory existed before the scrape (beforeStart threw) + if (directoryWasValidated) { await fs.promises.rm(absoluteDirectoryPath, {force: true, recursive: true}); } }); diff --git a/lib/request.js b/lib/request.js index 9002c00e..60bb401f 100644 --- a/lib/request.js +++ b/lib/request.js @@ -61,18 +61,19 @@ function requestWithRetry (requestOptions) { async function applyAfterResponse ({ responseData, headers, afterResponse }) { const originalStream = responseData.stream; - let bufferedBody = null; + // memoize the promise, not the value - concurrent getBody calls must not + // both start reading the same one-shot stream + let bufferedBodyPromise = null; - const getBody = async () => { - if (bufferedBody === null) { - bufferedBody = await readStreamToBuffer(originalStream); + const getBody = () => { + if (bufferedBodyPromise === null) { + bufferedBodyPromise = readStreamToBuffer(originalStream); } - return bufferedBody; + return bufferedBodyPromise; }; - let result; - try { - result = await afterResponse({ + const transform = async () => { + const result = await afterResponse({ response: { url: responseData.url, statusCode: responseData.statusCode, @@ -80,46 +81,55 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { getBody } }); - } catch (err) { - originalStream.destroy(); - throw err; - } - if (result === null || result === undefined) { - originalStream.destroy(); - return null; - } + if (result === null || result === undefined) { + return null; + } - if (typeof result !== 'object' || Array.isArray(result) || result instanceof Error) { - originalStream.destroy(); - throwAfterResponseTypeError(result); - } + if (typeof result !== 'object' || Array.isArray(result) || result instanceof Error) { + throwAfterResponseTypeError(result); + } - const encoding = result.encoding || responseData.encoding; - const metadata = result.metadata || null; + const metadata = result.metadata || null; - if ('body' in result) { - let bodyBuffer; - if (Buffer.isBuffer(result.body)) { - bodyBuffer = result.body; - } else if (typeof result.body === 'string') { - bodyBuffer = Buffer.from(result.body, encoding); - } else { - originalStream.destroy(); - throwAfterResponseTypeError(result.body); + if ('body' in result) { + if (result.body === null || result.body === undefined) { + return null; // no body - skip the resource, same as returning null + } + // plugin-supplied bodies default to binary encoding, not the header-derived one + const encoding = result.encoding || 'binary'; + let bodyBuffer; + if (Buffer.isBuffer(result.body)) { + bodyBuffer = result.body; + } else if (typeof result.body === 'string') { + bodyBuffer = Buffer.from(result.body, encoding); + } else { + throwAfterResponseTypeError(result.body); + } + return extend(responseData, { stream: Readable.from(bodyBuffer), encoding, metadata }); } - if (bufferedBody === null) { - originalStream.destroy(); // original body replaced without being read + + const encoding = result.encoding || responseData.encoding; + + if (bufferedBodyPromise !== null) { + // body was read by the action - original stream is consumed, serve the buffered copy + return extend(responseData, { stream: Readable.from(await bufferedBodyPromise), encoding, metadata }); } - return extend(responseData, { stream: Readable.from(bodyBuffer), encoding, metadata }); - } - if (bufferedBody !== null) { - // body was read by the action - original stream is consumed, serve the buffered copy - return extend(responseData, { stream: Readable.from(bufferedBody), encoding, metadata }); - } + return extend(responseData, { encoding, metadata }); + }; - return extend(responseData, { encoding, metadata }); + let transformed = null; + try { + transformed = await transform(); + return transformed; + } finally { + if (!transformed || transformed.stream !== originalStream) { + // the live stream is not handed over to the caller - destroy it to release + // the socket (no-op when the body was already fully read via getBody) + originalStream.destroy(); + } + } } /* diff --git a/lib/resource.js b/lib/resource.js index b6a94aa9..d4f386a5 100644 --- a/lib/resource.js +++ b/lib/resource.js @@ -62,9 +62,13 @@ class Resource { this.contentStream = stream; } + setContentBuffer (buffer) { + this.contentBuffer = buffer; + } + /** * Returns resource content as a Readable stream. - * For buffered (html/css) resources a fresh stream is created on each call. + * For buffered (html/css text or contentBuffer) resources a fresh stream is created on each call. * For streamed resources the live response stream is returned - it can be consumed only once. * Returns null when content was already cleared (see clearContent). */ @@ -72,6 +76,9 @@ class Resource { if (this.contentStream) { return this.contentStream; } + if (this.contentBuffer) { + return Readable.from(this.contentBuffer); + } if (typeof this.text === 'string') { return Readable.from(Buffer.from(this.text, this.encoding)); } @@ -81,6 +88,7 @@ class Resource { clearContent () { this.text = null; this.contentStream = null; + this.contentBuffer = null; } getDepth () { diff --git a/lib/scraper.js b/lib/scraper.js index 591f09c8..b66d83b2 100644 --- a/lib/scraper.js +++ b/lib/scraper.js @@ -14,7 +14,7 @@ import { import * as utils from './utils/index.js'; const { extend, union, urlsEqual, getTypeByMime, getTypeByFilename, series } = utils; -import { readAll } from './utils/stream.js'; +import { buffer as readStreamToBuffer } from 'stream/consumers'; import NormalizedUrlMap from './utils/normalized-url-map.js'; const actionNames = [ @@ -53,6 +53,7 @@ class Scraper { this.requestedResourcePromises = new NormalizedUrlMap(); // Map url -> request promise this.loadedResources = new NormalizedUrlMap(); // Map url -> resource this.requestQueue = new PromiseQueue({concurrency: this.options.requestConcurrency}); + this.fatalError = null; // first error which should abort the scrape, see Scraper.waitForLoad } normalizeOptions (options) { @@ -130,10 +131,9 @@ class Scraper { await this.resourceHandler.handleResource(resource); if (resource.contentStream && this.actions.saveResource.length > 1) { - // live stream can be consumed only once - buffer content - // so each saveResource action can read it - resource.setText(await readAll(resource.contentStream, 'binary')); - resource.setEncoding('binary'); + // live stream can be consumed only once - buffer content (as a Buffer, + // strings cap at ~512 MiB) so each saveResource action can read it + resource.setContentBuffer(await readStreamToBuffer(resource.contentStream)); resource.setContentStream(null); } @@ -202,30 +202,44 @@ class Scraper { resource.setType(getTypeByMime(responseData.mimeType)); const { stream, ...responseInfo } = responseData; - const { filename } = await self.runActions('generateFilename', { resource, responseData: responseInfo }); - resource.setFilename(filename); - // if type was not determined by mime we can try to get it from filename after it was generated - if (!resource.getType()) { - resource.setType(getTypeByFilename(filename)); - } + try { + const { filename } = await self.runActions('generateFilename', { resource, responseData: responseInfo }); + resource.setFilename(filename); - if (responseData.metadata) { - resource.setMetadata(responseData.metadata); - } + // if type was not determined by mime we can try to get it from filename after it was generated + if (!resource.getType()) { + resource.setType(getTypeByFilename(filename)); + } + + if (responseData.metadata) { + resource.setMetadata(responseData.metadata); + } - resource.setEncoding(responseData.encoding); - - if (resource.isHtml() || resource.isCss()) { - // html and css need full text to find and update links to child resources, - // they are saved after their children are resolved - see Scraper.waitForLoad - resource.setText(await readAll(stream, responseData.encoding)); - self.loadResource(resource); - } else { - // other types need no modification - stream them to storage immediately - resource.setContentStream(stream); - self.loadResource(resource); - await self.persistResource(resource); + resource.setEncoding(responseData.encoding); + + if (resource.isHtml() || resource.isCss()) { + // html and css need full text to find and update links to child resources, + // they are saved after their children are resolved - see Scraper.waitForLoad + resource.setText((await readStreamToBuffer(stream)).toString(responseData.encoding)); + self.loadResource(resource); + } else { + // other types need no modification - stream them to storage immediately + resource.setContentStream(stream); + self.loadResource(resource); + try { + await self.saveResource(resource); + } catch (err) { + // saveResource already handled the error (warn log + onResourceError); + // don't rethrow - this task's rejection would be discarded by + // Promise.allSettled in downloadChildrenResources - remember the error + // instead so waitForLoad aborts the scrape + self.fatalError = self.fatalError || err; + } + } + } catch (err) { + stream.destroy(); // the stream was never handed over to a consumer - release the socket + throw err; } return resource; }); @@ -289,6 +303,13 @@ class Scraper { // 1. Get all not saved resources and save them // 2. Recursion if any new not saved resource were added during this time. If not, loading is done. async waitForLoad () { + if (this.fatalError) { + // saving a streamed resource failed inside a request queue task - its rejection + // was discarded by Promise.allSettled in downloadChildrenResources, re-throw + // it here to abort the scrape (see Scraper.createNewRequest) + throw this.fatalError; + } + const resourcesToSave = Array.from(this.loadedResources.values()).filter((r) => !r.isSaved()); const loadingIsFinished = resourcesToSave.length === 0; diff --git a/lib/utils/fs.js b/lib/utils/fs.js index d03f97e0..497cd2a9 100644 --- a/lib/utils/fs.js +++ b/lib/utils/fs.js @@ -3,13 +3,6 @@ import fs from 'fs/promises'; import { createWriteStream } from 'fs'; import { pipeline } from 'stream/promises'; -async function outputFile (file, data, encoding) { - const dir = path.dirname(file); - await fs.mkdir(dir, { recursive: true}); - - return fs.writeFile(file, data, { encoding: encoding }); -} - async function outputFileStream (file, stream) { const dir = path.dirname(file); await fs.mkdir(dir, { recursive: true}); @@ -18,6 +11,5 @@ async function outputFileStream (file, stream) { } export { - outputFile, outputFileStream }; diff --git a/lib/utils/stream.js b/lib/utils/stream.js deleted file mode 100644 index 07880483..00000000 --- a/lib/utils/stream.js +++ /dev/null @@ -1,11 +0,0 @@ -async function readAll (stream, encoding) { - const chunks = []; - for await (const chunk of stream) { - chunks.push(chunk); - } - return Buffer.concat(chunks).toString(encoding); -} - -export { - readAll -}; diff --git a/test/functional/streaming/streaming.test.js b/test/functional/streaming/streaming.test.js index ad4498c9..447e827f 100644 --- a/test/functional/streaming/streaming.test.js +++ b/test/functional/streaming/streaming.test.js @@ -156,6 +156,36 @@ describe('Functional: streaming resources to storage', function () { seenResources.should.be.eql(['http://example.com/a.png']); }); + it('should reject the scrape when saving a streamed child resource fails and ignoreErrors is false', async function () { + nock('http://example.com/').get('/').reply(200, '', {'content-type': 'text/html'}); + nock('http://example.com/').get('/a.png').reply(200, 'image content', {'content-type': 'image/png'}); + + const errors = []; + class FailingStoragePlugin { + apply (add) { + add('saveResource', async ({resource}) => { + if (resource.getFilename().endsWith('.png')) { + throw new Error('SAVE FAILED'); + } + }); + add('onResourceError', ({error}) => errors.push(error.message)); + } + } + + try { + await scrape({ + urls: ['http://example.com/'], + directory: testDirname, + ignoreErrors: false, + plugins: [new FailingStoragePlugin()] + }); + throw new Error('expected scrape to reject'); + } catch (err) { + err.message.should.be.eql('SAVE FAILED'); + } + errors.should.be.eql(['SAVE FAILED']); // handled exactly once + }); + // nock cannot simulate a connection dying mid-body, so these tests use a real local server describe('mid-stream failures', function () { let server, baseUrl; @@ -221,6 +251,70 @@ describe('Functional: streaming resources to storage', function () { }); }); + describe('stream cleanup when an action fails', function () { + let server, baseUrl, pngSocket; + + beforeEach(function (done) { + nock.enableNetConnect(/127\.0\.0\.1|localhost/); + server = http.createServer((req, res) => { + if (req.url === '/a.png') { + // never ends the body - the socket stays open unless the client destroys it + pngSocket = req.socket; + res.writeHead(200, {'content-type': 'image/png'}); + res.write('partial data'); + } else { + res.writeHead(200, {'content-type': 'text/html'}); + res.end(''); + } + }); + server.listen(0, '127.0.0.1', () => { + baseUrl = `http://127.0.0.1:${server.address().port}`; + done(); + }); + }); + + afterEach(function (done) { + if (pngSocket && !pngSocket.destroyed) { + pngSocket.destroy(); + } + server.close(done); + }); + + it('should destroy the response stream when generateFilename throws', async function () { + class FailingFilenamePlugin { + apply (add) { + add('generateFilename', ({resource}) => { + if (resource.getUrl().endsWith('.png')) { + throw new Error('FILENAME FAILED'); + } + return {filename: 'index.html'}; + }); + } + } + + await scrape({ + urls: [baseUrl + '/'], + directory: testDirname, + ignoreErrors: true, + plugins: [new FailingFilenamePlugin()] + }); + + // the failed resource's socket must be released, otherwise the process never exits + await new Promise((resolve, reject) => { + const startedAt = Date.now(); + (function checkSocketClosed () { + if (pngSocket.destroyed) { + return resolve(); + } + if (Date.now() - startedAt > 2000) { + return reject(new Error('response socket was not destroyed')); + } + setTimeout(checkSocketClosed, 20); + })(); + }); + }); + }); + it('should handle redirect to an already requested resource with requestConcurrency 1', async function () { this.timeout(3000); // a deadlock in the request queue would exceed this diff --git a/test/unit/plugins.test.js b/test/unit/plugins.test.js index 60540ee7..4e550768 100644 --- a/test/unit/plugins.test.js +++ b/test/unit/plugins.test.js @@ -79,6 +79,28 @@ describe('SaveResourceToFileSystemPlugin', () => { } fs.existsSync(testDirname + '/a.png').should.be.eql(false); }); + + it('should propagate the original save error when cleanup rm fails', async () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + actions.beforeStart({options: {directory: testDirname}}); + + // first resource occupies "about" as a directory + const first = new Resource('http://example.com/about/', 'about/index.html'); + first.setContentStream(Readable.from('page')); + await actions.saveResource({resource: first}); + + // second resource wants "about" as a plain file - createWriteStream fails with EISDIR, + // and the cleanup rm of a directory would fail too - it must not mask the save error + const second = new Resource('http://example.com/about', 'about'); + second.setContentStream(Readable.from('other')); + + try { + await actions.saveResource({resource: second}); + throw new Error('expected saveResource to reject'); + } catch (err) { + err.code.should.be.eql('EISDIR'); + } + }); }); describe('error', () => { @@ -104,5 +126,33 @@ describe('SaveResourceToFileSystemPlugin', () => { fs.existsSync(testDirname).should.be.eql(false); }); + + it('should remove directories created by a failed save even when no resource was saved', async () => { + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + actions.beforeStart({options: {directory: testDirname}}); + + const resource = new Resource('http://example.com/a.png', 'images/a.png'); + resource.setContentStream(new Readable({ + read () { + this.destroy(new Error('mid-stream failure')); + } + })); + await actions.saveResource({resource}).catch(() => {}); + fs.existsSync(testDirname).should.be.eql(true); // directories created before the stream failed + + await actions.error(); + + fs.existsSync(testDirname).should.be.eql(false); + }); + + it('should not remove a pre-existing directory when beforeStart failed', async () => { + fs.mkdirSync(testDirname, {recursive: true}); + const actions = applyPlugin(new SaveResourceToFileSystemPlugin()); + (() => actions.beforeStart({options: {directory: testDirname}})).should.throw(/Directory (.*?) exists/); + + await actions.error(); + + fs.existsSync(testDirname).should.be.eql(true); + }); }); }); diff --git a/test/unit/request-test.js b/test/unit/request-test.js index 071f023b..eb4e750c 100644 --- a/test/unit/request-test.js +++ b/test/unit/request-test.js @@ -152,6 +152,22 @@ describe('request', () => { await request.get({url, afterResponse}); }); + it('should return full body to concurrent getBody calls', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + + const afterResponse = async ({response}) => { + const [first, second] = await Promise.all([response.getBody(), response.getBody()]); + first.toString().should.eql('TEST BODY'); + second.should.be.equal(first); + return {}; + }; + + const data = await request.get({url, afterResponse}); + const body = await readStreamText(data.stream); + body.should.eql('TEST BODY'); + }); + it('should return null and skip resource when action returns null', async () => { const url = 'http://example.com'; nock(url).get('/').reply(200, 'TEST BODY'); @@ -177,6 +193,26 @@ describe('request', () => { body.should.eql('a'); }); + it('should return null and skip resource when action returns object with null body', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + const handlerStub = sinon.stub().resolves({body: null}); + + const data = await request.get({url, afterResponse: handlerStub}); + should.not.exist(data); + }); + + it('should decode string body as binary when no encoding returned', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY', {'content-type': 'text/html; charset=utf-8'}); + const handlerStub = sinon.stub().resolves({body: 'café'}); + + const data = await request.get({url, afterResponse: handlerStub}); + data.encoding.should.eql('binary'); + const body = await readStreamBuffer(data.stream); + body.should.eql(Buffer.from('café', 'binary')); + }); + it('should support Buffer body', async () => { const url = 'http://example.com'; nock(url).get('/').reply(200, 'TEST BODY'); diff --git a/test/unit/resource-test.js b/test/unit/resource-test.js index d7875b48..8024840d 100644 --- a/test/unit/resource-test.js +++ b/test/unit/resource-test.js @@ -68,13 +68,25 @@ describe('Resource', function() { const streamedBytes = await readStreamBuffer(resource.getContentStream()); streamedBytes.should.be.eql(originalBytes); }); + + it('should return fresh stream with content buffer on each call', async function () { + const resource = new Resource('http://example.com'); + const contentBuffer = Buffer.from([0xff, 0x00, 0xab, 0x10]); + resource.setContentBuffer(contentBuffer); + + const firstBytes = await readStreamBuffer(resource.getContentStream()); + const secondBytes = await readStreamBuffer(resource.getContentStream()); + firstBytes.should.be.eql(contentBuffer); + secondBytes.should.be.eql(contentBuffer); + }); }); describe('#clearContent', function () { - it('should clear text and content stream', function () { + it('should clear text, content stream and content buffer', function () { const resource = new Resource('http://example.com'); resource.setText('some text'); resource.setContentStream(Readable.from('stream')); + resource.setContentBuffer(Buffer.from('buffer')); resource.clearContent(); diff --git a/test/unit/utils/stream-test.js b/test/unit/utils/stream-test.js deleted file mode 100644 index c2c84111..00000000 --- a/test/unit/utils/stream-test.js +++ /dev/null @@ -1,34 +0,0 @@ -import * as chai from 'chai'; -import { Readable } from 'stream'; -import { readAll } from '../../../lib/utils/stream.js'; -chai.should(); - -describe('utils/stream', function () { - describe('#readAll', function () { - it('should read whole stream to string with given encoding', async function () { - const stream = Readable.from([Buffer.from('hello '), Buffer.from('world')]); - const text = await readAll(stream, 'utf8'); - text.should.be.eql('hello world'); - }); - - it('should keep bytes intact with binary encoding', async function () { - const bytes = Buffer.from([0xff, 0x00, 0xab, 0x10]); - const text = await readAll(Readable.from(bytes), 'binary'); - Buffer.from(text, 'binary').should.be.eql(bytes); - }); - - it('should reject when stream emits error', async function () { - const stream = new Readable({ - read () { - this.destroy(new Error('read error')); - } - }); - try { - await readAll(stream, 'utf8'); - throw new Error('expected readAll to reject'); - } catch (err) { - err.message.should.be.eql('read error'); - } - }); - }); -}); From 1da5e88cb3714cef07981162239c9ea13a25e31f Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Mon, 19 Oct 2026 21:29:14 +0200 Subject: [PATCH 09/18] Document that afterResponse null body skips the resource (#386) Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 9ce229bb..7fac6076 100644 --- a/README.md +++ b/README.md @@ -326,7 +326,7 @@ Parameters - object which includes: Return resolved `Promise` with: * object if the resource should be saved, object may contain next properties: - * `body` (string or Buffer, optional) - replaces resource content; if omitted, the original body is used without buffering it in memory + * `body` (string or Buffer, optional) - replaces resource content; if omitted, the original body is used without buffering it in memory. `null` body skips the resource, same as returning `null` * `encoding` (`binary` or `utf8`) is used to decode a string `body`, binary is used by default. * `metadata` (object) - everything you want to save for this resource (like headers, original text, timestamps, etc.), scraper will not use this field at all, it is only for the result * or null if the resource should be skipped From 8fb52a6f574f468d3138652869220688554cc64e Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Mon, 19 Oct 2026 21:34:33 +0200 Subject: [PATCH 10/18] Derive afterResponse string body encoding from content-type (#386) - Default encoding of a string body returned by afterResponse is now derived from the response content-type header instead of binary - Match utf-8 charset in content-type case-insensitively - Document the new default in README and MIGRATION.md Co-Authored-By: Claude Opus 5.5 --- MIGRATION.md | 4 ++++ README.md | 2 +- lib/request.js | 8 +++----- test/unit/request-test.js | 16 +++++++++++++--- 4 files changed, 21 insertions(+), 9 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index ce6886f2..8eb48423 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -30,6 +30,10 @@ Note: when more than one `saveResource` action is registered, resource content i The action is called when response headers arrive, before the body is downloaded. It receives `{response}` where response is `{url, statusCode, headers, getBody()}` instead of the full got response. Returning a string is not supported anymore - return an object or null. Return an object without `body` to keep the resource streaming (no buffering). +Other contract changes: + +* When a returned string `body` has no `encoding`, it is decoded with the encoding derived from the response `content-type` header (`utf8` if it contains utf-8, case-insensitive, `binary` otherwise). Previously `binary` was always the default - pass `encoding: 'binary'` explicitly to keep the old behavior. + ```javascript // before registerAction('afterResponse', ({response}) => { diff --git a/README.md b/README.md index 7fac6076..ffd099a8 100644 --- a/README.md +++ b/README.md @@ -327,7 +327,7 @@ Parameters - object which includes: Return resolved `Promise` with: * object if the resource should be saved, object may contain next properties: * `body` (string or Buffer, optional) - replaces resource content; if omitted, the original body is used without buffering it in memory. `null` body skips the resource, same as returning `null` - * `encoding` (`binary` or `utf8`) is used to decode a string `body`, binary is used by default. + * `encoding` (`binary` or `utf8`) is used to decode a string `body`; defaults to the encoding derived from the response `content-type` header (`utf8` if it contains utf-8, case-insensitive, `binary` otherwise) * `metadata` (object) - everything you want to save for this resource (like headers, original text, timestamps, etc.), scraper will not use this field at all, it is only for the result * or null if the resource should be skipped diff --git a/lib/request.js b/lib/request.js index 60bb401f..6f02cbf0 100644 --- a/lib/request.js +++ b/lib/request.js @@ -11,7 +11,8 @@ function getMimeType (contentType) { function extractEncodingFromHeader (headers) { const contentTypeHeader = headers['content-type']; - return contentTypeHeader && contentTypeHeader.includes('utf-8') ? 'utf8' : 'binary'; + // charset values are case-insensitive (RFC 9110), e.g. "UTF-8" is common + return contentTypeHeader && contentTypeHeader.toLowerCase().includes('utf-8') ? 'utf8' : 'binary'; } function throwAfterResponseTypeError (result) { @@ -91,13 +92,12 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { } const metadata = result.metadata || null; + const encoding = result.encoding || responseData.encoding; if ('body' in result) { if (result.body === null || result.body === undefined) { return null; // no body - skip the resource, same as returning null } - // plugin-supplied bodies default to binary encoding, not the header-derived one - const encoding = result.encoding || 'binary'; let bodyBuffer; if (Buffer.isBuffer(result.body)) { bodyBuffer = result.body; @@ -109,8 +109,6 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { return extend(responseData, { stream: Readable.from(bodyBuffer), encoding, metadata }); } - const encoding = result.encoding || responseData.encoding; - if (bufferedBodyPromise !== null) { // body was read by the action - original stream is consumed, serve the buffered copy return extend(responseData, { stream: Readable.from(await bufferedBodyPromise), encoding, metadata }); diff --git a/test/unit/request-test.js b/test/unit/request-test.js index eb4e750c..443bfc70 100644 --- a/test/unit/request-test.js +++ b/test/unit/request-test.js @@ -69,6 +69,16 @@ describe('request', () => { body.should.eql('Hello from Google!'); }); + it('should return utf8 encoding when charset in content-type header is uppercase', async () => { + const url = 'http://www.google.com'; + nock(url).get('/').reply(200, 'Hello from Google!', { + 'content-type': 'text/html; charset=UTF-8' + }); + + const data = await request.get({url}); + data.encoding.should.eql('utf8'); + }); + it('should return mimeType = null and binary encoding if content-type header was not found in response', async () => { const url = 'http://www.google.com'; nock(url).get('/').reply(200, 'Hello from Google!', {}); @@ -202,15 +212,15 @@ describe('request', () => { should.not.exist(data); }); - it('should decode string body as binary when no encoding returned', async () => { + it('should decode string body with header-derived encoding when no encoding returned', async () => { const url = 'http://example.com'; nock(url).get('/').reply(200, 'TEST BODY', {'content-type': 'text/html; charset=utf-8'}); const handlerStub = sinon.stub().resolves({body: 'café'}); const data = await request.get({url, afterResponse: handlerStub}); - data.encoding.should.eql('binary'); + data.encoding.should.eql('utf8'); const body = await readStreamBuffer(data.stream); - body.should.eql(Buffer.from('café', 'binary')); + body.should.eql(Buffer.from('café', 'utf8')); }); it('should support Buffer body', async () => { From 047f83c74d58ff67567adba5428a2c989a5e2ccd Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Mon, 19 Oct 2026 21:34:33 +0200 Subject: [PATCH 11/18] Use nullish assignment for getBody memoization Bump eslint ecmaVersion to 2025 to support it. Co-Authored-By: Claude Opus 5.5 --- eslint.config.js | 2 +- lib/request.js | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/eslint.config.js b/eslint.config.js index fbd8b306..af7c4f68 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -10,7 +10,7 @@ export default [ '@stylistic': stylistic }, languageOptions: { - ecmaVersion: 2020, + ecmaVersion: 2025, sourceType: 'module', globals: { ...globals.node, diff --git a/lib/request.js b/lib/request.js index 6f02cbf0..972ef857 100644 --- a/lib/request.js +++ b/lib/request.js @@ -67,9 +67,7 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { let bufferedBodyPromise = null; const getBody = () => { - if (bufferedBodyPromise === null) { - bufferedBodyPromise = readStreamToBuffer(originalStream); - } + bufferedBodyPromise ??= readStreamToBuffer(originalStream); return bufferedBodyPromise; }; From 516e7a1f400426154efe35e052d360c0c4fd7adb Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Mon, 19 Oct 2026 21:53:30 +0200 Subject: [PATCH 12/18] Reduce returns in applyAfterResponse by extracting helpers Co-Authored-By: Claude Opus 5.5 --- lib/request.js | 91 ++++++++++++++++++++++++++------------------------ 1 file changed, 47 insertions(+), 44 deletions(-) diff --git a/lib/request.js b/lib/request.js index 972ef857..ca9d6dfb 100644 --- a/lib/request.js +++ b/lib/request.js @@ -60,18 +60,58 @@ function requestWithRetry (requestOptions) { }); } +function afterResponseBodyToBuffer (body, encoding) { + if (Buffer.isBuffer(body)) { + return body; + } + if (typeof body !== 'string') { + throwAfterResponseTypeError(body); + } + return Buffer.from(body, encoding); +} + +/* + * Maps afterResponse result to responseData, or null if the resource should be skipped. + * bufferedBodyPromise is set when the action read the body via getBody. + */ +async function buildAfterResponseData ({ result, responseData, bufferedBodyPromise }) { + if (result === null || result === undefined) { + return null; + } + + if (typeof result !== 'object' || Array.isArray(result) || result instanceof Error) { + throwAfterResponseTypeError(result); + } + + const hasBody = 'body' in result; + if (hasBody && (result.body === null || result.body === undefined)) { + return null; // no body - skip the resource, same as returning null + } + + const overrides = { + encoding: result.encoding || responseData.encoding, + metadata: result.metadata || null + }; + + if (hasBody) { + overrides.stream = Readable.from(afterResponseBodyToBuffer(result.body, overrides.encoding)); + } else if (bufferedBodyPromise !== null) { + // body was read by the action - original stream is consumed, serve the buffered copy + overrides.stream = Readable.from(await bufferedBodyPromise); + } + + return extend(responseData, overrides); +} + async function applyAfterResponse ({ responseData, headers, afterResponse }) { const originalStream = responseData.stream; // memoize the promise, not the value - concurrent getBody calls must not // both start reading the same one-shot stream let bufferedBodyPromise = null; + const getBody = () => (bufferedBodyPromise ??= readStreamToBuffer(originalStream)); - const getBody = () => { - bufferedBodyPromise ??= readStreamToBuffer(originalStream); - return bufferedBodyPromise; - }; - - const transform = async () => { + let transformed = null; + try { const result = await afterResponse({ response: { url: responseData.url, @@ -80,44 +120,7 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { getBody } }); - - if (result === null || result === undefined) { - return null; - } - - if (typeof result !== 'object' || Array.isArray(result) || result instanceof Error) { - throwAfterResponseTypeError(result); - } - - const metadata = result.metadata || null; - const encoding = result.encoding || responseData.encoding; - - if ('body' in result) { - if (result.body === null || result.body === undefined) { - return null; // no body - skip the resource, same as returning null - } - let bodyBuffer; - if (Buffer.isBuffer(result.body)) { - bodyBuffer = result.body; - } else if (typeof result.body === 'string') { - bodyBuffer = Buffer.from(result.body, encoding); - } else { - throwAfterResponseTypeError(result.body); - } - return extend(responseData, { stream: Readable.from(bodyBuffer), encoding, metadata }); - } - - if (bufferedBodyPromise !== null) { - // body was read by the action - original stream is consumed, serve the buffered copy - return extend(responseData, { stream: Readable.from(await bufferedBodyPromise), encoding, metadata }); - } - - return extend(responseData, { encoding, metadata }); - }; - - let transformed = null; - try { - transformed = await transform(); + transformed = await buildAfterResponseData({ result, responseData, bufferedBodyPromise }); return transformed; } finally { if (!transformed || transformed.stream !== originalStream) { From cc1efcdd33603c4fe28c19b9773db9ec5e74e2f5 Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sun, 25 Oct 2026 18:48:31 +0100 Subject: [PATCH 13/18] Don't abort the scrape when a streamed child resource fails mid-body (#386) Moving body consumption into the save phase made a network failure while streaming a child resource fatal with ignoreErrors: false, while the same failure on an html/css child (or any child request error) is discarded by Promise.allSettled and the scrape completes - as on master. Classify the error at the catch site: a got RequestError from a child resource is a request error and does not set fatalError; storage/plugin failures and root resource failures still abort the scrape. Co-Authored-By: Claude Fable 5.1 --- lib/scraper.js | 19 +++++++++++----- test/functional/streaming/streaming.test.js | 24 +++++++++++++++++++++ 2 files changed, 38 insertions(+), 5 deletions(-) diff --git a/lib/scraper.js b/lib/scraper.js index b66d83b2..bbefc031 100644 --- a/lib/scraper.js +++ b/lib/scraper.js @@ -3,6 +3,7 @@ import logger from './logger.js'; import defaults from './config/defaults.js'; import recursiveSources from './config/recursive-sources.js'; import Resource from './resource.js'; +import { RequestError } from 'got'; import request from './request.js'; import ResourceHandler from './resource-handler/index.js'; import { @@ -230,11 +231,19 @@ class Scraper { try { await self.saveResource(resource); } catch (err) { - // saveResource already handled the error (warn log + onResourceError); - // don't rethrow - this task's rejection would be discarded by - // Promise.allSettled in downloadChildrenResources - remember the error - // instead so waitForLoad aborts the scrape - self.fatalError = self.fatalError || err; + // saveResource already handled the error (warn log + onResourceError) + // and rethrew because ignoreErrors is false. This task's rejection + // would be discarded by Promise.allSettled in downloadChildrenResources, + // so decide here whether the scrape must abort: + // - a network failure while reading the body (got RequestError) of a + // child resource is a request error - same as a child failing before + // the body arrived, it does not abort the scrape + // - anything else (storage/plugin failure, or a root resource failing) + // is remembered so that waitForLoad aborts the scrape + const isChildNetworkError = err instanceof RequestError && Boolean(resource.parent); + if (!isChildNetworkError) { + self.fatalError = self.fatalError || err; + } } } } catch (err) { diff --git a/test/functional/streaming/streaming.test.js b/test/functional/streaming/streaming.test.js index 447e827f..01593cbd 100644 --- a/test/functional/streaming/streaming.test.js +++ b/test/functional/streaming/streaming.test.js @@ -231,6 +231,30 @@ describe('Functional: streaming resources to storage', function () { fs.readFileSync(testDirname + '/images/fine.png').toString().should.be.eql('fine content'); }); + it('should continue scraping when a child fails mid-stream without ignoreErrors (same as a failed child request)', async function () { + const errors = []; + class ErrorCollectorPlugin { + apply (add) { + add('onResourceError', ({resource, error}) => errors.push({url: resource.getUrl(), message: error.message})); + } + } + + const result = await scrape({ + urls: [baseUrl + '/'], + directory: testDirname, + request: {retry: {limit: 0}}, + ignoreErrors: false, + plugins: [new ErrorCollectorPlugin()] + }); + + result.should.have.length(1); + fs.existsSync(testDirname + '/index.html').should.be.eql(true); + fs.existsSync(testDirname + '/images/fail.png').should.be.eql(false); // failed, partial file removed + fs.readFileSync(testDirname + '/images/fine.png').toString().should.be.eql('fine content'); + errors.should.have.length(1); // handled exactly once + errors[0].url.should.be.eql(baseUrl + '/fail.png'); + }); + it('should reject and remove directory when a root resource fails mid-stream without ignoreErrors', async function () { try { await scrape({ From 704720d1852fe8ae1a4cfe8bf699dd104c14680f Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sun, 25 Oct 2026 19:09:21 +0100 Subject: [PATCH 14/18] Fail the save when the live stream errors, even if the action never settles (#386) A saveResource action which does not propagate source stream errors (e.g. .pipe() without an error handler) never resolved when the body died mid-stream: the request queue slot was held forever and scrape() hung. Race the actions against the stream's 'error' event so a failed body always fails that resource's save and goes through the normal error path. Co-Authored-By: Claude Fable 5.1 --- lib/scraper.js | 26 +++++++++++++++++- test/functional/streaming/streaming.test.js | 29 +++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/lib/scraper.js b/lib/scraper.js index bbefc031..ee1a7216 100644 --- a/lib/scraper.js +++ b/lib/scraper.js @@ -139,7 +139,7 @@ class Scraper { } logger.info('saving resource ' + resource + ' to fs'); - await this.runActions('saveResource', {resource}); + await this.runSaveResourceActions(resource); } finally { const stream = resource.contentStream; if (stream && !stream.readableEnded) { @@ -153,6 +153,30 @@ class Scraper { this.runActions('onResourceSaved', {resource}); } + /* + * Runs saveResource actions. For a live stream the save also fails when the stream + * itself errors (e.g. connection reset mid-body), so that an action which does not + * propagate stream errors (e.g. .pipe() without an error handler) cannot hang the scrape. + */ + async runSaveResourceActions (resource) { + const stream = resource.contentStream; + if (!stream) { + return this.runActions('saveResource', {resource}); + } + + let onStreamError; + const streamFailed = new Promise((resolve, reject) => { + onStreamError = reject; + stream.once('error', onStreamError); + }); + + try { + return await Promise.race([this.runActions('saveResource', {resource}), streamFailed]); + } finally { + stream.off('error', onStreamError); + } + } + async saveResource (resource) { try { await this.persistResource(resource); diff --git a/test/functional/streaming/streaming.test.js b/test/functional/streaming/streaming.test.js index 01593cbd..6b20cc2f 100644 --- a/test/functional/streaming/streaming.test.js +++ b/test/functional/streaming/streaming.test.js @@ -3,6 +3,7 @@ const should = chai.should(); import '../../utils/assertions.js'; import nock from 'nock'; import fs from 'fs'; +import path from 'path'; import http from 'http'; import { buffer as readStreamBuffer } from 'stream/consumers'; import scrape from 'website-scraper'; @@ -255,6 +256,34 @@ describe('Functional: streaming resources to storage', function () { errors[0].url.should.be.eql(baseUrl + '/fail.png'); }); + it('should fail the save instead of hanging when a saveResource action does not propagate stream errors', async function () { + const errors = []; + // replaces the default fs saver, so it is the only saveResource action and receives the live stream + class PipeWithoutErrorHandlingPlugin { + apply (add) { + add('beforeStart', () => fs.mkdirSync(testDirname, {recursive: true})); + add('saveResource', ({resource}) => new Promise((resolve) => { + const ws = fs.createWriteStream(path.join(testDirname, path.basename(resource.getFilename()))); + // .pipe() does not forward source errors - 'finish' never fires if the body dies + resource.getContentStream().pipe(ws).on('finish', resolve); + })); + add('onResourceError', ({resource, error}) => errors.push({url: resource.getUrl(), message: error.message})); + } + } + + const result = await scrape({ + urls: [baseUrl + '/'], + request: {retry: {limit: 0}}, + ignoreErrors: true, + plugins: [new PipeWithoutErrorHandlingPlugin()] + }); + + result.should.have.length(1); + fs.readFileSync(testDirname + '/fine.png').toString().should.be.eql('fine content'); + errors.should.have.length(1); + errors[0].url.should.be.eql(baseUrl + '/fail.png'); + }); + it('should reject and remove directory when a root resource fails mid-stream without ignoreErrors', async function () { try { await scrape({ From 51166ee7ef98b9f8d15ea5a231f9e9f8ce5e199e Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sun, 25 Oct 2026 19:21:48 +0100 Subject: [PATCH 15/18] Retry mid-body errors for buffered resources (#386) html/css are buffered anyway, but going through got.stream with the 'retry' listener removed at response time they lost the mid-body retries the promise API gave them on master. request.get now resolves with readBody() in addition to stream: readBody keeps got's 'retry' listener attached while reading, so a transient mid-body error is retried by got (its limit, errorCodes and backoff) and the body is re-read from the retried response. The scraper uses it for html/css and afterResponse's getBody uses it too. Streamed resources keep consuming `stream` directly and are not retried mid-body, since the body is already being written to storage. Co-Authored-By: Claude Fable 5.1 --- MIGRATION.md | 2 +- lib/request.js | 73 +++++++++++++++++---- lib/scraper.js | 7 +- test/functional/streaming/streaming.test.js | 51 ++++++++++++++ test/unit/request-test.js | 69 +++++++++++++++++++ 5 files changed, 185 insertions(+), 17 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index 330ef541..f36203e4 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -73,7 +73,7 @@ Resource content is not retained after save: `resource.getText()` returns `null` #### Other behavior changes * Binary files are written byte-for-byte from the network (previously the content was round-tripped through a latin1 string). Output for correctly-served files is unchanged; edge-case encodings may differ. -* Transient request errors are retried as before, but only until response headers are received. A connection which dies mid-body is not retried. +* Transient request errors are retried as before. For streamed resources (everything except html and css) retries happen only until response headers are received - a connection which dies mid-body is not retried, since the body is already being written to storage. * `responseType` in the `request` option is ignored (responses are streamed). ## From version 4 to 5 diff --git a/lib/request.js b/lib/request.js index f3ff6e40..ac91a57a 100644 --- a/lib/request.js +++ b/lib/request.js @@ -28,12 +28,14 @@ function throwAfterResponseTypeError (result) { } /* - * Requests url with got.stream and resolves with { response, stream } when response headers arrive, - * before the body is consumed. Attaching the 'retry' listener is what enables got retries for - * streams - transient errors before/at response time are retried with a fresh stream. - * Mid-body errors after the stream was handed over to the consumer cannot be retried. + * Attaches got retry handling to a request stream and resolves with { response, stream } when + * response headers arrive, before the body is consumed. Attaching the 'retry' listener is what + * enables got retries for streams - transient errors before/at response time are retried with + * a fresh stream. The listener is removed once headers arrive: a consumer which streams the body + * (e.g. pipeline to storage) cannot be handed a replacement stream, see readBodyWithRetry for + * consumers which buffer the body and can. */ -function requestWithRetry (url, requestOptions) { +function waitForResponse (url, initialStream) { return new Promise((resolve, reject) => { const attempt = (stream) => { const onRetry = (retryCount, error, createRetryStream) => { @@ -45,6 +47,8 @@ function requestWithRetry (url, requestOptions) { const onResponse = (response) => { stream.off('retry', onRetry); stream.off('error', onError); + // a body error emitted before the consumer attaches its own handlers must not crash the process + stream.on('error', (err) => logger.debug(`[request] response stream error for ${response.url}: ${err.message}`)); resolve({ response, stream }); }; const onError = (error) => { @@ -56,8 +60,48 @@ function requestWithRetry (url, requestOptions) { stream.once('response', onResponse); stream.once('error', onError); }; - attempt(got.stream(url, requestOptions)); + attempt(initialStream); + }); +} + +function requestWithRetry (url, requestOptions) { + return waitForResponse(url, got.stream(url, requestOptions)); +} + +/* + * Reads the whole body of a response stream with the 'retry' listener attached, so that got + * retries a transient mid-body error (e.g. connection reset) within its retry rules instead + * of emitting it. Resolves with { body } or, when got retried, with { retryStream }. + */ +function readBodyOrRetry (url, stream) { + let onRetry; + const retried = new Promise((resolve) => { + onRetry = (retryCount, error, createRetryStream) => { + logger.debug(`[request] retrying request for ${url} after body error, attempt ${retryCount}, reason: ${error.message}`); + resolve({ retryStream: createRetryStream() }); + }; + stream.once('retry', onRetry); }); + // got destroys the stream right before emitting 'retry', so on retry the read fails + // a tick later - the race has already settled with retryStream by then + const read = readStreamToBuffer(stream).then((body) => ({ body })); + + return Promise.race([read, retried]).finally(() => stream.off('retry', onRetry)); +} + +/* + * Buffers the body of a response stream, re-reading it from the retried response whenever + * got retries the request mid-body (limit, errorCodes and backoff are got's). + */ +async function readBodyWithRetry (url, stream) { + let current = stream; + for (;;) { + const { body, retryStream } = await readBodyOrRetry(url, current); + if (body) { + return body; + } + ({ stream: current } = await waitForResponse(url, retryStream)); + } } function afterResponseBodyToBuffer (body, encoding) { @@ -103,6 +147,10 @@ async function buildAfterResponseData ({ result, responseData, bufferedBodyPromi overrides.stream = Readable.from(await bufferedBodyPromise); } + if (overrides.stream) { + overrides.readBody = () => readStreamToBuffer(overrides.stream); + } + return extend(responseData, overrides); } @@ -111,7 +159,7 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { // memoize the promise, not the value - concurrent getBody calls must not // both start reading the same one-shot stream let bufferedBodyPromise = null; - const getBody = () => (bufferedBodyPromise ??= readStreamToBuffer(originalStream)); + const getBody = () => (bufferedBodyPromise ??= responseData.readBody()); let transformed = null; try { @@ -136,8 +184,9 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { /* * Resolves when response headers arrive with - * { url, statusCode, mimeType, encoding, metadata, stream } - * The body is NOT buffered - it is available as a Readable in stream. + * { url, statusCode, mimeType, encoding, metadata, stream, readBody } + * The body is NOT buffered - it is available as a Readable in stream, or via readBody() + * which buffers it with got retries for mid-body errors. Use one or the other. * Returns null if the resource should be skipped (afterResponse returned null). */ async function getRequest ({ url, referer, options = {}, afterResponse }) { @@ -153,16 +202,14 @@ async function getRequest ({ url, referer, options = {}, afterResponse }) { const { response, stream } = await requestWithRetry(url, requestOptions); logger.debug(`[request] received response for ${response.url}, statusCode ${response.statusCode}`); - // a body error emitted before the consumer attaches its own handlers must not crash the process - stream.on('error', (err) => logger.debug(`[request] response stream error for ${response.url}: ${err.message}`)); - const responseData = { url: response.url, statusCode: response.statusCode, mimeType: getMimeType(response.headers['content-type']), encoding: extractEncodingFromHeader(response.headers), metadata: null, - stream + stream, + readBody: () => readBodyWithRetry(url, stream) }; if (!afterResponse) { diff --git a/lib/scraper.js b/lib/scraper.js index ee1a7216..4e4cc28f 100644 --- a/lib/scraper.js +++ b/lib/scraper.js @@ -226,7 +226,7 @@ class Scraper { resource.setType(getTypeByMime(responseData.mimeType)); - const { stream, ...responseInfo } = responseData; + const { stream, readBody, ...responseInfo } = responseData; try { const { filename } = await self.runActions('generateFilename', { resource, responseData: responseInfo }); @@ -245,8 +245,9 @@ class Scraper { if (resource.isHtml() || resource.isCss()) { // html and css need full text to find and update links to child resources, - // they are saved after their children are resolved - see Scraper.waitForLoad - resource.setText((await readStreamToBuffer(stream)).toString(responseData.encoding)); + // they are saved after their children are resolved - see Scraper.waitForLoad. + // readBody retries mid-body errors (the stream is not handed over to anyone) + resource.setText((await readBody()).toString(responseData.encoding)); self.loadResource(resource); } else { // other types need no modification - stream them to storage immediately diff --git a/test/functional/streaming/streaming.test.js b/test/functional/streaming/streaming.test.js index 6b20cc2f..07272c17 100644 --- a/test/functional/streaming/streaming.test.js +++ b/test/functional/streaming/streaming.test.js @@ -304,6 +304,57 @@ describe('Functional: streaming resources to storage', function () { }); }); + describe('mid-body retries', function () { + let server, baseUrl, attempts; + + beforeEach(function (done) { + nock.enableNetConnect(/127\.0\.0\.1|localhost/); + attempts = {}; + server = http.createServer((req, res) => { + attempts[req.url] = (attempts[req.url] || 0) + 1; + const isHtml = req.url === '/'; + res.writeHead(200, {'content-type': isHtml ? 'text/html' : 'image/png'}); + if (attempts[req.url] === 1) { + res.write(isHtml ? '' : 'partial'); + setTimeout(() => res.destroy(), 10); + } else { + res.end(isHtml ? '' : 'image content'); + } + }); + server.listen(0, '127.0.0.1', () => { + baseUrl = `http://127.0.0.1:${server.address().port}`; + done(); + }); + }); + + afterEach(function (done) { + server.close(done); + }); + + it('should retry html which fails mid-body, but not a streamed resource', async function () { + const errors = []; + class ErrorCollectorPlugin { + apply (add) { + add('onResourceError', ({resource}) => errors.push(resource.getUrl())); + } + } + + await scrape({ + urls: [baseUrl + '/'], + directory: testDirname, + request: {retry: {limit: 1}}, + ignoreErrors: false, + plugins: [new ErrorCollectorPlugin()] + }); + + attempts['/'].should.be.eql(2); // html body retried and buffered from the 2nd attempt + fs.readFileSync(testDirname + '/index.html').toString().should.contain(''); + attempts['/a.png'].should.be.eql(1); // streamed to storage - no mid-body retry + fs.existsSync(testDirname + '/images/a.png').should.be.eql(false); + errors.should.be.eql([baseUrl + '/a.png']); + }); + }); + describe('stream cleanup when an action fails', function () { let server, baseUrl, pngSocket; diff --git a/test/unit/request-test.js b/test/unit/request-test.js index f8cb498a..035a8193 100644 --- a/test/unit/request-test.js +++ b/test/unit/request-test.js @@ -2,6 +2,9 @@ import * as chai from 'chai'; const should = chai.should(); import sinon from 'sinon'; import nock from 'nock'; +import http from 'http'; +import { pipeline } from 'stream/promises'; +import { Writable } from 'stream'; import { text as readStreamText, buffer as readStreamBuffer } from 'stream/consumers'; import request from '../../lib/request.js'; @@ -112,6 +115,72 @@ describe('request', () => { } }); + // nock cannot simulate a connection dying mid-body, so these tests use a real local server + describe('mid-body errors', () => { + let server, url, attempts; + + beforeEach((done) => { + attempts = 0; + server = http.createServer((req, res) => { + attempts++; + res.writeHead(200, {'content-type': 'text/html'}); + if (attempts === 1) { + res.write('partial'); + setTimeout(() => res.destroy(), 10); + } else { + res.end('full body'); + } + }); + server.listen(0, '127.0.0.1', () => { + url = `http://127.0.0.1:${server.address().port}/`; + done(); + }); + }); + + afterEach((done) => server.close(done)); + + it('should retry and return full body from readBody', async () => { + const data = await request.get({url, options: {retry: {limit: 1}}}); + const body = await data.readBody(); + body.toString().should.eql('full body'); + attempts.should.eql(2); + }); + + it('should reject readBody with the request error when no retries are left', async () => { + const data = await request.get({url, options: {retry: {limit: 0}}}); + try { + await data.readBody(); + throw new Error('expected readBody to reject'); + } catch (err) { + err.name.should.eql('ReadError'); + } + attempts.should.eql(1); + }); + + it('should not retry when the body is consumed as a stream', async () => { + const data = await request.get({url, options: {retry: {limit: 1}}}); + try { + await pipeline(data.stream, new Writable({write (chunk, enc, cb) { cb(); }})); + throw new Error('expected pipeline to reject'); + } catch (err) { + err.name.should.eql('ReadError'); + } + attempts.should.eql(1); + }); + + it('should retry inside afterResponse getBody', async () => { + let receivedBody; + const afterResponse = async ({response}) => { + receivedBody = await response.getBody(); + return {}; + }; + const data = await request.get({url, options: {retry: {limit: 1}}, afterResponse}); + receivedBody.toString().should.eql('full body'); + (await readStreamText(data.stream)).should.eql('full body'); + attempts.should.eql(2); + }); + }); + describe('afterResponse', () => { it('should call afterResponse with url, statusCode, headers and getBody', async () => { const url = 'http://example.com'; From 18a1160fd8530d1e968e411d66c3b72d79a60a96 Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sun, 25 Oct 2026 19:24:33 +0100 Subject: [PATCH 16/18] Reject getBody() calls after the afterResponse action has returned (#386) Once the action returns, the live stream is handed over to the scraper and may be consumed by storage. A late getBody() call would read the same one-shot stream concurrently, silently truncating the saved file. Co-Authored-By: Claude Fable 5.1 --- lib/request.js | 11 ++++++++++- test/unit/request-test.js | 20 ++++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/lib/request.js b/lib/request.js index ac91a57a..65ff0cb8 100644 --- a/lib/request.js +++ b/lib/request.js @@ -159,7 +159,14 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { // memoize the promise, not the value - concurrent getBody calls must not // both start reading the same one-shot stream let bufferedBodyPromise = null; - const getBody = () => (bufferedBodyPromise ??= responseData.readBody()); + let actionReturned = false; + const getBody = () => { + if (actionReturned) { + // the live stream may already be consumed by storage - reading it here would corrupt both + return Promise.reject(new Error(`getBody() can only be called while the afterResponse action is running (${responseData.url})`)); + } + return bufferedBodyPromise ??= responseData.readBody(); + }; let transformed = null; try { @@ -171,9 +178,11 @@ async function applyAfterResponse ({ responseData, headers, afterResponse }) { getBody } }); + actionReturned = true; transformed = await buildAfterResponseData({ result, responseData, bufferedBodyPromise }); return transformed; } finally { + actionReturned = true; // also when the action threw if (!transformed || transformed.stream !== originalStream) { // the live stream is not handed over to the caller - destroy it to release // the socket (no-op when the body was already fully read via getBody) diff --git a/test/unit/request-test.js b/test/unit/request-test.js index 035a8193..e99ec690 100644 --- a/test/unit/request-test.js +++ b/test/unit/request-test.js @@ -247,6 +247,26 @@ describe('request', () => { body.should.eql('TEST BODY'); }); + it('should reject getBody called after the action has returned', async () => { + const url = 'http://example.com'; + nock(url).get('/').reply(200, 'TEST BODY'); + + let savedResponse; + const data = await request.get({url, afterResponse: ({response}) => { + savedResponse = response; + return {}; + }}); + + try { + await savedResponse.getBody(); + throw new Error('expected getBody to reject'); + } catch (err) { + err.message.should.contain('getBody() can only be called while the afterResponse action is running'); + } + // the stream handed over to the caller is untouched + (await readStreamText(data.stream)).should.eql('TEST BODY'); + }); + it('should return null and skip resource when action returns null', async () => { const url = 'http://example.com'; nock(url).get('/').reply(200, 'TEST BODY'); From 3cf773640444af8584e3e14bb0b7a61162ad6b96 Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sun, 25 Oct 2026 19:26:34 +0100 Subject: [PATCH 17/18] Hand out a live content stream only once from getContentStream() (#386) For buffered content getContentStream() returns a fresh stream on each call, but for streamed resources it returned the same live stream, so a second call got an already-ended stream and a plugin which e.g. hashed then wrote silently produced an empty file for binaries only. A second call now throws with an explanation. Co-Authored-By: Claude Fable 5.1 --- MIGRATION.md | 2 +- README.md | 2 +- lib/resource.js | 9 ++++++++- test/unit/resource-test.js | 18 ++++++++++++++++++ 4 files changed, 28 insertions(+), 3 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index f36203e4..000813ec 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -19,7 +19,7 @@ Version 7 stops storing resource content in memory ([#386](https://github.com/we #### saveResource action -Resource content is now exposed as a Readable stream via `resource.getContentStream()` instead of `resource.getText()`. For streamed resources the stream can be consumed only once - consume or destroy it within the action. +Resource content is now exposed as a Readable stream via `resource.getContentStream()` instead of `resource.getText()`. For streamed resources the stream can be consumed only once - consume or destroy it within the action. `getContentStream()` hands such a stream out only once, a second call throws. ```javascript // before diff --git a/README.md b/README.md index 31d72160..71b8ed17 100644 --- a/README.md +++ b/README.md @@ -430,7 +430,7 @@ Action saveResource is called to save file to some storage. Use it to save files Parameters - object which includes: * resource - [Resource](https://github.com/website-scraper/node-website-scraper/blob/master/lib/resource.js) object -Resource content is exposed as a Readable stream returned by `resource.getContentStream()`. For resources which are streamed directly from the network (all types except html and css) the stream can be consumed only once - consume it (or destroy it) within the action. +Resource content is exposed as a Readable stream returned by `resource.getContentStream()`. For resources which are streamed directly from the network (all types except html and css) the stream can be consumed only once - consume it (or destroy it) within the action. `getContentStream()` hands such a stream out only once, a second call throws - buffer the content yourself if you need it more than once. If multiple actions `saveResource` added - resource will be saved to multiple storages (the scraper buffers streamed content in that case so every action can read it). ```javascript diff --git a/lib/resource.js b/lib/resource.js index d4f386a5..20dcb5fb 100644 --- a/lib/resource.js +++ b/lib/resource.js @@ -60,6 +60,7 @@ class Resource { setContentStream (stream) { this.contentStream = stream; + this.contentStreamTaken = false; } setContentBuffer (buffer) { @@ -69,11 +70,16 @@ class Resource { /** * Returns resource content as a Readable stream. * For buffered (html/css text or contentBuffer) resources a fresh stream is created on each call. - * For streamed resources the live response stream is returned - it can be consumed only once. + * For streamed resources the live response stream is returned - it can be consumed only once, + * so it is handed out only once: a second call throws instead of returning an ended stream. * Returns null when content was already cleared (see clearContent). */ getContentStream () { if (this.contentStream) { + if (this.contentStreamTaken) { + throw new Error(`Content stream of ${this} was already taken - it is a live response stream which can be consumed only once. Buffer it in your action if you need the content more than once`); + } + this.contentStreamTaken = true; return this.contentStream; } if (this.contentBuffer) { @@ -88,6 +94,7 @@ class Resource { clearContent () { this.text = null; this.contentStream = null; + this.contentStreamTaken = false; this.contentBuffer = null; } diff --git a/test/unit/resource-test.js b/test/unit/resource-test.js index 8024840d..476c451d 100644 --- a/test/unit/resource-test.js +++ b/test/unit/resource-test.js @@ -49,6 +49,24 @@ describe('Resource', function() { resource.getContentStream().should.be.equal(stream); }); + it('should throw when a live stream is requested a second time', function () { + const resource = new Resource('http://example.com/a.png'); + resource.setContentStream(Readable.from(Buffer.from('binary data'))); + resource.getContentStream(); + + (() => resource.getContentStream()).should.throw(/was already taken/); + }); + + it('should hand out a newly set live stream again', function () { + const resource = new Resource('http://example.com/a.png'); + resource.setContentStream(Readable.from(Buffer.from('first'))); + resource.getContentStream(); + + const second = Readable.from(Buffer.from('second')); + resource.setContentStream(second); + resource.getContentStream().should.be.equal(second); + }); + it('should return fresh stream with text content on each call', async function () { const resource = new Resource('http://example.com'); resource.setEncoding('utf8'); From eff51466637c9b53ff51d4022baaa1c4f21051a5 Mon Sep 17 00:00:00 2001 From: Illia Antypenko Date: Sun, 25 Oct 2026 19:30:18 +0100 Subject: [PATCH 18/18] Remove the output directory on error only if the fs plugin wrote to it (#386) The error action removed the directory whenever beforeStart had validated it, so files placed there by other plugins during the run were deleted even when this plugin had not written anything. Track the first write instead (that is where directories get created), which still cleans up after a partial save. Co-Authored-By: Claude Fable 5.1 --- lib/plugins/save-resource-to-fs-plugin.js | 11 +++++---- .../error-handling/error-handling.test.js | 24 +++++++++++++++++++ 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/lib/plugins/save-resource-to-fs-plugin.js b/lib/plugins/save-resource-to-fs-plugin.js index 8ad2b304..4350bcd3 100644 --- a/lib/plugins/save-resource-to-fs-plugin.js +++ b/lib/plugins/save-resource-to-fs-plugin.js @@ -4,7 +4,7 @@ import { outputFileStream } from '../utils/fs.js'; class SaveResourceToFileSystemPlugin { apply (registerAction) { - let absoluteDirectoryPath, directoryWasValidated = false; + let absoluteDirectoryPath, wroteToDirectory = false; registerAction('beforeStart', ({options}) => { if (!options.directory || typeof options.directory !== 'string') { @@ -16,11 +16,11 @@ class SaveResourceToFileSystemPlugin { if (fs.existsSync(absoluteDirectoryPath)) { throw new Error(`Directory ${absoluteDirectoryPath} exists`); } - directoryWasValidated = true; }); registerAction('saveResource', async ({resource}) => { const filename = path.join(absoluteDirectoryPath, resource.getFilename()); + wroteToDirectory = true; // outputFileStream creates directories even if the write then fails try { await outputFileStream(filename, resource.getContentStream()); } catch (err) { @@ -34,9 +34,10 @@ class SaveResourceToFileSystemPlugin { }); registerAction('error', async () => { - // a failed save may leave directories behind even when no resource was saved - - // remove them unless the directory existed before the scrape (beforeStart threw) - if (directoryWasValidated) { + // remove what this plugin created (a failed save may leave directories behind even + // when no resource was saved), but nothing when it wrote nothing - the directory + // may then hold files placed there by other plugins + if (wroteToDirectory) { await fs.promises.rm(absoluteDirectoryPath, {force: true, recursive: true}); } }); diff --git a/test/functional/error-handling/error-handling.test.js b/test/functional/error-handling/error-handling.test.js index 5cd588b6..2affe3a8 100644 --- a/test/functional/error-handling/error-handling.test.js +++ b/test/functional/error-handling/error-handling.test.js @@ -91,6 +91,30 @@ describe('Functional error handling', function() { }); }); + describe('Error before anything was saved', function () { + it('should not remove the directory when the fs plugin wrote nothing to it', function () { + nock.cleanAll(); + nock('http://example.com/').get('/').replyWithError('REQUEST FAILED!'); + + class RequestLogPlugin { + apply (registerAction) { + registerAction('beforeRequest', ({requestOptions}) => { + fs.mkdirSync(testDirname, {recursive: true}); + fs.appendFileSync(testDirname + '/requests.log', 'request\n'); + return {requestOptions}; + }); + } + } + + return scrape({...options, ignoreErrors: false, plugins: [new RequestLogPlugin()]}).then(function () { + false.should.be.true; + }).catch(function (err) { + err.message.should.be.eql('REQUEST FAILED!'); + fs.existsSync(testDirname + '/requests.log').should.be.eql(true); + }); + }); + }); + describe('Resource Handler Error', function () { let scraper; let handleResourceStub;