diff --git a/src/error/QueueIsFullError.ts b/src/error/QueueIsFullError.ts new file mode 100644 index 0000000..5e5cbf7 --- /dev/null +++ b/src/error/QueueIsFullError.ts @@ -0,0 +1 @@ +export class QueueIsFullError extends Error {} diff --git a/src/error/index.ts b/src/error/index.ts index e6cccbd..c6fe528 100644 --- a/src/error/index.ts +++ b/src/error/index.ts @@ -1,2 +1,3 @@ export * from './BlockedError'; export * from './NotFoundError'; +export * from './QueueIsFullError'; diff --git a/src/utils/Fetcher.test.ts b/src/utils/Fetcher.test.ts index cd64853..7d50bc6 100644 --- a/src/utils/Fetcher.test.ts +++ b/src/utils/Fetcher.test.ts @@ -2,10 +2,10 @@ import winston from 'winston'; import fetchMock from 'fetch-mock'; import { Fetcher } from './Fetcher'; import { Context } from '../types'; -import { BlockedError, NotFoundError } from '../error'; +import { BlockedError, NotFoundError, QueueIsFullError } from '../error'; fetchMock.mockGlobal(); -const fetcher = new Fetcher(winston.createLogger({ transports: [new winston.transports.Console({ level: 'nope' })] })); +const fetcher = new Fetcher(winston.createLogger({ transports: [new winston.transports.Console({ level: 'nope' })] }), 3, 10); describe('fetch', () => { const ctx: Context = { id: 'id', ip: '127.0.0.1', config: { de: 'on' } }; @@ -106,15 +106,21 @@ describe('fetch', () => { }); test('times out after 10 seconds', async () => { - jest.useFakeTimers(); - fetchMock.get('https://some-timeout-url.test/', 200, { delay: 15000 }); + fetchMock.get('https://some-timeout-url.test/', 200, { delay: 20 }); - const promise = fetcher.text(ctx, new URL('https://some-timeout-url.test/')); + await expect(fetcher.text(ctx, new URL('https://some-timeout-url.test/'))).rejects.toBeInstanceOf(DOMException); + }); - jest.advanceTimersByTime(13000); + test('full queue throws an error', async () => { + fetchMock.get('https://some-full-queue-url.test/', 'some text'); - await expect(promise).rejects.toBeInstanceOf(DOMException); + const allPromises = Promise.all([ + fetcher.text(ctx, new URL('https://some-full-queue-url.test/')), + fetcher.text(ctx, new URL('https://some-full-queue-url.test/')), + fetcher.text(ctx, new URL('https://some-full-queue-url.test/')), + fetcher.text(ctx, new URL('https://some-full-queue-url.test/')), + ]); - jest.useRealTimers(); + await expect(allPromises).rejects.toBeInstanceOf(QueueIsFullError); }); }); diff --git a/src/utils/Fetcher.ts b/src/utils/Fetcher.ts index f87ba76..c78e0fa 100644 --- a/src/utils/Fetcher.ts +++ b/src/utils/Fetcher.ts @@ -2,17 +2,34 @@ import CachePolicy from 'http-cache-semantics'; import TTLCache from '@isaacs/ttlcache'; import winston from 'winston'; import { Context, TIMEOUT } from '../types'; -import { BlockedError, NotFoundError } from '../error'; +import { BlockedError, NotFoundError, QueueIsFullError } from '../error'; import { clearTimeout } from 'node:timers'; -interface HttpCacheItem { policy: CachePolicy; status: number; statusText: string; body: string } +interface HttpCacheItem { + policy: CachePolicy; + status: number; + statusText: string; + body: string; +} + +interface HostQueue { + promise: Promise; + count: number; +} export class Fetcher { private readonly logger: winston.Logger; - private readonly httpCache: TTLCache; + private readonly queueLimit: number; + private readonly timeout: number; - constructor(logger: winston.Logger) { + private readonly httpCache: TTLCache; + private readonly hostQueues = new Map(); + + constructor(logger: winston.Logger, queueLimit?: number, timeout?: number) { this.logger = logger; + this.queueLimit = queueLimit ?? 10; + this.timeout = timeout ?? 10000; + this.httpCache = new TTLCache(); } @@ -87,18 +104,7 @@ export class Fetcher { return this.handleHttpCacheItem(httpCacheItem); } - this.logger.info(`Fetch ${request.method} ${url}`, ctx); - - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(TIMEOUT), 10000); - - let response; - try { - response = await fetch(url, { ...newInit, signal: controller.signal }); - } finally { - clearTimeout(timer); - } - + const response = await this.queuedFetch(ctx, url, newInit); const body = await response.text(); const policy = new CachePolicy(request, { status: response.status, headers: this.headersToObject(response.headers) }, { shared: false }); @@ -118,4 +124,52 @@ export class Fetcher { return this.handleHttpCacheItem(httpCacheItem); }; + + private readonly fetchWithTimeout = async (ctx: Context, url: URL, init?: RequestInit): Promise => { + this.logger.info(`Fetch ${init?.method ?? 'GET'} ${url}`, ctx); + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(TIMEOUT), this.timeout); + + let response; + try { + response = await fetch(url, { ...init, signal: controller.signal }); + } finally { + clearTimeout(timer); + } + + return response; + }; + + private readonly queuedFetch = async (ctx: Context, url: URL, init?: RequestInit): Promise => { + let queue = this.hostQueues.get(url.host); + if (!queue) { + queue = { promise: Promise.resolve(), count: 0 }; + this.hostQueues.set(url.host, queue); + } + + queue.count++; + + if (queue.count > this.queueLimit) { + throw new QueueIsFullError(); + } + + let resolveCurrent: () => void; + const currentPromise = new Promise( + (resolve) => { resolveCurrent = resolve; }, + ); + + const fetchPromise = queue.promise.then(async () => { + try { + return await this.fetchWithTimeout(ctx, url, init); + } finally { + resolveCurrent(); + queue.count--; + } + }); + + queue.promise = currentPromise.catch(/* istanbul ignore next */ () => undefined); + + return fetchPromise; + }; } diff --git a/src/utils/StreamResolver.test.ts b/src/utils/StreamResolver.test.ts index d95c5ea..12a6f3a 100644 --- a/src/utils/StreamResolver.test.ts +++ b/src/utils/StreamResolver.test.ts @@ -4,7 +4,7 @@ import { StreamResolver } from './StreamResolver'; import { Handler, MeineCloud, MostraGuarda } from '../handler'; import { Fetcher } from './Fetcher'; import { Context, CountryCode, TIMEOUT, UrlResult } from '../types'; -import { BlockedError, NotFoundError } from '../error'; +import { BlockedError, NotFoundError, QueueIsFullError } from '../error'; jest.mock('../utils/Fetcher'); const logger = winston.createLogger({ transports: [new winston.transports.Console({ level: 'nope' })] }); @@ -99,6 +99,16 @@ describe('resolve', () => { countryCode: 'de', }, }, + { + url: new URL('https://example3.com'), + isExternal: true, + error: new QueueIsFullError(), + label: 'hoster.com', + sourceId: '', + meta: { + countryCode: 'de', + }, + }, ]; }; } diff --git a/src/utils/StreamResolver.ts b/src/utils/StreamResolver.ts index c73fedf..6f0f553 100644 --- a/src/utils/StreamResolver.ts +++ b/src/utils/StreamResolver.ts @@ -4,7 +4,7 @@ import winston from 'winston'; import bytes from 'bytes'; import { Context, TIMEOUT, UrlResult } from '../types'; import { Handler } from '../handler'; -import { BlockedError, NotFoundError } from '../error'; +import { BlockedError, NotFoundError, QueueIsFullError } from '../error'; import { languageFromCountryCode } from './languageFromCountryCode'; export class StreamResolver { @@ -125,6 +125,12 @@ export class StreamResolver { return '🐢 Request timed out.'; } + if (error instanceof QueueIsFullError) { + this.logger.warn(`${source}: Request queue is full.`, ctx); + + return '⏳ Request queue is full. Please try again later or consider self-hosting.'; + } + const cause = (error as Error & { cause?: unknown }).cause; this.logger.error(`${source} error: ${error}, cause: ${cause}`, ctx); diff --git a/src/utils/__snapshots__/StreamResolver.test.ts.snap b/src/utils/__snapshots__/StreamResolver.test.ts.snap index f0a580f..30811df 100644 --- a/src/utils/__snapshots__/StreamResolver.test.ts.snap +++ b/src/utils/__snapshots__/StreamResolver.test.ts.snap @@ -26,6 +26,14 @@ exports[`resolve adds error info 1`] = ` 🔗 hoster.com 🐢 Request timed out.", }, + { + "behaviorHints": {}, + "externalUrl": "https://example3.com/", + "name": "WebStreamr N/A ⚠️ external", + "title": "🌐 German 🇩🇪 +🔗 hoster.com +⏳ Request queue is full. Please try again later or consider self-hosting.", + }, ] `; @@ -55,6 +63,14 @@ exports[`resolve adds error info 2`] = ` 🐢 Request timed out.", "ytId": "E4WlUXrJgy4", }, + { + "behaviorHints": {}, + "name": "WebStreamr N/A", + "title": "🌐 German 🇩🇪 +🔗 hoster.com +⏳ Request queue is full. Please try again later or consider self-hosting.", + "ytId": "E4WlUXrJgy4", + }, ] `;