fix(rate-limit): send the backoff the strategy means, name an outage as one, and bucket messages by type

.changeset/rate-limit-wire-correctness.md

@@ -0,0 +1,15 @@
+---
+'nostream': patch
+---
+
+fix(rate-limit): send the backoff the strategy actually means, name an outage as an error, and bucket messages by type
+
+Three defects on one wire path, from the spike in #1004.
+
+**The EWMA backoff was a half-life, sent as a delay.** `RateLimit.period` is the decay half-life under `ewma`, so the time until a limit lifts is `period · log2(R_new / rate)`, not `period`. With `rate: 6, period: 60000`, a client four times over the limit was told to wait 60 s and needed 120 s. The Lua script now returns its score instead of a yes/no, and the limiter returns `period · log2(score / rate)`. It was only accidentally safe before: the hash carried a TTL of exactly `period`, so a quiet client's state expired anyway — an unstated coupling that would break the moment anyone lengthened the TTL to keep state warm.
+
+**A limiter outage was announced as `rate-limited`.** A failing backend made every connected client back off for a fault on this side. It now answers `error: rate limiter unavailable` and still rejects the event; fail-closed behaviour is unchanged.
+
+**The message bucket ignored the message type.** One bucket per client meant a `REQ` could be refused because its `EVENT`s had filled it — the wrong message throttled, and the wrong one told to back off. The bucket key now carries the type, read from a bounded prefix of the frame because the limiter runs before validation.
+
+`IRateLimiter.hit` returns `IRateLimiterResult` (`limited`, plus `retryAfterMs` when the strategy can estimate one) so each strategy answers in its own terms; a caller with no estimate keeps sending the period.

src/@types/utils.ts

@@ -3,6 +3,19 @@ export interface IRateLimiterOptions {
   rate: number
 }
 
+/**
+ * The outcome of one limiter hit.
+ *
+ * `retryAfterMs` is an estimate of how long the caller must wait before its
+ * next attempt can be accepted, and it is absent when the strategy cannot say
+ * so. Absence means "limited, no estimate" — it must never be read as "retry
+ * immediately", and the caller decides what to send instead.
+ */
+export interface IRateLimiterResult {
+  limited: boolean
+  retryAfterMs?: number
+}
+
 export interface IRateLimiter {
-  hit(key: string, step: number, options: IRateLimiterOptions): Promise<boolean>
+  hit(key: string, step: number, options: IRateLimiterOptions): Promise<IRateLimiterResult>
 }

src/adapters/web-socket-adapter.ts

@@ -35,6 +35,36 @@ import { Settings } from '../@types/settings'
 import { SocketAddress } from 'net'
 
 const logger = createLogger('web-socket-adapter')
