Кэширование и инвалидация

import { createHash } from 'node:crypto';
import { createServer } from 'node:http';
import { createProductService } from './products.ts';

const products = await createProductService('http');

const server = createServer(async (req, res) => {
  const match = req.url?.match(/^\/api\/products\/(\d+)$/);
  if (req.method !== 'GET' || !match) {
    res.writeHead(404).end();
    return;
  }
  let value;
  try {
    ({ value } = await products.getProduct(Number(match[1])));
  } catch (err) {
    console.error(err);
    res.writeHead(500, { 'Cache-Control': 'no-store' }).end(); // never cache an error
    return;
  }
  if (!value) {
    res.writeHead(404, { 'Cache-Control': 'public, max-age=30' }).end();
    return;
  }

  const body = JSON.stringify(value);
  const etag = `"${createHash('sha256').update(body).digest('base64url').slice(0, 22)}"`;
  const headers = {
    // browser: 60 s; CDN: 300 s; after that a stale copy may be served for 10 min while revalidating
    'Cache-Control': 'public, max-age=60, s-maxage=300, stale-while-revalidate=600',
    ETag: etag,
  };

  // Conditional request: the client already has this version
  const known = req.headers['if-none-match']?.split(',').map((t) => t.trim()) ?? [];
  if (known.includes(etag)) {
    res.writeHead(304, headers).end();
    return;
  }
  res.writeHead(200, { ...headers, 'Content-Type': 'application/json' }).end(body);
});

server.listen(3080, '127.0.0.1', () => console.log('http://127.0.0.1:3080'));

$ curl -si localhost:3080/api/products/1
HTTP/1.1 200 OK
Cache-Control: public, max-age=60, s-maxage=300, stale-while-revalidate=600
ETag: "QiAF3iaw-m9HY2CB31jgsu"
Content-Type: application/json

{"id":1,"title":"Latte","priceKzt":1800}
$ curl -si localhost:3080/api/products/1 -H 'If-None-Match: "QiAF3iaw-m9HY2CB31jgsu"'
HTTP/1.1 304 Not Modified
Cache-Control: public, max-age=60, s-maxage=300, stale-while-revalidate=600
ETag: "QiAF3iaw-m9HY2CB31jgsu"

import { createCache } from './cache.ts';
import { loadProduct, savePrice, type Product } from './db.ts';
import { connectRedis } from './redis.ts';

const CHANNEL = 'cache:invalidate';
const L1_TTL_MS = 30_000; // safety net: Pub/Sub may lose a message
const L1_MAX = 10_000;
const policy = { freshMs: 60_000, staleMs: 300_000 };

// One instance of the application: L1 in process memory, L2 in shared Redis
export async function createProductService(name: string) {
  const redis = await connectRedis();
  const cache = createCache(redis);
  const local = new Map<string, { value: Product | null; expires: number }>();

  // A subscribed connection cannot run other commands, so it is a separate one
  const sub = redis.duplicate();
  sub.on('error', (err) => console.error(`[${name}] sub error:`, err.message));
  sub.on('ready', () => local.clear()); // after a reconnect, missed messages are gone: drop L1
  await sub.connect();
  await sub.subscribe(CHANNEL, (key) => {
    local.delete(key);
    console.log(`[${name}] invalidated ${key}`);
  });

  // Bumping catalog:version switches every product to new keys at once
  async function keyFor(id: number) {
    const version = (await redis.get('catalog:version')) ?? '0';
    return `catalog:v${version}:product:${id}`;
  }

  async function getProduct(id: number) {
    const key = await keyFor(id);
    const l1 = local.get(key);
    if (l1 && l1.expires > Date.now()) return { key, value: l1.value, from: 'L1' };
    const { value, outcome } = await cache.get(key, () => loadProduct(id), policy);
    if (local.size >= L1_MAX) local.delete(local.keys().next().value!); // evict the oldest
    local.set(key, { value, expires: Date.now() + L1_TTL_MS });
    return { key, value, from: outcome === 'miss' ? 'db' : `L2 (${outcome})` };
  }

  async function changePrice(id: number, priceKzt: number) {
    await savePrice(id, priceKzt); // 1. the source of truth first
    const key = await keyFor(id);
    await redis.del(key); // 2. the shared cache
    await redis.publish(CHANNEL, key); // 3. local caches of every instance
  }

  async function close() {
    await sub.close();
    await redis.close();
  }

  return { getProduct, changePrice, redis, close };
}

