fix(nip77): address review findings in negentropy sessions

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

Author: Priyanshubhartistm

CONFIGURATION.md

@@ -232,7 +232,7 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | nip66.timeouts.nip11Ms                      | NIP-11 fetch timeout in milliseconds. Defaults to 10000. |
 | nip66.timeouts.tlsMs                        | TLS probe timeout in milliseconds. Defaults to 10000. |
 | nip66.timeouts.wsRttMs                      | WebSocket open RTT probe timeout in milliseconds. Defaults to 10000. |
-| nip77.enabled                               | Enable NIP-77 (Negentropy) set-reconciliation sync, so clients and other relays can find which events differ without downloading them all. Defaults to false. Only answers when `eventStore.backend` is `postgres`; other backends bring their own negentropy implementation. Each open session holds the ids of every event matching its filter in memory, so size the limits below for your hardware. |
+| nip77.enabled                               | Enable NIP-77 (Negentropy) set-reconciliation sync, so clients and other relays can find which events differ without downloading them all. Defaults to false. Only answers when `eventStore.backend` is `postgres`; other backends bring their own negentropy implementation. When `nip42.restrictedReads` is enabled, reconciliation would list the ids of restricted events, so a `NEG-OPEN` filter that could match one (it names a restricted kind or no kinds at all) is only accepted when scoped by `authors` or `#p` to pubkeys the client has authenticated as, and is re-checked on every `NEG-MSG`; otherwise the session is closed with `NEG-ERR auth-required:`. Each open session holds the ids of every event matching its filter in memory, so size the limits below for your hardware. |
 | nip77.maxEventsPerSession                   | Most events one session may cover. A `NEG-OPEN` whose filter matches more is rejected with `NEG-ERR blocked:` (clients should split the sync into `since`/`until` windows). The filter's `limit` is ignored, since reconciliation needs the whole set. Defaults to 100000. |
 | nip77.maxSessionsPerConnection              | Concurrent negentropy sessions one websocket connection may hold; opening an id that is already open replaces it. Further `NEG-OPEN`s are rejected with `NEG-ERR blocked:`. Defaults to 3. |
 | nip77.sessionIdleTimeoutMs                  | A session that receives no `NEG-MSG` for this many milliseconds is closed with `NEG-ERR closed:`. Defaults to 60000. |

src/@types/negentropy.ts

@@ -1,27 +1,39 @@
 import { Negentropy } from '../utils/negentropy'
-import { SubscriptionId } from './subscription'
+import { SubscriptionFilter, SubscriptionId } from './subscription'
 
 /** Proves a caller still owns a subscription id after awaiting; see `INegentropySessions.reserve`. */
 export type NegentropySessionToken = symbol
 
