Verify Nostr inbound per relay to avoid head-of-line blocking

The single global inbound consumer serialized ALL relay traffic behind
one Schnorr-verification queue, so a burst of EVENT frames from one
busy/malicious relay stalled DMs, OKs, EOSEs, and events from every other
relay (codex P2). Give each relay connection its own bounded AsyncStream +
detached serial consumer instead: N relays verify in parallel while each
relay's frames stay in arrival order (per-relay ordering preserves the
per-subscription ordering that actually matters, since a subscription's
events for a relay all arrive on that relay's socket).

Continuations live in a lock-guarded Sendable router so the non-isolated
socket receive callback can route a frame to the right relay stream with
no per-frame main hop; the main actor owns pipeline start/teardown, wired
into connect, disconnect, panic wipe, per-relay disconnect, retry, and the
default-relay revoke path. Streams use .bufferingNewest so a relay flooding
faster than it verifies sheds its OWN oldest frames — it can neither exhaust
memory nor starve other relays.

Security invariants are unchanged: signature verified exactly once, off
main; dedup pre-check-before / record-after-verify (forged copies still
can't poison the dedup set); the atomic main-actor check-and-record in
deliverVerifiedInboundEvent; the inner NIP-17 seal check untouched.

Tests: existing per-relay in-order-delivery and tampered-signature
dedup-poison tests still pass; add a cross-relay non-blocking test proving
a large backlog on relay A does not delay a later frame on relay B.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bitchat/Nostr/NostrRelayManager.swift

@@ -213,23 +213,34 @@ final class NostrRelayManager: ObservableObject {
     // Bump generation to invalidate scheduled reconnects when we reset/disconnect
     private var connectionGeneration: Int = 0
 
-    // Serial off-main inbound pipeline: raw socket frames are parsed and
+    // Per-relay off-main inbound pipeline: raw socket frames are parsed and
     // Schnorr-verified in arrival order OFF the main actor (this is the single
     // signature verification for the whole inbound path — downstream handlers
     // receive only verified events), then hop back to the main actor for dedup
-    // recording and handler dispatch. A single consumer task preserves
+    // recording and handler dispatch.
-    // per-subscription arrival order.
+    //
-    private struct InboundFrame {
+    // Each relay connection owns its OWN AsyncStream + consumer task, so N
-        let message: URLSessionWebSocketTask.Message
+    // relays verify in parallel while every relay's frames stay in arrival
-        let relayUrl: String
+    // order (a single subscription's events for a relay all arrive on that
-    }
+    // relay's socket, so per-relay ordering preserves per-subscription
-    private let inboundContinuation: AsyncStream<InboundFrame>.Continuation
+    // ordering). A burst of EVENT frames from one busy/malicious relay only
-    private var inboundTask: Task<Void, Never>?
+    // blocks that relay's own verification backlog — DMs, OKs, EOSEs, and
+    // events from every other relay keep flowing on their own pipelines.
+    //
+    // Each stream is bounded (`.bufferingNewest`) so a relay flooding faster
+    // than its verification drains sheds its own oldest frames instead of
+    // growing memory without bound; it can never starve other relays.
+    //
+    // Continuations live in a lock-guarded, `Sendable` router (see
+    // `InboundFrameRouter` at file scope) so the raw socket receive callback
+    // (which is NOT main-actor isolated) can route a frame to the right relay
+    // stream without a per-frame main hop, while the main actor owns pipeline
+    // creation/teardown. The expensive work (Schnorr verify) is what runs
+    // off-main; the yield stays cheap.
+    private let inboundRouter = InboundFrameRouter()
 
     init() {
         self.dependencies = .live()
-        let (inboundStream, inboundContinuation) = AsyncStream<InboundFrame>.makeStream()
-        self.inboundContinuation = inboundContinuation
         hasMutualFavorites = dependencies.hasMutualFavorites()
         hasLocationPermission = dependencies.hasLocationPermission()
         applyDefaultRelayPolicy(force: true)
@@ -253,13 +264,10 @@ final class NostrRelayManager: ObservableObject {
                 self.applyDefaultRelayPolicy()
             }
             .store(in: &cancellables)
-        startInboundPipeline(consuming: inboundStream)
     }
 
     internal init(dependencies: NostrRelayManagerDependencies) {
         self.dependencies = dependencies
-        let (inboundStream, inboundContinuation) = AsyncStream<InboundFrame>.makeStream()
-        self.inboundContinuation = inboundContinuation
         hasMutualFavorites = dependencies.hasMutualFavorites()
         hasLocationPermission = dependencies.hasLocationPermission()
         applyDefaultRelayPolicy(force: true)
@@ -283,17 +291,16 @@ final class NostrRelayManager: ObservableObject {
                 self.applyDefaultRelayPolicy()
             }
             .store(in: &cancellables)
-        startInboundPipeline(consuming: inboundStream)
     }
 
     deinit {
-        inboundContinuation.finish()
+        inboundRouter.finishAll()
-        inboundTask?.cancel()
     }
 
-    /// Starts the single consumer task behind `inboundContinuation`.
+    /// Ensure a serial off-main consumer pipeline exists for a relay. Called on
+    /// the main actor when a socket is (re)armed for receiving. Idempotent.
     ///
-    /// Ordering is deliberate and security/performance-critical:
+    /// Ordering within the relay is deliberate and security/performance-critical:
     /// 1. `precheckInboundEvent` (main hop): per-relay stats plus a cheap
     ///    duplicate LOOKUP — duplicate fan-in from multiple relays dominates
     ///    real traffic and must never pay for Schnorr verification.
@@ -304,33 +311,49 @@ final class NostrRelayManager: ObservableObject {
     ///    check-and-RECORD plus handler dispatch. Recording only after
     ///    verification means a forged-signature copy can never poison the
     ///    dedup cache and suppress the genuine event.
-    private func startInboundPipeline(consuming stream: AsyncStream<InboundFrame>) {
+    private func ensureRelayInboundPipeline(for relayUrl: String) {
-        inboundTask = Task.detached(priority: .userInitiated) { [weak self] in
+        let started = inboundRouter.startPipeline(for: relayUrl) { [weak self] stream in
-            for await frame in stream {
+            Task.detached(priority: .userInitiated) {
-                guard let parsed = ParsedInbound(frame.message) else { continue }
+                for await frame in stream {
-                guard let self else { return }
+                    guard let parsed = ParsedInbound(frame.message) else { continue }
-                switch parsed {
+                    guard let self else { return }
-                case .event(let subId, let event):
+                    switch parsed {
-                    guard await self.precheckInboundEvent(
+                    case .event(let subId, let event):
-                        subscriptionID: subId,
+                        guard await self.precheckInboundEvent(
-                        eventID: event.id,
+                            subscriptionID: subId,
-                        relayUrl: frame.relayUrl
+                            eventID: event.id,
-                    ) else {
+                            relayUrl: relayUrl
-                        continue
+                        ) else {
+                            continue
+                        }
+                        guard event.isValidSignature() else {
+                            SecureLogger.warning(
+                                "⚠️ Dropped invalid Nostr event id=\(event.id.prefix(16))… sub=\(subId) relay=\(relayUrl)",
+                                category: .session
+                            )
+                            continue
+                        }
+                        await self.deliverVerifiedInboundEvent(subscriptionID: subId, event: event, from: relayUrl)
+                    case .eose, .ok, .notice:
+                        await self.handleParsedMessage(parsed, from: relayUrl)
                     }
-                    guard event.isValidSignature() else {
-                        SecureLogger.warning(
-                            "⚠️ Dropped invalid Nostr event id=\(event.id.prefix(16))… sub=\(subId) relay=\(frame.relayUrl)",
-                            category: .session
-                        )
-                        continue
-                    }
-                    await self.deliverVerifiedInboundEvent(subscriptionID: subId, event: event, from: frame.relayUrl)
-                case .eose, .ok, .notice:
-                    await self.handleParsedMessage(parsed, from: frame.relayUrl)
                 }
             }
         }
+        if started {
+            SecureLogger.debug("🧵 Started inbound verify pipeline for \(relayUrl)", category: .session)
+        }
+    }
+
+    /// Tear down a relay's inbound pipeline (socket gone or state wiped). The
+    /// consumer drains any already-buffered frames before finishing, so
+    /// in-flight verified events are still delivered.
+    private func teardownRelayInboundPipeline(for relayUrl: String) {
+        inboundRouter.finishPipeline(for: relayUrl)
+    }
+
+    private func teardownAllRelayInboundPipelines() {
+        inboundRouter.finishAll()
     }
 
     /// Connect to all configured relays
@@ -347,6 +370,8 @@ final class NostrRelayManager: ObservableObject {
             task.cancel(with: .goingAway, reason: nil)
         }
         connections.removeAll()
+        // Sockets are gone; drop every relay's inbound verify pipeline.
+        teardownAllRelayInboundPipelines()
         markRelaySocketsClosed(resetState: false)
         // Sockets are gone, so per-relay subscription state is cleared — but
         // durable intent (subscriptionRequestState, messageHandlers, parked
@@ -376,6 +401,7 @@ final class NostrRelayManager: ObservableObject {
             task.cancel(with: .goingAway, reason: nil)
         }
         connections.removeAll()
+        teardownAllRelayInboundPipelines()
         markRelaySocketsClosed(resetState: true)
         subscriptions.removeAll()
         pendingSubscriptions.removeAll()
@@ -626,6 +652,7 @@ final class NostrRelayManager: ObservableObject {
                     connection.cancel(with: .goingAway, reason: nil)
                 }
                 connections.removeValue(forKey: url)
+                teardownRelayInboundPipeline(for: url)
                 subscriptions.removeValue(forKey: url)
                 pendingSubscriptions.removeValue(forKey: url)
             }
@@ -965,7 +992,11 @@ final class NostrRelayManager: ObservableObject {
         
         connections[urlString] = task
         task.resume()
-        
+
+        // Bring up this relay's own serial verify pipeline before arming the
+        // socket, so inbound frames have somewhere to land.
+        ensureRelayInboundPipeline(for: urlString)
+
         // Start receiving messages
         receiveMessage(from: task, relayUrl: urlString)
         
@@ -1024,9 +1055,12 @@ final class NostrRelayManager: ObservableObject {
             
             switch result {
             case .success(let message):
-                // Hand the raw frame to the serial inbound pipeline: parsing
+                // Hand the raw frame to this relay's serial inbound pipeline:
-                // and signature verification run off-main, in arrival order.
+                // parsing and signature verification run off-main, in arrival
-                self.inboundContinuation.yield(InboundFrame(message: message, relayUrl: relayUrl))
+                // order, independently of every other relay's pipeline. Routing
+                // through the lock-guarded router keeps this off the main actor
+                // (no per-frame main hop).
+                self.inboundRouter.yield(InboundFrame(message: message), to: relayUrl)
 
                 // Continue receiving
                 Task { @MainActor in
@@ -1188,6 +1222,7 @@ final class NostrRelayManager: ObservableObject {
 
     private func handleDisconnection(relayUrl: String, error: Error) {
         connections.removeValue(forKey: relayUrl)
+        teardownRelayInboundPipeline(for: relayUrl)
         subscriptions.removeValue(forKey: relayUrl)
         updateRelayStatus(relayUrl, isConnected: false, error: error)
         settleEOSETrackers(droppingRelay: relayUrl)
@@ -1276,8 +1311,9 @@ final class NostrRelayManager: ObservableObject {
         if let connection = connections[normalizedRelayUrl] {
             connection.cancel(with: .goingAway, reason: nil)
             connections.removeValue(forKey: normalizedRelayUrl)
+            teardownRelayInboundPipeline(for: normalizedRelayUrl)
         }
-        
+
         // Attempt immediate reconnection
         connectToRelay(normalizedRelayUrl)
     }
@@ -1370,6 +1406,75 @@ final class NostrRelayManager: ObservableObject {
 
 // MARK: - Off-main inbound parsing helpers (file scope, non-isolated)
 
+/// A single raw socket frame awaiting off-main parse + Schnorr verification.
+private struct InboundFrame: Sendable {
+    let message: URLSessionWebSocketTask.Message
+}
+
+/// Lock-guarded registry of per-relay inbound streams.
+///
+/// The raw WebSocket receive callback is not main-actor isolated, so it needs a
+/// `Sendable` path to route a frame to the correct relay's stream without a
+/// per-frame hop onto the main actor. Pipeline lifecycle (start/finish) is
+/// driven from the main actor; frame delivery (`yield`) can come from any
+/// thread. All access is serialized by a single lock — contention is negligible
+/// because the guarded critical section is only a dictionary lookup + yield.
+private final class InboundFrameRouter: @unchecked Sendable {
+    private let lock = NSLock()
+    private var continuations: [String: AsyncStream<InboundFrame>.Continuation] = [:]
+    private var tasks: [String: Task<Void, Never>] = [:]
+
+    /// Start a relay's stream + consumer if one does not already exist.
+    /// Returns true when a new pipeline was created. The bounded
+    /// `.bufferingNewest` policy makes a single relay shed its OWN oldest
+    /// frames under a flood, never other relays' frames and never memory.
+    func startPipeline(
+        for relayUrl: String,
+        makeConsumer: (AsyncStream<InboundFrame>) -> Task<Void, Never>
+    ) -> Bool {
+        lock.lock()
+        defer { lock.unlock() }
+        if continuations[relayUrl] != nil { return false }
+        let (stream, continuation) = AsyncStream<InboundFrame>.makeStream(
+            bufferingPolicy: .bufferingNewest(TransportConfig.nostrInboundPerRelayBufferCap)
+        )
+        continuations[relayUrl] = continuation
+        tasks[relayUrl] = makeConsumer(stream)
+        return true
+    }
+
+    /// Route a frame to a relay's stream. No-op if the relay has no live
+    /// pipeline (socket already torn down) — the frame is simply dropped, which
+    /// is safe for best-effort Nostr inbound.
+    func yield(_ frame: InboundFrame, to relayUrl: String) {
+        lock.lock()
+        let continuation = continuations[relayUrl]
+        lock.unlock()
+        continuation?.yield(frame)
+    }
+
+    /// Finish a relay's stream. The consumer drains any already-buffered frames
+    /// before exiting, so in-flight verified events are still delivered.
+    func finishPipeline(for relayUrl: String) {
+        lock.lock()
+        let continuation = continuations.removeValue(forKey: relayUrl)
+        tasks.removeValue(forKey: relayUrl)
+        lock.unlock()
+        continuation?.finish()
+    }
+
+    func finishAll() {
+        lock.lock()
+        let allContinuations = continuations
+        continuations.removeAll()
+        tasks.removeAll()
+        lock.unlock()
+        for continuation in allContinuations.values {
+            continuation.finish()
+        }
+    }
+}
+
 private enum ParsedInbound {
     case event(subId: String, event: NostrEvent)
     case ok(eventId: String, success: Bool, reason: String)

bitchat/Services/TransportConfig.swift

@@ -55,6 +55,14 @@ enum TransportConfig {
     static let nostrDuplicateEventLogInterval: Int = 50
     // Sample interval for per-event debug logs on the inbound hot path.
     static let nostrInboundEventLogInterval: Int = 100
+    // Bounded per-relay inbound frame buffer. Each relay connection owns its
+    // own serial verify pipeline; if a relay floods faster than its Schnorr
+    // verification drains, the oldest buffered frames for THAT relay are
+    // dropped (bufferingNewest) so one relay can neither exhaust memory nor
+    // stall other relays. Nostr inbound is already best-effort (relays are
+    // redundant and events replay), so dropping a flooding relay's backlog is
+    // safe.
+    static let nostrInboundPerRelayBufferCap: Int = 256
 
     // Conversation store diagnostics (field observability)
     // Sample interval for the periodic store-audit "OK" heartbeat line

bitchatTests/Services/NostrRelayManagerTests.swift

@@ -767,9 +767,9 @@ final class NostrRelayManagerTests: XCTestCase {
         XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, 0)
     }
 
-    /// Signature verification moved off-main into a single serial consumer;
+    /// Signature verification runs off-main in a per-relay serial consumer;
     /// several frames buffered on one socket must still be delivered to the
-    /// handler in arrival order.
+    /// handler in that relay's arrival order.
     func test_receiveEvent_deliversBackToBackEventsInArrivalOrder() async throws {
         let relayURL = "wss://ordered.example"
         let context = makeContext(permission: .denied)
@@ -795,6 +795,64 @@ final class NostrRelayManagerTests: XCTestCase {
         XCTAssertEqual(receivedIDs, events.map(\.id))
     }
 
+    /// Each relay owns its own off-main verify pipeline, so a large backlog of
+    /// EVENT frames on one relay must NOT head-of-line-block a frame that
+    /// arrives on a different relay. Under the previous single global consumer,
+    /// relay B's single event (emitted after relay A's whole burst) could only
+    /// be delivered once every frame in A's backlog had been Schnorr-verified;
+    /// with per-relay pipelines B verifies concurrently and lands before A's
+    /// backlog drains.
+    func test_receiveEvent_busyRelayDoesNotBlockOtherRelayDelivery() async throws {
+        let busyRelayURL = "wss://busy-relay.example"
+        let quietRelayURL = "wss://quiet-relay.example"
+        let context = makeContext(permission: .denied)
+
+        // Distinct subscriptions per relay so dedup never coalesces A vs. B.
+        let busyEvents = try (0..<200).map { try makeSignedEvent(content: "busy-\($0)") }
+        let quietEvent = try makeSignedEvent(content: "quiet")
+
+        var busyDeliveredCount = 0
+        var quietDeliveredAfterBusyCount = -1 // busy-count observed when B lands
+
+        context.manager.subscribe(filter: makeFilter(), id: "busy", relayUrls: [busyRelayURL]) { _ in
+            busyDeliveredCount += 1
+        }
+        context.manager.subscribe(filter: makeFilter(), id: "quiet", relayUrls: [quietRelayURL]) { _ in
+            if quietDeliveredAfterBusyCount < 0 {
+                quietDeliveredAfterBusyCount = busyDeliveredCount
+            }
+        }
+        let subscribed = await waitUntil {
+            context.sessionFactory.latestConnection(for: busyRelayURL)?.sentStrings.count == 1 &&
+            context.sessionFactory.latestConnection(for: quietRelayURL)?.sentStrings.count == 1
+        }
+        XCTAssertTrue(subscribed)
+
+        // Flood relay A first, then emit a single frame on relay B.
+        for event in busyEvents {
+            try context.sessionFactory.latestConnection(for: busyRelayURL)?.emitEventMessage(subscriptionID: "busy", event: event)
+        }
+        try context.sessionFactory.latestConnection(for: quietRelayURL)?.emitEventMessage(subscriptionID: "quiet", event: quietEvent)
+
+        let quietDelivered = await waitUntil(timeout: 5.0) { quietDeliveredAfterBusyCount >= 0 }
+        XCTAssertTrue(quietDelivered, "relay B's event was never delivered")
+
+        // The signal: B did not have to wait for A's entire backlog. If the two
+        // pipelines were globally serialized, B could only land after all 200 of
+        // A's frames, so busyDeliveredCount would be 200 when B arrived.
+        XCTAssertLessThan(
+            quietDeliveredAfterBusyCount,
+            busyEvents.count,
+            "relay B was head-of-line blocked behind relay A's backlog"
+        )
+
+        // Both relays still drain fully and in order.
+        let allDelivered = await waitUntil(timeout: 5.0) {
+            busyDeliveredCount == busyEvents.count
+        }
+        XCTAssertTrue(allDelivered)
+    }
+
     func test_receiveEvent_withoutHandlerStillTracksReceivedCount() async throws {
         let relayURL = "wss://missing-handler.example"
         let context = makeContext(permission: .denied)