From a21f811e87d06c66a7130b02dbde2589920a079e Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 3 Jun 2026 20:15:06 +0000 Subject: [PATCH] refactor(quartz): clarity pass on relay server internals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Behaviour-preserving readability improvements (full suite green): - LimitsPolicy: drop the generic reject helpers (which forced rejectSubId(...) call-site type args). Each check is now a "rejection reason or null" function (eventRejection / subscriptionRejection) and the accept overloads just wrap a non-null reason — the three overloads read almost identically. - Extract ConnectionRegistry: NostrServer and ReqResponderServer duplicated the connection bookkeeping (stable-id keying, active gauge, once-only teardown accounting). That subtle logic now lives in one named class both servers delegate to; their connect()/close() shrink to the parts that actually differ (the backend and what else teardown closes). - HyperLogLog.addPubKey: rename `ri` -> `registerIndex`. https://claude.ai/code/session_016YBS2pWCBSDgAMthHzfCTr --- .../relay/server/ConnectionRegistry.kt | 78 +++++++++++++++++++ .../nip01Core/relay/server/NostrServer.kt | 43 +++------- .../relay/server/ReqResponderServer.kt | 41 +++------- .../relay/server/policies/LimitsPolicy.kt | 55 ++++++------- .../quartz/nip45Count/HyperLogLog.kt | 4 +- 5 files changed, 123 insertions(+), 98 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ConnectionRegistry.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ConnectionRegistry.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ConnectionRegistry.kt new file mode 100644 index 0000000000..a21f2c615e --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ConnectionRegistry.kt @@ -0,0 +1,78 @@ +/* + * 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.server + +import com.vitorpamplona.quartz.utils.cache.LargeCache +import kotlin.concurrent.atomics.AtomicLong +import kotlin.concurrent.atomics.ExperimentalAtomicApi + +/** + * Tracks the live [RelaySession]s of one relay server and fires the + * [RelayServerListener] as they come and go. Shared by [NostrServer] and + * [ReqResponderServer] so the connection bookkeeping — the stable-id keying, + * the [active] gauge, and the once-only teardown accounting — lives in exactly + * one place. + */ +@OptIn(ExperimentalAtomicApi::class) +class ConnectionRegistry( + private val listener: RelayServerListener, +) { + private val connections = LargeCache() + private val activeCount = AtomicLong(0L) + + /** Number of connections currently registered. */ + val active: Long get() = activeCount.load() + + /** Records [session] and fires [RelayServerListener.onConnect]. Returns it for chaining. */ + fun register(session: RelaySession): RelaySession { + connections.put(session.id, session) + activeCount.addAndFetch(1L) + listener.onConnect(session.id) + return session + } + + /** + * Drops the connection with [id], decrementing [active] and firing + * [RelayServerListener.onDisconnect] — but only on the first call for a + * given connection, so a double `close()` can't underflow the gauge or + * double-fire the listener. + */ + fun unregister(id: Long) { + if (connections.remove(id) != null) { + activeCount.addAndFetch(-1L) + listener.onDisconnect(id) + } + } + + /** + * Cancels every still-open connection's subscriptions and fires + * [RelayServerListener.onDisconnect] for each, then resets the registry. + * Used by the server's `close()`. + */ + fun closeAll() { + connections.forEach { _, session -> + session.cancelAllSubscriptions() + listener.onDisconnect(session.id) + } + connections.clear() + activeCount.store(0L) + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index 23599b33bd..adb3ee61b3 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -25,12 +25,9 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.LimitsPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings -import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel -import kotlin.concurrent.atomics.AtomicLong -import kotlin.concurrent.atomics.ExperimentalAtomicApi import kotlin.coroutines.CoroutineContext /** @@ -57,22 +54,18 @@ import kotlin.coroutines.CoroutineContext * subscription caps) and advertised via [RelayLimits.toNip11Limitation]. * Null disables limit enforcement. */ -@OptIn(ExperimentalAtomicApi::class) class NostrServer( private val store: IEventStore, private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy }, private val parentContext: CoroutineContext = SupervisorJob(), parallelVerify: Boolean = false, private val negentropySettings: NegentropySettings = NegentropySettings.Default, - private val listener: RelayServerListener = RelayServerListener.None, + listener: RelayServerListener = RelayServerListener.None, val limits: RelayLimits? = null, ) : AutoCloseable { /** Scope for all subscriptions. */ private val scope = CoroutineScope(parentContext + SupervisorJob()) - /** Live count of registered connections; backs [activeConnections]. */ - private val activeCount = AtomicLong(0L) - /** * Group-commit writer shared across every connected session. * Sessions hand off EVENT publishes here instead of awaiting @@ -89,11 +82,10 @@ class NostrServer( private val subStore = LiveEventStore(store, ingest) - /** Active client sessions keyed by [RelaySession.id]. */ - private val connections = LargeCache() + private val connections = ConnectionRegistry(listener) /** Number of connections currently registered with this server. */ - val activeConnections: Long get() = activeCount.load() + val activeConnections: Long get() = connections.active /** * Builds the per-connection policy, prepending a [LimitsPolicy] when @@ -111,29 +103,17 @@ class NostrServer( * @param send Callback the server uses to send JSON messages to this client. * Implementations must be safe to call from any coroutine. */ - fun connect(send: (String) -> Unit): RelaySession { - val session = + fun connect(send: (String) -> Unit): RelaySession = + connections.register( RelaySession( policy = buildPolicy(), store = subStore, scope = scope, onSend = send, - onClose = { closed -> - // Idempotent: only account for the first teardown of a - // given connection so a double close() can't underflow - // the gauge or double-fire the listener. - if (connections.remove(closed.id) != null) { - activeCount.addAndFetch(-1L) - listener.onDisconnect(closed.id) - } - }, + onClose = { connections.unregister(it.id) }, negentropySettings = negentropySettings, - ) - connections.put(session.id, session) - activeCount.addAndFetch(1L) - listener.onConnect(session.id) - return session - } + ), + ) /** * Registers a new client connection and serves it for the duration of @@ -159,12 +139,7 @@ class NostrServer( * Shuts down the server, cancelling all subscriptions and closing the store. */ override fun close() { - connections.forEach { _, session -> - session.cancelAllSubscriptions() - listener.onDisconnect(session.id) - } - connections.clear() - activeCount.store(0L) + connections.closeAll() ingest.close() scope.cancel() store.close() diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponderServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponderServer.kt index a758635da1..c28d62f5db 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponderServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponderServer.kt @@ -23,12 +23,9 @@ package com.vitorpamplona.quartz.nip01Core.relay.server import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.LimitsPolicy import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings -import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel -import kotlin.concurrent.atomics.AtomicLong -import kotlin.concurrent.atomics.ExperimentalAtomicApi import kotlin.coroutines.CoroutineContext /** @@ -72,13 +69,12 @@ import kotlin.coroutines.CoroutineContext * subscription caps) and advertised via [RelayLimits.toNip11Limitation]. * Null disables limit enforcement. */ -@OptIn(ExperimentalAtomicApi::class) class ReqResponderServer( responder: ReqResponder, private val policyBuilder: () -> IRelayPolicy = { EmptyPolicy }, parentContext: CoroutineContext = SupervisorJob(), private val negentropySettings: NegentropySettings = NegentropySettings.Default, - private val listener: RelayServerListener = RelayServerListener.None, + listener: RelayServerListener = RelayServerListener.None, val limits: RelayLimits? = null, ) : AutoCloseable { /** Scope for all subscriptions. */ @@ -86,14 +82,10 @@ class ReqResponderServer( private val backend = ReqResponderBackend(responder) - /** Live count of registered connections; backs [activeConnections]. */ - private val activeCount = AtomicLong(0L) - - /** Active client sessions keyed by [RelaySession.id]. */ - private val connections = LargeCache() + private val connections = ConnectionRegistry(listener) /** Number of connections currently registered with this server. */ - val activeConnections: Long get() = activeCount.load() + val activeConnections: Long get() = connections.active private fun buildPolicy(): IRelayPolicy { val base = policyBuilder() @@ -106,27 +98,17 @@ class ReqResponderServer( * @param send Callback the server uses to send JSON messages to this client. * Implementations must be safe to call from any coroutine. */ - fun connect(send: (String) -> Unit): RelaySession { - val session = + fun connect(send: (String) -> Unit): RelaySession = + connections.register( RelaySession( policy = buildPolicy(), store = backend, scope = scope, onSend = send, - onClose = { closed -> - // Idempotent teardown accounting (see NostrServer.connect). - if (connections.remove(closed.id) != null) { - activeCount.addAndFetch(-1L) - listener.onDisconnect(closed.id) - } - }, + onClose = { connections.unregister(it.id) }, negentropySettings = negentropySettings, - ) - connections.put(session.id, session) - activeCount.addAndFetch(1L) - listener.onConnect(session.id) - return session - } + ), + ) /** * Registers a new client connection and serves it for the duration of @@ -150,12 +132,7 @@ class ReqResponderServer( /** Shuts down the server, cancelling all subscriptions. */ override fun close() { - connections.forEach { _, session -> - session.cancelAllSubscriptions() - listener.onDisconnect(session.id) - } - connections.clear() - activeCount.store(0L) + connections.closeAll() scope.cancel() } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/LimitsPolicy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/LimitsPolicy.kt index e8c085af90..d413f59566 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/LimitsPolicy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/LimitsPolicy.kt @@ -20,8 +20,8 @@ */ package com.vitorpamplona.quartz.nip01Core.relay.server.policies +import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix -import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CountCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd @@ -42,13 +42,16 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.RelayLimits * caps ([RelayLimits.maxMessageLength], [RelayLimits.maxSubscriptions]) are * enforced through the [acceptMessage] / [acceptSubscription] policy hooks, so * everything limit-related composes uniformly across a [PolicyStack]. + * + * Each check is expressed as a "rejection reason or null" helper; the command + * `accept` overloads just wrap a non-null reason in [PolicyResult.Rejected]. */ class LimitsPolicy( private val limits: RelayLimits, ) : PassThroughPolicy() { override fun acceptMessage(message: String): String? { val max = limits.maxMessageLength ?: return null - return if (message.length > max) MachineReadablePrefix.INVALID.format("message too large (max $max)") else null + return if (message.length > max) invalid("message too large (max $max)") else null } override fun acceptSubscription( @@ -63,49 +66,41 @@ class LimitsPolicy( } } - override fun accept(cmd: EventCmd): PolicyResult { - val event = cmd.event - limits.maxContentLength?.let { - if (event.content.length > it) return reject("content too large (max $it)") - } - limits.maxEventTags?.let { - if (event.tags.size > it) return reject("too many tags (max $it)") - } - limits.createdAtLowerLimit?.let { - if (event.createdAt < it) return reject("created_at is before the relay's lower limit") - } - limits.createdAtUpperLimit?.let { - if (event.createdAt > it) return reject("created_at is after the relay's upper limit") - } - return PolicyResult.Accepted(cmd) - } + override fun accept(cmd: EventCmd): PolicyResult = eventRejection(cmd.event)?.let { PolicyResult.Rejected(it) } ?: PolicyResult.Accepted(cmd) override fun accept(cmd: ReqCmd): PolicyResult { - rejectSubId(cmd.subId)?.let { return it } - rejectFilterCount(cmd.filters)?.let { return it } + subscriptionRejection(cmd.subId, cmd.filters)?.let { return PolicyResult.Rejected(it) } val clamped = clampLimits(cmd.filters) return PolicyResult.Accepted(if (clamped === cmd.filters) cmd else ReqCmd(cmd.subId, clamped)) } override fun accept(cmd: CountCmd): PolicyResult { - rejectSubId(cmd.queryId)?.let { return it } - rejectFilterCount(cmd.filters)?.let { return it } + subscriptionRejection(cmd.queryId, cmd.filters)?.let { return PolicyResult.Rejected(it) } val clamped = clampLimits(cmd.filters) return PolicyResult.Accepted(if (clamped === cmd.filters) cmd else CountCmd(cmd.queryId, clamped)) } - private fun reject(message: String): PolicyResult = PolicyResult.Rejected(MachineReadablePrefix.INVALID.format(message)) - - private fun rejectSubId(subId: String): PolicyResult? { - val max = limits.maxSubidLength ?: return null - return if (subId.length > max) reject("subscription id too long (max $max)") else null + /** The reason an EVENT violates a limit, or null when it's within bounds. */ + private fun eventRejection(event: Event): String? { + limits.maxContentLength?.let { if (event.content.length > it) return invalid("content too large (max $it)") } + limits.maxEventTags?.let { if (event.tags.size > it) return invalid("too many tags (max $it)") } + limits.createdAtLowerLimit?.let { if (event.createdAt < it) return invalid("created_at is before the relay's lower limit") } + limits.createdAtUpperLimit?.let { if (event.createdAt > it) return invalid("created_at is after the relay's upper limit") } + return null } - private fun rejectFilterCount(filters: List): PolicyResult? { - val max = limits.maxFilters ?: return null - return if (filters.size > max) reject("too many filters (max $max)") else null + /** The reason a REQ/COUNT violates a limit, or null when it's within bounds. */ + private fun subscriptionRejection( + subId: String, + filters: List, + ): String? { + limits.maxSubidLength?.let { if (subId.length > it) return invalid("subscription id too long (max $it)") } + limits.maxFilters?.let { if (filters.size > it) return invalid("too many filters (max $it)") } + return null } + private fun invalid(message: String): String = MachineReadablePrefix.INVALID.format(message) + /** Returns the same list reference when nothing changes, so callers can skip rebuilding the command. */ private fun clampLimits(filters: List): List { if (limits.maxLimit == null && limits.defaultLimit == null) return filters diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip45Count/HyperLogLog.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip45Count/HyperLogLog.kt index bfda58cb6c..e00ca410bc 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip45Count/HyperLogLog.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip45Count/HyperLogLog.kt @@ -131,9 +131,9 @@ object HyperLogLog { offset: Int, ) { if (offset < 0 || offset >= pubKey.size) return - val ri = pubKey[offset].toInt() and 0xFF + val registerIndex = pubKey[offset].toInt() and 0xFF val value = (leadingZeroBits(pubKey, offset + 1) + 1).coerceAtMost(0xFF) - if (value > (registers[ri].toInt() and 0xFF)) registers[ri] = value.toByte() + if (value > (registers[registerIndex].toInt() and 0xFF)) registers[registerIndex] = value.toByte() } /**