Two-tier distributed cache with stampede protection, tag invalidation, and O(1) full invalidation

Cache

Working example → Read-through fetch with dynamic tags/TTL, tag + generation invalidation, and batch warm-up: apps/backend/src/services/caches.ts

Cache() defines a distributed, stampede-protected cache with two tiers: L1 (per-process LRU, ~µs reads) and L2 (the hot backend — Redis in production, in-process in memory mode). L1 stays coherent across nodes through pub/sub invalidation.

The cache is a pure data-plane module: entries live only on the hot backend; the durable plane stores just the cache definition and pause state. It creates no jobs and no runs.

Quick start

caches/products.ts

import { Cache } from "oqronkit";

export const products = Cache<Product>({
  name: "products",
  ttlMs: 60_000, // default entry TTL
  ttlJitter: 0.1, // ±10% randomization — prevents synchronized mass expiry
});

service.ts

// Read-through with stampede protection
const product = await products.getOrFetch("p_123", {
  fetcher: async (key) => db.products.findById(key),
  tags: ["products"],
});

// Direct operations
await products.set("p_123", product, { ttlMs: 30_000, tags: ["products"] });
const hit = await products.get("p_123"); // T | null
await products.delete("p_123");

Lookup flow

getOrFetch(key)
  │
  ├─ 1. L1 (process memory)  → HIT? return
  ├─ 2. L2 (hot backend)     → HIT? populate L1, return
  │
  ├─ 3. Stampede protection
  │    ├─ same-process callers join ONE in-flight fetch (single-flight)
  │    └─ cross-node: distributed lock — one node fetches,
  │       losers poll L2 for the winner's value (lockWaitMs),
  │       then FAIL OPEN (fetch anyway — a cache never deadlocks)
  │
  ├─ 4. fetcher(key, fctx) — optional timeoutMs race
  └─ 5. write L2 + L1 (+ tag log), return value

The stampede guarantee: N concurrent getOrFetch calls for the same key run the fetcher exactly once — within a process via single-flight, across nodes via the distributed lock.

The fetcher context

The fetcher receives a context to set tags and TTL dynamically, based on what it actually fetched:

await products.getOrFetch("p_123", {
  fetcher: async (key, fctx) => {
    const product = await db.products.findById(key);
    fctx.tags([`vendor:${product.vendorId}`]); // dynamic tags
    if (product.volatile) fctx.ttl(5_000); // dynamic TTL
    fctx.log("info", `fetched ${key} from db`);
    return product;
  },
  timeoutMs: 3_000, // reject a hanging fetcher
  forceRefresh: false, // true → skip cache, refetch
  ignoreCacheWrite: false, // true → return fetched value without caching it
});

Invalidation

Three mechanisms, chosen for cross-node correctness:

// 1. Exact key — L1 + L2 + pub/sub broadcast to every node
await products.invalidate("p_123");

// 2. By tags — replays the per-tag key log, exact-invalidating each key
await products.invalidateTags(["vendor:v9"]); // → number of entries removed

// 3. Everything — O(1) generation bump
await products.invalidateAll();

invalidateAll() is O(1) regardless of cache size: each cache carries a generation counter embedded in its storage keys. Bumping it makes every existing entry unreachable instantly (they expire from Redis by TTL). No scans, no key deletion storms.

Tag logs are capped (10k keys per tag) and best-effort under concurrent cross-node writes. For guaranteed removal of a specific entry, use invalidate(key). Prefix invalidation is not yet supported.

Batch operations

const values = await products.getMany(["p_1", "p_2", "p_3"]);
// { p_1: {...}, p_2: {...}, p_3: null }

const res = await products.setMany([
  { key: "p_1", value: a },
  { key: "p_2", value: b, opts: { ttlMs: 10_000 } },
]);
// CacheBatchResult: { ok, total, succeeded: [...], failed: [...], errors: {...} }

await products.deleteMany(["p_1", "p_2"]);

TTL jitter

When thousands of entries are written together (bulk import, deploy warm-up), identical TTLs expire together — and the refill stampedes your database. ttlJitter: 0.1 spreads each entry's effective TTL randomly within ±10%:

Cache<Product>({ name: "products", ttlMs: 300_000, ttlJitter: 0.15 });

Error semantics

Reads never throw. A broken hot backend degrades get to a miss (logged/evented) — your fallback path is the fetcher or the database, never an exception from the cache.
Writes are best-effort. A failed set logs and continues.
Stampede lock failures fail open — the caller fetches directly.
Cached null is indistinguishable from a miss (v1 limitation — avoid caching null; cache a sentinel object instead).

Observability

Events (always emitted): cache:hit, cache:miss, cache:set, cache:delete, cache:invalidate, cache:instance:enabled/disabled.
Stats — this node's counters:

await products.stats();
// { hits, misses, sets, deletes, fetches, fetchErrors, invalidations }

Persisted logs (opt-in) — debug level records every hit/miss/set safely (writes are rate-capped); see Observability & Control:

Cache<Product>({
  name: "products",
  logs: { level: "debug", maxWritesPerSec: 50 },
});

Control & environments

// Cache only active in production
Cache<Product>({ name: "products", environments: ["production"] });

Standard admin surface on the engine: list(), get(name), pauseInstance(name), resumeInstance(name). A paused cache bypasses — reads miss, writes no-op, getOrFetch calls the fetcher directly without caching. Existing entries survive the pause and are visible again on resume.

Configuration reference

Option	Default	Purpose
`name`	—	Unique cache name (storage namespace)
`ttlMs`	`60_000`	Default entry TTL
`ttlJitter`	`0`	0..1 — randomize effective TTL ± factor
`lockWaitMs`	`5_000`	How long stampede losers wait for the winner
`logs`	off	Opt-in persisted logs (`true` or `{ level, maxEntries, ttlMs, maxWritesPerSec }`)
`environments`	all	Environment allow-list
`enabled`	`true`	Definition-level switch
`version`	`0`	Definition version (downgrade protection)

Cache

On this page