From 12b28ddcad20b9133b8ce5162f06a137ddfab822 Mon Sep 17 00:00:00 2001 From: mrbobbytables Date: Mon, 2 Nov 2026 15:32:19 +0000 Subject: [PATCH] fix(nps-relay): release reservations on pre-commit failures Implement approved option A with structured request-owned compensation and a table-driven failure gate. Preserve committed quotas and surface cleanup failures without losing the original error. Hive-Run: hivecommons/docs#561 Hive-Plan: issue-561-option-A Hive-Spec: issue-561#post-reservation-cleanup Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: mrbobbytables --- .../fixed-561-nps-reservation-cleanup.md | 2 + netlify/nps-relay/relay.failure-paths.test.ts | 439 +++++++++++++++++- netlify/nps-relay/relay.ts | 228 +++++---- 3 files changed, 560 insertions(+), 109 deletions(-) create mode 100644 changelog.d/fixed-561-nps-reservation-cleanup.md diff --git a/changelog.d/fixed-561-nps-reservation-cleanup.md b/changelog.d/fixed-561-nps-reservation-cleanup.md new file mode 100644 index 00000000..4014d242 --- /dev/null +++ b/changelog.d/fixed-561-nps-reservation-cleanup.md @@ -0,0 +1,2 @@ +- Release NPS relay rate-limit reservations when registration or submission fails + before commit, while retaining quotas for committed operations. diff --git a/netlify/nps-relay/relay.failure-paths.test.ts b/netlify/nps-relay/relay.failure-paths.test.ts index 4d6e252a..c0dfbcda 100644 --- a/netlify/nps-relay/relay.failure-paths.test.ts +++ b/netlify/nps-relay/relay.failure-paths.test.ts @@ -5,16 +5,23 @@ import { sign, type KeyObject, } from "node:crypto"; -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ACK_PATH, + CAS_MAX_ATTEMPTS, + ENTRY_PREFIX, HEADER_INSTALL_ID, HEADER_NONCE, HEADER_SIGNATURE, HEADER_TIMESTAMP, INSTALL_PREFIX, + INSTALL_RATE_PREFIX, + IP_RATE_PREFIX, PENDING_PATH, PURPOSE_REGISTER, + PURPOSE_SUBMIT, + REGISTER_DAY_PREFIX, + REGISTER_IP_RATE_PREFIX, REGISTER_PATH, RELAY_BASE_PATH, BodyTooLargeError, @@ -41,6 +48,8 @@ function sha256(value: string): string { /** Minimal in-memory store; every method can be overridden per test. */ class MemoryStore implements RelayStore { data = new Map(); + private etags = new Map(); + private version = 0; async get(key: string): Promise { return this.data.has(key) @@ -52,7 +61,7 @@ class MemoryStore implements RelayStore { key: string ): Promise<{ data: unknown; etag?: string } | null> { const data = await this.get(key); - return data === null ? null : { data, etag: '"v1"' }; + return data === null ? null : { data, etag: this.etags.get(key) }; } async setJSON( @@ -61,12 +70,20 @@ class MemoryStore implements RelayStore { options?: ConditionalSetOptions ): Promise { if (options?.onlyIfNew && this.data.has(key)) return { modified: false }; + if ( + options?.onlyIfMatch !== undefined && + options.onlyIfMatch !== this.etags.get(key) + ) + return { modified: false }; + const etag = `"v${++this.version}"`; this.data.set(key, JSON.parse(JSON.stringify(value))); - return { modified: true, etag: '"v2"' }; + this.etags.set(key, etag); + return { modified: true, etag }; } async delete(key: string): Promise { this.data.delete(key); + this.etags.delete(key); } async list({ @@ -113,32 +130,40 @@ function deps(overrides: Partial = {}): RelayDeps { } function registerRequest(hive: Hive): Request { - const body = JSON.stringify({ - install_id: hive.installId, - public_key: hive.publicKey, - }); + return signedRequest(hive, PURPOSE_REGISTER); +} + +function signedRequest(hive: Hive, purpose: string): Request { + const body = JSON.stringify( + purpose === PURPOSE_REGISTER + ? { install_id: hive.installId, public_key: hive.publicKey } + : { hive_id: "failure-test-hive", score: 4 } + ); const ts = String(Math.floor(T0 / MS_PER_SECOND)); const nonce = randomBytes(NONCE_BYTES).toString("hex"); const message = signingInput( - PURPOSE_REGISTER, + purpose, hive.installId, ts, nonce, Buffer.from(body, "utf8") ); - return new Request(ORIGIN + REGISTER_PATH, { - method: "POST", - headers: { - "content-type": "application/json", - [HEADER_INSTALL_ID]: hive.installId, - [HEADER_TIMESTAMP]: ts, - [HEADER_NONCE]: nonce, - [HEADER_SIGNATURE]: sign(null, message, hive.privateKey).toString( - "base64" - ), - }, - body, - }); + return new Request( + ORIGIN + (purpose === PURPOSE_REGISTER ? REGISTER_PATH : RELAY_BASE_PATH), + { + method: "POST", + headers: { + "content-type": "application/json", + [HEADER_INSTALL_ID]: hive.installId, + [HEADER_TIMESTAMP]: ts, + [HEADER_NONCE]: nonce, + [HEADER_SIGNATURE]: sign(null, message, hive.privateKey).toString( + "base64" + ), + }, + body, + } + ); } function hubHeaders(): Record { @@ -339,6 +364,7 @@ describe("registration race loser", () => { public_key: winner.publicKey, registered_at: new Date(T0).toISOString(), }); + return { modified: false }; } return super.setJSON(k, value, options); @@ -356,6 +382,14 @@ describe("registration race loser", () => { expect(await racing.get(key)).toMatchObject({ public_key: winner.publicKey, }); + expect( + await racing.get(`${REGISTER_IP_RATE_PREFIX}${sha256("203.0.113.9")}`) + ).toEqual({ times: [] }); + expect( + await racing.get( + `${REGISTER_DAY_PREFIX}${new Date(T0).toISOString().slice(0, 10)}` + ) + ).toEqual({ count: 0 }); }); it("answers 200 registered:false when the write loses to the same key", async () => { @@ -383,5 +417,368 @@ describe("registration race loser", () => { ); expect(res.status).toBe(200); expect(await res.json()).toEqual({ ok: true, registered: false }); + expect( + await racing.get(`${REGISTER_IP_RATE_PREFIX}${sha256("203.0.113.9")}`) + ).toEqual({ times: [] }); + expect( + await racing.get( + `${REGISTER_DAY_PREFIX}${new Date(T0).toISOString().slice(0, 10)}` + ) + ).toEqual({ count: 0 }); }); }); + +describe("post-reservation failure gate", () => { + const ip = "203.0.113.9"; + const dayKey = `${REGISTER_DAY_PREFIX}${new Date(T0).toISOString().slice(0, 10)}`; + const registerIpKey = `${REGISTER_IP_RATE_PREFIX}${sha256(ip)}`; + const submitIpKey = `${IP_RATE_PREFIX}${sha256(ip)}`; + + beforeEach(() => { + vi.spyOn(console, "error").mockImplementation(() => {}); + vi.spyOn(console, "warn").mockImplementation(() => {}); + }); + + afterEach(() => vi.restoreAllMocks()); + + async function arrange(purpose: string) { + const hive = newHive(); + const installRateKey = `${INSTALL_RATE_PREFIX}${hive.installId}`; + // Seed existing quota records so the gate compares their exact JSON, not + // just counts. New empty records are harmless and cannot be safely deleted + // without a conditional-delete API. + await store.setJSON(registerIpKey, { times: [T0 - 20] }); + await store.setJSON(dayKey, { count: 7 }); + await store.setJSON(submitIpKey, { times: [] }); + await store.setJSON(installRateKey, { times: [T0 - 10] }); + if (purpose === PURPOSE_SUBMIT) { + await store.setJSON(`${INSTALL_PREFIX}${hive.installId}`, { + public_key: hive.publicKey, + }); + } + return { + hive, + installRateKey, + secondKey: purpose === PURPOSE_REGISTER ? dayKey : installRateKey, + commitPrefix: + purpose === PURPOSE_REGISTER ? INSTALL_PREFIX : ENTRY_PREFIX, + }; + } + + function rateSnapshot() { + return [...store.data] + .filter(([key]) => key.startsWith("rate/")) + .sort(([a], [b]) => a.localeCompare(b)) + .map(([key, value]) => [key, JSON.stringify(value)]); + } + + const failurePoints = [ + { + step: "second reservation read", + method: "getWithMetadata", + mode: "throw", + status: 500, + }, + { + step: "second reservation write", + method: "setJSON", + mode: "throw", + status: 500, + }, + { + step: "second reservation contention", + method: "setJSON", + mode: "refuse", + status: 429, + }, + { + step: "commit write", + method: "setJSON", + mode: "throw", + status: 500, + }, + { + step: "commit refusal", + method: "setJSON", + mode: "refuse", + status: 500, + }, + ] as const; + + it.each( + [PURPOSE_REGISTER, PURPOSE_SUBMIT].flatMap(purpose => + failurePoints.map(point => ({ purpose, ...point })) + ) + )( + "$purpose: refunds exact quotas after $step", + async ({ purpose, step, method, mode, status }) => { + const { hive, secondKey, commitPrefix } = await arrange(purpose); + const before = rateSnapshot(); + const isCommit = step.startsWith("commit"); + const target = isCommit ? commitPrefix : secondKey; + let injected = 0; + if (method === "getWithMetadata") { + const read = store.getWithMetadata.bind(store); + vi.spyOn(store, method).mockImplementation(async key => { + if (key === target && injected++ === 0) + throw new Error("read failed"); + return read(key); + }); + } else { + const write = store.setJSON.bind(store); + vi.spyOn(store, method).mockImplementation( + async (key, value, options) => { + if ( + key.startsWith(target) && + injected < (mode === "refuse" ? CAS_MAX_ATTEMPTS : 1) + ) { + injected++; + if (mode === "throw") throw new Error("write failed"); + return { modified: false }; + } + return write(key, value, options); + } + ); + } + const res = await handleRelayRequest( + signedRequest(hive, purpose), + deps() + ); + expect(injected).toBeGreaterThan(0); + expect(res.status).toBe( + purpose === PURPOSE_REGISTER && step === "commit refusal" ? 409 : status + ); + expect(rateSnapshot()).toEqual(before); + expect( + [...store.data.keys()].filter(key => key.startsWith(commitPrefix)) + ).toHaveLength(0); + + vi.restoreAllMocks(); + // Submit retries use a fresh signed nonce; replay protection is unchanged. + const retry = await handleRelayRequest( + signedRequest(hive, purpose), + deps() + ); + expect(retry.status).toBe(201); + } + ); + + it("refunds both registration reservations if reading the race winner throws", async () => { + const { hive } = await arrange(PURPOSE_REGISTER); + const before = rateSnapshot(); + const key = `${INSTALL_PREFIX}${hive.installId}`; + const write = store.setJSON.bind(store); + let lost = false; + vi.spyOn(store, "setJSON").mockImplementation(async (k, value, options) => { + if (k === key) { + lost = true; + return { modified: false }; + } + return write(k, value, options); + }); + const read = store.get.bind(store); + vi.spyOn(store, "get").mockImplementation(async k => { + if (k === key && lost) throw new Error("winner lookup failed"); + return read(k); + }); + expect( + (await handleRelayRequest(registerRequest(hive), deps())).status + ).toBe(500); + expect(lost).toBe(true); + expect(rateSnapshot()).toEqual(before); + }); + + it("keeps committed submission quotas if rolling-window listing fails", async () => { + const { hive, installRateKey } = await arrange(PURPOSE_SUBMIT); + vi.spyOn(store, "list").mockRejectedValue(new Error("listing failed")); + expect( + (await handleRelayRequest(signedRequest(hive, PURPOSE_SUBMIT), deps())) + .status + ).toBe(500); + expect(await store.get(submitIpKey)).toEqual({ times: [T0] }); + expect(await store.get(installRateKey)).toEqual({ + times: [T0 - 10, T0], + }); + expect( + [...store.data.keys()].filter(key => key.startsWith(ENTRY_PREFIX)) + ).toHaveLength(1); + expect( + (await handleRelayRequest(signedRequest(hive, PURPOSE_SUBMIT), deps())) + .status + ).toBe(429); + }); + + it("keeps committed registration quotas if constructing the response throws", async () => { + const { hive } = await arrange(PURPOSE_REGISTER); + const stringify = JSON.stringify; + vi.spyOn(JSON, "stringify").mockImplementation((value, ...args) => { + if (value?.ok && value?.registered === true) { + throw new Error("response construction failed"); + } + return stringify(value, ...args); + }); + expect( + (await handleRelayRequest(registerRequest(hive), deps())).status + ).toBe(500); + expect(await store.get(registerIpKey)).toEqual({ + times: [T0 - 20, T0], + }); + expect(await store.get(dayKey)).toEqual({ count: 8 }); + const beforeRetry = rateSnapshot(); + const retry = await handleRelayRequest(registerRequest(hive), deps()); + expect(retry.status).toBe(200); + expect(await retry.json()).toEqual({ + ok: true, + registered: false, + }); + expect(rateSnapshot()).toEqual(beforeRetry); + }); + + it.each([PURPOSE_REGISTER, PURPOSE_SUBMIT])( + "%s: a failed first request leaves empty quotas available for retry", + async purpose => { + const hive = newHive(); + if (purpose === PURPOSE_SUBMIT) { + await store.setJSON(`${INSTALL_PREFIX}${hive.installId}`, { + public_key: hive.publicKey, + }); + } + const prefix = + purpose === PURPOSE_REGISTER ? INSTALL_PREFIX : ENTRY_PREFIX; + const write = store.setJSON.bind(store); + vi.spyOn(store, "setJSON").mockImplementation( + async (key, value, options) => { + if (key.startsWith(prefix)) throw new Error("commit failed"); + return write(key, value, options); + } + ); + expect( + (await handleRelayRequest(signedRequest(hive, purpose), deps())).status + ).toBe(500); + for (const [key, value] of store.data) { + if (key.startsWith("rate/")) { + expect(value).toEqual(key === dayKey ? { count: 0 } : { times: [] }); + } + } + vi.mocked(store.setJSON).mockRestore(); + expect( + (await handleRelayRequest(signedRequest(hive, purpose), deps())).status + ).toBe(201); + } + ); + + it("surfaces cleanup contention instead of returning a normal rate-limit refusal", async () => { + const { hive } = await arrange(PURPOSE_REGISTER); + await store.setJSON(dayKey, { count: 500 }); + const write = store.setJSON.bind(store); + let ipWrites = 0; + vi.spyOn(store, "setJSON").mockImplementation( + async (key, value, options) => { + if (key === registerIpKey && ipWrites++ > 0) return { modified: false }; + return write(key, value, options); + } + ); + expect( + (await handleRelayRequest(registerRequest(hive), deps())).status + ).toBe(500); + expect(console.error).toHaveBeenCalledWith( + "nps relay: internal error:", + "ReservationCleanupError", + "ReservationReleaseContention" + ); + }); + + it.each([PURPOSE_REGISTER, PURPOSE_SUBMIT])( + "%s: retries cleanup CAS without erasing a concurrent successful request", + async purpose => { + const { hive, secondKey, commitPrefix } = await arrange(purpose); + const other = purpose === PURPOSE_REGISTER ? newHive() : hive; + const write = store.setJSON.bind(store); + let failed = false; + let interleaved = false; + vi.spyOn(store, "setJSON").mockImplementation( + async (key, value, options) => { + if (!failed && key.startsWith(commitPrefix)) { + failed = true; + throw new Error("commit failed"); + } + // Interleave after cleanup has read the counter but before its CAS. + if (failed && !interleaved && key === secondKey) { + interleaved = true; + const concurrent = await handleRelayRequest( + signedRequest(other, purpose), + deps({ + clientIp: purpose === PURPOSE_REGISTER ? ip : "203.0.113.10", + }) + ); + expect(concurrent.status).toBe(201); + } + return write(key, value, options); + } + ); + expect( + (await handleRelayRequest(signedRequest(hive, purpose), deps())).status + ).toBe(500); + expect(interleaved).toBe(true); + expect(await store.get(secondKey)).toEqual( + purpose === PURPOSE_REGISTER ? { count: 8 } : { times: [T0 - 10, T0] } + ); + expect( + await store.get( + purpose === PURPOSE_REGISTER ? registerIpKey : submitIpKey + ) + ).toEqual({ + times: purpose === PURPOSE_REGISTER ? [T0 - 20, T0] : [], + }); + expect( + [...store.data.keys()].filter(key => key.startsWith(commitPrefix)) + ).toHaveLength(1); + } + ); + + it.each(["throw", "refuse"] as const)( + "surfaces cleanup %s, preserves the original error, and attempts the other release", + async mode => { + const { hive, installRateKey } = await arrange(PURPOSE_SUBMIT); + const write = store.setJSON.bind(store); + let failed = false; + let cleanupAttempts = 0; + const original = new Error("private commit failure details"); + original.name = "CommitUnavailable"; + const cleanup = new Error("private cleanup failure details"); + cleanup.name = "CleanupUnavailable"; + vi.spyOn(store, "setJSON").mockImplementation( + async (key, value, options) => { + if (key.startsWith(ENTRY_PREFIX)) { + failed = true; + throw original; + } + if (failed && key === installRateKey) { + cleanupAttempts++; + if (mode === "throw") throw cleanup; + return { modified: false }; + } + return write(key, value, options); + } + ); + expect( + (await handleRelayRequest(signedRequest(hive, PURPOSE_SUBMIT), deps())) + .status + ).toBe(500); + expect(cleanupAttempts).toBe(mode === "throw" ? 1 : CAS_MAX_ATTEMPTS); + expect(await store.get(submitIpKey)).toEqual({ times: [] }); + expect(await store.get(installRateKey)).toEqual({ + times: [T0 - 10, T0], + }); + expect(console.error).toHaveBeenCalledWith( + "nps relay: internal error:", + "ReservationCleanupError", + "CommitUnavailable", + mode === "throw" ? "CleanupUnavailable" : "ReservationReleaseContention" + ); + expect(JSON.stringify(vi.mocked(console.error).mock.calls)).not.toContain( + "private" + ); + } + ); +}); diff --git a/netlify/nps-relay/relay.ts b/netlify/nps-relay/relay.ts index 8056e525..5bc9bb17 100644 --- a/netlify/nps-relay/relay.ts +++ b/netlify/nps-relay/relay.ts @@ -585,6 +585,51 @@ function releaseSlot( }); } +type ReleaseReservation = () => Promise; + +class ReservationCleanupError extends AggregateError { + constructor(errors: unknown[], cause: unknown) { + super(errors, "reservation cleanup failed", { cause }); + this.name = "ReservationCleanupError"; + } +} + +/** + * Register compensation only after a successful reservation; clear it at the + * commit write, not at the response, since post-commit errors must not refund. + */ +async function withReservations( + action: (releases: ReleaseReservation[]) => Promise +): Promise { + const releases: ReleaseReservation[] = []; + let outcome: { response: Response } | { error: unknown }; + try { + outcome = { response: await action(releases) }; + } catch (err) { + outcome = { error: err }; + } + const errors: unknown[] = []; + for (const release of releases.reverse()) { + try { + if (await release()) { + const err = new Error("reservation release exhausted CAS retries"); + err.name = "ReservationReleaseContention"; + throw err; + } + } catch (err) { + errors.push(err); + } + } + if (errors.length) { + throw new ReservationCleanupError( + "error" in outcome ? [outcome.error, ...errors] : errors, + "error" in outcome ? outcome.error : undefined + ); + } + if ("error" in outcome) throw outcome.error; + return outcome.response; +} + /** Whether a windowed rate record is already at `max`, without reserving. */ async function atLimit( store: RelayStore, @@ -742,60 +787,59 @@ async function handleRegister( return json(409, { error: "install_id is registered to a different key" }); } - // Reserve the per-IP and per-day slots before creating the install, so a - // limit that is hit never leaves a registration behind. The two writes are - // not atomic together: if the day cap refuses, the IP slot is released so a - // rejected registration does not consume the independent IP quota. + // These reservations are independent; compensate every pre-commit failure. const ipKey = `${REGISTER_IP_RATE_PREFIX}${ipHash(deps)}`; - const ipDenied = await reserveSlot( - store, - ipKey, - MAX_REGISTRATIONS_PER_IP_PER_WINDOW, - now - ); - if (ipDenied) return ipDenied; - const dayKey = `${REGISTER_DAY_PREFIX}${new Date(now).toISOString().slice(0, ISO_DATE_LENGTH)}`; - const dayDenied = await compareAndSwap(store, dayKey, current => { - const count = - current && typeof (current as DayCounter).count === "number" - ? (current as DayCounter).count - : 0; - if (count >= MAX_REGISTRATIONS_PER_DAY) return { reject: RATE_LIMITED() }; - return { next: { count: count + 1 } satisfies DayCounter }; - }); - if (dayDenied) { - await releaseSlot(store, ipKey, now); - return dayDenied; - } - - const record: InstallRecord = { - public_key: reg.public_key, - ...(reg.hive_version ? { hive_version: reg.hive_version } : {}), - registered_at: new Date(now).toISOString(), - }; - // Create-only: two registrations racing for the same install id cannot both - // win, so the id is bound to exactly one key — the first write's. - const created = await store.setJSON( - `${INSTALL_PREFIX}${reg.install_id}`, - record, - { onlyIfNew: true } - ); - if (!created.modified) { - // The registration was not created here, so give back both reservations. - await releaseSlot(store, ipKey, now); - await compareAndSwap(store, dayKey, current => { + return withReservations(async releases => { + const ipDenied = await reserveSlot( + store, + ipKey, + MAX_REGISTRATIONS_PER_IP_PER_WINDOW, + now + ); + if (ipDenied) return ipDenied; + releases.push(() => releaseSlot(store, ipKey, now)); + const dayKey = `${REGISTER_DAY_PREFIX}${new Date(now).toISOString().slice(0, ISO_DATE_LENGTH)}`; + const dayDenied = await compareAndSwap(store, dayKey, current => { const count = current && typeof (current as DayCounter).count === "number" ? (current as DayCounter).count : 0; - return { next: { count: Math.max(0, count - 1) } satisfies DayCounter }; + if (count >= MAX_REGISTRATIONS_PER_DAY) return { reject: RATE_LIMITED() }; + return { next: { count: count + 1 } satisfies DayCounter }; }); - const winner = await getInstall(store, reg.install_id); - if (winner && winner.public_key === reg.public_key) - return json(200, { ok: true, registered: false }); - return json(409, { error: "install_id is registered to a different key" }); - } - return json(201, { ok: true, registered: true }); + if (dayDenied) return dayDenied; + releases.push(() => + compareAndSwap(store, dayKey, current => { + const count = + current && typeof (current as DayCounter).count === "number" + ? (current as DayCounter).count + : 0; + return { next: { count: Math.max(0, count - 1) } satisfies DayCounter }; + }) + ); + + const record: InstallRecord = { + public_key: reg.public_key, + ...(reg.hive_version ? { hive_version: reg.hive_version } : {}), + registered_at: new Date(now).toISOString(), + }; + // Create-only: an install id stays bound to the first write's key. + const created = await store.setJSON( + `${INSTALL_PREFIX}${reg.install_id}`, + record, + { onlyIfNew: true } + ); + if (!created.modified) { + const winner = await getInstall(store, reg.install_id); + if (winner && winner.public_key === reg.public_key) + return json(200, { ok: true, registered: false }); + return json(409, { + error: "install_id is registered to a different key", + }); + } + releases.length = 0; + return json(201, { ok: true, registered: true }); + }); } /** @@ -864,46 +908,49 @@ async function handleSubmit( if (await atLimit(store, installKey, MAX_PER_INSTALL_PER_WINDOW, now)) return RATE_LIMITED(); - // Reserve each slot with its own conditional write before storing the - // entry, so concurrent submissions cannot all squeeze into the same - // remaining slot. The two writes are not atomic together: if the install - // reservation is refused, the IP slot is released so a rejected submission - // does not consume the independent IP quota. - const ipDenied = await reserveSlot(store, ipKey, MAX_PER_IP_PER_WINDOW, now); - if (ipDenied) return ipDenied; - const installDenied = await reserveSlot( - store, - installKey, - MAX_PER_INSTALL_PER_WINDOW, - now - ); - if (installDenied) { - await releaseSlot(store, ipKey, now); - return installDenied; - } - - const entry: RelayEntry = { - id: newEntryId(now), - install_id: signed.installId, - hive_id: parsed.value.hive_id, - score: parsed.value.score, - ...(parsed.value.feedback ? { feedback: parsed.value.feedback } : {}), - ...(parsed.value.dashboard_version - ? { dashboard_version: parsed.value.dashboard_version } - : {}), - timestamp: new Date(now).toISOString(), - }; - await store.setJSON(`${ENTRY_PREFIX}${entry.id}`, entry); - - // Rolling window: drop the oldest entries beyond the cap. - const keys = await entryKeysOldestFirst(store); - const overflow = keys.length - MAX_ENTRIES; - if (overflow > 0) { - await Promise.allSettled( - keys.slice(0, overflow).map(key => store.delete(key)) + return withReservations(async releases => { + const ipDenied = await reserveSlot( + store, + ipKey, + MAX_PER_IP_PER_WINDOW, + now ); - } - return json(201, { ok: true }); + if (ipDenied) return ipDenied; + releases.push(() => releaseSlot(store, ipKey, now)); + const installDenied = await reserveSlot( + store, + installKey, + MAX_PER_INSTALL_PER_WINDOW, + now + ); + if (installDenied) return installDenied; + releases.push(() => releaseSlot(store, installKey, now)); + + const entry: RelayEntry = { + id: newEntryId(now), + install_id: signed.installId, + hive_id: parsed.value.hive_id, + score: parsed.value.score, + ...(parsed.value.feedback ? { feedback: parsed.value.feedback } : {}), + ...(parsed.value.dashboard_version + ? { dashboard_version: parsed.value.dashboard_version } + : {}), + timestamp: new Date(now).toISOString(), + }; + const created = await store.setJSON(`${ENTRY_PREFIX}${entry.id}`, entry); + if (!created.modified) return json(500, { error: "internal server error" }); + releases.length = 0; + + // Rolling window: drop the oldest entries beyond the cap. + const keys = await entryKeysOldestFirst(store); + const overflow = keys.length - MAX_ENTRIES; + if (overflow > 0) { + await Promise.allSettled( + keys.slice(0, overflow).map(key => store.delete(key)) + ); + } + return json(201, { ok: true }); + }); } function hubAuthorized(req: Request, env: RelayEnv): Response | null { @@ -1020,7 +1067,12 @@ async function routeRelayRequest( // Never echo request data (signatures and secrets live in headers) into logs. console.error( "nps relay: internal error:", - err instanceof Error ? err.name : "unknown" + err instanceof Error ? err.name : "unknown", + ...(err instanceof ReservationCleanupError + ? err.errors.map(cause => + cause instanceof Error ? cause.name : "unknown" + ) + : []) ); return json(500, { error: "internal server error" }); }