الـ Concurrency Limits والـ Bulkheads: dependency بطيئة ما توقعش الـ app كله

// load.js: أربعين user فتحوا صفحة فيها recommendations في نفس الوقت، وخمسة بيدفعوا
const B = "http://localhost:3000";
const t = async (p) => { const s = performance.now(); const r = await fetch(B + p); await r.text(); return { status: r.status, ms: performance.now() - s }; };
await t("/checkout?order=0"); // warm-up

const recsP = Array.from({ length: 40 }, (_, i) => t(`/recs?user=u${i}`));
await new Promise((r) => setTimeout(r, 100));
const pays = [];
for (let i = 1; i <= 5; i++) pays.push(await t(`/checkout?order=${i}`));
const rs = await Promise.all(recsP);
const count = (s) => rs.filter((r) => r.status === s).length;
console.log(`checkout: ${pays.map((p) => `${p.status} في ${p.ms.toFixed(0)}ms`).join(" | ")}`);
console.log(`recs: ${count(200)} رجعوا 200، و ${count(503)} رجعوا 503 | أبطأ واحد ${Math.max(...rs.map((r) => r.ms)).toFixed(0)}ms`);

$ node server.js & S=$!; sleep 1.5; node load.js; kill $S
on 3000
checkout: 200 في 3914ms | 200 في 2ms | 200 في 2ms | 200 في 2ms | 200 في 2ms
recs: 40 رجعوا 200، و 0 رجعوا 503 | أبطأ واحد 4040ms

// limiter.js: concurrency limiter بطابور محدود و timeout، و middleware بيرجّع 503، و Little's law
export class LimitError extends Error {
  constructor(code) { super(code); this.code = code; }
}

export function createLimiter({ concurrency, maxQueue = 0, queueTimeoutMs = Infinity }) {
  let active = 0, rejected = 0;
  const queue = []; // كل واحد: { start, timer }

  const next = () => {
    active--;
    const w = queue.shift();
    if (w) { clearTimeout(w.timer); active++; w.start(); }
  };

  const run = (fn) => new Promise((resolve, reject) => {
    const start = () => Promise.resolve().then(fn).finally(next).then(resolve, reject);
    if (active < concurrency) { active++; return start(); }
    if (queue.length >= maxQueue) { rejected++; return reject(new LimitError("OVERLOADED")); }
    const w = { start };
    if (Number.isFinite(queueTimeoutMs)) {
      w.timer = setTimeout(() => {
        queue.splice(queue.indexOf(w), 1);
        rejected++;
        reject(new LimitError("QUEUE_TIMEOUT"));
      }, queueTimeoutMs);
    }
    queue.push(w);
  });

  return { run, stats: () => ({ active, queued: queue.length, rejected }) };
}

// الـ handler بيشتغل جُوّه الـ limiter، والـ slot بيتساب لما الـ response يخلص
export const limitMiddleware = (limiter, { retryAfterSec = 1 } = {}) => (req, res, next) => {
  limiter
    .run(() => new Promise((done) => { res.on("close", done); next(); }))
    .catch((e) => {
      if (!(e instanceof LimitError)) return next(e);
      if (res.headersSent) return;
      res.set("Retry-After", String(retryAfterSec)).status(503).json({ error: e.code });
    });
};

// L = λ × W: عدد الشغل اللي في النص = المعدل في التانية × الوقت بالتواني
export const littlesLaw = ({ rps, latencyMs }) => Math.ceil((rps * latencyMs) / 1000);

$ node server.js & S=$!; sleep 1.5; node load.js; kill $S
on 3000
checkout: 200 في 8ms | 200 في 2ms | 200 في 2ms | 200 في 2ms | 200 في 2ms
recs: 6 رجعوا 200، و 34 رجعوا 503 | أبطأ واحد 2051ms

// server.js: bulkheads: كل route ليه limiter، والـ recs ما تقدرش تاخد أكتر من تلات connections من العشرة
import express from "express";
import { recs, checkout, pool } from "./db.js";
import { createLimiter, limitMiddleware } from "./limiter.js";

const recsLimit = createLimiter({ concurrency: 3, maxQueue: 6, queueTimeoutMs: 1500 });
const payLimit = createLimiter({ concurrency: 6, maxQueue: 50, queueTimeoutMs: 3000 });

