mirror of
https://github.com/bulwarkmail/webmail.git
synced 2026-11-01 08:12:43 +00:00
feat: add a shared store that keeps today's files by default and can use a redis-protocol server #1066
This commit is contained in:
Linus Rath
@@ -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<void>;
|
||||
}
|
||||
|
||||
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> = {}): 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);
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -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(),
|
||||
});
|
||||
@@ -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();
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<string, { value: string; expiresAt: number }>;
|
||||
kvSets: number;
|
||||
counters: Map<string, { count: number; resetAt: number }>;
|
||||
queues: Map<string, Promise<unknown>>;
|
||||
pruned: Map<string, number>;
|
||||
leases: Map<string, { holder: string; expiresAt: number }>;
|
||||
epochs: Map<string, number>;
|
||||
replicas: Map<string, ReplicaInfo>;
|
||||
}
|
||||
|
||||
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<T>(key: string, task: () => Promise<T>): Promise<T> {
|
||||
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<ReserveResult> {
|
||||
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<StoredDoc | null> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<PutResult> {
|
||||
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<string[]> {
|
||||
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<void> {
|
||||
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];
|
||||
}
|
||||
@@ -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<string | number | Buffer>) => Promise<unknown> };
|
||||
|
||||
// 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<T>(fn: () => Promise<T>): Promise<T> {
|
||||
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<string | number> = [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<string | number> = [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<StoredDoc | null> {
|
||||
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<PutResult> {
|
||||
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<void> | 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<never>((_, 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<void>((resolve) => client.once('ready', () => resolve()));
|
||||
return readyPromise;
|
||||
},
|
||||
async close() {
|
||||
client.disconnect();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function backoff(attempt: number): Promise<void> {
|
||||
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');
|
||||
}
|
||||
@@ -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<string | null>;
|
||||
/** Returns false only when `ifAbsent` was set and the key existed. */
|
||||
set(key: string, value: string, options?: KvSetOptions): Promise<boolean>;
|
||||
del(key: string): Promise<void>;
|
||||
/** Read and delete in one step: of several callers, exactly one gets the value. */
|
||||
take(key: string): Promise<string | null>;
|
||||
/**
|
||||
* 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<string | null | undefined>;
|
||||
}
|
||||
|
||||
// ─── 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<ReserveResult>;
|
||||
/**
|
||||
* 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<void>;
|
||||
/** 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<void>;
|
||||
}
|
||||
|
||||
// ─── 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: `<target>.<pid>.<random>.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<StoredDoc | null>;
|
||||
put(id: string, data: Buffer | string, options?: PutOptions): Promise<PutResult>;
|
||||
/**
|
||||
* 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<boolean>;
|
||||
/** Ids under a prefix (`''` for all), sorted. */
|
||||
list(prefix?: string): Promise<string[]>;
|
||||
}
|
||||
|
||||
// ─── 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<void>;
|
||||
/** Every line, oldest first (local: the current file only, as today). */
|
||||
readAll(): Promise<string[]>;
|
||||
length(): Promise<number>;
|
||||
}
|
||||
|
||||
// ─── Coordination ──────────────────────────────────────────────────────────
|
||||
|
||||
export interface LeaseStore {
|
||||
/** Take or renew a lease. True when `holder` holds it afterwards. */
|
||||
acquire(name: string, holder: string, ttlMs: number): Promise<boolean>;
|
||||
/** Give it up, if `holder` still holds it. */
|
||||
release(name: string, holder: string): Promise<void>;
|
||||
holder(name: string): Promise<string | null>;
|
||||
}
|
||||
|
||||
export interface EpochStore {
|
||||
/** Add one to an area's epoch and return it. */
|
||||
bump(area: string): Promise<number>;
|
||||
all(): Promise<Record<string, number>>;
|
||||
}
|
||||
|
||||
export interface ReplicaInfo {
|
||||
id: string;
|
||||
version: string;
|
||||
startedAt: number;
|
||||
lastSeenAt: number;
|
||||
}
|
||||
|
||||
export interface ReplicaRegistry {
|
||||
beat(info: Omit<ReplicaInfo, 'lastSeenAt'>): Promise<void>;
|
||||
/** Replicas that beat within the last `withinMs`. */
|
||||
recent(withinMs: number): Promise<ReplicaInfo[]>;
|
||||
}
|
||||
|
||||
// ─── 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<void>;
|
||||
status(): StoreStatus;
|
||||
/** Wait for the first connection (shared mode). Local: resolves at once. */
|
||||
ready(): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
+4
-2
@@ -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.
|
||||
|
||||
Generated
+140
-2
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user