refactor(negentropy): audit fixes for interop tests
- Delete `strfryDrivesGeodeAsServer` — boots strfry but uses
kmp-negentropy ↔ Geode, no actual strfry-vs-geode interop.
Duplicates `Nip77NegentropyTest.negentropyComputesSymmetricDifference`.
- Strip speculative `negentropy { enabled = ... }` and `nofiles`
blocks from the strfry config; defaults are what we want to test.
- `InteropSyncDriver.reconcile` → `negotiate`. The function
computes the symmetric difference; it doesn't move events.
Convert to `suspend fun` to drop a nested `runBlocking` that
could deadlock under dispatcher pressure.
- Inline `pullSync` helper in GeodeVsGeodeNegentropySyncTest —
it was a one-line wrapper with a misleading name.
- Batch `relayB.preload(needFromA)` instead of looping.
- Tighten `idsOnRelay(NormalizedRelayUrl)` signature (was
re-parsing the string on every call).
- Drop redundant `assertTrue(size > cap)` after `assertEquals(11)`.
This commit is contained in:
Claude
+24
-53
@@ -29,6 +29,7 @@ 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.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||
@@ -112,25 +113,7 @@ class GeodeVsGeodeNegentropySyncTest {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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,
|
||||
)
|
||||
private val driver by lazy { InteropSyncDriver(httpClient) }
|
||||
|
||||
@Test
|
||||
fun bidirectionalSyncConvergesTwoRelays() =
|
||||
@@ -145,71 +128,59 @@ class GeodeVsGeodeNegentropySyncTest {
|
||||
relayB.preload(bEvents)
|
||||
|
||||
val filter = Filter(kinds = listOf(1))
|
||||
val urlA = serverA.url.normalizeRelayUrl()
|
||||
val urlB = serverB.url.normalizeRelayUrl()
|
||||
|
||||
// --- B pulls from A ---
|
||||
val pullAtoB = pullSync(serverA.url, bEvents, filter)
|
||||
assertNull(pullAtoB.error, "A→B reconciliation must not error")
|
||||
// B negotiates the symmetric difference with A.
|
||||
val diff = driver.negotiate(serverA.url, filter, bEvents)
|
||||
assertNull(diff.error, "negotiation must not error: ${diff.error}")
|
||||
// From B's perspective: needs A-only, has B-only.
|
||||
assertEquals(
|
||||
aEvents.subList(0, 5).map { it.id }.toSet(),
|
||||
pullAtoB.needIds,
|
||||
diff.needIds,
|
||||
"B should NEED [0..4] from A",
|
||||
)
|
||||
assertEquals(
|
||||
bEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
pullAtoB.haveIds,
|
||||
diff.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)
|
||||
client.fetchAll(relay = urlA, filter = Filter(ids = diff.needIds.toList()))
|
||||
relayB.preload(needFromA)
|
||||
|
||||
for (id in pullAtoB.haveIds) {
|
||||
val ev = bEvents.first { it.id == id }
|
||||
client.publishAndConfirm(ev, setOf(serverA.url.normalizeRelayUrl()))
|
||||
for (id in diff.haveIds) {
|
||||
client.publishAndConfirm(bEvents.first { it.id == id }, setOf(urlA))
|
||||
}
|
||||
|
||||
// --- 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")
|
||||
val expected = all.map { it.id }.toSet()
|
||||
assertEquals(expected, idsOnRelay(urlA, filter), "Relay A should hold every event")
|
||||
assertEquals(expected, idsOnRelay(urlB, filter), "Relay B should hold every event")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun negentropyConvergesInBoundedRoundsOnSmallCorpus() =
|
||||
runBlocking {
|
||||
val all = makeEvents(200)
|
||||
val aEvents = all.subList(0, 150)
|
||||
relayA.preload(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)
|
||||
val res = driver.negotiate(serverA.url, Filter(kinds = listOf(1)), bEvents)
|
||||
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.
|
||||
// strfry typically converges these in ≤5 rounds; ≤16 is
|
||||
// generous headroom that still catches regressions.
|
||||
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,
|
||||
/** Every event id matching [filter] visible on the relay at [url] via REQ. */
|
||||
private suspend fun idsOnRelay(
|
||||
url: NormalizedRelayUrl,
|
||||
filter: Filter,
|
||||
): Set<HexKey> =
|
||||
client
|
||||
.fetchAll(
|
||||
relay = wsUrl.normalizeRelayUrl(),
|
||||
filter = filter,
|
||||
).map { it.id }
|
||||
.toSet()
|
||||
): Set<HexKey> = client.fetchAll(relay = url, filter = filter).map { it.id }.toSet()
|
||||
}
|
||||
|
||||
+61
-151
@@ -20,129 +20,100 @@
|
||||
*/
|
||||
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.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 java.io.File
|
||||
import java.net.ServerSocket
|
||||
import java.net.Socket
|
||||
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.
|
||||
* Reciprocal interop test: Geode's NIP-77 client driving 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:
|
||||
* **Opt-in.** Skipped unless `STRFRY_BIN` env var (or
|
||||
* `-Dstrfry.bin=...`) points at a `strfry` binary. When unset, the
|
||||
* test prints a `[skip]` line and returns. Mirrors the gate
|
||||
* `LoadBenchmark` uses for `-DrunLoadBenchmark`:
|
||||
*
|
||||
* 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".
|
||||
* temp LMDB dir. We feed events into it via the NIP-01 EVENT wire
|
||||
* (no `strfry import`), then run [InteropSyncDriver] against it.
|
||||
* Strfry's `RelayNegentropy.cpp` answers the same NIP-77 wire we test
|
||||
* against Geode — passing both is 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)
|
||||
}
|
||||
private val httpClient by lazy { OkHttpClient.Builder().build() }
|
||||
|
||||
@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.
|
||||
* Boots a strfry instance in a temp LMDB dir on a free port.
|
||||
* Writes the smallest config strfry accepts, starts the
|
||||
* subprocess, and polls until the WebSocket port is reachable.
|
||||
*
|
||||
* The config is intentionally minimal — strfry's defaults
|
||||
* (negentropy on, sane limits) are what we want to test against.
|
||||
* Adding speculative config keys risks failing on schema drift.
|
||||
*/
|
||||
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 =
|
||||
strfryProcess =
|
||||
ProcessBuilder(strfryBin, "--config", configFile.absolutePath, "relay")
|
||||
.redirectErrorStream(true)
|
||||
.redirectOutput(File(strfryDir, "strfry.log"))
|
||||
strfryProcess = pb.start()
|
||||
.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!!
|
||||
Socket("127.0.0.1", port).close()
|
||||
return "ws://127.0.0.1:$port/"
|
||||
}
|
||||
Thread.sleep(100)
|
||||
}
|
||||
@@ -161,37 +132,33 @@ class GeodeVsStrfryNegentropySyncTest {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Push every event in [events] to the relay at [wsUrl] over a
|
||||
* one-shot WebSocket, waiting for an `OK` response per event.
|
||||
*
|
||||
* Used to seed strfry from the same `Event` objects Geode's
|
||||
* `Relay.preload` accepts — that way both sides start from a
|
||||
* byte-identical corpus.
|
||||
*/
|
||||
private fun publishToStrfry(
|
||||
private suspend 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 incoming = Channel<String>(UNLIMITED)
|
||||
val ws =
|
||||
httpClient.newWebSocket(
|
||||
okhttp3.Request
|
||||
.Builder()
|
||||
.url(wsUrl.replace("ws://", "http://"))
|
||||
.build(),
|
||||
object : okhttp3.WebSocketListener() {
|
||||
Request.Builder().url(wsUrl.replace("ws://", "http://")).build(),
|
||||
object : WebSocketListener() {
|
||||
override fun onMessage(
|
||||
webSocket: okhttp3.WebSocket,
|
||||
webSocket: WebSocket,
|
||||
text: String,
|
||||
) {
|
||||
incoming.trySend(text)
|
||||
}
|
||||
|
||||
override fun onFailure(
|
||||
webSocket: okhttp3.WebSocket,
|
||||
webSocket: WebSocket,
|
||||
t: Throwable,
|
||||
response: okhttp3.Response?,
|
||||
response: Response?,
|
||||
) {
|
||||
incoming.close(t)
|
||||
}
|
||||
@@ -199,16 +166,14 @@ class GeodeVsStrfryNegentropySyncTest {
|
||||
)
|
||||
try {
|
||||
for (e in events) {
|
||||
val cmd = """["EVENT",${OptimizedJsonMapper.toJson(e)}]"""
|
||||
check(ws.send(cmd)) { "publish to strfry failed" }
|
||||
check(ws.send("""["EVENT",${OptimizedJsonMapper.toJson(e)}]""")) {
|
||||
"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()
|
||||
}
|
||||
var oks = 0
|
||||
withTimeout(30_000) {
|
||||
while (oks < events.size) {
|
||||
if (incoming.receive().startsWith("[\"OK\"")) oks++
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
@@ -225,86 +190,31 @@ class GeodeVsStrfryNegentropySyncTest {
|
||||
}
|
||||
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].
|
||||
// Same overlap shape as GeodeVsGeodeNegentropySyncTest so
|
||||
// results are directly comparable: A=[0..14], local=[5..19].
|
||||
val all = makeEvents(20)
|
||||
val strfryEvents = all.subList(0, 15)
|
||||
val geodeEvents = all.subList(5, 20)
|
||||
val localEvents = 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)
|
||||
// The wire we care about: kmp-negentropy (client) talking
|
||||
// to strfry's `Negentropy ne(storage, 500'000)` (server).
|
||||
// Symmetric difference must match the Geode-vs-Geode case.
|
||||
val res = InteropSyncDriver(httpClient).negotiate(strfryWs, filter, localEvents)
|
||||
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",
|
||||
"client should NEED [0..4] from strfry",
|
||||
)
|
||||
assertEquals(
|
||||
geodeEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
localEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
res.haveIds,
|
||||
"Geode should announce HAVE for [15..19]",
|
||||
"client 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -30,7 +30,6 @@ 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
|
||||
@@ -41,27 +40,27 @@ 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`).
|
||||
* tests. Drives one round of NIP-77 *negotiation* (NEG-OPEN /
|
||||
* NEG-MSG / NEG-CLOSE) against a real `ws://` endpoint — a Geode
|
||||
* `LocalRelayServer`, a strfry process, or any other NIP-77 relay.
|
||||
*
|
||||
* 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.
|
||||
* to keep the wire format under 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.
|
||||
* Note: this driver only computes the symmetric difference. The
|
||||
* actual *sync* (REQ for `needIds`, EVENT for `haveIds`) is the
|
||||
* caller's job; that's a NIP-01 follow-up, not part of NIP-77.
|
||||
*/
|
||||
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.
|
||||
* Negotiates the symmetric difference between `localEvents` and
|
||||
* the relay at [wsUrl] under [filter]. Returns the id sets so
|
||||
* the caller can close the loop with REQ / EVENT.
|
||||
*
|
||||
* @param wsUrl the source relay's `ws://…` URL.
|
||||
* @param filter NEG-OPEN filter — usually the broadest filter the
|
||||
@@ -72,12 +71,12 @@ class InteropSyncDriver(
|
||||
* 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
|
||||
* @param maxRounds upper bound on round trips. Strfry typically
|
||||
* converges in ≤5 rounds for 100 k corpora; 64 is a generous
|
||||
* safety net that catches pathological splits without hanging
|
||||
* tests forever.
|
||||
*/
|
||||
fun reconcile(
|
||||
suspend fun negotiate(
|
||||
wsUrl: String,
|
||||
filter: Filter,
|
||||
localEvents: List<Event>,
|
||||
@@ -128,10 +127,7 @@ class InteropSyncDriver(
|
||||
val needIds = mutableSetOf<HexKey>()
|
||||
var rounds = 0
|
||||
while (rounds < maxRounds) {
|
||||
val raw =
|
||||
runBlocking {
|
||||
withTimeout(timeoutMs) { incoming.receive() }
|
||||
}
|
||||
val raw = withTimeout(timeoutMs) { incoming.receive() }
|
||||
val msg = OptimizedJsonMapper.fromJsonToMessage(raw)
|
||||
rounds++
|
||||
when (msg) {
|
||||
|
||||
Reference in New Issue
Block a user