feat(geode): mirror directions — dir = "down" | "up" | "both" (strfry-router parity)

Each [[mirror]] entry now takes strfry-router's dir: - down (default): pull — subscribe to the upstream and ingest, exactly as before. - up: push — an in-process session on the LOCAL relay subscribes with the same scoped filter (so backfill_seconds and filter behave identically in both directions, and the relay's own policy chain gates what leaves), and every matching event is handed to the client's outbox for the upstream, which owns delivery and re-sends across reconnects. - both: pull and push, with echo suppression: a per-upstream LRU of recently exchanged ids keeps an event pulled down from being pushed straight back (and vice versa when the upstream fans our own publish back). Eviction only costs a duplicate round trip — the stores' unique-id constraints stay the correctness backstop. trusted (verify skip) remains a down-only concept; the up direction never verifies since the upstream does its own gatekeeping. Tests: up pushes both the backfill window and the live tail; both converges two stores with disjoint content and holds exact counts after the echo settles (no ping-pong). Live-published test events are genuinely signed — the local publish path verifies, which is also what the debugging showed: the mechanism was fine, the first version of the tests was pushing forged events into a verifying relay. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A
2026-08-09 08:04:45 +00:00 · 2026-07-04 01:32:18 +00:00
parent cc73639ae5
commit 3d23f563b1
5 changed files with 266 additions and 45 deletions
@@ -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"]}'
@@ -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<String>) {
                        }
                    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 }) {
@@ -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",
    )

    /**
@@ -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<String, Boolean>(capacity, 0.75f, true) {
+                override fun removeEldestEntry(eldest: Map.Entry<String, Boolean>) = 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<AutoCloseable>()
+
    /** 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<Filter>?,
-                    ) {
-                        // 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<Filter>?,
-                    ) {
-                        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<Filter>?,
+                ) {
+                    // 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<Filter>?,
+                ) {
+                    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
    }
 }
@@ -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<Event>(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<Event>(Filter()).map { it.id }.toSet())
+            assertEquals(expected, upstreamStore.query<Event>(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 {