import { setTimeout as sleep } from 'node:timers/promises';
import { createProductService } from './products.ts';

// Two application instances; in production they are separate processes
const a = await createProductService('A');
const b = await createProductService('B');

async function show(name: string, svc: typeof a, id: number) {
  const { key, value, from } = await svc.getProduct(id);
  console.log(`${name} get ${key}: ${value?.priceKzt} KZT from ${from}`);
}

await show('A', a, 1);
await show('B', b, 1);
await show('A', a, 1);
await show('B', b, 1);

console.log('--- A changes the price of product 1');
await a.changePrice(1, 2000);
await sleep(50); // let the Pub/Sub message arrive
await show('B', b, 1);
await show('A', a, 1);

console.log('--- bulk price update: bump catalog:version');
await show('A', a, 2);
console.log('catalog:version =', await a.redis.incr('catalog:version'));
await show('A', a, 2);
await show('B', b, 1);

await a.close();
await b.close();

$ node src/demo-invalidation.ts
A get catalog:v0:product:1: 1800 KZT from db
B get catalog:v0:product:1: 1800 KZT from L2 (hit)
A get catalog:v0:product:1: 1800 KZT from L1
B get catalog:v0:product:1: 1800 KZT from L1
--- A changes the price of product 1
[B] invalidated catalog:v0:product:1
[A] invalidated catalog:v0:product:1
B get catalog:v0:product:1: 2000 KZT from db
A get catalog:v0:product:1: 2000 KZT from L2 (hit)
--- bulk price update: bump catalog:version
A get catalog:v0:product:2: 2400 KZT from db
catalog:version = 1
A get catalog:v1:product:2: 2400 KZT from db
B get catalog:v1:product:1: 2000 KZT from db

docker run -d --rm --name kb-redis-cache -p 127.0.0.1:56390:6379 redis:8.10-alpine

mkdir cache-demo && cd cache-demo
npm init -y && npm pkg set type=module
npm install redis@6.3.0
npm install -D typescript@7.0.2 @types/node@24

{
  "compilerOptions": {
    "target": "es2024",
    "module": "nodenext",
    "strict": true,
    "noEmit": true,
    "allowImportingTsExtensions": true,
    "erasableSyntaxOnly": true,
    "verbatimModuleSyntax": true,
    "skipLibCheck": true
  },
  "include": ["src"]
}

import { createClient } from 'redis';

export async function connectRedis() {
  const client = createClient({ url: process.env.REDIS_URL ?? 'redis://localhost:56390' });
  client.on('error', (err) => console.error('redis error:', err.message));
  await client.connect();
  return client;
}

export type Redis = Awaited<ReturnType<typeof connectRedis>>;

// Stand-in for a slow database: in a real project this is SQL or an external API
export type Product = { id: number; title: string; priceKzt: number };

const products = new Map<number, Product>([
  [1, { id: 1, title: 'Latte', priceKzt: 1800 }],
  [2, { id: 2, title: 'Cheesecake', priceKzt: 2400 }],
]);
export const db = { calls: 0 };
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

export async function loadProduct(id: number): Promise<Product | null> {
  db.calls++;
  await sleep(200); // a heavy query
  const product = products.get(id);
  return product ? structuredClone(product) : null;
}

export async function savePrice(id: number, priceKzt: number): Promise<void> {
  await sleep(10);
  const product = products.get(id);
  if (!product) throw new Error(`product ${id} not found`);
  product.priceKzt = priceKzt;
}

import { randomUUID } from 'node:crypto';
import { setTimeout as sleep } from 'node:timers/promises';
import type { Redis } from './redis.ts';

export type Outcome = 'hit' | 'stale' | 'miss' | 'waited' | 'fallback';
type Entry<T> = { value: T; freshUntil: number }; // freshness lives inside the value
type Policy = { freshMs: number; staleMs: number };

const LOCK_MS = 5000; // longer than the slowest load, otherwise two loads may overlap
const POLL_MS = 25;
const WAIT_MS = 2000;

