laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-10 00:16:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(geode): stream haveIds in the mirror up-catch-up instead of materializing

Browse Source 
runCatchUpUp used negentropyReconcileIds, which builds the FULL need+have
id lists in memory even though the up direction only needs haveIds — on a
large window (e.g. 1M local events against an empty upstream) that is a
~100 MB+ heap spike per convergence round, plus a needIds list built and
immediately discarded.

Switch to the streaming negentropyReconcile: publish each haveIds batch as
it arrives (bounded to one batch of ids) and drop the need direction via a
no-op onNeedIds. The publish still suspends the reconcile round, so the
back-pressure and the reconcile-as-delivery-check convergence loop are
unchanged — only the peak memory drops from O(window) to O(batch).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012EZeWww5TJnzBZKPoc6mvU
This commit is contained in:
Claude
2026-07-05 04:55:36 +00:00
parent 7aa6144dd1
commit d567aec6de
1 changed files with 40 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt
+40 -32
View File
@@ -23,7 +23,7 @@ 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.accessories.negentropyReconcileIds
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcile
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncOrFetch
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
@@ -499,41 +499,49 @@ class MirrorWorker(
try {
var round = 0
while (round < MAX_UP_SYNC_ROUNDS) {
// Reconcile (need ids are the down catch-up's job — ignore them);
// `have` = events we hold that the upstream still lacks.
val diff =
client.negentropyReconcileIds(
relay = up.url,
filter = catchUpFilter,
localEntries = localEntries,
)
if (diff.haveIds.isEmpty()) {
// Reconcile and PUBLISH the `have` ids (events we hold the
// upstream lacks) as each batch streams in — never materialising
// the full diff. On a large window the id list is millions of
// entries; the streaming reconcile keeps memory at one batch and
// still back-pressures the relay because publishing suspends the
// round. The `need` direction is the down catch-up's job, so its
// ids are discarded (not accumulated) here.
var haveCount = 0
var pushed = 0
client.negentropyReconcile(
relay = up.url,
filter = catchUpFilter,
localEntries = localEntries,
batchSize = HAVE_FETCH_BATCH,
onHaveIds = { batch ->
haveCount += batch.size
for (event in localStore.query<Event>(Filter(ids = batch))) {
// Scope containment: a scoped upstream only receives
// in-scope events. Echo suppression: record the id so
// a BOTH mirror doesn't re-ingest its own push on the
// down sub (but always re-publish — a straggler stays
// in `exchanged` yet still needs delivering).
if (up.filter != null && !up.filter.match(event)) continue
exchanged?.add(event.id)
client.publish(event, setOf(up.url))
pushed++
}
// Pace the outbox so a batch drains before the next.
delay(UP_PUBLISH_PACING_MS)
},
onNeedIds = { },
)
if (haveCount == 0) {
Log.i("MirrorWorker") { "up catch-up to ${up.url.url}: converged after $round round(s)" }
return
}
// Publish the remaining local-only events. `client.publish`'s
// outbox is best-effort under a bulk burst (each publish also
// churns a reconnect), so instead of trusting one pass we
// re-reconcile next round and re-push only what didn't land —
// the reconcile is the delivery check, so the push converges
// to lossless.
var pushed = 0
for (batch in diff.haveIds.chunked(HAVE_FETCH_BATCH)) {
for (event in localStore.query<Event>(Filter(ids = batch))) {
// Scope containment: a scoped upstream only receives
// in-scope events. Echo suppression: record the id so a
// BOTH mirror doesn't re-ingest its own push on the down
// sub (but always re-publish — a straggler stays in
// `exchanged` yet still needs delivering).
if (up.filter != null && !up.filter.match(event)) continue
exchanged?.add(event.id)
client.publish(event, setOf(up.url))
pushed++
}
delay(UP_PUBLISH_PACING_MS)
}
// `client.publish`'s outbox is best-effort under a bulk burst
// (each publish also churns a reconnect), so instead of trusting
// one pass we re-reconcile next round and re-push only what didn't
// land — the reconcile is the delivery check, so the push
// converges to lossless.
sentUp.addAndGet(pushed.toLong())
Log.i("MirrorWorker") { "up catch-up to ${up.url.url}: round $round pushed $pushed (had ${diff.haveIds.size} to go)" }
Log.i("MirrorWorker") { "up catch-up to ${up.url.url}: round $round pushed $pushed (had $haveCount to go)" }
round++
// Let the upstream ingest + OK before the next reconcile, so the
// diff reflects what actually landed rather than what's in flight.