feat(geode): [[mirror]] upstream streaming with relay-to-relay trust

Backlog item 1 of the relay performance campaign: skip signature
verification for events ingested from explicitly configured trusted
upstream relays, strfry-router style.

geode now dials each [[mirror]] url from the config, subscribes to
everything newer than now - backfill_seconds, and feeds the stream
through NostrServer.ingest (same group-commit writer + live fanout as
client publishes). The NostrClient underneath owns reconnects, backoff
and REQ re-sync; duplicate replays after a reconnect are rejected by
the store's unique-id constraint and only surface as counters.

trusted = true is the per-upstream trust switch: events from that
connection skip Schnorr verify. The trusted identity is the URL this
relay dialed (TLS-authenticated for wss://), never anything an inbound
peer claims. Default is false — mirror-but-verify — and a relay with no
[[mirror]] entries behaves exactly as before.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A

geode/build.gradle.kts

@@ -85,6 +85,11 @@ dependencies {
     implementation(libs.jackson.module.kotlin)
     implementation(libs.kotlinx.serialization.json)
 
+    // Outbound WebSockets for the [[mirror]] upstream streams (quartz's
+    // BasicOkHttpWebSocket transport). Same OkHttp the rest of the repo
+    // already ships (Apache-2.0).
+    implementation(libs.okhttp)
+
     // Bundled SQLite driver — Relay's default in-memory EventStore creates
     // an in-memory DB at runtime.
     implementation(libs.androidx.sqlite.bundled.jvm)

geode/config.example.toml

@@ -86,6 +86,24 @@ require_auth = false
 # kind_whitelist = [0, 1, 3, 7, 1059, 30023]
 # kind_blacklist = [4]
 
+# Mirror upstream relays (strfry-router style, "down" direction): the
+# relay dials each [[mirror]] url, subscribes to everything newer than
+# now - backfill_seconds, and ingests the stream alongside client
+# publishes. Reconnects and re-subscribes automatically.
+#
+# `trusted = true` is the relay-to-relay trust switch: events from that
+# upstream skip Schnorr signature verification (the upstream already
+# verified its own ingest; re-verifying burns ~8% of ingest CPU). The
+# trusted identity is the URL *this* relay dialed — TLS-authenticated
+# for wss:// — so an inbound client can never claim it. Default false:
+# mirror-but-verify. Only trust relays you operate or whose ingest
+# discipline you'd stake your own db on.
+#
+# [[mirror]]
+# url = "wss://upstream.example.com/"
+# trusted = true
+# backfill_seconds = 3600
+
 [admin]
 # NIP-86 relay management API. When `pubkeys` is non-empty, the relay
 # accepts HTTP POST application/nostr+json+rpc on the same URL,

geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt

@@ -24,6 +24,8 @@ import com.vitorpamplona.geode.config.BannedEntry
 import com.vitorpamplona.geode.config.RuntimeConfig
 import com.vitorpamplona.geode.config.RuntimeConfigData
 import com.vitorpamplona.geode.config.StaticConfig
+import com.vitorpamplona.geode.mirror.MirrorUpstream
+import com.vitorpamplona.geode.mirror.MirrorWorker
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
@@ -172,10 +174,37 @@ fun main(args: Array<String>) {
             callGroupSize = config.network.call_group_size,
         ).start()
 
+    // `[[mirror]]` upstreams: dial each configured relay and stream its
+    // events into the local store. `trusted = true` entries skip Schnorr
+    // verification for that connection (relay-to-relay trust) — only
+    // meaningful while verify is on; with --no-verify nothing verifies
+    // anyway. Never mirror ourselves: a self-URL would echo every local
+    // publish back forever.
+    val upstreams =
+        config.mirror.map {
+            MirrorUpstream(
+                url = it.url.normalizeRelayUrl(),
+                trusted = it.trusted,
+                backfillSeconds = it.backfill_seconds,
+            )
+        }
+    require(upstreams.none { it.url == advertisedUrl }) {
+        "[[mirror]] must not list this relay's own URL ($advertisedUrl)"
+    }
+    val mirror =
+        if (upstreams.isEmpty()) {
+            null
+        } else {
+            MirrorWorker(upstreams, relay.server).also { it.start() }
+        }
+
     Runtime.getRuntime().addShutdownHook(
         Thread {
-            // Each step wrapped so a throw in `server.stop()` doesn't
-            // skip `relay.close()` (which closes the SQLite store).
+            // Each step wrapped so a throw in any stage doesn't skip
+            // `relay.close()` (which closes the SQLite store). The mirror
+            // goes first: stop pulling new events before the queue and
+            // store beneath it shut down.
+            runCatching { mirror?.close() }
             runCatching { server.stop() }
             runCatching { relay.close() }
         },
@@ -183,6 +212,10 @@ fun main(args: Array<String>) {
 
     println("geode listening on ${server.url}")
     println("NIP-11 info doc: curl -H 'Accept: application/nostr+json' http://$advertisedHost:$port$path")
+    if (upstreams.isNotEmpty()) {
+        val trusted = upstreams.count { it.trusted }
+        println("mirroring ${upstreams.size} upstream relay(s), $trusted trusted (signature verification skipped)")
+    }
 
     // Park the main thread; shutdown hook handles teardown.
     Thread.currentThread().join()

geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt

@@ -45,6 +45,8 @@ data class StaticConfig(
     val authorization: AuthorizationSection = AuthorizationSection(),
     val admin: AdminSection = AdminSection(),
     val negentropy: NegentropySection = NegentropySection(),
+    /** `[[mirror]]` entries — upstream relays this relay streams from. */
+    val mirror: List<MirrorSection> = emptyList(),
 ) {
     fun resolveInfo(fullTextSearch: Boolean = true): RelayInfo =
         RelayInfo(
@@ -160,6 +162,27 @@ data class StaticConfig(
         val kind_blacklist: List<Int> = emptyList(),
     )
 
+    /**
+     * One upstream relay to mirror, declared as a `[[mirror]]` TOML array
+     * entry. The relay dials [url] itself, subscribes to everything newer
+     * than `now - backfill_seconds`, and ingests the stream through the
+     * same group-commit writer as client publishes (reconnects and
+     * re-subscribes automatically).
+     *
+     * [trusted] is the relay-to-relay trust switch (strfry's model):
+     * `true` skips Schnorr signature verification for events arriving on
+     * this connection — sound only when the upstream verifies its own
+     * ingest, which is why it defaults to `false` (mirror-but-verify).
+     * The identity being trusted is the URL this relay dialed (TLS-
+     * authenticated for `wss://`), never anything a peer claims.
+     */
+    data class MirrorSection(
+        val url: String,
+        val trusted: Boolean = false,
+        /** How far back the initial subscription reaches. 0 = live-only. */
+        val backfill_seconds: Long = 0L,
+    )
+
     /**
      * NIP-86 admin. [pubkeys] non-empty opens the POST endpoint at the
      * relay path; only NIP-98 tokens signed by these pubkeys dispatch.

geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt

@@ -0,0 +1,190 @@
+/*
+ * 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.geode.mirror
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import com.vitorpamplona.quartz.utils.Log
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.launch
+import okhttp3.OkHttpClient
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * One upstream relay this relay mirrors, from the `[[mirror]]` config.
+ *
+ * [trusted] is the relay-to-relay trust switch: events streamed from this
+ * upstream skip Schnorr signature verification on ingest. The trusted
+ * identity is [url] — the address *this* relay dialed (TLS-authenticated
+ * for `wss://`), never anything the peer claims — so the skip can't be
+ * hijacked by an inbound client.
+ */
+class MirrorUpstream(
+    val url: NormalizedRelayUrl,
+    val trusted: Boolean,
+    /** How far back the initial REQ reaches. 0 = live-only from connect. */
+    val backfillSeconds: Long = 0L,
+)
+
+/**
+ * Streams events from configured upstream relays into the local relay —
+ * geode's equivalent of `strfry router` in the "down" direction.
+ *
+ * One [NostrClient] holds every upstream connection; the client owns
+ * reconnects, exponential backoff, and re-sending the REQ after a drop.
+ * Each upstream gets its own subscription over an open filter
+ * (`since = now - backfill`), and every EVENT that arrives is handed to
+ * [NostrServer.ingest] — the same group-commit writer and live fanout a
+ * client publish takes — with `skipVerify` set for [MirrorUpstream.trusted]
+ * upstreams (the upstream already verified its ingest; re-verifying here
+ * only burns CPU — Schnorr verify profiles at ~8% of busy ingest CPU).
+ *
+ * After a reconnect the upstream replays everything since the boot-time
+ * `since`; replayed duplicates are rejected by the store's unique id
+ * constraint and only show up in [rejected].
+ *
+ * Listener callbacks can't suspend, so events funnel through an unbounded
+ * [inbound] channel into one consumer coroutine whose [NostrServer.ingest]
+ * call suspends on the ingest queue's backpressure. The buffer is unbounded
+ * for the same reason the client's receive channels are (see
+ * `BasicOkHttpWebSocket`): blocking the socket reader parks the backlog on
+ * infrastructure that isn't ours.
+ */
+class MirrorWorker(
+    private val upstreams: List<MirrorUpstream>,
+    private val server: NostrServer,
+    /**
+     * Transport override for tests (e.g. `InProcessRelays`). Defaults to
+     * a real OkHttp WebSocket per upstream.
+     */
+    websocketBuilder: WebsocketBuilder? = null,
+) : AutoCloseable {
+    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
+
+    private val okhttp: OkHttpClient? = if (websocketBuilder == null) OkHttpClient.Builder().build() else null
+
+    private val client =
+        NostrClient(
+            websocketBuilder = websocketBuilder ?: BasicOkHttpWebSocket.Builder { okhttp!! },
+            parentScope = scope,
+        )
+
+    private class Inbound(
+        val event: Event,
+        val skipVerify: Boolean,
+    )
+
+    private val inbound = Channel<Inbound>(Channel.UNLIMITED)
+
+    /** Events accepted into the local store (excludes duplicates). */
+    val accepted = AtomicLong(0)
+
+    /** Events the store rejected — mostly duplicate replays after a reconnect. */
+    val rejected = AtomicLong(0)
+
+    /** Dials every upstream and starts streaming. Call once. */
+    fun start() {
+        scope.launch {
+            for (msg in inbound) {
+                try {
+                    server.ingest(msg.event, msg.skipVerify) { outcome ->
+                        when (outcome) {
+                            IEventStore.InsertOutcome.Accepted -> accepted.incrementAndGet()
+                            is IEventStore.InsertOutcome.Rejected -> {
+                                rejected.incrementAndGet()
+                                Log.d("MirrorWorker") { "rejected ${msg.event.id}: ${outcome.reason}" }
+                            }
+                        }
+                    }
+                } catch (e: CancellationException) {
+                    throw e
+                } catch (e: Throwable) {
+                    // A server shutting down closes its ingest queue while
+                    // events may still be buffered here; an uncaught throw
+                    // would leak into the scope's default handler (and
+                    // poison unrelated runTest tests on CI). Stop pulling —
+                    // the relay beneath us is going away.
+                    Log.w("MirrorWorker") { "ingest failed, stopping mirror consumer: ${e.message}" }
+                    break
+                }
+            }
+        }
+
+        val since = TimeUtils.now()
+        upstreams.forEachIndexed { i, up ->
+            val listener =
+                object : SubscriptionListener {
+                    override fun onEvent(
+                        event: Event,
+                        isLive: Boolean,
+                        relay: NormalizedRelayUrl,
+                        forFilters: List<Filter>?,
+                    ) {
+                        inbound.trySend(Inbound(event, up.trusted))
+                    }
+
+                    override fun onCannotConnect(
+                        relay: NormalizedRelayUrl,
+                        message: String,
+                        forFilters: List<Filter>?,
+                    ) {
+                        Log.w("MirrorWorker") { "cannot reach upstream ${relay.url}: $message" }
+                    }
+                }
+            client.subscribe(
+                subId = "geode-mirror-$i",
+                filters = mapOf(up.url to listOf(Filter(since = since - up.backfillSeconds))),
+                listener = listener,
+            )
+        }
+        client.connect()
+    }
+
+    /**
+     * Stops pulling from every upstream. In-flight ingest submissions
+     * drain through the server's queue; events still buffered in
+     * [inbound] are dropped — the next boot's `since` overlaps only if
+     * the operator configured a backfill window, which is the documented
+     * trade-off of a live mirror.
+     */
+    override fun close() {
+        // Close the client first so no listener callback races the
+        // channel close below.
+        runCatching { client.close() }
+        inbound.close()
+        scope.cancel()
+        okhttp?.dispatcher?.executorService?.shutdown()
+        okhttp?.connectionPool?.evictAll()
+    }
+}

geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt

@@ -46,6 +46,36 @@ class StaticConfigTest {
         assertEquals(false, c.options.verify_signatures)
     }
 
+    @Test
+    fun mirrorSectionDefaultsToEmpty() {
+        assertTrue(StaticConfig.fromToml("").mirror.isEmpty())
+    }
+
+    @Test
+    fun parsesMirrorUpstreams() {
+        val toml =
+            """
+            [[mirror]]
+            url = "wss://trusted.upstream.example/"
+            trusted = true
+            backfill_seconds = 3600
+
+            [[mirror]]
+            url = "wss://public.upstream.example/"
+            """.trimIndent()
+
+        val c = StaticConfig.fromToml(toml)
+
+        assertEquals(2, c.mirror.size)
+        assertEquals("wss://trusted.upstream.example/", c.mirror[0].url)
+        assertEquals(true, c.mirror[0].trusted)
+        assertEquals(3600L, c.mirror[0].backfill_seconds)
+        // Trust is opt-in per upstream: the default is mirror-but-verify.
+        assertEquals("wss://public.upstream.example/", c.mirror[1].url)
+        assertEquals(false, c.mirror[1].trusted)
+        assertEquals(0L, c.mirror[1].backfill_seconds)
+    }
+
     @Test
     fun parsesAllSectionsTogether() {
         val toml =

geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt

@@ -0,0 +1,165 @@
+/*
+ * 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.geode.mirror
+
+import com.vitorpamplona.geode.InProcessRelays
+import com.vitorpamplona.geode.RelayEngine
+import com.vitorpamplona.geode.testing.preload
+import com.vitorpamplona.geode.testing.publish
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.toHexKey
+import com.vitorpamplona.quartz.nip01Core.crypto.EventAssembler
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.junit.After
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * End-to-end `[[mirror]]` behavior over the in-process transport: a
+ * downstream relay (verification ON via the parallel-verify queue) dials
+ * an upstream and streams its events.
+ *
+ *  - `trusted = true` — the relay-to-relay trust switch — must land
+ *    events the downstream could never verify itself.
+ *  - `trusted = false` must keep verify-everything semantics: forged
+ *    events from the upstream are dropped, valid ones land.
+ */
+class MirrorWorkerTest {
+    private val upstreamUrl = RelayUrlNormalizer.normalize("ws://upstream.relay/")
+    private val downstreamUrl = RelayUrlNormalizer.normalize("ws://downstream.relay/")
+
+    /** Upstream side: no verification (EmptyPolicy), so forged events store fine. */
+    private val hub = InProcessRelays()
+
+    /** Downstream side: signature verification on, in the IngestQueue. */
+    private val downstreamStore = EventStore(null)
+    private val downstream =
+        RelayEngine(
+            url = downstreamUrl,
+            store = downstreamStore,
+            parallelVerify = true,
+        )
+
+    private var worker: MirrorWorker? = null
+
+    private val signer = KeyPair()
+
+    @After
+    fun tearDown() {
+        worker?.close()
+        downstream.close()
+        hub.close()
+    }
+
+    private fun startMirror(trusted: Boolean): MirrorWorker =
+        MirrorWorker(
+            upstreams = listOf(MirrorUpstream(upstreamUrl, trusted = trusted, backfillSeconds = 3600)),
+            server = downstream.server,
+            websocketBuilder = hub,
+        ).also {
+            worker = it
+            it.start()
+        }
+
+    private fun forgedEvent(idSeed: Int): Event =
+        Event(
+            id = idSeed.toString().padStart(64, '0'),
+            pubKey = "1".repeat(64),
+            createdAt = TimeUtils.now() - idSeed,
+            kind = 1,
+            tags = emptyArray(),
+            content = "forged $idSeed",
+            sig = "f".repeat(128),
+        )
+
+    private fun signedEvent(content: String): Event =
+        EventAssembler.hashAndSign(
+            pubKey = signer.pubKey.toHexKey(),
+            createdAt = TimeUtils.now() - 5,
+            kind = 1,
+            tags = emptyArray(),
+            content = content,
+            privKey = signer.privKey!!,
+        )
+
+    private suspend fun awaitDownstreamCount(expected: Int) =
+        withTimeout(15_000) {
+            while (downstreamStore.count(Filter()) < expected) delay(25)
+        }
+
+    private suspend fun await(condition: suspend () -> Boolean) =
+        withTimeout(15_000) {
+            while (!condition()) delay(25)
+        }
+
+    // NOTE: assertions are on the downstream STORE, not on exact counter
+    // values — the client may legitimately re-send its REQ while settling
+    // (connect + filter sync), so an upstream can replay an event twice and
+    // the duplicate shows up as one extra rejection. That's the documented
+    // mirror behavior, not a failure.
+
+    @Test
+    fun trustedUpstreamLandsEventsTheDownstreamCannotVerify() =
+        runBlocking {
+            val stored = forgedEvent(1)
+            val live = forgedEvent(2)
+
+            // Stored replay: exists on the upstream before the mirror dials.
+            hub.getOrCreate(upstreamUrl).preload(stored)
+
+            startMirror(trusted = true)
+            awaitDownstreamCount(1)
+
+            // Live tail: published upstream after the mirror subscribed.
+            hub.getOrCreate(upstreamUrl).publish(live)
+            awaitDownstreamCount(2)
+
+            val ids = downstreamStore.query<Event>(Filter()).map { it.id }.toSet()
+            assertEquals(setOf(stored.id, live.id), ids)
+        }
+
+    @Test
+    fun untrustedUpstreamStillVerifiesEverything() =
+        runBlocking {
+            val forged = forgedEvent(1)
+            val valid = signedEvent("the real one")
+            hub.getOrCreate(upstreamUrl).preload(forged, valid)
+
+            val mirror = startMirror(trusted = false)
+
+            // The valid event lands; the forged one is verified and dropped
+            // (a forged event can never land untrusted, so once both have
+            // been processed the store can only hold the valid one).
+            await { mirror.rejected.get() >= 1 && downstreamStore.count(Filter()) == 1 }
+
+            val stored = downstreamStore.query<Event>(Filter()).single()
+            assertEquals(valid.id, stored.id)
+            assertTrue(downstreamStore.query<Event>(Filter(ids = listOf(forged.id))).isEmpty())
+        }
+}