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 c7ad0ce781..3410a33b4d 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 @@ -26,6 +26,7 @@ import com.vitorpamplona.amethyst.cli.DataDir 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.DELETION_PROPAGATION_KINDS 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 @@ -38,6 +39,7 @@ import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicInteger /** @@ -57,15 +59,24 @@ 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. + * Deletions ride a side-channel, and run in three phases so both sides fully + * reflect each other. 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. + * + * 1. Whenever the filter pins `kinds` and excludes 5/62, reconcile + * `{kinds:[5,62], authors:}` **both directions regardless of + * --up/--down** ([negentropyPropagateDeletions]) — FIRST, before content, + * so every deletion is applied on both sides before the content diff is + * taken. Local deletions are pushed up (a kind-62 vanish only to a relay it + * targets); the relay's deletions are pulled down and applied locally. + * 2. Content reconcile, over a local snapshot taken AFTER phase 1 so it never + * re-offers (and cannot resurrect on the relay) an event just deleted. + * 3. Backstop: if the relay rejected a content push — usually because it holds + * a deletion we lack — pull that author's kind-5/62 and apply it locally so + * we stop re-offering the dead event. + * + * Pass `--no-sync-deletions` to opt out of all three. * * Both directions are pipelined with the reconcile: need-id batches feed * [DOWNLOAD_WORKERS] concurrent by-id REQ drains and have-ids feed a single @@ -111,12 +122,53 @@ object SyncCommand { Context.open(dataDir).use { ctx -> ctx.prepare() + + val deletionsDown = AtomicInteger(0) + val deletionsUp = AtomicInteger(0) + + // ── Phase 1: deletions first, both directions ──────────────────── + // Propagate kind 5/62 before content so both stores have applied every + // deletion by the time content is diffed. Ordering is load-bearing: a + // deletion pulled down here removes a local event, so the content + // snapshot MUST be taken AFTER this phase — a snapshot taken before + // would still list the just-deleted event and re-offer it up, which the + // relay would either reject or (if it also lacks the deletion) resurrect. + // No-op (returns null) when the content filter already covers 5/62. + 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 + } + + // ── Phase 2: content ───────────────────────────────────────────── + // Snapshot the local set AFTER the deletion phase so it reflects any + // deletion just applied — never re-offering an event we just deleted. val localEvents = ctx.store.query(filter) val localById = localEvents.associateBy { it.id } val localEntries = localEvents.map { IdAndTime(it.createdAt, it.id) } val downloaded = AtomicInteger(0) val uploaded = AtomicInteger(0) + // Authors whose content push the relay rejected — a rejection usually + // means the relay holds a deletion we lack (Phase 3 reconciles them). + val blockedAuthors = ConcurrentHashMap.newKeySet() val result = try { @@ -144,7 +196,14 @@ object SyncCommand { for (id in batch) { val ev = localById[id] ?: continue val ack = ctx.publish(ev, setOf(relay)) - if (ack.values.any { it }) uploaded.incrementAndGet() + if (ack.values.any { it }) { + uploaded.incrementAndGet() + } else if (syncDeletions) { + // Relay refused it — most often because it holds a + // deletion for this id that we lack. Remember the + // author so Phase 3 can pull that deletion down. + blockedAuthors.add(ev.pubKey) + } } } } @@ -174,33 +233,19 @@ 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 - } + // ── Phase 3: reject-reaction backstop ──────────────────────────── + // A content push the relay blocked usually means the relay deleted that + // id and holds the deletion we lack. Pull that author's deletions and + // ingest them locally so we stop re-offering the dead event. Cheap — + // only fires on an actual rejection, and verify-by-fetch: only a real + // kind-5/62 that the store accepts has any effect. Rarely triggers once + // Phase 1 ran, but it is the net when deletions are otherwise skipped. + if (blockedAuthors.isNotEmpty()) { + ctx.drain( + mapOf(relay to listOf(Filter(kinds = DELETION_PROPAGATION_KINDS, authors = blockedAuthors.toList()))), + timeoutMs, + ) + } Output.emit( mapOf( 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 8abf263d95..2155353ed7 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt @@ -484,11 +484,13 @@ class MirrorWorker( } 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. + // Deletions first, so a deletion already lands (or the reject-trigger + // is armed) before the content pull can add its target — no + // add-then-delete churn. They 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, not the window. if (deletionScope != null) pull(deletionScope) + pull(catchUpFilter) } catch (e: CancellationException) { throw e } catch (e: Throwable) { @@ -577,10 +579,13 @@ class MirrorWorker( } try { - pushUp(catchUpFilter, "content") { event -> up.filter == null || up.filter.match(event) } + // Deletions first: push a deletion up before its target, so the + // upstream's reject-trigger blocks the target instead of ingesting + // then deleting it. if (deletionScope != null) { pushUp(deletionScope, "deletions") { event -> shouldPropagateDeletionUp(event, up.url) } } + pushUp(catchUpFilter, "content") { event -> up.filter == null || up.filter.match(event) } } catch (e: CancellationException) { throw e } catch (e: Throwable) { diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt index 8b0be63d5e..7d37ee8389 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt @@ -111,4 +111,58 @@ class MirrorDeletionSyncTest { "mirror ingested the deletion event itself", ) } + + /** + * The authoritative-push case: the local relay holds the deletion (its note + * already removed), the remote still holds the note, and a scoped `dir = up` + * mirror must push the kind-5 up so the remote reflects the local state. + */ + @Test + fun scopedUpMirrorPushesDeletion() = + 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)) + + // Local (downstream) is authoritative: it already applied the deletion. + downstreamStore.insert(note) + downstreamStore.insert(deletion) + assertEquals(0, downstreamStore.count(Filter(ids = listOf(note.id))), "local removed its note") + + // Remote (upstream, the sink geode dials) still holds the note. + upstreamStore.insert(note) + assertEquals(1, upstreamStore.count(Filter(ids = listOf(note.id))), "remote still has 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 = false, + backfillSeconds = 86_400, + direction = MirrorDirection.UP, + filter = Filter(kinds = listOf(1)), + ), + ), + server = downstream.server, + store = downstreamStore, + negentropyBackfill = true, + ).also { it.start() } + + val gone = + withTimeoutOrNull(30_000) { + while (upstreamStore.count(Filter(ids = listOf(note.id))) > 0) delay(200) + true + } + + assertTrue(gone == true, "scoped up mirror did not push the deletion to the remote") + assertEquals( + 1, + upstreamStore.count(Filter(kinds = listOf(DeletionEvent.KIND))), + "remote received the deletion event", + ) + } }