mirror of
https://github.com/bulwarkmail/webmail.git
synced 2026-11-02 08:15:55 +00:00
462 lines
18 KiB
TypeScript
462 lines
18 KiB
TypeScript
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,
|
|
isStoreUnavailableError,
|
|
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 (isStoreUnavailableError(error)) 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');
|
|
}
|