spb/cancerindex

1import { setTimeout as sleep } from 'node:timers/promises';2import type { ConnectorManifest } from './manifest.js';34export interface HttpStats {5  requests: number;6  failures: number;7  rateLimitEvents: number;8  retries: number;9}1011export class BodyTimeoutError extends Error {12  constructor(13    public readonly url: string,14    public readonly timeoutMs: number,15  ) {16    super(`body read timed out after ${timeoutMs} ms for ${url}`);17    this.name = 'BodyTimeoutError';18  }19}2021export class HttpError extends Error {22  constructor(23    public readonly status: number,24    public readonly url: string,25    public readonly bodySnippet: string,26  ) {27    super(`HTTP ${status} for ${url}: ${bodySnippet.slice(0, 200)}`);28  }29}3031/** Token bucket rate limiter (CLAUDE.md §229). */32class TokenBucket {33  private tokens: number;34  private last = Date.now();35  constructor(36    private readonly ratePerSec: number,37    private readonly burst: number,38  ) {39    this.tokens = burst;40  }41  async take(): Promise<void> {42    for (;;) {43      const now = Date.now();44      this.tokens = Math.min(this.burst, this.tokens + ((now - this.last) / 1000) * this.ratePerSec);45      this.last = now;46      if (this.tokens >= 1) {47        this.tokens -= 1;48        return;49      }50      await sleep(Math.ceil(((1 - this.tokens) / this.ratePerSec) * 1000));51    }52  }53}5455class Semaphore {56  private queue: Array<() => void> = [];57  private active = 0;58  constructor(private readonly max: number) {}59  async acquire(): Promise<() => void> {60    if (this.active >= this.max) await new Promise<void>((resolve) => this.queue.push(resolve));61    this.active++;62    let released = false;63    return () => {64      if (released) return;65      released = true;66      this.active--;67      const next = this.queue.shift();68      if (next) next();69    };70  }71}7273export interface HttpClientOptions {74  userAgent?: string;75  timeoutMs?: number;76  headers?: Record<string, string>;77}7879/**80 * Source-aware HTTP client: token bucket, bounded concurrency, exponential backoff with jitter,81 * Retry-After respect, request accounting for the ingest run.82 */83export class HttpClient {84  readonly stats: HttpStats = { requests: 0, failures: 0, rateLimitEvents: 0, retries: 0 };85  private readonly bucket: TokenBucket | null;86  private readonly sem: Semaphore;87  private readonly retry: NonNullable<ConnectorManifest['retryPolicy']>;88  private readonly ua: string;89  private readonly timeoutMs: number;90  private readonly headers: Record<string, string>;9192  constructor(manifest: ConnectorManifest, opts: HttpClientOptions = {}) {93    const rps = manifest.rateLimits.requestsPerSecond ?? (manifest.rateLimits.requestsPerMinute ? manifest.rateLimits.requestsPerMinute / 60 : undefined);94    this.bucket = rps ? new TokenBucket(rps, Math.max(1, Math.ceil(rps))) : null;95    this.sem = new Semaphore(manifest.rateLimits.maxConcurrency);96    this.retry = manifest.retryPolicy;97    // Plain product token only: some WAFs (e.g. oncotree.mskcc.org, Akamai) reject UAs containing URLs/parentheses.98    this.ua = opts.userAgent ?? `CancerIndex/0.1`;99    this.timeoutMs = opts.timeoutMs ?? 60_000;100    this.headers = opts.headers ?? {};101  }102103  /** One attempt: returns the Response (any status) or throws on network error. */104  private async attempt(url: string, init: RequestInit): Promise<Response> {105    const release = await this.sem.acquire();106    try {107      if (this.bucket) await this.bucket.take();108      this.stats.requests++;109      const controller = new AbortController();110      const timer = setTimeout(() => controller.abort(), this.timeoutMs);111      try {112        return await fetch(url, {113          ...init,114          headers: { 'user-agent': this.ua, accept: 'application/json, text/plain, */*', ...this.headers, ...(init.headers as Record<string, string> | undefined) },115          signal: controller.signal,116        });117      } finally {118        clearTimeout(timer);119      }120    } finally {121      release();122    }123  }124125  async request(url: string, init: RequestInit = {}): Promise<Response> {126    for (let attempt = 0; ; attempt++) {127      let res: Response;128      try {129        res = await this.attempt(url, init);130      } catch (err) {131        if (attempt >= this.retry.maxRetries) {132          this.stats.failures++;133          throw err;134        }135        this.stats.retries++;136        await sleep(this.backoff(attempt));137        continue;138      }139      if (res.ok) return res;140      const body = await res.text().catch(() => '');141      if (res.status === 429) this.stats.rateLimitEvents++;142      const retryable = res.status === 429 || res.status >= 500 || res.status === 408;143      if (!retryable || attempt >= this.retry.maxRetries) {144        this.stats.failures++;145        throw new HttpError(res.status, url, body);146      }147      const retryAfter = Number(res.headers.get('retry-after'));148      const delay = Number.isFinite(retryAfter) && retryAfter > 0 ? Math.min(retryAfter * 1000, 120_000) : this.backoff(attempt);149      this.stats.retries++;150      await sleep(delay);151    }152  }153154  /**155   * Read a body under the same time budget as the headers. Without this a stalled body (server156   * accepted the request, sent headers, then went silent — observed with ChEMBL) hangs forever157   * because the AbortController is cleared once headers arrive.158   */159  private async readBody<T>(res: Response, url: string, read: (r: Response) => Promise<T>): Promise<T> {160    let timer: ReturnType<typeof setTimeout> | undefined;161    const timeout = new Promise<never>((_, reject) => {162      timer = setTimeout(() => {163        res.body?.cancel().catch(() => {});164        reject(new BodyTimeoutError(url, this.timeoutMs));165      }, this.timeoutMs);166    });167    try {168      return await Promise.race([read(res), timeout]);169    } finally {170      clearTimeout(timer);171    }172  }173174  /** Request + body read with retries on body stalls (same backoff as request()). */175  private async withBody<T>(url: string, init: RequestInit | undefined, read: (r: Response) => Promise<T>): Promise<T> {176    for (let attempt = 0; ; attempt++) {177      const res = await this.request(url, init);178      try {179        return await this.readBody(res, url, read);180      } catch (err) {181        if (!(err instanceof BodyTimeoutError) || attempt >= this.retry.maxRetries) {182          this.stats.failures++;183          throw err;184        }185        this.stats.retries++;186        await sleep(this.backoff(attempt));187      }188    }189  }190191  async json<T = unknown>(url: string, init?: RequestInit): Promise<T> {192    return this.withBody(url, init, (r) => r.json() as Promise<T>);193  }194195  async text(url: string, init?: RequestInit): Promise<string> {196    return this.withBody(url, init, (r) => r.text());197  }198199  async postJson<T = unknown>(url: string, body: unknown, init: RequestInit = {}): Promise<T> {200    return this.json<T>(url, { ...init, method: 'POST', headers: { 'content-type': 'application/json', ...(init.headers as Record<string, string> | undefined) }, body: JSON.stringify(body) });201  }202203  private backoff(attempt: number): number {204    const base = Math.min(this.retry.maxDelayMs, this.retry.baseDelayMs * 2 ** attempt);205    return Math.round(base / 2 + Math.random() * (base / 2));206  }207}208