chore: implement queued fetching

src/error/QueueIsFullError.ts

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

src/error/index.ts

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

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

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<void>;
+  count: number;
+}
 
 export class Fetcher {
   private readonly logger: winston.Logger;
-  private readonly httpCache: TTLCache<string, HttpCacheItem>;
+  private readonly queueLimit: number;
+  private readonly timeout: number;
 
-  constructor(logger: winston.Logger) {
+  private readonly httpCache: TTLCache<string, HttpCacheItem>;
+  private readonly hostQueues = new Map<string, HostQueue>();
+
+  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<Response> => {
+    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<Response> => {
+    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<void>(
+      (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;
+  };
 }

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',
+            },
+          },
         ];
       };
     }

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

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",
+  },
 ]
 `;