From f7042b94ef7da5de1c5d581ee5236df910508317 Mon Sep 17 00:00:00 2001 From: WebStreamr <210764791+webstreamr@users.noreply.github.com> Date: Tue, 17 Jun 2025 19:09:22 +0000 Subject: [PATCH] chore(fetcher): remember recent timeouts and throw if count is too high This should avoid issues with longer outages of hosters like we had with DoodStream now twice recently. --- src/error/TooManyTimeoutsError.ts | 1 + src/error/index.ts | 3 +- src/utils/Fetcher.test.ts | 10 ++++- src/utils/Fetcher.ts | 38 +++++++++++++++---- src/utils/StreamResolver.test.ts | 12 +++++- src/utils/StreamResolver.ts | 16 +++++++- .../__snapshots__/StreamResolver.test.ts.snap | 16 ++++++++ 7 files changed, 85 insertions(+), 11 deletions(-) create mode 100644 src/error/TooManyTimeoutsError.ts diff --git a/src/error/TooManyTimeoutsError.ts b/src/error/TooManyTimeoutsError.ts new file mode 100644 index 0000000..1dff207 --- /dev/null +++ b/src/error/TooManyTimeoutsError.ts @@ -0,0 +1 @@ +export class TooManyTimeoutsError extends Error {} diff --git a/src/error/index.ts b/src/error/index.ts index 831e515..7eb30ce 100644 --- a/src/error/index.ts +++ b/src/error/index.ts @@ -2,5 +2,6 @@ export * from './BlockedError'; export * from './HttpError'; export * from './NotFoundError'; export * from './QueueIsFullError'; -export * from './TooManyRequestsError'; export * from './TimeoutError'; +export * from './TooManyRequestsError'; +export * from './TooManyTimeoutsError'; diff --git a/src/utils/Fetcher.test.ts b/src/utils/Fetcher.test.ts index e311e75..30e9737 100644 --- a/src/utils/Fetcher.test.ts +++ b/src/utils/Fetcher.test.ts @@ -2,7 +2,7 @@ import winston from 'winston'; import fetchMock from 'fetch-mock'; import { Fetcher } from './Fetcher'; import { Context } from '../types'; -import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TimeoutError, TooManyRequestsError } from '../error'; +import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TimeoutError, TooManyRequestsError, TooManyTimeoutsError } from '../error'; fetchMock.mockGlobal(); const fetcher = new Fetcher(winston.createLogger({ transports: [new winston.transports.Console({ level: 'nope' })] })); @@ -198,4 +198,12 @@ describe('fetch', () => { await expect(allPromises).rejects.toBeInstanceOf(QueueIsFullError); }); + + test('throws if too many recent requests timed-out', async () => { + fetchMock.get('https://too-many-recent-timeouts-url.test/', 200, { delay: 20 }); + + await expect(fetcher.text(ctx, new URL('https://too-many-recent-timeouts-url.test/'), { timeout: 10, timeoutsCountThrow: 2 })).rejects.toBeInstanceOf(TimeoutError); + await expect(fetcher.text(ctx, new URL('https://too-many-recent-timeouts-url.test/'), { timeout: 10, timeoutsCountThrow: 2 })).rejects.toBeInstanceOf(TimeoutError); + await expect(fetcher.text(ctx, new URL('https://too-many-recent-timeouts-url.test/'), { timeout: 10, timeoutsCountThrow: 2 })).rejects.toBeInstanceOf(TooManyTimeoutsError); + }); }); diff --git a/src/utils/Fetcher.ts b/src/utils/Fetcher.ts index a864185..141a3c1 100644 --- a/src/utils/Fetcher.ts +++ b/src/utils/Fetcher.ts @@ -1,10 +1,10 @@ import CachePolicy from 'http-cache-semantics'; import TTLCache from '@isaacs/ttlcache'; import winston from 'winston'; -import { Semaphore, SemaphoreInterface, withTimeout } from 'async-mutex'; +import { Mutex, Semaphore, SemaphoreInterface, withTimeout } from 'async-mutex'; import { Cookie, CookieJar } from 'tough-cookie'; import { BlockedReason, Context } from '../types'; -import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TimeoutError, TooManyRequestsError } from '../error'; +import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TimeoutError, TooManyTimeoutsError, TooManyRequestsError } from '../error'; import { clearTimeout } from 'node:timers'; import { envGet } from './env'; @@ -41,6 +41,7 @@ interface FlareSolverrResult { } export type CustomRequestInit = RequestInit & { + timeoutsCountThrow?: number; noFlareSolverr?: boolean; queueLimit?: number; queueTimeout?: number; @@ -52,20 +53,22 @@ export class Fetcher { private readonly DEFAULT_TIMEOUT = 10000; private readonly DEFAULT_QUEUE_LIMIT = 5; private readonly DEFAULT_QUEUE_TIMEOUT = 5000; + private readonly DEFAULT_FAILED_REQUEST_COUNT_THROW = 10; + private readonly TIMEOUT_CACHE_TTL = 3600000; // 1h private readonly logger: winston.Logger; - private readonly httpCache: TTLCache; - private readonly rateLimitedCache: TTLCache; + private readonly httpCache = new TTLCache(); + private readonly rateLimitedCache = new TTLCache(); private readonly semaphores = new Map(); private readonly hostUserAgentMap = new Map(); private readonly cookieJar = new CookieJar(); + private readonly timeoutsCountCache = new TTLCache(); + private readonly timeoutsCountMutex = new Mutex(); + public constructor(logger: winston.Logger) { this.logger = logger; - - this.httpCache = new TTLCache(); - this.rateLimitedCache = new TTLCache(); } public async text(ctx: Context, url: URL, init?: CustomRequestInit): Promise { @@ -228,6 +231,10 @@ export class Fetcher { throw new TooManyRequestsError(this.rateLimitedCache.getRemainingTTL(url.host) / 1000); } + if ((this.timeoutsCountCache.get(url.host) ?? 0) >= (init?.timeoutsCountThrow ?? this.DEFAULT_FAILED_REQUEST_COUNT_THROW)) { + throw new TooManyTimeoutsError(); + } + const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), init?.timeout ?? this.DEFAULT_TIMEOUT); @@ -236,6 +243,7 @@ export class Fetcher { response = await fetch(url, { ...init, keepalive: true, signal: controller.signal }); } catch (error) { if (error instanceof DOMException && error.name === 'AbortError') { + await this.increaseTimeoutsCount(url); throw new TimeoutError(); } @@ -244,9 +252,25 @@ export class Fetcher { clearTimeout(timer); } + await this.decreaseTimeoutsCount(url); + return response; }; + private async increaseTimeoutsCount(url: URL) { + await this.timeoutsCountMutex.runExclusive(async () => { + const count = (this.timeoutsCountCache.get(url.host) ?? 0) + 1; + this.timeoutsCountCache.set(url.host, count, { ttl: this.TIMEOUT_CACHE_TTL }); + }); + } + + private async decreaseTimeoutsCount(url: URL) { + await this.timeoutsCountMutex.runExclusive(async () => { + const count = Math.max(0, (this.timeoutsCountCache.get(url.host) ?? 0) - 1); + this.timeoutsCountCache.set(url.host, count, { ttl: this.TIMEOUT_CACHE_TTL }); + }); + } + private getSemaphore(url: URL, queueLimit: number, queueTimeout: number): SemaphoreInterface { let sem = this.semaphores.get(url.host); diff --git a/src/utils/StreamResolver.test.ts b/src/utils/StreamResolver.test.ts index 5e7e0d9..361739b 100644 --- a/src/utils/StreamResolver.test.ts +++ b/src/utils/StreamResolver.test.ts @@ -4,7 +4,7 @@ import { createExtractors, Extractor, ExtractorRegistry } from '../extractor'; import { StreamResolver } from './StreamResolver'; import { Source, SourceResult, MeineCloud, MostraGuarda } from '../source'; import { BlockedReason, Context, CountryCode, UrlResult } from '../types'; -import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TimeoutError, TooManyRequestsError } from '../error'; +import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TimeoutError, TooManyTimeoutsError, TooManyRequestsError } from '../error'; import { ImdbId } from './id'; import { FetcherMock } from './FetcherMock'; @@ -115,6 +115,16 @@ describe('resolve', () => { countryCode: CountryCode.de, }, }, + { + url: new URL('https://example.com'), + isExternal: true, + error: new TooManyTimeoutsError(), + label: 'hoster.com', + sourceId: '', + meta: { + countryCode: CountryCode.de, + }, + }, { url: new URL('https://example2.com'), isExternal: true, diff --git a/src/utils/StreamResolver.ts b/src/utils/StreamResolver.ts index 6bfbdf0..83a6a91 100644 --- a/src/utils/StreamResolver.ts +++ b/src/utils/StreamResolver.ts @@ -3,7 +3,15 @@ import winston from 'winston'; import bytes from 'bytes'; import { Context, UrlResult } from '../types'; import { Source } from '../source'; -import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TimeoutError, TooManyRequestsError } from '../error'; +import { + BlockedError, + HttpError, + NotFoundError, + QueueIsFullError, + TimeoutError, + TooManyTimeoutsError, + TooManyRequestsError, +} from '../error'; import { flagFromCountryCode, languageFromCountryCode } from './language'; import { envGetAppName } from './env'; import { Id } from './id'; @@ -169,6 +177,12 @@ export class StreamResolver { return '🚦 Request was rate-limited. Please try again later or consider self-hosting.'; } + if (error instanceof TooManyTimeoutsError) { + this.logger.warn(`${source}: Too many timeouts.`, ctx); + + return '🚦 Too many recent timeouts. Please try again later.'; + } + if (error instanceof TimeoutError) { this.logger.warn(`${source}: Request timed out.`, ctx); diff --git a/src/utils/__snapshots__/StreamResolver.test.ts.snap b/src/utils/__snapshots__/StreamResolver.test.ts.snap index 07f803f..a4814d9 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`] = ` "title": "🌐 German 🇩🇪 🔗 hoster.com 🚦 Request was rate-limited. Please try again later or consider self-hosting.", + }, + { + "behaviorHints": {}, + "externalUrl": "https://example.com/", + "name": "WebStreamr N/A ⚠️ external", + "title": "🌐 German 🇩🇪 +🔗 hoster.com +🚦 Too many recent timeouts. Please try again later.", }, { "behaviorHints": {}, @@ -95,6 +103,14 @@ exports[`resolve adds error info 2`] = ` "name": "WebStreamr N/A", "title": "🌐 German 🇩🇪 🔗 hoster.com +🚦 Too many recent timeouts. Please try again later.", + "ytId": "E4WlUXrJgy4", + }, + { + "behaviorHints": {}, + "name": "WebStreamr N/A", + "title": "🌐 German 🇩🇪 +🔗 hoster.com ❌ Request failed. Request-id: id.", "ytId": "E4WlUXrJgy4", },