diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt index 9f878186d0..c7ad0ce781 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt @@ -27,6 +27,9 @@ import com.vitorpamplona.amethyst.cli.Output import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.deletionSideChannelFilter +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.excludesDeletionKinds +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyPropagateDeletions import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcile import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer @@ -54,6 +57,16 @@ import java.util.concurrent.atomic.AtomicInteger * Pass both for a full bidirectional sync. The filter flags are the same as * `fetch`/`subscribe`; an empty filter reconciles the whole store. * + * Deletions ride a side-channel. NIP-77 reconciles by id over the content + * filter, so a scoped sync (`--kind 1`) would never carry the kind-5/62 that + * deletes one of those notes — the deletion would be stuck on whichever side + * issued it. So whenever the filter pins `kinds` and excludes 5/62, a second + * reconcile over `{kinds:[5,62], authors:}` runs **both directions + * regardless of --up/--down** ([negentropyPropagateDeletions]): local deletions + * are pushed up so the relay applies them, and the relay's deletions are pulled + * down so the local store applies them. A kind-62 vanish is only pushed to a + * relay it actually targets. Pass `--no-sync-deletions` to opt out. + * * Both directions are pipelined with the reconcile: need-id batches feed * [DOWNLOAD_WORKERS] concurrent by-id REQ drains and have-ids feed a single * uploader, so downloads and uploads overlap the remaining reconcile rounds @@ -93,6 +106,7 @@ object SyncCommand { // Default direction is download; --up adds upload. val up = args.bool("up") val down = args.bool("down") || !up + val syncDeletions = !args.bool("no-sync-deletions") val filter = RawEventSupport.buildFilter(args) Context.open(dataDir).use { ctx -> @@ -160,6 +174,34 @@ object SyncCommand { return Output.error("sync_error", e.message ?: "negentropy sync failed") } + // Deletion side-channel: propagate kind 5/62 both ways, independent of + // the content filter, so a scoped sync still carries the deletions that + // apply to it. No-op (returns null) when the filter already covers 5/62. + val deletionsDown = AtomicInteger(0) + val deletionsUp = AtomicInteger(0) + val deletionResult = + if (syncDeletions && filter.excludesDeletionKinds()) { + try { + val localDeletions = ctx.store.query(filter.deletionSideChannelFilter()) + ctx.client.negentropyPropagateDeletions( + relay = relay, + contentFilter = filter, + localDeletions = localDeletions, + idleTimeoutMs = timeoutMs, + download = { batch -> + deletionsDown.addAndGet(ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs).size) + }, + upload = { event -> + if (ctx.publish(event, setOf(relay)).values.any { it }) deletionsUp.incrementAndGet() + }, + ) + } catch (e: NegentropySyncException) { + return Output.error("sync_error", e.message ?: "deletion sync failed") + } + } else { + null + } + Output.emit( mapOf( "relay" to relay.url, @@ -169,6 +211,10 @@ object SyncCommand { "have" to result.haveCount, "downloaded" to downloaded.get(), "uploaded" to uploaded.get(), + "deletions_need" to (deletionResult?.needCount ?: 0), + "deletions_have" to (deletionResult?.haveCount ?: 0), + "deletions_downloaded" to deletionsDown.get(), + "deletions_uploaded" to deletionsUp.get(), ), ) return 0 diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt index 62936d17fe..8abf263d95 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt @@ -23,8 +23,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.accessories.deletionSideChannelFilter +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.excludesDeletionKinds 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.accessories.shouldPropagateDeletionUp 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 @@ -239,17 +242,24 @@ class MirrorWorker( val listener: SubscriptionListener, initialSince: Long, val watermark: AtomicLong, + // The deletion side-channel filter (kinds 5/62), when the operator scope + // excludes them. Carried here so it rides every re-subscribe too — else a + // reconnect would silently drop live deletions. + val deletionScope: Filter?, ) { @Volatile var issuedSince: Long = initialSince + /** Content scope plus, when in force, the deletion side-channel — both at [since]. */ + fun filtersFrom(since: Long): List = listOfNotNull(scopedBase.copy(since = since), deletionScope?.copy(since = since)) + fun advanceSinceOnReconnect() { val candidate = watermark.get() - WATERMARK_OVERLAP_SECS if (candidate > issuedSince) { issuedSince = candidate client.subscribe( subId = subId, - filters = mapOf(up.url to listOf(scopedBase.copy(since = candidate))), + filters = mapOf(up.url to filtersFrom(candidate)), listener = listener, ) sinceAdvances.incrementAndGet() @@ -330,6 +340,15 @@ class MirrorWorker( val scopedBase = (up.filter ?: Filter()).copy(since = null, limit = null) val initialSince = since - up.backfillSeconds + // Deletion side-channel scope. NIP-77 (and the live REQ) reconcile by + // id over the operator filter, so a kind-scoped mirror (`kinds:[1]`) + // would drop the kind-5/62 that deletes one of those notes — the + // deletion would never reach the other side. When the operator filter + // excludes 5/62, mirror them on their own (same authors), in the + // mirror's configured direction. `null` when the filter already covers + // deletions (unscoped, or 5/62 explicitly listed) — nothing extra to do. + val deletionScope = up.filter?.takeIf { it.excludesDeletionKinds() }?.deletionSideChannelFilter() + // strfry's two-phase model, both directions (`strfry sync --dir // both` + the live router): a one-shot NIP-77 "sync" closes the // historical [initialSince, now] gap — down pulls what the upstream @@ -344,16 +363,16 @@ class MirrorWorker( val upLiveSince = if (catchUpUp) since else initialSince if (up.direction != MirrorDirection.UP) { - downSubs += startDown(i, up, scopedBase, downLiveSince, exchanged) + downSubs += startDown(i, up, scopedBase, downLiveSince, exchanged, deletionScope) } if (up.direction != MirrorDirection.DOWN) { - startUp(up, scopedBase.copy(since = upLiveSince), exchanged) + startUp(up, scopedBase.copy(since = upLiveSince), exchanged, deletionScope?.copy(since = upLiveSince)) } if (catchUpDown) { - scope.launch { runCatchUpDown(up, scopedBase, initialSince, since) } + scope.launch { runCatchUpDown(up, scopedBase, initialSince, since, deletionScope) } } if (catchUpUp) { - scope.launch { runCatchUpUp(up, scopedBase, initialSince, since, exchanged) } + scope.launch { runCatchUpUp(up, scopedBase, initialSince, since, exchanged, deletionScope) } } } client.connect() @@ -411,12 +430,14 @@ class MirrorWorker( scopedBase: Filter, initialSince: Long, until: Long, + deletionScope: Filter?, ) { val catchUpFilter = scopedBase.copy(since = initialSince, until = until) - // Reconcile against what we already hold in this window → download only - // the diff (like `strfry sync`). No store wired → empty local set → the - // whole window is downloaded and the store's unique-id constraint dedups. - val localEntries = store?.snapshotIdsForNegentropy(listOf(catchUpFilter)) ?: emptyList() + + // Even a trusted upstream may only inject events inside the declared + // scope — plus the deletion side-channel scope when one is in force, so a + // kind-scoped mirror still accepts the kind-5/62 that apply to it. + fun inScope(event: Event): Boolean = up.filter == null || up.filter.match(event) || deletionScope?.match(event) == true // Bounded hand-off → one ingest consumer. `onEvent` can't suspend, so it // blocks here when the sink falls behind; because negentropySyncOrFetch's @@ -442,26 +463,32 @@ class MirrorWorker( } } - try { + // Reconciles one filter against what we already hold → downloads only the + // diff (like `strfry sync`). No store wired → empty local set → the whole + // set is downloaded and the store's unique-id constraint dedups. + suspend fun pull(filter: Filter) { + val localEntries = store?.snapshotIdsForNegentropy(listOf(filter)) ?: emptyList() val result = client.negentropySyncOrFetch( relay = up.url, - filter = catchUpFilter, + filter = filter, localEntries = localEntries, onEvent = { event -> - // Same containment as the live path: even a trusted - // upstream may only inject events inside the declared scope. - if (up.filter == null || up.filter.match(event)) { - handoff.trySendBlocking(event) - } else { - filtered.incrementAndGet() - } + if (inScope(event)) handoff.trySendBlocking(event) else filtered.incrementAndGet() }, ) Log.i("MirrorWorker") { val how = if (result.pagedFallback) "paged REQ (upstream has no NIP-77)" else "negentropy" "catch-up from ${up.url.url}: ${result.downloaded} events via $how" } + } + + try { + pull(catchUpFilter) + // Deletions carry no time window: a deletion's created_at is when it + // was issued, not when its target was, so the whole deletion set for + // the scope is reconciled rather than the [initialSince, until] window. + if (deletionScope != null) pull(deletionScope) } catch (e: CancellationException) { throw e } catch (e: Throwable) { @@ -489,65 +516,71 @@ class MirrorWorker( initialSince: Long, until: Long, exchanged: RecentIds?, + deletionScope: Filter?, ) { val localStore = store ?: return val catchUpFilter = scopedBase.copy(since = initialSince, until = until) - // Our local set for this window is fixed; each round reconciles it - // against the upstream, which grows as we push, so the `have` diff - // shrinks to zero. - val localEntries = localStore.snapshotIdsForNegentropy(listOf(catchUpFilter)) - try { + + // Reconcile [filter] against the upstream and PUSH the events we hold that + // it lacks (the `have` ids), re-reconciling each round until the diff is + // empty — the reconcile is the delivery check, so the push converges to + // lossless. [publishable] gates which local events actually go: scope + // containment for content, and per-relay vanish targeting for kind-62 (a + // vanish is only sent to a relay it names). + suspend fun pushUp( + filter: Filter, + label: String, + publishable: (Event) -> Boolean, + ) { + // Our local set for this window is fixed; each round reconciles it + // against the upstream, which grows as we push, so the diff shrinks. + val localEntries = localStore.snapshotIdsForNegentropy(listOf(filter)) var round = 0 while (round < MAX_UP_SYNC_ROUNDS) { - // 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. + // Stream the `have` batches and publish as they arrive — never + // materialising the full diff. Publishing suspends the round, so + // the relay is back-pressured. The `need` direction is the down + // catch-up's job, so its ids are discarded here. var haveCount = 0 var pushed = 0 client.negentropyReconcile( relay = up.url, - filter = catchUpFilter, + filter = filter, localEntries = localEntries, batchSize = HAVE_FETCH_BATCH, onHaveIds = { batch -> haveCount += batch.size for (event in localStore.query(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 + if (!publishable(event)) continue + // 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). 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) + delay(UP_PUBLISH_PACING_MS) // pace the outbox }, onNeedIds = { }, ) if (haveCount == 0) { - Log.i("MirrorWorker") { "up catch-up to ${up.url.url}: converged after $round round(s)" } + Log.i("MirrorWorker") { "up catch-up ($label) to ${up.url.url}: converged after $round round(s)" } return } - // `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 $haveCount to go)" } + Log.i("MirrorWorker") { "up catch-up ($label) 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. - delay(UP_SYNC_SETTLE_MS) + delay(UP_SYNC_SETTLE_MS) // let the upstream ingest + OK before re-checking + } + Log.w("MirrorWorker") { "up catch-up ($label) to ${up.url.url}: did not fully converge in $MAX_UP_SYNC_ROUNDS rounds (live push continues)" } + } + + try { + pushUp(catchUpFilter, "content") { event -> up.filter == null || up.filter.match(event) } + if (deletionScope != null) { + pushUp(deletionScope, "deletions") { event -> shouldPropagateDeletionUp(event, up.url) } } - Log.w("MirrorWorker") { "up catch-up to ${up.url.url}: did not fully converge in $MAX_UP_SYNC_ROUNDS rounds (live push continues)" } } catch (e: CancellationException) { throw e } catch (e: Throwable) { @@ -562,6 +595,7 @@ class MirrorWorker( scopedBase: Filter, initialSince: Long, exchanged: RecentIds?, + deletionScope: Filter?, ): DownSub { // watermark tracks the newest created_at ingested from this // upstream; seeded at initialSince so a still-catching-up @@ -593,7 +627,7 @@ class MirrorWorker( // 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)) { + if (up.filter != null && !up.filter.match(event) && deletionScope?.match(event) != true) { filtered.incrementAndGet() Log.d("MirrorWorker") { "out-of-scope from ${relay.url}: ${event.id}" } return @@ -616,12 +650,13 @@ class MirrorWorker( } } val subId = "geode-mirror-$index" + val downSub = DownSub(subId, up, scopedBase, listener, initialSince, watermark, deletionScope) client.subscribe( subId = subId, - filters = mapOf(up.url to listOf(scopedBase.copy(since = initialSince))), + filters = mapOf(up.url to downSub.filtersFrom(initialSince)), listener = listener, ) - return DownSub(subId, up, scopedBase, listener, initialSince, watermark) + return downSub } /** @@ -638,6 +673,7 @@ class MirrorWorker( up: MirrorUpstream, scopedFilter: Filter, exchanged: RecentIds?, + deletionScope: Filter?, ) { val session = server.connect { json -> @@ -645,6 +681,9 @@ class MirrorWorker( val event = runCatching { (OptimizedJsonMapper.fromJsonToMessage(json) as? EventMessage)?.event } .getOrNull() ?: return@connect + // A kind-62 vanish only goes to a relay it targets; kind-5 always + // goes. Content events are already scoped by the session's REQ. + if (!shouldPropagateDeletionUp(event, up.url)) return@connect // BOTH: don't push back what we just pulled down. if (exchanged?.contains(event.id) == true) return@connect exchanged?.add(event.id) @@ -652,8 +691,9 @@ class MirrorWorker( sentUp.incrementAndGet() } upSessions += AutoCloseable { session.close() } + val reqFilters = listOfNotNull(scopedFilter, deletionScope) scope.launch { - session.receive(OptimizedJsonMapper.toJson(ReqCmd("geode-mirror-up", listOf(scopedFilter)))) + session.receive(OptimizedJsonMapper.toJson(ReqCmd("geode-mirror-up", reqFilters))) } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt new file mode 100644 index 0000000000..244f4916c9 --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt @@ -0,0 +1,194 @@ +/* + * 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 + +import com.vitorpamplona.geode.testing.RelayClientTest +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.HexKey +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.deletionSideChannelFilter +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.excludesDeletionKinds +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyPropagateDeletions +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.shouldPropagateDeletionUp +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * A NIP-77 sync reconciles by id over the content filter, so a scoped sync + * (`--kind 1`) never carries the kind-5/62 that would delete one of those notes + * — the deletion is stuck on whichever side issued it. The deletion side-channel + * ([negentropyPropagateDeletions]) closes that gap by reconciling kinds 5 & 62 + * on their own, both directions, independent of the content filter. + * + * Scenario under test (the user's "Relay A has a deletion, Relay B doesn't"): + * the note lives on both sides; a kind-5 deleting it lives on only one. After the + * side-channel runs, the deletion has reached the other side and the note is gone + * there too. + */ +class DeletionSyncTest : RelayClientTest() { + private val signer = NostrSignerSync(KeyPair()) + + private fun note(text: String): Event = signer.sign(TextNoteEvent.build(text)) + + private fun deletionOf(target: Event): Event = signer.sign(DeletionEvent.build(listOf(target), createdAt = target.createdAt + 1)) + + // ---- filter helpers (pure) -------------------------------------------- + + @Test + fun unscopedFilterNeedsNoSideChannel() { + assertFalse(Filter().excludesDeletionKinds(), "no kinds constraint already matches deletions") + assertFalse(Filter(kinds = listOf(1, 5)).excludesDeletionKinds(), "explicit kind 5 is covered") + assertFalse(Filter(kinds = listOf(62)).excludesDeletionKinds(), "explicit kind 62 is covered") + assertTrue(Filter(kinds = listOf(1)).excludesDeletionKinds(), "kind-1-only drops deletions") + } + + @Test + fun sideChannelFilterCarriesAuthorsNotWindow() { + val authored = Filter(kinds = listOf(1), authors = listOf("aa", "bb"), since = 100, until = 200) + val side = authored.deletionSideChannelFilter() + + assertEquals(listOf(DeletionEvent.KIND, RequestToVanishEvent.KIND), side.kinds) + assertEquals(listOf("aa", "bb"), side.authors, "author scope is inherited") + assertNull(side.since, "no time window: a deletion's created_at is not its target's") + assertNull(side.until) + } + + @Test + fun vanishGateHonorsDeclaredTargets() { + val here = defaultRelayUrl + val elsewhere = RelayUrlNormalizer.normalize("wss://elsewhere.example/") + + val delete = deletionOf(note("x")) + val vanishHere = signer.sign(RequestToVanishEvent.build(here)) + val vanishElsewhere = signer.sign(RequestToVanishEvent.build(elsewhere)) + val vanishEverywhere = signer.sign(RequestToVanishEvent.buildVanishFromEverywhere()) + + assertTrue(shouldPropagateDeletionUp(delete, elsewhere), "kind-5 is always safe to propagate") + assertTrue(shouldPropagateDeletionUp(vanishHere, here), "vanish targeting this relay goes") + assertFalse(shouldPropagateDeletionUp(vanishElsewhere, here), "vanish for another relay does not") + assertTrue(shouldPropagateDeletionUp(vanishEverywhere, here), "ALL_RELAYS vanish goes anywhere") + } + + @Test + fun noOpWhenFilterAlreadyCoversDeletions() = + runBlocking { + val result = + withTimeout(20_000) { + client.negentropyPropagateDeletions( + relay = defaultRelayUrl, + contentFilter = Filter(kinds = listOf(1, 5)), + localDeletions = listOf(deletionOf(note("x"))), + download = { error("must not download") }, + upload = { error("must not upload") }, + ) + } + assertNull(result, "a filter that already covers 5/62 skips the side-channel") + } + + // ---- up: local has the deletion, relay does not ----------------------- + + @Test + fun pushesLocalDeletionUpSoRelayRemovesTarget() = + runBlocking { + val note = note("delete me") + val deletion = deletionOf(note) + + // Relay B: has the note, no deletion. + defaultRelay.preload(listOf(note)) + assertEquals(1, defaultRelay.store.query(Filter(ids = listOf(note.id))).size) + + // Local side (Relay A) already applied the deletion, so it holds only + // the kind-5. A content sync over kind 1 would never carry it. + val uploaded = mutableListOf() + withTimeout(20_000) { + client.negentropyPropagateDeletions( + relay = defaultRelayUrl, + contentFilter = Filter(kinds = listOf(1)), + localDeletions = listOf(deletion), + download = { error("relay has no deletions to pull") }, + upload = { event -> + uploaded += event + defaultRelay.publish(event) + }, + ) + } + + assertEquals(listOf(deletion.id), uploaded.map { it.id }, "the deletion was pushed up") + assertTrue( + defaultRelay.store.query(Filter(ids = listOf(note.id))).isEmpty(), + "relay applied the pushed deletion and removed the note", + ) + } + + // ---- down: relay has the deletion, local does not --------------------- + + @Test + fun pullsRelayDeletionDownSoLocalRemovesTarget() = + runBlocking { + val note = note("delete me too") + val deletion = deletionOf(note) + + // Remote relay already applied the deletion → holds only the kind-5. + defaultRelay.preload(listOf(note, deletion)) + assertTrue( + defaultRelay.store.query(Filter(ids = listOf(note.id))).isEmpty(), + "precondition: relay removed the note when it ingested the deletion", + ) + + // Local side: a second store still holding the note, no deletion. + val localUrl = RelayUrlNormalizer.normalize("ws://local-a/") + val local = hub.getOrCreate(localUrl) + local.preload(listOf(note)) + assertEquals(1, local.store.query(Filter(ids = listOf(note.id))).size) + + withTimeout(20_000) { + client.negentropyPropagateDeletions( + relay = defaultRelayUrl, + contentFilter = Filter(kinds = listOf(1)), + localDeletions = emptyList(), + download = { ids: List -> + // Stand-in for REQ-by-id + verify + store: pull from the + // remote in-process store and ingest into the local one. + defaultRelay.store.query(Filter(ids = ids)).forEach { local.store.insert(it) } + }, + upload = { error("local has no deletions to push") }, + ) + } + + assertTrue( + local.store.query(Filter(ids = listOf(note.id))).isEmpty(), + "local store applied the pulled deletion and removed the note", + ) + } +} diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt new file mode 100644 index 0000000000..8b0be63d5e --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt @@ -0,0 +1,114 @@ +/* + * 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.KtorRelay +import com.vitorpamplona.geode.RelayEngine +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * A kind-scoped mirror (`filter = {kinds:[1]}`) must still propagate the kind-5/62 + * that delete those notes — otherwise the deletion is stuck upstream and the note + * lives forever on the mirror. [MirrorWorker]'s deletion side-channel reconciles + * kinds 5/62 on their own, in the mirror's configured direction, independent of + * the operator filter. + */ +class MirrorDeletionSyncTest { + private val upstreamStore = EventStore(null) + private val downstreamStore = EventStore(null) + + private val upstream = RelayEngine(url = "ws://127.0.0.1:7896/".normalizeRelayUrl(), store = upstreamStore) + private val downstream = + RelayEngine(url = "ws://127.0.0.1:7897/".normalizeRelayUrl(), store = downstreamStore, parallelVerify = true) + + private var server: KtorRelay? = null + private var worker: MirrorWorker? = null + + @AfterTest + fun tearDown() { + worker?.close() + server?.stop(gracePeriodMillis = 0, timeoutMillis = 1_000) + upstream.close() + downstream.close() + } + + @Test + fun scopedDownMirrorPropagatesDeletion() = + runBlocking { + val signer = NostrSignerSync(KeyPair()) + val note = signer.sign(TextNoteEvent.build("delete me")) + val deletion = signer.sign(DeletionEvent.build(listOf(note), createdAt = note.createdAt + 1)) + + // Upstream already applied the deletion → it holds only the kind-5. + upstreamStore.insert(note) + upstreamStore.insert(deletion) + assertEquals(0, upstreamStore.count(Filter(ids = listOf(note.id))), "upstream removed the note") + + // Downstream (the mirror) still holds the note, no deletion. + downstreamStore.insert(note) + assertEquals(1, downstreamStore.count(Filter(ids = listOf(note.id))), "mirror starts with the note") + + server = KtorRelay(upstream, host = "127.0.0.1", port = 7896).start() + + worker = + MirrorWorker( + upstreams = + listOf( + MirrorUpstream( + url = "ws://127.0.0.1:7896/".normalizeRelayUrl(), + trusted = true, + backfillSeconds = 86_400, + // Scoped to kind 1 — would drop the kind-5 without the side-channel. + filter = Filter(kinds = listOf(1)), + ), + ), + server = downstream.server, + store = downstreamStore, + negentropyBackfill = true, + ).also { it.start() } + + val gone = + withTimeoutOrNull(30_000) { + while (downstreamStore.count(Filter(ids = listOf(note.id))) > 0) delay(200) + true + } + + assertTrue(gone == true, "scoped down mirror did not propagate the deletion") + assertEquals( + 1, + downstreamStore.count(Filter(kinds = listOf(DeletionEvent.KIND))), + "mirror ingested the deletion event itself", + ) + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDeletionSyncExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDeletionSyncExt.kt new file mode 100644 index 0000000000..7e34191495 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDeletionSyncExt.kt @@ -0,0 +1,134 @@ +/* + * 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.quartz.nip01Core.relay.client.accessories + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent +import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent + +/** + * Event kinds that carry *deletion intent* — NIP-09 deletion requests (kind 5) + * and NIP-62 request-to-vanish (kind 62). They are propagation instructions, not + * content, so a sync should carry them **regardless of its content filter**: a + * `--kind 1` sync that dropped the kind-5 deleting one of those notes would leave + * the note un-deleted on the other side forever. + */ +val DELETION_PROPAGATION_KINDS = listOf(DeletionEvent.KIND, RequestToVanishEvent.KIND) + +/** + * Whether this content [Filter] would *exclude* the deletion kinds — i.e. it pins + * `kinds` and none of them is 5/62. A filter with no `kinds` constraint already + * matches deletions, so it needs no side-channel. Anything else (a scoped `kinds` + * list without 5/62) would silently drop deletions and wants + * [deletionSideChannelFilter]. + */ +fun Filter.excludesDeletionKinds(): Boolean { + val k = kinds ?: return false + return DELETION_PROPAGATION_KINDS.none { it in k } +} + +/** + * The companion "deletion side-channel" filter for a content [Filter]: kinds 5/62, + * scoped to the same `authors`. A kind-5/62 only affects its own author's events, so + * when the content sync is author-scoped the deletions worth reconciling are exactly + * those authors'. Carries **no** `since`/`until`: a deletion's `created_at` is when it + * was issued, not when its target was created, so inheriting the content window would + * drop a recent deletion of an old event (or an old deletion synced late). + */ +fun Filter.deletionSideChannelFilter(): Filter = Filter(kinds = DELETION_PROPAGATION_KINDS, authors = authors) + +/** + * Whether a local deletion-family [event] may be pushed UP to [relay]. + * + * - **kind 5** — always. A deletion request is owner-scoped (the relay only removes + * the deleting author's own events), so propagating it can never delete a third + * party's data; the worst case is a no-op the relay ignores. + * - **kind 62** — only when the request actually targets [relay] (its `relay` tags + * name that URL, or `ALL_RELAYS`). A vanish triggers a pubkey-wide mass delete on + * every relay that ingests it, so we must not fan one out to a relay the author + * never named — we honor the author's declared targets, no broader. + */ +fun shouldPropagateDeletionUp( + event: Event, + relay: NormalizedRelayUrl, +): Boolean = + when (event) { + is RequestToVanishEvent -> event.shouldVanishFrom(relay) + else -> true + } + +/** + * Propagates deletion-family events (kinds 5 & 62) between the local set and [relay], + * **independent of a content sync's [contentFilter]** and **always bidirectional**: + * + * - **down** — deletions the relay has and we lack are handed to [download]; feeding + * them into the local store lets NIP-09/62 remove the targets locally too (and its + * reject-trigger keeps them from being re-added by a later content sync). + * - **up** — deletions we have and the relay lacks are handed to [upload]; publishing + * them makes the relay apply the deletion. Kind-62 vanishes are gated by + * [shouldPropagateDeletionUp] so one is only sent to a relay it targets. + * + * A no-op returning `null` when [contentFilter] already covers kinds 5/62 (an + * unscoped or already-deletion-including sync carries them itself). Otherwise runs one + * extra [negentropyReconcile] over [deletionSideChannelFilter]; the set is tiny on any + * real store, so the cost is a single short reconcile, not a second full sync. + * + * The caller owns I/O: [download] fetches + ingests the given ids however it fetches + * content (REQ-by-id, verify, store), and [upload] publishes one local event. Both + * suspend the reconcile round that produced them, so the relay is back-pressured. + * + * @param localDeletions the local kind-5/62 events — both the reconcile set and the + * source the `have` ids resolve against for [upload]. + */ +suspend fun INostrClient.negentropyPropagateDeletions( + relay: NormalizedRelayUrl, + contentFilter: Filter, + localDeletions: List, + batchSize: Int = 500, + idleTimeoutMs: Long = 120_000L, + download: suspend (List) -> Unit, + upload: suspend (Event) -> Unit, +): NegentropyReconcileResult? { + if (!contentFilter.excludesDeletionKinds()) return null + + val byId = localDeletions.associateBy { it.id } + val localEntries = localDeletions.map { IdAndTime(it.createdAt, it.id) } + + return negentropyReconcile( + relay = relay, + filter = contentFilter.deletionSideChannelFilter(), + localEntries = localEntries, + batchSize = batchSize, + idleTimeoutMs = idleTimeoutMs, + onNeedIds = { batch -> download(batch) }, + onHaveIds = { batch -> + for (id in batch) { + val event = byId[id] ?: continue + if (shouldPropagateDeletionUp(event, relay)) upload(event) + } + }, + ) +}