Skip to content

Commit 85f7a0d

Browse files
fix(server): budget counters hold their limit per key, count in a registered cache, and keep the in-process fallback bounded (#29345)
<!-- claude-projects-attribution: {"github_login":"os-steve"} --> _Requested by **Steve** · [project thread](https://claude.ai/code/project/chan_01FdL1XeUyvVNxLreh3F4Qi9?thread=cmsg_01FdL1XeUyvVNxLreh3F4Qi9DDnuBdW9LnD9CpSuCNuHyX)_ Refs #29053 Clause-②: no Metadata-protocol: none ## 维护者速读 - 改了什么:把 #29216 给 plugin-auth 定下的计数规则推广到 server 车道的其余预算计数器:runtime 的入站限流与端点限流(令牌桶)同一个键的更新在进程内串行;runtime、短信每日配额、notify 节点按地址的邮件限额都按同一规则解析 `cache` 服务(值注册、工厂注册都能拿到,内核自带的 degraded 内存替身不算共享缓存);core 的内存 cache 兜底在写入时清掉已过期的键,不必等再读一次。 - 为什么改:按维护者裁决「A+C 同意」的 C 半,规则是:每个预算计数器在并发请求下守住限额(同一个键的更新进程内串行),注册了共享 cache 时计在其中(ADR-0069 D2),进程内兜底保持有界。 - 风险与回滚:同一个键的更新会排队,不同键互不影响;没有 cache 插件时 runtime 的限流桶改走有界的进程内兜底,并照旧告警一次。跨节点仍没有原子自增(契约在 spec,本 PR 不动)。回滚即撤销本 PR。 - 先看:`packages/runtime/src/security/inbound-rate-limit.ts`(`SharedTokenBucketLimiter.consume`)、`packages/runtime/src/security/kernel-cache.ts`、`packages/core/src/fallbacks/memory-cache.ts` 的写入清扫。 - 两分钟验证:`pnpm --filter @objectstack/runtime exec vitest run src/security/inbound-rate-limit-concurrency.test.ts src/security/kernel-cache.test.ts src/dispatcher-plugin.rate-limit-cache.integration.test.ts`,预期 3 个测试文件、15 个用例全部通过。 - 需要你做的:无,席位落地。 ## What changes The rule, for every budget counter in the server lane: **every budget counter holds its limit under concurrent requests (updates to one key are serialised in-process), counts in the registered shared cache when one is registered (ADR-0069 D2), and its in-process fallback stays bounded.** PR #29216 applied it to `@objectstack/plugin-auth`; this applies it to the rest. - **`@objectstack/runtime` token buckets.** `SharedTokenBucketLimiter.consume` runs its read, refill, spend and write for one key through `serializePerKey` (new, package-local `src/security/per-key-serial.ts`). That covers the server-level inbound limiter (`server.security.rateLimit`) and every declared endpoint `rateLimit`, which share the class. Other keys run concurrently. - **Cache resolution.** The dispatcher's inbound and endpoint limiters, the daily SMS send quota (`@objectstack/service-sms`) and the `notify` node's per-address email-field limit (`@objectstack/service-automation`) resolve the `cache` service through a package-local `resolveKernelCache` (`src/security/kernel-cache.ts`, `src/kernel-cache.ts`, `src/builtin/kernel-cache.ts`): a value registration and a factory registration both answer; the kernel's own in-memory stand-in, which self-describes `degraded` (ADR-0076 D12), does not, so those counters keep the bounded per-process fallback and its one-shot warning. - **`@objectstack/core` in-memory cache fallback.** Expired entries are evicted on write once the map reaches a sweep mark (twice the entries left after the previous sweep, at least 1024), so the map holds at most about twice its live entries without a key having to be read again. No live entry (no TTL, or a TTL still running) is evicted; the `ICacheService` contract is unchanged. - **Changeset**: `.changeset/29053-server-budget-counters.md`, patch for `@objectstack/runtime`, `@objectstack/service-sms`, `@objectstack/service-automation`, `@objectstack/core`. **Local copies, by the seat's decision.** `per-key-serial.ts` and `kernel-cache.ts` are copies of the two internal plugin-auth helpers #29216 added (same rule, same code), one per package that needs them. Re-exporting them from plugin-auth's published `./rate-limit-storage` entry would widen that entry, so they stay internal and are not re-exported from any package. ## Why ADR-0069 D2 promises a budget counted once across nodes when a shared cache is registered, and a counter's limit is only a limit if it holds when requests for one key arrive together. #29216 measured both halves for the auth counters; the server lane's other counters follow the same rule. ## Tests - `packages/runtime/src/security/inbound-rate-limit-concurrency.test.ts` (new, 6 cases): concurrent consumes of one key never exceed the budget, in a cache whose calls each cross a macrotask, in the per-process fallback through the middleware, and for a declared endpoint budget; serialisation is per key; a failing update releases the next; nothing is left in flight once the concurrent updates settle. - `packages/runtime/src/security/kernel-cache.test.ts` (new, 7 cases), against real `ObjectKernel` and `LiteKernel` contexts: factory, value, none, the `degraded` stand-in, a non-store value, and a bucket counted through it lands in the factory-registered cache. - `packages/runtime/src/dispatcher-plugin.rate-limit-cache.integration.test.ts` (new, 2 cases), the composed dispatcher on a real `ObjectKernel` over a socket: with a factory-registered cache the caller's bucket lives in it and the budget holds; with only the stand-in, the stand-in stays empty and the budget still holds. - `packages/services/service-sms/src/sms-daily-quota-cache.test.ts` (new, 2 cases) and `packages/services/service-automation/src/builtin/notify-email-field-cache.integration.test.ts` (new, 2 cases), on a real `ObjectKernel`: with a factory-registered cache the counter lives in it and the limit holds; with only the stand-in, the stand-in stays empty and the limit still holds. - `packages/core/src/fallbacks/memory-cache-bound.test.ts` (new, 3 cases): the fallback stays bounded after many distinct keys expire unread; live entries survive every sweep; an expired entry still reads as absent. - **Forced test edits**: none. `scripts/engine-double-contract.pinned.json` gains the new notify test's data double (written by `check-engine-double-contract.mjs --write`). - Each pin goes red with its check removed (eight ablations, each restored to the HEAD blob). ## Acceptance notes - **Across nodes, unchanged.** `ICacheService` has no atomic increment, so two nodes sharing one cache still read-modify-write against each other. Adding one is a `packages/spec` contract change, outside this lane. - **What the rule costs.** Updates of one key wait for each other; other keys are unaffected. The fallback cache's sweep costs O(1) per write amortised: on this shared box, one million writes of distinct keys took the same time as before within noise when all stay live, and less when they expire (the map stays small). - **The stand-in rule** matches #29216's choice: a cache-less `ObjectKernel` deployment's runtime buckets move from the stand-in to the bounded per-process fallback, which is just as process-local, and the "no shared cache" warning now fires there. ## Verification Final head `0c0c6185` (origin/main `39951d2a` merged in `edc9e043`; the two commits after it touch only test harness files: the notify pin's data double and its ledger row, and `service-sms`'s `vitest.config.ts`). Each reading names the head it was taken on. <details><summary>Tests and typecheck</summary> - On `edc9e043`, under the verify lock: `turbo run build` of the four packages and their dependencies, exit 0; `pnpm --filter PKG test`: `@objectstack/core` 130 files / 2912 tests passed; `@objectstack/service-sms` 8 / 109; `@objectstack/service-automation` 260 / 3337; `@objectstack/runtime` 518 files, 8346 passed, 19 skipped. `pnpm --filter PKG typecheck` exit 0 for core, service-sms and service-automation; runtime's first read was a missing optional-driver build (no `dist` for `@objectstack/driver-turso`), and after `turbo run build --filter='./packages/**'` it read exit 0 (`check:test-typecheck` OK, debt ledger held). - On `a7dd43d1`: the notify pin, 2 passed. - On `0c0c6185`: `@objectstack/service-sms` test 8 files / 109 tests passed, typecheck exit 0. </details> Each pin goes red with its check removed (eight ablations, each restored to the HEAD blob). <details><summary>Dispatch gates and lint</summary> - `node scripts/pm/dispatch-gates.mjs --commands --repo objectstack-ai/objectstack` on `0c0c6185` derived 70 commands (18 paths, 1005 changed lines, under the 3000-line threshold). All 70 run with stdin from /dev/null and the exit captured before any pipe: 70 exit 0. `--ran`: 70 derived, 70 run, 0 NOT-MEASURED, 0 UNRUN (a derived zero). - Two families read red on the earlier head and were fixed in the two commits after it: `check:engine-double-contract` (the notify pin's data double now routes `findOne` through `assertEngineFindOnePredicate`, and the pinned ledger gained its row via `--write`) and `check:test-source-alias` (`service-sms` aliases `@objectstack/core` and `@objectstack/spec/api` to source rather than widening the shrink-only registry). - `eslint --no-inline-config` over `packages/runtime`, `packages/core`, `packages/services/service-sms`, `packages/services/service-automation` on `edc9e043`: exit 0; over `packages/services/service-sms` on `0c0c6185`: exit 0. The repo-wide run is CI's. - `test:repo` is not run locally (no file of that project changed); it is CI's `Test Core`. - CI check-runs: not read yet at the time of writing. </details> ## Human Minimum - [x] 1. Schemas start as Zod; `packages/spec` holds no runtime logic. (No schema touched.) - [x] 2. Every name I added uses the casing for its kind. - [x] 3. No formatter ran; `pnpm lint` passes. (eslint over the four touched packages; the repo-wide run is CI's.) - [x] 4. One issue, one branch off an up-to-date `main`, one pull request. - [x] 5. A change that publishes carries a changeset, and both declaration lines are filled in here and in the changeset. - [x] 6. `packages/spec` touched: its generated artifacts are regenerated and committed. (Not touched.) - [x] 7. No hand edits to `content/docs/references/`, `packages/*/CHANGELOG.md` or `content/docs/releases/`. - [x] 8. Title, body and comments are in English. (The 维护者速读 section is Chinese by the repo's maintainer rule.) - [x] The *Before Submitting* list in CONTRIBUTING.md is done. <!-- signed: domain:server#5 · os-steve · session_01ADHkmPcNztYJ49AQY8UaeN --> --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent f2f6bf0 commit 85f7a0d

18 files changed

Lines changed: 965 additions & 43 deletions
Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
---
2+
'@objectstack/runtime': patch
3+
'@objectstack/service-sms': patch
4+
'@objectstack/service-automation': patch
5+
'@objectstack/core': patch
6+
---
7+
8+
fix(server): every budget counter holds its limit under concurrent requests, counts in a registered `cache` service, and keeps its in-process fallback bounded
9+
10+
Clause-②: no
11+
Metadata-protocol: none
12+
13+
The rule `@objectstack/plugin-auth`'s counters already follow now holds for the server's other budget counters (ADR-0069 D2).
14+
15+
- **`@objectstack/runtime`: a token bucket holds its budget under concurrency inside one process.** Updates of one bucket key are serialised in the process, so the inbound limiter (`server.security.rateLimit`) and a declared endpoint `rateLimit` admit no more requests than their budget for one caller however many arrive at once. Other callers keep counting concurrently.
16+
- **The counters count in the registered `cache`.** When a `cache` service is registered, as a value or as a factory, the runtime's inbound and endpoint buckets, the daily SMS send quota (`@objectstack/service-sms`) and the `notify` node's per-address email-field limit (`@objectstack/service-automation`) count in it. The kernel's own in-memory stand-in is not a shared store: with no cache plugin they keep their bounded per-process fallback and say so once.
17+
- **`@objectstack/core`: the in-memory `cache` fallback stays bounded.** Entries whose time-to-live has passed leave it without having to be read again, so it holds at most about twice its live entries. Nothing live is evicted; the `ICacheService` contract is unchanged.
18+
- **Across nodes, unchanged.** `ICacheService` has no atomic increment, so two nodes that share one cache can still over-admit slightly between them, as documented.
Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
/**
4+
* The rule these pin: the kernel's in-memory `cache` stand-in stays bounded by
5+
* what is live. Entries whose TTL has passed leave the map without having to
6+
* be read again, and nothing live is ever evicted to get there.
7+
*/
8+
9+
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
10+
11+
import { createMemoryCache } from './memory-cache';
12+
13+
const WAVE = 1_000;
14+
const WAVES = 20;
15+
16+
describe('createMemoryCache — bounded by its live entries', () => {
17+
beforeEach(() => {
18+
vi.useFakeTimers();
19+
vi.setSystemTime(1_000_000);
20+
});
21+
afterEach(() => {
22+
vi.useRealTimers();
23+
});
24+
25+
it('many distinct keys that expire unread do not accumulate', async () => {
26+
const cache = createMemoryCache();
27+
for (let wave = 0; wave < WAVES; wave++) {
28+
for (let i = 0; i < WAVE; i++) await cache.set(`wave${wave}:key${i}`, { n: 1 }, 1);
29+
// Let the whole wave expire; none of its keys is ever read again.
30+
vi.advanceTimersByTime(2_000);
31+
}
32+
const { keyCount } = await cache.stats();
33+
// Unbounded, the map would hold every key ever written (WAVES × WAVE).
34+
expect(keyCount).toBeLessThan(3 * WAVE);
35+
});
36+
37+
it('never evicts a live entry: no TTL, or a TTL still running, survives every sweep', async () => {
38+
const cache = createMemoryCache();
39+
await cache.set('pinned', 'forever');
40+
await cache.set('long', 'still-live', 3_600);
41+
for (let wave = 0; wave < WAVES; wave++) {
42+
for (let i = 0; i < WAVE; i++) await cache.set(`wave${wave}:key${i}`, i, 1);
43+
vi.advanceTimersByTime(2_000);
44+
}
45+
// The last wave was written with time running; write one more, still live.
46+
for (let i = 0; i < WAVE; i++) await cache.set(`live:key${i}`, i, 60);
47+
expect(await cache.get('pinned')).toBe('forever');
48+
expect(await cache.get('long')).toBe('still-live');
49+
for (let i = 0; i < WAVE; i++) expect(await cache.get(`live:key${i}`)).toBe(i);
50+
});
51+
52+
it('an expired entry still reads as absent before any sweep reaches it', async () => {
53+
const cache = createMemoryCache();
54+
await cache.set('k', 'v', 1);
55+
vi.advanceTimersByTime(1_500);
56+
expect(await cache.get('k')).toBeUndefined();
57+
});
58+
});

‎packages/core/src/fallbacks/memory-cache.ts‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
22

3+
/** The fewest entries at which a write sweeps expired entries out. */
4+
const SWEEP_FLOOR = 1024;
5+
36
/**
47
* In-memory Map-backed cache fallback.
58
*
@@ -15,11 +18,28 @@
1518
* it as fully `available`. `handlerReady: false` because
1619
* no HTTP surface is mounted for `cache` at all — the same reason realtime
1720
* reports false.
21+
*
22+
* Bounded by what is live: an entry whose TTL has passed is evicted without
23+
* having to be read again. A `get` of an expired key drops it as before, and a
24+
* `set` that grows the map to the next sweep mark walks it once and drops every
25+
* expired entry. The mark is then set to twice the entries left (never below
26+
* {@link SWEEP_FLOOR}), so the map holds at most about twice its live entries
27+
* and the walks cost O(1) per write, amortised. A key that is written once and
28+
* never read again — a per-caller counter, say — therefore does not stay
29+
* forever. Nothing live is ever evicted: an entry with no TTL, or a TTL still
30+
* running, survives every sweep, so the `ICacheService` contract is unchanged.
1831
*/
1932
export function createMemoryCache() {
2033
const store = new Map<string, { value: unknown; expires?: number }>();
2134
let hits = 0;
2235
let misses = 0;
36+
let sweepAt = SWEEP_FLOOR;
37+
const sweepExpired = (now: number): void => {
38+
for (const [key, entry] of store) {
39+
if (entry.expires && now > entry.expires) store.delete(key);
40+
}
41+
sweepAt = Math.max(SWEEP_FLOOR, store.size * 2);
42+
};
2343
return {
2444
__serviceInfo: {
2545
status: 'degraded' as const,
@@ -38,7 +58,9 @@ export function createMemoryCache() {
3858
return entry.value as T;
3959
},
4060
async set<T = unknown>(key: string, value: T, ttl?: number): Promise<void> {
41-
store.set(key, { value, expires: ttl ? Date.now() + ttl * 1000 : undefined });
61+
const now = Date.now();
62+
store.set(key, { value, expires: ttl ? now + ttl * 1000 : undefined });
63+
if (store.size >= sweepAt) sweepExpired(now);
4264
},
4365
async delete(key: string): Promise<boolean> { return store.delete(key); },
4466
async has(key: string): Promise<boolean> { return store.has(key); },
Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
/**
4+
* The rule these pin (ADR-0069 D2), through the composed dispatcher: the
5+
* inbound limiter's buckets count in the registered `cache` service — a value
6+
* or a FACTORY registration — and only when none is registered (the kernel's
7+
* own `degraded` stand-in included) in the bounded per-process fallback.
8+
*
9+
* Booted on a real `ObjectKernel`, the kernel that supports factory
10+
* registrations, over a real socket.
11+
*/
12+
13+
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
14+
import { createMemoryCache, ObjectKernel, type Plugin, type PluginContext } from '@objectstack/core';
15+
import { HonoServerPlugin } from '@objectstack/plugin-hono-server';
16+
import type { IHttpServer } from '@objectstack/spec/contracts';
17+
18+
import { createDispatcherPlugin } from './dispatcher-plugin.js';
19+
20+
const BUDGET = { enabled: true, windowMs: 60_000, maxRequests: 2 };
21+
22+
function fakeAuthPlugin(): Plugin {
23+
return {
24+
name: 'com.objectstack.test.fake-auth',
25+
version: '1.0.0',
26+
init: async (ctx: PluginContext) => {
27+
ctx.registerService('auth', {
28+
api: {
29+
getSession: async ({ headers }: { headers: any }) => {
30+
const uid = typeof headers?.get === 'function' ? headers.get('x-test-user') : headers?.['x-test-user'];
31+
return uid ? { user: { id: uid } } : undefined;
32+
},
33+
},
34+
});
35+
},
36+
};
37+
}
38+
39+
function cachePlugin(register: (ctx: PluginContext) => void): Plugin {
40+
return { name: 'com.objectstack.test.cache', version: '1.0.0', init: async (ctx: PluginContext) => register(ctx) };
41+
}
42+
43+
async function boot(register: (ctx: PluginContext) => void) {
44+
const kernel = new ObjectKernel({ logger: { level: 'silent' }, gracefulShutdown: false, skipSystemValidation: true });
45+
await kernel.use(fakeAuthPlugin());
46+
await kernel.use(cachePlugin(register));
47+
await kernel.use(new HonoServerPlugin({ port: 0 }));
48+
await kernel.use(createDispatcherPlugin({
49+
prefix: '/api/v1',
50+
securityHeaders: false,
51+
rateLimit: { budget: BUDGET, trustProxy: false },
52+
}));
53+
await kernel.bootstrap();
54+
const httpServer = kernel.getService<IHttpServer>('http.server');
55+
return { kernel, baseUrl: `http://127.0.0.1:${httpServer.getPort!()}` };
56+
}
57+
58+
async function shutdown(kernel: ObjectKernel | undefined) {
59+
if (!kernel) return;
60+
await Promise.race([kernel.shutdown(), new Promise<void>((resolve) => setTimeout(resolve, 10_000))]);
61+
}
62+
63+
async function statusesFor(baseUrl: string, user: string, n: number): Promise<number[]> {
64+
const statuses: number[] = [];
65+
for (let i = 0; i < n; i++) {
66+
statuses.push((await fetch(`${baseUrl}/api/v1/health`, { headers: { 'x-test-user': user } })).status);
67+
}
68+
return statuses;
69+
}
70+
71+
describe('inbound limiter, `cache` registered by FACTORY', () => {
72+
const entries = new Map<string, unknown>();
73+
let kernel: ObjectKernel;
74+
let baseUrl: string;
75+
76+
beforeAll(async () => {
77+
({ kernel, baseUrl } = await boot((ctx) => ctx.registerServiceFactory('cache', () => ({
78+
get: async (key: string) => entries.get(key),
79+
set: async (key: string, value: unknown) => { entries.set(key, value); },
80+
}))));
81+
}, 30_000);
82+
afterAll(() => shutdown(kernel), 30_000);
83+
84+
it("the caller's bucket lives in the registered cache, and the budget holds", async () => {
85+
expect(await statusesFor(baseUrl, 'usr_factory', BUDGET.maxRequests + 1)).toEqual([200, 200, 429]);
86+
expect([...entries.keys()]).toContain('principal:usr_factory');
87+
});
88+
});
89+
90+
describe("inbound limiter, only the kernel's `degraded` stand-in registered", () => {
91+
const standIn = createMemoryCache();
92+
let kernel: ObjectKernel;
93+
let baseUrl: string;
94+
95+
beforeAll(async () => {
96+
({ kernel, baseUrl } = await boot((ctx) => ctx.registerService('cache', standIn)));
97+
}, 30_000);
98+
afterAll(() => shutdown(kernel), 30_000);
99+
100+
it('the stand-in is not counted in: the bounded fallback holds the budget', async () => {
101+
expect(await statusesFor(baseUrl, 'usr_standin', BUDGET.maxRequests + 1)).toEqual([200, 200, 429]);
102+
expect((await standIn.stats()).keyCount).toBe(0);
103+
});
104+
});

‎packages/runtime/src/dispatcher-plugin.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@ import { Plugin, PluginContext, IHttpServer, ANONYMOUS_DENY_BODY, ANONYMOUS_DENY
44
import { looksLikeInternalErrorLeak, INTERNAL_ERROR_MESSAGE, resolveThrownHttpError, serverFaultProvenance, declaredRefusalMessage, demotedDeclaredCode, logServerFault, describeFaultRequest, inProcessSessionReadInput } from '@objectstack/types';
55
import { DispatcherErrorCode } from '@objectstack/spec/api';
66
import type { IAuthService, IMetadataService, Logger } from '@objectstack/spec/contracts';
7-
import type { CounterStore } from '@objectstack/plugin-auth/rate-limit-storage';
87
import { HttpDispatcher, HttpDispatcherResult, type HttpProtocolContext } from './http-dispatcher.js';
98
import { isServiceServeable } from './service-serveable.js';
109
import { deferRequestBody } from './deferred-request-body.js';
@@ -21,6 +20,7 @@ import {
2120
} from './security/index.js';
2221
import { resolveSessionData, resolveSessionPrincipalId } from './security/resolve-session-principal.js';
2322
import { buildActorUser } from './security/actor-user.js';
23+
import { resolveKernelCache } from './security/kernel-cache.js';
2424
import {
2525
NoopMetricsRegistry,
2626
NoopErrorReporter,
@@ -911,7 +911,7 @@ export function createDispatcherPlugin(config: DispatcherPluginConfig = {}): Plu
911911
safeGetService<IAuthService>(ctx, 'auth'),
912912
headers as Record<string, unknown>,
913913
),
914-
resolveCache: async () => safeGetService<CounterStore>(ctx, 'cache'),
914+
resolveCache: () => resolveKernelCache(ctx),
915915
logger: ctx.logger,
916916
});
917917
// `null` = no budget declared, or declared disabled. Nothing is
@@ -2190,7 +2190,7 @@ export function createDispatcherPlugin(config: DispatcherPluginConfig = {}): Plu
21902190
// Same `resolveCache` the server-level limiter uses, so one
21912191
// deployment has one counter backend, not two.
21922192
const endpointLimiters = createEndpointRateLimiterRegistry({
2193-
resolveCache: async () => safeGetService<CounterStore>(ctx, 'cache'),
2193+
resolveCache: () => resolveKernelCache(ctx),
21942194
logger: ctx.logger,
21952195
});
21962196

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
/**
4+
* The rule these pin (ADR-0069 D2): inside one process, a token bucket admits
5+
* no more than its budget however many requests for one key arrive at once —
6+
* in a networked cache whose read and write are separate awaited calls, and in
7+
* the per-process fallback alike. Serialisation is per KEY: other keys keep
8+
* counting concurrently, and nothing is left behind once a burst has drained.
9+
*
10+
* The same `SharedTokenBucketLimiter` backs the server-level inbound limiter
11+
* and the declarative endpoint limiters, so both are exercised through their
12+
* public builders.
13+
*/
14+
15+
import { describe, expect, it } from 'vitest';
16+
import type { CounterStore } from '@objectstack/plugin-auth/rate-limit-storage';
17+
18+
import { createEndpointRateLimiterRegistry } from '../endpoint-policy.js';
19+
import { createInboundRateLimitMiddleware, SharedTokenBucketLimiter } from './inbound-rate-limit.js';
20+
import { serializePerKey, serializedKeysInFlight } from './per-key-serial.js';
21+
22+
const BURST = 20;
23+
const CAPACITY = 3;
24+
/** A slow refill, so no token comes back while a burst is in flight. */
25+
const BUCKET = { capacity: CAPACITY, refillPerSec: 0.001 } as const;
26+
const CLOCK = (): number => 1_000_000;
27+
28+
/** A cache stand-in whose reads and writes each cross a macrotask, as a networked store's do. */
29+
function makeNetworkedCache(): CounterStore & { readonly store: Map<string, unknown> } {
30+
const store = new Map<string, unknown>();
31+
const hop = (): Promise<void> => new Promise((resolve) => setTimeout(resolve, 0));
32+
return {
33+
store,
34+
async get<T = unknown>(key: string): Promise<T | undefined> {
35+
await hop();
36+
return store.get(key) as T | undefined;
37+
},
38+
async set<T = unknown>(key: string, value: T): Promise<void> {
39+
await hop();
40+
store.set(key, value);
41+
},
42+
};
43+
}
44+
45+
async function admittedOfBurst(consume: () => Promise<{ allowed: boolean }>): Promise<number> {
46+
const decisions = await Promise.all(Array.from({ length: BURST }, consume));
47+
return decisions.filter((d) => d.allowed).length;
48+
}
49+
50+
describe('concurrent consumes of one key never exceed the budget (one process)', () => {
51+
it('networked cache: exactly `capacity` of the burst are admitted', async () => {
52+
const cache = makeNetworkedCache();
53+
const limiter = new SharedTokenBucketLimiter(BUCKET, async () => cache, CLOCK);
54+
expect(await admittedOfBurst(() => limiter.consume('principal:burst'))).toBe(CAPACITY);
55+
// The stored bucket is empty: every admitted request spent its token.
56+
expect(JSON.parse(cache.store.get('principal:burst') as string)).toMatchObject({ t: 0 });
57+
});
58+
59+
it('per-process fallback (no cache service): exactly `capacity` of the burst are admitted', async () => {
60+
const middleware = createInboundRateLimitMiddleware({
61+
budget: { enabled: true, maxRequests: CAPACITY, windowMs: 3_600_000 },
62+
resolveCache: async () => undefined,
63+
})!;
64+
let passed = 0;
65+
let refused = 0;
66+
await Promise.all(Array.from({ length: BURST }, async () => {
67+
const res: any = {
68+
status(code: number) { if (code === 429) refused++; return res; },
69+
header() { return res; },
70+
json() {},
71+
send() {},
72+
};
73+
const req: any = { method: 'GET', path: '/api/v1/x', headers: {}, query: {}, params: {}, remoteAddress: '10.9.9.9' };
74+
await middleware(req, res, async () => { passed++; });
75+
}));
76+
expect(passed).toBe(CAPACITY);
77+
expect(refused).toBe(BURST - CAPACITY);
78+
});
79+
80+
it('declared endpoint budget: exactly `capacity` of the burst are admitted', async () => {
81+
const cache = makeNetworkedCache();
82+
const registry = createEndpointRateLimiterRegistry({ resolveCache: async () => cache, now: CLOCK });
83+
const limiter = registry.limiterFor({
84+
name: 'burst_endpoint',
85+
rateLimit: { enabled: true, maxRequests: CAPACITY, windowMs: 3_600_000 },
86+
} as never)!;
87+
expect(await admittedOfBurst(() => limiter.consume('endpoint:burst_endpoint:ip:10.0.0.1'))).toBe(CAPACITY);
88+
});
89+
90+
it('serialisation is per key: a burst on one key does not change another key', async () => {
91+
const cache = makeNetworkedCache();
92+
const limiter = new SharedTokenBucketLimiter(BUCKET, async () => cache, CLOCK);
93+
const [a, b] = await Promise.all([
94+
admittedOfBurst(() => limiter.consume('ip:10.0.0.1')),
95+
admittedOfBurst(() => limiter.consume('ip:10.0.0.2')),
96+
]);
97+
expect([a, b]).toEqual([CAPACITY, CAPACITY]);
98+
});
99+
100+
it('a key parked inside its update does not hold up another key', async () => {
101+
let release!: () => void;
102+
const parked = serializePerKey('parked', () => new Promise<void>((resolve) => { release = resolve; }));
103+
await expect(serializePerKey('other', async () => 'ran')).resolves.toBe('ran');
104+
release();
105+
await parked;
106+
});
107+
108+
it('a failing update releases the next one, and no key is left behind once a burst drains', async () => {
109+
const failing = serializePerKey('k', async () => { throw new Error('store down'); });
110+
const next = serializePerKey('k', async () => 'next');
111+
await expect(failing).rejects.toThrow('store down');
112+
await expect(next).resolves.toBe('next');
113+
114+
const cache = makeNetworkedCache();
115+
const limiter = new SharedTokenBucketLimiter(BUCKET, async () => cache, CLOCK);
116+
await admittedOfBurst(() => limiter.consume('principal:drain'));
117+
await new Promise((resolve) => setTimeout(resolve, 0));
118+
expect(serializedKeysInFlight()).toBe(0);
119+
});
120+
});

0 commit comments

Comments
 (0)