From 50e259b495c5ac12bf9a74be26cb496ce9708036 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Jul 2026 21:22:46 +0000 Subject: [PATCH] feat(geode): [[mirror]] upstream streaming with relay-to-relay trust MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A --- geode/build.gradle.kts | 5 + geode/config.example.toml | 18 ++ .../kotlin/com/vitorpamplona/geode/Main.kt | 37 +++- .../geode/config/StaticConfig.kt | 23 +++ .../geode/mirror/MirrorWorker.kt | 190 ++++++++++++++++++ .../geode/config/StaticConfigTest.kt | 30 +++ .../geode/mirror/MirrorWorkerTest.kt | 165 +++++++++++++++ 7 files changed, 466 insertions(+), 2 deletions(-) create mode 100644 geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt create mode 100644 geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt diff --git a/geode/build.gradle.kts b/geode/build.gradle.kts index bc41967f3d..981d2a0abc 100644 --- a/geode/build.gradle.kts +++ b/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) diff --git a/geode/config.example.toml b/geode/config.example.toml index 73f7adca1c..98ad0fc2d4 100644 --- a/geode/config.example.toml +++ b/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, diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 963269af0d..89dc1c989f 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/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) { 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) { 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() 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 b9c9f6aacd..ba7291330a 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt +++ b/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 = emptyList(), ) { fun resolveInfo(fullTextSearch: Boolean = true): RelayInfo = RelayInfo( @@ -160,6 +162,27 @@ data class StaticConfig( val kind_blacklist: List = 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. diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt new file mode 100644 index 0000000000..f6f5b18468 --- /dev/null +++ b/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, + 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(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?, + ) { + 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-$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() + } +} diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt index e21469938e..2d478e52ef 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt +++ b/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 = diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt new file mode 100644 index 0000000000..440e40abf7 --- /dev/null +++ b/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(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(Filter()).single() + assertEquals(valid.id, stored.id) + assertTrue(downstreamStore.query(Filter(ids = listOf(forged.id))).isEmpty()) + } +}