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 09ef31b197..e4da7e9c99 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,11 +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 -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 @@ -40,7 +36,6 @@ 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 /** @@ -60,27 +55,13 @@ 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, in three phases, so both sides reflect each - * other. NIP-77 reconciles by id over the content filter, so a scoped sync - * (`--kind 1`) would never carry the kind-5 that deletes one of those notes — - * the deletion would be stuck on whichever side issued it. - * - * 1. Reconcile the missing deletion kinds — FIRST, before content, so every - * deletion is applied on both sides before the content diff is taken. The - * reconcile is **bounded to the authors we hold content for** (the filter's - * authors ∪ our local matched set's authors), never the relay's whole - * population — an author-less pull would otherwise import the relay's entire - * deletion history and mass-delete this local store. Skipped when we hold - * nothing in scope. Best-effort: a failure here never aborts the sync. - * 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 deletions and apply them locally. - * This is the down-convergence path for author-less syncs (phase 1 skipped). - * - * Kind-5 (precise, owner-scoped) propagates by default. Kind-62 Request-to-Vanish - * mass-deletes ALL of a pubkey's events, so it is opt-in via `--sync-vanish`. - * Pass `--no-sync-deletions` to disable deletion propagation entirely. + * Deletion propagation is deliberately narrow (on by default; disable with + * `--no-sync-deletions`): for each id the relay HAS that we LACK — the reconcile's + * need set — if we hold a **kind-5** deletion that targets it, that deletion is + * published up, so a note we deleted is deleted on the relay too instead of being + * re-downloaded. That is the whole feature: only the need ids, only kind-5, up + * only. Nothing is pulled down or applied locally, so it can never over-delete this + * store, and it needs no author scoping (the need set already bounds it). * * Both directions are pipelined with the reconcile: need-id batches feed * [DOWNLOAD_WORKERS] concurrent by-id REQ drains and have-ids feed a single @@ -121,86 +102,24 @@ object SyncCommand { // Default direction is download; --up adds upload. val up = args.bool("up") val down = args.bool("down") || !up + // Deletion propagation (on by default; --no-sync-deletions disables). Scope is + // exactly: for the ids the relay HAS that we LACK (the reconcile's need set), if + // we hold a kind-5 deletion targeting one of them, publish that deletion up so + // the relay deletes it too — instead of re-downloading a note we deleted. That + // is the whole feature: only these ids, only kind-5, up only. Nothing is pulled + // down or applied locally, so it can never over-delete this store. val syncDeletions = !args.bool("no-sync-deletions") - // Which deletion kinds the side-channel carries. Kind-5 (precise, owner-scoped) - // by default; kind-62 Request-to-Vanish is opt-in because it mass-deletes ALL of - // a pubkey's events, a blast radius that always exceeds a content sync's scope. - val deletionKinds = if (args.bool("sync-vanish")) DELETION_PROPAGATION_KINDS else listOf(DeletionEvent.KIND) val filter = RawEventSupport.buildFilter(args) Context.open(dataDir).use { ctx -> ctx.prepare() - - val deletionsDown = AtomicInteger(0) - val deletionsUp = AtomicInteger(0) - var deletionsError: String? = null - - // ── Phase 1: deletions first, both directions ──────────────────── - // Propagate deletions 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, resurrecting it on a relay - // that also lacks the deletion. - // - // SCOPE. The reconcile is bounded to the authors we actually hold content - // for (the content filter's authors ∪ the authors in our local matched - // set) — NEVER the relay's whole population. An author-less filter against a - // public relay would otherwise pull the relay's entire deletion history and - // apply it to this personal store, mass-deleting cached events far outside - // the sync's scope. When we hold nothing in scope there is nothing to - // reconcile, so the side-channel is skipped and Phase 3 covers the rest. - // Only compute the scope (a store read) when a side-channel could actually - // run — a whole-store sync already carries deletions and must not pay a - // full-store scan here. - val runSideChannel = syncDeletions && filter.excludesDeletionKinds(deletionKinds) - val scopeAuthors = - if (runSideChannel) { - ((filter.authors ?: emptyList()) + ctx.store.query(filter).map { it.pubKey }).distinct() - } else { - emptyList() - } - val deletionResult = - if (runSideChannel && scopeAuthors.isNotEmpty()) { - // Best-effort: a deletion-reconcile failure must NEVER abort the - // primary content sync (matches geode's mirror policy). Record it - // and fall through to content + the Phase-3 backstop. - try { - val localDeletions = ctx.store.query(filter.deletionSideChannelFilter(scopeAuthors, deletionKinds)) - ctx.client.negentropyPropagateDeletions( - relay = relay, - contentFilter = filter, - localDeletions = localDeletions, - scopeAuthors = scopeAuthors, - deletionKinds = deletionKinds, - 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) { - deletionsError = e.message ?: "deletion sync failed" - null - } - } 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 deletionsSent = AtomicInteger(0) val result = try { @@ -212,6 +131,8 @@ object SyncCommand { // Unbounded is fine here: have-ids reference events we already // hold locally, so memory is bounded by the local set. val haveBatches = Channel>(Channel.UNLIMITED) + // need-ids routed to the deletion sender (bounded → back-pressure). + val delBatches = Channel>(DOWNLOAD_WORKERS * 2) val downloaders = List(DOWNLOAD_WORKERS) { @@ -227,15 +148,24 @@ object SyncCommand { for (batch in haveBatches) { for (id in batch) { val ev = localById[id] ?: continue - val ack = ctx.publish(ev, setOf(relay)) - 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) - } + if (ctx.publish(ev, setOf(relay)).values.any { it }) uploaded.incrementAndGet() + } + } + } + // For each id the relay has that we lack, publish any local kind-5 + // deletion that targets it (queried by its `e` tag). A note we + // deleted then gets deleted on the relay too, instead of being + // re-downloaded. Most need-ids have no such deletion, so the query + // usually returns empty and nothing is sent. + val deletionSender = + launch { + for (batch in delBatches) { + val mine = + ctx.store.query( + Filter(kinds = listOf(DeletionEvent.KIND), tags = mapOf("e" to batch)), + ) + for (del in mine) { + if (ctx.publish(del, setOf(relay)).values.any { it }) deletionsSent.incrementAndGet() } } } @@ -250,36 +180,26 @@ object SyncCommand { idleTimeoutMs = timeoutMs, reconcileConcurrency = RECONCILE_CONCURRENCY, onHaveIds = if (up) { batch -> haveBatches.send(batch) } else null, - onNeedIds = { batch -> if (down) needBatches.send(batch) }, + onNeedIds = { batch -> + if (down) needBatches.send(batch) + if (syncDeletions) delBatches.send(batch) + }, ) } finally { needBatches.close() haveBatches.close() + delBatches.close() } downloaders.joinAll() uploader.join() + deletionSender.join() reconcile } } catch (e: NegentropySyncException) { return Output.error("sync_error", e.message ?: "negentropy sync failed") } - // ── 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 and - // precisely bounded — only fires on an actual rejection, only for the - // rejected authors, and verify-by-fetch: only a real deletion the store - // accepts has any effect. This is the primary down-convergence path for - // author-less syncs (where Phase 1 is intentionally skipped). - if (syncDeletions && blockedAuthors.isNotEmpty()) { - ctx.drain( - mapOf(relay to listOf(Filter(kinds = deletionKinds, authors = blockedAuthors.toList()))), - timeoutMs, - ) - } - Output.emit( mapOf( "relay" to relay.url, @@ -289,11 +209,7 @@ 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(), - "deletions_error" to (deletionsError ?: ""), + "deletions_sent" to deletionsSent.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 ae621a5415..62936d17fe 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt @@ -23,11 +23,8 @@ 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 @@ -242,24 +239,17 @@ 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 filtersFrom(candidate)), + filters = mapOf(up.url to listOf(scopedBase.copy(since = candidate))), listener = listener, ) sinceAdvances.incrementAndGet() @@ -340,15 +330,6 @@ 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 @@ -363,16 +344,16 @@ class MirrorWorker( val upLiveSince = if (catchUpUp) since else initialSince if (up.direction != MirrorDirection.UP) { - downSubs += startDown(i, up, scopedBase, downLiveSince, exchanged, deletionScope) + downSubs += startDown(i, up, scopedBase, downLiveSince, exchanged) } if (up.direction != MirrorDirection.DOWN) { - startUp(up, scopedBase.copy(since = upLiveSince), exchanged, deletionScope?.copy(since = upLiveSince)) + startUp(up, scopedBase.copy(since = upLiveSince), exchanged) } if (catchUpDown) { - scope.launch { runCatchUpDown(up, scopedBase, initialSince, since, deletionScope) } + scope.launch { runCatchUpDown(up, scopedBase, initialSince, since) } } if (catchUpUp) { - scope.launch { runCatchUpUp(up, scopedBase, initialSince, since, exchanged, deletionScope) } + scope.launch { runCatchUpUp(up, scopedBase, initialSince, since, exchanged) } } } client.connect() @@ -430,14 +411,12 @@ class MirrorWorker( scopedBase: Filter, initialSince: Long, until: Long, - deletionScope: Filter?, ) { val catchUpFilter = scopedBase.copy(since = initialSince, until = until) - - // 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 + // 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() // Bounded hand-off → one ingest consumer. `onEvent` can't suspend, so it // blocks here when the sink falls behind; because negentropySyncOrFetch's @@ -463,34 +442,26 @@ class MirrorWorker( } } - // 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() + try { val result = client.negentropySyncOrFetch( relay = up.url, - filter = filter, + filter = catchUpFilter, localEntries = localEntries, onEvent = { event -> - if (inScope(event)) handoff.trySendBlocking(event) else filtered.incrementAndGet() + // 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() + } }, ) 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 { - // 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) { @@ -518,80 +489,65 @@ class MirrorWorker( initialSince: Long, until: Long, exchanged: RecentIds?, - deletionScope: Filter?, ) { val localStore = store ?: return val catchUpFilter = scopedBase.copy(since = initialSince, until = until) - - // 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)) + // 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 { var round = 0 while (round < MAX_UP_SYNC_ROUNDS) { - // 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. + // Reconcile and PUBLISH the `have` ids (events we hold the + // upstream lacks) as each batch streams in — never materialising + // the full diff. On a large window the id list is millions of + // entries; the streaming reconcile keeps memory at one batch and + // still back-pressures the relay because publishing suspends the + // round. The `need` direction is the down catch-up's job, so its + // ids are discarded (not accumulated) here. var haveCount = 0 var pushed = 0 client.negentropyReconcile( relay = up.url, - filter = filter, + filter = catchUpFilter, localEntries = localEntries, batchSize = HAVE_FETCH_BATCH, onHaveIds = { batch -> haveCount += batch.size for (event in localStore.query(Filter(ids = batch))) { - 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). + // Scope containment: a scoped upstream only receives + // in-scope events. Echo suppression: record the id so + // a BOTH mirror doesn't re-ingest its own push on the + // down sub (but always re-publish — a straggler stays + // in `exchanged` yet still needs delivering). + if (up.filter != null && !up.filter.match(event)) continue exchanged?.add(event.id) client.publish(event, setOf(up.url)) pushed++ } - delay(UP_PUBLISH_PACING_MS) // pace the outbox + // Pace the outbox so a batch drains before the next. + delay(UP_PUBLISH_PACING_MS) }, onNeedIds = { }, ) - // Converge when nothing PUBLISHABLE remains to push — not on raw - // haveCount. A diff that is all un-publishable (e.g. a kind-62 vanish - // for a relay this upstream isn't, which shouldPropagateDeletionUp - // rightly refuses) would otherwise report a non-zero haveCount every - // round and burn all MAX_UP_SYNC_ROUNDS on every startup. - if (pushed == 0) { - val how = if (haveCount == 0) "converged" else "converged ($haveCount un-publishable left)" - Log.i("MirrorWorker") { "up catch-up ($label) to ${up.url.url}: $how after $round round(s)" } + if (haveCount == 0) { + Log.i("MirrorWorker") { "up catch-up 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 ($label) to ${up.url.url}: round $round pushed $pushed (had $haveCount to go)" } + Log.i("MirrorWorker") { "up catch-up to ${up.url.url}: round $round pushed $pushed (had $haveCount to go)" } round++ - delay(UP_SYNC_SETTLE_MS) // let the upstream ingest + OK before re-checking + // 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) } - 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 { - // 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) } + 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) { @@ -606,7 +562,6 @@ 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 @@ -638,7 +593,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) && deletionScope?.match(event) != true) { + if (up.filter != null && !up.filter.match(event)) { filtered.incrementAndGet() Log.d("MirrorWorker") { "out-of-scope from ${relay.url}: ${event.id}" } return @@ -661,13 +616,12 @@ 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 downSub.filtersFrom(initialSince)), + filters = mapOf(up.url to listOf(scopedBase.copy(since = initialSince))), listener = listener, ) - return downSub + return DownSub(subId, up, scopedBase, listener, initialSince, watermark) } /** @@ -684,7 +638,6 @@ class MirrorWorker( up: MirrorUpstream, scopedFilter: Filter, exchanged: RecentIds?, - deletionScope: Filter?, ) { val session = server.connect { json -> @@ -692,9 +645,6 @@ 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) @@ -702,9 +652,8 @@ class MirrorWorker( sentUp.incrementAndGet() } upSessions += AutoCloseable { session.close() } - val reqFilters = listOfNotNull(scopedFilter, deletionScope) scope.launch { - session.receive(OptimizedJsonMapper.toJson(ReqCmd("geode-mirror-up", reqFilters))) + session.receive(OptimizedJsonMapper.toJson(ReqCmd("geode-mirror-up", listOf(scopedFilter)))) } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt index e510aaf801..891a4426a9 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt @@ -24,37 +24,30 @@ 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.client.accessories.negentropyReconcileIds 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.nip01Core.store.IdAndTime 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. + * The `amy sync` deletion rule, end-to-end: for the ids the relay HAS that we LACK + * (the negentropy need set), if we hold a kind-5 deletion targeting one of them, + * publish that deletion up so the relay deletes it too — instead of re-downloading a + * note we deleted. Only these ids, only kind-5, up only; nothing is pulled down or + * applied locally, so the personal store can never be over-deleted. * - * 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. + * This exercises the exact wiring `SyncCommand` uses (reconcile → look up the local + * kind-5 by its `e` tag for the need ids → publish), which the CLI itself has no test + * harness for. */ class DeletionSyncTest : RelayClientTest() { private val signer = NostrSignerSync(KeyPair()) @@ -63,159 +56,74 @@ class DeletionSyncTest : RelayClientTest() { private fun deletionOf(target: Event): Event = signer.sign(DeletionEvent.build(listOf(target), createdAt = target.createdAt + 1)) - // ---- filter helpers (pure) -------------------------------------------- - - @Test - fun excludesDeletionKindsChecksEachKindIndependently() { - // No kinds constraint already matches every deletion kind. - assertFalse(Filter().excludesDeletionKinds()) - // Listing ONE deletion kind does NOT cover the other — the reconcile - // carries only the kinds actually listed. - assertTrue(Filter(kinds = listOf(1, 5)).excludesDeletionKinds(), "kind 5 listed, kind 62 still missing") - assertTrue(Filter(kinds = listOf(62)).excludesDeletionKinds(), "kind 62 listed, kind 5 still missing") - assertTrue(Filter(kinds = listOf(1)).excludesDeletionKinds(), "kind-1-only misses both") - // Both listed → nothing missing. - assertFalse(Filter(kinds = listOf(5, 62)).excludesDeletionKinds(), "both deletion kinds covered") - // With a restricted deletionKinds set, only kind 5 matters. - assertFalse(Filter(kinds = listOf(1, 5)).excludesDeletionKinds(listOf(DeletionEvent.KIND))) - } - - @Test - fun sideChannelFilterCarriesMissingKindsScopedAuthorsNoWindow() { - val authored = Filter(kinds = listOf(1, 5), authors = listOf("aa", "bb"), since = 100, until = 200) - val side = authored.deletionSideChannelFilter(authors = listOf("aa", "bb")) - - assertEquals(listOf(RequestToVanishEvent.KIND), side.kinds, "kind 5 already covered → only 62 missing") - assertEquals(listOf("aa", "bb"), side.authors, "explicit author scope") - 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 noOpWhenAllKindsCoveredOrScopeEmpty() = - runBlocking { - val d = deletionOf(note("x")) - // All deletion kinds already covered by content → skip. - assertNull( - withTimeout(20_000) { - client.negentropyPropagateDeletions( - relay = defaultRelayUrl, - contentFilter = Filter(kinds = listOf(5, 62)), - localDeletions = listOf(d), - scopeAuthors = listOf(signer.pubKey), - download = { error("must not download") }, - upload = { error("must not upload") }, - ) - }, - "a filter that already covers 5 AND 62 skips the side-channel", - ) - // Author-less scope → skip rather than reconcile the relay's whole history. - assertNull( - withTimeout(20_000) { - client.negentropyPropagateDeletions( - relay = defaultRelayUrl, - contentFilter = Filter(kinds = listOf(1)), - localDeletions = listOf(d), - scopeAuthors = emptyList(), - download = { error("must not download") }, - upload = { error("must not upload") }, - ) - }, - "an empty author scope disables the side-channel (no relay-wide pull)", - ) + /** The SyncCommand step under test: publish local kind-5 deletions targeting [needIds]. */ + private suspend fun sendDeletionsFor( + local: com.vitorpamplona.geode.RelayEngine, + needIds: List, + ): Int { + var sent = 0 + val mine = local.store.query(Filter(kinds = listOf(DeletionEvent.KIND), tags = mapOf("e" to needIds))) + for (del in mine) { + defaultRelay.publish(del) + sent++ } - - // ---- up: local has the deletion, relay does not ----------------------- + return sent + } @Test - fun pushesLocalDeletionUpSoRelayRemovesTarget() = + fun sendsDeletionForANeedIdWeDeleted() = runBlocking { val note = note("delete me") val deletion = deletionOf(note) - // Relay B: has the note, no deletion. + // Relay 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), - scopeAuthors = listOf(signer.pubKey), - download = { error("relay has no deletions to pull") }, - upload = { event -> - uploaded += event - defaultRelay.publish(event) - }, - ) - } + // Local already applied the deletion → it holds only the kind-5, so the + // note is a "need" (relay has it, we lack it). + val local = hub.getOrCreate(RelayUrlNormalizer.normalize("ws://local/")) + local.preload(listOf(note, deletion)) + assertTrue(local.store.query(Filter(ids = listOf(note.id))).isEmpty(), "local deleted the note") - assertEquals(listOf(deletion.id), uploaded.map { it.id }, "the deletion was pushed up") + val localKind1 = local.store.query(Filter(kinds = listOf(1))).map { IdAndTime(it.createdAt, it.id) } + val diff = + withTimeout(20_000) { + client.negentropyReconcileIds(relay = defaultRelayUrl, filter = Filter(kinds = listOf(1)), localEntries = localKind1) + } + assertEquals(setOf(note.id), diff.needIds.toSet(), "the deleted note is the only need id") + + val sent = sendDeletionsFor(local, diff.needIds) + + assertEquals(1, sent, "the deletion targeting the need id was sent") 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() = + fun sendsNothingForANeedIdWeNeverHad() = runBlocking { - val note = note("delete me too") - val deletion = deletionOf(note) + // Relay has a note we simply never had and never deleted — a plain download, + // no deletion to send. + val other = note("just never had this") + defaultRelay.preload(listOf(other)) - // 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", - ) + val local = hub.getOrCreate(RelayUrlNormalizer.normalize("ws://local2/")) + // Local holds an unrelated deletion (targets a different note) — must NOT be + // sent for `other`. + local.preload(listOf(deletionOf(note("unrelated")))) - // 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) + val diff = + withTimeout(20_000) { + client.negentropyReconcileIds(relay = defaultRelayUrl, filter = Filter(kinds = listOf(1)), localEntries = emptyList()) + } + assertTrue(other.id in diff.needIds, "the note is a need id") - withTimeout(20_000) { - client.negentropyPropagateDeletions( - relay = defaultRelayUrl, - contentFilter = Filter(kinds = listOf(1)), - localDeletions = emptyList(), - scopeAuthors = listOf(signer.pubKey), - 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") }, - ) - } + val sent = sendDeletionsFor(local, diff.needIds) - assertTrue( - local.store.query(Filter(ids = listOf(note.id))).isEmpty(), - "local store applied the pulled deletion and removed the note", - ) + assertEquals(0, sent, "no deletion targets the need id, so nothing is sent") + assertEquals(1, defaultRelay.store.query(Filter(ids = listOf(other.id))).size, "the note is untouched on the relay") } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt deleted file mode 100644 index 7d37ee8389..0000000000 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt +++ /dev/null @@ -1,168 +0,0 @@ -/* - * 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", - ) - } - - /** - * 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", - ) - } -} 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 deleted file mode 100644 index dfe345721a..0000000000 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDeletionSyncExt.kt +++ /dev/null @@ -1,160 +0,0 @@ -/* - * 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) - -/** - * The deletion kinds in [deletionKinds] that this content [Filter] does NOT already - * carry — i.e. the ones a side-channel still needs to reconcile. Empty when the filter - * has no `kinds` constraint (it already matches every deletion kind) or already lists - * all of them. NIP-77 reconciles strictly by the filter's `kinds`, so listing kind 5 - * does NOT cover kind 62: each is checked independently. - */ -fun Filter.missingDeletionKinds(deletionKinds: List = DELETION_PROPAGATION_KINDS): List { - val k = kinds ?: return emptyList() - return deletionKinds.filter { it !in k } -} - -/** - * Whether the content [Filter] would *exclude* at least one [deletionKinds], so a - * side-channel is needed. A filter with no `kinds` constraint matches every deletion - * kind (returns false); a scoped filter that omits kind 5 and/or 62 returns true. - */ -fun Filter.excludesDeletionKinds(deletionKinds: List = DELETION_PROPAGATION_KINDS): Boolean = missingDeletionKinds(deletionKinds).isNotEmpty() - -/** - * The companion "deletion side-channel" filter for a content [Filter]: the deletion - * kinds the filter doesn't already carry, scoped to [authors]. A kind-5/62 only affects - * its own author's events, so the deletions worth reconciling are exactly those - * authors' — and [authors] MUST be bounded (the content filter's authors, or, for a - * personal store, the authors actually held locally). An empty/`null` [authors] here - * means "every author on the relay", which for a personal store would pull the relay's - * ENTIRE deletion history and apply it locally — callers must not do that. - * - * 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( - authors: List? = this.authors, - deletionKinds: List = DELETION_PROPAGATION_KINDS, -): Filter = Filter(kinds = missingDeletionKinds(deletionKinds), 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 every [deletionKinds], - * or when [scopeAuthors] is empty (nothing to scope — an unscopeable deletion set must - * not be reconciled, or it would pull the relay's entire deletion history). - * - * **Scope is the caller's responsibility.** [scopeAuthors] bounds the reconcile — it - * MUST be a bounded author set (the content filter's authors, or the authors actually - * held locally). It defaults to `contentFilter.authors`, so an author-less content - * filter yields an EMPTY scope → this returns null rather than reconcile everything. - * - * 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 deletion events (in [deletionKinds]) — both the - * reconcile set and the source the `have` ids resolve against for [upload]. - * @param scopeAuthors the authors to bound the deletion reconcile to. Empty/`null` - * disables the side-channel (see above). - * @param deletionKinds which deletion kinds to propagate (default 5 & 62). Pass `[5]` - * to propagate only precise NIP-09 deletions and skip account-wide NIP-62 vanishes. - */ -suspend fun INostrClient.negentropyPropagateDeletions( - relay: NormalizedRelayUrl, - contentFilter: Filter, - localDeletions: List, - scopeAuthors: List? = contentFilter.authors, - deletionKinds: List = DELETION_PROPAGATION_KINDS, - batchSize: Int = 500, - idleTimeoutMs: Long = 120_000L, - download: suspend (List) -> Unit, - upload: suspend (Event) -> Unit, -): NegentropyReconcileResult? { - if (scopeAuthors.isNullOrEmpty()) return null - if (!contentFilter.excludesDeletionKinds(deletionKinds)) return null - - val byId = localDeletions.associateBy { it.id } - val localEntries = localDeletions.map { IdAndTime(it.createdAt, it.id) } - - return negentropyReconcile( - relay = relay, - filter = contentFilter.deletionSideChannelFilter(authors = scopeAuthors, deletionKinds = deletionKinds), - 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) - } - }, - ) -}