+export interface NegentropySession {
+  negentropy: Negentropy
+  /** The filter the session was opened with, so access can be re-checked on every message. */
+  filter: SubscriptionFilter
+}
+
 export interface INegentropySessions {
   /**
    * Claims `subscriptionId` for a new session, closing any session already open under that id.
-   * Returns undefined when the connection already holds `maxSessions` other sessions.
+   * Returns undefined when the connection already holds `maxLimit` other sessions, or already has
+   * `maxLimit` event reads in flight -- including reads of requests that have since been replaced,
+   * which keep running until `endRead` is called. The caller must call `endRead` for a token it got.
    */
-  reserve(subscriptionId: SubscriptionId, maxSessions: number): NegentropySessionToken | undefined
+  reserve(subscriptionId: SubscriptionId, maxLimit: number): NegentropySessionToken | undefined
+  /** Marks the event read started for a reservation as finished, successfully or not. */
+  endRead(token: NegentropySessionToken): void
+  /** Whether `token` is still the reservation for `subscriptionId`, i.e. nobody replaced or closed it. */
+  isCurrent(subscriptionId: SubscriptionId, token: NegentropySessionToken): boolean
   /**
    * Turns a reservation into a live session. Returns false when the reservation was closed or
    * replaced while the caller was busy (e.g. reading events), in which case nothing is stored.
    */
   activate(
     subscriptionId: SubscriptionId,
     token: NegentropySessionToken,
-    negentropy: Negentropy,
+    session: NegentropySession,
     idleTimeoutMs: number,
   ): boolean
   /** The live session for an id, with its idle timer restarted. */
-  touch(subscriptionId: SubscriptionId): Negentropy | undefined
+  touch(subscriptionId: SubscriptionId): NegentropySession | undefined
   /** Drops a reservation that never became a session; a no-op if it has since been replaced. */
   release(subscriptionId: SubscriptionId, token: NegentropySessionToken): void
   close(subscriptionId: SubscriptionId): boolean

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

@@ -6,6 +6,7 @@ import { Settings } from '../@types/settings'
 import { WebSocketAdapterEvent } from '../constants/adapter'
 import { createLogger } from '../factories/logger-factory'
 import { isNip77Active } from '../utils/nip77'
+import { isReconciliationAuthorized } from '../utils/nip42'
 
 const logger = createLogger('neg-msg-message-handler')
 
@@ -26,14 +27,23 @@ export class NegMsgMessageHandler implements IMessageHandler {
       return
     }
 
-    const negentropy = sessions.touch(subscriptionId)
-    if (!negentropy) {
+    const session = sessions.touch(subscriptionId)
+    if (!session) {
       this.reply(createNegErrMessage(subscriptionId, 'closed: unknown negentropy subscription'))
       return
     }
 
+    // Access can lapse while a session is open (e.g. NIP-42 auth expiring), so check it every time.
+    if (!isReconciliationAuthorized(this.settings(), session.filter, () => this.webSocket.getAuthenticatedPubkeys())) {
+      sessions.close(subscriptionId)
+      this.reply(
+        createNegErrMessage(subscriptionId, 'auth-required: authentication is required to sync these event kinds'),
+      )
+      return
+    }
+
     try {
-      const { output } = negentropy.reconcile(Buffer.from(message[2], 'hex'))
+      const { output } = session.negentropy.reconcile(Buffer.from(message[2], 'hex'))
 
       this.reply(createNegMsgMessage(subscriptionId, Buffer.from(output as Uint8Array).toString('hex')))
     } catch (error) {

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

@@ -1,16 +1,17 @@
 import { IWebSocketAdapter } from '../@types/adapters'
 import { createNegErrMessage, createNegMsgMessage } from '../utils/messages'
 import { getNip77Limits, isNip77Active, NEGENTROPY_FRAME_SIZE_LIMIT } from '../utils/nip77'
+import { INegentropySessions } from '../@types/negentropy'
 import { Negentropy, NegentropyStorageVector } from '../utils/negentropy'
-import { IEventRepository } from '../@types/repositories'
+import { IEventRepository, ReconciliationItem } from '../@types/repositories'
 import { IMessageHandler } from '../@types/message-handlers'
 import { NegOpenMessage, OutgoingMessage } from '../@types/messages'
 import { Settings } from '../@types/settings'
 import { SubscriptionFilter, SubscriptionId } from '../@types/subscription'
 import { WebSocketAdapterEvent } from '../constants/adapter'
 import { countFilterValues } from '../utils/filter'
 import { createLogger } from '../factories/logger-factory'
-import { isCountAuthorized } from '../utils/nip42'
+import { isReconciliationAuthorized } from '../utils/nip42'
 
 const logger = createLogger('neg-open-message-handler')
 
@@ -31,41 +32,62 @@ export class NegOpenMessageHandler implements IMessageHandler {
     const initialMessage = Buffer.from(message[3], 'hex')
 
     const settings = this.settings()
+    const sessions = this.webSocket.getNegentropySessions()
+
     if (!isNip77Active(settings)) {
-      this.reply(createNegErrMessage(subscriptionId, 'blocked: NIP-77 is disabled by relay configuration'))
+      this.reject(sessions, subscriptionId, 'blocked: NIP-77 is disabled by relay configuration')
       return
     }
 
     const rejection = this.getRejection(subscriptionId, filter, initialMessage)
     if (rejection) {
       logger('NEG-OPEN %s with %o rejected: %s', subscriptionId, filter, rejection)
-      this.reply(createNegErrMessage(subscriptionId, rejection))
+      this.reject(sessions, subscriptionId, rejection)
       return
     }
 
     const { maxEventsPerSession, maxSessionsPerConnection, sessionIdleTimeoutMs } = getNip77Limits(settings)
-    const sessions = this.webSocket.getNegentropySessions()
 
     const token = sessions.reserve(subscriptionId, maxSessionsPerConnection)
     if (!token) {
-      this.reply(
-        createNegErrMessage(
-          subscriptionId,
-          `blocked: too many negentropy sessions on this connection (max ${maxSessionsPerConnection})`,
-        ),
+      this.reject(
+        sessions,
+        subscriptionId,
+        `blocked: too many negentropy sessions or reads in progress on this connection (max ${maxSessionsPerConnection})`,
       )
       return
     }
 
+    let items: ReconciliationItem[]
     try {
       // One row past the limit tells us the set is too big without loading all of it.
-      const items = await this.eventRepository.findReconciliationItems(filter, maxEventsPerSession + 1)
-      if (items.length > maxEventsPerSession) {
+      items = await this.eventRepository.findReconciliationItems(filter, maxEventsPerSession + 1)
+    } catch (error) {
+      if (sessions.isCurrent(subscriptionId, token)) {
         sessions.release(subscriptionId, token)
-        this.reply(createNegErrMessage(subscriptionId, 'blocked: this query is too big', maxEventsPerSession))
-        return
+        logger('NEG-OPEN %s failed: %o', subscriptionId, error)
+        // Keep this generic so internal errors are not leaked to clients.
+        this.reply(createNegErrMessage(subscriptionId, 'error: unable to process negentropy request'))
       }
+      return
+    } finally {
+      sessions.endRead(token)
+    }
+
+    // The client may have closed or replaced this NEG-OPEN while events were being read. Then the
+    // newer request owns the id: say nothing, since an error here would tell the client to drop it.
+    if (!sessions.isCurrent(subscriptionId, token)) {
+      return
+    }
 
+    // Nothing below awaits, so ownership cannot change before the reply.
+    if (items.length > maxEventsPerSession) {
+      sessions.release(subscriptionId, token)
+      this.reply(createNegErrMessage(subscriptionId, 'blocked: this query is too big', maxEventsPerSession))
+      return
+    }
+
+    try {
       const storage = new NegentropyStorageVector()
       for (const item of items) {
         storage.insert(item.event_created_at, item.event_id)
@@ -85,16 +107,11 @@ export class NegOpenMessageHandler implements IMessageHandler {
         return
       }
 
-      if (!sessions.activate(subscriptionId, token, negentropy, sessionIdleTimeoutMs)) {
-        // Closed or replaced by the client while we were reading events: stay silent.
-        return
-      }
-
+      sessions.activate(subscriptionId, token, { negentropy, filter }, sessionIdleTimeoutMs)
       this.reply(createNegMsgMessage(subscriptionId, Buffer.from(output).toString('hex')))
     } catch (error) {
       sessions.release(subscriptionId, token)
       logger('NEG-OPEN %s failed: %o', subscriptionId, error)
-      // Keep this generic so internal errors are not leaked to clients.
       this.reply(createNegErrMessage(subscriptionId, 'error: unable to process negentropy request'))
     }
   }
@@ -123,13 +140,20 @@ export class NegOpenMessageHandler implements IMessageHandler {
       return 'invalid: malformed negentropy message'
     }
 
-    // NIP-42: like COUNT, reconciling reveals which events exist, so restricted kinds must be
-    // scoped to the client's own pubkeys.
-    if (!isCountAuthorized(settings, [filter], () => this.webSocket.getAuthenticatedPubkeys())) {
+    // NIP-42: reconciling reveals the id of every event the filter matches, so a filter that could
+    // match a restricted kind must be scoped to the client's own pubkeys.
+    if (!isReconciliationAuthorized(settings, filter, () => this.webSocket.getAuthenticatedPubkeys())) {
       return 'auth-required: authentication is required to sync these event kinds'
     }
   }
 
+  // NIP-77: after a NEG-ERR the subscription is closed, so a rejected NEG-OPEN must also end any
+  // session the client already had under that id; otherwise it keeps serving, and holding a slot.
+  private reject(sessions: INegentropySessions, subscriptionId: SubscriptionId, reason: string): void {
+    sessions.close(subscriptionId)
+    this.reply(createNegErrMessage(subscriptionId, reason))
+  }
+
   private reply(message: OutgoingMessage): void {
     this.webSocket.emit(WebSocketAdapterEvent.Message, message)
   }

src/utils/negentropy-sessions.ts

@@ -1,61 +1,74 @@
-import { INegentropySessions, NegentropySessionToken } from '../@types/negentropy'
-import { Negentropy } from './negentropy'
+import { INegentropySessions, NegentropySession, NegentropySessionToken } from '../@types/negentropy'
 import { SubscriptionId } from '../@types/subscription'
 
 interface Entry {
   token: NegentropySessionToken
-  negentropy?: Negentropy
+  session?: NegentropySession
   idleTimeoutMs?: number
   idleTimer?: NodeJS.Timeout
 }
 
 /** The NIP-77 sessions of one websocket connection. Ids are a separate namespace from REQ ids. */
 export class NegentropySessions implements INegentropySessions {
   private readonly entries = new Map<SubscriptionId, Entry>()
+  // Event reads still running. A replaced request's read is not cancelled, so it stays counted here
+  // until it finishes; otherwise repeating NEG-OPEN on one id would stack up unbounded queries.
+  private readonly pendingReads = new Set<NegentropySessionToken>()
 
   public constructor(private readonly onIdleTimeout: (subscriptionId: SubscriptionId) => void) {}
 
-  public reserve(subscriptionId: SubscriptionId, maxSessions: number): NegentropySessionToken | undefined {
+  public reserve(subscriptionId: SubscriptionId, maxLimit: number): NegentropySessionToken | undefined {
     // NIP-77: opening an id that is already open closes the old session first.
     this.close(subscriptionId)
 
-    if (this.entries.size >= maxSessions) {
+    if (this.entries.size >= maxLimit || this.pendingReads.size >= maxLimit) {
       return undefined
     }
 
     const token = Symbol(subscriptionId)
     this.entries.set(subscriptionId, { token })
+    this.pendingReads.add(token)
 
     return token
   }
 
+  public endRead(token: NegentropySessionToken): void {
+    this.pendingReads.delete(token)
+  }
+
+  public isCurrent(subscriptionId: SubscriptionId, token: NegentropySessionToken): boolean {
+    return this.entries.get(subscriptionId)?.token === token
+  }
+
   public activate(
     subscriptionId: SubscriptionId,
     token: NegentropySessionToken,
-    negentropy: Negentropy,
+    session: NegentropySession,
     idleTimeoutMs: number,
   ): boolean {
     const entry = this.entries.get(subscriptionId)
     if (!entry || entry.token !== token) {
       return false
     }
 
-    entry.negentropy = negentropy
+    // Activating means the read is over.
+    this.pendingReads.delete(token)
+    entry.session = session
     entry.idleTimeoutMs = idleTimeoutMs
     this.startIdleTimer(subscriptionId, entry)
 
     return true
   }
 
-  public touch(subscriptionId: SubscriptionId): Negentropy | undefined {
+  public touch(subscriptionId: SubscriptionId): NegentropySession | undefined {
     const entry = this.entries.get(subscriptionId)
-    if (!entry?.negentropy) {
+    if (!entry?.session) {
       return undefined
     }
 
     this.startIdleTimer(subscriptionId, entry)
 
-    return entry.negentropy
+    return entry.session
   }
 
   public release(subscriptionId: SubscriptionId, token: NegentropySessionToken): void {
@@ -83,7 +96,7 @@ export class NegentropySessions implements INegentropySessions {
   }
 
   public getSessionIds(): SubscriptionId[] {
-    return [...this.entries.entries()].filter(([, entry]) => entry.negentropy).map(([subscriptionId]) => subscriptionId)
+    return [...this.entries.entries()].filter(([, entry]) => entry.session).map(([subscriptionId]) => subscriptionId)
   }
 
   private startIdleTimer(subscriptionId: SubscriptionId, entry: Entry): void {

src/utils/nip42.ts

@@ -89,6 +89,32 @@ export const isSubscriptionAuthRequired = (
   return getAuthenticatedPubkeys().size === 0
 }
 
+// NIP-77 reconciliation, like COUNT, can't be filtered per event: it reveals the id of every event the
+// filter matches. REQ hides restricted events per event, so a filter that *could* match one -- one
+// naming a restricted kind or naming no kinds at all -- must be scoped to the client's own pubkeys.
+export const isReconciliationAuthorized = (
+  settings: Settings | undefined,
+  filter: SubscriptionFilter,
+  getAuthenticatedPubkeys: () => ReadonlySet<Pubkey>,
+): boolean => {
+  const restrictedKinds = getRestrictedReadKinds(settings)
+  if (!restrictedKinds.length) {
+    return true
+  }
+
+  const mayMatchRestricted =
+    !Array.isArray(filter.kinds) || filter.kinds.some((kind) => isKindRestricted(restrictedKinds, kind))
+  if (!mayMatchRestricted) {
+    return true
+  }
+
+  const authenticatedPubkeys = getAuthenticatedPubkeys()
+  const isScopedToClient = (values?: Pubkey[]) =>
+    Array.isArray(values) && values.length > 0 && values.every((value) => authenticatedPubkeys.has(value))
+
+  return isScopedToClient(filter.authors) || isScopedToClient(filter['#p'])
+}
+
 // COUNT can't be filtered per event, so a restricted-kind filter must be
 // scoped to the client's own pubkeys via authors/#p.
 export const isCountAuthorized = (

test/unit/adapters/web-socket-adapter.spec.ts

@@ -988,7 +988,12 @@ describe('WebSocketAdapter', () => {
 
     const openSession = (id: string, idleTimeoutMs = 60000) => {
       const sessions = adapter.getNegentropySessions()
-      sessions.activate(id, sessions.reserve(id, 3) as symbol, emptyStorage(), idleTimeoutMs)
+      sessions.activate(
+        id,
+        sessions.reserve(id, 3) as symbol,
+        { negentropy: emptyStorage(), filter: {} },
+        idleTimeoutMs,
+      )
     }
 
     const sentMessages = () => client.send.getCalls().map((call: any) => JSON.parse(call.args[0]))

test/unit/handlers/neg-close-message-handler.spec.ts

@@ -17,7 +17,7 @@ describe('NegCloseMessageHandler', () => {
   const openSession = (id: string) => {
     const storage = new NegentropyStorageVector()
     storage.seal()
-    sessions.activate(id, sessions.reserve(id, 3) as symbol, new Negentropy(storage), 60000)
+    sessions.activate(id, sessions.reserve(id, 3) as symbol, { negentropy: new Negentropy(storage), filter: {} }, 60000)
   }
 
   beforeEach(() => {