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 d26e37e594..4c70c12498 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,7 +24,6 @@ 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 @@ -78,10 +77,13 @@ class OkHttpWebSocket( ) : okhttp3.WebSocketListener() { val scope = CoroutineScope(Dispatchers.IO + exceptionHandler) - // 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) + // UNLIMITED on purpose — do NOT bound this channel. The app holds + // 2000+ relay connections; a bounded buffer under a slow consumer + // would block OkHttp reader threads and park the backlog on the + // relay's outbound buffers — infrastructure that isn't ours. Drain + // the remote as fast as it can send; consumer speed is handled + // downstream. + val incomingMessages: Channel = Channel(Channel.UNLIMITED) 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 234307a084..4fd84f3a5d 100644 --- a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md +++ b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md @@ -497,18 +497,26 @@ 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) +### Bounded per-connection receive buffer (REVERTED — deliberate design choice) -`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. +A 4096-frame bound on the reader→consumer channel was tried (backpressure +via TCP flow control instead of unbounded heap; proven lossless by a +slow-consumer test) and then **reverted by explicit maintainer decision**, +for two reasons that outweigh the heap tail-risk: + +1. **The remote infrastructure isn't ours.** Backpressure parks the backlog + in the RELAY's outbound buffers/TCP window — a client should release the + relay from its duties as fast as the relay can send, and own the + buffering itself. +2. **The app holds 2000+ simultaneous relay connections.** A bounded buffer + under a slow consumer blocks OkHttp reader THREADS; at that connection + count, blocked readers are a thread-starvation hazard far worse than the + heap growth they prevent. + +The `Channel.UNLIMITED` receive queues now carry an explicit "do not bound" +comment with this rationale. The slow-consumer risk is addressed from the +other side instead: make the consumer fast enough that backlogs don't form +(CachingEventDecoder, ParallelEventVerifier, PoolRequests sharding). ### ParallelEventVerifier: batched verify off the receiver coroutine (done) 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 1bdef6d0ab..f0a9c61cec 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,7 +42,6 @@ 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 @@ -50,23 +49,6 @@ 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 @@ -79,7 +61,17 @@ class BasicOkHttpWebSocket( val listener = object : OkHttpWebSocketListener() { val scope = CoroutineScope(Dispatchers.IO + exceptionHandler) - val incomingMessages: Channel = Channel(receiveBufferFrames) + + // UNLIMITED on purpose — do NOT bound this channel. The app + // holds 2000+ relay connections; a bounded buffer under a + // slow consumer would block OkHttp reader threads (thread + // starvation at that connection count) and park the backlog + // on the RELAY's outbound buffers via TCP backpressure — + // infrastructure that isn't ours. We drain the remote as + // fast as it can send and own the buffering; consumer speed + // is handled downstream (CachingEventDecoder, + // ParallelEventVerifier). + val incomingMessages: Channel = Channel(Channel.UNLIMITED) val job = // Launch a coroutine to process messages from the channel. scope.launch { for (message in incomingMessages) { @@ -99,8 +91,8 @@ class BasicOkHttpWebSocket( webSocket: OkHttpWebSocket, text: String, ) { - // Blocks the OkHttp reader thread when the buffer is full — - // that's the backpressure mechanism, see RECEIVE_BUFFER_FRAMES. + // Never blocks (unlimited channel): the OkHttp reader + // thread must stay free to keep draining the socket. 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 deleted file mode 100644 index 07feaccdc5..0000000000 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BoundedReceiveBufferTest.kt +++ /dev/null @@ -1,131 +0,0 @@ -/* - * 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() - } - } -}