commit e330fa0221133d89dee7df5ad381f20890f47f28
parent 85312f48466959b9955502f9285afbddaa2ee9f7
Author: Michael Camilleri <[email protected]>
Date: Fri, 3 Jul 2026 09:22:01 +0900
Sweep stale rate-limit keys in the push worker
The push worker's Durable Object wrote sliding-window rate entries per
rotatable identity — IP, deviceID, credID, push address — and never
deleted them, so storage grew without bound once an identity stopped
appearing. This commit ports the room worker's alarm sweep:
checkRateLimit arms the alarm if none is pending, and
PushRegistry.alarm() prunes `rate:` keys whose newest entry is older
than the sweep horizon, re-arming only while live keys remain so an idle
registry goes quiet.
Unlike the room worker's single window, the push buckets span 60-second
publish limits to one-hour attest limits, and a bucket's window cannot
be recovered from its HMAC-digested storage key. The sweep therefore
uses one horizon — the largest configured window — which every bucket
has outlived by the time its key is pruned. The per-bucket defaults are
hoisted into a shared RATE_LIMIT_DEFAULTS table read by both the
enforcement call sites and the horizon, so the two cannot drift. The
change needs a deploy of the push worker; no Durable Object migration is
required.
This commit also introduces a worker test suite (Tests/Workers/, run
via `bash Scripts/test-workers.sh`). Node's built-in runner drives the
push and room limiters against a stubbed Durable Object storage,
covering limit enforcement, window sliding, per-identity isolation,
and both workers' sweep behaviour.
Co-Authored-By: Claude Fable 5 <[email protected]>
Diffstat:
6 files changed, 341 insertions(+), 11 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
@@ -41,3 +41,4 @@ Puzzles are plain-text `.xd` files. Other supported puzzle formats are converted
bash Scripts/test-unit.sh > /tmp/test.log 2>&1; grep "Test case" /tmp/test.log; echo "---"; tail -15 /tmp/test.log
```
This surfaces every per-case result and the trailing summary (totals on success, `.xcresult` path on failure). For more context on a failure, grep around the failure line in the same log file or open the `.xcresult`.
+- Run the Cloudflare Worker tests via `bash Scripts/test-workers.sh` (Node's built-in test runner, no dependencies; suite lives in `Tests/Workers/`). These are fast — no capture-to-file pattern needed.
diff --git a/Scripts/test-workers.sh b/Scripts/test-workers.sh
@@ -0,0 +1,9 @@
+#!/bin/bash
+set -euo pipefail
+
+# Runs the Cloudflare Worker unit tests with Node's built-in test runner (no
+# dependencies to install). The tests import the worker modules directly and
+# stub the Durable Object storage surface, so they cover the workers' pure
+# logic — not the Cloudflare runtime itself.
+cd "$(dirname "${BASH_SOURCE[0]}")/.."
+node --test "Tests/Workers/*.test.mjs"
diff --git a/Tests/Workers/helpers.mjs b/Tests/Workers/helpers.mjs
@@ -0,0 +1,52 @@
+// Map-backed stand-in for a Durable Object's transactional storage, covering
+// only the surface the rate limiters touch (get/put/delete/list + alarms).
+// Tests import the worker modules directly and construct the classes against
+// this stub, so they exercise the pure logic — not the workers runtime.
+export class StubStorage {
+ constructor() {
+ this.map = new Map();
+ this.alarmAt = null;
+ }
+
+ async get(key) {
+ return this.map.get(key);
+ }
+
+ async put(key, value) {
+ this.map.set(key, value);
+ }
+
+ async delete(key) {
+ this.map.delete(key);
+ }
+
+ async list({ prefix }) {
+ return new Map([...this.map].filter(([key]) => key.startsWith(prefix)));
+ }
+
+ async getAlarm() {
+ return this.alarmAt;
+ }
+
+ async setAlarm(at) {
+ this.alarmAt = at;
+ }
+
+ keys(prefix = "") {
+ return [...this.map.keys()].filter((key) => key.startsWith(prefix));
+ }
+
+ // Rewinds every timestamp in a stored rate entry by `millis`, simulating
+ // the passage of time without the test having to wait for it. Handles both
+ // rate-value shapes: plain timestamp arrays (push worker) and
+ // `{room, at}` entry arrays (room worker).
+ age(key, millis) {
+ const stored = this.map.get(key);
+ this.map.set(
+ key,
+ stored.map((entry) =>
+ typeof entry === "number" ? entry - millis : { ...entry, at: entry.at - millis }
+ )
+ );
+ }
+}
diff --git a/Tests/Workers/push-worker.test.mjs b/Tests/Workers/push-worker.test.mjs
@@ -0,0 +1,122 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import { PushRegistry } from "../../Workers/push-worker.js";
+import { StubStorage } from "./helpers.mjs";
+
+function makeRegistry(env = {}) {
+ const storage = new StubStorage();
+ return { registry: new PushRegistry({ storage }, env), storage };
+}
+
+function attestRequest(ip) {
+ return new Request("https://push.example/attest/register", {
+ method: "POST",
+ headers: { "CF-Connecting-IP": ip }
+ });
+}
+
+test("rate limit config uses table defaults", () => {
+ const { registry } = makeRegistry();
+ assert.deepEqual(registry.rateLimitConfig("PUBLISH_CRED"), { limit: 60, windowSeconds: 60 });
+ assert.deepEqual(registry.rateLimitConfig("APP_ATTEST_REGISTER_IP"), { limit: 30, windowSeconds: 60 * 60 });
+});
+
+test("rate limit config honours env overrides and rejects junk", () => {
+ const { registry } = makeRegistry({
+ PUBLISH_CRED_LIMIT: "5",
+ PUBLISH_CRED_WINDOW_SECONDS: "120",
+ PUBLISH_ADDRESS_LIMIT: "not-a-number"
+ });
+ assert.deepEqual(registry.rateLimitConfig("PUBLISH_CRED"), { limit: 5, windowSeconds: 120 });
+ // A malformed override falls back to the table default rather than NaN.
+ assert.equal(registry.rateLimitConfig("PUBLISH_ADDRESS").limit, 30);
+});
+
+test("requests beyond the limit are rejected with a retry hint", async () => {
+ const { registry } = makeRegistry();
+ const config = registry.rateLimitConfig("PUBLISH_CRED");
+ for (let i = 0; i < config.limit; i++) {
+ const result = await registry.checkRateLimit("publish:cred", "cred-a", config);
+ assert.equal(result.ok, true, `call ${i + 1} should be allowed`);
+ }
+ const over = await registry.checkRateLimit("publish:cred", "cred-a", config);
+ assert.equal(over.ok, false);
+ assert.ok(over.retryAfterSeconds >= 1);
+});
+
+test("the window slides: aged entries free quota again", async () => {
+ const { registry, storage } = makeRegistry({ PUBLISH_CRED_LIMIT: "2" });
+ const config = registry.rateLimitConfig("PUBLISH_CRED");
+ await registry.checkRateLimit("publish:cred", "cred-a", config);
+ await registry.checkRateLimit("publish:cred", "cred-a", config);
+ assert.equal((await registry.checkRateLimit("publish:cred", "cred-a", config)).ok, false);
+
+ storage.age(storage.keys("rate:")[0], config.windowSeconds * 1000 + 1000);
+ assert.equal((await registry.checkRateLimit("publish:cred", "cred-a", config)).ok, true);
+});
+
+test("identities have independent buckets", async () => {
+ const { registry } = makeRegistry({ PUBLISH_CRED_LIMIT: "1" });
+ const config = registry.rateLimitConfig("PUBLISH_CRED");
+ assert.equal((await registry.checkRateLimit("publish:cred", "cred-a", config)).ok, true);
+ assert.equal((await registry.checkRateLimit("publish:cred", "cred-a", config)).ok, false);
+ assert.equal((await registry.checkRateLimit("publish:cred", "cred-b", config)).ok, true);
+});
+
+test("attestation gate returns a 429 response once the device limit is hit", async () => {
+ const { registry } = makeRegistry({ APP_ATTEST_REGISTER_DEVICE_LIMIT: "2" });
+ const request = attestRequest("203.0.113.9");
+ assert.equal(await registry.checkAttestationRateLimit(request, "register", "device-1"), null);
+ assert.equal(await registry.checkAttestationRateLimit(request, "register", "device-1"), null);
+ const limited = await registry.checkAttestationRateLimit(request, "register", "device-1");
+ assert.equal(limited.status, 429);
+ assert.ok(Number(limited.headers.get("Retry-After")) >= 1);
+});
+
+test("first write arms the sweep alarm at the horizon; later writes leave it", async () => {
+ const { registry, storage } = makeRegistry();
+ const config = registry.rateLimitConfig("PUBLISH_CRED");
+ const before = Date.now();
+ await registry.checkRateLimit("publish:cred", "cred-a", config);
+ const horizon = registry.rateSweepHorizonMillis();
+ // The default horizon is the largest table window: the 1h attest buckets.
+ assert.equal(horizon, 60 * 60 * 1000);
+ assert.ok(storage.alarmAt >= before + horizon);
+
+ const armedAt = storage.alarmAt;
+ await registry.checkRateLimit("publish:cred", "cred-b", config);
+ assert.equal(storage.alarmAt, armedAt, "arm-if-unarmed must not reschedule");
+});
+
+test("an env window override raises the sweep horizon", () => {
+ const { registry } = makeRegistry({ APP_ATTEST_REGISTER_IP_WINDOW_SECONDS: "7200" });
+ assert.equal(registry.rateSweepHorizonMillis(), 7200 * 1000);
+});
+
+test("sweep prunes only stale rate keys and re-arms while live ones remain", async () => {
+ const { registry, storage } = makeRegistry();
+ const config = registry.rateLimitConfig("PUBLISH_CRED");
+ await registry.checkRateLimit("publish:cred", "cred-a", config);
+ await registry.checkRateLimit("publish:cred", "cred-b", config);
+ const [staleKey, freshKey] = storage.keys("rate:");
+ storage.age(staleKey, registry.rateSweepHorizonMillis() + 60 * 1000);
+ await storage.put("gamecred:keep-me", { secret: "s" });
+
+ await registry.alarm();
+ assert.equal(storage.map.has(staleKey), false);
+ assert.equal(storage.map.has(freshKey), true);
+ assert.equal(storage.map.has("gamecred:keep-me"), true, "sweep must stay inside the rate: prefix");
+ assert.ok(storage.alarmAt > Date.now(), "alarm re-arms while live keys remain");
+});
+
+test("sweep goes quiet once every rate key has expired", async () => {
+ const { registry, storage } = makeRegistry();
+ const config = registry.rateLimitConfig("PUBLISH_CRED");
+ await registry.checkRateLimit("publish:cred", "cred-a", config);
+ storage.age(storage.keys("rate:")[0], registry.rateSweepHorizonMillis() + 60 * 1000);
+ storage.alarmAt = null;
+
+ await registry.alarm();
+ assert.deepEqual(storage.keys("rate:"), []);
+ assert.equal(storage.alarmAt, null, "no live keys, no re-arm");
+});
diff --git a/Tests/Workers/room-worker.test.mjs b/Tests/Workers/room-worker.test.mjs
@@ -0,0 +1,89 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import { EngagementRegisterLimiter } from "../../Workers/room-worker.js";
+import { StubStorage } from "./helpers.mjs";
+
+function makeLimiter(env = {}) {
+ const storage = new StubStorage();
+ return { limiter: new EngagementRegisterLimiter({ storage }, env), storage };
+}
+
+function registerRequest(roomID, ip = "203.0.113.9") {
+ const url = new URL("https://room.example/");
+ if (roomID) url.searchParams.set("roomID", roomID);
+ return new Request(url, { headers: { "CF-Connecting-IP": ip } });
+}
+
+test("a register probe without a room ID is a bad request", async () => {
+ const { limiter } = makeLimiter();
+ const response = await limiter.fetch(new Request("https://room.example/"));
+ assert.equal(response.status, 400);
+});
+
+test("distinct rooms are limited per IP, with a retry hint", async () => {
+ const { limiter } = makeLimiter({ ROOM_REGISTER_IP_LIMIT: "2" });
+ assert.equal((await limiter.fetch(registerRequest("room-1"))).status, 204);
+ assert.equal((await limiter.fetch(registerRequest("room-2"))).status, 204);
+ const limited = await limiter.fetch(registerRequest("room-3"));
+ assert.equal(limited.status, 429);
+ assert.ok(Number(limited.headers.get("Retry-After")) >= 1);
+});
+
+test("re-registering the same room is idempotent and consumes no quota", async () => {
+ const { limiter } = makeLimiter({ ROOM_REGISTER_IP_LIMIT: "2" });
+ for (let i = 0; i < 5; i++) {
+ assert.equal((await limiter.fetch(registerRequest("room-1"))).status, 204, `re-register ${i + 1}`);
+ }
+ // The window still has one fresh-room slot left.
+ assert.equal((await limiter.fetch(registerRequest("room-2"))).status, 204);
+ assert.equal((await limiter.fetch(registerRequest("room-3"))).status, 429);
+});
+
+test("IPs have independent budgets", async () => {
+ const { limiter } = makeLimiter({ ROOM_REGISTER_IP_LIMIT: "1" });
+ assert.equal((await limiter.fetch(registerRequest("room-1", "203.0.113.9"))).status, 204);
+ assert.equal((await limiter.fetch(registerRequest("room-2", "203.0.113.9"))).status, 429);
+ assert.equal((await limiter.fetch(registerRequest("room-2", "198.51.100.7"))).status, 204);
+});
+
+test("the window slides: an aged room entry frees its slot", async () => {
+ const { limiter, storage } = makeLimiter({
+ ROOM_REGISTER_IP_LIMIT: "1",
+ ROOM_REGISTER_IP_WINDOW_SECONDS: "60"
+ });
+ assert.equal((await limiter.fetch(registerRequest("room-1"))).status, 204);
+ assert.equal((await limiter.fetch(registerRequest("room-2"))).status, 429);
+
+ storage.age(storage.keys("rate:")[0], 61 * 1000);
+ assert.equal((await limiter.fetch(registerRequest("room-2"))).status, 204);
+});
+
+test("registering arms the sweep alarm once", async () => {
+ const { limiter, storage } = makeLimiter({ ROOM_REGISTER_IP_WINDOW_SECONDS: "60" });
+ const before = Date.now();
+ await limiter.fetch(registerRequest("room-1"));
+ assert.ok(storage.alarmAt >= before + 60 * 1000);
+
+ const armedAt = storage.alarmAt;
+ await limiter.fetch(registerRequest("room-2"));
+ assert.equal(storage.alarmAt, armedAt, "arm-if-unarmed must not reschedule");
+});
+
+test("sweep prunes stale IP keys, keeps live ones, and re-arms accordingly", async () => {
+ const { limiter, storage } = makeLimiter({ ROOM_REGISTER_IP_WINDOW_SECONDS: "60" });
+ await limiter.fetch(registerRequest("room-1", "203.0.113.9"));
+ await limiter.fetch(registerRequest("room-2", "198.51.100.7"));
+ const [staleKey, liveKey] = storage.keys("rate:");
+ storage.age(staleKey, 61 * 1000);
+
+ await limiter.alarm();
+ assert.equal(storage.map.has(staleKey), false);
+ assert.equal(storage.map.has(liveKey), true);
+ assert.ok(storage.alarmAt > Date.now(), "alarm re-arms while live keys remain");
+
+ storage.age(liveKey, 61 * 1000);
+ storage.alarmAt = null;
+ await limiter.alarm();
+ assert.deepEqual(storage.keys("rate:"), []);
+ assert.equal(storage.alarmAt, null, "no live keys, no re-arm");
+});
diff --git a/Workers/push-worker.js b/Workers/push-worker.js
@@ -1,3 +1,15 @@
+// Single source for every rate bucket's defaults (env-overridable via
+// `<PREFIX>_LIMIT` / `<PREFIX>_WINDOW_SECONDS`), shared by the enforcement
+// call sites and the sweep horizon so the two cannot drift.
+const RATE_LIMIT_DEFAULTS = {
+ APP_ATTEST_CHALLENGE_IP: { limit: 60, windowSeconds: 60 * 60 },
+ APP_ATTEST_CHALLENGE_DEVICE: { limit: 10, windowSeconds: 5 * 60 },
+ APP_ATTEST_REGISTER_IP: { limit: 30, windowSeconds: 60 * 60 },
+ APP_ATTEST_REGISTER_DEVICE: { limit: 5, windowSeconds: 10 * 60 },
+ PUBLISH_CRED: { limit: 60, windowSeconds: 60 },
+ PUBLISH_ADDRESS: { limit: 30, windowSeconds: 60 }
+};
+
export class PushRegistry {
constructor(state, env) {
this.state = state;
@@ -506,11 +518,11 @@ export class PushRegistry {
async checkAttestationRateLimit(request, endpoint, deviceID) {
const ip = request.headers.get("CF-Connecting-IP") || "unknown";
const ipLimit = endpoint === "register"
- ? this.rateLimitConfig("APP_ATTEST_REGISTER_IP", 30, 60 * 60)
- : this.rateLimitConfig("APP_ATTEST_CHALLENGE_IP", 60, 60 * 60);
+ ? this.rateLimitConfig("APP_ATTEST_REGISTER_IP")
+ : this.rateLimitConfig("APP_ATTEST_CHALLENGE_IP");
const deviceLimit = endpoint === "register"
- ? this.rateLimitConfig("APP_ATTEST_REGISTER_DEVICE", 5, 10 * 60)
- : this.rateLimitConfig("APP_ATTEST_CHALLENGE_DEVICE", 10, 5 * 60);
+ ? this.rateLimitConfig("APP_ATTEST_REGISTER_DEVICE")
+ : this.rateLimitConfig("APP_ATTEST_CHALLENGE_DEVICE");
const ipResult = await this.checkRateLimit(`attest:${endpoint}:ip`, ip, ipLimit);
if (!ipResult.ok) return rateLimitedResponse(ipResult);
@@ -524,7 +536,7 @@ export class PushRegistry {
const result = await this.checkRateLimit(
"publish:cred",
credID,
- this.rateLimitConfig("PUBLISH_CRED", 60, 60)
+ this.rateLimitConfig("PUBLISH_CRED")
);
return result.ok ? null : rateLimitedResponse(result);
}
@@ -541,21 +553,22 @@ export class PushRegistry {
const result = await this.checkRateLimit(
"publish:address",
address,
- this.rateLimitConfig("PUBLISH_ADDRESS", 30, 60)
+ this.rateLimitConfig("PUBLISH_ADDRESS")
);
if (!result.ok) return rateLimitedResponse(result);
}
return null;
}
- rateLimitConfig(prefix, defaultLimit, defaultWindowSeconds) {
- const limit = Number(this.env[`${prefix}_LIMIT`] || String(defaultLimit));
- const windowSeconds = Number(this.env[`${prefix}_WINDOW_SECONDS`] || String(defaultWindowSeconds));
+ rateLimitConfig(prefix) {
+ const defaults = RATE_LIMIT_DEFAULTS[prefix];
+ const limit = Number(this.env[`${prefix}_LIMIT`] || String(defaults.limit));
+ const windowSeconds = Number(this.env[`${prefix}_WINDOW_SECONDS`] || String(defaults.windowSeconds));
return {
- limit: Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : defaultLimit,
+ limit: Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : defaults.limit,
windowSeconds: Number.isFinite(windowSeconds) && windowSeconds > 0
? Math.floor(windowSeconds)
- : defaultWindowSeconds
+ : defaults.windowSeconds
};
}
@@ -571,13 +584,57 @@ export class PushRegistry {
if (recent.length >= config.limit) {
const retryAfterSeconds = Math.max(1, Math.ceil((recent[0] + windowMillis - now) / 1000));
await this.state.storage.put(key, recent);
+ await this.ensureRateSweepScheduled();
return { ok: false, retryAfterSeconds };
}
recent.push(now);
await this.state.storage.put(key, recent);
+ await this.ensureRateSweepScheduled();
return { ok: true };
}
+ // Rate keys are written per rotatable identity (IP, deviceID, credID,
+ // address) and would otherwise persist forever once that identity stops
+ // appearing. Arm-if-unarmed keeps the sweep at most one per horizon even
+ // under constant traffic — the same pattern as the room worker's
+ // EngagementRegisterLimiter.
+ async ensureRateSweepScheduled() {
+ const scheduled = await this.state.storage.getAlarm();
+ if (scheduled === null) {
+ await this.state.storage.setAlarm(Date.now() + this.rateSweepHorizonMillis());
+ }
+ }
+
+ // The sweep horizon is the largest configured window across all buckets: a
+ // key untouched for that long has expired in every bucket, whatever its own
+ // window, so one horizon safely serves keys whose bucket (and window) can't
+ // be recovered from the HMAC-digested storage key.
+ rateSweepHorizonMillis() {
+ const windows = Object.keys(RATE_LIMIT_DEFAULTS).map(
+ (prefix) => this.rateLimitConfig(prefix).windowSeconds
+ );
+ return Math.max(...windows) * 1000;
+ }
+
+ async alarm() {
+ const cutoff = Date.now() - this.rateSweepHorizonMillis();
+ const entries = await this.state.storage.list({ prefix: "rate:" });
+ let liveKeys = 0;
+ for (const [key, stored] of entries) {
+ const newest = Array.isArray(stored)
+ ? stored.reduce((max, timestamp) => (typeof timestamp === "number" && timestamp > max ? timestamp : max), 0)
+ : 0;
+ if (newest <= cutoff) {
+ await this.state.storage.delete(key);
+ } else {
+ liveKeys += 1;
+ }
+ }
+ if (liveKeys > 0) {
+ await this.state.storage.setAlarm(Date.now() + this.rateSweepHorizonMillis());
+ }
+ }
+
rateLimitKeySecret() {
return this.env.RATE_LIMIT_HASH_KEY || this.env.APNS_KEY || "crossmate-rate-limit-v1";
}
