Backpressure and Load Shedding

// Express middleware that sheds the cheapest work first, using Google's
// criticality ladder as the ordering and a local health signal as the input.
const CRITICALITY = {
  CRITICAL_PLUS: 0,   // user-visible and unrecoverable: a payment
  CRITICAL: 1,        // user-visible: loading the page
  SHEDDABLE_PLUS: 2,  // batch work that tolerates partial failure
  SHEDDABLE: 3,       // prefetches, analytics beacons, speculative work
};

function criticalityOf(req) {
  if (req.method === 'POST' && req.path.startsWith('/checkout')) return CRITICALITY.CRITICAL_PLUS;
  if (req.get('x-request-source') === 'batch') return CRITICALITY.SHEDDABLE_PLUS;
  if (req.path.startsWith('/beacon')) return CRITICALITY.SHEDDABLE;
  return CRITICALITY.CRITICAL;
}

// Shed from the bottom up as pressure rises, so the system degrades in a
// known order instead of dropping whatever happened to arrive during a spike.
function sheddingLevel() {
  const lagMs = loopDelayHistogram.percentile(99) / 1e6;
  if (lagMs > 500 || admissions.queue.length > 240) return CRITICALITY.CRITICAL;
  if (lagMs > 200 || admissions.queue.length > 160) return CRITICALITY.SHEDDABLE_PLUS;
  if (lagMs > 100 || admissions.queue.length > 80) return CRITICALITY.SHEDDABLE;
  return 99; // shed nothing
}

app.use((req, res, next) => {
  const level = sheddingLevel();
  if (criticalityOf(req) >= level) {
    shedCounter.inc({ path: req.path }); // ALWAYS count what you shed
    // 503 with Retry-After: this is overload, not a client error, so do not
    // use 429 here. 429 means "you specifically are over your limit".
    return res.set('Retry-After', '2').status(503).json({ error: 'overloaded' });
  }
  next();
});

// Shedding at the server still costs the server a request. Google's SRE book
// pushes part of the job into the client: once a backend starts rejecting,
// the client rejects locally and the request never leaves the process.
const WINDOW_MS = 120_000; // the book uses a 2-minute window
const K = 2;               // recommended multiplier

class AdaptiveThrottle {
  constructor() { this.events = []; }

  prune(now) {
    const cutoff = now - WINDOW_MS;
    while (this.events.length && this.events[0].at < cutoff) this.events.shift();
  }

  record(accepted) { this.events.push({ at: Date.now(), accepted }); }

  // P(reject) = max(0, (requests - K * accepts) / (requests + 1))
  rejectionProbability() {
    const now = Date.now();
    this.prune(now);
    const requests = this.events.length;
    const accepts = this.events.reduce((n, e) => n + (e.accepted ? 1 : 0), 0);
    return Math.max(0, (requests - K * accepts) / (requests + 1));
  }

  async call(fn) {
    if (Math.random() < this.rejectionProbability()) {
      // Local rejection: no connection, no backend work, no queue slot.
      throw Object.assign(new Error('throttled locally'), { code: 'CLIENT_THROTTLED' });
    }
    try {
      const out = await fn();
      this.record(true);
      return out;
    } catch (err) {
      // A backend rejection (429/503) is NOT an accept. A 404 is: the backend
      // did the work and answered, so it must not throttle us.
      this.record(!(err.status === 429 || err.status === 503));
      throw err;
    }
  }
}

// Node streams have backpressure built in, and the shorthand hides it.
// LONGHAND: writable.write() returns false when the buffer is past
// highWaterMark; you must stop and wait for 'drain'.
function writeAll(rows, out, done) {
  let i = 0;
  (function pump() {
    while (i < rows.length) {
      const ok = out.write(JSON.stringify(rows[i++]) + '\n');
      if (!ok) {
        // Ignoring this return value is the classic leak: the loop keeps
        // buffering in memory at the speed of the producer, not the consumer.
        out.once('drain', pump);
        return;
      }
    }
    out.end(done);
  })();
}

// SHORTHAND: pipeline() does exactly the same thing, plus error and cleanup
// propagation in both directions. stream/promises is Node v15.0.0+.
const { pipeline } = require('node:stream/promises');
const { Readable } = require('node:stream');

await pipeline(
  Readable.from(rows),                          // async iterables also work
  async function* (source) {
    for await (const row of source) yield JSON.stringify(row) + '\n';
  },
  out,
);