รูปแบบ Circuit Breaker และการแยกทรัพยากรด้วย Bulkhead

async function callDependency(attempt) {
  // Simulate a dependency that fails 70% of the time
  if (Math.random() < 0.7) {
    throw new Error('dependency timeout');
  }
  return 'ok';
}

async function withBlindRetry(maxRetries) {
  let physicalCalls = 0;
  for (let i = 0; i <= maxRetries; i++) {
    physicalCalls++;
    try {
      const res = await callDependency(i);
      console.log(`Success after ${physicalCalls} physical call(s)`);
      return res;
    } catch (e) {
      console.log(`Attempt ${i + 1} failed: ${e.message}`);
    }
  }
  console.log(`Gave up after ${physicalCalls} physical calls`);
}

withBlindRetry(3);

class CircuitBreaker {
  constructor(fn, { threshold = 3, resetMs = 5000 } = {}) {
    this.fn = fn;
    this.threshold = threshold;
    this.resetMs = resetMs;
    this.failures = 0;
    this.state = 'CLOSED';
    this.nextTry = 0;
  }

  async exec(...args) {
    if (this.state === 'OPEN') {
      if (Date.now() < this.nextTry) {
        throw new Error('Circuit OPEN - failing fast');
      }
      this.state = 'HALF_OPEN';
    }
    try {
      const result = await this.fn(...args);
      this.failures = 0;
      this.state = 'CLOSED';
      return result;
    } catch (err) {
      this.failures++;
      if (this.failures >= this.threshold) {
        this.state = 'OPEN';
        this.nextTry = Date.now() + this.resetMs;
      }
      throw err;
    }
  }
}

let n = 0;
const flaky = async () => { n++; if (n <= 5) throw new Error('boom'); return 'ok'; };
const breaker = new CircuitBreaker(flaky, { threshold: 3, resetMs: 1000 });

(async () => {
  for (let i = 0; i < 4; i++) {
    try { console.log(await breaker.exec()); }
    catch (e) { console.log(`call ${i}: ${e.message} [state=${breaker.state}]`); }
  }
})();

function estimate(reqs, timeoutMs, openAfter) {
  let totalMs = 0;
  for (let i = 0; i < reqs; i++) {
    if (i < openAfter) {
      totalMs += timeoutMs; // these waited for the full timeout
    } else {
      totalMs += 0.01; // breaker rejected instantly
    }
  }
  return totalMs;
}

const reqs = 100, timeout = 2000, openAfter = 5;
const withBreaker = estimate(reqs, timeout, openAfter);
const noBreaker = reqs * timeout;
console.log(`No breaker: ${noBreaker} ms of blocked time`);
console.log(`With breaker: ${withBreaker.toFixed(2)} ms of blocked time`);
console.log(`Saved: ${(noBreaker - withBreaker).toFixed(0)} ms`);

const CircuitBreaker = require('opossum');

async function getInventory(sku) {
  const res = await fetch(`http://inventory.internal/items/${sku}`);
  if (!res.ok) throw new Error(`HTTP ${res.status}`);
  return res.json();
}

const breaker = new CircuitBreaker(getInventory, {
  timeout: 3000,                 // fail a call after 3s
  errorThresholdPercentage: 50,  // open at 50% failures in the window
  resetTimeout: 10000,           // try again after 10s
  rollingCountTimeout: 10000,    // 10s stats window
});

// Serve stale/cached data instead of erroring out
breaker.fallback((sku) => ({ sku, stock: 'unknown', cached: true }));

breaker.on('open', () => console.warn('inventory breaker OPEN'));
breaker.on('halfOpen', () => console.info('inventory breaker probing'));

module.exports = (sku) => breaker.fire(sku);

class Bulkhead {
  constructor(maxConcurrent, maxQueue = 0) {
    this.max = maxConcurrent;
    this.maxQueue = maxQueue;
    this.active = 0;
    this.queue = [];
  }

  async run(task) {
    if (this.active >= this.max) {
      if (this.queue.length >= this.maxQueue) {
        throw new Error('Bulkhead full - rejected');
      }
      await new Promise((resolve) => this.queue.push(resolve));
    }
    this.active++;
    try {
      return await task();
    } finally {
      this.active--;
      const next = this.queue.shift();
      if (next) next();
    }
  }
}

const bh = new Bulkhead(2, 2);
const slow = (id) => () => new Promise((r) => setTimeout(() => { console.log('done', id); r(id); }, 50));

(async () => {
  const results = await Promise.allSettled(
    [1, 2, 3, 4, 5, 6].map((id) => bh.run(slow(id)))
  );
  results.forEach((r, i) =>
    console.log(`task ${i + 1}: ${r.status}${r.reason ? ' - ' + r.reason.message : ''}`)
  );
})();

const http = require('http');

// One isolated pool per downstream service
const paymentsAgent = new http.Agent({
  keepAlive: true,
  maxSockets: 10,        // at most 10 concurrent connections to payments
  maxFreeSockets: 5,
});

const inventoryAgent = new http.Agent({
  keepAlive: true,
  maxSockets: 20,        // inventory gets its own, independent budget
});

function callPayments(path) {
  return new Promise((resolve, reject) => {
    const req = http.request(
      { host: 'payments.internal', path, agent: paymentsAgent, timeout: 3000 },
      (res) => { res.resume(); res.on('end', resolve); }
    );
    req.on('timeout', () => req.destroy(new Error('payments timeout')));
    req.on('error', reject);
    req.end();
  });
}

module.exports = { callPayments, paymentsAgent, inventoryAgent };

// Compose: breaker(bulkhead(timeout(call)))
function withTimeout(fn, ms) {
  return (...args) => Promise.race([
    fn(...args),
    new Promise((_, rej) => setTimeout(() => rej(new Error('timeout')), ms)),
  ]);
}

function resilient(rawCall, { bulkhead, breaker, timeoutMs }) {
  const timed = withTimeout(rawCall, timeoutMs);
  // breaker on the OUTSIDE: it can fail fast before we ever take a bulkhead slot
  return (...args) => breaker.exec(() => bulkhead.run(() => timed(...args)));
}

async function demo() {
  await withTimeout(() => Promise.resolve('fast'), 50)().then(console.log);
  try { await withTimeout(() => new Promise(() => {}), 30)(); }
  catch (e) { console.log('slow call ->', e.message); }
  console.log('Compose order: breaker -> bulkhead -> timeout -> rawCall');
}
demo();