diff --git a/lib/shared-store/__tests__/contract.ts b/lib/shared-store/__tests__/contract.ts new file mode 100644 index 000000000..055b685e1 --- /dev/null +++ b/lib/shared-store/__tests__/contract.ts @@ -0,0 +1,353 @@ +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterAll, beforeEach, describe, expect, it } from 'vitest'; +import type { CollectionOptions, LogOptions, SharedStore } from '../types'; + +/** + * One contract for both stores, so local mode and the shared store can't + * drift apart (#1066). `makeStores(n)` returns n handles on one store: for + * the shared store n separate connections, standing in for n replicas. + */ +export interface ContractTarget { + name: string; + makeStores: (n: number) => SharedStore[]; + /** Called after each test file; drop the keys the run used. */ + cleanup?: () => Promise; +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +export function runStoreContract(target: ContractTarget): void { + describe(`shared store contract: ${target.name}`, () => { + let dir: string; + let run = 0; + const roots: string[] = []; + + beforeEach(() => { + dir = mkdtempSync(path.join(tmpdir(), 'bulwark-store-')); + roots.push(dir); + run++; + }); + + afterAll(async () => { + for (const root of roots) rmSync(root, { recursive: true, force: true }); + await target.cleanup?.(); + }); + + const coll = (name: string, extra: Partial = {}): CollectionOptions => ({ + local: { dir: () => path.join(dir, name) }, + ...extra, + }); + const key = (k: string) => `t${run}:${k}`; + const open = async (n: number) => { + const stores = target.makeStores(n); + await Promise.all(stores.map((store) => store.ready())); + return stores; + }; + + describe('kv', () => { + it('sets, gets and deletes', async () => { + const [s] = await open(1); + expect(await s.kv.get(key('a'))).toBeNull(); + expect(await s.kv.set(key('a'), 'one')).toBe(true); + expect(await s.kv.get(key('a'))).toBe('one'); + await s.kv.del(key('a')); + expect(await s.kv.get(key('a'))).toBeNull(); + }); + + it('expires values after their ttl', async () => { + const [s] = await open(1); + await s.kv.set(key('t'), 'x', { ttlMs: 150 }); + expect(await s.kv.get(key('t'))).toBe('x'); + await sleep(400); + expect(await s.kv.get(key('t'))).toBeNull(); + }); + + it('sets ifAbsent only once, across replicas', async () => { + const stores = await open(3); + const results = await Promise.all( + Array.from({ length: 12 }, (_, i) => stores[i % 3].kv.set(key('once'), `v${i}`, { ifAbsent: true })), + ); + expect(results.filter(Boolean)).toHaveLength(1); + const winner = results.indexOf(true); + expect(await stores[1].kv.get(key('once'))).toBe(`v${winner}`); + }); + + it('hands a taken value to exactly one caller', async () => { + const stores = await open(3); + await stores[0].kv.set(key('code'), 'grant'); + const taken = await Promise.all(Array.from({ length: 12 }, (_, i) => stores[i % 3].kv.take(key('code')))); + expect(taken.filter((v) => v === 'grant')).toHaveLength(1); + expect(await stores[2].kv.get(key('code'))).toBeNull(); + }); + + it('updates without losing concurrent changes', async () => { + const stores = await open(3); + await Promise.all(Array.from({ length: 30 }, (_, i) => + stores[i % 3].kv.update(key('n'), (cur) => String(Number(cur ?? '0') + 1)), + )); + expect(await stores[0].kv.get(key('n'))).toBe('30'); + }); + + it('update can delete or leave a value', async () => { + const [s] = await open(1); + await s.kv.set(key('u'), 'a'); + expect(await s.kv.update(key('u'), () => undefined)).toBeUndefined(); + expect(await s.kv.get(key('u'))).toBe('a'); + expect(await s.kv.update(key('u'), () => null)).toBeNull(); + expect(await s.kv.get(key('u'))).toBeNull(); + }); + + it('refuses keys with control characters', async () => { + const [s] = await open(1); + await expect(s.kv.get('bad\nkey')).rejects.toThrow(/Invalid store key/); + }); + }); + + describe('counters', () => { + it('reserves up to the max and refuses without counting', async () => { + const [s] = await open(1); + const limit = [{ key: key('ip'), max: 3, windowMs: 60_000 }]; + for (let i = 1; i <= 3; i++) expect((await s.counters.reserve(limit)).ok).toBe(true); + const refused = await s.counters.reserve(limit); + expect(refused.ok).toBe(false); + expect(refused.retryAfterMs).toBeGreaterThan(55_000); + expect(refused.counts).toEqual([3]); + expect((await s.counters.get(key('ip')))?.count).toBe(3); + }); + + it('checks every key before counting any (a full global ceiling stops a fresh client)', async () => { + const [s] = await open(1); + const global = { key: key('all'), max: 2, windowMs: 60_000 }; + expect((await s.counters.reserve([{ key: key('c1'), max: 5, windowMs: 60_000 }, global])).ok).toBe(true); + expect((await s.counters.reserve([{ key: key('c2'), max: 5, windowMs: 60_000 }, global])).ok).toBe(true); + const third = await s.counters.reserve([{ key: key('c3'), max: 5, windowMs: 60_000 }, global]); + expect(third.ok).toBe(false); + expect(await s.counters.get(key('c3'))).toBeNull(); + // A client over its own limit doesn't use up the ceiling either. + const own = { key: key('own'), max: 1, windowMs: 60_000 }; + const wide = { key: key('wide'), max: 10, windowMs: 60_000 }; + await s.counters.reserve([own, wide]); + for (let i = 0; i < 5; i++) await s.counters.reserve([own, wide]); + expect((await s.counters.get(key('wide')))?.count).toBe(1); + }); + + it('counts together across replicas', async () => { + const stores = await open(3); + const limit = [{ key: key('shared'), max: 5, windowMs: 60_000 }]; + const results = await Promise.all(Array.from({ length: 12 }, (_, i) => stores[i % 3].counters.reserve(limit))); + expect(results.filter((r) => r.ok)).toHaveLength(5); + }); + + it('starts a new window after windowMs, by the caller\'s clock', async () => { + const [s] = await open(1); + const t0 = Date.now(); + const limit = [{ key: key('w'), max: 1, windowMs: 1000 }]; + expect((await s.counters.reserve(limit, t0)).ok).toBe(true); + expect((await s.counters.reserve(limit, t0 + 999)).ok).toBe(false); + const next = await s.counters.reserve(limit, t0 + 1000); + expect(next.ok).toBe(true); + expect(next.resetAts[0]).toBe(t0 + 2000); + }); + + it('refunds only the window it reserved in', async () => { + const [s] = await open(1); + const t0 = Date.now(); + const limit = [{ key: key('r'), max: 2, windowMs: 1000 }]; + const first = await s.counters.reserve(limit, t0); + await s.counters.reserve(limit, t0 + 10); + await s.counters.refund([{ key: key('r'), resetAt: first.resetAts[0] }], t0 + 20); + expect((await s.counters.get(key('r'), t0 + 30))?.count).toBe(1); + // A new window: the old reservation's refund doesn't touch it. + await s.counters.reserve(limit, t0 + 1500); + await s.counters.refund([{ key: key('r'), resetAt: first.resetAts[0] }], t0 + 1510); + expect((await s.counters.get(key('r'), t0 + 1520))?.count).toBe(1); + // Never below zero. + await s.counters.refund([{ key: key('r') }, { key: key('r') }, { key: key('r') }], t0 + 1530); + expect((await s.counters.get(key('r'), t0 + 1540))?.count).toBe(0); + }); + + it('hits, gets and resets', async () => { + const [s] = await open(1); + const t0 = Date.now(); + expect(await s.counters.hit(key('h'), 5000, t0)).toEqual({ count: 1, resetAt: t0 + 5000 }); + expect((await s.counters.hit(key('h'), 5000, t0 + 1)).count).toBe(2); + await s.counters.reset(key('h')); + expect(await s.counters.get(key('h'), t0 + 2)).toBeNull(); + }); + }); + + describe('documents', () => { + it('round-trips bytes and changes the version on every write', async () => { + const [s] = await open(1); + const docs = s.collection('things', coll('things')); + const bytes = Buffer.from([0, 1, 2, 255, 10, 13]); + const put = await docs.put('a.bin', bytes); + expect(put.ok).toBe(true); + const got = await docs.get('a.bin'); + expect(got?.data.equals(bytes)).toBe(true); + expect(got?.version).toBe(put.ok && put.version); + const again = await docs.put('a.bin', 'text'); + expect(again.ok && again.version).not.toBe(got?.version); + expect((await docs.get('a.bin'))?.data.toString()).toBe('text'); + }); + + it('creates only once with ifVersion: null, across replicas', async () => { + const stores = await open(3); + const results = await Promise.all(Array.from({ length: 9 }, (_, i) => + stores[i % 3].collection('once', coll('once')).put('x', `v${i}`, { ifVersion: null }), + )); + expect(results.filter((r) => r.ok)).toHaveLength(1); + }); + + it('refuses a stale version and returns the current document', async () => { + const [s] = await open(1); + const docs = s.collection('cas', coll('cas')); + const first = await docs.put('p', 'one'); + await docs.put('p', 'two'); + const stale = await docs.put('p', 'three', { ifVersion: first.ok ? first.version : '' }); + expect(stale.ok).toBe(false); + expect(!stale.ok && stale.current?.data.toString()).toBe('two'); + expect((await docs.get('p'))?.data.toString()).toBe('two'); + }); + + it('update loses no concurrent change across replicas', async () => { + const stores = await open(3); + await Promise.all(Array.from({ length: 24 }, (_, i) => + stores[i % 3].collection('upd', coll('upd')).update('list.json', (cur) => { + const list = cur ? (JSON.parse(cur.toString()) as number[]) : []; + return JSON.stringify([...list, i]); + }, { maxRetries: 100 }), + )); + const final = JSON.parse((await stores[0].collection('upd', coll('upd')).get('list.json'))!.data.toString()) as number[]; + expect(final.sort((a, b) => a - b)).toEqual(Array.from({ length: 24 }, (_, i) => i)); + }); + + it('update leaves the document alone on undefined and deletes on null', async () => { + const [s] = await open(1); + const docs = s.collection('upd2', coll('upd2')); + await docs.put('d', 'keep'); + const left = await docs.update('d', () => undefined); + expect(left.data).toBeUndefined(); + expect((await docs.get('d'))?.data.toString()).toBe('keep'); + await docs.update('d', () => null); + expect(await docs.get('d')).toBeNull(); + // Deleting what isn't there is fine. + expect((await docs.update('d', () => null)).data).toBeNull(); + }); + + it('deletes, conditionally too', async () => { + const [s] = await open(1); + const docs = s.collection('del', coll('del')); + const put = await docs.put('d', 'x'); + expect(await docs.delete('d', { ifVersion: 'nope' })).toBe(false); + expect(await docs.delete('d', { ifVersion: put.ok ? put.version : '' })).toBe(true); + expect(await docs.get('d')).toBeNull(); + expect(await docs.delete('d')).toBe(true); + }); + + it('lists ids by prefix, nested ones included', async () => { + const [s] = await open(1); + const docs = s.collection('tree', coll('tree')); + await docs.put('history/a/1.0.0/bundle.js', 'a1'); + await docs.put('history/a/1.1.0/bundle.js', 'a2'); + await docs.put('history/b/1.0.0/bundle.js', 'b1'); + await docs.put('registry.json', '{}'); + expect(await docs.list('history/a/')).toEqual(['history/a/1.0.0/bundle.js', 'history/a/1.1.0/bundle.js']); + expect(await docs.list()).toHaveLength(4); + await docs.delete('history/a/1.0.0/bundle.js'); + expect(await docs.list('history/a/')).toEqual(['history/a/1.1.0/bundle.js']); + }); + + it('refuses ids that would leave the collection', async () => { + const [s] = await open(1); + const docs = s.collection('safe', coll('safe')); + for (const bad of ['../x', 'a/../../x', '/abs', '', 'a//b', '.', 'a/./b', 'sp ace']) { + await expect(docs.put(bad, 'x')).rejects.toThrow(); + } + }); + + it('expires documents of a ttl collection', async () => { + const [s] = await open(1); + const docs = s.collection('ttl', coll('ttl', { ttlMs: 300 })); + expect((await docs.put('m', '', { ifVersion: null })).ok).toBe(true); + expect((await docs.put('m', '', { ifVersion: null })).ok).toBe(false); + await sleep(1200); + expect(await docs.get('m')).toBeNull(); + expect(await docs.list()).toEqual([]); + expect((await docs.put('m', '', { ifVersion: null })).ok).toBe(true); + }); + }); + + describe('log', () => { + const logOptions = (): LogOptions => ({ + local: { path: () => path.join(dir, 'audit.log'), maxBytes: 10 * 1024 * 1024, rotations: 3 }, + maxEntries: 10_000, + }); + + it('keeps every line from concurrent writers, untorn and in each writer\'s order', async () => { + const stores = await open(3); + const logs = stores.map((s) => s.log(`audit-${run}`, logOptions())); + await Promise.all(logs.map((log, r) => Promise.all( + Array.from({ length: 50 }, (_, i) => log.append(JSON.stringify({ r, i, pad: 'x'.repeat(200) }))), + ))); + const lines = await logs[0].readAll(); + expect(lines).toHaveLength(150); + const parsed = lines.map((l) => JSON.parse(l) as { r: number; i: number }); + for (let r = 0; r < 3; r++) { + expect(parsed.filter((p) => p.r === r).map((p) => p.i)).toEqual(Array.from({ length: 50 }, (_, i) => i)); + } + expect(await logs[1].length()).toBe(150); + }); + + it('never writes a line break into a line', async () => { + const [s] = await open(1); + const log = s.log(`audit-${run}`, logOptions()); + await log.append('one\ntwo'); + expect(await log.readAll()).toEqual(['one two']); + }); + }); + + describe('leases', () => { + it('has one holder at a time and hands over after expiry', async () => { + const [a, b] = await open(2); + const name = `lease-${run}`; + expect(await a.leases.acquire(name, 'A', 300)).toBe(true); + expect(await b.leases.acquire(name, 'B', 300)).toBe(false); + expect(await a.leases.acquire(name, 'A', 300)).toBe(true); // renew + expect(await b.leases.holder(name)).toBe('A'); + await sleep(500); + expect(await b.leases.acquire(name, 'B', 300)).toBe(true); + await a.leases.release(name, 'A'); // not A's any more + expect(await a.leases.holder(name)).toBe('B'); + await b.leases.release(name, 'B'); + expect(await a.leases.holder(name)).toBeNull(); + }); + }); + + describe('epochs and replicas', () => { + it('bumps epochs that every replica reads', async () => { + const [a, b] = await open(2); + const area = `area-${run}`; + const before = (await b.epochs.all())[area] ?? 0; + expect(await a.epochs.bump(area)).toBe(before + 1); + expect((await b.epochs.all())[area]).toBe(before + 1); + }); + + it('lists replicas that beat recently', async () => { + const [a, b] = await open(2); + await a.replicas.beat({ id: `r${run}-a`, version: '1.0.0', startedAt: 1 }); + await b.replicas.beat({ id: `r${run}-b`, version: '1.0.0', startedAt: 2 }); + const ids = (await a.replicas.recent(60_000)).map((r) => r.id); + expect(ids).toEqual(expect.arrayContaining([`r${run}-a`, `r${run}-b`])); + }); + + it('answers a ping', async () => { + const [s] = await open(1); + await expect(s.ping()).resolves.toBeUndefined(); + expect(s.status().connected || s.mode === 'redis').toBe(true); + }); + }); + }); +} diff --git a/lib/shared-store/__tests__/local-store.test.ts b/lib/shared-store/__tests__/local-store.test.ts new file mode 100644 index 000000000..c92f0a8f7 --- /dev/null +++ b/lib/shared-store/__tests__/local-store.test.ts @@ -0,0 +1,12 @@ +import { createLocalStore, resetLocalStoreForTests } from '../local'; +import { runStoreContract } from './contract'; + +runStoreContract({ + name: 'local (memory + files)', + makeStores: (n) => { + // One process: every handle is the same store, as every module copy is. + const store = createLocalStore(); + return Array.from({ length: n }, () => store); + }, + cleanup: async () => resetLocalStoreForTests(), +}); diff --git a/lib/shared-store/__tests__/redis-store.test.ts b/lib/shared-store/__tests__/redis-store.test.ts new file mode 100644 index 000000000..48418352b --- /dev/null +++ b/lib/shared-store/__tests__/redis-store.test.ts @@ -0,0 +1,42 @@ +import { randomBytes } from 'node:crypto'; +import Redis from 'ioredis'; +import { describe, it } from 'vitest'; +import { createRedisStore } from '../redis'; +import type { SharedStore } from '../types'; +import { runStoreContract } from './contract'; + +/** + * The contract against a real Redis-protocol server. Runs only when + * IT_SHARED_STORE_URL is set (e.g. a Valkey on the test VM through an SSH + * tunnel), so the unit run in CI stays offline. Keys get a random prefix and + * are removed afterwards. + */ +const url = process.env.IT_SHARED_STORE_URL; + +if (!url) { + describe('shared store contract: redis', () => { + it.skip('needs IT_SHARED_STORE_URL', () => {}); + }); +} else { + const prefix = `bulwark-test:${randomBytes(4).toString('hex')}:`; + const opened: SharedStore[] = []; + runStoreContract({ + name: `redis (${prefix})`, + makeStores: (n) => { + const stores = Array.from({ length: n }, () => createRedisStore({ url, keyPrefix: prefix })); + opened.push(...stores); + return stores; + }, + cleanup: async () => { + await Promise.all(opened.map((s) => s.close())); + const admin = new Redis(url); + let cursor = '0'; + do { + const [next, keys] = await admin.scan(cursor, 'MATCH', `${prefix}*`, 'COUNT', 500); + cursor = next; + if (keys.length) await admin.del(...keys); + } while (cursor !== '0'); + admin.disconnect(); + }, + }); +} diff --git a/lib/shared-store/index.ts b/lib/shared-store/index.ts new file mode 100644 index 000000000..a32423336 --- /dev/null +++ b/lib/shared-store/index.ts @@ -0,0 +1,83 @@ +import { hostname } from 'node:os'; +import { readFileEnv } from '@/lib/read-file-env'; +import { createLocalStore } from './local'; +import { createRedisStore, randomHolderSuffix } from './redis'; +import type { SharedStore } from './types'; + +export * from './types'; + +/** + * The one store of this process. On globalThis, because Next evaluates this + * module once per bundle entry (proxy, instrumentation, each route bundle) + * and they must all share one connection and one set of local maps. + */ +interface StoreGlobals { + store?: SharedStore; + replicaId?: string; + boot?: BootState; +} + +/** + * starting: connecting to the store and loading. ready: serving. + * store-lost: the store was emptied under a running deployment (see + * changes.ts); the replica keeps its last loaded state and isn't ready. + * blocked: this replica's local data conflicts with an already initialised + * store (see boot.ts); it won't serve until an admin resolves it. + */ +export type BootState = 'starting' | 'ready' | 'store-lost' | 'blocked'; + +const KEY = Symbol.for('bulwark.sharedStore'); + +function globals(): StoreGlobals { + const g = globalThis as typeof globalThis & { [KEY]?: StoreGlobals }; + return (g[KEY] ??= {}); +} + +/** The shared store URL from REDIS_URL or REDIS_URL_FILE, or null for local mode. */ +export function sharedStoreUrl(): string | null { + return process.env.REDIS_URL || readFileEnv(process.env.REDIS_URL_FILE) || null; +} + +export function getStore(): SharedStore { + const g = globals(); + if (!g.store) { + const url = sharedStoreUrl(); + g.store = url + ? createRedisStore({ url, keyPrefix: process.env.REDIS_KEY_PREFIX || 'bulwark:' }) + : createLocalStore(); + } + return g.store; +} + +/** True when this process runs with a shared store (several replicas possible). */ +export function isSharedMode(): boolean { + return getStore().mode !== 'local'; +} + +/** Stable for the life of the process: host name plus a random suffix. */ +export function replicaId(): string { + const g = globals(); + g.replicaId ??= `${hostname()}-${randomHolderSuffix()}`; + return g.replicaId; +} + +/** + * How far the boot chain got. Local mode starts out ready, as it always + * has; shared mode is "starting" until instrumentation finished connecting, + * importing and loading (see instrumentation.node.ts). + */ +export function bootState(): BootState { + const g = globals(); + return g.boot ?? (isSharedMode() ? 'starting' : 'ready'); +} + +export function setBootState(state: BootState): void { + globals().boot = state; +} + +/** Replace the store (tests). Pass null to go back to the one from the environment. */ +export function setStoreForTests(store: SharedStore | null): void { + const g = globals(); + g.store = store ?? undefined; + g.boot = undefined; +} diff --git a/lib/shared-store/keys.ts b/lib/shared-store/keys.ts new file mode 100644 index 000000000..fff65a176 --- /dev/null +++ b/lib/shared-store/keys.ts @@ -0,0 +1,36 @@ +import { createHash } from 'node:crypto'; +import { InvalidStoreKeyError } from './types'; + +// One path segment: letters, digits and . _ @ + ~ -, but never "." or "..". +const SEGMENT = /^(?!\.{1,2}$)[A-Za-z0-9._@+~-]{1,200}$/; +const NAME = /^[a-z0-9][a-z0-9-]{0,63}$/; + +/** A document id: one or more segments joined by "/". Throws on anything else. */ +export function assertDocId(id: string): void { + if (typeof id !== 'string' || id.length === 0 || id.length > 1024) throw new InvalidStoreKeyError(String(id)); + for (const segment of id.split('/')) { + if (!SEGMENT.test(segment)) throw new InvalidStoreKeyError(id); + } +} + +/** A collection, log or lease name. */ +export function assertName(name: string): void { + if (!NAME.test(name)) throw new InvalidStoreKeyError(name); +} + +/** A kv or counter key: anything printable without control characters, at most 512 chars. */ +export function assertKey(key: string): void { + // eslint-disable-next-line no-control-regex + if (typeof key !== 'string' || key.length === 0 || key.length > 512 || /[\u0000-\u001f\u007f]/.test(key)) { + throw new InvalidStoreKeyError(String(key)); + } +} + +/** The version of a document: changes whenever its bytes do. */ +export function versionOf(data: Buffer): string { + return createHash('sha256').update(data).digest('base64url').slice(0, 22); +} + +export function toBuffer(data: Buffer | string): Buffer { + return typeof data === 'string' ? Buffer.from(data, 'utf8') : data; +} diff --git a/lib/shared-store/local.ts b/lib/shared-store/local.ts new file mode 100644 index 000000000..781174a6e --- /dev/null +++ b/lib/shared-store/local.ts @@ -0,0 +1,483 @@ +import { randomBytes } from 'node:crypto'; +import { existsSync } from 'node:fs'; +import { appendFile, mkdir, readFile, readdir, rename, stat, unlink, writeFile } from 'node:fs/promises'; +import path from 'node:path'; +import { assertDocId, assertKey, assertName, toBuffer, versionOf } from './keys'; +import type { + AppendLog, + CollectionOptions, + CounterLimit, + CounterStore, + DocCollection, + EpochStore, + KvSetOptions, + KvStore, + LeaseStore, + LogOptions, + PutOptions, + PutResult, + ReplicaInfo, + ReplicaRegistry, + ReserveResult, + SharedStore, + StoredDoc, +} from './types'; + +/** + * The default store: one process, today's behaviour. Short-lived values and + * counters live in memory, documents in the files their collection names, + * written through a temp file and a rename. Everything that must be shared + * by the module copies Next makes of this file (proxy, instrumentation, + * routes) sits on globalThis. + */ + +interface LocalState { + kv: Map; + kvSets: number; + counters: Map; + queues: Map>; + pruned: Map; + leases: Map; + epochs: Map; + replicas: Map; +} + +const STATE_KEY = Symbol.for('bulwark.sharedStore.local'); +const MAX_COUNTERS = 50_000; + +function state(): LocalState { + const g = globalThis as typeof globalThis & { [STATE_KEY]?: LocalState }; + if (!g[STATE_KEY]) { + g[STATE_KEY] = { + kv: new Map(), + kvSets: 0, + counters: new Map(), + queues: new Map(), + pruned: new Map(), + leases: new Map(), + epochs: new Map(), + replicas: new Map(), + }; + } + return g[STATE_KEY]; +} + +/** Run tasks for one key one after another, across every module copy in this process. */ +export function serializeLocal(key: string, task: () => Promise): Promise { + const queues = state().queues; + const run = (queues.get(key) ?? Promise.resolve()).then(task, task); + const tail = run.catch(() => {}); + queues.set(key, tail); + void tail.then(() => { + if (queues.get(key) === tail) queues.delete(key); + }); + return run; +} + +function isMissing(error: unknown): boolean { + return (error as NodeJS.ErrnoException)?.code === 'ENOENT'; +} + +// ─── kv ──────────────────────────────────────────────────────────────────── + +function kvLive(key: string, now = Date.now()): { value: string; expiresAt: number } | null { + const entry = state().kv.get(key); + if (!entry) return null; + if (entry.expiresAt <= now) { + state().kv.delete(key); + return null; + } + return entry; +} + +function kvSweep(): void { + const s = state(); + if (++s.kvSets % 1000 !== 0) return; + const now = Date.now(); + for (const [key, entry] of s.kv) if (entry.expiresAt <= now) s.kv.delete(key); +} + +function kvWrite(key: string, value: string, ttlMs: number | undefined): void { + state().kv.set(key, { value, expiresAt: ttlMs && ttlMs > 0 ? Date.now() + ttlMs : Infinity }); + kvSweep(); +} + +const kv: KvStore = { + async get(key) { + assertKey(key); + return kvLive(key)?.value ?? null; + }, + async set(key, value, options: KvSetOptions = {}) { + assertKey(key); + if (options.ifAbsent && kvLive(key)) return false; + kvWrite(key, value, options.ttlMs); + return true; + }, + async del(key) { + assertKey(key); + state().kv.delete(key); + }, + async take(key) { + assertKey(key); + const entry = kvLive(key); + if (!entry) return null; + state().kv.delete(key); + return entry.value; + }, + async update(key, fn, options = {}) { + assertKey(key); + // One thread, and fn is synchronous: nothing can interleave here. + const entry = kvLive(key); + const next = fn(entry?.value ?? null); + if (next === undefined) return undefined; + if (next === null) state().kv.delete(key); + else kvWrite(key, next, options.ttlMs ?? (entry && entry.expiresAt !== Infinity ? entry.expiresAt - Date.now() : undefined)); + return next; + }, +}; + +// ─── counters ────────────────────────────────────────────────────────────── + +function counterLive(key: string, now: number): { count: number; resetAt: number } | null { + const entry = state().counters.get(key); + if (!entry || entry.resetAt <= now) return null; + return entry; +} + +function counterPrune(now: number): void { + const counters = state().counters; + if (counters.size < MAX_COUNTERS) return; + for (const [key, entry] of counters) if (entry.resetAt <= now) counters.delete(key); + // Still full of live windows: drop the oldest. + while (counters.size >= MAX_COUNTERS) { + const oldest = counters.keys().next().value; + if (oldest === undefined) break; + counters.delete(oldest); + } +} + +const counters: CounterStore = { + async reserve(limits: CounterLimit[], now = Date.now()): Promise { + const current = limits.map((limit) => { + assertKey(limit.key); + return counterLive(limit.key, now); + }); + let retryAfterMs = 0; + let ok = true; + limits.forEach((limit, i) => { + const entry = current[i]; + if (entry && entry.count >= limit.max) { + ok = false; + retryAfterMs = Math.max(retryAfterMs, entry.resetAt - now); + } + }); + if (!ok) return { ok, retryAfterMs, counts: current.map((e) => e?.count ?? 0), resetAts: current.map((e) => e?.resetAt ?? 0) }; + counterPrune(now); + const after = limits.map((limit, i) => { + const entry = current[i] ?? { count: 0, resetAt: now + limit.windowMs }; + entry.count++; + state().counters.set(limit.key, entry); + return entry; + }); + return { ok, retryAfterMs: 0, counts: after.map((e) => e.count), resetAts: after.map((e) => e.resetAt) }; + }, + async refund(entries, now = Date.now()) { + for (const { key, resetAt } of entries) { + assertKey(key); + const entry = counterLive(key, now); + if (!entry) continue; + if (resetAt !== undefined && entry.resetAt !== resetAt) continue; + entry.count = Math.max(0, entry.count - 1); + } + }, + async hit(key, windowMs, now = Date.now()) { + assertKey(key); + counterPrune(now); + const entry = counterLive(key, now) ?? { count: 0, resetAt: now + windowMs }; + entry.count++; + state().counters.set(key, entry); + return { count: entry.count, resetAt: entry.resetAt }; + }, + async get(key, now = Date.now()) { + assertKey(key); + const entry = counterLive(key, now); + return entry ? { count: entry.count, resetAt: entry.resetAt } : null; + }, + async reset(key) { + assertKey(key); + state().counters.delete(key); + }, +}; + +// ─── documents ───────────────────────────────────────────────────────────── + +/** Files of a collection on disk, as today. Also used by the shared store for read-only config and the import. */ +export function localCollection(name: string, options: CollectionOptions): DocCollection { + assertName(name); + const layout = options.local; + + function fileFor(id: string): string { + assertDocId(id); + const dir = path.resolve(layout.dir()); + const file = path.resolve(dir, layout.file ? layout.file(id) : id); + if (!file.startsWith(dir + path.sep)) throw new Error(`Invalid path for ${name}/${id}`); + return file; + } + + async function readCurrent(file: string): Promise { + let data: Buffer; + try { + if (options.ttlMs) { + const info = await stat(file); + if (info.mtimeMs + options.ttlMs <= Date.now()) { + await unlink(file).catch(() => {}); + return null; + } + } + data = await readFile(file); + } catch (error) { + if (isMissing(error)) return null; + throw error; + } + return { data, version: versionOf(data) }; + } + + async function writeCurrent(file: string, data: Buffer): Promise { + await mkdir(path.dirname(file), { recursive: true }); + const tmp = layout.tmpName ? layout.tmpName(file) : `${file}.${process.pid}.${randomBytes(4).toString('hex')}.tmp`; + await writeFile(tmp, data, layout.mode !== undefined ? { mode: layout.mode } : undefined); + try { + await rename(tmp, file); + } catch (error) { + await unlink(tmp).catch(() => {}); + throw error; + } + } + + async function removeCurrent(file: string): Promise { + try { + await unlink(file); + } catch (error) { + if (!isMissing(error)) throw error; + } + } + + // Expired documents of a TTL collection are deleted at most every 10 s. + async function pruneExpired(): Promise { + if (!options.ttlMs) return; + const key = `${name}\0${path.resolve(layout.dir())}`; + const last = state().pruned.get(key) ?? 0; + if (Date.now() - last < 10_000) return; + state().pruned.set(key, Date.now()); + const cutoff = Date.now() - options.ttlMs; + for (const file of await walk(path.resolve(layout.dir()))) { + try { + if ((await stat(file)).mtimeMs <= cutoff) await unlink(file); + } catch { + /* gone already */ + } + } + } + + const queueKey = (file: string) => `doc\0${file}`; + + return { + name, + async get(id) { + return readCurrent(fileFor(id)); + }, + async put(id, data, putOptions: PutOptions = {}): Promise { + const file = fileFor(id); + const bytes = toBuffer(data); + await pruneExpired(); + return serializeLocal(queueKey(file), async () => { + if (putOptions.ifVersion !== undefined) { + const current = await readCurrent(file); + const matches = putOptions.ifVersion === null ? current === null : current?.version === putOptions.ifVersion; + if (!matches) return { ok: false, current }; + } + await writeCurrent(file, bytes); + return { ok: true, version: versionOf(bytes) }; + }); + }, + async update(id, fn) { + const file = fileFor(id); + return serializeLocal(queueKey(file), async () => { + const current = await readCurrent(file); + const next = fn(current?.data ?? null); + if (next === undefined) return { data: undefined, version: current?.version ?? null }; + if (next === null) { + if (current) await removeCurrent(file); + return { data: null, version: null }; + } + const bytes = toBuffer(next); + await writeCurrent(file, bytes); + return { data: bytes, version: versionOf(bytes) }; + }); + }, + async delete(id, putOptions: PutOptions = {}) { + const file = fileFor(id); + return serializeLocal(queueKey(file), async () => { + if (putOptions.ifVersion !== undefined) { + const current = await readCurrent(file); + const matches = putOptions.ifVersion === null ? current === null : current?.version === putOptions.ifVersion; + if (!matches) return false; + } + await removeCurrent(file); + return true; + }); + }, + async list(prefix = '') { + const dir = path.resolve(layout.dir()); + const ids: string[] = []; + for (const file of await walk(dir)) { + const relative = path.relative(dir, file).split(path.sep).join('/'); + const id = layout.idOf ? layout.idOf(relative) : relative; + if (id === null || !id.startsWith(prefix)) continue; + try { + assertDocId(id); + } catch { + continue; + } + if (options.ttlMs) { + try { + if ((await stat(file)).mtimeMs + options.ttlMs <= Date.now()) continue; + } catch { + continue; + } + } + ids.push(id); + } + return ids.sort(); + }, + }; +} + +/** Every regular file under `dir`, skipping temp files. */ +async function walk(dir: string): Promise { + let entries; + try { + entries = await readdir(dir, { withFileTypes: true }); + } catch (error) { + if (isMissing(error) || (error as NodeJS.ErrnoException)?.code === 'ENOTDIR') return []; + throw error; + } + const files: string[] = []; + for (const entry of entries) { + const full = path.join(dir, entry.name); + if (entry.isDirectory()) files.push(...(await walk(full))); + else if (entry.isFile() && !entry.name.endsWith('.tmp')) files.push(full); + } + return files; +} + +// ─── log ─────────────────────────────────────────────────────────────────── + +export function localLog(name: string, options: LogOptions): AppendLog { + assertName(name); + const { local } = options; + + async function rotateIfNeeded(file: string): Promise { + try { + const info = await stat(file); + if (info.size < local.maxBytes) return; + // log.N is dropped, log.(N-1) → log.N, …, log → log.1 + for (let i = local.rotations; i >= 1; i--) { + const from = i === 1 ? file : `${file}.${i - 1}`; + if (existsSync(from)) { + try { await rename(from, `${file}.${i}`); } catch { /* target may exist on overwrite */ } + } + } + } catch { + // stat failed: the file doesn't exist yet + } + } + + return { + async append(line) { + const file = path.resolve(local.path()); + await serializeLocal(`log\0${file}`, async () => { + await mkdir(path.dirname(file), { recursive: true }); + await appendFile(file, line.replace(/[\r\n]+/g, ' ') + '\n', 'utf-8'); + await rotateIfNeeded(file); + }); + }, + async readAll() { + try { + const content = await readFile(path.resolve(local.path()), 'utf-8'); + return content.split('\n').filter(Boolean); + } catch (error) { + if (isMissing(error)) return []; + throw error; + } + }, + async length() { + return (await this.readAll()).length; + }, + }; +} + +// ─── coordination ────────────────────────────────────────────────────────── + +const leases: LeaseStore = { + async acquire(name, holder, ttlMs) { + const current = state().leases.get(name); + const now = Date.now(); + if (current && current.expiresAt > now && current.holder !== holder) return false; + state().leases.set(name, { holder, expiresAt: now + ttlMs }); + return true; + }, + async release(name, holder) { + if (state().leases.get(name)?.holder === holder) state().leases.delete(name); + }, + async holder(name) { + const current = state().leases.get(name); + return current && current.expiresAt > Date.now() ? current.holder : null; + }, +}; + +const epochs: EpochStore = { + async bump(area) { + const next = (state().epochs.get(area) ?? 0) + 1; + state().epochs.set(area, next); + return next; + }, + async all() { + return Object.fromEntries(state().epochs); + }, +}; + +const replicas: ReplicaRegistry = { + async beat(info) { + state().replicas.set(info.id, { ...info, lastSeenAt: Date.now() }); + }, + async recent(withinMs) { + const cutoff = Date.now() - withinMs; + return [...state().replicas.values()].filter((r) => r.lastSeenAt >= cutoff); + }, +}; + +export function createLocalStore(): SharedStore { + return { + mode: 'local', + kv, + counters, + collection: localCollection, + log: localLog, + leases, + epochs, + replicas, + async ping() {}, + status() { + return { mode: 'local', location: 'local files', connected: true }; + }, + async ready() {}, + async close() {}, + }; +} + +/** Forget all in-memory state (tests). */ +export function resetLocalStoreForTests(): void { + const g = globalThis as typeof globalThis & { [STATE_KEY]?: LocalState }; + delete g[STATE_KEY]; +} diff --git a/lib/shared-store/redis.ts b/lib/shared-store/redis.ts new file mode 100644 index 000000000..ef6445124 --- /dev/null +++ b/lib/shared-store/redis.ts @@ -0,0 +1,460 @@ +import { randomBytes } from 'node:crypto'; +import Redis, { type RedisOptions } from 'ioredis'; +import { isConfigReadOnly } from '@/lib/admin/paths'; +import { assertDocId, assertKey, assertName, toBuffer, versionOf } from './keys'; +import { localCollection } from './local'; +import { + StoreUnavailableError, + type AppendLog, + type CollectionOptions, + type CounterStore, + type DocCollection, + type EpochStore, + type KvStore, + type LeaseStore, + type LogOptions, + type PutOptions, + type PutResult, + type ReplicaInfo, + type ReplicaRegistry, + type SharedStore, + type StoredDoc, +} from './types'; + +/** + * The shared store on a Redis-protocol server. Every check-then-write is a + * Lua script, so it is atomic on the server: WATCH would be tied to the + * connection, which every request in this process shares. Counter windows + * use the caller's clock, as today's limiters do (a few seconds of skew + * between replicas don't matter for 15-minute windows). Single endpoint only (no Cluster: the scripts touch + * several keys). + */ + +const SCRIPTS = { + // KEYS[1] doc hash, KEYS[2] index set. ARGV: data, version, mode (any|absent|version), expected, id, ttlMs + docPut: ` +local cur = redis.call('HGET', KEYS[1], 'v') +if ARGV[3] == 'absent' and cur then return 0 end +if ARGV[3] == 'version' and cur ~= ARGV[4] then return 0 end +redis.call('HSET', KEYS[1], 'd', ARGV[1], 'v', ARGV[2]) +if tonumber(ARGV[6]) > 0 then redis.call('PEXPIRE', KEYS[1], ARGV[6]) else redis.call('PERSIST', KEYS[1]) end +redis.call('SADD', KEYS[2], ARGV[5]) +return 1`, + // KEYS[1] doc hash, KEYS[2] index set. ARGV: mode (any|absent|version), expected, id + docDelete: ` +local cur = redis.call('HGET', KEYS[1], 'v') +if ARGV[1] == 'absent' and cur then return 0 end +if ARGV[1] == 'version' and cur ~= ARGV[2] then return 0 end +redis.call('DEL', KEYS[1]) +redis.call('SREM', KEYS[2], ARGV[3]) +return 1`, + // KEYS[1] key. ARGV: mode (absent|value), expected, new value ('' with ARGV[4]=1 means delete), delete flag, ttlMs (-1 keep, 0 none) + kvCas: ` +local cur = redis.call('GET', KEYS[1]) +if ARGV[1] == 'absent' and cur then return 0 end +if ARGV[1] == 'value' and cur ~= ARGV[2] then return 0 end +if ARGV[4] == '1' then redis.call('DEL', KEYS[1]) return 1 end +local ttl = tonumber(ARGV[5]) +if ttl == -1 then + local left = redis.call('PTTL', KEYS[1]) + if left > 0 then redis.call('SET', KEYS[1], ARGV[3], 'PX', left) else redis.call('SET', KEYS[1], ARGV[3]) end +elseif ttl > 0 then redis.call('SET', KEYS[1], ARGV[3], 'PX', ttl) +else redis.call('SET', KEYS[1], ARGV[3]) end +return 1`, + kvTake: ` +local v = redis.call('GET', KEYS[1]) +if v then redis.call('DEL', KEYS[1]) end +return v`, + // KEYS: counter hashes {n, r}. ARGV: now, max1, window1, max2, window2, … (max < 0 = no limit). + // Returns {ok, retryAfterMs, n1, r1, n2, r2, …} + counterReserve: ` +local now = tonumber(ARGV[1]) +local n, r = {}, {} +local ok, retry = 1, 0 +for i, k in ipairs(KEYS) do + local v = redis.call('HMGET', k, 'n', 'r') + local ni, ri = tonumber(v[1] or '0'), tonumber(v[2] or '0') + if ri <= now then ni, ri = 0, 0 end + n[i], r[i] = ni, ri + local max = tonumber(ARGV[2 * i]) + if max >= 0 and ni >= max then + ok = 0 + if ri - now > retry then retry = ri - now end + end +end +if ok == 1 then + for i, k in ipairs(KEYS) do + if r[i] == 0 then r[i] = now + tonumber(ARGV[2 * i + 1]) end + n[i] = n[i] + 1 + redis.call('HSET', k, 'n', n[i], 'r', r[i]) + -- Kept a minute past the window, so clock skew between replicas can't drop it early. + redis.call('PEXPIRE', k, math.max(r[i] - now, 0) + 60000) + end +end +local out = {ok, retry} +for i = 1, #KEYS do out[#out + 1] = n[i]; out[#out + 1] = r[i] end +return out`, + // KEYS: counters. ARGV: now, then resetAt per key (0 = any window) + counterRefund: ` +local now = tonumber(ARGV[1]) +for i, k in ipairs(KEYS) do + local v = redis.call('HMGET', k, 'n', 'r') + local n, r = tonumber(v[1] or '0'), tonumber(v[2] or '0') + local want = tonumber(ARGV[i + 1]) + if r > now and (want == 0 or want == r) and n > 0 then redis.call('HSET', k, 'n', n - 1) end +end +return 1`, + // KEYS[1] counter. ARGV: now + counterGet: ` +local now = tonumber(ARGV[1]) +local v = redis.call('HMGET', KEYS[1], 'n', 'r') +local n, r = tonumber(v[1] or '0'), tonumber(v[2] or '0') +if r <= now then return {0, 0} end +return {n, r}`, + // KEYS[1] lease. ARGV: holder, ttlMs + leaseAcquire: ` +local h = redis.call('GET', KEYS[1]) +if h == ARGV[1] then redis.call('PEXPIRE', KEYS[1], ARGV[2]) return 1 end +if h then return 0 end +redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2]) +return 1`, + leaseRelease: ` +if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('DEL', KEYS[1]) end +return 1`, +} as const; + +type ScriptName = keyof typeof SCRIPTS; +type Scripted = Redis & { [K in ScriptName]: (...args: Array) => Promise }; + +// Server replies that mean "not available right now", not "bug in our script". +const UNAVAILABLE_REPLY = /^(NOAUTH|WRONGPASS|LOADING|READONLY|MASTERDOWN|BUSY|CLUSTERDOWN|TRYAGAIN|ERR max number of clients)/; + +function unavailable(error: unknown): never { + if (error instanceof StoreUnavailableError) throw error; + const name = (error as Error)?.name; + const message = (error as Error)?.message ?? ''; + if (name === 'ReplyError' && !UNAVAILABLE_REPLY.test(message)) throw error; + throw new StoreUnavailableError(`The shared store is unavailable: ${message}`, { cause: error }); +} + +async function call(fn: () => Promise): Promise { + try { + return await fn(); + } catch (error) { + unavailable(error); + } +} + +/** The URL without its password, for logs and the admin overview. */ +export function redactStoreUrl(url: string): string { + try { + const u = new URL(url); + if (u.password) u.password = '***'; + return u.toString(); + } catch { + return 'redis://(invalid url)'; + } +} + +export interface RedisStoreOptions { + url: string; + /** Prefix for every key, so several deployments can share one server. Default `bulwark:`. */ + keyPrefix?: string; + /** Per-command timeout (ms). Default 2000. */ + commandTimeoutMs?: number; +} + +export function createRedisStore(options: RedisStoreOptions): SharedStore { + const prefix = options.keyPrefix ?? 'bulwark:'; + const redisOptions: RedisOptions = { + // Fail fast while disconnected, so callers see StoreUnavailableError at + // once and can fail closed, instead of piling up behind a queue. + enableOfflineQueue: false, + maxRetriesPerRequest: 1, + commandTimeout: options.commandTimeoutMs ?? 2000, + connectTimeout: 5000, + retryStrategy: (times) => Math.min(times * 250, 2000), + lazyConnect: false, + }; + const client = new Redis(options.url, redisOptions) as Scripted; + for (const [name, lua] of Object.entries(SCRIPTS)) { + client.defineCommand(name, { lua, numberOfKeys: name.startsWith('counter') ? undefined : name.startsWith('doc') ? 2 : 1 }); + } + // A failed connection is reported through commands and ping(); the + // client's own error events would otherwise go to stderr as unhandled. + client.on('error', () => {}); + + const k = { + kv: (key: string) => `${prefix}kv:${key}`, + ctr: (key: string) => `${prefix}ctr:${key}`, + doc: (coll: string, id: string) => `${prefix}doc:${coll}:${id}`, + idx: (coll: string) => `${prefix}docidx:${coll}`, + log: (name: string) => `${prefix}log:${name}`, + lease: (name: string) => `${prefix}lease:${name}`, + epochs: `${prefix}epochs`, + replicas: `${prefix}replicas`, + seen: `${prefix}replicas:seen`, + }; + + // ─── kv ────────────────────────────────────────────────────────────────── + + const kv: KvStore = { + async get(key) { + assertKey(key); + return call(() => client.get(k.kv(key))); + }, + async set(key, value, opts = {}) { + assertKey(key); + const args: Array = [k.kv(key), value]; + if (opts.ttlMs && opts.ttlMs > 0) args.push('PX', Math.ceil(opts.ttlMs)); + if (opts.ifAbsent) args.push('NX'); + const result = await call(() => client.call('SET', ...args)); + return result === 'OK'; + }, + async del(key) { + assertKey(key); + await call(() => client.del(k.kv(key))); + }, + async take(key) { + assertKey(key); + return (await call(() => client.kvTake(k.kv(key)))) as string | null; + }, + async update(key, fn, opts = {}) { + assertKey(key); + for (let attempt = 0; attempt < 20; attempt++) { + const current = await call(() => client.get(k.kv(key))); + const next = fn(current); + if (next === undefined) return undefined; + const ttl = opts.ttlMs && opts.ttlMs > 0 ? Math.ceil(opts.ttlMs) : -1; + const ok = await call(() => client.kvCas( + k.kv(key), current === null ? 'absent' : 'value', current ?? '', next ?? '', next === null ? '1' : '0', ttl, + )); + if (ok === 1) return next; + await backoff(attempt); + } + throw new Error(`Too many concurrent writes to ${key}`); + }, + }; + + // ─── counters ──────────────────────────────────────────────────────────── + + const counters: CounterStore = { + async reserve(limits, now = Date.now()) { + limits.forEach((l) => assertKey(l.key)); + const args: Array = [limits.length, ...limits.map((l) => k.ctr(l.key)), now]; + for (const l of limits) args.push(Number.isFinite(l.max) ? l.max : -1, Math.ceil(l.windowMs)); + const out = (await call(() => client.counterReserve(...args))) as number[]; + const counts: number[] = []; + const resetAts: number[] = []; + for (let i = 0; i < limits.length; i++) { + counts.push(Number(out[2 + 2 * i])); + resetAts.push(Number(out[3 + 2 * i])); + } + return { ok: Number(out[0]) === 1, retryAfterMs: Number(out[1]), counts, resetAts }; + }, + async refund(entries, now = Date.now()) { + if (entries.length === 0) return; + entries.forEach((e) => assertKey(e.key)); + await call(() => client.counterRefund(entries.length, ...entries.map((e) => k.ctr(e.key)), now, ...entries.map((e) => e.resetAt ?? 0))); + }, + async hit(key, windowMs, now = Date.now()) { + const result = await counters.reserve([{ key, max: Infinity, windowMs }], now); + return { count: result.counts[0], resetAt: result.resetAts[0] }; + }, + async get(key, now = Date.now()) { + assertKey(key); + const [n, r] = ((await call(() => client.counterGet(1, k.ctr(key), now))) as number[]).map(Number); + return r === 0 ? null : { count: n, resetAt: r }; + }, + async reset(key) { + assertKey(key); + await call(() => client.del(k.ctr(key))); + }, + }; + + // ─── documents ─────────────────────────────────────────────────────────── + + function collection(name: string, opts: CollectionOptions): DocCollection { + assertName(name); + // Read-only config comes from the mounted files on every replica. + if (opts.configDir && isConfigReadOnly()) return localCollection(name, opts); + const ttlMs = opts.ttlMs ?? 0; + + async function get(id: string): Promise { + assertDocId(id); + const [data, version] = await call(() => client.hmgetBuffer(k.doc(name, id), 'd', 'v')); + if (!data || !version) return null; + return { data, version: version.toString('utf8') }; + } + + async function put(id: string, data: Buffer | string, putOpts: PutOptions = {}): Promise { + assertDocId(id); + const bytes = toBuffer(data); + const version = versionOf(bytes); + const mode = putOpts.ifVersion === undefined ? 'any' : putOpts.ifVersion === null ? 'absent' : 'version'; + const ok = await call(() => client.docPut(k.doc(name, id), k.idx(name), bytes, version, mode, putOpts.ifVersion ?? '', id, ttlMs)); + if (ok === 1) return { ok: true, version }; + return { ok: false, current: await get(id) }; + } + + return { + name, + get, + put, + async update(id, fn, updateOpts = {}) { + const retries = updateOpts.maxRetries ?? 10; + for (let attempt = 0; attempt <= retries; attempt++) { + const current = await get(id); + const next = fn(current?.data ?? null); + if (next === undefined) return { data: undefined, version: current?.version ?? null }; + if (next === null) { + if (!current) return { data: null, version: null }; + const ok = await call(() => client.docDelete(k.doc(name, id), k.idx(name), 'version', current.version, id)); + if (ok === 1) return { data: null, version: null }; + } else { + const result = await put(id, next, { ifVersion: current?.version ?? null }); + if (result.ok) return { data: toBuffer(next), version: result.version }; + } + await backoff(attempt); + } + throw new Error(`Too many concurrent writes to ${name}/${id}`); + }, + async delete(id, putOpts: PutOptions = {}) { + assertDocId(id); + const mode = putOpts.ifVersion === undefined ? 'any' : putOpts.ifVersion === null ? 'absent' : 'version'; + return (await call(() => client.docDelete(k.doc(name, id), k.idx(name), mode, putOpts.ifVersion ?? '', id))) === 1; + }, + async list(prefix = '') { + const ids = (await call(() => client.smembers(k.idx(name)))).filter((id) => id.startsWith(prefix)); + if (ids.length === 0) return []; + // Expired documents (TTL collections) leave their id in the index. + const exists = await call(() => { + const pipeline = client.pipeline(); + for (const id of ids) pipeline.exists(k.doc(name, id)); + return pipeline.exec(); + }); + const live: string[] = []; + const stale: string[] = []; + ids.forEach((id, i) => ((exists?.[i]?.[1] as number) === 1 ? live : stale).push(id)); + if (stale.length) await call(() => client.srem(k.idx(name), ...stale)); + return live.sort(); + }, + }; + } + + // ─── log ───────────────────────────────────────────────────────────────── + + function log(name: string, opts: LogOptions): AppendLog { + assertName(name); + const key = k.log(name); + return { + async append(line) { + const clean = line.replace(/[\r\n]+/g, ' '); + await call(() => client.multi().rpush(key, clean).ltrim(key, -opts.maxEntries, -1).exec()); + }, + async readAll() { + return call(() => client.lrange(key, 0, -1)); + }, + async length() { + return call(() => client.llen(key)); + }, + }; + } + + // ─── coordination ──────────────────────────────────────────────────────── + + const leases: LeaseStore = { + async acquire(name, holder, ttlMs) { + assertName(name); + return (await call(() => client.leaseAcquire(k.lease(name), holder, Math.ceil(ttlMs)))) === 1; + }, + async release(name, holder) { + assertName(name); + await call(() => client.leaseRelease(k.lease(name), holder)); + }, + async holder(name) { + assertName(name); + return call(() => client.get(k.lease(name))); + }, + }; + + const epochs: EpochStore = { + async bump(area) { + return Number(await call(() => client.hincrby(k.epochs, area, 1))); + }, + async all() { + const raw = await call(() => client.hgetall(k.epochs)); + return Object.fromEntries(Object.entries(raw).map(([a, n]) => [a, Number(n)])); + }, + }; + + const REPLICA_KEEP_MS = 10 * 60 * 1000; + const replicas: ReplicaRegistry = { + async beat(info) { + const now = Date.now(); + await call(() => client.multi() + .hset(k.replicas, info.id, JSON.stringify({ ...info, lastSeenAt: now })) + .zadd(k.seen, now, info.id) + .exec()); + const old = await call(() => client.zrangebyscore(k.seen, '-inf', now - REPLICA_KEEP_MS)); + if (old.length) await call(() => client.multi().zrem(k.seen, ...old).hdel(k.replicas, ...old).exec()); + }, + async recent(withinMs) { + const ids = await call(() => client.zrangebyscore(k.seen, Date.now() - withinMs, '+inf')); + if (ids.length === 0) return []; + const raw = await call(() => client.hmget(k.replicas, ...ids)); + return raw.flatMap((r) => { + if (!r) return []; + try { return [JSON.parse(r) as ReplicaInfo]; } catch { return []; } + }); + }, + }; + + let readyPromise: Promise | null = null; + + return { + mode: 'redis', + kv, + counters, + collection, + log, + leases, + epochs, + replicas, + async ping(timeoutMs = 1000) { + let timer: NodeJS.Timeout | undefined; + try { + await Promise.race([ + call(() => client.ping()), + new Promise((_, reject) => { + timer = setTimeout(() => reject(new StoreUnavailableError('The shared store did not answer in time')), timeoutMs); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } + }, + status() { + return { mode: 'redis', location: redactStoreUrl(options.url), connected: client.status === 'ready' }; + }, + ready() { + readyPromise ??= client.status === 'ready' + ? Promise.resolve() + : new Promise((resolve) => client.once('ready', () => resolve())); + return readyPromise; + }, + async close() { + client.disconnect(); + }, + }; +} + +function backoff(attempt: number): Promise { + const ms = Math.min(5 + attempt * 10, 100) * (0.5 + Math.random()); + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** An id unique to this process, for leases and the replica list. */ +export function randomHolderSuffix(): string { + return randomBytes(3).toString('hex'); +} diff --git a/lib/shared-store/types.ts b/lib/shared-store/types.ts new file mode 100644 index 000000000..2d852c86e --- /dev/null +++ b/lib/shared-store/types.ts @@ -0,0 +1,256 @@ +/** + * The storage interface every piece of server state that must be the same on + * every replica goes through (#1066). Two implementations: + * + * - local (default): today's behaviour. Short-lived state in memory on + * globalThis, durable state in the files each collection declares, with + * the same paths and formats as before. + * - redis: one Redis-protocol server (Redis, Valkey, KeyDB, Dragonfly) + * shared by every replica, selected by REDIS_URL. + * + * Stores never talk to Redis directly and never ask which implementation + * they have, except to choose their behaviour when the shared store is down + * (StoreUnavailableError). + */ + +export type StoreMode = 'local' | 'redis'; + +/** The shared store could not be reached or didn't answer in time. */ +export class StoreUnavailableError extends Error { + constructor(message = 'The shared store is unavailable', options?: { cause?: unknown }) { + super(message, options); + this.name = 'StoreUnavailableError'; + } +} + +/** A collection or document id that would leave its namespace. */ +export class InvalidStoreKeyError extends Error { + constructor(key: string) { + super(`Invalid store key: ${JSON.stringify(key)}`); + this.name = 'InvalidStoreKeyError'; + } +} + +// ─── Short-lived values ──────────────────────────────────────────────────── + +export interface KvSetOptions { + /** Expire after this many milliseconds. Without it the value stays until deleted. */ + ttlMs?: number; + /** Only set when the key doesn't exist. `set` then returns false if it did. */ + ifAbsent?: boolean; +} + +/** + * String values that may expire. Local: memory only, gone on restart, as + * today's in-process maps are. + */ +export interface KvStore { + get(key: string): Promise; + /** Returns false only when `ifAbsent` was set and the key existed. */ + set(key: string, value: string, options?: KvSetOptions): Promise; + del(key: string): Promise; + /** Read and delete in one step: of several callers, exactly one gets the value. */ + take(key: string): Promise; + /** + * Read, change and write one value atomically. `fn` gets the current value + * (null if missing) and returns the new one, or null to delete it, or + * undefined to leave it as it is. Returns what `fn` returned. `fn` may run + * more than once in shared mode (on a conflicting write), so it must not + * have side effects. `ttlMs` applies to the written value. + */ + update(key: string, fn: (current: string | null) => string | null | undefined, options?: { ttlMs?: number }): Promise; +} + +// ─── Counters for limits ─────────────────────────────────────────────────── + +export interface CounterLimit { + key: string; + /** Reserve fails when the counter already holds this many. `Infinity` for no limit. */ + max: number; + /** Fixed window: it starts with the first count and resets after this long. */ + windowMs: number; +} + +export interface ReserveResult { + ok: boolean; + /** When refused: how long until the fullest refusing window resets. */ + retryAfterMs: number; + /** Counts after the reservation (or as they were, when refused), in the order given. */ + counts: number[]; + /** When each window resets (ms), in the order given; 0 for a counter that doesn't exist. Pass to refund. */ + resetAts: number[]; +} + +/** + * Fixed-window counters, as every login limit uses today. Local: memory. + */ +export interface CounterStore { + /** + * Atomically: if every counter is below its max, add one to each and + * return ok; otherwise change nothing. `now` (ms) is the caller's clock, + * as today's limiters take it; default Date.now(). + */ + reserve(limits: CounterLimit[], now?: number): Promise; + /** + * Take back one count from each key, never below zero. With `resetAt` + * (from reserve) only from that same window: a window that has since + * reset is left alone. + */ + refund(entries: Array<{ key: string; resetAt?: number }>, now?: number): Promise; + /** Add one and return the new count and when the window resets. */ + hit(key: string, windowMs: number, now?: number): Promise<{ count: number; resetAt: number }>; + get(key: string, now?: number): Promise<{ count: number; resetAt: number } | null>; + reset(key: string): Promise; +} + +// ─── Durable documents ───────────────────────────────────────────────────── + +/** Where a collection lives in local mode. */ +export interface LocalLayout { + /** The directory, resolved on every call so env changes in tests apply. */ + dir: () => string; + /** File name for an id; default: the id itself. May contain `/` for subfolders. */ + file?: (id: string) => string; + /** Map a relative file path back to its id, for list(); default: identity. Return null to skip a file. */ + idOf?: (relativePath: string) => string | null; + /** File mode for writes, e.g. 0o600 for files holding secrets. */ + mode?: number; + /** + * Today's temp-file name for a target, when a store's tests or tooling + * depend on it. Default: `...tmp`. + */ + tmpName?: (target: string) => string; +} + +export interface CollectionOptions { + local: LocalLayout; + /** + * The collection is part of the admin config dir. When + * ADMIN_CONFIG_READONLY is set it is read from the mounted files on every + * replica, in shared mode too, and writes are refused upstream. + */ + configDir?: boolean; + /** + * Documents expire this long after they were written (local: by file + * mtime, pruned when the collection is used). + */ + ttlMs?: number; +} + +export interface StoredDoc { + data: Buffer; + /** Changes on every write. Opaque. */ + version: string; +} + +export interface PutOptions { + /** + * Only write when the document is at this version; `null` = only when it + * doesn't exist. Omit for an unconditional write. + */ + ifVersion?: string | null; +} + +export type PutResult = { ok: true; version: string } | { ok: false; current: StoredDoc | null }; + +export interface UpdateOptions { + /** Conflicting writes in shared mode are retried this often (default 10). */ + maxRetries?: number; +} + +export interface DocCollection { + readonly name: string; + get(id: string): Promise; + put(id: string, data: Buffer | string, options?: PutOptions): Promise; + /** + * Read, change and write one document without losing a concurrent write. + * `fn` gets the current data (null if missing) and returns new data, null + * to delete, or undefined to leave it. It may run more than once, so no + * side effects. Returns `fn`'s last result and the version written. + */ + update( + id: string, + fn: (current: Buffer | null) => Buffer | string | null | undefined, + options?: UpdateOptions, + ): Promise<{ data: Buffer | null | undefined; version: string | null }>; + /** Returns false when `ifVersion` didn't match. */ + delete(id: string, options?: PutOptions): Promise; + /** Ids under a prefix (`''` for all), sorted. */ + list(prefix?: string): Promise; +} + +// ─── Append-only log ─────────────────────────────────────────────────────── + +export interface LogOptions { + local: { + /** The log file; rotated to `.1`… when it passes `maxBytes`. */ + path: () => string; + maxBytes: number; + rotations: number; + }; + /** Shared mode keeps the newest this many lines. */ + maxEntries: number; +} + +export interface AppendLog { + /** One line, without a newline. Lines are never interleaved or torn. */ + append(line: string): Promise; + /** Every line, oldest first (local: the current file only, as today). */ + readAll(): Promise; + length(): Promise; +} + +// ─── Coordination ────────────────────────────────────────────────────────── + +export interface LeaseStore { + /** Take or renew a lease. True when `holder` holds it afterwards. */ + acquire(name: string, holder: string, ttlMs: number): Promise; + /** Give it up, if `holder` still holds it. */ + release(name: string, holder: string): Promise; + holder(name: string): Promise; +} + +export interface EpochStore { + /** Add one to an area's epoch and return it. */ + bump(area: string): Promise; + all(): Promise>; +} + +export interface ReplicaInfo { + id: string; + version: string; + startedAt: number; + lastSeenAt: number; +} + +export interface ReplicaRegistry { + beat(info: Omit): Promise; + /** Replicas that beat within the last `withinMs`. */ + recent(withinMs: number): Promise; +} + +// ─── The store ───────────────────────────────────────────────────────────── + +export interface StoreStatus { + mode: StoreMode; + /** Where it is, with any password removed, e.g. `redis://valkey:6379/0`. */ + location: string; + connected: boolean; +} + +export interface SharedStore { + readonly mode: StoreMode; + readonly kv: KvStore; + readonly counters: CounterStore; + collection(name: string, options: CollectionOptions): DocCollection; + log(name: string, options: LogOptions): AppendLog; + readonly leases: LeaseStore; + readonly epochs: EpochStore; + readonly replicas: ReplicaRegistry; + /** Resolves when the store answered; throws StoreUnavailableError otherwise. */ + ping(timeoutMs?: number): Promise; + status(): StoreStatus; + /** Wait for the first connection (shared mode). Local: resolves at once. */ + ready(): Promise; + close(): Promise; +} diff --git a/next.config.ts b/next.config.ts index 4e72c6948..03ddb9b74 100644 --- a/next.config.ts +++ b/next.config.ts @@ -59,8 +59,10 @@ const nextConfig: NextConfig = { basePath: basePath || undefined, // esbuild ships native binaries + a README the bundler can't parse; load // it from node_modules at runtime instead of trying to bundle it. Used by - // PLUGIN_DEV_DIR's on-the-fly bundler. - serverExternalPackages: ["esbuild"], + // PLUGIN_DEV_DIR's on-the-fly bundler. ioredis (the shared store, #1066) + // is loaded from node_modules too, so the proxy and the route bundles + // don't each carry a copy. + serverExternalPackages: ["esbuild", "ioredis"], // The build's type check leaves out tests, which are a third of the files // and push tsc past the default heap on a 4 GB machine (#1431). // `npm run typecheck` still checks them with tsconfig.json. diff --git a/package-lock.json b/package-lock.json index b657198c3..878d8cb72 100644 --- a/package-lock.json +++ b/package-lock.json @@ -35,6 +35,7 @@ "clsx": "^2.1.1", "date-fns": "^4.4.0", "dompurify": "^3.4.12", + "ioredis": "^5.11.1", "jalaali-js": "^2.0.1", "jszip": "^3.10.2", "next": "^16.4.3", @@ -1921,6 +1922,12 @@ "url": "https://opencollective.com/libvips" } }, + "node_modules/@ioredis/commands": { + "version": "1.10.0", + "resolved": "https://registry.npmjs.org/@ioredis/commands/-/commands-1.10.0.tgz", + "integrity": "sha512-UmeW7z4LfctwoQ5wkhVzgq8tXkreED2xZGpX+Bg+zA+WJFZCT6c062AfCK/Dfk81xZnnwdhJCUMkitihRaoC2Q==", + "license": "MIT" + }, "node_modules/@jridgewell/gen-mapping": { "version": "0.3.13", "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.13.tgz", @@ -3427,6 +3434,72 @@ "node": ">=14.0.0" } }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/core": { + "version": "1.11.1", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@emnapi/wasi-threads": "1.2.2", + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/runtime": { + "version": "1.11.1", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/wasi-threads": { + "version": "1.2.2", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@napi-rs/wasm-runtime": { + "version": "1.1.4", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@tybys/wasm-util": "^0.10.1" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/Brooooooklyn" + }, + "peerDependencies": { + "@emnapi/core": "^1.7.1", + "@emnapi/runtime": "^1.7.1" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@tybys/wasm-util": { + "version": "0.10.2", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/tslib": { + "version": "2.8.1", + "dev": true, + "inBundle": true, + "license": "0BSD", + "optional": true + }, "node_modules/@tailwindcss/oxide-win32-arm64-msvc": { "version": "4.3.3", "resolved": "https://registry.npmjs.org/@tailwindcss/oxide-win32-arm64-msvc/-/oxide-win32-arm64-msvc-4.3.3.tgz", @@ -5069,6 +5142,15 @@ "node": ">=6" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.1.tgz", + "integrity": "sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/color-convert": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/color-convert/-/color-convert-2.0.1.tgz", @@ -5231,7 +5313,6 @@ "version": "4.4.3", "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", - "dev": true, "license": "MIT", "dependencies": { "ms": "^2.1.3" @@ -5304,6 +5385,15 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/denque": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz", + "integrity": "sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10" + } + }, "node_modules/dequal": { "version": "2.0.3", "resolved": "https://registry.npmjs.org/dequal/-/dequal-2.0.3.tgz", @@ -6572,6 +6662,28 @@ "integrity": "sha512-j8cUmOJzVgkHuS0QiQ6ga76UIoLOFSAMWhs7aZJztH3aAdCOAE6vpC8KVvFB4cU10ON0y2/5oOVmPJ43s2lTwA==", "license": "MIT" }, + "node_modules/ioredis": { + "version": "5.11.1", + "resolved": "https://registry.npmjs.org/ioredis/-/ioredis-5.11.1.tgz", + "integrity": "sha512-ehuGcf94bQXhfagULNXrJdfnWO38v070jxSx/qE87Kjzmu2fU7ro5EFAb+OPituLqgfyuQaym5DlrNydW2sJ9A==", + "license": "MIT", + "dependencies": { + "@ioredis/commands": "1.10.0", + "cluster-key-slot": "1.1.1", + "debug": "4.4.3", + "denque": "2.1.0", + "redis-errors": "1.2.0", + "redis-parser": "3.0.0", + "standard-as-callback": "2.1.0" + }, + "engines": { + "node": ">=12.22.0" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/ioredis" + } + }, "node_modules/is-array-buffer": { "version": "3.0.5", "resolved": "https://registry.npmjs.org/is-array-buffer/-/is-array-buffer-3.0.5.tgz", @@ -7607,7 +7719,6 @@ "version": "2.1.3", "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", - "dev": true, "license": "MIT" }, "node_modules/nanoid": { @@ -8512,6 +8623,27 @@ "node": ">=8" } }, + "node_modules/redis-errors": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz", + "integrity": "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/redis-parser": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/redis-parser/-/redis-parser-3.0.0.tgz", + "integrity": "sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==", + "license": "MIT", + "dependencies": { + "redis-errors": "^1.0.0" + }, + "engines": { + "node": ">=4" + } + }, "node_modules/reflect.getprototypeof": { "version": "1.0.10", "resolved": "https://registry.npmjs.org/reflect.getprototypeof/-/reflect.getprototypeof-1.0.10.tgz", @@ -8957,6 +9089,12 @@ "node": ">=0.10.0" } }, + "node_modules/standard-as-callback": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/standard-as-callback/-/standard-as-callback-2.1.0.tgz", + "integrity": "sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==", + "license": "MIT" + }, "node_modules/std-env": { "version": "4.3.0", "resolved": "https://registry.npmjs.org/std-env/-/std-env-4.3.0.tgz", diff --git a/package.json b/package.json index d51009a05..2f34fcd6b 100644 --- a/package.json +++ b/package.json @@ -65,6 +65,7 @@ "clsx": "^2.1.1", "date-fns": "^4.4.0", "dompurify": "^3.4.12", + "ioredis": "^5.11.1", "jalaali-js": "^2.0.1", "jszip": "^3.10.2", "next": "^16.4.3",