diff --git a/.changeset/peering-request-admin.md b/.changeset/peering-request-admin.md new file mode 100644 index 000000000..7bb176f5e --- /dev/null +++ b/.changeset/peering-request-admin.md @@ -0,0 +1,11 @@ +--- +"nostream": minor +--- + +feat(admin): account for peer on-demand requests + +Records on-demand `REQ`, `COUNT`, and `NEG-OPEN` requests and the events served +for each authenticated peer pubkey in Redis. The admin console's Peering table +shows the request and event-served counts per peer, and the diagnostics export +includes the same counters. Accounting is best-effort and never fails the +request it describes. diff --git a/docs/PEERING.md b/docs/PEERING.md index d5aa328f4..7fa82e9de 100644 --- a/docs/PEERING.md +++ b/docs/PEERING.md @@ -179,6 +179,11 @@ unavailable. The same request pattern from a non-peer connection is handled as ordinary client traffic and is not given the peer budget. +Request counts and the number of events served are recorded per peer pubkey in +Redis and shown in the admin console's Peering table alongside sync state. The +same counters are included in diagnostics exports. Accounting is best-effort +and never fails the request it describes. + 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/admin/assets/admin-ops.js b/resources/admin/assets/admin-ops.js index 6a86563ab..6d423ae66 100644 --- a/resources/admin/assets/admin-ops.js +++ b/resources/admin/assets/admin-ops.js @@ -145,13 +145,14 @@ const renderPeering = (rows) => { if (!Array.isArray(rows) || !rows.length) { elements.peeringBody.innerHTML = - 'No peering peers configured.' + 'No peering peers configured.' return } elements.peeringBody.innerHTML = rows .map((row) => { const state = row.state + const requests = row.requests const status = state?.status ? String(state.status) : 'no sync yet' const title = [row.relayUrl, row.pubkey, status].filter(Boolean).join(' · ') return ` @@ -160,6 +161,8 @@ ${escapeHtml(formatTime(state?.lastReconciledAt))} ${escapeHtml(formatLag(state?.lagSeconds))} ${escapeHtml(state?.missingEvents ?? '—')} + ${escapeHtml(requests?.requests ?? 0)} + ${escapeHtml(requests?.eventsServed ?? 0)} ${peeringSignalBadge(row)} ` }) diff --git a/resources/admin/dashboard.html b/resources/admin/dashboard.html index 4631f0147..40771553e 100644 --- a/resources/admin/dashboard.html +++ b/resources/admin/dashboard.html @@ -239,11 +239,13 @@

Peering

Last reconciliation Lag Missing + Requests + Events served Signal - No peering peers configured. + No peering peers configured. diff --git a/src/adapters/web-socket-adapter.ts b/src/adapters/web-socket-adapter.ts index 2299b9745..9d175180e 100644 --- a/src/adapters/web-socket-adapter.ts +++ b/src/adapters/web-socket-adapter.ts @@ -1,40 +1,40 @@ import cluster from 'cluster' -import { EventEmitter } from 'stream' import { IncomingMessage as IncomingHttpMessage } from 'http' +import { SocketAddress } from 'net' +import { EventEmitter } from 'stream' import { WebSocket } from 'ws' import { ZodError } from 'zod' - +import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters' import { ContextMetadata, Factory } from '../@types/base' -import { - createAuthChallengeMessage, - createClosedMessage, - createNegErrMessage, - createNoticeMessage, - createOutgoingEventMessage, -} from '../utils/messages' +import { Event } from '../@types/event' import { IAbortable, IMessageHandler } from '../@types/message-handlers' import { IncomingMessage, MessageType, OutgoingMessage } from '../@types/messages' import { INegentropySessions } from '../@types/negentropy' -import { NegentropySessions } from '../utils/negentropy-sessions' -import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters' +import { Settings } from '../@types/settings' import { SubscriptionFilter, SubscriptionId } from '../@types/subscription' +import { IRateLimiter } from '../@types/utils' import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter' -import { attemptValidation } from '../utils/validation' import { ContextMetadataKey } from '../constants/base' import { createLogger } from '../factories/logger-factory' +import { messageSchema } from '../schemas/message-schema' import { recordWebsocketConnectionClosed, recordWebsocketConnectionOpened } from '../telemetry/event-metrics' -import { Event } from '../@types/event' +import { isEventMatchingFilter } from '../utils/event' +import { isHidden } from '../utils/hidden-content-cache' import { getRemoteAddress } from '../utils/http' +import { + createAuthChallengeMessage, + createClosedMessage, + createNegErrMessage, + createNoticeMessage, + createOutgoingEventMessage, +} from '../utils/messages' +import { NegentropySessions } from '../utils/negentropy-sessions' 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' -import { messageSchema } from '../schemas/message-schema' -import { Settings } from '../@types/settings' -import { SocketAddress } from 'net' +import { isConfiguredPeeringPeer } from '../utils/peering-auth' +import { PeeringRequestStatsStore } from '../utils/peering-request-stats' +import { attemptValidation } from '../utils/validation' const logger = createLogger('web-socket-adapter') @@ -90,6 +90,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter private readonly createMessageHandler: Factory, private readonly rateLimiter: Factory, private readonly settings: Factory, + private readonly requestStats?: PeeringRequestStatsStore, ) { super() this.alive = true @@ -110,7 +111,9 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter this.client .on('error', (error) => { if (error.name === 'RangeError' && error.message === 'Max payload size exceeded') { - logger.error(`web-socket-adapter: client ${this.clientId} (${this.getClientAddress()}) sent payload too large`) + logger.error( + `web-socket-adapter: client ${this.clientId} (${this.getClientAddress()}) sent payload too large`, + ) } else if (error.name === 'RangeError' && error.message === 'Invalid WebSocket frame: RSV1 must be clear') { logger(`client ${this.clientId} (${this.getClientAddress()}) enabled compression`) } else { @@ -206,6 +209,13 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter this.pendingSend = undefined } }) + + if (message[0] === MessageType.EVENT) { + const peerPubkey = this.getPeerPubkey() + if (peerPubkey) { + void this.requestStats?.addEventsServed(peerPubkey, 1) + } + } } public waitForWritable(): Promise { diff --git a/src/controllers/admin/get-admin-peering-controller.ts b/src/controllers/admin/get-admin-peering-controller.ts index 3ae4d470e..4bd00f118 100644 --- a/src/controllers/admin/get-admin-peering-controller.ts +++ b/src/controllers/admin/get-admin-peering-controller.ts @@ -3,6 +3,7 @@ import { Request, Response } from 'express' import { IController } from '../../@types/controllers' import { Settings } from '../../@types/settings' import { mergePeeringRows } from '../../utils/admin-peering' +import { PeeringRequestStatsStore } from '../../utils/peering-request-stats' import { PeeringSyncStateStore } from '../../utils/peering-sync-state' import { loadMergedSettings } from '../../utils/settings-config' @@ -10,11 +11,16 @@ export class GetAdminPeeringController implements IController { public constructor( private readonly stateStore: PeeringSyncStateStore, private readonly getSettings: () => Settings = loadMergedSettings, + private readonly requestStatsStore?: PeeringRequestStatsStore, ) {} public async handleRequest(_request: Request, response: Response): Promise { - const [states, settings] = await Promise.all([this.stateStore.list(), this.getSettings()]) - const peers = mergePeeringRows(settings, states) + const [states, settings, requestStats] = await Promise.all([ + this.stateStore.list(), + this.getSettings(), + this.requestStatsStore?.list() ?? [], + ]) + const peers = mergePeeringRows(settings, states, undefined, requestStats) response.status(200).setHeader('content-type', 'application/json').send({ peers }) } diff --git a/src/factories/controllers/get-admin-peering-controller-factory.ts b/src/factories/controllers/get-admin-peering-controller-factory.ts index bfc2bd47f..7eb74ac64 100644 --- a/src/factories/controllers/get-admin-peering-controller-factory.ts +++ b/src/factories/controllers/get-admin-peering-controller-factory.ts @@ -2,6 +2,7 @@ import { IController } from '../../@types/controllers' import { RedisAdapter } from '../../adapters/redis-adapter' import { getCacheClient } from '../../cache/client' import { GetAdminPeeringController } from '../../controllers/admin/get-admin-peering-controller' +import { PeeringRequestStatsStore } from '../../utils/peering-request-stats' import { PeeringSyncStateStore } from '../../utils/peering-sync-state' export const createGetAdminPeeringController = (): IController => { @@ -9,8 +10,10 @@ export const createGetAdminPeeringController = (): IController => { // adapter keeps repeated dashboard polls from stacking listeners on the // shared cache client. if (!sharedController) { - const stateStore = new PeeringSyncStateStore(new RedisAdapter(getCacheClient())) - sharedController = new GetAdminPeeringController(stateStore) + const cache = new RedisAdapter(getCacheClient()) + const stateStore = new PeeringSyncStateStore(cache) + const requestStatsStore = new PeeringRequestStatsStore(cache) + sharedController = new GetAdminPeeringController(stateStore, undefined, requestStatsStore) } return sharedController diff --git a/src/factories/message-handler-factory.ts b/src/factories/message-handler-factory.ts index eb41efbd0..3fe597969 100644 --- a/src/factories/message-handler-factory.ts +++ b/src/factories/message-handler-factory.ts @@ -20,6 +20,7 @@ import { NegOpenMessageHandler } from '../handlers/neg-open-message-handler' import { SubscribeMessageHandler } from '../handlers/subscribe-message-handler' import { UnsubscribeMessageHandler } from '../handlers/unsubscribe-message-handler' import { UserSubscriptionRepository } from '../repositories/user-subscription-repository' +import { PeeringRequestStatsStore } from '../utils/peering-request-stats' import { eventStrategyFactory } from './event-strategy-factory' import { rateLimiterFactory } from './rate-limiter-factory' import { createSettings } from './settings-factory' @@ -41,6 +42,14 @@ const getUserSubscriptionRepository = (): UserSubscriptionRepository => { return userSubscriptionRepository } +let peerRequestStatsStore: PeeringRequestStatsStore | undefined = undefined +export const getPeerRequestStatsStore = (): PeeringRequestStatsStore => { + if (!peerRequestStatsStore) { + peerRequestStatsStore = new PeeringRequestStatsStore(getCache()) + } + return peerRequestStatsStore +} + export const messageHandlerFactory = ( eventRepository: IEventRepository, @@ -81,15 +90,28 @@ export const messageHandlerFactory = createSettings, inviteCodeRepository, rateLimiterFactory, + getPeerRequestStatsStore(), ) case MessageType.CLOSE: return new UnsubscribeMessageHandler(adapter) case MessageType.COUNT: - return new CountMessageHandler(adapter, eventRepository, createSettings, rateLimiterFactory) + return new CountMessageHandler( + adapter, + eventRepository, + createSettings, + rateLimiterFactory, + getPeerRequestStatsStore(), + ) case MessageType.AUTH: return new AuthMessageHandler(adapter, createSettings) case MessageType.NEG_OPEN: - return new NegOpenMessageHandler(adapter, eventRepository, createSettings, rateLimiterFactory) + return new NegOpenMessageHandler( + adapter, + eventRepository, + createSettings, + rateLimiterFactory, + getPeerRequestStatsStore(), + ) case MessageType.NEG_MSG: return new NegMsgMessageHandler(adapter, createSettings) case MessageType.NEG_CLOSE: diff --git a/src/factories/websocket-adapter-factory.ts b/src/factories/websocket-adapter-factory.ts index 0d0246ff3..9fef63750 100644 --- a/src/factories/websocket-adapter-factory.ts +++ b/src/factories/websocket-adapter-factory.ts @@ -1,6 +1,6 @@ import { IncomingMessage } from 'http' import { WebSocket } from 'ws' - +import { IWebSocketServerAdapter } from '../@types/adapters' import { IDvmJobRepository, IEventRepository, @@ -9,11 +9,10 @@ import { IReportRepository, IUserRepository, } from '../@types/repositories' -import { createSettings } from './settings-factory' -import { IWebSocketServerAdapter } from '../@types/adapters' -import { messageHandlerFactory } from './message-handler-factory' -import { rateLimiterFactory } from './rate-limiter-factory' import { WebSocketAdapter } from '../adapters/web-socket-adapter' +import { getPeerRequestStatsStore, messageHandlerFactory } from './message-handler-factory' +import { rateLimiterFactory } from './rate-limiter-factory' +import { createSettings } from './settings-factory' export const webSocketAdapterFactory = ( @@ -39,4 +38,5 @@ export const webSocketAdapterFactory = ), rateLimiterFactory, createSettings, + getPeerRequestStatsStore(), ) diff --git a/src/handlers/count-message-handler.ts b/src/handlers/count-message-handler.ts index a08dfbe07..1f5140b4f 100644 --- a/src/handlers/count-message-handler.ts +++ b/src/handlers/count-message-handler.ts @@ -15,6 +15,7 @@ import { createClosedMessage, createCountResultMessage } from '../utils/messages import { isCountAuthorized } from '../utils/nip42' import { subscriptionLimitsFor } from '../utils/peer-limits' import { getPeerRequestRetryAfterMs } from '../utils/peer-requests' +import { PeeringRequestStatsStore } from '../utils/peering-request-stats' const debug = createLogger('count-message-handler') @@ -24,6 +25,7 @@ export class CountMessageHandler implements IMessageHandler { private readonly eventRepository: IEventRepository, private readonly settings: () => Settings, private readonly rateLimiter?: Factory, + private readonly requestStats?: PeeringRequestStatsStore, ) {} public async handleMessage(message: CountMessage): Promise { @@ -47,6 +49,8 @@ export class CountMessageHandler implements IMessageHandler { ) return } + + void this.requestStats?.recordRequest(peerPubkey, 'COUNT') } // Some clients send the same filter more than once. Bound the raw list diff --git a/src/handlers/neg-open-message-handler.ts b/src/handlers/neg-open-message-handler.ts index b56890493..d8b9c1de3 100644 --- a/src/handlers/neg-open-message-handler.ts +++ b/src/handlers/neg-open-message-handler.ts @@ -16,6 +16,7 @@ import { isReconciliationAuthorized } from '../utils/nip42' import { getNip77Limits, isNip77Active, NEGENTROPY_FRAME_SIZE_LIMIT } from '../utils/nip77' import { subscriptionLimitsFor } from '../utils/peer-limits' import { getPeerRequestRetryAfterMs } from '../utils/peer-requests' +import { PeeringRequestStatsStore } from '../utils/peering-request-stats' const logger = createLogger('neg-open-message-handler') @@ -29,6 +30,7 @@ export class NegOpenMessageHandler implements IMessageHandler { private readonly eventRepository: IEventRepository, private readonly settings: () => Settings, private readonly rateLimiter?: Factory, + private readonly requestStats?: PeeringRequestStatsStore, ) {} public async handleMessage(message: NegOpenMessage): Promise { @@ -83,6 +85,10 @@ export class NegOpenMessageHandler implements IMessageHandler { return } + if (peerPubkey) { + void this.requestStats?.recordRequest(peerPubkey, 'NEG-OPEN') + } + let items: ReconciliationItem[] try { // One row past the limit tells us the set is too big without loading all of it. diff --git a/src/handlers/subscribe-message-handler.ts b/src/handlers/subscribe-message-handler.ts index fd0ace6f6..cd81daeb4 100644 --- a/src/handlers/subscribe-message-handler.ts +++ b/src/handlers/subscribe-message-handler.ts @@ -1,7 +1,18 @@ import { anyPass, equals, isNil, map, omit, propSatisfies, uniqWith } from 'ramda' import { addAbortSignal } from 'stream' import { pipeline } from 'stream/promises' - +import { IWebSocketAdapter } from '../@types/adapters' +import { Factory, Pubkey } from '../@types/base' +import { Event } from '../@types/event' +import { IAbortable, IMessageHandler } from '../@types/message-handlers' +import { SubscribeMessage } from '../@types/messages' +import { IEventRepository, IInviteCodeRepository } from '../@types/repositories' +import { Settings } from '../@types/settings' +import { SubscriptionFilter, SubscriptionId } from '../@types/subscription' +import { IRateLimiter } from '../@types/utils' +import { WebSocketAdapterEvent } from '../constants/adapter' +import { createLogger } from '../factories/logger-factory' +import { getPublicKey, getRelayPrivateKey, isEventMatchingFilter, isExpiredEvent, toNostrEvent } from '../utils/event' import { countFilterValues } from '../utils/filter' import { createClosedMessage, @@ -10,23 +21,12 @@ import { createOutgoingEventMessage, } from '../utils/messages' import { createReadAuthorizationGuard, isSubscriptionAuthRequired } from '../utils/nip42' -import { getPublicKey, getRelayPrivateKey, isEventMatchingFilter, isExpiredEvent, toNostrEvent } from '../utils/event' -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 { streamEachAsync, streamEnd, streamFilter, streamMap } from '../utils/stream' -import { SubscriptionFilter, SubscriptionId } from '../@types/subscription' -import { createLogger } from '../factories/logger-factory' -import { Event } from '../@types/event' -import { Factory, Pubkey } from '../@types/base' -import { IRateLimiter } from '../@types/utils' -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' import { getPeerRequestRetryAfterMs } from '../utils/peer-requests' +import { PeeringRequestStatsStore } from '../utils/peering-request-stats' +import { streamEachAsync, streamEnd, streamFilter, streamMap } from '../utils/stream' const logger = createLogger('subscribe-message-handler') @@ -42,6 +42,7 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { private readonly settings: () => Settings, private readonly inviteCodeRepository: IInviteCodeRepository, private readonly rateLimiter: Factory, + private readonly requestStats?: PeeringRequestStatsStore, ) { this.abortController = new AbortController() } @@ -75,6 +76,8 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { ) return } + + void this.requestStats?.recordRequest(peerPubkey, 'REQ') } // Bound before uniqWith(): deduplicating an unbounded list is O(n^2). The @@ -91,7 +94,7 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { // NIP-50: strip search from filters when disabled so isEventMatchingFilter ignores it const nip50Enabled = this.settings()?.nip50?.enabled ?? false - const filters = nip50Enabled ? rawFilters : rawFilters.map(omit(['search'])) as SubscriptionFilter[] + const filters = nip50Enabled ? rawFilters : (rawFilters.map(omit(['search'])) as SubscriptionFilter[]) const reason = this.canSubscribe(subscriptionId, filters) if (reason) { @@ -258,9 +261,8 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { } // NIP-42: drop restricted-kind events the client isn't authorized to read. - const isReadAuthorized = createReadAuthorizationGuard( - this.settings(), - () => this.webSocket.getAuthenticatedPubkeys(), + const isReadAuthorized = createReadAuthorizationGuard(this.settings(), () => + this.webSocket.getAuthenticatedPubkeys(), ) // A peer's omitted `limit` means the full stored feed, not the client-sized @@ -304,10 +306,7 @@ 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 = subscriptionLimitsFor( - this.settings(), - this.webSocket.getPeerPubkey() !== undefined, - ) + const subscriptionLimits = subscriptionLimitsFor(this.settings(), this.webSocket.getPeerPubkey() !== undefined) if (existingSubscription?.length && equals(filters, existingSubscription)) { return `Duplicate subscription ${subscriptionId}: Ignoring` diff --git a/src/utils/admin-diagnostics-export.ts b/src/utils/admin-diagnostics-export.ts index ade762e62..9d588c609 100644 --- a/src/utils/admin-diagnostics-export.ts +++ b/src/utils/admin-diagnostics-export.ts @@ -7,6 +7,7 @@ import { getCacheClient } from '../cache/client' import { collectAdminHealthSnapshot } from './admin-health' import { getRedactedAdminNotifications } from './admin-notifications-settings' import { AdminOpsStore } from './admin-ops-store' +import { PeeringRequestStatsStore } from './peering-request-stats' import { PeeringSyncStateStore } from './peering-sync-state' import { RelayProbeSnapshotStore } from './relay-probe-snapshot' import { loadMergedSettings } from './settings-config' @@ -24,6 +25,7 @@ export interface AdminDiagnosticsExport { operations: { workers: Awaited> peering: Awaited> + peeringRequests: Awaited> } } @@ -44,12 +46,14 @@ export const buildAdminDiagnosticsExport = async ( const opsStore = new AdminOpsStore(cache) const snapshotStore = new RelayProbeSnapshotStore(cache) const peeringStore = new PeeringSyncStateStore(cache) + const peeringRequestStore = new PeeringRequestStatsStore(cache) - const [health, workers, latest, peering] = await Promise.all([ + const [health, workers, latest, peering, peeringRequests] = await Promise.all([ collectAdminHealthSnapshot(), opsStore.listWorkerHeartbeats(), snapshotStore.getLatest(), peeringStore.list(), + peeringRequestStore.list(), ]) const settings = redactSettingsSecrets(getSettings()) as Settings @@ -61,6 +65,6 @@ export const buildAdminDiagnosticsExport = async ( settings, notifications: getRedactedAdminNotifications(), networkHealth: { latest }, - operations: { workers, peering }, + operations: { workers, peering, peeringRequests }, } } diff --git a/src/utils/admin-peering.ts b/src/utils/admin-peering.ts index ca37f8abd..b35ab09ca 100644 --- a/src/utils/admin-peering.ts +++ b/src/utils/admin-peering.ts @@ -1,6 +1,7 @@ import { PeeringPeerSyncState } from '../@types/peering' import { Settings } from '../@types/settings' import { parsePeeringPeerPubkey } from './peering-auth' +import { PeeringRequestStats } from './peering-request-stats' import { isPeeringSyncStateStale, PEERING_SYNC_STATE_STALE_AFTER_SECONDS } from './peering-sync-state' export interface AdminPeeringRow { @@ -10,14 +11,17 @@ export interface AdminPeeringRow { enabled: boolean stale: boolean state?: PeeringPeerSyncState + requests?: PeeringRequestStats } export const mergePeeringRows = ( settings: Settings, states: PeeringPeerSyncState[], staleAfterSeconds = PEERING_SYNC_STATE_STALE_AFTER_SECONDS, + requestStats: PeeringRequestStats[] = [], ): AdminPeeringRow[] => { const statesByPubkey = new Map(states.map((state) => [state.pubkey.toLowerCase(), state])) + const requestsByPubkey = new Map(requestStats.map((stats) => [stats.pubkey.toLowerCase(), stats])) const rows: AdminPeeringRow[] = [] const matched = new Set() @@ -36,6 +40,7 @@ export const mergePeeringRows = ( enabled, stale: enabled && (!state || isPeeringSyncStateStale(state.updatedAt, staleAfterSeconds)), state, + requests: requestsByPubkey.get(normalized), }) } @@ -53,6 +58,7 @@ export const mergePeeringRows = ( enabled: false, stale: isPeeringSyncStateStale(state.updatedAt, staleAfterSeconds), state, + requests: requestsByPubkey.get(state.pubkey.toLowerCase()), }) } diff --git a/src/utils/peering-request-stats.ts b/src/utils/peering-request-stats.ts new file mode 100644 index 000000000..55b3fe7ba --- /dev/null +++ b/src/utils/peering-request-stats.ts @@ -0,0 +1,238 @@ +import { ICacheAdapter } from '../@types/adapters' +import { createLogger } from '../factories/logger-factory' + +const logger = createLogger('peering-request-stats') + +export const PEERING_REQUEST_STATS_INDEX_KEY = 'admin:peering:requests:index' +export const PEERING_REQUEST_STATS_TTL_SECONDS = 7 * 24 * 60 * 60 + +export interface PeeringRequestStats { + pubkey: string + requests: number + eventsServed: number + lastRequestAt?: string + lastRequestType?: string +} + +const statsKey = (pubkey: string): string => `admin:peering:requests:${pubkey.toLowerCase()}` + +const RECORD_SCRIPT = ` +local key = KEYS[1] +local index = KEYS[2] +redis.call('HINCRBY', key, 'requests', tonumber(ARGV[1])) +redis.call('HINCRBY', key, 'eventsServed', tonumber(ARGV[2])) +redis.call('HSET', key, 'pubkey', ARGV[3]) +if ARGV[4] ~= '' then + redis.call('HSET', key, 'lastRequestAt', ARGV[4], 'lastRequestType', ARGV[5]) +end +redis.call('EXPIRE', key, tonumber(ARGV[6])) +redis.call('SADD', index, ARGV[3]) +redis.call('EXPIRE', index, tonumber(ARGV[6])) +return 1 +` + +const READ_SCRIPT = ` +local raw = redis.call('HGETALL', KEYS[1]) +if #raw == 0 then + redis.call('SREM', KEYS[2], ARGV[1]) +end +return raw +` + +const parseStats = (raw: unknown): PeeringRequestStats | undefined => { + if (!Array.isArray(raw)) { + return undefined + } + + const fields = new Map() + for (let index = 0; index + 1 < raw.length; index += 2) { + fields.set(String(raw[index]), String(raw[index + 1])) + } + + const pubkey = fields.get('pubkey') + if (!pubkey) { + return undefined + } + + const requests = Number(fields.get('requests') ?? 0) + const eventsServed = Number(fields.get('eventsServed') ?? 0) + + return { + pubkey: pubkey.toLowerCase(), + requests: Number.isFinite(requests) ? requests : 0, + eventsServed: Number.isFinite(eventsServed) ? eventsServed : 0, + ...(fields.get('lastRequestAt') ? { lastRequestAt: fields.get('lastRequestAt') } : {}), + ...(fields.get('lastRequestType') ? { lastRequestType: fields.get('lastRequestType') } : {}), + } +} + +interface PendingStats { + requests: number + eventsServed: number + lastRequestAt?: string + lastRequestType?: string +} + +const ACCOUNTING_FLUSH_DELAY_MS = 250 +const MAX_CONCURRENT_ACCOUNTING_WRITES = 25 + +export class PeeringRequestStatsStore { + private pending = new Map() + private flushTimer: NodeJS.Timeout | undefined + private flushPromise: Promise | undefined + + public constructor(private readonly cache: ICacheAdapter) {} + + public recordRequest(pubkey: string, type: string): void { + this.enqueue(pubkey, { + requests: 1, + eventsServed: 0, + lastRequestAt: new Date().toISOString(), + lastRequestType: type, + }) + } + + public addEventsServed(pubkey: string, count: number): void { + if (count <= 0) { + return + } + + this.enqueue(pubkey, { requests: 0, eventsServed: count }) + } + + public async list(): Promise { + // Flush the bounded pending batch so a dashboard poll sees the latest + // counters without adding one Redis write per delivered event. + await this.flushPending() + + let pubkeys: string[] + try { + pubkeys = await this.cache.getSetMembers(PEERING_REQUEST_STATS_INDEX_KEY) + } catch (error) { + // A stats-index read failure must not hide the Peering table's sync rows. + logger.warn('failed to read peering request stats index: %o', error) + return [] + } + + const stats: PeeringRequestStats[] = [] + + for (const pubkey of pubkeys) { + try { + // Reading and pruning are one script: a request may recreate the hash + // between an empty HGETALL and a separate SREM, which would otherwise + // drop fresh counters from the index. + const parsed = parseStats( + await this.cache.eval(READ_SCRIPT, [statsKey(pubkey), PEERING_REQUEST_STATS_INDEX_KEY], [pubkey]), + ) + if (parsed) { + stats.push(parsed) + } + } catch (error) { + logger.warn('failed to read peering request stats for %s: %o', pubkey, error) + } + } + + return stats.sort((left, right) => left.pubkey.localeCompare(right.pubkey)) + } + + /** Flush pending counters now. Exposed for admin reads and deterministic tests. */ + public async flushPending(): Promise { + if (this.flushTimer) { + clearTimeout(this.flushTimer) + this.flushTimer = undefined + } + + if (this.flushPromise) { + await this.flushPromise + } + + if (this.pending.size === 0) { + return + } + + const batch = this.pending + this.pending = new Map() + const entries = [...batch.entries()] + + this.flushPromise = (async () => { + for (let offset = 0; offset < entries.length; offset += MAX_CONCURRENT_ACCOUNTING_WRITES) { + await Promise.all( + entries.slice(offset, offset + MAX_CONCURRENT_ACCOUNTING_WRITES).map(([pubkey, stats]) => + this.record(pubkey, stats.requests, stats.eventsServed, { + at: stats.lastRequestAt, + type: stats.lastRequestType, + }), + ), + ) + } + })() + + try { + await this.flushPromise + } finally { + this.flushPromise = undefined + if (this.pending.size > 0) { + this.scheduleFlush() + } + } + } + + private enqueue(pubkey: string, delta: PendingStats): void { + const normalized = pubkey.toLowerCase() + const current = this.pending.get(normalized) ?? { requests: 0, eventsServed: 0 } + + this.pending.set(normalized, { + requests: current.requests + delta.requests, + eventsServed: current.eventsServed + delta.eventsServed, + ...(delta.lastRequestAt + ? { lastRequestAt: delta.lastRequestAt } + : current.lastRequestAt + ? { lastRequestAt: current.lastRequestAt } + : {}), + ...(delta.lastRequestType + ? { lastRequestType: delta.lastRequestType } + : current.lastRequestType + ? { lastRequestType: current.lastRequestType } + : {}), + }) + + this.scheduleFlush() + } + + private scheduleFlush(): void { + if (this.flushTimer || this.flushPromise) { + return + } + + this.flushTimer = setTimeout(() => { + this.flushTimer = undefined + void this.flushPending() + }, ACCOUNTING_FLUSH_DELAY_MS) + this.flushTimer.unref() + } + + private async record( + pubkey: string, + requests: number, + eventsServed: number, + last?: { at?: string; type?: string }, + ): Promise { + try { + await this.cache.eval( + RECORD_SCRIPT, + [statsKey(pubkey), PEERING_REQUEST_STATS_INDEX_KEY], + [ + String(requests), + String(eventsServed), + pubkey.toLowerCase(), + last?.at ?? '', + last?.type ?? '', + String(PEERING_REQUEST_STATS_TTL_SECONDS), + ], + ) + } catch (error) { + // Accounting must never fail the peer request it is describing. + logger.warn('failed to record peering request stats for %s: %o', pubkey, error) + } + } +} diff --git a/test/unit/adapters/web-socket-adapter.spec.ts b/test/unit/adapters/web-socket-adapter.spec.ts index 35da6f764..2b35d28e4 100644 --- a/test/unit/adapters/web-socket-adapter.spec.ts +++ b/test/unit/adapters/web-socket-adapter.spec.ts @@ -1,21 +1,20 @@ -import EventEmitter from 'events' -import { WebSocket } from 'ws' - import chai from 'chai' import chaiAsPromised from 'chai-as-promised' +import EventEmitter from 'events' import Sinon from 'sinon' import sinonChai from 'sinon-chai' +import { WebSocket } from 'ws' chai.use(sinonChai) chai.use(chaiAsPromised) const { expect } = chai -import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../../../src/constants/adapter' import { IWebSocketServerAdapter } from '../../../src/@types/adapters' -import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' import { WebSocketAdapter } from '../../../src/adapters/web-socket-adapter' +import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../../../src/constants/adapter' import { markActionableTarget, resetHiddenContentCache } from '../../../src/utils/hidden-content-cache' +import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' describe('WebSocketAdapter', () => { let sandbox: Sinon.SinonSandbox @@ -25,6 +24,7 @@ describe('WebSocketAdapter', () => { let createMessageHandler: Sinon.SinonStub let slidingWindowRateLimiter: Sinon.SinonStub let settingsFactory: Sinon.SinonStub + let requestStats: { addEventsServed: Sinon.SinonStub } let adapter: WebSocketAdapter let originalConsoleError: typeof console.error @@ -71,6 +71,7 @@ describe('WebSocketAdapter', () => { }, }, }) + requestStats = { addEventsServed: sandbox.stub().resolves() } adapter = new WebSocketAdapter( client, @@ -79,6 +80,7 @@ describe('WebSocketAdapter', () => { createMessageHandler, slidingWindowRateLimiter, settingsFactory, + requestStats as any, ) // Reset send history so existing tests see a clean slate @@ -279,6 +281,28 @@ describe('WebSocketAdapter', () => { expect(sent[2]).to.deep.equal(event) }) + it('counts each EVENT frame delivered to a peer', () => { + const peer = 'd'.repeat(64) + sandbox.stub(adapter, 'getPeerPubkey').returns(peer) + client.readyState = WebSocket.OPEN + adapter.onSubscribed('sub-1', [{ kinds: [1] }]) + + const event = { + id: 'a'.repeat(64), + pubkey: 'b'.repeat(64), + kind: 1, + content: 'hello', + created_at: 1000000, + sig: 'c'.repeat(128), + tags: [], + } + + adapter.emit(WebSocketAdapterEvent.Event, event) + + expect(client.send).to.have.been.calledOnce + expect(requestStats.addEventsServed).to.have.been.calledOnceWithExactly(peer, 1) + }) + it('does not send event not matching any filter', () => { client.readyState = WebSocket.OPEN adapter.onSubscribed('sub-1', [{ kinds: [999] }]) @@ -645,7 +669,7 @@ describe('WebSocketAdapter', () => { // The strategy's own estimate wins when it has one: a sliding window's // period is the upper bound, an EWMA half-life is not. - it('sends the limiter\'s own backoff when it has one', async () => { + it("sends the limiter's own backoff when it has one", async () => { slidingWindowRateLimiter.returns({ hit: sandbox.stub().resolves({ limited: true, retryAfterMs: 120000 }), }) @@ -663,11 +687,7 @@ describe('WebSocketAdapter', () => { await onMessage(Buffer.from(JSON.stringify(['REQ', 'sub1', { kinds: [1] }]))) - expect(hitStub).to.have.been.calledOnceWithExactly( - '127.0.0.1:message:REQ:60000', - 1, - { period: 60000, rate: 1 }, - ) + expect(hitStub).to.have.been.calledOnceWithExactly('127.0.0.1:message:REQ:60000', 1, { period: 60000, rate: 1 }) }) it('uses peer limits and a separate bucket for a recognized peer', async () => { @@ -691,11 +711,10 @@ describe('WebSocketAdapter', () => { 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 }, - ) + 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 diff --git a/test/unit/controllers/admin/get-admin-peering-controller.spec.ts b/test/unit/controllers/admin/get-admin-peering-controller.spec.ts index 3abcd102f..50c58b25d 100644 --- a/test/unit/controllers/admin/get-admin-peering-controller.spec.ts +++ b/test/unit/controllers/admin/get-admin-peering-controller.spec.ts @@ -22,13 +22,24 @@ describe('GetAdminPeeringController', () => { lagSeconds: 3, } const stateStore = { list: Sinon.stub().resolves([state]) } + const requestStatsStore = { + list: Sinon.stub().resolves([ + { + pubkey: state.pubkey, + requests: 12, + eventsServed: 34, + lastRequestAt: '2026-01-01T00:00:00.000Z', + lastRequestType: 'REQ', + }, + ]), + } const settings = { peering: { enabled: true, peers: [{ pubkey: state.pubkey, relayUrl: state.relayUrl }], }, } as Settings - const controller = new GetAdminPeeringController(stateStore as any, () => settings) + const controller = new GetAdminPeeringController(stateStore as any, () => settings, requestStatsStore as any) const response = { status: Sinon.stub().returnsThis(), setHeader: Sinon.stub().returnsThis(), @@ -46,5 +57,12 @@ describe('GetAdminPeeringController', () => { stale: false, }) expect(payload.peers[0].state.lagSeconds).to.equal(3) + expect(payload.peers[0].requests).to.deep.equal({ + pubkey: state.pubkey, + requests: 12, + eventsServed: 34, + lastRequestAt: '2026-01-01T00:00:00.000Z', + lastRequestType: 'REQ', + }) }) }) diff --git a/test/unit/handlers/count-message-handler.spec.ts b/test/unit/handlers/count-message-handler.spec.ts index 9d930dcae..172eb011c 100644 --- a/test/unit/handlers/count-message-handler.spec.ts +++ b/test/unit/handlers/count-message-handler.spec.ts @@ -20,6 +20,7 @@ describe('CountMessageHandler', () => { let sandbox: Sinon.SinonSandbox let rateLimiterHitStub: Sinon.SinonStub let rateLimiterFactory: Sinon.SinonStub + let requestStats: { recordRequest: Sinon.SinonStub } beforeEach(() => { sandbox = Sinon.createSandbox() @@ -30,20 +31,28 @@ describe('CountMessageHandler', () => { rateLimiterHitStub = sandbox.stub().resolves({ limited: false }) rateLimiterFactory = sandbox.stub().returns({ hit: rateLimiterHitStub }) + requestStats = { recordRequest: sandbox.stub().resolves() } webSocket = new EventEmitter() as any webSocket.getPeerPubkey = sandbox.stub().returns(undefined) - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - limits: { - client: { - subscription: { - maxFilters: 10, - maxSubscriptionIdLength: 256, + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + limits: { + client: { + subscription: { + maxFilters: 10, + maxSubscriptionIdLength: 256, + }, + }, }, - }, - }, - }) as Settings, rateLimiterFactory) + }) as Settings, + rateLimiterFactory, + requestStats as any, + ) }) afterEach(() => { @@ -79,16 +88,21 @@ describe('CountMessageHandler', () => { }) it('returns CLOSED when the request has too many filters', async () => { - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - limits: { - client: { - subscription: { - maxFilters: 1, - maxSubscriptionIdLength: 256, + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + limits: { + client: { + subscription: { + maxFilters: 1, + maxSubscriptionIdLength: 256, + }, + }, }, - }, - }, - }) as Settings) + }) as Settings, + ) const message = [MessageType.COUNT, 'q1', { kinds: [1] }, { kinds: [2] }] as any @@ -101,12 +115,19 @@ describe('CountMessageHandler', () => { }) 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, rateLimiterFactory) + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + limits: { + client: { subscription: { maxFilters: 1, maxSubscriptionIdLength: 256 } }, + peer: { subscription: { maxFilters: 5, maxSubscriptionIdLength: 256 } }, + }, + }) as Settings, + rateLimiterFactory, + requestStats as any, + ) ;(webSocket.getPeerPubkey as Sinon.SinonStub).returns('a'.repeat(64)) const message = [MessageType.COUNT, 'q1', { kinds: [1] }, { kinds: [2] }] as any @@ -115,15 +136,22 @@ describe('CountMessageHandler', () => { expect(eventRepository.countByFilters).to.have.been.calledOnceWithExactly([{ kinds: [1] }, { kinds: [2] }]) expect(webSocketOnMessageStub).to.have.been.calledOnceWithExactly([MessageType.COUNT, 'q1', { count: 7 }]) + expect(requestStats.recordRequest).to.have.been.calledOnceWithExactly('a'.repeat(64), 'COUNT') }) it('closes a peer count when the on-demand request limit is exceeded', async () => { const peer = 'a'.repeat(64) ;(webSocket.getPeerPubkey as Sinon.SinonStub).returns(peer) rateLimiterHitStub.resolves({ limited: true, retryAfterMs: 321 }) - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - limits: { peer: { request: { rateLimits: [{ period: 60_000, rate: 1 }] } } }, - }) as Settings, rateLimiterFactory) + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + limits: { peer: { request: { rateLimits: [{ period: 60_000, rate: 1 }] } } }, + }) as Settings, + rateLimiterFactory, + ) await handler.handleMessage([MessageType.COUNT, 'q1', { kinds: [1] }] as any) @@ -141,9 +169,14 @@ describe('CountMessageHandler', () => { }) it('bounds raw filters before deduplicating them', async () => { - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - limits: { client: { subscription: { maxFilters: 10, maxSubscriptionIdLength: 256 } } }, - }) as Settings) + 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 @@ -155,17 +188,22 @@ describe('CountMessageHandler', () => { }) it('returns CLOSED when a filter has too many values', async () => { - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - limits: { - client: { - subscription: { - maxFilters: 10, - maxFilterValues: 2, - maxSubscriptionIdLength: 256, + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + limits: { + client: { + subscription: { + maxFilters: 10, + maxFilterValues: 2, + maxSubscriptionIdLength: 256, + }, + }, }, - }, - }, - }) as Settings) + }) as Settings, + ) const message = [MessageType.COUNT, 'q1', { authors: ['aa', 'bb', 'cc'] }] as any @@ -180,16 +218,21 @@ describe('CountMessageHandler', () => { }) it('returns CLOSED when the query ID is too long', async () => { - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - limits: { - client: { - subscription: { - maxFilters: 10, - maxSubscriptionIdLength: 2, + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + limits: { + client: { + subscription: { + maxFilters: 10, + maxSubscriptionIdLength: 2, + }, + }, }, - }, - }, - }) as Settings) + }) as Settings, + ) const message = [MessageType.COUNT, 'q123', {}] as any @@ -217,9 +260,14 @@ describe('CountMessageHandler', () => { it('returns CLOSED with auth-required when counting restricted kinds unauthenticated', async () => { webSocket.getAuthenticatedPubkeys = sandbox.stub().returns(new Set()) - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - nip42: { restrictedReads: { enabled: true } }, - }) as Settings) + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + nip42: { restrictedReads: { enabled: true } }, + }) as Settings, + ) const message = [MessageType.COUNT, 'q1', { kinds: [1059], '#p': ['f'.repeat(64)] }] as any @@ -235,9 +283,14 @@ describe('CountMessageHandler', () => { it('returns CLOSED when counting restricted kinds scoped to another pubkey', async () => { webSocket.getAuthenticatedPubkeys = sandbox.stub().returns(new Set(['a'.repeat(64)])) - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - nip42: { restrictedReads: { enabled: true } }, - }) as Settings) + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + nip42: { restrictedReads: { enabled: true } }, + }) as Settings, + ) const message = [MessageType.COUNT, 'q1', { kinds: [1059], '#p': ['f'.repeat(64)] }] as any @@ -250,9 +303,14 @@ describe('CountMessageHandler', () => { it('returns COUNT when restricted kinds are scoped to the authenticated client', async () => { const pubkey = 'f'.repeat(64) webSocket.getAuthenticatedPubkeys = sandbox.stub().returns(new Set([pubkey])) - handler = new CountMessageHandler(webSocket, eventRepository, () => ({ - nip42: { restrictedReads: { enabled: true } }, - }) as Settings) + handler = new CountMessageHandler( + webSocket, + eventRepository, + () => + ({ + nip42: { restrictedReads: { enabled: true } }, + }) as Settings, + ) const filter = { kinds: [1059], '#p': [pubkey] } const message = [MessageType.COUNT, 'q1', filter] as any diff --git a/test/unit/handlers/neg-open-message-handler.spec.ts b/test/unit/handlers/neg-open-message-handler.spec.ts index 00fd94e32..84d512007 100644 --- a/test/unit/handlers/neg-open-message-handler.spec.ts +++ b/test/unit/handlers/neg-open-message-handler.spec.ts @@ -53,6 +53,7 @@ describe('NegOpenMessageHandler', () => { let handler: NegOpenMessageHandler let rateLimiterHitStub: Sinon.SinonStub let rateLimiterFactory: Sinon.SinonStub + let requestStats: { recordRequest: Sinon.SinonStub } const open = (id = 'sync1', filter: any = { kinds: [1] }, hex = emptyClientMessageHex) => handler.handleMessage([MessageType.NEG_OPEN, id, filter, hex] as any) @@ -74,8 +75,15 @@ describe('NegOpenMessageHandler', () => { rateLimiterHitStub = sandbox.stub().resolves({ limited: false }) rateLimiterFactory = sandbox.stub().returns({ hit: rateLimiterHitStub }) - - handler = new NegOpenMessageHandler(webSocket, eventRepository, () => settings, rateLimiterFactory) + requestStats = { recordRequest: sandbox.stub().resolves() } + + handler = new NegOpenMessageHandler( + webSocket, + eventRepository, + () => settings, + rateLimiterFactory, + requestStats as any, + ) }) afterEach(() => { @@ -235,6 +243,7 @@ describe('NegOpenMessageHandler', () => { expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) expect(findReconciliationItems).to.have.been.called + expect(requestStats.recordRequest).to.have.been.calledOnceWithExactly('a'.repeat(64), 'NEG-OPEN') }) it('replies NEG-ERR when the peer on-demand request limit is exceeded', async () => { diff --git a/test/unit/handlers/subscribe-message-handler.spec.ts b/test/unit/handlers/subscribe-message-handler.spec.ts index 7fbdaa340..389aaa29f 100644 --- a/test/unit/handlers/subscribe-message-handler.spec.ts +++ b/test/unit/handlers/subscribe-message-handler.spec.ts @@ -50,6 +50,7 @@ describe('SubscribeMessageHandler', () => { let rateLimiterHitStub: Sinon.SinonStub let rateLimiterFactory: Sinon.SinonStub let endSubscriptionStub: Sinon.SinonStub + let requestStats: { recordRequest: Sinon.SinonStub } let sandbox: Sinon.SinonSandbox @@ -64,6 +65,9 @@ describe('SubscribeMessageHandler', () => { webSocket.waitForWritable = sandbox.stub().resolves() endSubscriptionStub = sandbox.stub() webSocket.endSubscription = endSubscriptionStub + requestStats = { + recordRequest: sandbox.stub().resolves(), + } settingsFactory = sandbox.stub() stream = new PassThrough({ objectMode: true, @@ -94,6 +98,7 @@ describe('SubscribeMessageHandler', () => { settingsFactory, inviteCodeRepository, rateLimiterFactory, + requestStats as any, ) }) @@ -224,6 +229,19 @@ describe('SubscribeMessageHandler', () => { expect(rateLimiterHitStub).not.to.have.been.called }) + it('records an accepted peer request', async () => { + const peer = 'a'.repeat(64) + canSubscribeStub.returns(undefined) + ;(webSocket.getPeerPubkey as Sinon.SinonStub).returns(peer) + settingsFactory.returns({ limits: { peer: { request: { rateLimits: [] } } } }) + message = [MessageType.REQ, subscriptionId, ...filters] as any + + await handler.handleMessage(message) + + expect(requestStats.recordRequest).to.have.been.calledOnceWithExactly(peer, 'REQ') + expect(fetchAndSendStub).to.have.been.calledOnce + }) + it('emits subscribe event if subscription is accepted', async () => { canSubscribeStub.returns(undefined) message = [MessageType.REQ, subscriptionId, ...filters] as any diff --git a/test/unit/utils/admin-peering.spec.ts b/test/unit/utils/admin-peering.spec.ts index a5bf6e080..8cc03d552 100644 --- a/test/unit/utils/admin-peering.spec.ts +++ b/test/unit/utils/admin-peering.spec.ts @@ -72,4 +72,24 @@ describe('admin peering rows', () => { expect(rows[0].state).to.equal(undefined) expect(rows[0].stale).to.equal(true) }) + + it('attaches request accounting by peer pubkey', () => { + const rows = mergePeeringRows( + { + peering: { + enabled: true, + peers: [{ pubkey: 'a'.repeat(64), relayUrl: 'wss://peer.example.com' }], + }, + } as Settings, + [state()], + undefined, + [{ pubkey: 'a'.repeat(64), requests: 4, eventsServed: 7 }], + ) + + expect(rows[0].requests).to.deep.equal({ + pubkey: 'a'.repeat(64), + requests: 4, + eventsServed: 7, + }) + }) }) diff --git a/test/unit/utils/peering-request-stats.spec.ts b/test/unit/utils/peering-request-stats.spec.ts new file mode 100644 index 000000000..b36b3f86b --- /dev/null +++ b/test/unit/utils/peering-request-stats.spec.ts @@ -0,0 +1,132 @@ +import chai from 'chai' +import Sinon from 'sinon' +import sinonChai from 'sinon-chai' + +import { ICacheAdapter } from '../../../src/@types/adapters' +import { PEERING_REQUEST_STATS_INDEX_KEY, PeeringRequestStatsStore } from '../../../src/utils/peering-request-stats' + +chai.use(sinonChai) +const { expect } = chai + +describe('PeeringRequestStatsStore', () => { + let sandbox: Sinon.SinonSandbox + let evalStub: Sinon.SinonStub + let getSetMembersStub: Sinon.SinonStub + let store: PeeringRequestStatsStore + + beforeEach(() => { + sandbox = Sinon.createSandbox() + evalStub = sandbox.stub().resolves(1) + getSetMembersStub = sandbox.stub().resolves([]) + store = new PeeringRequestStatsStore({ + eval: evalStub, + getSetMembers: getSetMembersStub, + } as unknown as ICacheAdapter) + }) + + afterEach(() => { + sandbox.restore() + }) + + it('records a request with its type and timestamp', async () => { + const pubkey = 'A'.repeat(64) + + store.recordRequest(pubkey, 'REQ') + await store.flushPending() + + const [, keys, args] = evalStub.firstCall.args + expect(keys).to.deep.equal([`admin:peering:requests:${pubkey.toLowerCase()}`, PEERING_REQUEST_STATS_INDEX_KEY]) + expect(args[0]).to.equal('1') + expect(args[1]).to.equal('0') + expect(args[2]).to.equal(pubkey.toLowerCase()) + expect(args[3]).to.be.a('string') + expect(args[4]).to.equal('REQ') + }) + + it('adds events served without counting another request', async () => { + store.addEventsServed('a'.repeat(64), 5) + await store.flushPending() + + const [, , args] = evalStub.firstCall.args + expect(args[0]).to.equal('0') + expect(args[1]).to.equal('5') + expect(args[3]).to.equal('') + }) + + it('coalesces request and event increments into one write per peer', async () => { + const pubkey = 'a'.repeat(64) + + store.recordRequest(pubkey, 'REQ') + store.addEventsServed(pubkey, 5) + store.recordRequest(pubkey, 'COUNT') + await store.flushPending() + + expect(evalStub).to.have.been.calledOnce + const [, , args] = evalStub.firstCall.args + expect(args[0]).to.equal('2') + expect(args[1]).to.equal('5') + expect(args[4]).to.equal('COUNT') + }) + + it('reads and parses all recorded peers', async () => { + const pubkey = 'a'.repeat(64) + getSetMembersStub.resolves([pubkey]) + evalStub.resolves([ + 'pubkey', + pubkey, + 'requests', + '3', + 'eventsServed', + '9', + 'lastRequestAt', + '2026-01-01T00:00:00.000Z', + 'lastRequestType', + 'REQ', + ]) + + const result = await store.list() + + expect(result).to.deep.equal([ + { + pubkey, + requests: 3, + eventsServed: 9, + lastRequestAt: '2026-01-01T00:00:00.000Z', + lastRequestType: 'REQ', + }, + ]) + }) + + it('prunes index members whose hash expired in the same read', async () => { + const pubkey = 'a'.repeat(64) + getSetMembersStub.resolves([pubkey]) + evalStub.resolves([]) + + const result = await store.list() + + expect(result).to.deep.equal([]) + expect(evalStub).to.have.been.calledOnce + expect(evalStub.firstCall.args[1]).to.deep.equal([ + `admin:peering:requests:${pubkey}`, + PEERING_REQUEST_STATS_INDEX_KEY, + ]) + expect(evalStub.firstCall.args[2]).to.deep.equal([pubkey]) + }) + + it('returns an empty list when the stats index cannot be read', async () => { + getSetMembersStub.rejects(new Error('redis down')) + + const result = await store.list() + + expect(result).to.deep.equal([]) + expect(evalStub).not.to.have.been.called + }) + + it('does not fail the request path when accounting storage is unavailable', async () => { + evalStub.rejects(new Error('redis down')) + + store.recordRequest('a'.repeat(64), 'REQ') + store.addEventsServed('a'.repeat(64), 2) + await store.flushPending() + }) +})