feat(admin): account for peer on-demand requests

.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.

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

resources/admin/assets/admin-ops.js

@@ -145,13 +145,14 @@
   const renderPeering = (rows) => {
     if (!Array.isArray(rows) || !rows.length) {
       elements.peeringBody.innerHTML =
-        '<tr><td colspan="6" class="admin-muted small">No peering peers configured.</td></tr>'
+        '<tr><td colspan="8" class="admin-muted small">No peering peers configured.</td></tr>'
       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 `<tr>
@@ -160,6 +161,8 @@
           <td class="small">${escapeHtml(formatTime(state?.lastReconciledAt))}</td>
           <td>${escapeHtml(formatLag(state?.lagSeconds))}</td>
           <td>${escapeHtml(state?.missingEvents ?? '—')}</td>
+          <td>${escapeHtml(requests?.requests ?? 0)}</td>
+          <td>${escapeHtml(requests?.eventsServed ?? 0)}</td>
           <td>${peeringSignalBadge(row)}</td>
         </tr>`
       })

resources/admin/dashboard.html

@@ -239,11 +239,13 @@ <h2 class="section-title">Peering</h2>
                     <th scope="col">Last reconciliation</th>
                     <th scope="col">Lag</th>
                     <th scope="col">Missing</th>
+                    <th scope="col">Requests</th>
+                    <th scope="col">Events served</th>
                     <th scope="col">Signal</th>
                   </tr>
                 </thead>
                 <tbody id="peering-body">
-                  <tr><td colspan="6" class="admin-muted small">No peering peers configured.</td></tr>
+                  <tr><td colspan="8" class="admin-muted small">No peering peers configured.</td></tr>
                 </tbody>
               </table>
             </div>

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<IMessageHandler, [IncomingMessage, IWebSocketAdapter]>,
     private readonly rateLimiter: Factory<IRateLimiter>,
     private readonly settings: Factory<Settings>,
+    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<void> {

src/controllers/admin/get-admin-peering-controller.ts

@@ -3,18 +3,24 @@ 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'
 
 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<void> {
-    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 })
   }

src/factories/controllers/get-admin-peering-controller-factory.ts

@@ -2,15 +2,18 @@ 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 => {
   // The RedisAdapter constructor attaches connection listeners. Reusing one
   // 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

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:

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(),
     )

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<IRateLimiter>,
+    private readonly requestStats?: PeeringRequestStatsStore,
   ) {}
 
   public async handleMessage(message: CountMessage): Promise<void> {
@@ -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

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<IRateLimiter>,
+    private readonly requestStats?: PeeringRequestStatsStore,
   ) {}
 
   public async handleMessage(message: NegOpenMessage): Promise<void> {
@@ -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.