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

@@ -0,0 +1 @@
+export class TooManyTimeoutsError extends Error {}

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';

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);
+  });
 });

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<string, HttpCacheItem>;
-  private readonly rateLimitedCache: TTLCache<string, undefined>;
+  private readonly httpCache = new TTLCache<string, HttpCacheItem>();
+  private readonly rateLimitedCache = new TTLCache<string, undefined>();
   private readonly semaphores = new Map<string, SemaphoreInterface>();
   private readonly hostUserAgentMap = new Map<string, string>();
   private readonly cookieJar = new CookieJar();
 
+  private readonly timeoutsCountCache = new TTLCache<string, number>();
+  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<string> {
@@ -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);
 

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,

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);
 

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",
     },