feat(peering): treat authenticated peers as feeds

.changeset/peering-feed-treatment.md

@@ -0,0 +1,26 @@
+---
+"nostream": minor
+---
+
+feat(peering): treat authenticated peers as feeds, not clients
+
+Adds `limits.peer`, a separate limit set for connections that have proven a
+configured `peering.peers` pubkey with NIP-42 AUTH. Peer subscriptions use
+`limits.peer.subscription` instead of `limits.client.subscription`, and peer
+messages use `limits.peer.message` in a separate rate-limit bucket, so a
+full-feed subscription cannot trip the client EWMA limiter or the client
+`max_subscriptions` cap.
+
+Peers are eligible for full feeds by default: `maxLimit` is uncapped and the
+subscription caps are higher. An omitted peer REQ `limit` queries the full
+stored feed instead of the client 500-event default, and peer filter counts are
+not rejected by the protocol's former fixed ten-filter cap. Raw filter counts are
+still bounded before deduplication, and stored reads wait for the socket's
+send-flush callback so a slow peer cannot grow worker memory. Deleted and expired rows
+are excluded in SQL before the limit, so they cannot consume a page slot.
+
+A high but finite peer message rate keeps the credential from becoming an
+unbounded egress path. Unauthenticated or unconfigured connections keep the
+existing client limits exactly as before. The new limits are validated with the
+rest of the settings document and documented in `CONFIGURATION.md` and
+`docs/PEERING.md`.

CONFIGURATION.md

@@ -195,6 +195,13 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | limits.client.subscription.maxFilters       | Maximum number of filters per subscription. Defaults to 10. Disabled when set to zero. |
 | limits.message.rateLimits[].period | Rate limiting period in milliseconds. For `sliding_window`: the time window. For `ewma`: the half-life of the decay function. |
 | limits.message.rateLimits[].rate            | Maximum number of messages during period. |
+| limits.peer.message.rateLimits[].period     | Rate limit period in milliseconds for authenticated peering peers. Separate from `limits.message`, so a peer feed never spends client message budget. |
+| limits.peer.message.rateLimits[].rate       | Maximum peer messages during period. Defaults to 6000/min. |
+| limits.peer.subscription.maxFilters         | Maximum number of filters per peer subscription. Defaults to 100. Disabled when set to zero. |
+| limits.peer.subscription.maxFilterValues    | Maximum number of values allowed in each peer filter. Defaults to 10000. Disabled when set to zero. |
+| limits.peer.subscription.maxLimit           | Maximum REQ `limit` for peers. Defaults to 0 (no cap), so a peer can request a full feed. |
+| limits.peer.subscription.maxSubscriptionIdLength | Maximum peer subscription id length. Defaults to 256. |
+| limits.peer.subscription.maxSubscriptions   | Maximum number of subscriptions per authenticated peer. Defaults to 100. Disabled when set to zero. |
 | mirroring.static                            | Removed: the static mirroring worker no longer exists, so this setting is ignored and logs a warning at start-up. To sync events with another relay, use NIP-77 (Negentropy) from a client that speaks it, for example `strfry sync`, against a relay with `nip77.enabled` set. |
 | network.maxPayloadSize                      | Maximum number of bytes accepted per WebSocket frame |
 | network.remoteIpHeader                      | HTTP header from proxy containing IP address from client. |

docs/PEERING.md

@@ -81,6 +81,34 @@ authenticated for the caller to use for a feed or negentropy session. This
 proves our identity to the peer; it does not verify the peer's identity in
 return.
 
