chore: implement rate-limit detection

src/error/TooManyRequestsError.ts

@@ -0,0 +1,9 @@
+export class TooManyRequestsError extends Error {
+  public readonly retryAfter: number;
+
+  public constructor(retryAfter: number) {
+    super();
+
+    this.retryAfter = retryAfter;
+  }
+}

src/error/index.ts

@@ -2,3 +2,4 @@ export * from './BlockedError';
 export * from './HttpError';
 export * from './NotFoundError';
 export * from './QueueIsFullError';
+export * from './TooManyRequestsError';

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 } from '../error';
+import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TooManyRequestsError } from '../error';
 fetchMock.mockGlobal();
 
 const fetcher = new Fetcher(winston.createLogger({ transports: [new winston.transports.Console({ level: 'nope' })] }));
@@ -130,6 +130,37 @@ describe('fetch', () => {
     }
   });
 
+  test('converts 429 to custom TooManyRequestsError', async () => {
+    fetchMock.get('https://some-rate-limited-url.test/', { status: 429 });
+
+    try {
+      await fetcher.text(ctx, new URL('https://some-rate-limited-url.test/'));
+      fail();
+    } catch (error) {
+      expect(error).toBeInstanceOf(TooManyRequestsError);
+      expect(error).toMatchObject({ retryAfter: NaN });
+    }
+  });
+
+  test('converts 429 to custom TooManyRequestsError and blocks consecutive requests', async () => {
+    fetchMock.get('https://some-rate-limited-retry-after-url.test/', { status: 429, headers: { 'retry-after': '10' } });
+
+    try {
+      await fetcher.text(ctx, new URL('https://some-rate-limited-retry-after-url.test/'));
+      fail();
+    } catch (error) {
+      expect(error).toBeInstanceOf(TooManyRequestsError);
+      expect(error).toMatchObject({ retryAfter: 10 });
+    }
+
+    try {
+      await fetcher.text(ctx, new URL('https://some-rate-limited-retry-after-url.test/'));
+      fail();
+    } catch (error) {
+      expect(error).toBeInstanceOf(TooManyRequestsError);
+    }
+  });
+
   test('passes through other error as HttpError', async () => {
     fetchMock.get('https://some-error-url.test/', { status: 500, headers: { 'x-foo': 'bar' } });
 

src/utils/Fetcher.ts

@@ -4,7 +4,7 @@ import winston from 'winston';
 import { Mutex } from 'async-mutex';
 import { Cookie, CookieJar } from 'tough-cookie';
 import { BlockedReason, Context, TIMEOUT } from '../types';
-import { BlockedError, HttpError, NotFoundError, QueueIsFullError } from '../error';
+import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TooManyRequestsError } from '../error';
 import { clearTimeout } from 'node:timers';
 import { envGet } from './env';
 
@@ -51,6 +51,7 @@ export class Fetcher {
   private readonly logger: winston.Logger;
 
   private readonly httpCache: TTLCache<string, HttpCacheItem>;
+  private readonly rateLimitedCache: TTLCache<string, undefined>;
   private readonly hostUserAgentMap = new Map<string, string>();
   private readonly cookieJar = new CookieJar();
 
@@ -61,6 +62,7 @@ export class Fetcher {
     this.logger = logger;
 
     this.httpCache = new TTLCache();
+    this.rateLimitedCache = new TTLCache();
   }
 
   public async text(ctx: Context, url: URL, init?: CustomRequestInit): Promise<string> {
@@ -155,6 +157,15 @@ export class Fetcher {
       throw new BlockedError(BlockedReason.unknown, httpCacheItem.policy.responseHeaders());
     }
 
+    if (httpCacheItem.status === 429) {
+      const retryAfter = parseInt(`${httpCacheItem.policy.responseHeaders()['retry-after']}`);
+      if (!isNaN(retryAfter)) {
+        this.rateLimitedCache.set(url.host, undefined, { ttl: retryAfter * 1000 });
+      }
+
+      throw new TooManyRequestsError(retryAfter);
+    }
+
     throw new HttpError(httpCacheItem.status, httpCacheItem.statusText, responseHeaders);
   };
 
@@ -205,6 +216,10 @@ export class Fetcher {
   protected async fetchWithTimeout(ctx: Context, url: URL, init?: CustomRequestInit): Promise<Response> {
     this.logger.info(`Fetch ${init?.method ?? 'GET'} ${url}`, ctx);
 
+    if (this.rateLimitedCache.has(url.host)) {
+      throw new TooManyRequestsError(this.rateLimitedCache.getRemainingTTL(url.host) / 1000);
+    }
+
     const controller = new AbortController();
     const timer = setTimeout(() => controller.abort(TIMEOUT), init?.timeout ?? 5000);
 

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, TIMEOUT, UrlResult } from '../types';
-import { BlockedError, HttpError, NotFoundError, QueueIsFullError } from '../error';
+import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TooManyRequestsError } from '../error';
 import { ImdbId } from './id';
 import { FetcherMock } from './FetcherMock';
 
@@ -105,6 +105,16 @@ describe('resolve', () => {
               countryCode: CountryCode.de,
             },
           },
+          {
+            url: new URL('https://example.com'),
+            isExternal: true,
+            error: new TooManyRequestsError(10),
+            label: 'hoster.com',
+            sourceId: '',
+            meta: {
+              countryCode: CountryCode.de,
+            },
+          },
           {
             url: new URL('https://example2.com'),
             isExternal: true,

src/utils/StreamResolver.ts

@@ -3,7 +3,7 @@ import winston from 'winston';
 import bytes from 'bytes';
 import { Context, TIMEOUT, UrlResult } from '../types';
 import { Source } from '../source';
-import { BlockedError, HttpError, NotFoundError, QueueIsFullError } from '../error';
+import { BlockedError, HttpError, NotFoundError, QueueIsFullError, TooManyRequestsError } from '../error';
 import { flagFromCountryCode, languageFromCountryCode } from './language';
 import { envGetAppName } from './env';
 import { Id } from './id';
@@ -163,6 +163,12 @@ export class StreamResolver {
       return '⚠️ Request was blocked.';
     }
 
+    if (error instanceof TooManyRequestsError) {
+      this.logger.warn(`${source}: Rate limited for ${error.retryAfter} seconds.`, ctx);
+
+      return '🚦 Request was rate-limited. Please try again later or consider self-hosting.';
+    }
+
     if (error === TIMEOUT) {
       this.logger.warn(`${source}: Request timed out.`, ctx);
 

src/utils/__snapshots__/StreamResolver.test.ts.snap

@@ -18,6 +18,14 @@ exports[`resolve adds error info 1`] = `
       "title": "🌐 German 🇩🇪
 🔗 hoster.com
 ⚠️ Request was blocked.",
+    },
+    {
+      "behaviorHints": {},
+      "externalUrl": "https://example.com/",
+      "name": "WebStreamr N/A ⚠️ external",
+      "title": "🌐 German 🇩🇪
+🔗 hoster.com
+🚦 Request was rate-limited. Please try again later or consider self-hosting.",
     },
     {
       "behaviorHints": {},
@@ -79,6 +87,14 @@ exports[`resolve adds error info 2`] = `
       "name": "WebStreamr N/A",
       "title": "🌐 German 🇩🇪
 🔗 hoster.com
+🚦 Request was rate-limited. Please try again later or consider self-hosting.",
+      "ytId": "E4WlUXrJgy4",
+    },
+    {
+      "behaviorHints": {},
+      "name": "WebStreamr N/A",
+      "title": "🌐 German 🇩🇪
+🔗 hoster.com
 ❌ Request failed. Request-id: id.",
       "ytId": "E4WlUXrJgy4",
     },