diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt index c89de32554..d26e37e594 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt +++ b/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 = 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 = Channel(BasicOkHttpWebSocket.RECEIVE_BUFFER_FRAMES) val job = // Launch a coroutine to process messages from the channel. scope.launch { for (message in incomingMessages) { diff --git a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md index 38d519f32d..6a9b4b1d12 100644 --- a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md +++ b/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 diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt index a8fdb1d78d..1bdef6d0ab 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt +++ b/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 = Channel(Channel.UNLIMITED) + val incomingMessages: Channel = 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) } diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BoundedReceiveBufferTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BoundedReceiveBufferTest.kt new file mode 100644 index 0000000000..07feaccdc5 --- /dev/null +++ b/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() + } + } +}