diff --git a/.changeset/peering-feed-treatment.md b/.changeset/peering-feed-treatment.md new file mode 100644 index 000000000..14707e5d4 --- /dev/null +++ b/.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`. diff --git a/CONFIGURATION.md b/CONFIGURATION.md index c177ca376..c7567ec71 100644 --- a/CONFIGURATION.md +++ b/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. | diff --git a/docs/PEERING.md b/docs/PEERING.md index 375b7c702..e8cb9122c 100644 --- a/docs/PEERING.md +++ b/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 diff --git a/resources/default-settings.yaml b/resources/default-settings.yaml index 44d375add..f53357362 100755 --- a/resources/default-settings.yaml +++ b/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: diff --git a/src/@types/adapters.ts b/src/@types/adapters.ts index cd6cb96f9..9e44c6b0f 100644 --- a/src/@types/adapters.ts +++ b/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 drainAndClose(reason?: string): void /** NIP-77 sessions of this connection. */ getNegentropySessions(): INegentropySessions diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts index e64c2873e..3da242738 100644 --- a/src/@types/repositories.ts +++ b/src/@types/repositories.ts @@ -40,12 +40,25 @@ export interface IQueryResult extends Pick, keyof Promise & Exp stream(options?: Record): PassThrough & AsyncIterable } +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 createMany(events: Event[]): Promise upsert(event: Event): Promise upsertMany(events: Event[]): Promise - findByFilters(filters: SubscriptionFilter[]): IQueryResult + findByFilters(filters: SubscriptionFilter[], options?: EventQueryOptions): IQueryResult countByFilters(filters: SubscriptionFilter[], options?: { timeoutMs?: number }): Promise findReconciliationItems(filter: SubscriptionFilter, limit?: number): IQueryResult deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise diff --git a/src/@types/settings.ts b/src/@types/settings.ts index eaf5c71bd..1788dd13c 100644 --- a/src/@types/settings.ts +++ b/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 } diff --git a/src/adapters/web-socket-adapter.ts b/src/adapters/web-socket-adapter.ts index 3f64e5f55..f4de701c2 100644 --- a/src/adapters/web-socket-adapter.ts +++ b/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 | undefined private subscriptions: Map 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((resolve) => { + this.client.send(JSON.stringify(message), () => resolve()) + }) + this.pendingSend = pending + void pending.then(() => { + if (this.pendingSend === pending) { + this.pendingSend = undefined + } + }) + } + + public waitForWritable(): Promise { + 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,7 +365,8 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter * the wrong one to throttle. */ private async isRateLimited(client: string, type: string): Promise { - 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 @@ -350,8 +374,11 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter 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) diff --git a/src/handlers/count-message-handler.ts b/src/handlers/count-message-handler.ts index 2a5084ee4..a1db59215 100644 --- a/src/handlers/count-message-handler.ts +++ b/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) { diff --git a/src/handlers/neg-open-message-handler.ts b/src/handlers/neg-open-message-handler.ts index 03c0e7ce4..eef934bf4 100644 --- a/src/handlers/neg-open-message-handler.ts +++ b/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) { diff --git a/src/handlers/subscribe-message-handler.ts b/src/handlers/subscribe-message-handler.ts index ef5667fb9..a71a503bf 100644 --- a/src/handlers/subscribe-message-handler.ts +++ b/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 { 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 { 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` diff --git a/src/repositories/event-repository.ts b/src/repositories/event-repository.ts index fe201d89f..69dd1e0d8 100644 --- a/src/repositories/event-repository.ts +++ b/src/repositories/event-repository.ts @@ -25,6 +25,7 @@ import { DatabaseClient, EventId } from '../@types/base' import { DBEvent, Event } from '../@types/event' import { EventPurgeCounts, + EventQueryOptions, EventRetentionOptions, IEventRepository, IQueryResult, @@ -61,11 +62,16 @@ export class EventRepository implements IEventRepository { private readonly settings?: () => Settings, ) {} - public findByFilters(filters: SubscriptionFilter[]): IQueryResult { + public findByFilters( + filters: SubscriptionFilter[], + options?: EventQueryOptions, + ): IQueryResult { logger('querying for %o', filters) if (!Array.isArray(filters) || !filters.length) { throw new Error('Filters cannot be empty') } + const defaultLimit = options?.defaultLimit ?? DEFAULT_FILTER_LIMIT + const now = Math.floor(Date.now() / 1000) const isSearchQueries: boolean[] = [] const queries = filters.map((currentFilter) => { const builder = this.readReplicaDbClient('events') @@ -73,6 +79,14 @@ export class EventRepository implements IEventRepository { const { isSearchQuery } = this.applyFilterConditions(builder, currentFilter) isSearchQueries.push(isSearchQuery) + if (options?.excludeDeleted) { + // Handler-side filtering is too late once LIMIT has been applied: a + // soft-deleted or expired row would otherwise use a page slot. + builder.whereNull('events.deleted_at').andWhere((bd) => { + bd.whereNull('events.expires_at').orWhere('events.expires_at', '>', now) + }) + } + if (isSearchQuery) { const tsConfig = this.getNip50Language() const nip50Settings = this.settings?.() @@ -85,7 +99,11 @@ export class EventRepository implements IEventRepository { builder.select(searchSelection) } - builder.limit(typeof currentFilter.limit === 'number' ? currentFilter.limit : DEFAULT_FILTER_LIMIT) + if (typeof currentFilter.limit === 'number') { + builder.limit(currentFilter.limit) + } else if (defaultLimit > 0) { + builder.limit(defaultLimit) + } this.applyFilterOrder(builder, currentFilter, isSearchQuery) return builder diff --git a/src/schemas/message-schema.ts b/src/schemas/message-schema.ts index 6a928d4e4..edff2098f 100644 --- a/src/schemas/message-schema.ts +++ b/src/schemas/message-schema.ts @@ -19,14 +19,6 @@ export const reqMessageSchema = z inclusive: true, message: 'REQ message must contain at least one filter', }) - } else if (val.length > 12) { - ctx.addIssue({ - code: z.ZodIssueCode.too_big, - maximum: 12, - type: 'array', - inclusive: true, - message: 'REQ message must contain at most 12 elements', - }) } }) @@ -42,14 +34,6 @@ export const countMessageSchema = z inclusive: true, message: 'COUNT message must contain at least one filter', }) - } else if (val.length > 12) { - ctx.addIssue({ - code: z.ZodIssueCode.too_big, - maximum: 12, - type: 'array', - inclusive: true, - message: 'COUNT message must contain at most 12 elements', - }) } }) diff --git a/src/utils/peer-limits.ts b/src/utils/peer-limits.ts new file mode 100644 index 000000000..f0947521c --- /dev/null +++ b/src/utils/peer-limits.ts @@ -0,0 +1,10 @@ +import { ClientSubscriptionLimits, MessageLimits, Settings } from '../@types/settings' + +export const subscriptionLimitsFor = ( + settings: Settings | undefined, + isPeer: boolean, +): ClientSubscriptionLimits | undefined => + isPeer ? settings?.limits?.peer?.subscription : settings?.limits?.client?.subscription + +export const messageLimitsFor = (settings: Settings | undefined, isPeer: boolean): MessageLimits | undefined => + isPeer ? settings?.limits?.peer?.message : settings?.limits?.message diff --git a/src/utils/settings-config.ts b/src/utils/settings-config.ts index cdf8c68b9..8274d0ee3 100644 --- a/src/utils/settings-config.ts +++ b/src/utils/settings-config.ts @@ -1214,6 +1214,7 @@ export const validateSettings = (settings: Settings): ValidationIssue[] => { const limits = settings.limits if (limits) { validateRateLimitPeriods(limits.event?.rateLimits, 'limits.event.rateLimits', issues) + validateRateLimitPeriods(limits.peer?.message?.rateLimits, 'limits.peer.message.rateLimits', issues) validateRateLimitPeriods(limits.message?.rateLimits, 'limits.message.rateLimits', issues) validateRateLimitPeriods(limits.connection?.rateLimits, 'limits.connection.rateLimits', issues) validateRateLimitPeriods(limits.invoice?.rateLimits, 'limits.invoice.rateLimits', issues) diff --git a/src/utils/stream.ts b/src/utils/stream.ts index e247280ce..c38e27122 100644 --- a/src/utils/stream.ts +++ b/src/utils/stream.ts @@ -17,6 +17,14 @@ export const streamEach = (writeFn: (chunk: any) => void) => }, }) +export const streamEachAsync = (writeFn: (chunk: any) => Promise) => + new Transform({ + objectMode: true, + transform(chunk, _encoding, callback) { + writeFn(chunk).then(() => callback(null), callback) + }, + }) + export const streamFilter = (predicate: (chunk: any) => boolean) => new Transform({ objectMode: true, diff --git a/test/unit/adapters/web-socket-adapter.spec.ts b/test/unit/adapters/web-socket-adapter.spec.ts index 1c5f53d45..35da6f764 100644 --- a/test/unit/adapters/web-socket-adapter.spec.ts +++ b/test/unit/adapters/web-socket-adapter.spec.ts @@ -36,7 +36,8 @@ describe('WebSocketAdapter', () => { client = { on: sandbox.stub().returnsThis(), - send: sandbox.stub(), + // `ws` invokes the send callback once the frame is flushed. + send: sandbox.stub().callsFake((_data: string, callback?: () => void) => callback?.()), close: sandbox.stub(), ping: sandbox.stub(), pong: sandbox.stub(), @@ -240,7 +241,8 @@ describe('WebSocketAdapter', () => { adapter.emit(WebSocketAdapterEvent.Message, message) - expect(client.send).to.have.been.calledOnceWithExactly(JSON.stringify(message)) + expect(client.send).to.have.been.calledOnce + expect(client.send.firstCall.args[0]).to.equal(JSON.stringify(message)) }) it('does nothing when WebSocket is not OPEN', () => { @@ -668,6 +670,34 @@ describe('WebSocketAdapter', () => { ) }) + it('uses peer limits and a separate bucket for a recognized peer', async () => { + const peerPubkey = 'a'.repeat(64) + settingsFactory.returns({ + network: { remoteIpHeader: '' }, + peering: { + enabled: true, + peers: [{ pubkey: peerPubkey, relayUrl: 'wss://peer.example.com' }], + }, + limits: { + message: { rateLimits: [{ period: 60000, rate: 1 }], ipWhitelist: [] }, + peer: { message: { rateLimits: [{ period: 60000, rate: 100 }], ipWhitelist: [] } }, + }, + }) + adapter.addAuthenticatedPubkey(peerPubkey, '1'.repeat(64)) + adapter.setPeerPubkey(peerPubkey) + + const hitStub = sandbox.stub().resolves({ limited: false }) + slidingWindowRateLimiter.returns({ hit: hitStub }) + + await onMessage(Buffer.from(JSON.stringify(['REQ', 'sub1', { kinds: [1] }]))) + + expect(hitStub).to.have.been.calledOnceWithExactly( + '127.0.0.1:peer-message:REQ:60000', + 1, + { period: 60000, rate: 100 }, + ) + }) + // A closed set of types, or a client could rotate invented ones and keep // every bucket empty. it('buckets an invented message type with everything else unrecognised', async () => { @@ -803,6 +833,40 @@ describe('WebSocketAdapter', () => { expect(client.send).not.to.have.been.called }) + it('accepts a REQ with more than ten filters so peer limits can apply', async () => { + client.readyState = WebSocket.OPEN + settingsFactory.returns({ network: { remoteIpHeader: '' }, limits: { message: {} } }) + + const handleMessage = sandbox.stub().resolves() + createMessageHandler.returns({ handleMessage }) + + const messageCall = client.on.getCalls().find((call: any) => call.args[0] === 'message') + const onMessage = messageCall.args[1] + const filters = Array.from({ length: 11 }, (_, index) => ({ kinds: [index + 1] })) + + await onMessage(Buffer.from(JSON.stringify(['REQ', 'sub-1', ...filters]))) + + expect(handleMessage).to.have.been.calledOnce + expect(client.send).not.to.have.been.called + }) + + it('accepts a COUNT with more than ten filters so peer limits can apply', async () => { + client.readyState = WebSocket.OPEN + settingsFactory.returns({ network: { remoteIpHeader: '' }, limits: { message: {} } }) + + const handleMessage = sandbox.stub().resolves() + createMessageHandler.returns({ handleMessage }) + + const messageCall = client.on.getCalls().find((call: any) => call.args[0] === 'message') + const onMessage = messageCall.args[1] + const filters = Array.from({ length: 11 }, (_, index) => ({ kinds: [index + 1] })) + + await onMessage(Buffer.from(JSON.stringify(['COUNT', 'q-1', ...filters]))) + + expect(handleMessage).to.have.been.calledOnce + expect(client.send).not.to.have.been.called + }) + it('returns early when no handler is found for message', async () => { client.readyState = WebSocket.OPEN @@ -1067,6 +1131,43 @@ describe('WebSocketAdapter', () => { }) }) + describe('waitForWritable', () => { + it('resolves immediately when nothing has been sent', async () => { + await adapter.waitForWritable() + + expect(client.send).not.to.have.been.called + }) + + it('resolves once the send callback reports the frame as flushed', async () => { + let sendCallback: (() => void) | undefined + ;(client.send as Sinon.SinonStub).callsFake((_data: string, callback: () => void) => { + sendCallback = callback + }) + + adapter.emit(WebSocketAdapterEvent.Message, ['NOTICE', 'hello']) + + let resolved = false + const pending = adapter.waitForWritable().then(() => { + resolved = true + }) + await Promise.resolve() + expect(resolved).to.equal(false) + + sendCallback?.() + await pending + + expect(resolved).to.equal(true) + }) + + it('resolves immediately when the socket is not open', async () => { + client.readyState = WebSocket.CLOSED + + await adapter.waitForWritable() + + expect(client.send).not.to.have.been.called + }) + }) + describe('NIP-77 negentropy sessions', () => { const emptyStorage = () => { const storage = new NegentropyStorageVector() diff --git a/test/unit/handlers/count-message-handler.spec.ts b/test/unit/handlers/count-message-handler.spec.ts index d48581afa..367ba39c2 100644 --- a/test/unit/handlers/count-message-handler.spec.ts +++ b/test/unit/handlers/count-message-handler.spec.ts @@ -27,6 +27,7 @@ describe('CountMessageHandler', () => { } as any webSocket = new EventEmitter() as any + webSocket.getPeerPubkey = sandbox.stub().returns(undefined) handler = new CountMessageHandler(webSocket, eventRepository, () => ({ limits: { @@ -94,6 +95,37 @@ describe('CountMessageHandler', () => { expect(webSocketOnMessageStub.firstCall.args[0][1]).to.equal('q1') }) + it('uses peer limits instead of client limits for a recognized peer', async () => { + handler = new CountMessageHandler(webSocket, eventRepository, () => ({ + limits: { + client: { subscription: { maxFilters: 1, maxSubscriptionIdLength: 256 } }, + peer: { subscription: { maxFilters: 5, maxSubscriptionIdLength: 256 } }, + }, + }) as Settings) + ;(webSocket.getPeerPubkey as Sinon.SinonStub).returns('a'.repeat(64)) + + const message = [MessageType.COUNT, 'q1', { kinds: [1] }, { kinds: [2] }] as any + + await handler.handleMessage(message) + + expect(eventRepository.countByFilters).to.have.been.calledOnceWithExactly([{ kinds: [1] }, { kinds: [2] }]) + expect(webSocketOnMessageStub).to.have.been.calledOnceWithExactly([MessageType.COUNT, 'q1', { count: 7 }]) + }) + + it('bounds raw filters before deduplicating them', async () => { + handler = new CountMessageHandler(webSocket, eventRepository, () => ({ + limits: { client: { subscription: { maxFilters: 10, maxSubscriptionIdLength: 256 } } }, + }) as Settings) + const duplicate = { kinds: [1] } + const message = [MessageType.COUNT, 'q1', ...Array.from({ length: 11 }, () => duplicate)] as any + + await handler.handleMessage(message) + + expect(eventRepository.countByFilters).not.to.have.been.called + expect(webSocketOnMessageStub).to.have.been.calledOnce + expect(webSocketOnMessageStub.firstCall.args[0][0]).to.equal(MessageType.CLOSED) + }) + it('returns CLOSED when a filter has too many values', async () => { handler = new CountMessageHandler(webSocket, eventRepository, () => ({ limits: { diff --git a/test/unit/handlers/neg-open-message-handler.spec.ts b/test/unit/handlers/neg-open-message-handler.spec.ts index 6212ce5bd..f91e4ad52 100644 --- a/test/unit/handlers/neg-open-message-handler.spec.ts +++ b/test/unit/handlers/neg-open-message-handler.spec.ts @@ -65,6 +65,7 @@ describe('NegOpenMessageHandler', () => { webSocket = new EventEmitter() as any webSocket.getNegentropySessions = () => sessions webSocket.getAuthenticatedPubkeys = () => new Set() + webSocket.getPeerPubkey = () => undefined emitted = sandbox.stub() webSocket.on(WebSocketAdapterEvent.Message, emitted) @@ -214,6 +215,22 @@ describe('NegOpenMessageHandler', () => { expect(emitted.firstCall.args[0][2]).to.equal('blocked: too many filter values (max 2)') expect(findReconciliationItems).to.not.have.been.called }) + + it('uses peer limits instead of client limits for a recognized peer', async () => { + settings = { + nip77: { enabled: true }, + limits: { + client: { subscription: { maxFilterValues: 1 } }, + peer: { subscription: { maxFilterValues: 10 } }, + }, + } as Settings + webSocket.getPeerPubkey = () => 'a'.repeat(64) + + await open('sync1', { authors: ['a', 'b'] }) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + expect(findReconciliationItems).to.have.been.called + }) }) describe('NIP-42', () => { diff --git a/test/unit/handlers/subscribe-message-handler.spec.ts b/test/unit/handlers/subscribe-message-handler.spec.ts index b14562cc9..af767f411 100644 --- a/test/unit/handlers/subscribe-message-handler.spec.ts +++ b/test/unit/handlers/subscribe-message-handler.spec.ts @@ -60,6 +60,8 @@ describe('SubscribeMessageHandler', () => { webSocket = new EventEmitter() as any webSocketGetSubscriptionsStub = sandbox.stub().returns(subscriptions) webSocket.getSubscriptions = webSocketGetSubscriptionsStub + webSocket.getPeerPubkey = sandbox.stub().returns(undefined) + webSocket.waitForWritable = sandbox.stub().resolves() settingsFactory = sandbox.stub() stream = new PassThrough({ objectMode: true, @@ -123,6 +125,20 @@ describe('SubscribeMessageHandler', () => { expect(webSocketOnMessageStub).to.have.been.calledOnceWithExactly(['NOTICE', 'Subscription rejected: reason']) }) + it('bounds raw filters before deduplicating them', async () => { + settingsFactory.returns({ limits: { client: { subscription: { maxFilters: 10 } } } }) + const duplicate = { kinds: [1] } + message = [MessageType.REQ, subscriptionId, ...Array.from({ length: 11 }, () => duplicate)] as any + + await handler.handleMessage(message) + + expect(webSocketOnMessageStub).to.have.been.calledOnceWithExactly([ + 'NOTICE', + 'Subscription rejected: Too many filters: Number of filters per susbscription must be less then or equal to 10', + ]) + expect(webSocketOnSubscribeStub).not.to.have.been.called + }) + it('emits subscribe event if subscription is accepted', async () => { canSubscribeStub.returns(undefined) message = [MessageType.REQ, subscriptionId, ...filters] as any @@ -544,6 +560,24 @@ describe('SubscribeMessageHandler', () => { expect(eventRepositoryFindByFiltersStub).to.have.been.calledOnceWithExactly(filters) }) + it('passes an uncapped defaultLimit to the repository for a peer', async () => { + isClientSubscribedToEventStub.returns(always(false)) + ;(webSocket.getPeerPubkey as Sinon.SinonStub).returns('a'.repeat(64)) + settingsFactory.returns({ limits: { peer: { subscription: { maxLimit: 0 } } } }) + + const promise = (handler as any).fetchAndSend(subscriptionId, filters) + + stream.write(toDbEvent(event)) + stream.end() + + await promise + + expect(eventRepositoryFindByFiltersStub).to.have.been.calledOnceWithExactly(filters, { + defaultLimit: 0, + excludeDeleted: true, + }) + }) + it('sends event if client is subscribed', async () => { isClientSubscribedToEventStub.returns(always(true)) @@ -756,6 +790,34 @@ describe('SubscribeMessageHandler', () => { expect((handler as any).canSubscribe(subscriptionId, filters)).to.be.undefined }) + it('uses peer limits instead of client limits for a recognized peer', () => { + settingsFactory.returns({ + limits: { + client: { subscription: { maxSubscriptions: 1 } }, + peer: { subscription: { maxSubscriptions: 5 } }, + }, + }) + ;(webSocket.getPeerPubkey as Sinon.SinonStub).returns('a'.repeat(64)) + subscriptions.set('other-1', []) + subscriptions.set('other-2', []) + + expect((handler as any).canSubscribe(subscriptionId, filters)).to.be.undefined + }) + + it('still applies the client limit to an unauthenticated connection', () => { + settingsFactory.returns({ + limits: { + client: { subscription: { maxSubscriptions: 1 } }, + peer: { subscription: { maxSubscriptions: 5 } }, + }, + }) + subscriptions.set('other-1', []) + + expect((handler as any).canSubscribe(subscriptionId, filters)).to.equal( + 'Too many subscriptions: Number of subscriptions must be less than or equal to 1', + ) + }) + it('returns undefined if max subscription limit is disabled', () => { settingsFactory.returns({ limits: { diff --git a/test/unit/repositories/event-repository.spec.ts b/test/unit/repositories/event-repository.spec.ts index 8cba85b66..503d5a1bf 100644 --- a/test/unit/repositories/event-repository.spec.ts +++ b/test/unit/repositories/event-repository.spec.ts @@ -51,6 +51,37 @@ describe('EventRepository', () => { expect(() => repository.findByFilters([])).to.throw(Error, 'Filters cannot be empty') }) + it('omits LIMIT when defaultLimit is 0 (peer full feed)', () => { + const query = repository.findByFilters([{}], { defaultLimit: 0 }).toString() + + expect(query).to.equal('select * from "events" order by "event_created_at" asc, "event_id" asc') + }) + + it('uses a positive defaultLimit for filters that omit one', () => { + const query = repository.findByFilters([{}], { defaultLimit: 1000 }).toString() + + expect(query).to.equal('select * from "events" order by "event_created_at" asc, "event_id" asc limit 1000') + }) + + it('keeps an explicit filter limit over defaultLimit', () => { + const query = repository.findByFilters([{ limit: 25 }], { defaultLimit: 1000 }).toString() + + expect(query).to.equal('select * from "events" order by "event_created_at" DESC, "event_id" asc limit 25') + }) + + it('excludes deleted and expired rows before applying the peer limit', () => { + sandbox.stub(Date, 'now').returns(1_700_000_000_000) + + const query = repository + .findByFilters([{}], { defaultLimit: 100, excludeDeleted: true }) + .toString() + + expect(query).to.include('"deleted_at" is null') + expect(query).to.include('"expires_at" is null') + expect(query).to.include('"expires_at" > 1700000000') + expect(query).to.include('limit 100') + }) + describe('1 filter', () => { it('selects all events', () => { const filters = [{}] diff --git a/test/unit/schemas/message-schema.spec.ts b/test/unit/schemas/message-schema.spec.ts index f60c45271..3610d3079 100644 --- a/test/unit/schemas/message-schema.spec.ts +++ b/test/unit/schemas/message-schema.spec.ts @@ -104,12 +104,13 @@ describe('NIP-01', () => { expect(result).to.have.property('error').that.is.not.undefined }) - it('returns error if there are too many filters', () => { + it('accepts more filters than the client cap so per-connection limits decide', () => { ;(message as any[]).splice(2, 2) ;(message as any[]).push(...range(0, 11).map(() => ({}))) const result = validateSchema(messageSchema)(message) - expect(result).to.have.property('error').that.is.not.undefined + expect(result.error).to.be.undefined + expect(result).to.have.deep.property('value', message) }) }) @@ -159,12 +160,13 @@ describe('NIP-01', () => { expect(result).to.have.property('error').that.is.not.undefined }) - it('returns error if there are too many filters', () => { + it('accepts more filters than the client cap so per-connection limits decide', () => { ;(message as any[]).splice(2, 1) ;(message as any[]).push(...range(0, 11).map(() => ({}))) const result = validateSchema(messageSchema)(message) - expect(result).to.have.property('error').that.is.not.undefined + expect(result.error).to.be.undefined + expect(result).to.have.deep.property('value', message) }) }) diff --git a/test/unit/utils/settings-config.spec.ts b/test/unit/utils/settings-config.spec.ts index 7430b2e39..3c723f05c 100644 --- a/test/unit/utils/settings-config.spec.ts +++ b/test/unit/utils/settings-config.spec.ts @@ -400,6 +400,10 @@ describe('settings-config', () => { issuePath: 'limits.message.rateLimits[0].period', limits: { message: { rateLimits: [{ period: 0, rate: 1 }] } }, }, + { + issuePath: 'limits.peer.message.rateLimits[0].period', + limits: { peer: { message: { rateLimits: [{ period: 0, rate: 1 }] } } }, + }, { issuePath: 'limits.connection.rateLimits[0].period', limits: { connection: { rateLimits: [{ period: -1, rate: 1 }] } },