+
+/**
+ * The NIP-01 type of a client frame, or `OTHER` when it is not one.
+ *
+ * The message limiter runs before validation — it has to, or a client could
+ * spend the parse budget first — so it reads a bounded prefix instead of
+ * decoding the frame. A frame that does not start with `["WORD"` shares the
+ * `OTHER` bucket; it is refused a moment later anyway.
+ */
+const THROTTLED_MESSAGE_TYPES = new Set<string>([
+  'REQ',
+  'EVENT',
+  'COUNT',
+  'CLOSE',
+  'AUTH',
+  'NEG-OPEN',
+  'NEG-MSG',
+  'NEG-CLOSE',
+])
+
+const messageType = (raw: Buffer): string => {
+  const prefix = raw.toString('utf8', 0, Math.min(raw.length, 32))
+  const match = /^\s*\[\s*"([A-Za-z-]{1,8})"/.exec(prefix)
+  const type = match ? match[1].toUpperCase() : ''
+
+  // A closed set, not whatever the frame claims. Bucketing by an arbitrary word
+  // would let a client rotate invented types (`["AAAA"]`, `["BBBB"]`, …) and
+  // keep every bucket empty; anything unrecognised shares one bucket.
+  return THROTTLED_MESSAGE_TYPES.has(type) ? type : 'OTHER'
+}
 const debugHeartbeat = logger.extend('heartbeat')
 
 const abortableMessageHandlers: WeakMap<WebSocket, IAbortable[]> = new WeakMap()
@@ -218,7 +248,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
     let abortable = false
     let messageHandler: (IMessageHandler & IAbortable) | undefined = undefined
     try {
-      const retryAfterMs = await this.isRateLimited(this.clientAddress.address)
+      const retryAfterMs = await this.isRateLimited(this.clientAddress.address, messageType(raw))
       if (retryAfterMs) {
         this.sendMessage(this.createRateLimitedMessage(raw, retryAfterMs))
         return
@@ -276,10 +306,15 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
   }
 
   /**
-   * Resolves to false when allowed; otherwise to the tripped window's period in ms,
-   * sent as the NIP-01 `rate-limited` backoff.
+   * Resolves to false when allowed; otherwise to the backoff in ms, sent as the
+   * NIP-01 `rate-limited` parameter.
+   *
+   * The bucket carries the message type: a client told to back off should back
+   * off *the message that tripped the limit*. One shared bucket refused a REQ
+   * because its EVENTs had filled it, which is the wrong message to punish and
+   * the wrong one to throttle.
    */
-  private async isRateLimited(client: string): Promise<false | number> {
+  private async isRateLimited(client: string, type: string): Promise<false | number> {
     const { rateLimits, ipWhitelist = [] } = this.settings().limits?.message ?? {}
 
     if (!Array.isArray(rateLimits) || !rateLimits.length || ipWhitelist.includes(client)) {
@@ -288,15 +323,16 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
 
     const rateLimiter = this.rateLimiter()
 
-    const hit = (period: number, rate: number) => rateLimiter.hit(`${client}:message:${period}`, 1, { period, rate })
+    const hit = (period: number, rate: number) =>
+      rateLimiter.hit(`${client}:message:${type}:${period}`, 1, { period, rate })
 
     for (const { rate, period } of rateLimits) {
-      const isRateLimited = await hit(period, rate)
+      const result = await hit(period, rate)
 
-      if (isRateLimited) {
-        logger('rate limited %s: %d messages / %d ms exceeded', client, rate, period)
+      if (result.limited) {
+        logger('rate limited %s (%s): %d messages / %d ms exceeded', client, type, rate, period)
 
-        return Math.max(1, Math.ceil(period))
+        return result.retryAfterMs ?? Math.max(1, Math.ceil(period))
       }
     }
 

src/controllers/admission/get-admission-check-controller.ts

@@ -53,7 +53,7 @@ export class GetSubmissionCheckController implements IController {
     if (Array.isArray(ipWhitelist) && !ipWhitelist.includes(remoteAddress)) {
       const rateLimiter = this.rateLimiter()
       for (const { rate, period } of rateLimits) {
-        if (await rateLimiter.hit(`${remoteAddress}:admission-check:${period}`, 1, { period, rate })) {
+        if ((await rateLimiter.hit(`${remoteAddress}:admission-check:${period}`, 1, { period, rate })).limited) {
           logger('rate limited %s: %d in %d milliseconds', remoteAddress, rate, period)
           limited = true
         }

src/controllers/invoices/post-invoice-controller.ts

@@ -183,7 +183,7 @@ export class PostInvoiceController implements IController {
     if (Array.isArray(ipWhitelist) && !ipWhitelist.includes(remoteAddress)) {
       const rateLimiter = this.rateLimiter()
       for (const { rate, period } of rateLimits) {
-        if (await rateLimiter.hit(`${remoteAddress}:invoice:${period}`, 1, { period, rate })) {
+        if ((await rateLimiter.hit(`${remoteAddress}:invoice:${period}`, 1, { period, rate })).limited) {
           logger('rate limited %s: %d in %d milliseconds', remoteAddress, rate, period)
           limited = true
         }

src/controllers/subscriptions/get-subscription-status-controller.ts

@@ -124,7 +124,7 @@ export class GetSubscriptionStatusController implements IController {
     if (Array.isArray(ipWhitelist) && !ipWhitelist.includes(remoteAddress)) {
       const rateLimiter = this.rateLimiter()
       for (const { rate, period } of rateLimits) {
-        if (await rateLimiter.hit(`${remoteAddress}:subscription-status:${period}`, 1, { period, rate })) {
+        if ((await rateLimiter.hit(`${remoteAddress}:subscription-status:${period}`, 1, { period, rate })).limited) {
           logger('rate limited %s: %d in %d milliseconds', remoteAddress, rate, period)
           limited = true
         }

src/handlers/event-message-handler.ts

@@ -61,13 +61,26 @@ import { createLogger } from '../factories/logger-factory'
 import { Factory } from '../@types/base'
 import { ICacheAdapter } from '../@types/adapters'
 import { IncomingEventMessage } from '../@types/messages'
-import { IRateLimiter } from '../@types/utils'
+import { IRateLimiter, IRateLimiterResult } from '../@types/utils'
 import { IWebSocketAdapter } from '../@types/adapters'
 import { IWotGraphService } from '../@types/services'
 import { Nip05Verification } from '../@types/nip05'
 import { WebSocketAdapterEvent } from '../constants/adapter'
 import { recordModerationRateLimitHitSafely } from '../utils/moderation-rate-limit-stats'
 
+/**
+ * Why an event was admitted or refused, in the terms the wire needs.
+ *
+ * `limited` always carries a backoff — the strategy's estimate when it has one,
+ * the period when it does not. `unavailable` is our fault, not the client's, and
+ * must not be announced as `rate-limited`: every client would back off for a
+ * backend outage.
+ */
+type RateLimitVerdict =
+  | { status: 'allowed' }
+  | { status: 'limited', retryAfterMs: number }
+  | { status: 'unavailable' }
+
 const logger = createLogger('event-message-handler')
 
 export class EventMessageHandler implements IMessageHandler {
@@ -106,17 +119,27 @@ export class EventMessageHandler implements IMessageHandler {
 
     event = this.addExpirationMetadata(event)
 
-    const rateLimited = await this.isRateLimited(event)
-    if (rateLimited) {
+    const rateLimit = await this.isRateLimited(event)
+    if (rateLimit.status === 'limited') {
       logger('event %s rejected: rate-limited', event.id)
-      const retryAfterMs = typeof rateLimited === 'number' ? String(rateLimited) : undefined
       void recordModerationRateLimitHitSafely(this.cache, {
         pubkey: event.pubkey,
         ip: event[ContextMetadataKey]?.remoteAddress?.address,
       })
       this.webSocket.emit(
         WebSocketAdapterEvent.Message,
-        createEventCommandResult(event.id, false, 'rate-limited: slow down', retryAfterMs),
+        createEventCommandResult(event.id, false, 'rate-limited: slow down', String(rateLimit.retryAfterMs)),
+      )
+      return
+    }
+
+    if (rateLimit.status === 'unavailable') {
+      // Not the client's fault, so it gets `error:` and no backoff: telling it to
+      // slow down would make every client back off at once for our outage.
+      logger('event %s rejected: rate limiter unavailable', event.id)
+      this.webSocket.emit(
+        WebSocketAdapterEvent.Message,
+        createEventCommandResult(event.id, false, 'error: rate limiter unavailable'),
       )
       return
     }
@@ -404,12 +427,17 @@ export class EventMessageHandler implements IMessageHandler {
   }
 
   /**
-   * Resolves to false when allowed; otherwise to the tripped window's period in ms,
-   * sent as the NIP-01 `rate-limited` backoff, or true when no backoff can be given.
+   * Why an event was not admitted, in the terms the wire needs.
+   *
+   * `limited` carries the backoff when the strategy can estimate one, and the
+   * caller sends the period when it cannot. `unavailable` is a different thing:
+   * the limiter backend failed, so the event is rejected fail-closed, but the
+   * client must not be told to slow down — every client would back off at once
+   * for a fault that is ours.
    */
-  protected async isRateLimited(event: Event): Promise<boolean | number> {
+  protected async isRateLimited(event: Event): Promise<RateLimitVerdict> {
     if (this.getRelayPublicKey() === event.pubkey) {
-      return false
+      return { status: 'allowed' }
     }
 
     const currentSettings = this.settings()
@@ -418,23 +446,23 @@ export class EventMessageHandler implements IMessageHandler {
     const rateLimits =
       (await this.getPlanRateLimits(event, currentSettings)) ?? currentSettings.limits?.event?.rateLimits
     if (!rateLimits || !rateLimits.length) {
-      return false
+      return { status: 'allowed' }
     }
 
     if (
       typeof whitelists?.pubkeys !== 'undefined' &&
       Array.isArray(whitelists?.pubkeys) &&
       whitelists.pubkeys.includes(event.pubkey)
     ) {
-      return false
+      return { status: 'allowed' }
     }
 
     if (
       typeof whitelists?.ipAddresses !== 'undefined' &&
       Array.isArray(whitelists?.ipAddresses) &&
       whitelists.ipAddresses.includes(this.webSocket.getClientAddress())
     ) {
-      return false
+      return { status: 'allowed' }
     }
 
     const rateLimiter = this.rateLimiter()
@@ -457,23 +485,28 @@ export class EventMessageHandler implements IMessageHandler {
         continue
       }
 
-      let isRateLimited = false
+      let result: IRateLimiterResult
       try {
-        isRateLimited = await hit({ period, rate, kinds })
+        result = await hit({ period, rate, kinds })
       } catch (error) {
-        // Fail closed when the rate limiter backend is unavailable.
+        // Fail closed when the rate limiter backend is unavailable, but say so:
+        // announcing `rate-limited` here made every connected client back off
+        // for a fault on this side.
         logger('rate limiter unavailable for %s (%d/%d): %o', event.pubkey, rate, period, error)
-        return true
+        return { status: 'unavailable' }
       }
 
-      if (isRateLimited) {
+      if (result.limited) {
         logger('rate limited %s: %d events / %d ms exceeded', event.pubkey, rate, period)
 
-        return Math.max(1, Math.ceil(period))
+        return {
+          status: 'limited',
+          retryAfterMs: result.retryAfterMs ?? Math.max(1, Math.ceil(period)),
+        }
       }
     }
 
-    return false
+    return { status: 'allowed' }
   }
 
   /** The rate limits of the plan the sender is subscribed to, if it has its own and the subscription still gives access. */

src/handlers/request-handlers/rate-limiter-middleware.ts

@@ -41,7 +41,7 @@ export async function isRateLimited(remoteAddress: string, settings: Settings):
 
   let limited = false
   for (const { rate, period } of rateLimits) {
-    const isRateLimited = await hit(period, rate)
+    const isRateLimited = (await hit(period, rate)).limited
 
     if (isRateLimited) {
       logger('rate limited %s: %d messages / %d ms exceeded', remoteAddress, rate, period)

src/handlers/subscribe-message-handler.ts

@@ -188,7 +188,7 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
     for (const { period, rate } of rateLimits) {
       let isRateLimited: boolean
       try {
-        isRateLimited = await rateLimiter.hit(`${pubkey}:invites:${period}`, 1, { period, rate })
+        isRateLimited = (await rateLimiter.hit(`${pubkey}:invites:${period}`, 1, { period, rate })).limited
       } catch (error) {
         // Fail closed: minting writes to the database, so an unavailable limiter
         // must not leave the gate open.

src/utils/admin-rate-limit.ts

@@ -27,7 +27,7 @@ export async function isAdminRateLimited(
   if (!Array.isArray(ipWhitelist) || !ipWhitelist.includes(remoteAddress)) {
     const rateLimiter = rateLimiterFactory()
     for (const { rate, period } of rateLimits) {
-      if (await rateLimiter.hit(`${remoteAddress}:admin-${scope}:${period}`, 1, { period, rate })) {
+      if ((await rateLimiter.hit(`${remoteAddress}:admin-${scope}:${period}`, 1, { period, rate })).limited) {
         logger('rate limited %s: %d in %d milliseconds', remoteAddress, rate, period)
         limited = true
       }

src/utils/ewma-rate-limiter.ts

@@ -1,4 +1,4 @@
-import { IRateLimiter, IRateLimiterOptions } from '../@types/utils'
+import { IRateLimiter, IRateLimiterOptions, IRateLimiterResult } from '../@types/utils'
 import { createLogger } from '../factories/logger-factory'
 import { ICacheAdapter } from '../@types/adapters'
 
@@ -21,11 +21,10 @@ const rateLimitScript = {
       redis.call('HSET', key, 'rate', R_new, 'timestamp', timestamp)
       redis.call('EXPIRE', key, math.ceil(period / 1000))
 
-      if R_new > rate then
-          return 1
-      else
-          return 0
-      end
+      -- A Lua number reaches the client truncated to an integer, and the score
+      -- is a float: tostring is what keeps the fraction the caller needs to
+      -- tell 6.4 hits from 24.
+      return tostring(R_new)
     `,
   }
 
@@ -63,17 +62,50 @@ export class EWMARateLimiter implements IRateLimiter {
     key: string,
     step: number,
     options: IRateLimiterOptions,
-  ): Promise<boolean> {
+  ): Promise<IRateLimiterResult> {
     const { rate, period } = options
 
     const result = await this.cache.eval(rateLimitScript.SCRIPT,
        [ewmaKey(key)],
        [Date.now().toString(), rate.toString(), period.toString(), step.toString()]
     )
 
-    debug('ewma rate limited on %s bucket: %s', key, result ? 'yes' : 'no')
+    // A null/undefined answer means the script returned nil, which is a protocol
+    // disagreement, not a score of zero: reading it as "allowed" would silently
+    // disable the limiter.
+    if (typeof result !== 'string' && typeof result !== 'number') {
+        throw new Error(`ewma rate limiter returned no score: ${String(result)}`)
+    }
 
-    return result === 1
+    const score = Number(result)
+    if (!Number.isFinite(score)) {
+        throw new Error(`ewma rate limiter returned a non-numeric score: ${String(result)}`)
+    }
+
+    const limited = score > rate
+
+    debug('ewma rate limited on %s bucket: %s (score %d, rate %d)', key, limited ? 'yes' : 'no', score, rate)
+
+    if (!limited) {
+        return { limited: false }
+    }
+
+    // `period` is the half-life, not the wait: the score decays as
+    // `score * 2^(-t/period)`, so it takes `period * log2(score / target)` to
+    // fall back to `target`. Reporting `period` understates a four-times
+    // overrun by half (rate 6, score 24, period 60 s: ~136 s, not 60 s).
+    //
+    // The target is `rate - step`, not `rate`: the script adds this hit's step
+    // before comparing, so a score that has decayed to exactly `rate` still
+    // exceeds it. With no headroom (`rate <= step`) no amount of decay admits
+    // another hit, and the key's TTL is what eventually clears it.
+    const headroom = rate - step
+    return {
+        limited: true,
+        retryAfterMs: headroom > 0
+            ? Math.max(1, Math.ceil(period * Math.log2(score / headroom)))
+            : Math.max(1, Math.ceil(period)),
+    }
   }
 
 }