feat(nip-01): send rate-limited backoff hint on CLOSED

.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", <id>, "rate-limited: slow down", "<ms>"]` with the tripped window's period as the backoff, instead of a `NOTICE`. Other message types keep the existing `NOTICE`.

src/@types/messages.ts

@@ -90,6 +90,7 @@ export interface ClosedMessage {
   0: MessageType.CLOSED
   1: SubscriptionId
   2: string
+  3?: string
 }
 
 // NIP-42

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<boolean> {
+  /**
+   * 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<false | number> {
     const { rateLimits, ipWhitelist = [] } = this.settings().limits?.message ?? {}
 
     if (!Array.isArray(rateLimits) || !rateLimits.length || ipWhitelist.includes(client)) {
@@ -273,13 +278,62 @@ 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'
+    ) {
+      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

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<void> {
     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

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

test/unit/adapters/web-socket-adapter.spec.ts

@@ -556,31 +556,96 @@ 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<void>
 
-      // 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]
       })
 
-      slidingWindowRateLimiter.returns({
-        hit: sandbox.stub().resolves(true),
+      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'])
       })
 
-      const messageCall = client.on.getCalls().find((call: any) => call.args[0] === 'message')
-      const onMessage = messageCall.args[1]
+      it('closes a REQ with the tripped window as backoff hint', async () => {
+        await onMessage(Buffer.from(JSON.stringify(['REQ', 'sub1', { 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', '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<void>((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)
 
-      expect(client.send).to.have.been.called
+        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] }])))
+
+        expect(client.send).to.have.been.calledOnce
+        expect(sentFrame()).to.deep.equal(['CLOSED', 'q1', 'rate-limited: slow down', '60000'])
+      })
+
+      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 () => {

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] }])