const app = express();
app.get("/recs", limitMiddleware(recsLimit, { retryAfterSec: 2 }), async (req, res) => res.json((await recs(req.query.user)).rows[0]));
app.get("/checkout", limitMiddleware(payLimit), async (req, res) => res.json((await checkout(Number(req.query.order))).rows[0]));
app.get("/stats", (req, res) => res.json({ recs: recsLimit.stats(), checkout: payLimit.stats(), pool: { total: pool.totalCount, waiting: pool.waitingCount } }));
app.listen(3000, () => console.log("on 3000"));

// db.js: pool واحد لكل الـ app: عشر connections بس (زي أي database حقيقية: الـ connections ليها حد)
import pg from "pg";

export const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL, max: 10 });
// recommendations: query تقيلة (analytics) بتاخد تانية. checkout: query سريعة
export const recs = (user) => pool.query("SELECT pg_sleep(1), $1::text AS user, 'منتجات مقترحة' AS items", [user]);
export const checkout = (order) => pool.query("SELECT $1::int AS order_id, 'paid' AS status", [order]);

import { LimitError, createLimiter, littlesLaw } from "./limiter.js";

let ok = 0, all = 0;
const check = (name, cond, extra = "") => { all++; if (cond) ok++; console.log(cond ? "✅" : "❌", name, extra); };
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const B = "http://localhost:3000";
const err = (p) => p.then(() => null, (e) => e);

// unit: الـ limiter لوحده
{
  const l = createLimiter({ concurrency: 2, maxQueue: 10 });
  let now = 0, max = 0;
  const job = async () => { now++; max = Math.max(max, now); await sleep(40); now--; return "تمام"; };
  const out = await Promise.all(Array.from({ length: 6 }, () => l.run(job)));
  check("concurrency اتنين: عمره ما عدّى اتنين، والستة خلصوا", max === 2 && out.every((x) => x === "تمام"), `max ${max}`);
}
{
  const l = createLimiter({ concurrency: 1, maxQueue: 1 });
  const a = l.run(() => sleep(60)), b = l.run(() => sleep(10));
  const t0 = performance.now();
  const e = await err(l.run(() => "مش هيشتغل"));
  check("الطابور مليان ← OVERLOADED على طول", e instanceof LimitError && e.code === "OVERLOADED" && performance.now() - t0 < 20);
  check("stats: active واحد و queued واحد و rejected واحد", JSON.stringify(l.stats()) === JSON.stringify({ active: 1, queued: 1, rejected: 1 }), JSON.stringify(l.stats()));
  await Promise.all([a, b]);
}
{
  const l = createLimiter({ concurrency: 1, maxQueue: 5, queueTimeoutMs: 50 });
  const a = l.run(() => sleep(200));
  const t0 = performance.now();
  const e = await err(l.run(() => "متأخر"));
  const waited = performance.now() - t0;
  check("استنى أكتر من queueTimeoutMs ← QUEUE_TIMEOUT", e?.code === "QUEUE_TIMEOUT" && waited >= 45 && waited < 150, `${waited.toFixed(0)}ms`);
  check("بعد الـ timeout الطابور فاضي", l.stats().queued === 0);
  await a;
}
{
  const l = createLimiter({ concurrency: 1, maxQueue: 10 });
  const order = [];
  await Promise.all([1, 2, 3, 4].map((i) => l.run(async () => { await sleep(5); order.push(i); })));
  check("الطابور FIFO", order.join() === "1,2,3,4", order.join());
}
{
  const l = createLimiter({ concurrency: 1 });
  const e = await err(l.run(() => { throw new Error("وقع"); }));
  const v = await l.run(() => 42);
  check("fn بتعمل throw ← نفس الـ error، والـ slot بيتساب", e?.message === "وقع" && v === 42 && l.stats().active === 0);
}
check("Little's law: مِيّتين في التانية × خمسين ms ← عشرة، و سبعة × مِيّة وخمسين ← اتنين", littlesLaw({ rps: 200, latencyMs: 50 }) === 10 && littlesLaw({ rps: 7, latencyMs: 150 }) === 2);

// integration: recs بتغرق، و checkout لازم يفضل سريع
await fetch(`${B}/checkout?order=0`);
const t = async (p) => { const s = performance.now(); const r = await fetch(B + p); return { status: r.status, ms: performance.now() - s, ra: r.headers.get("retry-after"), body: await r.json() }; };
const flood = Promise.all(Array.from({ length: 40 }, (_, i) => t(`/recs?user=t${i}`)));
await sleep(100);
const pays = [];
for (let i = 1; i <= 5; i++) pays.push(await t(`/checkout?order=${i}`));
const rs = await flood;
che
…