spb/rareindex

1/**2 * Per-host fault isolation (SPEC §26): a concurrency gate, a politeness interval and a circuit3 * breaker. One broken or hostile domain must never stall the whole ingestion pipeline — after4 * `failures` consecutive failures the circuit opens and requests fail fast until the cooldown ends.5 * A single trial request is allowed when the cooldown elapses (half-open); success closes it.6 */7export interface HostGateOptions {8  concurrency: number;9  minIntervalMs: number;10  circuitFailures: number;11  circuitCooldownMs: number;12}1314export type CircuitState = 'closed' | 'open' | 'half_open';1516interface HostState {17  inFlight: number;18  waiters: Array<() => void>;19  lastStart: number;20  consecutiveFailures: number;21  openedAt: number | null;22  state: CircuitState;23  trialInFlight: boolean;24}2526export class CircuitOpenError extends Error {27  constructor(public host: string, public retryAt: Date) {28    super(`circuit open for ${host} until ${retryAt.toISOString()}`);29  }30}3132export class HostGates {33  private hosts = new Map<string, HostState>();34  constructor(private resolve: (host: string) => HostGateOptions) {}3536  private get(host: string): HostState {37    let s = this.hosts.get(host);38    if (!s) {39      s = { inFlight: 0, waiters: [], lastStart: 0, consecutiveFailures: 0, openedAt: null, state: 'closed', trialInFlight: false };40      this.hosts.set(host, s);41    }42    return s;43  }4445  snapshot(host: string): { state: CircuitState; consecutiveFailures: number; inFlight: number; retryAt: Date | null } {46    const s = this.get(host);47    const o = this.resolve(host);48    return { state: s.state, consecutiveFailures: s.consecutiveFailures, inFlight: s.inFlight, retryAt: s.openedAt ? new Date(s.openedAt + o.circuitCooldownMs) : null };49  }5051  /** Acquire a slot; throws CircuitOpenError immediately when the circuit is open. */52  async acquire(host: string): Promise<() => void> {53    const s = this.get(host);54    const o = this.resolve(host);55    if (s.state === 'open') {56      if (s.openedAt !== null && Date.now() - s.openedAt >= o.circuitCooldownMs) {57        s.state = 'half_open';58      } else {59        throw new CircuitOpenError(host, new Date((s.openedAt ?? Date.now()) + o.circuitCooldownMs));60      }61    }62    if (s.state === 'half_open') {63      if (s.trialInFlight) throw new CircuitOpenError(host, new Date(Date.now() + 5_000));64      s.trialInFlight = true;65    }66    while (s.inFlight >= o.concurrency) await new Promise<void>((r) => s.waiters.push(r));67    s.inFlight++;68    const wait = s.lastStart + o.minIntervalMs - Date.now();69    if (wait > 0) await new Promise((r) => setTimeout(r, wait));70    s.lastStart = Date.now();71    let released = false;72    return () => {73      if (released) return;74      released = true;75      s.inFlight--;76      s.waiters.shift()?.();77    };78  }7980  /** Report the outcome of a request made under `acquire`. */81  report(host: string, ok: boolean): void {82    const s = this.get(host);83    const o = this.resolve(host);84    if (ok) {85      s.consecutiveFailures = 0;86      s.openedAt = null;87      s.state = 'closed';88      s.trialInFlight = false;89      return;90    }91    s.consecutiveFailures++;92    if (s.state === 'half_open') {93      s.trialInFlight = false;94      s.state = 'open';95      s.openedAt = Date.now();96      return;97    }98    if (s.consecutiveFailures >= o.circuitFailures) {99      s.state = 'open';100      s.openedAt = Date.now();101    }102  }103104  /** Manually reset a host (admin "resume"). */105  reset(host: string): void {106    this.hosts.delete(host);107  }108109  hostsSnapshot(): Array<{ host: string; state: CircuitState; consecutiveFailures: number; inFlight: number }> {110    return [...this.hosts.entries()].map(([host, s]) => ({ host, state: s.state, consecutiveFailures: s.consecutiveFailures, inFlight: s.inFlight }));111  }112}113