export function createCache(redis: Redis) {
  async function write<T>(key: string, value: T, { freshMs, staleMs }: Policy) {
    const entry: Entry<T> = { value, freshUntil: Date.now() + freshMs };
    // the key lives longer than it is fresh: the stale tail is served while rebuilding
    await redis.set(key, JSON.stringify(entry), {
      expiration: { type: 'PX', value: freshMs + staleMs },
    });
  }

  // Only the request that takes the lock goes to the source
  async function rebuild<T>(key: string, load: () => Promise<T>, policy: Policy) {
    const lockKey = `lock:${key}`;
    const token = randomUUID();
    const locked = await redis.set(lockKey, token, {
      condition: 'NX',
      expiration: { type: 'PX', value: LOCK_MS },
    });
    if (locked !== 'OK') return null; // somebody else is already rebuilding
    try {
      const value = await load();
      await write(key, value, policy);
      return { value };
    } finally {
      // Redis 8.4+: delete the lock only if it is still ours
      await redis.delEx(lockKey, { condition: 'IFEQ', matchValue: token });
    }
  }

  async function get<T>(key: string, load: () => Promise<T>, policy: Policy): Promise<{ value: T; outcome: Outcome }> {
    const raw = await redis.get(key);
    if (raw !== null) {
      const entry = JSON.parse(raw) as Entry<T>;
      if (Date.now() < entry.freshUntil) return { value: entry.value, outcome: 'hit' };
      // stale-while-revalidate: answer with the old value, rebuild in the background
      rebuild(key, load, policy).catch((err) => console.warn('background rebuild failed:', err.message));
      return { value: entry.value, outcome: 'stale' };
    }

    const built = await rebuild(key, load, policy);
    if (built) return { value: built.value, outcome: 'miss' };

    // Someone holds the lock: wait for their result instead of hitting the database
    for (let waited = 0; waited < WAIT_MS; waited += POLL_MS) {
      await sleep(POLL_MS);
      const ready = await redis.get(key);
      if (ready !== null) return { value: (JSON.parse(ready) as Entry<T>).value, outcome: 'waited' };
    }
    return { value: await load(), outcome: 'fallback' }; // the rebuild is too slow: go to the source
  }

  return { get };
}

// For comparison: plain cache-aside without any protection
export async function naiveGet<T>(redis: Redis, key: string, load: () => Promise<T>, ttlMs: number) {
  const raw = await redis.get(key);
  if (raw !== null) return JSON.parse(raw) as T;
  const value = await load();
  await redis.set(key, JSON.stringify(value), { expiration: { type: 'PX', value: ttlMs } });
  return value;
}

import { setTimeout as sleep } from 'node:timers/promises';
import { createCache, naiveGet, type Outcome } from './cache.ts';
import { db, loadProduct } from './db.ts';
import { connectRedis } from './redis.ts';

const redis = await connectRedis();
const cache = createCache(redis);
const policy = { freshMs: 1000, staleMs: 60_000 };
const N = 100; // simultaneous requests for the same popular key

async function burst(label: string, request: () => Promise<Outcome | void>) {
  const before = db.calls;
  const t0 = performance.now();
  const outcomes = await Promise.all(Array.from({ length: N }, request));
  const ms = Math.round(performance.now() - t0);
  const counts: Record<string, number> = {};
  for (const o of outcomes) if (o) counts[o] = (counts[o] ?? 0) + 1;
  console.log(`${label}: db +${db.calls - before} (total ${db.calls}), ${ms} ms`, Object.keys(counts).length ? counts : '');
}

await redis.del(['naive:product:1', 'product:1']);

await burst('naive, cold key   ', async () => {
  await naiveGet(redis, 'naive:product:1', () => loadProduct(1), 60_000);
});

await burst('locked, cold key  ', async () => (await cache.get('product:1', () => loadProduct(1), policy)).outcome);
await burst('locked, fresh key ', async () => (await cache.get('product:1', () => loadProduct(1), policy)).outcome);

await sleep(1100); // the entry is now stale but still in Redis
await burst('locked, stale key ', async () => (await cache.get('product:1', () => loadProduct(1), policy)).outcome);
await sleep(300); // the background rebuild has finished
await burst('after rebuild     ', async () => (await cache.get('product:1', () => loadProduct(1), policy)).outcome);

await redis.close();

$ node src/demo-stampede.ts
naive, cold key   : db +100 (total 100), 212 ms
locked, cold key  : db +1 (total 101), 210 ms { miss: 1, waited: 99 }
locked, fresh key : db +0 (total 101), 1 ms { hit: 100 }
locked, stale key : db +0 (total 101), 7 ms { stale: 100 }
after rebuild     : db +0 (total 102), 2 ms { hit: 100 }
$ docker exec kb-redis-cache redis-cli INFO stats | grep keyspace_
keyspace_hits:401
keyspace_misses:695