openvibers/openvibe.tools Commits
Tools guard review fixes (plan T8): one Lua call decides and charges a run's buckets and day allowances all or nothing (a run refused on the day leg spent bucket tokens, so refused requests could drain a bucket); the store's day counters are one atomic check-and-increment (HGET then HINCRBYFLOAT let concurrent runs overshoot the day); the salt is asked of the store every time (one cached before Valkey answered stayed process-local all day); each app prunes only its own guard_abuse rows. guard.test.js covers each; guard-valkey.test.js runs the Lua on a real Valkey under test:pg.
Changes
@@@ -106,12 +106,9 @@ function createGuard(o = {}) {const store = o.store || createGuardStore({ db: o.db || null, valkey: o.valkey || null, app: appName, now, log });if (!o.store) void store.warm();-- let saltDay = null, saltValue = null;-- const salt = () => {-- const d = dayOf(now());-- if (d !== saltDay) { saltValue = store.salt(d); saltDay = d; }-- return saltValue;-- };++ // Asked of the store every time (a Map lookup), never cached here: a salt handed out before Valkey++ // answered is a process-local stand-in, and the shared one must replace it as soon as it loads.++ const salt = () => store.salt(dayOf(now()));// Network's sign-out-everywhere cutoffs (PostgreSQL, read through a short shared cache): the gateway// writes them (apps/gateway/server/revocation-events.js); every app reads them here.@@@ -3,10 +3,12 @@// Tiered quotas: a token bucket per (key, class) and a UTC-day allowance per (key, class). A run takes// its descriptor's `cost` from both, all or nothing.//--// The day allowance is shared across processes through the guard store (Valkey, plan T8 decision 4) and--// so are the minute/burst buckets when a Valkey connection is passed: one Lua call refills and spends--// one bucket atomically, so two hosts enforce one quota. Without Valkey (or on an outage) both legs use--// this process's own counters — a throttle that never fails open and never fails every request.++// The day allowance and the minute/burst buckets are shared across processes in Valkey (plan T8++// decision 4) when a connection is passed: ONE Lua call refills every bucket a run touches, checks++// every bucket and every day allowance, and charges them all only when all of them fit — so two hosts++// enforce one quota, and a refused run (say, on the day leg) never spends a bucket token. Without Valkey++// (or on an outage) the buckets are this process's and the day allowances go through the guard store's++// own atomic check-and-charge — a throttle that never fails open and never fails every request.//// Who is charged (keys come from the caller resolver, addresses already hashed):// anonymous its address (IPv6: its /64) at the anonymous tier@@@ -21,27 +23,56 @@// ═══════════════════════════════════════════════════════════════const { SESSION_IP_SHARE } = require('./limits');--const { dayOf, DAY_MS } = require('./store');++const { dayOf, dayTtl, dayKey, dayField, DAY_MS } = require('./store');--// Refill a bucket (tokens + elapsed × rate, capped at burst) and spend `need`, atomically. It only--// consumes when the bucket holds `need`: a refused run changes nothing. Returns "tokens:ok".--// ARGV: burst, rate (tokens/ms), need, now (ms), ttl (s).--const SPEND = `--local key = KEYS[1]--local burst = tonumber(ARGV[1])--local rate = tonumber(ARGV[2])--local need = tonumber(ARGV[3])--local now = tonumber(ARGV[4])--local ttl = tonumber(ARGV[5])--local d = redis.call('HMGET', key, 'tokens', 'at')--local tokens = (d[1] and tonumber(d[1])) or burst--local at = (d[2] and tonumber(d[2])) or now--tokens = math.min(burst, tokens + (now - at) * rate)--if tokens < need then return tostring(tokens)..':0' end--local left = tokens - need--redis.call('HSET', key, 'tokens', tostring(left), 'at', tostring(now))--redis.call('EXPIRE', key, ttl)--return tostring(left)..':1'`;++// The whole quota decision for one run, atomically: refill each bucket (tokens + elapsed × rate, capped++// at burst), read each day allowance, and only when every bucket holds its `need` and every allowance++// has room for `cost` — or in report mode (the work happens anyway) — take them all. A refused run++// changes nothing. KEYS[1..n] the buckets, KEYS[n+1] the day hash (guard_day, shared with the store).++// ARGV: n, now (ms), bucket ttl (s), day ttl (s), report (1/0), then per charge: burst, rate++// (tokens/ms), need, perDay (0 = none), cost, day field. Returns per charge "tokens,used,minuteOk,dayOk"++// (tokens and used before this run), joined with ";".++const CHARGE = `++local n = tonumber(ARGV[1])++local now = tonumber(ARGV[2])++local bttl = tonumber(ARGV[3])++local dttl = tonumber(ARGV[4])++local report = ARGV[5] == '1'++local day = KEYS[n + 1]++local ok = true++local st = {}++local out = {}++for i = 1, n do++ local b = 5 + (i - 1) * 6++ local burst = tonumber(ARGV[b + 1])++ local rate = tonumber(ARGV[b + 2])++ local need = tonumber(ARGV[b + 3])++ local perDay = tonumber(ARGV[b + 4])++ local cost = tonumber(ARGV[b + 5])++ local field = ARGV[b + 6]++ local d = redis.call('HMGET', KEYS[i], 'tokens', 'at')++ local tokens = (d[1] and tonumber(d[1])) or burst++ local at = (d[2] and tonumber(d[2])) or now++ tokens = math.min(burst, tokens + math.max(0, now - at) * rate)++ local used = 0++ if perDay > 0 then used = tonumber(redis.call('HGET', day, field) or '0') end++ local minuteOk = tokens >= need++ local dayOk = perDay <= 0 or used + cost <= perDay++ if not (minuteOk and dayOk) then ok = false end++ st[i] = { tokens, need, perDay, cost, field }++ out[i] = tostring(tokens) .. ',' .. tostring(used) .. ',' .. (minuteOk and '1' or '0') .. ',' .. (dayOk and '1' or '0')++end++if ok or report then++ local days = false++ for i = 1, n do++ local c = st[i]++ redis.call('HSET', KEYS[i], 'tokens', tostring(math.max(0, c[1] - c[2])), 'at', tostring(now))++ redis.call('EXPIRE', KEYS[i], bttl)++ if c[3] > 0 then redis.call('HINCRBYFLOAT', day, c[5], c[4]); days = true end++ end++ if days then redis.call('EXPIRE', day, dttl) end++end++return table.concat(out, ';')`;// Refill and give `need` back (a run that cost less than the up-front charge), capped at burst.const REFUND = `local key = KEYS[1]@@@ -61,17 +92,27 @@ return tostring(left)`;function createValkeyBuckets(valkey, log) {const c = valkey.client;-- if (typeof c.ovBucketSpend !== 'function') c.defineCommand('ovBucketSpend', { lua: SPEND });++ if (typeof c.ovQuotaCharge !== 'function') c.defineCommand('ovQuotaCharge', { lua: CHARGE });if (typeof c.ovBucketRefund !== 'function') c.defineCommand('ovBucketRefund', { lua: REFUND });let up = true;return {get down() { return !up; },-- /** { left, ok } — ok false means the bucket was short and nothing was taken. null: Valkey is down. */-- async spend(id, lim, need, t) {++ /**++ * Decide and charge a run in one call. charges: [{ id, lim, need, cost, field }].++ * → [{ tokens, used, minuteOk, dayOk }] (before this run); null: Valkey is down.++ */++ async charge(charges, day, t, report) {try {-- const r = String(await c.ovBucketSpend(1, valkey.key('bucket', id), String(lim.burst), String(lim.perMinute / 60_000), String(need), String(t), String(600)));-- const [left, ok] = r.split(':');-- return { left: Number(left), ok: ok === '1' };++ const args = [String(charges.length), String(t), '600', String(dayTtl(t)), report ? '1' : '0'];++ for (const x of charges) {++ args.push(String(x.lim.burst), String(x.lim.perMinute / 60_000), String(x.need), String(x.lim.perDay > 0 ? x.lim.perDay : 0), String(x.cost), x.field);++ }++ const keys = charges.map((x) => valkey.key('bucket', x.id)).concat(dayKey(valkey, day));++ const r = String(await c.ovQuotaCharge(keys.length, ...keys, ...args));++ return r.split(';').map((row) => {++ const [tokens, used, m, d] = row.split(',');++ return { tokens: Number(tokens), used: Number(used), minuteOk: m === '1', dayOk: d === '1' };++ });} catch (err) { up = false; log.warn(`[Quota] Valkey buckets unavailable (${err.message}); counting in this process`); return null; }},async refund(id, lim, need, t) {@@@ -82,7 +123,7 @@ function createValkeyBuckets(valkey, log) {/*** @param {object} o-- * @param {object} o.store guard store (async dayUsed, dayAdd)++ * @param {object} o.store guard store (async dayCharge: the day allowances without Valkey)* @param {() => object} o.quotas the QUOTAS table (limits.quotas())* @param {() => number} [o.now]* @param {number} [o.maxBuckets]@@@ -132,34 +173,43 @@ function createQuotas(o) {const t = now();sweep(t);const table = o.quotas();-- const charges = chargesFor(caller, quotaClass, table);++ const charges = chargesFor(caller, quotaClass, table)++ .filter((c) => c.lim.burst > 0 && c.lim.perMinute > 0)++ .map((c) => ({++ ...c, cost, id: `${quotaClass}|${c.key}`, field: dayField(c.key, quotaClass),++ need: Math.min(cost, c.lim.burst), // one run never costs more than a full bucket++ }));if (!charges.length) return { ok: true, limit: null, remaining: null, reset: null, retryAfter: 0 };const day = dayOf(t);const nextDay = Math.ceil((Math.floor(t / DAY_MS) + 1) * DAY_MS / 1000) - Math.floor(t / 1000);++++ // Shared (Valkey): the whole decision and the charge in one atomic call. Otherwise (no Valkey, or++ // an outage) this process's buckets, and the store's atomic day charge — committed only when the++ // buckets fit too, so a refused run takes nothing from either leg.++ let states = shared && !shared.down ? await shared.charge(charges, day, t, !!opt.report) : null;++ if (!states) {++ const levels = charges.map((c) => level(c.id, c.lim, t));++ const minuteOk = charges.map((c, i) => levels[i] >= c.need);++ const allMinute = minuteOk.every(Boolean);++ const withDay = charges.filter((c) => c.lim.perDay > 0);++ const d = withDay.length++ ? await o.store.dayCharge(day, withDay.map((c) => ({ key: c.key, cls: quotaClass, n: cost, limit: c.lim.perDay })), { commit: allMinute, force: !!opt.report })++ : { ok: true, used: [] };++ const usedOf = new Map(withDay.map((c, i) => [c, d.used[i] || 0]));++ states = charges.map((c, i) => {++ const used = usedOf.get(c) || 0;++ return { tokens: levels[i], used, minuteOk: minuteOk[i], dayOk: !(c.lim.perDay > 0) || used + cost <= c.lim.perDay };++ });++ if ((allMinute && d.ok) || opt.report) {++ for (const c of charges) spendLocal(c.id, c.lim, c.need, t);++ }++ }++let ok = true, retryAfter = 0, binding = null, scope = null;-- const views = [];-- for (const c of charges) {-- const { lim } = c;-- if (!(lim.burst > 0) || !(lim.perMinute > 0)) continue;-- const need = Math.min(cost, lim.burst); // one run never costs more than a full bucket-- const id = `${quotaClass}|${c.key}`;++ const views = charges.map((c, i) => {++ const { lim, need } = c;++ const { tokens, used, minuteOk, dayOk } = states[i];const rate = lim.perMinute / 60_000;-- // Shared (Valkey): refill and spend in one atomic call, and decide from its answer. Otherwise-- // (no Valkey, or an outage) decide and spend from this process's own map below.-- let tokens, minuteOk, done = false;-- if (shared && !shared.down) {-- const r = await shared.spend(id, lim, need, t);-- if (r) { tokens = r.ok ? r.left + need : r.left; minuteOk = r.ok; done = true; }-- }-- if (!done) { tokens = level(id, lim, t); minuteOk = tokens >= need; }-- const used = lim.perDay > 0 ? await o.store.dayUsed(day, c.key, quotaClass) : 0;-- const dayOk = !(lim.perDay > 0) || used + cost <= lim.perDay;-- const after = Math.max(0, tokens - need);-- views.push({-- c, need, tokens, rate, dayUsed: used, done,-- minute: { limit: lim.burst, remaining: Math.floor(after), reset: Math.ceil((lim.burst - after) / rate / 1000) },-- day: lim.perDay > 0 ? { limit: lim.perDay, remaining: Math.max(0, Math.floor(lim.perDay - used - cost)), reset: nextDay } : null,-- });if (!minuteOk) {ok = false;const wait = Math.ceil((need - tokens) / rate / 1000);@@@ -169,13 +219,12 @@ function createQuotas(o) {ok = false;if (nextDay > retryAfter) { retryAfter = nextDay; binding = 'day'; scope = c.scope; }}-- }-- if (ok || opt.report) {-- for (const v of views) {-- if (!v.done) await spend(`${quotaClass}|${v.c.key}`, v.c.lim, v.need, t, v.tokens);-- if (v.c.lim.perDay > 0) await o.store.dayAdd(day, v.c.key, quotaClass, cost);-- }-- }++ const after = Math.max(0, tokens - need);++ return {++ minute: { limit: lim.burst, remaining: Math.floor(after), reset: Math.ceil((lim.burst - after) / rate / 1000) },++ day: lim.perDay > 0 ? { limit: lim.perDay, remaining: Math.max(0, Math.floor(lim.perDay - used - cost)), reset: nextDay } : null,++ };++ });// The headers describe the allowance closest to running out, in units of this run's cost.let head = null;for (const v of views) {@@@ -191,13 +240,9 @@ function createQuotas(o) {};}-- /** Take `need` from a bucket: shared (Valkey) when it is up, this process otherwise. */-- async function spend(id, lim, need, t, peekedTokens) {-- if (shared && !shared.down) {-- if (await shared.spend(id, lim, need, t) != null) return;-- }-- const left = Math.max(0, peekedTokens - need);-- buckets.set(id, { tokens: left, at: t });++ /** Take `need` from this process's bucket, refilled to now (re-read: another run may have spent since). */++ function spendLocal(id, lim, need, t) {++ buckets.set(id, { tokens: Math.max(0, level(id, lim, t) - need), at: t });if (buckets.size > maxBuckets) buckets.delete(buckets.keys().next().value);}@@@ -26,6 +26,36 @@ const MAX_ABUSE = 10000;/** Seconds until just after the next UTC midnight (a day counter's TTL). */function dayTtl(t) { return Math.ceil((Math.floor(t / DAY_MS) + 1) * DAY_MS / 1000) - Math.floor(t / 1000) + 3600; }++/** The guard_day hash of one UTC day, and one allowance's field in it (quota.js charges the same ones). */++const dayKey = (valkey, day) => valkey.key('guard_day', day);++const dayField = (key, cls) => `${key}|${cls}`;++++// Check every allowance in one day hash and, when they all fit (or `force`, report mode), add to each:++// all or nothing, atomically. KEYS[1] the day hash. ARGV: ttl (s), commit (1: charge when it fits,++// 0: only look), force (1: charge even when it does not fit), then per allowance: field, n, limit++// (0 = none). Returns "<1|0>:<used before, comma-separated>".++const DAY_CHARGE = `++local ttl = tonumber(ARGV[1])++local commit = ARGV[2] == '1'++local force = ARGV[3] == '1'++local m = (#ARGV - 3) / 3++local ok = true++local used = {}++for i = 1, m do++ local b = 3 + (i - 1) * 3++ local u = tonumber(redis.call('HGET', KEYS[1], ARGV[b + 1]) or '0')++ used[i] = tostring(u)++ local lim = tonumber(ARGV[b + 3])++ if lim > 0 and u + tonumber(ARGV[b + 2]) > lim then ok = false end++end++if (ok and commit) or force then++ for i = 1, m do++ local b = 3 + (i - 1) * 3++ if tonumber(ARGV[b + 2]) > 0 then redis.call('HINCRBYFLOAT', KEYS[1], ARGV[b + 1], ARGV[b + 2]) end++ end++ redis.call('EXPIRE', KEYS[1], ttl)++end++return (ok and '1' or '0') .. ':' .. table.concat(used, ',')`;/*** @param {object} o@@@ -66,7 +96,7 @@ function createGuardStore(o = {}) {v = await valkey.client.get(key);}if (v) {-- salts.set(day, v);++ salts.set(day, v); // replaces the process-local stand-in: salt() answers the shared one from now onfor (const k of salts.keys()) if (k !== day) salts.delete(k);}} catch (err) {@@@ -78,29 +108,37 @@ function createGuardStore(o = {}) {async function warm() { if (valkey) await loadSalt(dayOf(now())); }// ── guard_day (Valkey) ───────────────────────────────────-- async function dayUsed(day, key, cls) {++ // One run charges one or more day allowances (a session and its address) all or nothing, and the++ // check and the increment are one Lua call: two requests can never both see room for the last unit.++ function dayCharge(day, entries, opt = {}) {if (cacheDay !== day) { days.clear(); cacheDay = day; }-- const k = `${day}|${key}|${cls}`;if (dayUp) {-- try {-- const v = await valkey.client.hget(valkey.key('guard_day', day), `${key}|${cls}`);-- const n = v == null ? 0 : Number(v);-- days.set(k, n);-- return n;-- } catch (err) { dayUp = false; log.warn(`[Guard] ${app}: Valkey day counters unavailable (${err.message}); counting in this process`); }++ return valkeyDayCharge(day, entries, opt).catch((err) => {++ dayUp = false;++ log.warn(`[Guard] ${app}: Valkey day counters unavailable (${err.message}); counting in this process`);++ return localDayCharge(day, entries, opt);++ });}-- return days.get(k) || 0;++ return Promise.resolve(localDayCharge(day, entries, opt));}-- async function dayAdd(day, key, cls, n) {-- if (!(n > 0)) return;-- const k = `${day}|${key}|${cls}`;-- days.set(k, (days.get(k) || 0) + n);-- if (!dayUp) return;-- try {-- const h = valkey.key('guard_day', day);-- await valkey.client.hincrbyfloat(h, `${key}|${cls}`, n);-- await valkey.client.expire(h, dayTtl(now()));-- } catch (err) { dayUp = false; log.warn(`[Guard] ${app}: Valkey day counters unavailable (${err.message}); counting in this process`); }++ /** In-process: synchronous, so nothing interleaves between the check and the increment. */++ function localDayCharge(day, entries, { commit = true, force = false } = {}) {++ const used = entries.map((e) => days.get(`${day}|${e.key}|${e.cls}`) || 0);++ const ok = entries.every((e, i) => !(e.limit > 0) || used[i] + e.n <= e.limit);++ if ((ok && commit) || force) {++ entries.forEach((e, i) => { if (e.n > 0) days.set(`${day}|${e.key}|${e.cls}`, used[i] + e.n); });++ }++ return { ok, used };++ }++ async function valkeyDayCharge(day, entries, { commit = true, force = false } = {}) {++ const c = valkey.client;++ if (typeof c.ovGuardDayCharge !== 'function') c.defineCommand('ovGuardDayCharge', { lua: DAY_CHARGE });++ const args = [String(dayTtl(now())), commit ? '1' : '0', force ? '1' : '0'];++ for (const e of entries) args.push(dayField(e.key, e.cls), String(e.n), String(e.limit > 0 ? e.limit : 0));++ const [ok, list] = String(await c.ovGuardDayCharge(1, dayKey(valkey, day), ...args)).split(':');++ const used = list ? list.split(',').map(Number) : [];++ entries.forEach((e, i) => days.set(`${day}|${e.key}|${e.cls}`, used[i] + ((ok === '1' && commit) || force ? e.n : 0)));++ return { ok: ok === '1', used };}// ── guard_abuse (PostgreSQL) ─────────────────────────────@@@ -123,7 +161,7 @@ function createGuardStore(o = {}) {async function pruneAbuse(beforeMs) {for (let i = abuse.length - 1; i >= 0; i--) if (abuse[i].at < beforeMs) abuse.splice(i, 1);if (!db) return 0;-- const n = await db.exec('DELETE FROM guard_abuse WHERE at < $1', [beforeMs]);++ const n = await db.exec('DELETE FROM guard_abuse WHERE app = $1 AND at < $2', [app, beforeMs]); // this app's rows only: the table is sharedreturn n || 0;}// guard_day rows live in Valkey and expire on their own.@@@ -131,7 +169,7 @@ function createGuardStore(o = {}) {async function close() { /* the app owns db and valkey */ }-- return { kind: db ? 'postgres' : 'memory', app, db, valkey, salt, warm, dayUsed, dayAdd, logAbuse, abuseRows, pruneAbuse, pruneDays, close };++ return { kind: db ? 'postgres' : 'memory', app, db, valkey, salt, warm, dayCharge, logAbuse, abuseRows, pruneAbuse, pruneDays, close };}--module.exports = { createGuardStore, dayOf, DAY_MS };++module.exports = { createGuardStore, dayOf, dayTtl, dayKey, dayField, DAY_MS };nnew file mode 100644@@@ -0,0 +1,83 @@++'use strict';++// The guard's Lua on a real Valkey (npm run test:pg sets VALKEY_URL; guard.test.js runs the same rules++// on an in-process emulation): quota.js CHARGE decides and charges every bucket and day allowance of a++// run at once, so a run refused on the day leg spends no bucket token and concurrent runs never++// overshoot the day; store.js DAY_CHARGE is the same all-or-nothing check-and-increment for the day++// counters alone. Keys live under a random prefix and are deleted at the end.++const assert = require('assert');++const crypto = require('crypto');++const { dep } = require('./deps');++const { createQuotas } = require('../guard/quota');++const { createGuardStore } = require('../guard/store');++++const quiet = { log() {}, warn() {}, error() {} };++++(async () => {++ if (!process.env.VALKEY_URL) {++ console.log('guard valkey: skipped (no VALKEY_URL: npm run test:pg runs the Lua on a real Valkey)');++ return;++ }++ const prefix = `ov:tools-test:${crypto.randomBytes(6).toString('hex')}:`;++ const valkey = dep('openvibe-sdk/valkey').createValkey({ url: process.env.VALKEY_URL, prefix, log: quiet });++ try {++ const table = {++ fetch: { anonymous: { perMinute: 1, burst: 5, perDay: 3 } },++ wide: { anonymous: { perMinute: 600, burst: 600, perDay: 3 } },++ session: { session: { perMinute: 60, burst: 10, perDay: 100 }, anonymous: { perMinute: 60, burst: 10, perDay: 100 } },++ };++ let clock = Date.UTC(2026, 8, 23, 23, 59, 58);++ const store = createGuardStore({ valkey, app: 'test', now: () => clock, log: quiet });++ const q = createQuotas({ store, valkey, quotas: () => table, now: () => clock, log: quiet });++ const anon = (ip) => ({ tier: 'anonymous', ipKey: `ip:${ip}`, key: `ip:${ip}` });++++ // A day refusal spends nothing: 2 runs fit, cost-2 runs are refused on the day leg (2 + 2 > 3)++ // though the bucket holds 3, and the next day a cost-3 run still finds those 3 tokens.++ const c = anon('a');++ for (let i = 0; i < 2; i++) assert.ok((await q.check(c, { quotaClass: 'fetch', cost: 1 })).ok);++ for (let i = 0; i < 5; i++) assert.strictEqual((await q.check(c, { quotaClass: 'fetch', cost: 2 })).binding, 'day');++ clock += 2000;++ assert.strictEqual((await q.check(c, { quotaClass: 'fetch', cost: 3 })).ok, true, 'the refused runs took no token');++ const short = await q.check(c, { quotaClass: 'fetch', cost: 1 });++ assert.strictEqual(short.ok, false, 'and now the bucket is empty');++++ // Report mode charges a refused run (the work happens anyway).++ const rep = anon('r');++ for (let i = 0; i < 3; i++) assert.ok((await q.check(rep, { quotaClass: 'wide', cost: 1 })).ok);++ assert.strictEqual((await q.check(rep, { quotaClass: 'wide', cost: 1 }, { report: true })).ok, false);++ const used = await store.dayCharge(new Date(clock).toISOString().slice(0, 10), [{ key: 'ip:r', cls: 'wide', n: 0, limit: 0 }], { commit: false });++ assert.deepStrictEqual(used.used, [4], 'report mode counted the refused run');++++ // Concurrency: 30 runs at once against a day allowance of 3.++ const all = await Promise.all(Array.from({ length: 30 }, () => q.check(anon('c'), { quotaClass: 'wide', cost: 1 })));++ assert.strictEqual(all.filter((r) => r.ok).length, 3, 'exactly 3 of 30 concurrent runs');++++ // A session run charges the session and its address together, all or nothing.++ const s1 = { tier: 'session', key: 'session:1', ipKey: 'ip:s' };++ assert.ok((await q.check(s1, { quotaClass: 'session', cost: 10 })).ok);++ const s2 = { tier: 'session', key: 'session:2', ipKey: 'ip:s' };++ const r = await q.check(s2, { quotaClass: 'session', cost: 10 });++ assert.ok(r.ok || r.scope === 'address', 'the address share binds, not the fresh session');++++ // DAY_CHARGE alone (quotas without Valkey): atomic and all or nothing.++ const day = '2026-09-24';++ const direct = await Promise.all(Array.from({ length: 12 }, () => store.dayCharge(day, [{ key: 'k', cls: 'c', n: 1, limit: 4 }])));++ assert.strictEqual(direct.filter((x) => x.ok).length, 4, 'dayCharge: 4 of 12');++ const both = await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 1, limit: 5 }, { key: 'k', cls: 'c', n: 1, limit: 4 }]);++ assert.strictEqual(both.ok, false);++ assert.deepStrictEqual((await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 0, limit: 0 }], { commit: false })).used, [0], 'k2 untouched');++ const forced = await store.dayCharge(day, [{ key: 'k', cls: 'c', n: 1, limit: 4 }], { force: true });++ assert.deepStrictEqual([forced.ok, forced.used], [false, [4]], 'force (report mode) charges past the limit');++++ // The salt: one per day, shared.++ await store.warm();++ const other = createGuardStore({ valkey, app: 'other', now: () => clock, log: quiet });++ await other.warm();++ assert.strictEqual(other.salt(), store.salt(), 'one salt in Valkey');++++ console.log('guard valkey: CHARGE (a refused run spends nothing, report mode counts, 3 of 30 concurrent, session + address), DAY_CHARGE (atomic, all or nothing, force), shared salt: all checks passed');++ } finally {++ const keys = await valkey.client.keys(`${prefix}*`);++ if (keys.length) await valkey.client.del(...keys);++ await valkey.close();++ }++})().catch((err) => { console.error(err); process.exit(1); });@@@ -44,8 +44,8 @@ const tmp = () => fs.mkdtempSync(path.join(os.tmpdir(), 'ov-guard-'));const quiet = { log() {}, warn() {}, error() {} };// A stand-in for openvibe-sdk/valkey: the same key()/client shape over one Map, so two guards can--// share "one Valkey" in-process. The bucket's two Lua commands are emulated in JS (the real ones are--// atomic; a single-threaded test has no race to lose).++// share "one Valkey" in-process. The guard's Lua commands are emulated in JS (the real ones are atomic,++// and guard-valkey.test.js runs them on a real Valkey under npm run test:pg).function fakeValkey({ down = false } = {}) {const store = new Map();const guard = () => { if (down) throw new Error('valkey down'); };@@@ -57,22 +57,56 @@ function fakeValkey({ down = false } = {}) {async expire() { guard(); return 1; },async incr(k) { guard(); const v = Number(store.get(k) || 0) + 1; store.set(k, String(v)); return v; },defineCommand(name) {-- client[name] = async (_numKeys, key, burst, rate, need, now) => {-- guard();-- const h = store.get(key) || {};-- let tokens = h.tokens != null ? Number(h.tokens) : Number(burst);-- const at = h.at != null ? Number(h.at) : Number(now);-- tokens = Math.min(Number(burst), tokens + (Number(now) - at) * Number(rate));-- if (name === 'ovBucketRefund') { // give back, capped at burst-- const left = Math.min(Number(burst), tokens + Number(need));++ const num = (x) => Number(x);++ const refill = (h, burst, rate, now) => {++ const tokens = h.tokens != null ? num(h.tokens) : burst;++ const at = h.at != null ? num(h.at) : now;++ return Math.min(burst, tokens + Math.max(0, now - at) * rate);++ };++ if (name === 'ovBucketRefund') { // give back, capped at burst++ client[name] = async (_numKeys, key, burst, rate, need, now) => {++ guard();++ const left = Math.min(num(burst), refill(store.get(key) || {}, num(burst), num(rate), num(now)) + num(need));store.set(key, { tokens: String(left), at: String(now) });return String(left);-- }-- if (tokens < Number(need)) return `${tokens}:0`; // short: take nothing-- const left = tokens - Number(need);-- store.set(key, { tokens: String(left), at: String(now) });-- return `${left}:1`;-- };++ };++ } else if (name === 'ovQuotaCharge') { // quota.js CHARGE: every bucket and day allowance, all or nothing++ client[name] = async (numKeys, ...rest) => {++ guard();++ const keys = rest.slice(0, numKeys), a = rest.slice(numKeys);++ const n = num(a[0]), now = num(a[1]), report = a[4] === '1';++ const dh = store.get(keys[n]) || {};++ let ok = true;++ const st = [];++ for (let i = 0; i < n; i++) {++ const [burst, rate, need, perDay, cost, field] = a.slice(5 + i * 6, 11 + i * 6);++ const tokens = refill(store.get(keys[i]) || {}, num(burst), num(rate), now);++ const used = num(perDay) > 0 ? num(dh[field] || 0) : 0;++ const minuteOk = tokens >= num(need), dayOk = !(num(perDay) > 0) || used + num(cost) <= num(perDay);++ if (!(minuteOk && dayOk)) ok = false;++ st.push({ tokens, used, need: num(need), perDay: num(perDay), cost: num(cost), field, minuteOk, dayOk });++ }++ if (ok || report) {++ st.forEach((c, i) => {++ store.set(keys[i], { tokens: String(Math.max(0, c.tokens - c.need)), at: String(now) });++ if (c.perDay > 0) dh[c.field] = String(num(dh[c.field] || 0) + c.cost);++ });++ store.set(keys[n], dh);++ }++ return st.map((c) => `${c.tokens},${c.used},${c.minuteOk ? 1 : 0},${c.dayOk ? 1 : 0}`).join(';');++ };++ } else if (name === 'ovGuardDayCharge') { // store.js DAY_CHARGE++ client[name] = async (_numKeys, key, _ttl, commit, force, ...triples) => {++ guard();++ const h = store.get(key) || {};++ const e = [];++ for (let i = 0; i < triples.length; i += 3) e.push({ field: triples[i], n: num(triples[i + 1]), limit: num(triples[i + 2]) });++ const used = e.map((x) => num(h[x.field] || 0));++ const ok = e.every((x, i) => !(x.limit > 0) || used[i] + x.n <= x.limit);++ if ((ok && commit === '1') || force === '1') { e.forEach((x, i) => { if (x.n > 0) h[x.field] = String(used[i] + x.n); }); store.set(key, h); }++ return `${ok ? 1 : 0}:${used.join(',')}`;++ };++ } else throw new Error(`fakeValkey: no emulation of ${name}`);},};return { client, key: (...p) => p.join(':'), ready: async () => (down ? { ok: false, error: 'down' } : { ok: true }), close: async () => {} };@@@ -316,6 +350,106 @@ async function serve(app) {g.close();}++ // ── A refused run takes nothing: a day refusal never drains the minute bucket (shared and local) ──++ // Just before UTC midnight: 2 runs fit the day (the bucket 5 → 3), ten more are refused on the day++ // leg. Two seconds later a new day starts with the bucket still at 3 (+ a trickle): 3 runs fit, the++ // 4th is refused on the minute leg. If the refused runs had spent tokens, the new day would start empty.++ for (const withValkey of [true, false]) {++ let clock = Date.UTC(2026, 8, 23, 23, 59, 58);++ const env = { TOOLS_GUARD: 'enforce', TOOLS_GUARD_LIMITS: JSON.stringify({ 'tools-fetch': { anonymous: { perMinute: 1, burst: 5, perDay: 2 } } }) };++ const g = makeGuard({ valkey: withValkey ? fakeValkey() : undefined, env, now: () => clock });++ await g.store.warm();++ const c = g.resolveCaller({ ip: '198.51.100.40', headers: {}, cookies: {} });++ const run = () => g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 1 });++ const label = withValkey ? 'shared' : 'local';++ assert.ok((await run()).ok && (await run()).ok, `${label}: the day allowance of 2`);++ for (let i = 0; i < 10; i++) assert.strictEqual((await run()).binding, 'day', `${label}: refused on the day leg`);++ clock += 2000;++ const next = [];++ for (let i = 0; i < 4; i++) next.push(await run());++ assert.deepStrictEqual(next.map(r => r.ok), [true, true, false, false], `${label}: a new day: the bucket still holds 3, the day allowance is 2 again`);++ assert.strictEqual(next[2].binding, 'day');++ g.close();++ }++ // The same with a bigger refused run: cost 2 fits the bucket (3 left) but not the day (2 + 2 > 3). Had++ // the refused runs spent, the next day's cost-3 run would find about 1 token instead of 3.++ for (const withValkey of [true, false]) {++ let clock = Date.UTC(2026, 8, 23, 23, 59, 58);++ const env = { TOOLS_GUARD: 'enforce', TOOLS_GUARD_LIMITS: JSON.stringify({ 'tools-fetch': { anonymous: { perMinute: 1, burst: 5, perDay: 3 } } }) };++ const g = makeGuard({ valkey: withValkey ? fakeValkey() : undefined, env, now: () => clock });++ await g.store.warm();++ const c = g.resolveCaller({ ip: '198.51.100.41', headers: {}, cookies: {} });++ const label = withValkey ? 'shared' : 'local';++ for (let i = 0; i < 2; i++) assert.ok((await g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 1 })).ok);++ for (let i = 0; i < 5; i++) assert.strictEqual((await g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 2 })).binding, 'day');++ clock += 2000;++ const r = await g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 3 });++ assert.strictEqual(r.ok, true, `${label}: the 3 tokens the refused runs never took`);++ g.close();++ }++++ // ── The day allowance is one atomic check and charge: concurrent runs never overshoot it ──++ {++ const env = { TOOLS_GUARD: 'enforce', TOOLS_GUARD_LIMITS: JSON.stringify({ 'tools-fetch': { anonymous: { perMinute: 600, burst: 600, perDay: 3 } } }) };++ // Two processes, one Valkey, twenty runs at once: exactly 3 fit.++ const valkey = fakeValkey();++ const a = makeGuard({ valkey, env }), b = makeGuard({ valkey, env });++ await a.store.warm(); await b.store.warm();++ const req = { ip: '198.51.100.50', headers: {}, cookies: {} };++ const ca = a.resolveCaller(req), cb = b.resolveCaller(req);++ const all = await Promise.all(Array.from({ length: 20 }, (_, i) => (i % 2 ? a : b).quotas.check(i % 2 ? ca : cb, { quotaClass: 'tools-fetch', cost: 1 })));++ assert.strictEqual(all.filter(r => r.ok).length, 3, 'shared buckets: 3 of 20 concurrent runs');++ a.close(); b.close();++ // The store's own charge (quotas without Valkey, day counters in Valkey): the same.++ const v2 = fakeValkey();++ const store = createGuardStore({ valkey: v2, log: quiet });++ const q = createQuotas({ store, quotas: () => JSON.parse(env.TOOLS_GUARD_LIMITS), log: quiet });++ const caller = { tier: 'anonymous', ipKey: 'ip:concurrent', key: 'ip:concurrent' };++ const r2 = await Promise.all(Array.from({ length: 20 }, () => q.check(caller, { quotaClass: 'tools-fetch', cost: 1 })));++ assert.strictEqual(r2.filter(r => r.ok).length, 3, 'store day charge: 3 of 20 concurrent runs');++ const day = new Date().toISOString().slice(0, 10);++ const direct = await Promise.all(Array.from({ length: 10 }, () => store.dayCharge(day, [{ key: 'k', cls: 'c', n: 1, limit: 4 }])));++ assert.strictEqual(direct.filter(r => r.ok).length, 4, 'dayCharge: 4 of 10');++ const both = await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 1, limit: 5 }, { key: 'k', cls: 'c', n: 1, limit: 4 }]);++ assert.strictEqual(both.ok, false, 'all or nothing: k is full');++ assert.deepStrictEqual((await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 0, limit: 5 }], { commit: false })).used, [0], 'so k2 was not charged');++ }++++ // ── A salt handed out before Valkey answered is replaced by the shared one as soon as it loads ──++ {++ const valkey = fakeValkey();++ let release;++ const gate = new Promise(r => { release = r; });++ const get = valkey.client.get;++ valkey.client.get = async (k) => { await gate; return get(k); };++ const req = { ip: '198.51.100.70', headers: {}, cookies: {} };++ const a = makeGuard({ valkey });++ const early = a.resolveCaller(req).ipKey; // Valkey has not answered: a process-local salt++ release();++ await a.store.warm();++ const b = makeGuard({ valkey });++ await b.store.warm();++ const later = a.resolveCaller(req).ipKey;++ assert.notStrictEqual(later, early, 'the process-local stand-in is gone once the shared salt loads');++ assert.strictEqual(later, b.resolveCaller(req).ipKey, 'and both processes agree on the key');++ a.close(); b.close();++ }++++ // ── One app's prune leaves the other apps' abuse rows alone (one shared guard_abuse table) ──++ {++ const dir = tmp();++ const db = await testDb(dir);++ const old = now0 - 40 * 24 * 3600 * 1000;++ for (const app of ['img', 'docs']) {++ await db.query('INSERT INTO guard_abuse (app, at, minute, ip_hash, principal, tool, reason, enforced) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)',++ [app, old, Math.floor(old / 60_000), 'h', '', 'png', 'quota', 1]);++ }++ const img = createGuardStore({ db, app: 'img', log: quiet });++ assert.strictEqual(await img.pruneAbuse(now0), 1, "img prunes its own row");++ const left = await db.query('SELECT app FROM guard_abuse ORDER BY app');++ assert.deepStrictEqual(left.map(r => r.app), ['docs'], "docs' row is untouched");++ }++// ── The abuse log: no raw address anywhere, 30 days ──{let clock = now0;@@@ -569,5 +703,5 @@ async function serve(app) {}await closeAllTestDbs();-- console.log('guard: callers (/64, one loopback hop, services by token, aud-checked users, sessions, sandbox), bucket math + cost, session sharing, report vs enforce, headers + problem+json, two processes one Valkey quota + outage fallback, sniffing, target throttle, port cap, abuse log + prune + metric, semaphore, job bounds (queue, disk, per address), challenge hook, Origin check (CSRF): all checks passed');++ console.log('guard: callers (/64, one loopback hop, services by token, aud-checked users, sessions, sandbox), bucket math + cost, session sharing, report vs enforce, headers + problem+json, two processes one Valkey quota + outage fallback, refused runs take nothing, atomic day charge, salt refresh, per-app prune, sniffing, target throttle, port cap, abuse log + prune + metric, semaphore, job bounds (queue, disk, per address), challenge hook, Origin check (CSRF): all checks passed');})().catch((err) => { console.error(err); process.exit(1); });
@@ -106,12 +106,9 @@ function createGuard(o = {}) {
@@ -106,12 +106,9 @@ function createGuard(o = {}) {
const store = o.store || createGuardStore({ db: o.db || null, valkey: o.valkey || null, app: appName, now, log });
const store = o.store || createGuardStore({ db: o.db || null, valkey: o.valkey || null, app: appName, now, log });
if (!o.store) void store.warm();
if (!o.store) void store.warm();
- let saltDay = null, saltValue = null;
- const salt = () => {
- const d = dayOf(now());
- if (d !== saltDay) { saltValue = store.salt(d); saltDay = d; }
- return saltValue;
- };
+ // Asked of the store every time (a Map lookup), never cached here: a salt handed out before Valkey
+ // answered is a process-local stand-in, and the shared one must replace it as soon as it loads.
+ const salt = () => store.salt(dayOf(now()));
// Network's sign-out-everywhere cutoffs (PostgreSQL, read through a short shared cache): the gateway
// Network's sign-out-everywhere cutoffs (PostgreSQL, read through a short shared cache): the gateway
// writes them (apps/gateway/server/revocation-events.js); every app reads them here.
// writes them (apps/gateway/server/revocation-events.js); every app reads them here.
@@ -3,10 +3,12 @@
@@ -3,10 +3,12 @@
// Tiered quotas: a token bucket per (key, class) and a UTC-day allowance per (key, class). A run takes
// Tiered quotas: a token bucket per (key, class) and a UTC-day allowance per (key, class). A run takes
// its descriptor's `cost` from both, all or nothing.
// its descriptor's `cost` from both, all or nothing.
//
//
-// The day allowance is shared across processes through the guard store (Valkey, plan T8 decision 4) and
-// so are the minute/burst buckets when a Valkey connection is passed: one Lua call refills and spends
-// one bucket atomically, so two hosts enforce one quota. Without Valkey (or on an outage) both legs use
-// this process's own counters — a throttle that never fails open and never fails every request.
+// The day allowance and the minute/burst buckets are shared across processes in Valkey (plan T8
+// decision 4) when a connection is passed: ONE Lua call refills every bucket a run touches, checks
+// every bucket and every day allowance, and charges them all only when all of them fit — so two hosts
+// enforce one quota, and a refused run (say, on the day leg) never spends a bucket token. Without Valkey
+// (or on an outage) the buckets are this process's and the day allowances go through the guard store's
+// own atomic check-and-charge — a throttle that never fails open and never fails every request.
//
//
// Who is charged (keys come from the caller resolver, addresses already hashed):
// Who is charged (keys come from the caller resolver, addresses already hashed):
// anonymous its address (IPv6: its /64) at the anonymous tier
// anonymous its address (IPv6: its /64) at the anonymous tier
@@ -21,27 +23,56 @@
@@ -21,27 +23,56 @@
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
const { SESSION_IP_SHARE } = require('./limits');
const { SESSION_IP_SHARE } = require('./limits');
-const { dayOf, DAY_MS } = require('./store');
+const { dayOf, dayTtl, dayKey, dayField, DAY_MS } = require('./store');
-// Refill a bucket (tokens + elapsed × rate, capped at burst) and spend `need`, atomically. It only
-// consumes when the bucket holds `need`: a refused run changes nothing. Returns "tokens:ok".
-// ARGV: burst, rate (tokens/ms), need, now (ms), ttl (s).
-const SPEND = `
-local key = KEYS[1]
-local burst = tonumber(ARGV[1])
-local rate = tonumber(ARGV[2])
-local need = tonumber(ARGV[3])
-local now = tonumber(ARGV[4])
-local ttl = tonumber(ARGV[5])
-local d = redis.call('HMGET', key, 'tokens', 'at')
-local tokens = (d[1] and tonumber(d[1])) or burst
-local at = (d[2] and tonumber(d[2])) or now
-tokens = math.min(burst, tokens + (now - at) * rate)
-if tokens < need then return tostring(tokens)..':0' end
-local left = tokens - need
-redis.call('HSET', key, 'tokens', tostring(left), 'at', tostring(now))
-redis.call('EXPIRE', key, ttl)
-return tostring(left)..':1'`;
+// The whole quota decision for one run, atomically: refill each bucket (tokens + elapsed × rate, capped
+// at burst), read each day allowance, and only when every bucket holds its `need` and every allowance
+// has room for `cost` — or in report mode (the work happens anyway) — take them all. A refused run
+// changes nothing. KEYS[1..n] the buckets, KEYS[n+1] the day hash (guard_day, shared with the store).
+// ARGV: n, now (ms), bucket ttl (s), day ttl (s), report (1/0), then per charge: burst, rate
+// (tokens/ms), need, perDay (0 = none), cost, day field. Returns per charge "tokens,used,minuteOk,dayOk"
+// (tokens and used before this run), joined with ";".
+const CHARGE = `
+local n = tonumber(ARGV[1])
+local now = tonumber(ARGV[2])
+local bttl = tonumber(ARGV[3])
+local dttl = tonumber(ARGV[4])
+local report = ARGV[5] == '1'
+local day = KEYS[n + 1]
+local ok = true
+local st = {}
+local out = {}
+for i = 1, n do
+ local b = 5 + (i - 1) * 6
+ local burst = tonumber(ARGV[b + 1])
+ local rate = tonumber(ARGV[b + 2])
+ local need = tonumber(ARGV[b + 3])
+ local perDay = tonumber(ARGV[b + 4])
+ local cost = tonumber(ARGV[b + 5])
+ local field = ARGV[b + 6]
+ local d = redis.call('HMGET', KEYS[i], 'tokens', 'at')
+ local tokens = (d[1] and tonumber(d[1])) or burst
+ local at = (d[2] and tonumber(d[2])) or now
+ tokens = math.min(burst, tokens + math.max(0, now - at) * rate)
+ local used = 0
+ if perDay > 0 then used = tonumber(redis.call('HGET', day, field) or '0') end
+ local minuteOk = tokens >= need
+ local dayOk = perDay <= 0 or used + cost <= perDay
+ if not (minuteOk and dayOk) then ok = false end
+ st[i] = { tokens, need, perDay, cost, field }
+ out[i] = tostring(tokens) .. ',' .. tostring(used) .. ',' .. (minuteOk and '1' or '0') .. ',' .. (dayOk and '1' or '0')
+end
+if ok or report then
+ local days = false
+ for i = 1, n do
+ local c = st[i]
+ redis.call('HSET', KEYS[i], 'tokens', tostring(math.max(0, c[1] - c[2])), 'at', tostring(now))
+ redis.call('EXPIRE', KEYS[i], bttl)
+ if c[3] > 0 then redis.call('HINCRBYFLOAT', day, c[5], c[4]); days = true end
+ end
+ if days then redis.call('EXPIRE', day, dttl) end
+end
+return table.concat(out, ';')`;
// Refill and give `need` back (a run that cost less than the up-front charge), capped at burst.
// Refill and give `need` back (a run that cost less than the up-front charge), capped at burst.
const REFUND = `
const REFUND = `
local key = KEYS[1]
local key = KEYS[1]
@@ -61,17 +92,27 @@ return tostring(left)`;
@@ -61,17 +92,27 @@ return tostring(left)`;
function createValkeyBuckets(valkey, log) {
function createValkeyBuckets(valkey, log) {
const c = valkey.client;
const c = valkey.client;
- if (typeof c.ovBucketSpend !== 'function') c.defineCommand('ovBucketSpend', { lua: SPEND });
+ if (typeof c.ovQuotaCharge !== 'function') c.defineCommand('ovQuotaCharge', { lua: CHARGE });
if (typeof c.ovBucketRefund !== 'function') c.defineCommand('ovBucketRefund', { lua: REFUND });
if (typeof c.ovBucketRefund !== 'function') c.defineCommand('ovBucketRefund', { lua: REFUND });
let up = true;
let up = true;
return {
return {
get down() { return !up; },
get down() { return !up; },
- /** { left, ok } — ok false means the bucket was short and nothing was taken. null: Valkey is down. */
- async spend(id, lim, need, t) {
+ /**
+ * Decide and charge a run in one call. charges: [{ id, lim, need, cost, field }].
+ * → [{ tokens, used, minuteOk, dayOk }] (before this run); null: Valkey is down.
+ */
+ async charge(charges, day, t, report) {
try {
try {
- const r = String(await c.ovBucketSpend(1, valkey.key('bucket', id), String(lim.burst), String(lim.perMinute / 60_000), String(need), String(t), String(600)));
- const [left, ok] = r.split(':');
- return { left: Number(left), ok: ok === '1' };
+ const args = [String(charges.length), String(t), '600', String(dayTtl(t)), report ? '1' : '0'];
+ for (const x of charges) {
+ args.push(String(x.lim.burst), String(x.lim.perMinute / 60_000), String(x.need), String(x.lim.perDay > 0 ? x.lim.perDay : 0), String(x.cost), x.field);
+ }
+ const keys = charges.map((x) => valkey.key('bucket', x.id)).concat(dayKey(valkey, day));
+ const r = String(await c.ovQuotaCharge(keys.length, ...keys, ...args));
+ return r.split(';').map((row) => {
+ const [tokens, used, m, d] = row.split(',');
+ return { tokens: Number(tokens), used: Number(used), minuteOk: m === '1', dayOk: d === '1' };
+ });
} catch (err) { up = false; log.warn(`[Quota] Valkey buckets unavailable (${err.message}); counting in this process`); return null; }
} catch (err) { up = false; log.warn(`[Quota] Valkey buckets unavailable (${err.message}); counting in this process`); return null; }
},
},
async refund(id, lim, need, t) {
async refund(id, lim, need, t) {
@@ -82,7 +123,7 @@ function createValkeyBuckets(valkey, log) {
@@ -82,7 +123,7 @@ function createValkeyBuckets(valkey, log) {
/**
/**
* @param {object} o
* @param {object} o
- * @param {object} o.store guard store (async dayUsed, dayAdd)
+ * @param {object} o.store guard store (async dayCharge: the day allowances without Valkey)
* @param {() => object} o.quotas the QUOTAS table (limits.quotas())
* @param {() => object} o.quotas the QUOTAS table (limits.quotas())
* @param {() => number} [o.now]
* @param {() => number} [o.now]
* @param {number} [o.maxBuckets]
* @param {number} [o.maxBuckets]
@@ -132,34 +173,43 @@ function createQuotas(o) {
@@ -132,34 +173,43 @@ function createQuotas(o) {
const t = now();
const t = now();
sweep(t);
sweep(t);
const table = o.quotas();
const table = o.quotas();
- const charges = chargesFor(caller, quotaClass, table);
+ const charges = chargesFor(caller, quotaClass, table)
+ .filter((c) => c.lim.burst > 0 && c.lim.perMinute > 0)
+ .map((c) => ({
+ ...c, cost, id: `${quotaClass}|${c.key}`, field: dayField(c.key, quotaClass),
+ need: Math.min(cost, c.lim.burst), // one run never costs more than a full bucket
+ }));
if (!charges.length) return { ok: true, limit: null, remaining: null, reset: null, retryAfter: 0 };
if (!charges.length) return { ok: true, limit: null, remaining: null, reset: null, retryAfter: 0 };
const day = dayOf(t);
const day = dayOf(t);
const nextDay = Math.ceil((Math.floor(t / DAY_MS) + 1) * DAY_MS / 1000) - Math.floor(t / 1000);
const nextDay = Math.ceil((Math.floor(t / DAY_MS) + 1) * DAY_MS / 1000) - Math.floor(t / 1000);
+
+ // Shared (Valkey): the whole decision and the charge in one atomic call. Otherwise (no Valkey, or
+ // an outage) this process's buckets, and the store's atomic day charge — committed only when the
+ // buckets fit too, so a refused run takes nothing from either leg.
+ let states = shared && !shared.down ? await shared.charge(charges, day, t, !!opt.report) : null;
+ if (!states) {
+ const levels = charges.map((c) => level(c.id, c.lim, t));
+ const minuteOk = charges.map((c, i) => levels[i] >= c.need);
+ const allMinute = minuteOk.every(Boolean);
+ const withDay = charges.filter((c) => c.lim.perDay > 0);
+ const d = withDay.length
+ ? await o.store.dayCharge(day, withDay.map((c) => ({ key: c.key, cls: quotaClass, n: cost, limit: c.lim.perDay })), { commit: allMinute, force: !!opt.report })
+ : { ok: true, used: [] };
+ const usedOf = new Map(withDay.map((c, i) => [c, d.used[i] || 0]));
+ states = charges.map((c, i) => {
+ const used = usedOf.get(c) || 0;
+ return { tokens: levels[i], used, minuteOk: minuteOk[i], dayOk: !(c.lim.perDay > 0) || used + cost <= c.lim.perDay };
+ });
+ if ((allMinute && d.ok) || opt.report) {
+ for (const c of charges) spendLocal(c.id, c.lim, c.need, t);
+ }
+ }
+
let ok = true, retryAfter = 0, binding = null, scope = null;
let ok = true, retryAfter = 0, binding = null, scope = null;
- const views = [];
- for (const c of charges) {
- const { lim } = c;
- if (!(lim.burst > 0) || !(lim.perMinute > 0)) continue;
- const need = Math.min(cost, lim.burst); // one run never costs more than a full bucket
- const id = `${quotaClass}|${c.key}`;
+ const views = charges.map((c, i) => {
+ const { lim, need } = c;
+ const { tokens, used, minuteOk, dayOk } = states[i];
const rate = lim.perMinute / 60_000;
const rate = lim.perMinute / 60_000;
- // Shared (Valkey): refill and spend in one atomic call, and decide from its answer. Otherwise
- // (no Valkey, or an outage) decide and spend from this process's own map below.
- let tokens, minuteOk, done = false;
- if (shared && !shared.down) {
- const r = await shared.spend(id, lim, need, t);
- if (r) { tokens = r.ok ? r.left + need : r.left; minuteOk = r.ok; done = true; }
- }
- if (!done) { tokens = level(id, lim, t); minuteOk = tokens >= need; }
- const used = lim.perDay > 0 ? await o.store.dayUsed(day, c.key, quotaClass) : 0;
- const dayOk = !(lim.perDay > 0) || used + cost <= lim.perDay;
- const after = Math.max(0, tokens - need);
- views.push({
- c, need, tokens, rate, dayUsed: used, done,
- minute: { limit: lim.burst, remaining: Math.floor(after), reset: Math.ceil((lim.burst - after) / rate / 1000) },
- day: lim.perDay > 0 ? { limit: lim.perDay, remaining: Math.max(0, Math.floor(lim.perDay - used - cost)), reset: nextDay } : null,
- });
if (!minuteOk) {
if (!minuteOk) {
ok = false;
ok = false;
const wait = Math.ceil((need - tokens) / rate / 1000);
const wait = Math.ceil((need - tokens) / rate / 1000);
@@ -169,13 +219,12 @@ function createQuotas(o) {
@@ -169,13 +219,12 @@ function createQuotas(o) {
ok = false;
ok = false;
if (nextDay > retryAfter) { retryAfter = nextDay; binding = 'day'; scope = c.scope; }
if (nextDay > retryAfter) { retryAfter = nextDay; binding = 'day'; scope = c.scope; }
}
}
- }
- if (ok || opt.report) {
- for (const v of views) {
- if (!v.done) await spend(`${quotaClass}|${v.c.key}`, v.c.lim, v.need, t, v.tokens);
- if (v.c.lim.perDay > 0) await o.store.dayAdd(day, v.c.key, quotaClass, cost);
- }
- }
+ const after = Math.max(0, tokens - need);
+ return {
+ minute: { limit: lim.burst, remaining: Math.floor(after), reset: Math.ceil((lim.burst - after) / rate / 1000) },
+ day: lim.perDay > 0 ? { limit: lim.perDay, remaining: Math.max(0, Math.floor(lim.perDay - used - cost)), reset: nextDay } : null,
+ };
+ });
// The headers describe the allowance closest to running out, in units of this run's cost.
// The headers describe the allowance closest to running out, in units of this run's cost.
let head = null;
let head = null;
for (const v of views) {
for (const v of views) {
@@ -191,13 +240,9 @@ function createQuotas(o) {
@@ -191,13 +240,9 @@ function createQuotas(o) {
};
};
}
}
- /** Take `need` from a bucket: shared (Valkey) when it is up, this process otherwise. */
- async function spend(id, lim, need, t, peekedTokens) {
- if (shared && !shared.down) {
- if (await shared.spend(id, lim, need, t) != null) return;
- }
- const left = Math.max(0, peekedTokens - need);
- buckets.set(id, { tokens: left, at: t });
+ /** Take `need` from this process's bucket, refilled to now (re-read: another run may have spent since). */
+ function spendLocal(id, lim, need, t) {
+ buckets.set(id, { tokens: Math.max(0, level(id, lim, t) - need), at: t });
if (buckets.size > maxBuckets) buckets.delete(buckets.keys().next().value);
if (buckets.size > maxBuckets) buckets.delete(buckets.keys().next().value);
}
}
@@ -26,6 +26,36 @@ const MAX_ABUSE = 10000;
@@ -26,6 +26,36 @@ const MAX_ABUSE = 10000;
/** Seconds until just after the next UTC midnight (a day counter's TTL). */
/** Seconds until just after the next UTC midnight (a day counter's TTL). */
function dayTtl(t) { return Math.ceil((Math.floor(t / DAY_MS) + 1) * DAY_MS / 1000) - Math.floor(t / 1000) + 3600; }
function dayTtl(t) { return Math.ceil((Math.floor(t / DAY_MS) + 1) * DAY_MS / 1000) - Math.floor(t / 1000) + 3600; }
+/** The guard_day hash of one UTC day, and one allowance's field in it (quota.js charges the same ones). */
+const dayKey = (valkey, day) => valkey.key('guard_day', day);
+const dayField = (key, cls) => `${key}|${cls}`;
+
+// Check every allowance in one day hash and, when they all fit (or `force`, report mode), add to each:
+// all or nothing, atomically. KEYS[1] the day hash. ARGV: ttl (s), commit (1: charge when it fits,
+// 0: only look), force (1: charge even when it does not fit), then per allowance: field, n, limit
+// (0 = none). Returns "<1|0>:<used before, comma-separated>".
+const DAY_CHARGE = `
+local ttl = tonumber(ARGV[1])
+local commit = ARGV[2] == '1'
+local force = ARGV[3] == '1'
+local m = (#ARGV - 3) / 3
+local ok = true
+local used = {}
+for i = 1, m do
+ local b = 3 + (i - 1) * 3
+ local u = tonumber(redis.call('HGET', KEYS[1], ARGV[b + 1]) or '0')
+ used[i] = tostring(u)
+ local lim = tonumber(ARGV[b + 3])
+ if lim > 0 and u + tonumber(ARGV[b + 2]) > lim then ok = false end
+end
+if (ok and commit) or force then
+ for i = 1, m do
+ local b = 3 + (i - 1) * 3
+ if tonumber(ARGV[b + 2]) > 0 then redis.call('HINCRBYFLOAT', KEYS[1], ARGV[b + 1], ARGV[b + 2]) end
+ end
+ redis.call('EXPIRE', KEYS[1], ttl)
+end
+return (ok and '1' or '0') .. ':' .. table.concat(used, ',')`;
/**
/**
* @param {object} o
* @param {object} o
@@ -66,7 +96,7 @@ function createGuardStore(o = {}) {
@@ -66,7 +96,7 @@ function createGuardStore(o = {}) {
v = await valkey.client.get(key);
v = await valkey.client.get(key);
}
}
if (v) {
if (v) {
- salts.set(day, v);
+ salts.set(day, v); // replaces the process-local stand-in: salt() answers the shared one from now on
for (const k of salts.keys()) if (k !== day) salts.delete(k);
for (const k of salts.keys()) if (k !== day) salts.delete(k);
}
}
} catch (err) {
} catch (err) {
@@ -78,29 +108,37 @@ function createGuardStore(o = {}) {
@@ -78,29 +108,37 @@ function createGuardStore(o = {}) {
async function warm() { if (valkey) await loadSalt(dayOf(now())); }
async function warm() { if (valkey) await loadSalt(dayOf(now())); }
// ── guard_day (Valkey) ───────────────────────────────────
// ── guard_day (Valkey) ───────────────────────────────────
- async function dayUsed(day, key, cls) {
+ // One run charges one or more day allowances (a session and its address) all or nothing, and the
+ // check and the increment are one Lua call: two requests can never both see room for the last unit.
+ function dayCharge(day, entries, opt = {}) {
if (cacheDay !== day) { days.clear(); cacheDay = day; }
if (cacheDay !== day) { days.clear(); cacheDay = day; }
- const k = `${day}|${key}|${cls}`;
if (dayUp) {
if (dayUp) {
- try {
- const v = await valkey.client.hget(valkey.key('guard_day', day), `${key}|${cls}`);
- const n = v == null ? 0 : Number(v);
- days.set(k, n);
- return n;
- } catch (err) { dayUp = false; log.warn(`[Guard] ${app}: Valkey day counters unavailable (${err.message}); counting in this process`); }
+ return valkeyDayCharge(day, entries, opt).catch((err) => {
+ dayUp = false;
+ log.warn(`[Guard] ${app}: Valkey day counters unavailable (${err.message}); counting in this process`);
+ return localDayCharge(day, entries, opt);
+ });
}
}
- return days.get(k) || 0;
+ return Promise.resolve(localDayCharge(day, entries, opt));
}
}
- async function dayAdd(day, key, cls, n) {
- if (!(n > 0)) return;
- const k = `${day}|${key}|${cls}`;
- days.set(k, (days.get(k) || 0) + n);
- if (!dayUp) return;
- try {
- const h = valkey.key('guard_day', day);
- await valkey.client.hincrbyfloat(h, `${key}|${cls}`, n);
- await valkey.client.expire(h, dayTtl(now()));
- } catch (err) { dayUp = false; log.warn(`[Guard] ${app}: Valkey day counters unavailable (${err.message}); counting in this process`); }
+ /** In-process: synchronous, so nothing interleaves between the check and the increment. */
+ function localDayCharge(day, entries, { commit = true, force = false } = {}) {
+ const used = entries.map((e) => days.get(`${day}|${e.key}|${e.cls}`) || 0);
+ const ok = entries.every((e, i) => !(e.limit > 0) || used[i] + e.n <= e.limit);
+ if ((ok && commit) || force) {
+ entries.forEach((e, i) => { if (e.n > 0) days.set(`${day}|${e.key}|${e.cls}`, used[i] + e.n); });
+ }
+ return { ok, used };
+ }
+ async function valkeyDayCharge(day, entries, { commit = true, force = false } = {}) {
+ const c = valkey.client;
+ if (typeof c.ovGuardDayCharge !== 'function') c.defineCommand('ovGuardDayCharge', { lua: DAY_CHARGE });
+ const args = [String(dayTtl(now())), commit ? '1' : '0', force ? '1' : '0'];
+ for (const e of entries) args.push(dayField(e.key, e.cls), String(e.n), String(e.limit > 0 ? e.limit : 0));
+ const [ok, list] = String(await c.ovGuardDayCharge(1, dayKey(valkey, day), ...args)).split(':');
+ const used = list ? list.split(',').map(Number) : [];
+ entries.forEach((e, i) => days.set(`${day}|${e.key}|${e.cls}`, used[i] + ((ok === '1' && commit) || force ? e.n : 0)));
+ return { ok: ok === '1', used };
}
}
// ── guard_abuse (PostgreSQL) ─────────────────────────────
// ── guard_abuse (PostgreSQL) ─────────────────────────────
@@ -123,7 +161,7 @@ function createGuardStore(o = {}) {
@@ -123,7 +161,7 @@ function createGuardStore(o = {}) {
async function pruneAbuse(beforeMs) {
async function pruneAbuse(beforeMs) {
for (let i = abuse.length - 1; i >= 0; i--) if (abuse[i].at < beforeMs) abuse.splice(i, 1);
for (let i = abuse.length - 1; i >= 0; i--) if (abuse[i].at < beforeMs) abuse.splice(i, 1);
if (!db) return 0;
if (!db) return 0;
- const n = await db.exec('DELETE FROM guard_abuse WHERE at < $1', [beforeMs]);
+ const n = await db.exec('DELETE FROM guard_abuse WHERE app = $1 AND at < $2', [app, beforeMs]); // this app's rows only: the table is shared
return n || 0;
return n || 0;
}
}
// guard_day rows live in Valkey and expire on their own.
// guard_day rows live in Valkey and expire on their own.
@@ -131,7 +169,7 @@ function createGuardStore(o = {}) {
@@ -131,7 +169,7 @@ function createGuardStore(o = {}) {
async function close() { /* the app owns db and valkey */ }
async function close() { /* the app owns db and valkey */ }
- return { kind: db ? 'postgres' : 'memory', app, db, valkey, salt, warm, dayUsed, dayAdd, logAbuse, abuseRows, pruneAbuse, pruneDays, close };
+ return { kind: db ? 'postgres' : 'memory', app, db, valkey, salt, warm, dayCharge, logAbuse, abuseRows, pruneAbuse, pruneDays, close };
}
}
-module.exports = { createGuardStore, dayOf, DAY_MS };
+module.exports = { createGuardStore, dayOf, dayTtl, dayKey, dayField, DAY_MS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,83 @@
@@ -0,0 +1,83 @@
+'use strict';
+// The guard's Lua on a real Valkey (npm run test:pg sets VALKEY_URL; guard.test.js runs the same rules
+// on an in-process emulation): quota.js CHARGE decides and charges every bucket and day allowance of a
+// run at once, so a run refused on the day leg spends no bucket token and concurrent runs never
+// overshoot the day; store.js DAY_CHARGE is the same all-or-nothing check-and-increment for the day
+// counters alone. Keys live under a random prefix and are deleted at the end.
+const assert = require('assert');
+const crypto = require('crypto');
+const { dep } = require('./deps');
+const { createQuotas } = require('../guard/quota');
+const { createGuardStore } = require('../guard/store');
+
+const quiet = { log() {}, warn() {}, error() {} };
+
+(async () => {
+ if (!process.env.VALKEY_URL) {
+ console.log('guard valkey: skipped (no VALKEY_URL: npm run test:pg runs the Lua on a real Valkey)');
+ return;
+ }
+ const prefix = `ov:tools-test:${crypto.randomBytes(6).toString('hex')}:`;
+ const valkey = dep('openvibe-sdk/valkey').createValkey({ url: process.env.VALKEY_URL, prefix, log: quiet });
+ try {
+ const table = {
+ fetch: { anonymous: { perMinute: 1, burst: 5, perDay: 3 } },
+ wide: { anonymous: { perMinute: 600, burst: 600, perDay: 3 } },
+ session: { session: { perMinute: 60, burst: 10, perDay: 100 }, anonymous: { perMinute: 60, burst: 10, perDay: 100 } },
+ };
+ let clock = Date.UTC(2026, 8, 23, 23, 59, 58);
+ const store = createGuardStore({ valkey, app: 'test', now: () => clock, log: quiet });
+ const q = createQuotas({ store, valkey, quotas: () => table, now: () => clock, log: quiet });
+ const anon = (ip) => ({ tier: 'anonymous', ipKey: `ip:${ip}`, key: `ip:${ip}` });
+
+ // A day refusal spends nothing: 2 runs fit, cost-2 runs are refused on the day leg (2 + 2 > 3)
+ // though the bucket holds 3, and the next day a cost-3 run still finds those 3 tokens.
+ const c = anon('a');
+ for (let i = 0; i < 2; i++) assert.ok((await q.check(c, { quotaClass: 'fetch', cost: 1 })).ok);
+ for (let i = 0; i < 5; i++) assert.strictEqual((await q.check(c, { quotaClass: 'fetch', cost: 2 })).binding, 'day');
+ clock += 2000;
+ assert.strictEqual((await q.check(c, { quotaClass: 'fetch', cost: 3 })).ok, true, 'the refused runs took no token');
+ const short = await q.check(c, { quotaClass: 'fetch', cost: 1 });
+ assert.strictEqual(short.ok, false, 'and now the bucket is empty');
+
+ // Report mode charges a refused run (the work happens anyway).
+ const rep = anon('r');
+ for (let i = 0; i < 3; i++) assert.ok((await q.check(rep, { quotaClass: 'wide', cost: 1 })).ok);
+ assert.strictEqual((await q.check(rep, { quotaClass: 'wide', cost: 1 }, { report: true })).ok, false);
+ const used = await store.dayCharge(new Date(clock).toISOString().slice(0, 10), [{ key: 'ip:r', cls: 'wide', n: 0, limit: 0 }], { commit: false });
+ assert.deepStrictEqual(used.used, [4], 'report mode counted the refused run');
+
+ // Concurrency: 30 runs at once against a day allowance of 3.
+ const all = await Promise.all(Array.from({ length: 30 }, () => q.check(anon('c'), { quotaClass: 'wide', cost: 1 })));
+ assert.strictEqual(all.filter((r) => r.ok).length, 3, 'exactly 3 of 30 concurrent runs');
+
+ // A session run charges the session and its address together, all or nothing.
+ const s1 = { tier: 'session', key: 'session:1', ipKey: 'ip:s' };
+ assert.ok((await q.check(s1, { quotaClass: 'session', cost: 10 })).ok);
+ const s2 = { tier: 'session', key: 'session:2', ipKey: 'ip:s' };
+ const r = await q.check(s2, { quotaClass: 'session', cost: 10 });
+ assert.ok(r.ok || r.scope === 'address', 'the address share binds, not the fresh session');
+
+ // DAY_CHARGE alone (quotas without Valkey): atomic and all or nothing.
+ const day = '2026-09-24';
+ const direct = await Promise.all(Array.from({ length: 12 }, () => store.dayCharge(day, [{ key: 'k', cls: 'c', n: 1, limit: 4 }])));
+ assert.strictEqual(direct.filter((x) => x.ok).length, 4, 'dayCharge: 4 of 12');
+ const both = await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 1, limit: 5 }, { key: 'k', cls: 'c', n: 1, limit: 4 }]);
+ assert.strictEqual(both.ok, false);
+ assert.deepStrictEqual((await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 0, limit: 0 }], { commit: false })).used, [0], 'k2 untouched');
+ const forced = await store.dayCharge(day, [{ key: 'k', cls: 'c', n: 1, limit: 4 }], { force: true });
+ assert.deepStrictEqual([forced.ok, forced.used], [false, [4]], 'force (report mode) charges past the limit');
+
+ // The salt: one per day, shared.
+ await store.warm();
+ const other = createGuardStore({ valkey, app: 'other', now: () => clock, log: quiet });
+ await other.warm();
+ assert.strictEqual(other.salt(), store.salt(), 'one salt in Valkey');
+
+ console.log('guard valkey: CHARGE (a refused run spends nothing, report mode counts, 3 of 30 concurrent, session + address), DAY_CHARGE (atomic, all or nothing, force), shared salt: all checks passed');
+ } finally {
+ const keys = await valkey.client.keys(`${prefix}*`);
+ if (keys.length) await valkey.client.del(...keys);
+ await valkey.close();
+ }
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -44,8 +44,8 @@ const tmp = () => fs.mkdtempSync(path.join(os.tmpdir(), 'ov-guard-'));
@@ -44,8 +44,8 @@ const tmp = () => fs.mkdtempSync(path.join(os.tmpdir(), 'ov-guard-'));
const quiet = { log() {}, warn() {}, error() {} };
const quiet = { log() {}, warn() {}, error() {} };
// A stand-in for openvibe-sdk/valkey: the same key()/client shape over one Map, so two guards can
// A stand-in for openvibe-sdk/valkey: the same key()/client shape over one Map, so two guards can
-// share "one Valkey" in-process. The bucket's two Lua commands are emulated in JS (the real ones are
-// atomic; a single-threaded test has no race to lose).
+// share "one Valkey" in-process. The guard's Lua commands are emulated in JS (the real ones are atomic,
+// and guard-valkey.test.js runs them on a real Valkey under npm run test:pg).
function fakeValkey({ down = false } = {}) {
function fakeValkey({ down = false } = {}) {
const store = new Map();
const store = new Map();
const guard = () => { if (down) throw new Error('valkey down'); };
const guard = () => { if (down) throw new Error('valkey down'); };
@@ -57,22 +57,56 @@ function fakeValkey({ down = false } = {}) {
@@ -57,22 +57,56 @@ function fakeValkey({ down = false } = {}) {
async expire() { guard(); return 1; },
async expire() { guard(); return 1; },
async incr(k) { guard(); const v = Number(store.get(k) || 0) + 1; store.set(k, String(v)); return v; },
async incr(k) { guard(); const v = Number(store.get(k) || 0) + 1; store.set(k, String(v)); return v; },
defineCommand(name) {
defineCommand(name) {
- client[name] = async (_numKeys, key, burst, rate, need, now) => {
- guard();
- const h = store.get(key) || {};
- let tokens = h.tokens != null ? Number(h.tokens) : Number(burst);
- const at = h.at != null ? Number(h.at) : Number(now);
- tokens = Math.min(Number(burst), tokens + (Number(now) - at) * Number(rate));
- if (name === 'ovBucketRefund') { // give back, capped at burst
- const left = Math.min(Number(burst), tokens + Number(need));
+ const num = (x) => Number(x);
+ const refill = (h, burst, rate, now) => {
+ const tokens = h.tokens != null ? num(h.tokens) : burst;
+ const at = h.at != null ? num(h.at) : now;
+ return Math.min(burst, tokens + Math.max(0, now - at) * rate);
+ };
+ if (name === 'ovBucketRefund') { // give back, capped at burst
+ client[name] = async (_numKeys, key, burst, rate, need, now) => {
+ guard();
+ const left = Math.min(num(burst), refill(store.get(key) || {}, num(burst), num(rate), num(now)) + num(need));
store.set(key, { tokens: String(left), at: String(now) });
store.set(key, { tokens: String(left), at: String(now) });
return String(left);
return String(left);
- }
- if (tokens < Number(need)) return `${tokens}:0`; // short: take nothing
- const left = tokens - Number(need);
- store.set(key, { tokens: String(left), at: String(now) });
- return `${left}:1`;
- };
+ };
+ } else if (name === 'ovQuotaCharge') { // quota.js CHARGE: every bucket and day allowance, all or nothing
+ client[name] = async (numKeys, ...rest) => {
+ guard();
+ const keys = rest.slice(0, numKeys), a = rest.slice(numKeys);
+ const n = num(a[0]), now = num(a[1]), report = a[4] === '1';
+ const dh = store.get(keys[n]) || {};
+ let ok = true;
+ const st = [];
+ for (let i = 0; i < n; i++) {
+ const [burst, rate, need, perDay, cost, field] = a.slice(5 + i * 6, 11 + i * 6);
+ const tokens = refill(store.get(keys[i]) || {}, num(burst), num(rate), now);
+ const used = num(perDay) > 0 ? num(dh[field] || 0) : 0;
+ const minuteOk = tokens >= num(need), dayOk = !(num(perDay) > 0) || used + num(cost) <= num(perDay);
+ if (!(minuteOk && dayOk)) ok = false;
+ st.push({ tokens, used, need: num(need), perDay: num(perDay), cost: num(cost), field, minuteOk, dayOk });
+ }
+ if (ok || report) {
+ st.forEach((c, i) => {
+ store.set(keys[i], { tokens: String(Math.max(0, c.tokens - c.need)), at: String(now) });
+ if (c.perDay > 0) dh[c.field] = String(num(dh[c.field] || 0) + c.cost);
+ });
+ store.set(keys[n], dh);
+ }
+ return st.map((c) => `${c.tokens},${c.used},${c.minuteOk ? 1 : 0},${c.dayOk ? 1 : 0}`).join(';');
+ };
+ } else if (name === 'ovGuardDayCharge') { // store.js DAY_CHARGE
+ client[name] = async (_numKeys, key, _ttl, commit, force, ...triples) => {
+ guard();
+ const h = store.get(key) || {};
+ const e = [];
+ for (let i = 0; i < triples.length; i += 3) e.push({ field: triples[i], n: num(triples[i + 1]), limit: num(triples[i + 2]) });
+ const used = e.map((x) => num(h[x.field] || 0));
+ const ok = e.every((x, i) => !(x.limit > 0) || used[i] + x.n <= x.limit);
+ if ((ok && commit === '1') || force === '1') { e.forEach((x, i) => { if (x.n > 0) h[x.field] = String(used[i] + x.n); }); store.set(key, h); }
+ return `${ok ? 1 : 0}:${used.join(',')}`;
+ };
+ } else throw new Error(`fakeValkey: no emulation of ${name}`);
},
},
};
};
return { client, key: (...p) => p.join(':'), ready: async () => (down ? { ok: false, error: 'down' } : { ok: true }), close: async () => {} };
return { client, key: (...p) => p.join(':'), ready: async () => (down ? { ok: false, error: 'down' } : { ok: true }), close: async () => {} };
@@ -316,6 +350,106 @@ async function serve(app) {
@@ -316,6 +350,106 @@ async function serve(app) {
g.close();
g.close();
}
}
+ // ── A refused run takes nothing: a day refusal never drains the minute bucket (shared and local) ──
+ // Just before UTC midnight: 2 runs fit the day (the bucket 5 → 3), ten more are refused on the day
+ // leg. Two seconds later a new day starts with the bucket still at 3 (+ a trickle): 3 runs fit, the
+ // 4th is refused on the minute leg. If the refused runs had spent tokens, the new day would start empty.
+ for (const withValkey of [true, false]) {
+ let clock = Date.UTC(2026, 8, 23, 23, 59, 58);
+ const env = { TOOLS_GUARD: 'enforce', TOOLS_GUARD_LIMITS: JSON.stringify({ 'tools-fetch': { anonymous: { perMinute: 1, burst: 5, perDay: 2 } } }) };
+ const g = makeGuard({ valkey: withValkey ? fakeValkey() : undefined, env, now: () => clock });
+ await g.store.warm();
+ const c = g.resolveCaller({ ip: '198.51.100.40', headers: {}, cookies: {} });
+ const run = () => g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 1 });
+ const label = withValkey ? 'shared' : 'local';
+ assert.ok((await run()).ok && (await run()).ok, `${label}: the day allowance of 2`);
+ for (let i = 0; i < 10; i++) assert.strictEqual((await run()).binding, 'day', `${label}: refused on the day leg`);
+ clock += 2000;
+ const next = [];
+ for (let i = 0; i < 4; i++) next.push(await run());
+ assert.deepStrictEqual(next.map(r => r.ok), [true, true, false, false], `${label}: a new day: the bucket still holds 3, the day allowance is 2 again`);
+ assert.strictEqual(next[2].binding, 'day');
+ g.close();
+ }
+ // The same with a bigger refused run: cost 2 fits the bucket (3 left) but not the day (2 + 2 > 3). Had
+ // the refused runs spent, the next day's cost-3 run would find about 1 token instead of 3.
+ for (const withValkey of [true, false]) {
+ let clock = Date.UTC(2026, 8, 23, 23, 59, 58);
+ const env = { TOOLS_GUARD: 'enforce', TOOLS_GUARD_LIMITS: JSON.stringify({ 'tools-fetch': { anonymous: { perMinute: 1, burst: 5, perDay: 3 } } }) };
+ const g = makeGuard({ valkey: withValkey ? fakeValkey() : undefined, env, now: () => clock });
+ await g.store.warm();
+ const c = g.resolveCaller({ ip: '198.51.100.41', headers: {}, cookies: {} });
+ const label = withValkey ? 'shared' : 'local';
+ for (let i = 0; i < 2; i++) assert.ok((await g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 1 })).ok);
+ for (let i = 0; i < 5; i++) assert.strictEqual((await g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 2 })).binding, 'day');
+ clock += 2000;
+ const r = await g.quotas.check(c, { quotaClass: 'tools-fetch', cost: 3 });
+ assert.strictEqual(r.ok, true, `${label}: the 3 tokens the refused runs never took`);
+ g.close();
+ }
+
+ // ── The day allowance is one atomic check and charge: concurrent runs never overshoot it ──
+ {
+ const env = { TOOLS_GUARD: 'enforce', TOOLS_GUARD_LIMITS: JSON.stringify({ 'tools-fetch': { anonymous: { perMinute: 600, burst: 600, perDay: 3 } } }) };
+ // Two processes, one Valkey, twenty runs at once: exactly 3 fit.
+ const valkey = fakeValkey();
+ const a = makeGuard({ valkey, env }), b = makeGuard({ valkey, env });
+ await a.store.warm(); await b.store.warm();
+ const req = { ip: '198.51.100.50', headers: {}, cookies: {} };
+ const ca = a.resolveCaller(req), cb = b.resolveCaller(req);
+ const all = await Promise.all(Array.from({ length: 20 }, (_, i) => (i % 2 ? a : b).quotas.check(i % 2 ? ca : cb, { quotaClass: 'tools-fetch', cost: 1 })));
+ assert.strictEqual(all.filter(r => r.ok).length, 3, 'shared buckets: 3 of 20 concurrent runs');
+ a.close(); b.close();
+ // The store's own charge (quotas without Valkey, day counters in Valkey): the same.
+ const v2 = fakeValkey();
+ const store = createGuardStore({ valkey: v2, log: quiet });
+ const q = createQuotas({ store, quotas: () => JSON.parse(env.TOOLS_GUARD_LIMITS), log: quiet });
+ const caller = { tier: 'anonymous', ipKey: 'ip:concurrent', key: 'ip:concurrent' };
+ const r2 = await Promise.all(Array.from({ length: 20 }, () => q.check(caller, { quotaClass: 'tools-fetch', cost: 1 })));
+ assert.strictEqual(r2.filter(r => r.ok).length, 3, 'store day charge: 3 of 20 concurrent runs');
+ const day = new Date().toISOString().slice(0, 10);
+ const direct = await Promise.all(Array.from({ length: 10 }, () => store.dayCharge(day, [{ key: 'k', cls: 'c', n: 1, limit: 4 }])));
+ assert.strictEqual(direct.filter(r => r.ok).length, 4, 'dayCharge: 4 of 10');
+ const both = await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 1, limit: 5 }, { key: 'k', cls: 'c', n: 1, limit: 4 }]);
+ assert.strictEqual(both.ok, false, 'all or nothing: k is full');
+ assert.deepStrictEqual((await store.dayCharge(day, [{ key: 'k2', cls: 'c', n: 0, limit: 5 }], { commit: false })).used, [0], 'so k2 was not charged');
+ }
+
+ // ── A salt handed out before Valkey answered is replaced by the shared one as soon as it loads ──
+ {
+ const valkey = fakeValkey();
+ let release;
+ const gate = new Promise(r => { release = r; });
+ const get = valkey.client.get;
+ valkey.client.get = async (k) => { await gate; return get(k); };
+ const req = { ip: '198.51.100.70', headers: {}, cookies: {} };
+ const a = makeGuard({ valkey });
+ const early = a.resolveCaller(req).ipKey; // Valkey has not answered: a process-local salt
+ release();
+ await a.store.warm();
+ const b = makeGuard({ valkey });
+ await b.store.warm();
+ const later = a.resolveCaller(req).ipKey;
+ assert.notStrictEqual(later, early, 'the process-local stand-in is gone once the shared salt loads');
+ assert.strictEqual(later, b.resolveCaller(req).ipKey, 'and both processes agree on the key');
+ a.close(); b.close();
+ }
+
+ // ── One app's prune leaves the other apps' abuse rows alone (one shared guard_abuse table) ──
+ {
+ const dir = tmp();
+ const db = await testDb(dir);
+ const old = now0 - 40 * 24 * 3600 * 1000;
+ for (const app of ['img', 'docs']) {
+ await db.query('INSERT INTO guard_abuse (app, at, minute, ip_hash, principal, tool, reason, enforced) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)',
+ [app, old, Math.floor(old / 60_000), 'h', '', 'png', 'quota', 1]);
+ }
+ const img = createGuardStore({ db, app: 'img', log: quiet });
+ assert.strictEqual(await img.pruneAbuse(now0), 1, "img prunes its own row");
+ const left = await db.query('SELECT app FROM guard_abuse ORDER BY app');
+ assert.deepStrictEqual(left.map(r => r.app), ['docs'], "docs' row is untouched");
+ }
+
// ── The abuse log: no raw address anywhere, 30 days ──
// ── The abuse log: no raw address anywhere, 30 days ──
{
{
let clock = now0;
let clock = now0;
@@ -569,5 +703,5 @@ async function serve(app) {
@@ -569,5 +703,5 @@ async function serve(app) {
}
}
await closeAllTestDbs();
await closeAllTestDbs();
- console.log('guard: callers (/64, one loopback hop, services by token, aud-checked users, sessions, sandbox), bucket math + cost, session sharing, report vs enforce, headers + problem+json, two processes one Valkey quota + outage fallback, sniffing, target throttle, port cap, abuse log + prune + metric, semaphore, job bounds (queue, disk, per address), challenge hook, Origin check (CSRF): all checks passed');
+ console.log('guard: callers (/64, one loopback hop, services by token, aud-checked users, sessions, sandbox), bucket math + cost, session sharing, report vs enforce, headers + problem+json, two processes one Valkey quota + outage fallback, refused runs take nothing, atomic day charge, salt refresh, per-app prune, sniffing, target throttle, port cap, abuse log + prune + metric, semaphore, job bounds (queue, disk, per address), challenge hook, Origin check (CSRF): all checks passed');
})().catch((err) => { console.error(err); process.exit(1); });
})().catch((err) => { console.error(err); process.exit(1); });