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 1ec4386619..81326a5bbf 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,6 +20,7 @@ */ 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.backend.IngestQueue import com.vitorpamplona.quartz.nip01Core.relay.server.backend.LiveEventStore @@ -95,7 +96,28 @@ class NostrServer( }, ) - override val backend: SessionBackend = LiveEventStore(store, ingest) + private val liveStore = LiveEventStore(store, ingest) + + override val backend: SessionBackend = liveStore + + /** + * Local ingestion path for events that did not arrive over a client + * connection — e.g. a mirror worker streaming a trusted upstream + * relay, or an import job. Routes through the same group-commit + * [IngestQueue] and live fanout as a client EVENT publish, but skips + * the per-connection policy chain (there is no connection). + * + * [skipVerify] exempts the event from the parallel signature-verify + * hook — the relay-to-relay trust model: pass `true` only for events + * from an explicitly configured upstream that already verified them + * (Schnorr verify profiles at ~8% of busy ingest CPU). The default + * `false` keeps verify-everything semantics. + */ + suspend fun ingest( + event: Event, + skipVerify: Boolean = false, + onComplete: (IEventStore.InsertOutcome) -> Unit, + ) = liveStore.submit(event, skipVerify, onComplete) init { // Deferred-FTS catch-up worker: tokenizes in the gaps between diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt index 4e32fba5ad..39f1191882 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/IngestQueue.kt @@ -27,7 +27,6 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.async -import kotlinx.coroutines.awaitAll import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.ClosedReceiveChannelException @@ -115,9 +114,15 @@ class IngestQueue( /** * One outstanding ingest request: the event to insert plus the * callback the writer fires once the row's outcome is known. + * [skipVerify] exempts this row from the [verify] hook — the + * relay-to-relay trust model: set by local ingestion paths for + * events streamed from an explicitly configured upstream relay + * that already verified them (see + * [com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer.ingest]). */ class Submission( val event: Event, + val skipVerify: Boolean, val onComplete: (IEventStore.InsertOutcome) -> Unit, ) @@ -159,11 +164,12 @@ class IngestQueue( */ suspend fun submit( event: Event, + skipVerify: Boolean = false, onComplete: (IEventStore.InsertOutcome) -> Unit, ) { ensureWriterStarted() pending.addAndFetch(1) - incoming.send(Submission(event, onComplete)) + incoming.send(Submission(event, skipVerify, onComplete)) } private fun ensureWriterStarted() { @@ -234,15 +240,26 @@ class IngestQueue( * 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. + * direct call to avoid coroutine-scope overhead. Rows flagged + * [Submission.skipVerify] (trusted publishers) pass without + * invoking the hook. */ private suspend fun verifyBatch(batch: List): BooleanArray? { val hook = verify ?: return null - if (batch.size == 1) return BooleanArray(1) { hook(batch[0].event) } + if (batch.size == 1) { + val sub = batch[0] + return BooleanArray(1) { sub.skipVerify || hook(sub.event) } + } + if (batch.all { it.skipVerify }) return BooleanArray(batch.size) { true } return coroutineScope { batch - .map { sub -> async(Dispatchers.Default) { hook(sub.event) } } - .awaitAll() + .map { sub -> + if (sub.skipVerify) { + null + } else { + async(Dispatchers.Default) { hook(sub.event) } + } + }.map { it?.await() ?: true } .toBooleanArray() } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt index 89f1fe98fe..6bed5eae8d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt @@ -91,8 +91,23 @@ class LiveEventStore( override suspend fun submit( event: Event, onComplete: (IEventStore.InsertOutcome) -> Unit, + ) = submit(event, skipVerify = false, onComplete = onComplete) + + /** + * [submit] variant for locally-originated traffic (mirror/sync + * workers rather than client connections). [skipVerify] exempts + * this event from the [IngestQueue]'s signature-verification hook — + * the relay-to-relay trust model: set it only for events streamed + * from a configured upstream relay that already verified them. + * Accepted events fan out to live subscribers exactly like a + * client publish. + */ + suspend fun submit( + event: Event, + skipVerify: Boolean, + onComplete: (IEventStore.InsertOutcome) -> Unit, ) { - ingest.submit(event) { outcome -> + ingest.submit(event, skipVerify) { outcome -> if (outcome is IEventStore.InsertOutcome.Accepted) { writeGeneration.addAndFetch(1L) fanout(event) diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerIngestTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerIngestTest.kt new file mode 100644 index 0000000000..085edb3098 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerIngestTest.kt @@ -0,0 +1,162 @@ +/* + * 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.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy +import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore +import com.vitorpamplona.quartz.utils.DeterministicSigner +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * The local-ingest path ([NostrServer.ingest]) and its relay-to-relay + * trust switch: `skipVerify = true` (a mirror streaming from a trusted + * upstream) must land forged-signature events, while the default keeps + * verify-everything semantics through the IngestQueue's parallel hook. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class NostrServerIngestTest { + private val signer = DeterministicSigner(KeyPair()) + + private fun signedEvent(content: String = "hello"): Event = signer.sign(createdAt = 1000L, kind = 1, tags = emptyArray(), content = content) + + /** A structurally valid event whose signature verifies against nothing. */ + private fun forgedEvent(id: Int = 1): Event = + Event( + id = id.toString().padStart(64, '0'), + pubKey = signer.pubKey, + createdAt = 1000L + id, + kind = 1, + tags = emptyArray(), + content = "forged", + sig = "f".repeat(128), + ) + + private fun createServer(dispatcher: CoroutineDispatcher): NostrServer = + NostrServer( + store = EventStore(null), + policyBuilder = { EmptyPolicy }, + parentContext = dispatcher, + parallelVerify = true, + ) + + private suspend fun NostrServer.ingestOutcome( + event: Event, + skipVerify: Boolean, + ): IEventStore.InsertOutcome { + val outcome = CompletableDeferred() + ingest(event, skipVerify) { outcome.complete(it) } + return outcome.await() + } + + @Test + fun forgedEventIsRejectedByDefault() = + runTest { + val server = createServer(UnconfinedTestDispatcher(testScheduler)) + + val outcome = server.ingestOutcome(forgedEvent(), skipVerify = false) + + assertTrue(outcome is IEventStore.InsertOutcome.Rejected) + assertTrue(outcome.reason.contains("signature")) + + server.close() + } + + @Test + fun forgedEventLandsWhenTrusted() = + runTest { + val server = createServer(UnconfinedTestDispatcher(testScheduler)) + + val outcome = server.ingestOutcome(forgedEvent(), skipVerify = true) + + assertEquals(IEventStore.InsertOutcome.Accepted, outcome) + + server.close() + } + + @Test + fun validEventLandsEitherWay() = + runTest { + val server = createServer(UnconfinedTestDispatcher(testScheduler)) + + assertEquals( + IEventStore.InsertOutcome.Accepted, + server.ingestOutcome(signedEvent("via verify"), skipVerify = false), + ) + assertEquals( + IEventStore.InsertOutcome.Accepted, + server.ingestOutcome(signedEvent("via trust"), skipVerify = true), + ) + + server.close() + } + + @Test + fun trustIsPerSubmissionNotPerQueue() = + runTest { + // A trusted mirror and an untrusted publisher share the same + // IngestQueue; the skip must apply row-by-row, never leak from + // one submission to the next. + val server = createServer(UnconfinedTestDispatcher(testScheduler)) + + assertEquals( + IEventStore.InsertOutcome.Accepted, + server.ingestOutcome(forgedEvent(1), skipVerify = true), + ) + val untrusted = server.ingestOutcome(forgedEvent(2), skipVerify = false) + assertTrue(untrusted is IEventStore.InsertOutcome.Rejected) + + server.close() + } + + @Test + fun trustedIngestFansOutToLiveSubscribers() = + runTest { + // Mirror traffic must feed live REQs exactly like a client + // publish — the skip changes verification, not delivery. + val dispatcher = UnconfinedTestDispatcher(testScheduler) + val server = createServer(dispatcher) + + val received = mutableListOf() + val session = server.connect { received.add(it) } + session.receive("""["REQ","live",{"kinds":[1]}]""") + assertTrue(received.any { it.startsWith("[\"EOSE\"") }) + + val forged = forgedEvent() + assertEquals( + IEventStore.InsertOutcome.Accepted, + server.ingestOutcome(forged, skipVerify = true), + ) + + assertTrue(received.any { it.startsWith("[\"EVENT\",\"live\"") && it.contains(forged.id) }) + + server.close() + } +}