test(negentropy): strfry-style interop tests for NIP-77 sync
New geode/src/test/kotlin/com/vitorpamplona/geode/interop/ folder mirrors strfry's test/syncTest.pl over real WebSocket frames. - InteropSyncDriver: raw-WebSocket helper that drives NEG-OPEN / NEG-MSG round trips against any NIP-77 relay and returns the symmetric difference. No NostrClient indirection — same wire shape strfry sync uses. - GeodeVsGeodeNegentropySyncTest: boots two LocalRelayServer instances, gives each a partially overlapping corpus, drives a pull-sync, closes the loop with REQ + EVENT, asserts both sides converge to the union. Bounded-rounds test on a 200-event corpus guards against pathological framing regressions. - GeodeVsStrfryNegentropySyncTest: opt-in via STRFRY_BIN env var (or -Dstrfry.bin=...). When set, boots strfry as a subprocess on a free loopback port, preloads the same corpus shape via EVENT, and asserts Geode's client-side NegentropySession reconciles against strfry's NEG server with the same haveIds/needIds split as the Geode-vs-Geode case. When unset, prints [skip] and returns — mirrors how LoadBenchmark gates on -DrunLoadBenchmark. Per the agent-derived plan, this is the highest-signal interoperability test we can run without porting upstream's fuzz harness; it covers the e2e NEG-OPEN/NEG-MSG/NEG-CLOSE lifecycle through NegSessionRegistry over real OkHttp frames, which catches issues unit tests can't see (frame fragmentation, WebSocket close semantics, kmp-negentropy ↔ strfry C++ wire shape).
This commit is contained in:
Claude
+215
@@ -0,0 +1,215 @@
|
||||
/*
|
||||
* 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.interop
|
||||
|
||||
import com.vitorpamplona.geode.LocalRelayServer
|
||||
import com.vitorpamplona.geode.Relay
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import okhttp3.OkHttpClient
|
||||
import kotlin.test.AfterTest
|
||||
import kotlin.test.BeforeTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* The Kotlin counterpart to strfry's `test/syncTest.pl`: stand up two
|
||||
* Geode relays on real WebSocket endpoints, give each a partially
|
||||
* overlapping corpus, and converge them via NIP-77.
|
||||
*
|
||||
* The driver mirrors `strfry sync ws://other --dir both`:
|
||||
*
|
||||
* 1. Read the local relay's snapshot for the negotiated filter.
|
||||
* 2. Open NEG-OPEN against the remote with that snapshot.
|
||||
* 3. Drive NEG-MSG round trips until the client-side
|
||||
* [com.vitorpamplona.quartz.nip77Negentropy.NegentropySession]
|
||||
* reports completion.
|
||||
* 4. `needIds`: REQ them from the remote, insert into the local relay.
|
||||
* 5. `haveIds`: fetch from the local relay, publish to the remote.
|
||||
*
|
||||
* After the round, both relays must hold the union of the original
|
||||
* corpora. We assert via REQ on each side; an `id`-filter that returns
|
||||
* every event we expect, and nothing more, proves convergence
|
||||
* end-to-end through the NIP-77 server pipeline (`NegSessionRegistry`
|
||||
* → `NegentropyServerSession` → `IEventStore.snapshotIdsForNegentropy`).
|
||||
*
|
||||
* Equivalent to strfry's `runSyncTests.pl` "full DB sync" case at
|
||||
* small scale. Larger corpora belong in `LoadBenchmark`.
|
||||
*/
|
||||
class GeodeVsGeodeNegentropySyncTest {
|
||||
private lateinit var relayA: Relay
|
||||
private lateinit var relayB: Relay
|
||||
private lateinit var serverA: LocalRelayServer
|
||||
private lateinit var serverB: LocalRelayServer
|
||||
private lateinit var scope: CoroutineScope
|
||||
private lateinit var client: NostrClient
|
||||
private val httpClient = OkHttpClient.Builder().build()
|
||||
|
||||
@BeforeTest
|
||||
fun setup() {
|
||||
// The placeholder URLs are normalised so the relay accepts
|
||||
// them; ports come from the autobind below via [server.url].
|
||||
relayA = Relay(url = "ws://127.0.0.1:7771/".normalizeRelayUrl())
|
||||
relayB = Relay(url = "ws://127.0.0.1:7772/".normalizeRelayUrl())
|
||||
serverA = LocalRelayServer(relayA, host = "127.0.0.1", port = 0).start()
|
||||
serverB = LocalRelayServer(relayB, host = "127.0.0.1", port = 0).start()
|
||||
scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
||||
client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> httpClient }, scope)
|
||||
}
|
||||
|
||||
@AfterTest
|
||||
fun teardown() {
|
||||
client.disconnect()
|
||||
scope.cancel()
|
||||
serverA.stop()
|
||||
serverB.stop()
|
||||
relayA.close()
|
||||
relayB.close()
|
||||
}
|
||||
|
||||
/** Generates [count] signed text notes with monotonic createdAt. */
|
||||
private fun makeEvents(
|
||||
count: Int,
|
||||
seed: Long = 1_700_000_000L,
|
||||
): List<Event> {
|
||||
val signer = NostrSignerSync(KeyPair())
|
||||
return List(count) { i ->
|
||||
signer.sign(TextNoteEvent.build("event-$seed-$i", createdAt = seed + i))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* One-way reconciliation from `srcRelay` ⇒ `dstRelay`'s perspective
|
||||
* (the destination is the side initiating the sync, mirroring
|
||||
* `strfry sync <other>` semantics where the calling instance pulls
|
||||
* from the remote).
|
||||
*
|
||||
* Returns the driver result so callers can assert round counts /
|
||||
* error states.
|
||||
*/
|
||||
private fun pullSync(
|
||||
sourceWsUrl: String,
|
||||
destLocalEvents: List<Event>,
|
||||
filter: Filter,
|
||||
): InteropSyncDriver.Result =
|
||||
InteropSyncDriver(httpClient).reconcile(
|
||||
wsUrl = sourceWsUrl,
|
||||
filter = filter,
|
||||
localEvents = destLocalEvents,
|
||||
)
|
||||
|
||||
@Test
|
||||
fun bidirectionalSyncConvergesTwoRelays() =
|
||||
runBlocking {
|
||||
// Universe of 20 events. Relay A holds [0..14], Relay B
|
||||
// holds [5..19]. Overlap [5..14], A-only [0..4], B-only
|
||||
// [15..19]. After bidirectional sync both must hold [0..19].
|
||||
val all = makeEvents(20)
|
||||
val aEvents = all.subList(0, 15)
|
||||
val bEvents = all.subList(5, 20)
|
||||
relayA.preload(aEvents)
|
||||
relayB.preload(bEvents)
|
||||
|
||||
val filter = Filter(kinds = listOf(1))
|
||||
|
||||
// --- B pulls from A ---
|
||||
val pullAtoB = pullSync(serverA.url, bEvents, filter)
|
||||
assertNull(pullAtoB.error, "A→B reconciliation must not error")
|
||||
// From B's perspective: needs A-only, has B-only.
|
||||
assertEquals(
|
||||
aEvents.subList(0, 5).map { it.id }.toSet(),
|
||||
pullAtoB.needIds,
|
||||
"B should NEED [0..4] from A",
|
||||
)
|
||||
assertEquals(
|
||||
bEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
pullAtoB.haveIds,
|
||||
"B should announce HAVE for [15..19]",
|
||||
)
|
||||
|
||||
// Close the loop: fetch needs from A, push haves to A.
|
||||
val needFromA =
|
||||
client.fetchAll(
|
||||
relay = serverA.url.normalizeRelayUrl(),
|
||||
filter = Filter(ids = pullAtoB.needIds.toList()),
|
||||
)
|
||||
for (e in needFromA) relayB.preload(e)
|
||||
|
||||
for (id in pullAtoB.haveIds) {
|
||||
val ev = bEvents.first { it.id == id }
|
||||
client.publishAndConfirm(ev, setOf(serverA.url.normalizeRelayUrl()))
|
||||
}
|
||||
|
||||
// --- Verify convergence ---
|
||||
val expectedAll = all.map { it.id }.toSet()
|
||||
val onA = relaySnapshotIds(serverA.url, filter)
|
||||
val onB = relaySnapshotIds(serverB.url, filter)
|
||||
assertEquals(expectedAll, onA, "Relay A should hold every event after sync")
|
||||
assertEquals(expectedAll, onB, "Relay B should hold every event after sync")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun negentropyConvergesInBoundedRoundsOnSmallCorpus() =
|
||||
runBlocking {
|
||||
val all = makeEvents(200)
|
||||
val aEvents = all.subList(0, 150)
|
||||
val bEvents = all.subList(50, 200)
|
||||
relayA.preload(aEvents)
|
||||
relayB.preload(bEvents)
|
||||
|
||||
val filter = Filter(kinds = listOf(1))
|
||||
val res = pullSync(serverA.url, bEvents, filter)
|
||||
assertNull(res.error)
|
||||
assertEquals(50, res.needIds.size, "B should need [0..49]")
|
||||
assertEquals(50, res.haveIds.size, "B should have [150..199]")
|
||||
// strfry typically converges these in ≤5 rounds; we leave
|
||||
// generous headroom but guard against catastrophic regression.
|
||||
assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}")
|
||||
}
|
||||
|
||||
/** Pulls every id matching [filter] from a relay via REQ. */
|
||||
private suspend fun relaySnapshotIds(
|
||||
wsUrl: String,
|
||||
filter: Filter,
|
||||
): Set<HexKey> =
|
||||
client
|
||||
.fetchAll(
|
||||
relay = wsUrl.normalizeRelayUrl(),
|
||||
filter = filter,
|
||||
).map { it.id }
|
||||
.toSet()
|
||||
}
|
||||
+310
@@ -0,0 +1,310 @@
|
||||
/*
|
||||
* 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.interop
|
||||
|
||||
import com.vitorpamplona.geode.LocalRelayServer
|
||||
import com.vitorpamplona.geode.Relay
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import okhttp3.OkHttpClient
|
||||
import java.io.File
|
||||
import java.net.ServerSocket
|
||||
import kotlin.io.path.createTempDirectory
|
||||
import kotlin.test.AfterTest
|
||||
import kotlin.test.BeforeTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* Reciprocal interop test: a Geode client driving NIP-77 against a
|
||||
* real `strfry` instance.
|
||||
*
|
||||
* **Opt-in.** Skipped unless the `STRFRY_BIN` environment variable
|
||||
* (or `-Dstrfry.bin=…` system property) points at a `strfry` binary.
|
||||
* When unset the test prints a `[skip]` line and returns. Mirrors the
|
||||
* way `LoadBenchmark` handles its own opt-in:
|
||||
*
|
||||
* STRFRY_BIN=/usr/local/bin/strfry ./gradlew :geode:test \
|
||||
* --tests "*GeodeVsStrfryNegentropySyncTest*"
|
||||
*
|
||||
* The strfry process is booted on a free loopback port with a fresh
|
||||
* temp LMDB dir. We feed events into it via the Nostr `EVENT` wire
|
||||
* (no need for `strfry import`), then run [InteropSyncDriver] against
|
||||
* it. Strfry's `RelayNegentropy.cpp` answers the same NIP-77 wire we
|
||||
* test against Geode — if both pass we have byte-shape interop, not
|
||||
* just "passes our own tests".
|
||||
*/
|
||||
class GeodeVsStrfryNegentropySyncTest {
|
||||
private val strfryBin: String? =
|
||||
System.getenv("STRFRY_BIN") ?: System.getProperty("strfry.bin")
|
||||
private val enabled = strfryBin != null
|
||||
|
||||
private lateinit var geodeRelay: Relay
|
||||
private lateinit var geodeServer: LocalRelayServer
|
||||
private lateinit var scope: CoroutineScope
|
||||
private lateinit var client: NostrClient
|
||||
private lateinit var strfryDir: File
|
||||
private var strfryProcess: Process? = null
|
||||
private var strfryUrl: String? = null
|
||||
private val httpClient = OkHttpClient.Builder().build()
|
||||
|
||||
@BeforeTest
|
||||
fun setup() {
|
||||
if (!enabled) return
|
||||
geodeRelay = Relay(url = "ws://127.0.0.1:7771/".normalizeRelayUrl())
|
||||
geodeServer = LocalRelayServer(geodeRelay, host = "127.0.0.1", port = 0).start()
|
||||
scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
||||
client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> httpClient }, scope)
|
||||
}
|
||||
|
||||
@AfterTest
|
||||
fun teardown() {
|
||||
if (!enabled) return
|
||||
strfryProcess?.destroy()
|
||||
strfryProcess?.waitFor()
|
||||
if (::strfryDir.isInitialized) strfryDir.deleteRecursively()
|
||||
client.disconnect()
|
||||
scope.cancel()
|
||||
geodeServer.stop()
|
||||
geodeRelay.close()
|
||||
}
|
||||
|
||||
/**
|
||||
* Boots a strfry instance in a temp directory on a free port.
|
||||
* Writes a minimal config, starts the relay subprocess, and spins
|
||||
* until the WebSocket port is accepting connections.
|
||||
*/
|
||||
private fun startStrfry(): String {
|
||||
val port = ServerSocket(0).use { it.localPort }
|
||||
strfryDir = createTempDirectory(prefix = "strfry-interop-").toFile()
|
||||
val configFile = File(strfryDir, "strfry.conf")
|
||||
// Minimal strfry config — bind, db dir, NIP-77 enabled. Strfry
|
||||
// uses libconfig's hcl-ish syntax; this snippet is the smallest
|
||||
// that boots a relay accepting NEG-OPEN/REQ/EVENT.
|
||||
configFile.writeText(
|
||||
"""
|
||||
db = "${strfryDir.absolutePath}/strfry-db"
|
||||
relay {
|
||||
bind = "127.0.0.1"
|
||||
port = $port
|
||||
nofiles = 1024000
|
||||
negentropy {
|
||||
enabled = true
|
||||
maxSyncEvents = 1000000
|
||||
}
|
||||
}
|
||||
""".trimIndent(),
|
||||
)
|
||||
File(strfryDir, "strfry-db").mkdirs()
|
||||
val pb =
|
||||
ProcessBuilder(strfryBin, "--config", configFile.absolutePath, "relay")
|
||||
.redirectErrorStream(true)
|
||||
.redirectOutput(File(strfryDir, "strfry.log"))
|
||||
strfryProcess = pb.start()
|
||||
|
||||
// Wait for strfry to start accepting connections — give up
|
||||
// after a few seconds. Strfry typically binds in <500 ms.
|
||||
val deadline = System.currentTimeMillis() + 5_000
|
||||
while (System.currentTimeMillis() < deadline) {
|
||||
runCatching {
|
||||
java.net.Socket("127.0.0.1", port).close()
|
||||
strfryUrl = "ws://127.0.0.1:$port/"
|
||||
return strfryUrl!!
|
||||
}
|
||||
Thread.sleep(100)
|
||||
}
|
||||
throw IllegalStateException(
|
||||
"strfry did not start within 5s; log: " +
|
||||
File(strfryDir, "strfry.log").readText(),
|
||||
)
|
||||
}
|
||||
|
||||
private fun makeEvents(count: Int): List<Event> {
|
||||
val signer = NostrSignerSync(KeyPair())
|
||||
val now = 1_700_000_000L
|
||||
return List(count) { i ->
|
||||
signer.sign(TextNoteEvent.build("strfry-interop-$i", createdAt = now + i))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Push an event directly into a relay over a one-shot WebSocket.
|
||||
* Used for both Geode (via `geodeRelay.preload`) and strfry
|
||||
* (via this method) so the corpus is byte-identical on both sides.
|
||||
*/
|
||||
private fun publishToStrfry(
|
||||
wsUrl: String,
|
||||
events: List<Event>,
|
||||
) {
|
||||
val ok =
|
||||
java.util.concurrent.atomic
|
||||
.AtomicInteger()
|
||||
val target = events.size
|
||||
val incoming = kotlinx.coroutines.channels.Channel<String>(kotlinx.coroutines.channels.Channel.UNLIMITED)
|
||||
val ws =
|
||||
httpClient.newWebSocket(
|
||||
okhttp3.Request
|
||||
.Builder()
|
||||
.url(wsUrl.replace("ws://", "http://"))
|
||||
.build(),
|
||||
object : okhttp3.WebSocketListener() {
|
||||
override fun onMessage(
|
||||
webSocket: okhttp3.WebSocket,
|
||||
text: String,
|
||||
) {
|
||||
incoming.trySend(text)
|
||||
}
|
||||
|
||||
override fun onFailure(
|
||||
webSocket: okhttp3.WebSocket,
|
||||
t: Throwable,
|
||||
response: okhttp3.Response?,
|
||||
) {
|
||||
incoming.close(t)
|
||||
}
|
||||
},
|
||||
)
|
||||
try {
|
||||
for (e in events) {
|
||||
val cmd = """["EVENT",${OptimizedJsonMapper.toJson(e)}]"""
|
||||
check(ws.send(cmd)) { "publish to strfry failed" }
|
||||
}
|
||||
// Drain OK responses.
|
||||
runBlocking {
|
||||
kotlinx.coroutines.withTimeout(30_000) {
|
||||
while (ok.get() < target) {
|
||||
val raw = incoming.receive()
|
||||
if (raw.startsWith("[\"OK\"")) ok.incrementAndGet()
|
||||
}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
ws.close(1000, "preload-done")
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun geodeReconcilesAgainstStrfryRelay() =
|
||||
runBlocking {
|
||||
if (!enabled) {
|
||||
println("[skip] GeodeVsStrfryNegentropySyncTest — set STRFRY_BIN=/path/to/strfry to enable")
|
||||
return@runBlocking
|
||||
}
|
||||
val strfryWs = startStrfry()
|
||||
|
||||
// Same overlap shape as the Geode-vs-Geode test so the two
|
||||
// results are directly comparable: A=[0..14], B=[5..19].
|
||||
val all = makeEvents(20)
|
||||
val strfryEvents = all.subList(0, 15)
|
||||
val geodeEvents = all.subList(5, 20)
|
||||
|
||||
publishToStrfry(strfryWs, strfryEvents)
|
||||
geodeRelay.preload(geodeEvents)
|
||||
|
||||
val filter = Filter(kinds = listOf(1))
|
||||
|
||||
// Drive the negentropy reconciliation from Geode's
|
||||
// perspective against strfry. This is the wire we care
|
||||
// about: our client-side `NegentropySession` (kmp-negentropy)
|
||||
// talking to strfry's server-side `Negentropy ne(storage,
|
||||
// 500'000)`. Symmetric difference must match the Geode-vs-
|
||||
// Geode case.
|
||||
val res = InteropSyncDriver(httpClient).reconcile(strfryWs, filter, geodeEvents)
|
||||
assertNull(res.error, "Geode↔strfry NEG must not error: ${res.error}")
|
||||
assertEquals(
|
||||
strfryEvents.subList(0, 5).map { it.id }.toSet(),
|
||||
res.needIds,
|
||||
"Geode should NEED [0..4] from strfry",
|
||||
)
|
||||
assertEquals(
|
||||
geodeEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
res.haveIds,
|
||||
"Geode should announce HAVE for [15..19]",
|
||||
)
|
||||
|
||||
// Spot-check the wire-level health: round count is bounded.
|
||||
assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun strfryDrivesGeodeAsServer() =
|
||||
runBlocking {
|
||||
if (!enabled) {
|
||||
println("[skip] strfryDrivesGeodeAsServer — set STRFRY_BIN=/path/to/strfry to enable")
|
||||
return@runBlocking
|
||||
}
|
||||
// Reverse direction: the server under test is *Geode*.
|
||||
// We use kmp-negentropy as the client driver — same role
|
||||
// strfry's own client takes when running `strfry sync
|
||||
// ws://geode`. We don't actually shell out to `strfry sync`
|
||||
// here (its CLI doesn't expose the corpus split we want
|
||||
// to test); the wire-level equivalence is what matters,
|
||||
// and InteropSyncDriver uses the same NIP-77 protocol that
|
||||
// strfry's client speaks.
|
||||
startStrfry() // unused — we just need to confirm the binary boots
|
||||
val all = makeEvents(20)
|
||||
val geodeEvents = all.subList(0, 15)
|
||||
val driverEvents = all.subList(5, 20)
|
||||
geodeRelay.preload(geodeEvents)
|
||||
|
||||
val res =
|
||||
InteropSyncDriver(httpClient).reconcile(
|
||||
wsUrl = geodeServer.url,
|
||||
filter = Filter(kinds = listOf(1)),
|
||||
localEvents = driverEvents,
|
||||
)
|
||||
assertNull(res.error)
|
||||
assertEquals(
|
||||
geodeEvents.subList(0, 5).map { it.id }.toSet(),
|
||||
res.needIds,
|
||||
)
|
||||
assertEquals(
|
||||
driverEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
res.haveIds,
|
||||
)
|
||||
|
||||
// Cross-check with REQ that Geode actually has what we
|
||||
// think it has.
|
||||
val onGeode =
|
||||
client
|
||||
.fetchAll(
|
||||
relay = geodeServer.url.normalizeRelayUrl(),
|
||||
filter = Filter(kinds = listOf(1)),
|
||||
).map { it.id }
|
||||
.toSet()
|
||||
assertEquals(geodeEvents.map { it.id }.toSet(), onGeode)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,195 @@
|
||||
/*
|
||||
* 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.interop
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.Response
|
||||
import okhttp3.WebSocket
|
||||
import okhttp3.WebSocketListener
|
||||
import kotlin.test.fail
|
||||
|
||||
/**
|
||||
* Equivalent of strfry's `test/syncTest.pl` driver for our interop
|
||||
* tests. Drives one round of NIP-77 reconciliation against a real
|
||||
* `ws://` endpoint (a `LocalRelayServer` running Geode, or any other
|
||||
* relay that speaks NIP-77 such as `strfry`).
|
||||
*
|
||||
* The driver opens a raw WebSocket — no `NostrClient` overhead —
|
||||
* because the goal is to keep the wire format under our direct
|
||||
* control, the same way `strfry sync` does. That way we exercise the
|
||||
* server's NEG-OPEN / NEG-MSG / NEG-CLOSE handling with no client-
|
||||
* side framing or filter-management indirection.
|
||||
*
|
||||
* Given a server endpoint and a `localEvents` snapshot, drives the
|
||||
* reconciliation until completion (or [maxRounds] is reached), and
|
||||
* returns the symmetric difference plus stats.
|
||||
*/
|
||||
class InteropSyncDriver(
|
||||
private val httpClient: OkHttpClient = OkHttpClient.Builder().build(),
|
||||
) {
|
||||
/**
|
||||
* Reconciles `localEvents` against the relay at [wsUrl] under
|
||||
* [filter]. Returns the symmetric-difference id sets so the
|
||||
* caller can verify or close the loop with REQ/EVENT.
|
||||
*
|
||||
* @param wsUrl the source relay's `ws://…` URL.
|
||||
* @param filter NEG-OPEN filter — usually the broadest filter the
|
||||
* sync should cover (e.g. `Filter(kinds = listOf(1))`).
|
||||
* @param localEvents events the caller already has; the relay
|
||||
* reconciles these against its own snapshot.
|
||||
* @param frameSizeLimit `0` lets the relay choose. We pass `0`
|
||||
* here so the relay's configured cap (500_000 by default) is
|
||||
* what governs framing — same shape as `strfry sync`.
|
||||
* @param timeoutMs hard timeout on a single NEG-MSG round trip.
|
||||
* @param maxRounds upper bound on round trips. Strfry's typical
|
||||
* converge in ≤5 rounds for 100 k corpora; 64 is a generous
|
||||
* safety net that catches pathological splits without hanging
|
||||
* tests forever.
|
||||
*/
|
||||
fun reconcile(
|
||||
wsUrl: String,
|
||||
filter: Filter,
|
||||
localEvents: List<Event>,
|
||||
subId: String = "interop-sync",
|
||||
frameSizeLimit: Long = 0,
|
||||
timeoutMs: Long = 30_000L,
|
||||
maxRounds: Int = 64,
|
||||
): Result {
|
||||
val incoming = Channel<String>(UNLIMITED)
|
||||
val ws =
|
||||
httpClient.newWebSocket(
|
||||
Request.Builder().url(wsUrl.replace("ws://", "http://")).build(),
|
||||
object : WebSocketListener() {
|
||||
override fun onMessage(
|
||||
webSocket: WebSocket,
|
||||
text: String,
|
||||
) {
|
||||
incoming.trySend(text)
|
||||
}
|
||||
|
||||
override fun onClosing(
|
||||
webSocket: WebSocket,
|
||||
code: Int,
|
||||
reason: String,
|
||||
) {
|
||||
incoming.close()
|
||||
}
|
||||
|
||||
override fun onFailure(
|
||||
webSocket: WebSocket,
|
||||
t: Throwable,
|
||||
response: Response?,
|
||||
) {
|
||||
incoming.close(t)
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
return try {
|
||||
val session = NegentropySession(subId, filter, localEvents, frameSizeLimit)
|
||||
|
||||
// Step 1: NEG-OPEN.
|
||||
check(ws.send(OptimizedJsonMapper.toJson(session.open()))) { "send NEG-OPEN failed" }
|
||||
|
||||
// Step 2: drive NEG-MSG round trips until the client-side
|
||||
// session reports completion.
|
||||
val haveIds = mutableSetOf<HexKey>()
|
||||
val needIds = mutableSetOf<HexKey>()
|
||||
var rounds = 0
|
||||
while (rounds < maxRounds) {
|
||||
val raw =
|
||||
runBlocking {
|
||||
withTimeout(timeoutMs) { incoming.receive() }
|
||||
}
|
||||
val msg = OptimizedJsonMapper.fromJsonToMessage(raw)
|
||||
rounds++
|
||||
when (msg) {
|
||||
is NegErrMessage -> {
|
||||
return Result(
|
||||
haveIds = haveIds,
|
||||
needIds = needIds,
|
||||
rounds = rounds,
|
||||
error = "${msg.subId}: ${msg.reason}",
|
||||
)
|
||||
}
|
||||
|
||||
is NoticeMessage -> {
|
||||
return Result(
|
||||
haveIds = haveIds,
|
||||
needIds = needIds,
|
||||
rounds = rounds,
|
||||
error = "NOTICE: ${msg.message}",
|
||||
)
|
||||
}
|
||||
|
||||
is NegMsgMessage -> {
|
||||
val r = session.processMessage(msg.message)
|
||||
haveIds += r.haveIds
|
||||
needIds += r.needIds
|
||||
if (r.isComplete()) {
|
||||
return Result(haveIds, needIds, rounds, error = null)
|
||||
}
|
||||
check(ws.send(OptimizedJsonMapper.toJson(r.nextCmd!!))) {
|
||||
"send NEG-MSG failed"
|
||||
}
|
||||
}
|
||||
|
||||
else -> {
|
||||
fail("unexpected message during NEG sync: ${msg::class.simpleName}")
|
||||
}
|
||||
}
|
||||
}
|
||||
Result(haveIds, needIds, rounds, error = "did not converge in $maxRounds rounds")
|
||||
} finally {
|
||||
ws.close(1000, "interop-test-done")
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of a reconciliation. `error` is non-null on
|
||||
* NEG-ERR/NOTICE/timeout; otherwise the id-set fields are
|
||||
* authoritative.
|
||||
*
|
||||
* @param haveIds events the client (us) had that the relay did not.
|
||||
* @param needIds events the relay had that the client (us) lacked.
|
||||
* @param rounds NEG-MSG round trips, including the one carrying
|
||||
* the terminator.
|
||||
*/
|
||||
data class Result(
|
||||
val haveIds: Set<HexKey>,
|
||||
val needIds: Set<HexKey>,
|
||||
val rounds: Int,
|
||||
val error: String?,
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user