diff --git a/geode/config.example.toml b/geode/config.example.toml index 17f15c8d31..1e5d9df79e 100644 --- a/geode/config.example.toml +++ b/geode/config.example.toml @@ -129,8 +129,14 @@ require_auth = false # everything; for several disjoint scopes, repeat [[mirror]] with the # same url. # +# `dir` (strfry-router parity) sets the flow direction: "down" pulls +# from the upstream (default), "up" pushes this relay's matching +# events to it, "both" does both with echo suppression so the two +# directions don't ping-pong the same event. +# # [[mirror]] # url = "wss://upstream.example.com/" +# dir = "both" # trusted = true # backfill_seconds = 3600 # filter = '{"kinds":[0,1,3,7],"#t":["nostr"]}' diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 4a7b1e6179..8bb7895249 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -24,6 +24,7 @@ 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.MirrorDirection import com.vitorpamplona.geode.mirror.MirrorUpstream import com.vitorpamplona.geode.mirror.MirrorWorker import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper @@ -220,11 +221,17 @@ fun main(args: Array) { } parsed.copy(since = null, limit = null) } + val direction = + MirrorDirection.parse(m.dir) + ?: throw IllegalArgumentException( + "[[mirror]] dir for ${m.url} must be \"down\", \"up\" or \"both\" (got \"${m.dir}\")", + ) MirrorUpstream( url = m.url.normalizeRelayUrl(), trusted = m.trusted, backfillSeconds = m.backfill_seconds, filter = scope, + direction = direction, ) } require(upstreams.none { it.url == advertisedUrl }) { diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt index 7145139f57..c1559f6726 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt @@ -223,6 +223,13 @@ data class StaticConfig( * several `[[mirror]]` entries with the same url. */ val filter: String? = null, + /** + * Flow direction, strfry-router's `dir`: `"down"` (pull from the + * upstream — the default), `"up"` (push this relay's matching + * events to it), or `"both"`. Both-way mirrors suppress echoes + * (an event pulled down is not pushed straight back). + */ + val dir: String = "down", ) /** diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt index f87737d89c..60643a24a7 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt @@ -21,8 +21,11 @@ package com.vitorpamplona.geode.mirror import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer @@ -44,6 +47,33 @@ import okhttp3.OkHttpClient import java.time.Duration import java.util.concurrent.atomic.AtomicLong +/** + * Which way events flow between this relay and one upstream — + * strfry-router's per-stream `dir`. + */ +enum class MirrorDirection { + /** Pull: subscribe to the upstream and ingest what it sends. */ + DOWN, + + /** Push: publish this relay's matching events to the upstream. */ + UP, + + /** Pull and push. Echo suppression keeps the two from ping-ponging. */ + BOTH, + ; + + companion object { + /** Parses strfry's `"down"` / `"up"` / `"both"`; null if unknown. */ + fun parse(value: String): MirrorDirection? = + when (value.lowercase()) { + "down" -> DOWN + "up" -> UP + "both" -> BOTH + else -> null + } + } +} + /** * One upstream relay this relay mirrors, from the `[[mirror]]` config. * @@ -51,23 +81,28 @@ import java.util.concurrent.atomic.AtomicLong * 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. + * hijacked by an inbound client. Only meaningful for [MirrorDirection.DOWN] + * / [MirrorDirection.BOTH]; the up direction never verifies (the upstream + * does its own gatekeeping). */ class MirrorUpstream( val url: NormalizedRelayUrl, val trusted: Boolean, - /** How far back the initial REQ reaches. 0 = live-only from connect. */ + /** How far back the initial replay reaches, in BOTH directions. 0 = live-only from connect. */ val backfillSeconds: Long = 0L, /** * Optional scope for this upstream (strfry-router's per-stream - * `filter`). Used twice: it shapes the REQ sent upstream, and every - * delivered event is re-checked against it before ingest — so even a - * [trusted] upstream can only inject events inside the declared - * scope. Its `since`/`limit` are ignored ([backfillSeconds] owns the - * time window; the subscription is unbounded). `null` mirrors + * `filter`). Applied symmetrically: down, it shapes the REQ sent + * upstream AND every delivered event is re-checked before ingest — + * so even a [trusted] upstream can only inject events inside the + * declared scope; up, it selects which local events are pushed. Its + * `since`/`limit` are ignored ([backfillSeconds] owns the time + * window; the subscriptions are unbounded). `null` mirrors * everything. */ val filter: Filter? = null, + /** Flow direction — strfry-router's `dir`. Defaults to pull-only. */ + val direction: MirrorDirection = MirrorDirection.DOWN, ) /** @@ -151,6 +186,38 @@ class MirrorWorker( */ val filtered = AtomicLong(0) + /** Local events handed to the client's outbox for an up-direction upstream. */ + val sentUp = AtomicLong(0) + + /** + * Recently exchanged event ids, one set per up-capable upstream — + * the echo suppressor for [MirrorDirection.BOTH]. An event pulled + * DOWN from an upstream must not be pushed straight back UP to it + * (and one we pushed up must not be re-ingested when the upstream + * fans it back on our down subscription). Bounded LRU: eviction only + * costs a wasted round trip that the stores' unique-id constraints + * absorb, so correctness never depends on it. + */ + private class RecentIds( + private val capacity: Int, + ) { + private val map = + object : LinkedHashMap(capacity, 0.75f, true) { + override fun removeEldestEntry(eldest: Map.Entry) = size > capacity + } + + @Synchronized + fun add(id: String) { + map[id] = true + } + + @Synchronized + fun contains(id: String): Boolean = map.containsKey(id) + } + + /** Open in-process sessions feeding the up direction; closed with the worker. */ + private val upSessions = mutableListOf() + /** Dials every upstream and starts streaming. Call once. */ fun start() { scope.launch { @@ -181,48 +248,25 @@ class MirrorWorker( val since = TimeUtils.now() upstreams.forEachIndexed { i, up -> - val listener = - object : SubscriptionListener { - override fun onEvent( - event: Event, - isLive: Boolean, - relay: NormalizedRelayUrl, - forFilters: List?, - ) { - // strfry-router parity: never take the upstream's - // word for what matched. Re-checking the configured - // scope here means even a trusted (skip-verify) - // upstream can only inject events the operator - // declared — the REQ shapes what we ask for, this - // shapes what we accept. - if (up.filter != null && !up.filter.match(event)) { - filtered.incrementAndGet() - Log.d("MirrorWorker") { "out-of-scope from ${relay.url}: ${event.id}" } - return - } - inbound.trySend(Inbound(event, up.trusted)) - } + // Echo suppression only matters when events can flow both + // ways on the same upstream. + val exchanged = if (up.direction == MirrorDirection.BOTH) RecentIds(EXCHANGED_IDS_CAPACITY) else null - override fun onCannotConnect( - relay: NormalizedRelayUrl, - message: String, - forFilters: List?, - ) { - Log.w("MirrorWorker") { "cannot reach upstream ${relay.url}: $message" } - } - } - // The operator's filter scopes the REQ; the mirror owns the - // time window (since) and never bounds the result (limit). - val reqFilter = + // The operator's filter scopes the subscriptions in both + // directions; the mirror owns the time window (since) and + // never bounds the result (limit). + val scopedFilter = (up.filter ?: Filter()).copy( since = since - up.backfillSeconds, limit = null, ) - client.subscribe( - subId = "geode-mirror-$i", - filters = mapOf(up.url to listOf(reqFilter)), - listener = listener, - ) + + if (up.direction != MirrorDirection.UP) { + startDown(i, up, scopedFilter, exchanged) + } + if (up.direction != MirrorDirection.DOWN) { + startUp(up, scopedFilter, exchanged) + } } client.connect() @@ -242,6 +286,87 @@ class MirrorWorker( } } + /** Down direction: subscribe to the upstream, ingest what it sends. */ + private fun startDown( + index: Int, + up: MirrorUpstream, + scopedFilter: Filter, + exchanged: RecentIds?, + ) { + val listener = + object : SubscriptionListener { + override fun onEvent( + event: Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + // strfry-router parity: never take the upstream's + // word for what matched. Re-checking the configured + // scope here means even a trusted (skip-verify) + // upstream can only inject events the operator + // declared — the REQ shapes what we ask for, this + // shapes what we accept. + if (up.filter != null && !up.filter.match(event)) { + filtered.incrementAndGet() + Log.d("MirrorWorker") { "out-of-scope from ${relay.url}: ${event.id}" } + return + } + // BOTH: an event we just pushed up is fanned back on + // this subscription — it already exists locally. + if (exchanged?.contains(event.id) == true) return + exchanged?.add(event.id) + inbound.trySend(Inbound(event, up.trusted)) + } + + override fun onCannotConnect( + relay: NormalizedRelayUrl, + message: String, + forFilters: List?, + ) { + Log.w("MirrorWorker") { "cannot reach upstream ${relay.url}: $message" } + } + } + client.subscribe( + subId = "geode-mirror-$index", + filters = mapOf(up.url to listOf(scopedFilter)), + listener = listener, + ) + } + + /** + * Up direction: an in-process session on the LOCAL relay subscribes + * with the same scoped filter — stored replay covers the backfill + * window, the live tail covers everything after — and each matching + * event is handed to the client's outbox for [MirrorUpstream.url]. + * The outbox owns delivery: it re-sends on reconnect until the + * upstream OKs, and the upstream's own duplicate handling absorbs + * replays. Going through a real session (not the store) means the + * relay's policy chain gates what leaves, same as any client. + */ + private fun startUp( + up: MirrorUpstream, + scopedFilter: Filter, + exchanged: RecentIds?, + ) { + val session = + server.connect { json -> + if (!json.startsWith("[\"EVENT\"")) return@connect + val event = + runCatching { (OptimizedJsonMapper.fromJsonToMessage(json) as? EventMessage)?.event } + .getOrNull() ?: return@connect + // BOTH: don't push back what we just pulled down. + if (exchanged?.contains(event.id) == true) return@connect + exchanged?.add(event.id) + client.publish(event, setOf(up.url)) + sentUp.incrementAndGet() + } + upSessions += AutoCloseable { session.close() } + scope.launch { + session.receive(OptimizedJsonMapper.toJson(ReqCmd("geode-mirror-up", listOf(scopedFilter)))) + } + } + /** * Stops pulling from every upstream. In-flight ingest submissions * drain through the server's queue; events still buffered in @@ -252,6 +377,7 @@ class MirrorWorker( override fun close() { // Close the client first so no listener callback races the // channel close below. + upSessions.forEach { runCatching { it.close() } } runCatching { client.close() } inbound.close() scope.cancel() @@ -265,5 +391,13 @@ class MirrorWorker( /** How often the retry pump nudges disconnected upstreams. */ const val RECONNECT_POKE_MS = 5_000L + + /** + * Per-upstream echo-suppression LRU size (BOTH direction only). + * Covers the burst window between pulling an event down and the + * up-session seeing its local fanout; eviction only costs a + * duplicate round trip. + */ + const val EXCHANGED_IDS_CAPACITY = 8_192 } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt index e2f62e2531..49bcb87785 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt @@ -80,9 +80,19 @@ class MirrorWorkerTest { private fun startMirror( trusted: Boolean, filter: Filter? = null, + direction: MirrorDirection = MirrorDirection.DOWN, ): MirrorWorker = MirrorWorker( - upstreams = listOf(MirrorUpstream(upstreamUrl, trusted = trusted, backfillSeconds = 3600, filter = filter)), + upstreams = + listOf( + MirrorUpstream( + upstreamUrl, + trusted = trusted, + backfillSeconds = 3600, + filter = filter, + direction = direction, + ), + ), server = downstream.server, websocketBuilder = hub, ).also { @@ -173,6 +183,63 @@ class MirrorWorkerTest { assertTrue(stored.all { it.kind == 1 }) } + @Test + fun upDirectionPushesLocalEventsToTheUpstream() = + runBlocking { + // Pre-existing local event: the up replay (backfill window) + // must carry it; then a live local publish must follow. + val preexisting = forgedEvent(1) + downstream.preload(preexisting) + + startMirror(trusted = false, direction = MirrorDirection.UP) + + val upstreamStore = hub.getOrCreate(upstreamUrl).store + await { upstreamStore.count(Filter()) >= 1 } + + // Live events go through the verifying publish path, so they + // must be genuinely signed (preload bypasses verification). + val live = signedEvent("up live") + downstream.publish(live) + await { upstreamStore.count(Filter()) >= 2 } + + val ids = upstreamStore.query(Filter()).map { it.id }.toSet() + assertEquals(setOf(preexisting.id, live.id), ids) + } + + @Test + fun bothDirectionConvergesWithoutPingPong() = + runBlocking { + // One event only the upstream has, one only the local relay + // has. BOTH must converge the two stores; echo suppression + // (plus store dedup as the backstop) must keep the shared + // events from bouncing. + val upstreamOnly = forgedEvent(1) + val localOnly = forgedEvent(2) + hub.getOrCreate(upstreamUrl).preload(upstreamOnly) + downstream.preload(localOnly) + + val mirror = startMirror(trusted = true, direction = MirrorDirection.BOTH) + + val upstreamStore = hub.getOrCreate(upstreamUrl).store + await { downstreamStore.count(Filter()) == 2 && upstreamStore.count(Filter()) == 2 } + + val expected = setOf(upstreamOnly.id, localOnly.id) + assertEquals(expected, downstreamStore.query(Filter()).map { it.id }.toSet()) + assertEquals(expected, upstreamStore.query(Filter()).map { it.id }.toSet()) + + // Live: an event published locally reaches the upstream AND + // its echo back down doesn't disturb either store. Signed, + // because the local publish path verifies. + val live = signedEvent("both live") + downstream.publish(live) + await { upstreamStore.count(Filter()) == 3 } + // Let any echo settle, then confirm counts are exact. + delay(500) + assertEquals(3, downstreamStore.count(Filter())) + assertEquals(3, upstreamStore.count(Filter())) + assertTrue(mirror.sentUp.get() >= 2) + } + @Test fun untrustedUpstreamStillVerifiesEverything() = runBlocking {