diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 19648f07cc..a731890e88 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.KindAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PubkeyAllowDenyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RejectFutureEventsPolicy +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore @@ -104,10 +105,7 @@ fun main(args: Array) { val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl) val policyBuilder: () -> IRelayPolicy = { - // When parallel verify is enabled the IngestQueue runs - // Schnorr verify off the WS pump, so the policy chain skips - // VerifyPolicy to avoid double-verifying every event. - composePolicy(config, advertisedUrl, requireAuth, verifySigs && !parallelVerify) + composePolicy(config, advertisedUrl, requireAuth, verifySigs, parallelVerify) } val stateFile = config.admin.state_file?.let { File(it) } @@ -168,6 +166,7 @@ private fun composePolicy( advertisedUrl: com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl, requireAuth: Boolean, verifySigs: Boolean, + parallelVerify: Boolean, ): IRelayPolicy { val pieces = mutableListOf() @@ -188,7 +187,11 @@ private fun composePolicy( } if (verifySigs) { - pieces += VerifyPolicy + // When parallel verify is on, the IngestQueue handles EVENT + // verification on the writer's CPU fan-out — but AUTH events + // bypass the queue, so we still need the policy chain to + // verify those. `VerifyAuthOnlyPolicy` does exactly that. + pieces += if (parallelVerify) VerifyAuthOnlyPolicy else VerifyPolicy } return pieces.fold(EmptyPolicy) { acc, p -> diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt index b93b236ae9..5b4876693c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt @@ -30,6 +30,7 @@ import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.ClosedReceiveChannelException import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import kotlin.coroutines.CoroutineContext @@ -168,43 +169,53 @@ class IngestQueue( processBatch(batch) batch.clear() } - } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) { + } catch (_: ClosedReceiveChannelException) { // Normal shutdown via close(). } } private suspend fun processBatch(batch: List) { - // Tier 3: parallel pre-insert verify. Each batch entry that - // fails verification is pre-marked Rejected and excluded from - // the SQLite transaction. The remaining (verified) events go - // through batchInsert in original order; we re-stitch outcomes - // back to the full batch by index. - val verifyResults: BooleanArray? = - verify?.let { hook -> - coroutineScope { - batch - .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } - .awaitAll() - .toBooleanArray() - } - } + val verifyResults = verifyBatch(batch) + val finalOutcomes = runInsertStage(batch, verifyResults) + dispatchOutcomes(batch, finalOutcomes) + } - val toInsert: List - val insertIndices: IntArray - if (verifyResults == null) { - toInsert = batch.map { it.event } - insertIndices = IntArray(batch.size) { it } - } else { - val accepted = ArrayList(batch.size) - val mapping = ArrayList(batch.size) - for (i in batch.indices) { - if (verifyResults[i]) { - accepted.add(batch[i].event) - mapping.add(i) - } + /** + * Tier 3: per-row verify. Returns `null` when no [verify] hook is + * configured (skip the stage entirely). For multi-event batches + * each verify runs as its own `async(Default)` so they spread + * across CPU cores; single-event batches short-circuit to a + * direct call to avoid coroutine-scope overhead. + */ + private suspend fun verifyBatch(batch: List): BooleanArray? { + val hook = verify ?: return null + if (batch.size == 1) return BooleanArray(1) { hook(batch[0].event) } + return coroutineScope { + batch + .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } + .awaitAll() + .toBooleanArray() + } + } + + /** + * Run the SQLite transaction for the verified subset of [batch] + * and stitch outcomes back to a per-batch-index array. Failed + * verifies pre-mark `Rejected` and skip the insert. A whole-batch + * commit failure converts every persisted entry to `Rejected` + * with the throw message. + */ + private suspend fun runInsertStage( + batch: List, + verifyResults: BooleanArray?, + ): Array { + val toInsert = ArrayList(batch.size) + val insertIndices = ArrayList(batch.size) + for (i in batch.indices) { + if (verifyResults == null || verifyResults[i]) { + toInsert.add(batch[i].event) + insertIndices.add(i) } - toInsert = accepted - insertIndices = mapping.toIntArray() } val insertOutcomes: List = @@ -220,11 +231,10 @@ class IngestQueue( } } - // Build a per-batch-index outcome array. val finalOutcomes = arrayOfNulls(batch.size) for (j in insertIndices.indices) { - finalOutcomes[insertIndices[j]] = insertOutcomes.getOrNull(j) - ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + finalOutcomes[insertIndices[j]] = + insertOutcomes.getOrNull(j) ?: missingOutcome } if (verifyResults != null) { for (i in batch.indices) { @@ -233,12 +243,16 @@ class IngestQueue( } } } + return finalOutcomes + } + private fun dispatchOutcomes( + batch: List, + outcomes: Array, + ) { for (i in batch.indices) { val sub = batch[i] - val outcome = - finalOutcomes[i] - ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + val outcome = outcomes[i] ?: missingOutcome try { sub.onComplete(outcome) } catch (e: Throwable) { @@ -262,6 +276,14 @@ class IngestQueue( } companion object { + /** + * Default for a missing per-row outcome — only reachable on a + * contract violation (the store returned fewer outcomes than + * inserts), so the message is informational, not user-facing. + */ + private val missingOutcome = + IEventStore.InsertOutcome.Rejected("internal error: missing outcome") + /** * Cap per batch. Sized to keep per-batch latency low (each * transaction holds the SQLite writer mutex; over-large 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 54c23fbe00..0ec8e602be 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 @@ -20,7 +20,6 @@ */ package com.vitorpamplona.quartz.nip01Core.relay.server -import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.crypto.verify import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore @@ -65,7 +64,7 @@ class NostrServer( IngestQueue( store = store, parentContext = parentContext, - verify = if (parallelVerify) ::verifyEvent else null, + verify = if (parallelVerify) ({ it.verify() }) else null, ) private val subStore = LiveEventStore(store, ingest) @@ -128,11 +127,4 @@ class NostrServer( */ @Deprecated("Use close() instead", replaceWith = ReplaceWith("close()")) fun shutdown() = close() - - private companion object { - // Function reference (`Event::verify`) wrapper so the - // ingest hook keeps a single instance per server rather - // than allocating a fresh lambda on every call site. - private fun verifyEvent(event: Event): Boolean = event.verify() - } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 5c2c56f33b..c1698492a3 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -43,6 +43,7 @@ import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.launch /** @@ -147,7 +148,7 @@ class RelaySession( } } } - } catch (_: kotlinx.coroutines.channels.ClosedSendChannelException) { + } catch (_: ClosedSendChannelException) { // Server is shutting down — the queue is closed. Reply // with a transient failure so a client re-trying against // the next instance gets a sane signal; the WS itself is diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt index 283fda6707..e59bd53088 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt @@ -31,13 +31,24 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.IRelayPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.PolicyResult /** - * Allows all commands without authentication. This is the default policy. + * Verifies the Schnorr signature + id hash of every incoming + * `EVENT` and `AUTH` event. Other commands pass through. + * + * The default `VerifyPolicy` singleton verifies both. The + * [VerifyAuthOnlyPolicy] singleton skips the EVENT path — use it + * when the [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue] + * is doing parallel verify (Tier 3 of the event-ingestion plan) + * so EVENTs aren't verified twice. AUTH is still verified inline + * because the AUTH path bypasses the queue entirely; without it, + * a forged event could mark a pubkey as authenticated. */ -object VerifyPolicy : IRelayPolicy { +open class VerifyEventsAndAuthPolicy( + private val verifyEvents: Boolean, +) : IRelayPolicy { override fun onConnect(send: (Message) -> Unit) { } override fun accept(cmd: EventCmd) = - if (cmd.event.verify()) { + if (!verifyEvents || cmd.event.verify()) { PolicyResult.Accepted(cmd) } else { PolicyResult.Rejected("invalid: bad signature or id") @@ -56,3 +67,18 @@ object VerifyPolicy : IRelayPolicy { override fun canSendToSession(event: Event) = true } + +/** + * Default verify policy — checks every EVENT and every AUTH. Use + * this when nothing else in the stack verifies signatures. + */ +object VerifyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = true) + +/** + * Verify policy that skips the EVENT path. Use when the + * `IngestQueue` is configured with `parallelVerify`, so EVENTs are + * verified once on the writer's CPU fan-out instead of inline on + * the WebSocket pump. AUTH is still verified inline because that + * command never reaches the queue. + */ +object VerifyAuthOnlyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = false) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt index eedca33713..cb18c13a96 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -137,8 +137,11 @@ class ObservableEventStore( } } - @Suppress("UNCHECKED_CAST") - return outcomes.toList() as List + // Every index in `events.indices` was populated above (either + // from the ephemeral pre-pass or the `inner.batchInsert` + // result), so no nulls remain. `requireNoNulls()` enforces + // that with a runtime check instead of an unchecked cast. + return outcomes.requireNoNulls().asList() } override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) {