+## Feed treatment
+
+Once a connection has proven a configured peer pubkey, its subscription and
+message limits come from `limits.peer`, not `limits.client` or `limits.message`:
+
+- Peer subscription caps (`maxSubscriptions`, `maxFilters`, `maxFilterValues`,
+  `maxSubscriptionIdLength`) are separate and larger by default.
+- `limits.peer.subscription.maxLimit: 0` leaves REQ `limit` uncapped and makes an
+  omitted `limit` query the full stored feed instead of the client-sized
+  500-event default. A positive value becomes the peer's default page size.
+- Peer filter counts are bounded by `limits.peer.subscription.maxFilters`, not
+  the client cap, so a full-feed filter set is not rejected before the handler
+  can apply peer limits. The raw count is checked before duplicate removal, so
+  a large set of duplicate filters cannot stall the worker in `uniqWith`.
+- Stored reads wait for the WebSocket send callback to report the previous
+  frame as flushed, so a slow peer cannot grow the worker's outbound memory.
+  Soft-deleted and expired events are excluded in SQL before the limit, so dead
+  rows cannot consume a page slot.
+- Peer messages are counted in a separate rate-limit bucket, so a feed never
+  trips the client EWMA limiter. The default is a high but finite bound, not an
+  exemption.
+- Only a verified AUTH from an enabled `peering.peers` pubkey selects these
+  limits. An unauthenticated or unconfigured connection stays on the client
+  limits exactly as before.
+
+Scope enforcement (only the agreed kinds and time ranges) is part of the sync
+work; this phase grants feed-sized limits to trusted peers.
+
 Only the public half of the keypair is ever visible on the wire. Inbound AUTH is
 verified by `AuthMessageHandler`: event id, signature, timestamp, challenge, and
 relay host. Outbound AUTH proves this relay's identity to the peer; it does not

resources/default-settings.yaml

@@ -409,6 +409,25 @@ limits:
       maxSubscriptionIdLength: 256
       maxLimit: 5000
       minPrefixLength: 4
+  peer:
+    # Configured peering peers get feed treatment, not client treatment. These
+    # caps are separate from limits.client and limits.message, so a full-feed
+    # subscription does not trip the client EWMA bucket or max_subscriptions.
+    # A value of 0 disables that cap.
+    subscription:
+      maxSubscriptions: 100
+      maxFilters: 100
+      maxFilterValues: 10000
+      maxSubscriptionIdLength: 256
+      # 0 = no cap, so a peer can request a full feed.
+      maxLimit: 0
+      minPrefixLength: 4
+    message:
+      # period is milliseconds (60000 = one minute)
+      rateLimits:
+        - description: 6000 peer messages/min
+          period: 60000
+          rate: 6000
   message:
     # period is milliseconds (60000 = one minute)
     rateLimits:

src/@types/adapters.ts

@@ -28,6 +28,12 @@ export type IWebSocketAdapter = EventEmitter & {
   getPeerPubkey(): string | undefined
   /** Mark `pubkey` as the peer identity for this socket. Ignored if it is not authenticated. */
   setPeerPubkey(pubkey: string): void
+  /**
+   * Resolves once the previous send has flushed to the underlying socket, or
+   * immediately when there is nothing pending. Used to pause a full-feed read
+   * while the client is slow.
+   */
+  waitForWritable(): Promise<void>
   drainAndClose(reason?: string): void
   /** NIP-77 sessions of this connection. */
   getNegentropySessions(): INegentropySessions

src/@types/repositories.ts

@@ -40,12 +40,25 @@ export interface IQueryResult<T> extends Pick<Promise<T>, keyof Promise<T> & Exp
   stream(options?: Record<string, any>): PassThrough & AsyncIterable<T>
 }
 
