Memcached Meta Protocol (mg/ms) Anti-Stampede & Stale-While-Revalidate

'use strict';
// Zero-Dependency Memcached 1.6+ Meta Protocol Anti-Stampede & SWR Helper (Node.js)
// Uses native 'mg' flags:
//   v   = return value        | t = return remaining TTL
//   N30 = auto-vivify stub w/ 30s lease on miss (returns 'W' to ONLY ONE winning caller)
//   R15 = proactive recache if remaining TTL < 15s (winner gets 'W' alongside live value!)

const net = require('node:net');

function sendMetaCommand(socket, line) {
  return new Promise((resolve, reject) => {
    let buf = Buffer.alloc(0);
    const onData = (chunk) => {
      buf = Buffer.concat([buf, chunk]);
      const nl = buf.indexOf(0x0d); // '\r\n'
      if (nl === -1) return;

      const header = buf.subarray(0, nl).toString('utf8');
      const parts = header.split(' ');
      const code = parts[0];

      if (code === 'VA') {
        const byteLen = Number(parts[1]);
        const totalNeeded = nl + 2 + byteLen + 2;
        if (buf.length < totalNeeded) return;
        cleanup();
        const flags = new Set(parts.slice(2));
        const body = buf.subarray(nl + 2, nl + 2 + byteLen).toString('utf8');
        resolve({ status: 'VA', flags, body });
      } else {
        cleanup();
        resolve({ status: code, flags: new Set(parts.slice(1)), body: null });
      }
    };
    const onErr = (err) => { cleanup(); reject(err); };
    const cleanup = () => {
      socket.off('data', onData);
      socket.off('error', onErr);
    };
    socket.on('data', onData);
    socket.on('error', onErr);
    socket.write(line + '\r\n');
  });
}

async function getWithStampedeGuard(socket, key, ttlSec, loaderFn) {
  // Request value, remaining TTL, 20s miss-lock lease (N20), and early refresh under 15s (R15)
  const res = await sendMetaCommand(socket, `mg ${key} v t N20 R15`);
  const wonRecacheLock = res.flags.has('W');
  const isStaleOrStub = res.flags.has('X') || res.body === '' || res.body === null;

  // Case 1: Hot hit approaching expiry — serve value immediately & refresh in background
  if (res.status === 'VA' && !isStaleOrStub) {
    if (wonRecacheLock) {
      Promise.resolve(loaderFn()).then((fresh) => {
        const payload = JSON.stringify(fresh);
        return sendMetaCommand(socket, `ms ${key} ${Buffer.byteLength(payload)} T${ttlSec}\r\n${payload}`);
      }).catch(() => {});
    }
    return JSON.parse(res.body);
  }

  // Case 2: Cache miss where THIS worker won 'W' token — populate cache once for everyone
  if (wonRecacheLock) {
    const fresh = await loaderFn();
    const payload = JSON.stringify(fresh);
    await sendMetaCommand(socket, `ms ${key} ${Buffer.byteLength(payload)} T${ttlSec}\r\n${payload}`);
    return fresh;
  }

  // Case 3: Concurrent loser received 'Z' token — brief jittered wait then read populated cache
  await new Promise((r) => setTimeout(r, 35 + Math.floor(Math.random() * 25)));
  const retry = await sendMetaCommand(socket, `mg ${key} v`);
  return retry.status === 'VA' && retry.body ? JSON.parse(retry.body) : loaderFn();
}

module.exports = { sendMetaCommand, getWithStampedeGuard };