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

refactor: reduce deletion sync to "send deletions for need ids", nothing else

Browse Source 
Per the actual requirement, deletion propagation is exactly: 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 a note we deleted is
deleted on the relay too instead of being re-downloaded. Only the need ids,
only kind-5, up only.

This removes all the machinery the earlier approach accreted and that the audit
flagged as over-broad / data-loss-prone:
- deleted NostrClientDeletionSyncExt (the bidirectional side-channel, author
  scoping, vanish gating, kind selection);
- reverted geode MirrorWorker to base (no deletion side-channel, live-sub
  changes, catch-up ordering, or convergence changes);
- dropped the 3-phase SyncCommand flow (deletions-first pull, author-scope
  derivation, reject-reaction backstop, --sync-vanish, deletions_* output).

The new path pulls nothing down and applies nothing locally, so it cannot
over-delete the store, and it needs no author scoping — the need set already
bounds it. Kind-62 is intentionally excluded: a vanish is not "of an id".

Emits deletions_sent. DeletionSyncTest now exercises the exact wiring
(reconcile → look up local kind-5 by its e tag for the need ids → publish),
including the negative case (a need id we never had sends nothing).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JgL1WTV4Hkp2uuXcUHCHGt
This commit is contained in:
Claude
2026-07-08 02:29:52 +00:00
parent 17687e43fc
commit e09a939b08
5 changed files with 152 additions and 707 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt
+41 -125
View File
@@ -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<Event>(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<Event>(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<Event>(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<HexKey>()
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<List<HexKey>>(Channel.UNLIMITED)
// need-ids routed to the deletion sender (bounded → back-pressure).
val delBatches = Channel<List<HexKey>>(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<Event>(
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
geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt
+55 -106
View File
@@ -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<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 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<Event>(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))))
}
}
geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt
+56 -148
View File
@@ -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<String>,
): Int {
var sent = 0
val mine = local.store.query<Event>(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<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),
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<Event>(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<Event>(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<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() =
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<Event>(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<Event>(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<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") },
)
}
val sent = sendDeletionsFor(local, diff.needIds)
assertTrue(
local.store.query<Event>(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<Event>(Filter(ids = listOf(other.id))).size, "the note is untouched on the relay")
}
}
geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorDeletionSyncTest.kt
-168
View File
@@ -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",
)
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDeletionSyncExt.kt
-160
View File
@@ -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<Int> = DELETION_PROPAGATION_KINDS): List<Int> {
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<Int> = 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<HexKey>? = this.authors,
deletionKinds: List<Int> = 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<Event>,
scopeAuthors: List<HexKey>? = contentFilter.authors,
deletionKinds: List<Int> = DELETION_PROPAGATION_KINDS,
batchSize: Int = 500,
idleTimeoutMs: Long = 120_000L,
download: suspend (List<HexKey>) -> 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)
}
},
)
}