+export interface EventQueryOptions {
+  /**
+   * Limit applied to filters that omit `limit`. Defaults to
+   * DEFAULT_FILTER_LIMIT. `0` means no limit, used for peer full feeds.
+   */
+  defaultLimit?: number
+  /**
+   * Exclude soft-deleted and expired rows in SQL, before the limit is applied.
+   * Peer full feeds set this so dead rows cannot consume a page slot.
+   */
+  excludeDeleted?: boolean
+}
+
 export interface IEventRepository {
   create(event: Event): Promise<number>
   createMany(events: Event[]): Promise<number>
   upsert(event: Event): Promise<number>
   upsertMany(events: Event[]): Promise<number>
-  findByFilters(filters: SubscriptionFilter[]): IQueryResult<DBEvent[]>
+  findByFilters(filters: SubscriptionFilter[], options?: EventQueryOptions): IQueryResult<DBEvent[]>
   countByFilters(filters: SubscriptionFilter[], options?: { timeoutMs?: number }): Promise<number>
   findReconciliationItems(filter: SubscriptionFilter, limit?: number): IQueryResult<ReconciliationItem[]>
   deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise<number>

src/@types/settings.ts

@@ -163,6 +163,15 @@ export interface ClientLimits {
   subscription?: ClientSubscriptionLimits
 }
 
+/**
+ * Limits applied to configured peering peers instead of client limits. Peer
+ * feeds need larger caps than end-user clients, while still being bounded.
+ */
+export interface PeerLimits {
+  subscription?: ClientSubscriptionLimits
+  message?: MessageLimits
+}
+
 export interface MessageRateLimit extends RateLimit {
   type?: MessageType
 }
@@ -206,6 +215,7 @@ export interface Limits {
   admin?: AdminLimits
   connection?: ConnectionLimits
   client?: ClientLimits
+  peer?: PeerLimits
   event?: EventLimits
   message?: MessageLimits
 }

src/adapters/web-socket-adapter.ts

@@ -28,6 +28,7 @@ import { getRemoteAddress } from '../utils/http'
 import { createReadAuthorizationGuard } from '../utils/nip42'
 import { Nip42SessionManager } from '../utils/nip42-session'
 import { isConfiguredPeeringPeer } from '../utils/peering-auth'
+import { messageLimitsFor } from '../utils/peer-limits'
 import { IRateLimiter } from '../@types/utils'
 import { isEventMatchingFilter } from '../utils/event'
 import { isHidden } from '../utils/hidden-content-cache'
@@ -74,6 +75,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
   public clientId: string
   private clientAddress: SocketAddress
   private alive: boolean
+  private pendingSend: Promise<void> | undefined
   private subscriptions: Map<SubscriptionId, SubscriptionFilter[]>
   private readonly session: Nip42SessionManager
   /** Configured peering peer proven by AUTH on this socket, if any. */
@@ -192,7 +194,28 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
     if (this.client.readyState !== WebSocket.OPEN) {
       return
     }
-    this.client.send(JSON.stringify(message))
+
+    // `ws` does not emit `drain`; the send callback is its write-completion
+    // signal, fired once the frame has been handed to the underlying socket.
+    const pending = new Promise<void>((resolve) => {
+      this.client.send(JSON.stringify(message), () => resolve())
+    })
+    this.pendingSend = pending
+    void pending.then(() => {
+      if (this.pendingSend === pending) {
+        this.pendingSend = undefined
+      }
+    })
+  }
+
+  public waitForWritable(): Promise<void> {
+    if (this.client.readyState !== WebSocket.OPEN) {
+      return Promise.resolve()
+    }
+
+    // A full feed can outrun a slow peer; wait for the previous frame to flush
+    // instead of letting the outbound buffer grow without bound.
+    return this.pendingSend ?? Promise.resolve()
   }
 
   public onHeartbeat(): void {
@@ -342,16 +365,20 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
    * the wrong one to throttle.
    */
   private async isRateLimited(client: string, type: string): Promise<false | number> {
-    const { rateLimits, ipWhitelist = [] } = this.settings().limits?.message ?? {}
+    const isPeer = this.getPeerPubkey() !== undefined
+    const { rateLimits, ipWhitelist = [] } = messageLimitsFor(this.settings(), isPeer) ?? {}
 
     if (!Array.isArray(rateLimits) || !rateLimits.length || ipWhitelist.includes(client)) {
       return false
     }
 
     const rateLimiter = this.rateLimiter()
 
+    // Peer buckets are separate so a connection that spent client budget before
+    // AUTH does not carry that score into feed treatment.
+    const bucket = isPeer ? 'peer-message' : 'message'
     const hit = (period: number, rate: number) =>
-      rateLimiter.hit(`${client}:message:${type}:${period}`, 1, { period, rate })
+      rateLimiter.hit(`${client}:${bucket}:${type}:${period}`, 1, { period, rate })
 
     for (const { rate, period } of rateLimits) {
       const result = await hit(period, rate)

src/handlers/count-message-handler.ts

@@ -11,6 +11,7 @@ import { createLogger } from '../factories/logger-factory'
 import { countFilterValues } from '../utils/filter'
 import { createClosedMessage, createCountResultMessage } from '../utils/messages'
 import { isCountAuthorized } from '../utils/nip42'
+import { subscriptionLimitsFor } from '../utils/peer-limits'
 
 const debug = createLogger('count-message-handler')
 
@@ -29,9 +30,24 @@ export class CountMessageHandler implements IMessageHandler {
       return
     }
 
-    // Some clients send the same filter more than once.
-    // We remove duplicates so we do less DB work.
-    const filters = uniqWith(equals, message.slice(2)) as SubscriptionFilter[]
+    // Some clients send the same filter more than once. Bound the raw list
+    // before deduplicating it, since uniqWith() is O(n^2) and the schema no
+    // longer caps filter counts so peer limits can be larger.
+    const incomingFilters = message.slice(2) as SubscriptionFilter[]
+    const maxFilters =
+      subscriptionLimitsFor(this.settings(), this.webSocket.getPeerPubkey() !== undefined)?.maxFilters ?? 0
+    if (maxFilters > 0 && incomingFilters.length > maxFilters) {
+      this.webSocket.emit(
+        WebSocketAdapterEvent.Message,
+        createClosedMessage(
+          queryId,
+          `Too many filters: Number of filters per count query must be less than or equal to ${maxFilters}`,
+        ),
+      )
+      return
+    }
+
+    const filters = uniqWith(equals, incomingFilters) as SubscriptionFilter[]
 
     const reason = this.canCount(queryId, filters)
     if (reason) {
@@ -52,7 +68,10 @@ export class CountMessageHandler implements IMessageHandler {
   }
 
   private canCount(queryId: SubscriptionId, filters: SubscriptionFilter[]): string | undefined {
-    const subscriptionLimits = this.settings().limits?.client?.subscription
+    const subscriptionLimits = subscriptionLimitsFor(
+      this.settings(),
+      this.webSocket.getPeerPubkey() !== undefined,
+    )
     const maxFilters = subscriptionLimits?.maxFilters ?? 0
 
     if (maxFilters > 0 && filters.length > maxFilters) {

src/handlers/neg-open-message-handler.ts

@@ -12,6 +12,7 @@ import { WebSocketAdapterEvent } from '../constants/adapter'
 import { countFilterValues } from '../utils/filter'
 import { createLogger } from '../factories/logger-factory'
 import { isReconciliationAuthorized } from '../utils/nip42'
+import { subscriptionLimitsFor } from '../utils/peer-limits'
 
 const logger = createLogger('neg-open-message-handler')
 
@@ -122,7 +123,10 @@ export class NegOpenMessageHandler implements IMessageHandler {
     initialMessage: Buffer,
   ): string | undefined {
     const settings = this.settings()
-    const subscriptionLimits = settings.limits?.client?.subscription
+    const subscriptionLimits = subscriptionLimitsFor(
+      settings,
+      this.webSocket.getPeerPubkey() !== undefined,
+    )
 
     const maxFilterValues = subscriptionLimits?.maxFilterValues ?? 0
     if (maxFilterValues > 0 && countFilterValues(filter) > maxFilterValues) {

src/handlers/subscribe-message-handler.ts

@@ -15,7 +15,7 @@ import { IAbortable, IMessageHandler } from '../@types/message-handlers'
 import { IEventRepository, IInviteCodeRepository } from '../@types/repositories'
 import { isNip43InviteRequestFilter, isRelaySelfConsistent } from '../utils/nip43'
 import { buildInviteCodeEvent, issueInviteCode } from '../utils/nip43-invites'
-import { streamEach, streamEnd, streamFilter, streamMap } from '../utils/stream'
+import { streamEachAsync, streamEnd, streamFilter, streamMap } from '../utils/stream'
 import { SubscriptionFilter, SubscriptionId } from '../@types/subscription'
 import { createLogger } from '../factories/logger-factory'
 import { Event } from '../@types/event'
@@ -25,6 +25,7 @@ import { IWebSocketAdapter } from '../@types/adapters'
 import { Settings } from '../@types/settings'
 import { SubscribeMessage } from '../@types/messages'
 import { WebSocketAdapterEvent } from '../constants/adapter'
+import { subscriptionLimitsFor } from '../utils/peer-limits'
 
 const logger = createLogger('subscribe-message-handler')
 
@@ -51,7 +52,20 @@ 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[]
+    const incomingFilters = message.slice(2) as SubscriptionFilter[]
+
+    // Bound before uniqWith(): deduplicating an unbounded list is O(n^2). The
+    // schema no longer caps filter counts so peer limits can be larger.
+    const maxFilters =
+      subscriptionLimitsFor(this.settings(), this.webSocket.getPeerPubkey() !== undefined)?.maxFilters ?? 0
+    if (maxFilters > 0 && incomingFilters.length > maxFilters) {
+      const reason = `Too many filters: Number of filters per susbscription must be less then or equal to ${maxFilters}`
+      logger('subscription %s with %o rejected: %s', subscriptionId, incomingFilters, reason)
+      this.webSocket.emit(WebSocketAdapterEvent.Message, createNoticeMessage(`Subscription rejected: ${reason}`))
+      return
+    }
+
+    const rawFilters = uniqWith(equals, incomingFilters) as SubscriptionFilter[]
 
     // NIP-50: strip search from filters when disabled so isEventMatchingFilter ignores it
     const nip50Enabled = this.settings()?.nip50?.enabled ?? false
@@ -207,8 +221,10 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
 
   private async fetchAndSend(subscriptionId: string, filters: SubscriptionFilter[]): Promise<void> {
     logger('fetching events for subscription %s with filters %o', subscriptionId, filters)
-    const sendEvent = (event: Event) =>
+    const sendEvent = async (event: Event) => {
       this.webSocket.emit(WebSocketAdapterEvent.Message, createOutgoingEventMessage(subscriptionId, event))
+      await this.webSocket.waitForWritable()
+    }
     const sendEOSE = () =>
       this.webSocket.emit(WebSocketAdapterEvent.Message, createEndOfStoredEventsNoticeMessage(subscriptionId))
     const isSubscribedToEvent = SubscribeMessageHandler.isClientSubscribedToEvent(filters)
@@ -225,7 +241,15 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
       () => this.webSocket.getAuthenticatedPubkeys(),
     )
 
-    const findEvents = this.eventRepository.findByFilters(filters).stream()
+    // A peer's omitted `limit` means the full stored feed, not the client-sized
+    // 500-event default. `maxLimit: 0` opts out of a cap; a positive value
+    // becomes the peer's default page size.
+    const isPeer = this.webSocket.getPeerPubkey() !== undefined
+    const peerDefaultLimit = isPeer ? (this.settings().limits?.peer?.subscription?.maxLimit ?? 0) : undefined
+
+    const findEvents = isPeer
+      ? this.eventRepository.findByFilters(filters, { defaultLimit: peerDefaultLimit, excludeDeleted: true }).stream()
+      : this.eventRepository.findByFilters(filters).stream()
 
     const abortableFindEvents = addAbortSignal(this.abortController.signal, findEvents)
 
@@ -237,7 +261,7 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
         streamFilter(isTagUnexpired),
         streamFilter(isReadAuthorized),
         streamFilter(isSubscribedToEvent),
-        streamEach(sendEvent),
+        streamEachAsync(sendEvent),
         streamEnd(sendEOSE),
       )
     } catch (error) {
@@ -258,7 +282,10 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
   private canSubscribe(subscriptionId: SubscriptionId, filters: SubscriptionFilter[]): string | undefined {
     const subscriptions = this.webSocket.getSubscriptions()
     const existingSubscription = subscriptions.get(subscriptionId)
-    const subscriptionLimits = this.settings().limits?.client?.subscription
+    const subscriptionLimits = subscriptionLimitsFor(
+      this.settings(),
+      this.webSocket.getPeerPubkey() !== undefined,
+    )
 
     if (existingSubscription?.length && equals(filters, existingSubscription)) {
       return `Duplicate subscription ${subscriptionId}: Ignoring`