mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-09 08:04:45 +00:00
feat: deletions-first ordering + reject-reaction backstop in sync
Reorder amy sync so both sides fully reflect each other, including deletions: 1. Deletion side-channel now runs FIRST, before content, in both directions. The content snapshot is taken AFTER it, closing a resurrection bug: a deletion pulled down mid-sync removes a local event, but the old top-of-run snapshot still listed it and would re-offer it up — resurrecting it on a relay that also lacked the deletion. 2. Content reconcile, over the post-deletion snapshot. 3. Reject-reaction backstop: when the relay blocks a content push (usually it holds a deletion we lack), pull that author's kind-5/62 and ingest locally so we stop re-offering the dead event. Verify-by-fetch — only a real deletion the store accepts has any effect; fires only on an actual reject. The up-push of deletions is already verified per-event: ctx.publish awaits the relay's OK, and ingesting the kind-5 runs the delete synchronously, so OK=true confirms the remote applied it. Mirror catch-up gets the same deletions-first ordering (down and up), so a deletion lands, or the reject-trigger is armed, before its target — no add-then-delete churn. Tests: MirrorDeletionSyncTest gains scopedUpMirrorPushesDeletion (authoritative push — local holds the deletion, remote drops the note). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JgL1WTV4Hkp2uuXcUHCHGt
This commit is contained in:
Claude
@@ -26,6 +26,7 @@ import com.vitorpamplona.amethyst.cli.DataDir
|
||||
import com.vitorpamplona.amethyst.cli.Output
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.DELETION_PROPAGATION_KINDS
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.deletionSideChannelFilter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.excludesDeletionKinds
|
||||
@@ -38,6 +39,7 @@ import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.joinAll
|
||||
import kotlinx.coroutines.launch
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
/**
|
||||
@@ -57,15 +59,24 @@ import java.util.concurrent.atomic.AtomicInteger
|
||||
* Pass both for a full bidirectional sync. The filter flags are the same as
|
||||
* `fetch`/`subscribe`; an empty filter reconciles the whole store.
|
||||
*
|
||||
* Deletions ride a side-channel. NIP-77 reconciles by id over the content
|
||||
* filter, so a scoped sync (`--kind 1`) would never carry the kind-5/62 that
|
||||
* deletes one of those notes — the deletion would be stuck on whichever side
|
||||
* issued it. So whenever the filter pins `kinds` and excludes 5/62, a second
|
||||
* reconcile over `{kinds:[5,62], authors:<same>}` runs **both directions
|
||||
* regardless of --up/--down** ([negentropyPropagateDeletions]): local deletions
|
||||
* are pushed up so the relay applies them, and the relay's deletions are pulled
|
||||
* down so the local store applies them. A kind-62 vanish is only pushed to a
|
||||
* relay it actually targets. Pass `--no-sync-deletions` to opt out.
|
||||
* Deletions ride a side-channel, and run in three phases so both sides fully
|
||||
* reflect each other. NIP-77 reconciles by id over the content filter, so a
|
||||
* scoped sync (`--kind 1`) would never carry the kind-5/62 that deletes one of
|
||||
* those notes — the deletion would be stuck on whichever side issued it.
|
||||
*
|
||||
* 1. Whenever the filter pins `kinds` and excludes 5/62, reconcile
|
||||
* `{kinds:[5,62], authors:<same>}` **both directions regardless of
|
||||
* --up/--down** ([negentropyPropagateDeletions]) — FIRST, before content,
|
||||
* so every deletion is applied on both sides before the content diff is
|
||||
* taken. Local deletions are pushed up (a kind-62 vanish only to a relay it
|
||||
* targets); the relay's deletions are pulled down and applied locally.
|
||||
* 2. Content reconcile, over a local snapshot taken AFTER phase 1 so it never
|
||||
* re-offers (and cannot resurrect on the relay) an event just deleted.
|
||||
* 3. Backstop: if the relay rejected a content push — usually because it holds
|
||||
* a deletion we lack — pull that author's kind-5/62 and apply it locally so
|
||||
* we stop re-offering the dead event.
|
||||
*
|
||||
* Pass `--no-sync-deletions` to opt out of all three.
|
||||
*
|
||||
* Both directions are pipelined with the reconcile: need-id batches feed
|
||||
* [DOWNLOAD_WORKERS] concurrent by-id REQ drains and have-ids feed a single
|
||||
@@ -111,12 +122,53 @@ object SyncCommand {
|
||||
|
||||
Context.open(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
|
||||
val deletionsDown = AtomicInteger(0)
|
||||
val deletionsUp = AtomicInteger(0)
|
||||
|
||||
// ── Phase 1: deletions first, both directions ────────────────────
|
||||
// Propagate kind 5/62 before content so both stores have applied every
|
||||
// deletion by the time content is diffed. Ordering is load-bearing: a
|
||||
// deletion pulled down here removes a local event, so the content
|
||||
// snapshot MUST be taken AFTER this phase — a snapshot taken before
|
||||
// would still list the just-deleted event and re-offer it up, which the
|
||||
// relay would either reject or (if it also lacks the deletion) resurrect.
|
||||
// No-op (returns null) when the content filter already covers 5/62.
|
||||
val deletionResult =
|
||||
if (syncDeletions && filter.excludesDeletionKinds()) {
|
||||
try {
|
||||
val localDeletions = ctx.store.query<Event>(filter.deletionSideChannelFilter())
|
||||
ctx.client.negentropyPropagateDeletions(
|
||||
relay = relay,
|
||||
contentFilter = filter,
|
||||
localDeletions = localDeletions,
|
||||
idleTimeoutMs = timeoutMs,
|
||||
download = { batch ->
|
||||
deletionsDown.addAndGet(ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs).size)
|
||||
},
|
||||
upload = { event ->
|
||||
if (ctx.publish(event, setOf(relay)).values.any { it }) deletionsUp.incrementAndGet()
|
||||
},
|
||||
)
|
||||
} catch (e: NegentropySyncException) {
|
||||
return Output.error("sync_error", e.message ?: "deletion sync failed")
|
||||
}
|
||||
} else {
|
||||
null
|
||||
}
|
||||
|
||||
// ── 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 result =
|
||||
try {
|
||||
@@ -144,7 +196,14 @@ object SyncCommand {
|
||||
for (id in batch) {
|
||||
val ev = localById[id] ?: continue
|
||||
val ack = ctx.publish(ev, setOf(relay))
|
||||
if (ack.values.any { it }) uploaded.incrementAndGet()
|
||||
if (ack.values.any { it }) {
|
||||
uploaded.incrementAndGet()
|
||||
} else if (syncDeletions) {
|
||||
// Relay refused it — most often because it holds a
|
||||
// deletion for this id that we lack. Remember the
|
||||
// author so Phase 3 can pull that deletion down.
|
||||
blockedAuthors.add(ev.pubKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -174,33 +233,19 @@ object SyncCommand {
|
||||
return Output.error("sync_error", e.message ?: "negentropy sync failed")
|
||||
}
|
||||
|
||||
// Deletion side-channel: propagate kind 5/62 both ways, independent of
|
||||
// the content filter, so a scoped sync still carries the deletions that
|
||||
// apply to it. No-op (returns null) when the filter already covers 5/62.
|
||||
val deletionsDown = AtomicInteger(0)
|
||||
val deletionsUp = AtomicInteger(0)
|
||||
val deletionResult =
|
||||
if (syncDeletions && filter.excludesDeletionKinds()) {
|
||||
try {
|
||||
val localDeletions = ctx.store.query<Event>(filter.deletionSideChannelFilter())
|
||||
ctx.client.negentropyPropagateDeletions(
|
||||
relay = relay,
|
||||
contentFilter = filter,
|
||||
localDeletions = localDeletions,
|
||||
idleTimeoutMs = timeoutMs,
|
||||
download = { batch ->
|
||||
deletionsDown.addAndGet(ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs).size)
|
||||
},
|
||||
upload = { event ->
|
||||
if (ctx.publish(event, setOf(relay)).values.any { it }) deletionsUp.incrementAndGet()
|
||||
},
|
||||
)
|
||||
} catch (e: NegentropySyncException) {
|
||||
return Output.error("sync_error", e.message ?: "deletion sync failed")
|
||||
}
|
||||
} else {
|
||||
null
|
||||
}
|
||||
// ── Phase 3: reject-reaction backstop ────────────────────────────
|
||||
// A content push the relay blocked usually means the relay deleted that
|
||||
// id and holds the deletion we lack. Pull that author's deletions and
|
||||
// ingest them locally so we stop re-offering the dead event. Cheap —
|
||||
// only fires on an actual rejection, and verify-by-fetch: only a real
|
||||
// kind-5/62 that the store accepts has any effect. Rarely triggers once
|
||||
// Phase 1 ran, but it is the net when deletions are otherwise skipped.
|
||||
if (blockedAuthors.isNotEmpty()) {
|
||||
ctx.drain(
|
||||
mapOf(relay to listOf(Filter(kinds = DELETION_PROPAGATION_KINDS, authors = blockedAuthors.toList()))),
|
||||
timeoutMs,
|
||||
)
|
||||
}
|
||||
|
||||
Output.emit(
|
||||
mapOf(
|
||||
|
||||
Reference in New Issue
Block a user