feat: propagate deletions (NIP-09/62) across negentropy sync

NIP-77 reconciles by event id over the content filter, so a scoped sync (`--kind 1`) never carries the kind-5/62 that deletes one of those notes: the deletion stays stuck on whichever side issued it while the target lives on forever on the other. Add a deletion side-channel that reconciles kinds 5 & 62 on their own, independent of the content filter. quartz: NostrClientDeletionSyncExt — DELETION_PROPAGATION_KINDS, Filter.excludesDeletionKinds()/deletionSideChannelFilter() (kinds 5/62 scoped to the same authors, no time window since a deletion's created_at is not its target's), shouldPropagateDeletionUp() (kind-5 always; kind-62 only to a relay it targets, honoring the vanish's declared relays), and negentropyPropagateDeletions() — one bidirectional reconcile that streams have→upload and need→download. amy sync: run the side-channel bidirectionally regardless of --up/--down whenever the filter excludes 5/62; emits deletions_{need,have,downloaded, uploaded}; --no-sync-deletions opts out. geode MirrorWorker: thread a per-upstream deletionScope through both catch-up phases and both live subs (down + up), in the mirror's configured direction; relax down containment to accept in-scope deletions, gate kind-62 pushes by target relay, and carry the deletion filter on re-subscribe so a reconnect never drops it. Tests: DeletionSyncTest (up/down propagation + filter/vanish-gate units) and MirrorDeletionSyncTest (a kind-scoped down mirror still removes the note). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JgL1WTV4Hkp2uuXcUHCHGt
2026-08-12 09:13:23 +00:00 · 2026-07-08 00:00:24 +00:00
parent 16357b0a9f
commit 683f993c7b
5 changed files with 582 additions and 54 deletions
@@ -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:<same>}` 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<Event>(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
@@ -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<Filter> = 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<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
+                            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)))
        }
    }

@@ -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<Event>(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<Event>()
+            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<Event>(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<Event>(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<Event>(Filter(ids = listOf(note.id))).size)
+
+            withTimeout(20_000) {
+                client.negentropyPropagateDeletions(
+                    relay = defaultRelayUrl,
+                    contentFilter = Filter(kinds = listOf(1)),
+                    localDeletions = emptyList(),
+                    download = { ids: List<HexKey> ->
+                        // Stand-in for REQ-by-id + verify + store: pull from the
+                        // remote in-process store and ingest into the local one.
+                        defaultRelay.store.query<Event>(Filter(ids = ids)).forEach { local.store.insert(it) }
+                    },
+                    upload = { error("local has no deletions to push") },
+                )
+            }
+
+            assertTrue(
+                local.store.query<Event>(Filter(ids = listOf(note.id))).isEmpty(),
+                "local store applied the pulled deletion and removed the note",
+            )
+        }
+}
@@ -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",
+            )
+        }
+}
@@ -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<Event>,
+    batchSize: Int = 500,
+    idleTimeoutMs: Long = 120_000L,
+    download: suspend (List<HexKey>) -> 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)
+            }
+        },
+    )
+}