From a0fc75911acbadda5fe51d218b574171d38221c0 Mon Sep 17 00:00:00 2001 From: ABHAY PANDEY Date: Tue, 13 Oct 2026 12:38:19 +0530 Subject: [PATCH 1/2] feat(nip-01): send rate-limited backoff hint on CLOSED (#820) --- .changeset/closed-rate-limited-hint.md | 7 +++ src/@types/messages.ts | 1 + src/adapters/web-socket-adapter.ts | 35 +++++++++-- src/utils/messages.ts | 5 +- test/unit/adapters/web-socket-adapter.spec.ts | 61 +++++++++++++------ test/unit/utils/messages.spec.ts | 20 ++++++ 6 files changed, 105 insertions(+), 24 deletions(-) create mode 100644 .changeset/closed-rate-limited-hint.md diff --git a/.changeset/closed-rate-limited-hint.md b/.changeset/closed-rate-limited-hint.md new file mode 100644 index 000000000..706d2d32a --- /dev/null +++ b/.changeset/closed-rate-limited-hint.md @@ -0,0 +1,7 @@ +--- +"nostream": minor +--- + +feat(nip-01): send the `rate-limited` backoff hint on CLOSED + +When the per-IP message rate limit trips on a `REQ` or `COUNT`, the relay now replies `["CLOSED", , "rate-limited: slow down", ""]` with the tripped window's period as the backoff, instead of a `NOTICE`. Other message types keep the existing `NOTICE`. diff --git a/src/@types/messages.ts b/src/@types/messages.ts index 3ccffdf3d..591d5d94c 100644 --- a/src/@types/messages.ts +++ b/src/@types/messages.ts @@ -90,6 +90,7 @@ export interface ClosedMessage { 0: MessageType.CLOSED 1: SubscriptionId 2: string + 3?: string } // NIP-42 diff --git a/src/adapters/web-socket-adapter.ts b/src/adapters/web-socket-adapter.ts index cd32d7a9a..0cffc5f76 100644 --- a/src/adapters/web-socket-adapter.ts +++ b/src/adapters/web-socket-adapter.ts @@ -12,7 +12,7 @@ import { createOutgoingEventMessage, } from '../utils/messages' import { IAbortable, IMessageHandler } from '../@types/message-handlers' -import { IncomingMessage, OutgoingMessage } from '../@types/messages' +import { IncomingMessage, MessageType, OutgoingMessage } from '../@types/messages' import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters' import { SubscriptionFilter, SubscriptionId } from '../@types/subscription' import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter' @@ -200,8 +200,9 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter let abortable = false let messageHandler: (IMessageHandler & IAbortable) | undefined = undefined try { - if (await this.isRateLimited(this.clientAddress.address)) { - this.sendMessage(createNoticeMessage('rate limited')) + const retryAfterMs = await this.isRateLimited(this.clientAddress.address) + if (retryAfterMs) { + this.sendMessage(this.createRateLimitedMessage(raw, retryAfterMs)) return } @@ -256,7 +257,11 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter } } - private async isRateLimited(client: string): Promise { + /** + * Resolves to false when allowed; otherwise to the tripped window's period in ms, + * sent as the NIP-01 `rate-limited` backoff. + */ + private async isRateLimited(client: string): Promise { const { rateLimits, ipWhitelist = [] } = this.settings().limits?.message ?? {} if (!Array.isArray(rateLimits) || !rateLimits.length || ipWhitelist.includes(client)) { @@ -273,13 +278,33 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter if (isRateLimited) { logger('rate limited %s: %d messages / %d ms exceeded', client, rate, period) - return true + return Math.max(1, Math.ceil(period)) } } return false } + // NIP-01: reads (REQ, COUNT) are refused with CLOSED; other messages keep the NOTICE. + private createRateLimitedMessage(raw: Buffer, retryAfterMs: number): OutgoingMessage { + let message: unknown + try { + message = JSON.parse(raw.toString('utf8')) + } catch { + return createNoticeMessage('rate limited') + } + + if ( + Array.isArray(message) && + (message[0] === MessageType.REQ || message[0] === MessageType.COUNT) && + typeof message[1] === 'string' + ) { + return createClosedMessage(message[1], 'rate-limited: slow down', String(retryAfterMs)) + } + + return createNoticeMessage('rate limited') + } + private onClientPong() { debugHeartbeat('client %s pong', this.clientId) this.alive = true diff --git a/src/utils/messages.ts b/src/utils/messages.ts index c6f951347..3c2544421 100644 --- a/src/utils/messages.ts +++ b/src/utils/messages.ts @@ -40,8 +40,9 @@ export const createCountResultMessage = (queryId: SubscriptionId, payload: Count return [MessageType.COUNT, queryId, payload] } -export const createClosedMessage = (queryId: SubscriptionId, reason: string): ClosedMessage => { - return [MessageType.CLOSED, queryId, reason] +// NIP-01 optional params (e.g. rate-limited backoff in ms) +export const createClosedMessage = (queryId: SubscriptionId, reason: string, params?: string): ClosedMessage => { + return params === undefined ? [MessageType.CLOSED, queryId, reason] : [MessageType.CLOSED, queryId, reason, params] } // NIP-42 diff --git a/test/unit/adapters/web-socket-adapter.spec.ts b/test/unit/adapters/web-socket-adapter.spec.ts index fd09b2bb8..5dfb39057 100644 --- a/test/unit/adapters/web-socket-adapter.spec.ts +++ b/test/unit/adapters/web-socket-adapter.spec.ts @@ -556,31 +556,58 @@ describe('WebSocketAdapter', () => { expect(sent[0]).to.equal('NOTICE') }) - it('sends rate-limited notice when rate limited', async () => { - client.readyState = WebSocket.OPEN + describe('when rate limited', () => { + let onMessage: (raw: Buffer) => Promise - // Configure rate limiting to be active - settingsFactory.returns({ - network: { remoteIpHeader: '' }, - limits: { - message: { - rateLimits: [{ period: 60000, rate: 1 }], - ipWhitelist: [], + const sentFrame = () => JSON.parse(client.send.firstCall.args[0]) + + beforeEach(() => { + client.readyState = WebSocket.OPEN + + settingsFactory.returns({ + network: { remoteIpHeader: '' }, + limits: { + message: { + rateLimits: [{ period: 60000, rate: 1 }], + ipWhitelist: [], + }, }, - }, + }) + + slidingWindowRateLimiter.returns({ + hit: sandbox.stub().resolves(true), + }) + + onMessage = client.on.getCalls().find((call: any) => call.args[0] === 'message').args[1] + }) + + it('sends a rate-limited notice for non-read messages', async () => { + await onMessage(Buffer.from(JSON.stringify(['EVENT', {}]))) + + expect(client.send).to.have.been.calledOnce + expect(sentFrame()).to.deep.equal(['NOTICE', 'rate limited']) }) - slidingWindowRateLimiter.returns({ - hit: sandbox.stub().resolves(true), + it('closes a REQ with the tripped window as backoff hint', async () => { + await onMessage(Buffer.from(JSON.stringify(['REQ', 'sub1', { kinds: [1] }]))) + + expect(client.send).to.have.been.calledOnce + expect(sentFrame()).to.deep.equal(['CLOSED', 'sub1', 'rate-limited: slow down', '60000']) }) - const messageCall = client.on.getCalls().find((call: any) => call.args[0] === 'message') - const onMessage = messageCall.args[1] + it('closes a COUNT with the tripped window as backoff hint', async () => { + await onMessage(Buffer.from(JSON.stringify(['COUNT', 'q1', { kinds: [1] }]))) - // Valid JSON message that would pass parsing - await onMessage(Buffer.from(JSON.stringify(['EVENT', {}]))) + expect(client.send).to.have.been.calledOnce + expect(sentFrame()).to.deep.equal(['CLOSED', 'q1', 'rate-limited: slow down', '60000']) + }) - expect(client.send).to.have.been.called + it('falls back to the notice when the frame is not valid JSON', async () => { + await onMessage(Buffer.from('not-json')) + + expect(client.send).to.have.been.calledOnce + expect(sentFrame()).to.deep.equal(['NOTICE', 'rate limited']) + }) }) it('stops hitting subsequent rate limit windows once one is exceeded', async () => { diff --git a/test/unit/utils/messages.spec.ts b/test/unit/utils/messages.spec.ts index 45432c76f..cff5ad242 100644 --- a/test/unit/utils/messages.spec.ts +++ b/test/unit/utils/messages.spec.ts @@ -1,6 +1,7 @@ import { expect } from 'chai' import { + createClosedMessage, createCommandResult, createEndOfStoredEventsNoticeMessage, createNoticeMessage, @@ -72,6 +73,25 @@ describe('createCommandResult', () => { }) }) +describe('createClosedMessage', () => { + it('returns a CLOSED message without params', () => { + expect(createClosedMessage('sub1', 'error: shutting down')).to.deep.equal([ + MessageType.CLOSED, + 'sub1', + 'error: shutting down', + ]) + }) + + it('appends optional params as a fourth element', () => { + expect(createClosedMessage('sub1', 'rate-limited: slow down', '2500')).to.deep.equal([ + MessageType.CLOSED, + 'sub1', + 'rate-limited: slow down', + '2500', + ]) + }) +}) + describe('createSubscriptionMessage', () => { it('returns a REQ message with a single filter', () => { const result = createSubscriptionMessage('sub1', [{ kinds: [1] }]) From d3f047e9556a523de33183908ee542fe3dc5f25c Mon Sep 17 00:00:00 2001 From: ABHAY PANDEY Date: Tue, 13 Oct 2026 12:54:28 +0530 Subject: [PATCH 2/2] fix(nip-01): tear down subscription before rate-limited CLOSED Abort in-flight REQ reads and remove the active subscription when the per-IP message limit closes a subscription, so clients do not keep receiving EVENT/EOSE after CLOSED. --- src/adapters/web-socket-adapter.ts | 29 ++++++++++++++ src/handlers/subscribe-message-handler.ts | 4 ++ test/unit/adapters/web-socket-adapter.spec.ts | 38 +++++++++++++++++++ 3 files changed, 71 insertions(+) diff --git a/src/adapters/web-socket-adapter.ts b/src/adapters/web-socket-adapter.ts index 0cffc5f76..8f107f3c2 100644 --- a/src/adapters/web-socket-adapter.ts +++ b/src/adapters/web-socket-adapter.ts @@ -299,12 +299,41 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter (message[0] === MessageType.REQ || message[0] === MessageType.COUNT) && typeof message[1] === 'string' ) { + if (message[0] === MessageType.REQ) { + this.endSubscription(message[1]) + } + return createClosedMessage(message[1], 'rate-limited: slow down', String(retryAfterMs)) } return createNoticeMessage('rate limited') } + private endSubscription(subscriptionId: SubscriptionId): void { + this.abortInFlightSubscriptionReads(subscriptionId) + if (this.subscriptions.has(subscriptionId)) { + this.onUnsubscribed(subscriptionId) + } + } + + private abortInFlightSubscriptionReads(subscriptionId: SubscriptionId): void { + const handlers = abortableMessageHandlers.get(this.client) + if (!handlers?.length) { + return + } + + for (const handler of handlers) { + const abortable = handler as IAbortable & { activeSubscriptionId?: SubscriptionId } + if (abortable.activeSubscriptionId === subscriptionId && typeof abortable.abort === 'function') { + try { + abortable.abort() + } catch (error) { + logger.error('Unable to abort subscription handler for %s: %o', subscriptionId, error) + } + } + } + } + private onClientPong() { debugHeartbeat('client %s pong', this.clientId) this.alive = true diff --git a/src/handlers/subscribe-message-handler.ts b/src/handlers/subscribe-message-handler.ts index ae881afe8..f79621e95 100644 --- a/src/handlers/subscribe-message-handler.ts +++ b/src/handlers/subscribe-message-handler.ts @@ -31,6 +31,9 @@ const logger = createLogger('subscribe-message-handler') export class SubscribeMessageHandler implements IMessageHandler, IAbortable { private readonly abortController: AbortController + /** Set while `handleMessage` runs; used to abort in-flight reads when the sub is closed early. */ + public activeSubscriptionId: SubscriptionId | undefined + public constructor( private readonly webSocket: IWebSocketAdapter, private readonly eventRepository: IEventRepository, @@ -47,6 +50,7 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { public async handleMessage(message: SubscribeMessage): Promise { const subscriptionId = message[1] + this.activeSubscriptionId = subscriptionId const rawFilters = uniqWith(equals, message.slice(2)) as SubscriptionFilter[] // NIP-50: strip search from filters when disabled so isEventMatchingFilter ignores it diff --git a/test/unit/adapters/web-socket-adapter.spec.ts b/test/unit/adapters/web-socket-adapter.spec.ts index 5dfb39057..b5039996d 100644 --- a/test/unit/adapters/web-socket-adapter.spec.ts +++ b/test/unit/adapters/web-socket-adapter.spec.ts @@ -595,6 +595,44 @@ describe('WebSocketAdapter', () => { expect(sentFrame()).to.deep.equal(['CLOSED', 'sub1', 'rate-limited: slow down', '60000']) }) + it('ends an active subscription and aborts its in-flight read before CLOSED', async () => { + adapter.onSubscribed('sub1', [{ kinds: [1] }]) + + const abort = sandbox.stub() + let unblockFirstReq!: () => void + const firstReqBlocked = new Promise((resolve) => { + unblockFirstReq = resolve + }) + const inFlightHandler = { + activeSubscriptionId: 'sub1', + abort, + handleMessage: sandbox.stub().returns(firstReqBlocked), + } + + const hitStub = sandbox.stub() + hitStub.onFirstCall().resolves(false) + hitStub.resolves(true) + slidingWindowRateLimiter.returns({ hit: hitStub }) + createMessageHandler.returns(inFlightHandler) + + const firstReq = onMessage(Buffer.from(JSON.stringify(['REQ', 'sub1', { kinds: [1] }]))) + await Promise.resolve() + + await onMessage(Buffer.from(JSON.stringify(['REQ', 'sub1', { kinds: [7] }]))) + + expect(abort).to.have.been.calledOnce + expect(adapter.getSubscriptions().has('sub1')).to.equal(false) + expect(JSON.parse(client.send.lastCall.args[0])).to.deep.equal([ + 'CLOSED', + 'sub1', + 'rate-limited: slow down', + '60000', + ]) + + unblockFirstReq() + await firstReq.catch(() => undefined) + }) + it('closes a COUNT with the tripped window as backoff hint', async () => { await onMessage(Buffer.from(JSON.stringify(['COUNT', 'q1', { kinds: [1] }])))