perf: bound the per-connection websocket receive buffer

The reader-thread-to-consumer channel in BasicOkHttpWebSocket and the
app's OkHttpWebSocket was UNLIMITED: a consumer slower than the socket
accumulated frame Strings without bound (gigabytes over a multi-million
event download). Now capped at 4096 frames — when full, OkHttp's reader
thread blocks and TCP flow control pushes back on the relay instead of
the heap. The trade-off (a blocked reader delays PING/PONG handling) is
documented on the constant; the bound is deep enough that only a
pathologically slow consumer hits it.

BoundedReceiveBufferTest forces sustained backpressure (8-frame buffer,
sleeping consumer, real socket to a local geode relay) and asserts all
events plus EOSE arrive in order with no drops or deadlock.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018saXqYfAa3RvSJoDXK591R

amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt

@@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket.Companion.exceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -76,7 +77,11 @@ class OkHttpWebSocket(
         val out: WebSocketListener,
     ) : okhttp3.WebSocketListener() {
         val scope = CoroutineScope(Dispatchers.IO + exceptionHandler)
-        val incomingMessages: Channel<String> = Channel(Channel.UNLIMITED)
+
+        // Bounded: a consumer slower than the socket blocks OkHttp's reader
+        // thread (TCP backpressure) instead of accumulating unbounded heap.
+        // See BasicOkHttpWebSocket.RECEIVE_BUFFER_FRAMES for the rationale.
+        val incomingMessages: Channel<String> = Channel(BasicOkHttpWebSocket.RECEIVE_BUFFER_FRAMES)
         val job = // Launch a coroutine to process messages from the channel.
             scope.launch {
                 for (message in incomingMessages) {

quartz/plans/2026-07-02-nostrclient-receiver-perf.md

@@ -497,6 +497,19 @@ decoder 1.5µs/frame — **6.5×**, with duplicates costing ~0.3µs instead of
 10µs. 7 scan-safety unit tests (`CachingEventDecoderTest`) cover repost
 embedding, escaped subIds, malformed ids, cache rotation, non-EVENT frames.
 
+### Bounded per-connection receive buffer (done)
+
+`BasicOkHttpWebSocket` (quartz) and `OkHttpWebSocket` (amethyst) now bound
+the reader-thread→consumer channel at 4096 frames (was `UNLIMITED`). A
+consumer slower than the socket blocks OkHttp's reader thread — TCP flow
+control pushes back on the relay instead of accumulating frame Strings on
+our heap (a multi-million-event bulk download could previously queue
+gigabytes). Trade-off documented on the constant: a blocked reader also
+delays PING/PONG, so the bound is generous. Validated by
+`BoundedReceiveBufferTest`: an 8-frame buffer against a deliberately slow
+consumer over a real socket — every EVENT plus EOSE arrives, in order, no
+drops, no deadlock.
+
 ## Recommendations (in order of value/risk)
 
 1. **Move Schnorr verification off the receiver coroutine** in the app's

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt

@@ -42,6 +42,7 @@ class BasicOkHttpWebSocket(
     val url: NormalizedRelayUrl,
     val httpClient: (NormalizedRelayUrl) -> OkHttpClient,
     val out: WebSocketListener,
+    val receiveBufferFrames: Int = RECEIVE_BUFFER_FRAMES,
 ) : WebSocket {
     companion object {
         // Exists to avoid exceptions stopping the coroutine
@@ -49,6 +50,23 @@ class BasicOkHttpWebSocket(
             CoroutineExceptionHandler { _, throwable ->
                 Log.e("BasicOkHttpWebSocket", "WebsocketListener Caught exception: ${throwable.message}", throwable)
             }
+
+        /**
+         * Cap on frames buffered between OkHttp's reader thread and the
+         * consumer coroutine. Bounded ON PURPOSE: with an unlimited channel a
+         * consumer slower than the socket accumulates heap without limit (a
+         * multi-million-event bulk download can queue gigabytes of frame
+         * Strings). When full, [trySendBlocking] blocks the OkHttp reader
+         * thread, which stops reading the TCP socket — flow control then
+         * pushes back on the relay instead of on our heap.
+         *
+         * Trade-off: a blocked reader thread also delays OkHttp's PING/PONG
+         * handling, so the bound is generous (4096 frames ≈ 4–40 MB of text
+         * at typical event sizes) — deep enough that only a pathologically
+         * slow consumer ever hits it, and short stalls stay well under relay
+         * ping timeouts.
+         */
+        const val RECEIVE_BUFFER_FRAMES = 4096
     }
 
     private var socket: OkHttpWebSocket? = null
@@ -61,7 +79,7 @@ class BasicOkHttpWebSocket(
         val listener =
             object : OkHttpWebSocketListener() {
                 val scope = CoroutineScope(Dispatchers.IO + exceptionHandler)
-                val incomingMessages: Channel<String> = Channel(Channel.UNLIMITED)
+                val incomingMessages: Channel<String> = Channel(receiveBufferFrames)
                 val job = // Launch a coroutine to process messages from the channel.
                     scope.launch {
                         for (message in incomingMessages) {
@@ -81,9 +99,8 @@ class BasicOkHttpWebSocket(
                     webSocket: OkHttpWebSocket,
                     text: String,
                 ) {
-                    // Asynchronously send the received message to the channel.
-                    // `trySendBlocking` is used here for simplicity within the callback,
-                    // but it's important to understand potential thread blocking if the buffer is full.
+                    // Blocks the OkHttp reader thread when the buffer is full —
+                    // that's the backpressure mechanism, see RECEIVE_BUFFER_FRAMES.
                     incomingMessages.trySendBlocking(text)
                 }
 

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BoundedReceiveBufferTest.kt

@@ -0,0 +1,131 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp
+
+import com.vitorpamplona.geode.KtorRelay
+import com.vitorpamplona.geode.RelayEngine
+import com.vitorpamplona.geode.fixtures.SyntheticEvents
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
+import kotlinx.coroutines.runBlocking
+import okhttp3.OkHttpClient
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * The receive channel between OkHttp's reader thread and the consumer
+ * coroutine is bounded (see [BasicOkHttpWebSocket.RECEIVE_BUFFER_FRAMES]) so a
+ * slow consumer exerts TCP backpressure instead of growing the heap without
+ * limit. Bounding must never DROP frames or deadlock — this test forces
+ * sustained backpressure (an 8-frame buffer against a consumer that sleeps on
+ * every message) and asserts every EVENT plus the trailing EOSE still arrives,
+ * in order.
+ */
+class BoundedReceiveBufferTest {
+    companion object {
+        const val EVENTS = 600
+        const val CONSUMER_DELAY_MS = 3L
+    }
+
+    @Test
+    fun slowConsumerWithTinyBufferReceivesEverythingInOrder() {
+        val placeholder = "ws://127.0.0.1:7771/".normalizeRelayUrl()
+        val engine = RelayEngine(url = placeholder)
+        runBlocking {
+            val events =
+                (1..EVENTS).map {
+                    SyntheticEvents.fakeEvent(idSeed = it, kind = 1, createdAt = it.toLong())
+                }
+            events.chunked(500).forEach { engine.store.batchInsert(it) }
+        }
+        val server = KtorRelay(engine, port = 0).start()
+        val httpClient = OkHttpClient.Builder().build()
+
+        try {
+            val received = AtomicInteger(0)
+            val lastCreatedAt = AtomicInteger(Int.MAX_VALUE) // relays stream newest-first
+            val orderViolations = AtomicInteger(0)
+            val eose = CountDownLatch(1)
+
+            lateinit var socket: BasicOkHttpWebSocket
+            val listener =
+                object : WebSocketListener {
+                    override fun onOpen(
+                        pingMillis: Int,
+                        compression: Boolean,
+                    ) {
+                        socket.send("""["REQ","slow",{"kinds":[1]}]""")
+                    }
+
+                    override fun onMessage(text: String) {
+                        // Deliberately slower than the socket delivers.
+                        Thread.sleep(CONSUMER_DELAY_MS)
+                        if (text.startsWith("[\"EVENT\"")) {
+                            received.incrementAndGet()
+                            val ts = text.substringAfter("\"created_at\":").takeWhile { it.isDigit() }.toIntOrNull()
+                            if (ts != null) {
+                                if (ts > lastCreatedAt.get()) orderViolations.incrementAndGet()
+                                lastCreatedAt.set(ts)
+                            }
+                        } else if (text.startsWith("[\"EOSE\"")) {
+                            eose.countDown()
+                        }
+                    }
+
+                    override fun onClosed(
+                        code: Int,
+                        reason: String,
+                    ) {}
+
+                    override fun onFailure(
+                        t: Throwable,
+                        code: Int?,
+                        response: String?,
+                    ) {
+                        eose.countDown()
+                    }
+                }
+
+            socket =
+                BasicOkHttpWebSocket(
+                    url = server.url.normalizeRelayUrl(),
+                    httpClient = { httpClient },
+                    out = listener,
+                    receiveBufferFrames = 8,
+                )
+            socket.connect()
+
+            assertTrue(eose.await(60, TimeUnit.SECONDS), "EOSE must arrive despite sustained backpressure")
+            assertEquals(EVENTS, received.get(), "bounding the buffer must never drop frames")
+            assertEquals(0, orderViolations.get(), "frame order must be preserved under backpressure")
+
+            socket.disconnect()
+        } finally {
+            httpClient.dispatcher.executorService.shutdown()
+            server.stop(gracePeriodMillis = 200, timeoutMillis = 1_000)
+            engine.close()
+        }
+    }
+}