diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 2a019ff766..0d3293a869 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -160,6 +160,17 @@ Summarize the survey in your plan: for each component, note whether it's reused as-is, extracted from `amethyst/` to `commons/`, genuinely new (platform-specific only), or a duplicate of an existing pattern to avoid. +**Relay client ops already exist — don't hand-roll subscribe/REQ/publish loops.** +One-shot and high-level relay operations (fetch a set, fetch one, page past the +relay cap, publish-and-confirm, NIP-45 count, NIP-77 sync/reconcile) are +`INostrClient` **extension functions** in +`quartz/…/nip01Core/relay/client/accessories/` (+ `…/reqs/` for the flow/subscribe +helpers). Because they're extensions, they don't surface under "usages of +`NostrClient`" or in completion — grep that package (or read its `README.md`, which +catalogs them) before writing a new subscription/collect loop. Reuse `fetchAll`, +`fetchFirst`, `fetchAllPages`, `publishAndConfirm`, `count`, `negentropyReconcile`, +etc. instead of re-implementing them. + **Share vs keep platform-native:** - **Share** → `quartz/commonMain/` (business logic, data models, protocol) and diff --git a/.claude/skills/relay-client/SKILL.md b/.claude/skills/relay-client/SKILL.md index 0bdea82309..c1019cde8f 100644 --- a/.claude/skills/relay-client/SKILL.md +++ b/.claude/skills/relay-client/SKILL.md @@ -123,6 +123,12 @@ Each subscription tracks "End of Stored Events" per relay. The eose manager in ` ## Related +- **Headless / one-shot client ops** (CLI, geode, tests, non-compose code): don't go + through `Subscribable` — use the `INostrClient` extension functions in + `quartz/…/nip01Core/relay/client/accessories/` (`fetchAll`, `fetchFirst`, + `fetchAllPages`, `publishAndConfirm`, `count`, `negentropyReconcile`/`negentropySync`, + …). They're extensions, so they don't show up under "usages of `NostrClient`" — see + that package's `README.md` for the catalog before writing a raw subscribe/collect loop. - `nostr-expert/references/tag-patterns.md` — how tags inform what a filter needs to look for. - `kotlin-coroutines/references/relay-patterns.md` — relay pool internals (sibling layer beneath assemblers). - `feed-patterns` skill — feeds compose several Subscribables (content + metadata + reactions). 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 545b35cb50..35bd5df2c1 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,8 +26,10 @@ 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.DeletionSettleResult import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcile +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySettleDeletions import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import com.vitorpamplona.quartz.nip01Core.store.IdAndTime @@ -54,15 +56,29 @@ 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. * - * Both directions are pipelined with the reconcile: need-id batches feed - * [DOWNLOAD_WORKERS] concurrent by-id REQ drains and have-ids feed a single - * uploader, so downloads and uploads overlap the remaining reconcile rounds - * instead of waiting for the full diff. Every downloaded event funnels - * through `Context.drain`'s verify-and-store path, unchanged. + * Deletion propagation (on by default; disable with `--no-sync-deletions`) is a + * **second pass over the residual**, not per-event work in the content pass — so it + * costs the same whether the database is tiny or huge. After the content settle, a + * re-reconcile's leftover diff is (barring races) exactly the events a deletion kept + * from converging: * - * Thin assembly only: the windowing, streaming, and back-pressure live in - * quartz (`negentropyReconcile`); this file only routes ids to - * `Context.drain` / `Context.publish`. + * - a residual **need** (relay has it, we still lack it after `--down` tried to + * download) = we deleted it → publish OUR covering deletion up so the relay drops it; + * - a residual **have** (we have it, relay still lacks it after `--up` tried to upload) + * = the relay deleted it → pull the relay's covering kind-5 down and apply it locally. + * + * Coverage is any way a deletion reaches an event ([deletionsCovering]): a NIP-09 kind-5 + * by id (`e`) or address (`a`, cutoff-checked), or a NIP-62 vanish targeting this relay + * (up direction only — a pulled vanish is not auto-applied, its blast radius being the + * whole account). The residual is small (only real deletion mismatches), so only it is + * fetched — never the whole need set. The loop repeats until a round resolves nothing. + * So `amy sync` (default `--down`) makes the relay honor your deletions; `--up` makes + * your store honor the relay's; `--up --down` converges both ways. + * + * Content is pipelined with the reconcile: need-id batches feed [DOWNLOAD_WORKERS] + * concurrent by-id REQ drains and have-ids feed a single uploader. Thin assembly only: + * the windowing, streaming, and back-pressure live in quartz (`negentropyReconcile`); + * this file only routes ids to `Context.drain` / `Context.publish`. */ object SyncCommand { private const val ID_CHUNK = 500 @@ -78,6 +94,15 @@ object SyncCommand { /** Overlapped `created_at`-window reconciles after an over-cap split. */ private const val RECONCILE_CONCURRENCY = 2 + /** + * Cap on deletion-settle rounds. Each round resolves the residual it can and + * re-reconciles; a healthy sync converges in 1–2 (round N sends/applies, round + * N+1 confirms empty). The cap only bounds pathological non-convergence (e.g. a + * relay that refuses a deletion), which the "resolved nothing → stop" check + * normally catches first. + */ + private const val MAX_DELETION_ROUNDS = 4 + suspend fun run( dataDir: DataDir, rest: Array, @@ -93,6 +118,7 @@ object SyncCommand { // Default direction is download; --up adds upload. val up = args.bool("up") val down = args.bool("down") || !up + val syncDeletions = !args.bool("no-sync-deletions") val filter = RawEventSupport.buildFilter(args) Context.openOrAnonymous(dataDir).use { ctx -> @@ -104,23 +130,22 @@ object SyncCommand { val downloaded = AtomicInteger(0) val uploaded = AtomicInteger(0) + // ── Pass 1: content settle — download needs, upload haves. No deletion + // logic, so a plain sync costs exactly what it always did. val result = try { coroutineScope { // needIds = relay has, we lack; haveIds = we have, relay lacks. - // Bounded so a slow download back-pressures the reconcile - // rounds instead of piling ids up in memory. val needBatches = Channel>(DOWNLOAD_WORKERS * 2) - // 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) val downloaders = List(DOWNLOAD_WORKERS) { launch { for (batch in needBatches) { - val got = ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs) - downloaded.addAndGet(got.size) + // drain verifies + stores; anything we deleted is + // rejected by our own tombstone and stays a "need". + downloaded.addAndGet(ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs).size) } } } @@ -129,8 +154,7 @@ 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() + if (ctx.publish(ev, setOf(relay)).values.any { it }) uploaded.incrementAndGet() } } } @@ -160,6 +184,28 @@ object SyncCommand { return Output.error("sync_error", e.message ?: "negentropy sync failed") } + // ── Pass 2+: deletion settle. The reusable quartz accessory re-reconciles + // and resolves only the residual — send our deletions up for what we deleted + // (bounded by --down), apply the relay's kind-5 down for what it deleted + // (bounded by --up) — looping until stable. Cheap regardless of database size + // (see negentropySettleDeletions), and best-effort so it can't fail the sync. + val deletions = + if (syncDeletions) { + ctx.client.negentropySettleDeletions( + relay = relay, + filter = filter, + store = ctx.store, + sendUp = down, + applyDown = up, + batchSize = ID_CHUNK, + idleTimeoutMs = timeoutMs, + maxRounds = MAX_DELETION_ROUNDS, + reconcileConcurrency = RECONCILE_CONCURRENCY, + ) + } else { + DeletionSettleResult(0, 0, 0) + } + Output.emit( mapOf( "relay" to relay.url, @@ -169,6 +215,9 @@ object SyncCommand { "have" to result.haveCount, "downloaded" to downloaded.get(), "uploaded" to uploaded.get(), + "deletions_sent_up" to deletions.sentUp, + "deletions_applied_down" to deletions.appliedDown, + "deletion_rounds" to deletions.rounds, ), ) return 0 diff --git a/cli/tests/.gitignore b/cli/tests/.gitignore index 3d5dc27cf2..7f43ea55a5 100644 --- a/cli/tests/.gitignore +++ b/cli/tests/.gitignore @@ -3,3 +3,4 @@ marmot/state-headless/ dm/state-dm-headless/ nests/state/ clink/state-clink-headless/ +sync/state-sync-deletions/ diff --git a/cli/tests/sync/sync-deletions-headless.sh b/cli/tests/sync/sync-deletions-headless.sh new file mode 100755 index 0000000000..871649dc5e --- /dev/null +++ b/cli/tests/sync/sync-deletions-headless.sh @@ -0,0 +1,196 @@ +#!/usr/bin/env bash +# +# sync-deletions-headless.sh — drives the real `amy` binary against a real +# `amy serve` relay to prove NIP-77 deletion propagation end-to-end. +# +# `amy sync` converges deletions in a second pass over the reconcile residual +# (see quartz `negentropySettleDeletions`). This exercises both directions plus +# the opt-out: +# +# T1 (up) — we deleted a note the relay still has → `amy sync` sends our +# kind-5 up and the relay drops the note. Verified by an ISOLATED +# third account whose store reads the relay only (no tombstone). +# T2 (off) — same setup with `--no-sync-deletions` → the relay keeps the note +# and nothing is sent. +# T3 (down) — the relay deleted a note we still hold → `amy sync --up` pulls the +# relay's kind-5 down and applies it locally (converges on re-sync). +# +# Each amy account gets its OWN $HOME so their file stores don't share (accounts +# under one $HOME share ~/.amy/shared/events-store). The relay (amy serve) keeps +# a separate store from any client store. +# +# Usage: ./sync-deletions-headless.sh [--port N] [--no-build] +set -uo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd -- "$SCRIPT_DIR/../../.." && pwd)" +TESTS_DIR="$(cd -- "$SCRIPT_DIR/.." && pwd)" +STATE_DIR="$SCRIPT_DIR/state-sync-deletions" +LOG_DIR="$STATE_DIR/logs" +RUN_TS="$(date +%Y%m%d-%H%M%S)" +LOG_FILE="$LOG_DIR/run-$RUN_TS.log" +RESULTS_FILE="$STATE_DIR/results-$RUN_TS.tsv" + +AMY_BIN="$REPO_ROOT/cli/build/install/amy/bin/amy" +RELAY_HOST="127.0.0.1" +RELAY_PORT="${RELAY_PORT:-7790}" +RELAY_URL="ws://$RELAY_HOST:$RELAY_PORT" +NO_BUILD=0 + +while [[ $# -gt 0 ]]; do + case "$1" in + --port) RELAY_PORT="$2"; RELAY_URL="ws://$RELAY_HOST:$RELAY_PORT"; shift ;; + --no-build) NO_BUILD=1 ;; + *) echo "unknown arg: $1" >&2; exit 2 ;; + esac + shift +done + +# Fresh state every run — stale per-account $HOME dirs from a prior run must not +# leak into this one. +rm -rf "$STATE_DIR" +mkdir -p "$LOG_DIR" +: >"$RESULTS_FILE" + +# shellcheck source=../lib.sh +source "$TESTS_DIR/lib.sh" + +# Leniently-trimmed equality assertion (assert helpers live in the DM-specific +# helpers.sh, which hardcodes its own amy wrappers — so define our own here). +assert_eq() { + local actual="$1" expected="$2" test_id="$3" note="${4:-}" + if [[ "${actual// /}" == "${expected// /}" ]]; then + info "assert: $test_id \"$actual\" == \"$expected\"" + return 0 + fi + fail_msg "$test_id: expected \"$expected\", got \"$actual\" (${note:-})" + record_result "$test_id" fail "${note:-mismatch}" + return 1 +} + +SERVE_PID="" +RELAY_HOME="" +cleanup() { + [[ -n "$SERVE_PID" ]] && kill "$SERVE_PID" 2>/dev/null + trap - EXIT INT TERM HUP + print_summary +} +trap cleanup EXIT +trap 'exit 130' INT +trap 'exit 143' TERM + +banner "amy sync — NIP-77 deletion propagation headless ($RUN_TS)" + +# ---- build ------------------------------------------------------------------ +if [[ "$NO_BUILD" -eq 0 ]]; then + step "Building amy (installDist)…" + (cd "$REPO_ROOT" && ./gradlew -q :cli:installDist) >>"$LOG_FILE" 2>&1 \ + || { fail_msg "build failed (see $LOG_FILE)"; exit 1; } +fi +[[ -x "$AMY_BIN" ]] || { fail_msg "amy binary not found at $AMY_BIN"; exit 1; } + +# ---- amy wrappers (one isolated $HOME per account) -------------------------- +strip() { grep -vE "Picked up JAVA_TOOL|DEBUG:|INFO:|MarmotManager|MlsGroup"; } +mk_home() { mktemp -d "$STATE_DIR/home.XXXXXX"; } +# amy_run args... +amy_run() { + local home="$1" acct="$2"; shift 2 + HOME="$home" "$AMY_BIN" --account "$acct" --secret-backend plaintext --json "$@" 2>>"$LOG_FILE" | strip +} + +RELAY_HOME="$(mk_home)" +amy_run "$RELAY_HOME" a init >/dev/null + +step "Starting amy serve on $RELAY_URL…" +HOME="$RELAY_HOME" "$AMY_BIN" --account a --secret-backend plaintext \ + serve --host "$RELAY_HOST" --port "$RELAY_PORT" >>"$LOG_FILE" 2>&1 & +SERVE_PID=$! + +# Wait for the relay to accept connections (poll the serve log). +for _ in $(seq 1 60); do + grep -q "relay up at" "$LOG_FILE" && break + sleep 0.5 +done +grep -q "relay up at" "$LOG_FILE" || { fail_msg "relay did not come up"; exit 1; } + +# Isolated verifier: its own empty store, reads the relay only (no tombstone). +VERIFY_HOME="$(mk_home)" +amy_run "$VERIFY_HOME" v init >/dev/null +relay_count() { amy_run "$VERIFY_HOME" v fetch --id "$1" --relay "$RELAY_URL" | jq -r '.count // 0'; } + +# ============================================================================= +# T1 — up direction: we deleted it, the relay still has it → sync sends it up. +# ============================================================================= +banner "T1 — amy sync sends our deletion up (relay drops the note)" +NOTE="$(amy_run "$RELAY_HOME" a event --kind 1 --content "delete-me-t1" | jq -c '.event')" +NID="$(echo "$NOTE" | jq -r '.id')" +echo "$NOTE" | amy_run "$RELAY_HOME" a publish --relay "$RELAY_URL" >/dev/null + +before="$(relay_count "$NID")" +assert_eq "$before" "1" T1.setup "relay should hold the note before sync" \ + && record_result T1.setup pass "relay has the note" + +# Delete locally only (no --relay → stored, applied, not sent to the relay). +amy_run "$RELAY_HOME" a event --kind 5 --tags "[[\"e\",\"$NID\"]]" --content "" --publish >/dev/null + +SYNC="$(amy_run "$RELAY_HOME" a sync --relay "$RELAY_URL")" +info "sync: $SYNC" +sent="$(echo "$SYNC" | jq -r '.deletions_sent_up // 0')" +assert_eq "$sent" "1" T1.sent_up "sync should report one deletion sent up" \ + && record_result T1.sent_up pass "deletions_sent_up=1" + +sleep 1 +after="$(relay_count "$NID")" +assert_eq "$after" "0" T1.relay_dropped "relay must have removed the note after sync" \ + && record_result T1.relay_dropped pass "relay note count 1 → 0" + +# ============================================================================= +# T2 — opt-out: --no-sync-deletions leaves the relay untouched. +# ============================================================================= +banner "T2 — --no-sync-deletions propagates nothing" +NOTE2="$(amy_run "$RELAY_HOME" a event --kind 1 --content "keep-me-t2" | jq -c '.event')" +NID2="$(echo "$NOTE2" | jq -r '.id')" +echo "$NOTE2" | amy_run "$RELAY_HOME" a publish --relay "$RELAY_URL" >/dev/null +amy_run "$RELAY_HOME" a event --kind 5 --tags "[[\"e\",\"$NID2\"]]" --content "" --publish >/dev/null + +SYNC2="$(amy_run "$RELAY_HOME" a sync --relay "$RELAY_URL" --no-sync-deletions)" +info "sync: $SYNC2" +sent2="$(echo "$SYNC2" | jq -r '.deletions_sent_up // 0')" +assert_eq "$sent2" "0" T2.no_send "--no-sync-deletions must send nothing" \ + && record_result T2.no_send pass "deletions_sent_up=0" +sleep 1 +kept="$(relay_count "$NID2")" +assert_eq "$kept" "1" T2.relay_kept "relay must still hold the note" \ + && record_result T2.relay_kept pass "relay note untouched" + +# ============================================================================= +# T3 — down direction: the relay deleted it, we still hold it → sync --up pulls +# the relay's deletion down and applies it locally. +# ============================================================================= +banner "T3 — amy sync --up applies the relay's deletion locally" +BOB_HOME="$(mk_home)" +amy_run "$BOB_HOME" b init >/dev/null +NOTE3="$(amy_run "$RELAY_HOME" a event --kind 1 --content "delete-me-t3" | jq -c '.event')" +NID3="$(echo "$NOTE3" | jq -r '.id')" +echo "$NOTE3" | amy_run "$RELAY_HOME" a publish --relay "$RELAY_URL" >/dev/null +# bob's isolated store learns the note from the relay… +amy_run "$BOB_HOME" b fetch --id "$NID3" --relay "$RELAY_URL" >/dev/null +# …then the relay deletes it (author pushes a kind-5 straight to the relay). +amy_run "$RELAY_HOME" a event --kind 5 --tags "[[\"e\",\"$NID3\"]]" --content "" | jq -c '.event' \ + | amy_run "$RELAY_HOME" a publish --relay "$RELAY_URL" >/dev/null + +SYNC3="$(amy_run "$BOB_HOME" b sync --up --relay "$RELAY_URL")" +info "sync: $SYNC3" +applied="$(echo "$SYNC3" | jq -r '.deletions_applied_down // 0')" +assert_eq "$applied" "1" T3.applied_down "sync --up should apply one relay deletion locally" \ + && record_result T3.applied_down pass "deletions_applied_down=1" + +# Converged: a second --up sync finds nothing left to apply. +SYNC3B="$(amy_run "$BOB_HOME" b sync --up --relay "$RELAY_URL")" +applied2="$(echo "$SYNC3B" | jq -r '.deletions_applied_down // 0')" +assert_eq "$applied2" "0" T3.converged "re-sync applies nothing (converged)" \ + && record_result T3.converged pass "second sync stable" + +# print_summary runs from the cleanup trap; exit non-zero if any test failed. +grep -q $'\tfail\t' "$RESULTS_FILE" && exit 1 +exit 0 diff --git a/geode/build.gradle.kts b/geode/build.gradle.kts index 594458773f..ee636965c7 100644 --- a/geode/build.gradle.kts +++ b/geode/build.gradle.kts @@ -72,6 +72,9 @@ tasks.withType().configureEach { // NegentropyServerReconcileBenchmark opt-in + sizing. System.getProperty("negServerBench")?.let { systemProperty("negServerBench", it) } System.getProperty("negBenchN")?.let { systemProperty("negBenchN", it) } + // DeletionSettleBenchmark sizing. + System.getProperty("delBenchN")?.let { systemProperty("delBenchN", it) } + System.getProperty("delBenchK")?.let { systemProperty("delBenchK", it) } // MirrorSyncThroughputTest sizing + external-source opt-in. System.getProperty("syncN")?.let { systemProperty("syncN", it) } System.getProperty("syncExpect")?.let { systemProperty("syncExpect", it) } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSettleBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSettleBenchmark.kt new file mode 100644 index 0000000000..46e4b7a1ca --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSettleBenchmark.kt @@ -0,0 +1,128 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.geode + +import com.vitorpamplona.geode.testing.RelayClientTest +import com.vitorpamplona.geode.testing.preload +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcileIds +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySettleDeletions +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +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 com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals + +/** + * Cost of the deletion side-channel ([negentropySettleDeletions]) at database scale. + * + * The whole point of the two-pass design is that turning deletions on does NOT re-fetch + * content — the content sync already downloaded the need set, and the settle only touches + * the reconcile *residual* (the events a deletion stopped from converging). So its cost is + * one reconcile per round plus the residual, independent of how big the database is. + * + * This models the post-content-settle state: a relay holding N notes, and a local store + * holding the same N notes EXCEPT K it deleted (it keeps the K kind-5s). The residual is + * exactly those K — so a `sendUp` settle fetches K, not N. It prints the reconcile cost + * (the O(N) part it shares with any sync) next to the settle cost, so the deletion + * overhead is visible as "≈ a couple of reconciles + K", not "+ a content re-download". + * + * Why the printed settle can read as several× a bare reconcile at large N: the extra time + * is NOT the deletion algorithm (a phase breakdown showed reconciles stay ~sub-second at + * N=100k, and the settle re-fetches K=20, not N). It is entirely the K `publishAndConfirm` + * ingests into a large geode relay — publishing K *plain* notes costs the same — and that + * ingest path is JVM-cold on first use: consecutive K-note batches dropped monotonically + * (~3100 → ~570 ms) purely from JIT warmup. So the cost is O(K) relay-ingest dominated by + * one-time warmup, independent of N. + * + * Default N is small so it doubles as a fast correctness guard; scale it with + * `-DdelBenchN=200000` to see the shape at size. Not a speed assertion (container noise). + */ +class DeletionSettleBenchmark : RelayClientTest() { + private val signer = NostrSignerSync(KeyPair()) + private val local = EventStore(null) + + @AfterTest fun closeLocal() = local.close() + + private val n = System.getProperty("delBenchN")?.toInt() ?: 2_000 + private val k = System.getProperty("delBenchK")?.toInt() ?: 20 + + @Test + fun settleCostIsResidualNotDatabase() = + runBlocking { + val base = TimeUtils.now() - n + // N notes with monotonic created_at (sorted order == index order). + val notes = (0 until n).map { signer.sign(TextNoteEvent.build("n$it", createdAt = base + it.toLong())) } + // The last K are the ones we deleted locally. + val deleted = notes.takeLast(k) + val kept = notes.dropLast(k) + val deletions = deleted.map { signer.sign(DeletionEvent.build(listOf(it), createdAt = it.createdAt + 1)) } + + // Relay holds all N notes; we hold the N-K we didn't delete, plus the K kind-5s. + defaultRelay.preload(notes) + kept.forEach { local.insert(it) } + deletions.forEach { local.insert(it) } + assertEquals(n - k, local.query(Filter(kinds = listOf(1))).size, "local kept N-K notes") + + // Cost of one reconcile — the O(N) work every sync round already does. + val r0 = System.nanoTime() + val diff = + withTimeout(120_000) { + client.negentropyReconcileIds(defaultRelayUrl, Filter(kinds = listOf(1)), local.snapshotIdsForNegentropy(listOf(Filter(kinds = listOf(1))))) + } + val reconcileMs = (System.nanoTime() - r0) / 1e6 + assertEquals(k, diff.needIds.size, "the residual is exactly the K deleted notes, not N") + + // Cost of the whole settle: reconcile(s) + resolve the K-event residual. + val s0 = System.nanoTime() + val res = + withTimeout(120_000) { + client.negentropySettleDeletions( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + store = local, + sendUp = true, + applyDown = false, + idleTimeoutMs = 60_000, + ) + } + val settleMs = (System.nanoTime() - s0) / 1e6 + + assertEquals(k, res.sentUp, "sent exactly K deletions up") + assertEquals( + n - k, + defaultRelay.store.query(Filter(kinds = listOf(1))).size, + "relay converged: the K deleted notes are gone", + ) + + println("─ DeletionSettleBenchmark @ N=$n K=$k ─") + println(" one reconcile: ${"%.1f".format(reconcileMs)} ms (O(N), shared with any sync)") + println(" full settle: ${"%.1f".format(settleMs)} ms (${res.rounds} rounds, sentUp=${res.sentUp})") + println(" deletion cost: settle is ~${"%.1f".format(settleMs / reconcileMs)}× one reconcile — fetched K=$k, not N=$n") + } +} diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt new file mode 100644 index 0000000000..f3a7ef4306 --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/DeletionSyncTest.kt @@ -0,0 +1,267 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.geode + +import com.vitorpamplona.geode.testing.RelayClientTest +import com.vitorpamplona.geode.testing.preload +import com.vitorpamplona.geode.testing.publish +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcileIds +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySettleDeletions +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +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.nip01Core.store.deletionsCovering +import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore +import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent +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.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * The `amy sync` deletion rule: for the events the relay HAS that we LACK (the + * negentropy need set), publish the local deletions that would make the relay remove + * them — and only those. [deletionsCovering] is the core: it maps a set of server-held + * events to the local deletions that cover them, across id-based (NIP-09 `e`), + * address-based (NIP-09 `a`, cutoff-checked) and NIP-62 vanish (relay-targeted, cutoff). + */ +class DeletionSyncTest : RelayClientTest() { + private val signer = NostrSignerSync(KeyPair()) + private val here: NormalizedRelayUrl get() = defaultRelayUrl + private val elsewhere = RelayUrlNormalizer.normalize("wss://elsewhere.example/") + + private val store = EventStore(null) + + @AfterTest fun closeStore() = store.close() + + private fun note(text: String): Event = signer.sign(TextNoteEvent.build(text)) + + // ---- deletionsCovering: the three coverage forms -------------------------- + + @Test + fun idBasedDeletionCoversByETag() = + runBlocking { + val target = note("delete me") + val deletion = signer.sign(DeletionEvent.build(listOf(target), createdAt = target.createdAt + 1)) + store.insert(deletion) + + assertEquals(listOf(deletion.id), store.deletionsCovering(listOf(target), here).map { it.id }) + // A different note the deletion doesn't name is not covered. + assertTrue(store.deletionsCovering(listOf(note("unrelated")), here).isEmpty()) + } + + @Test + fun addressBasedDeletionCoversByATagWithCutoff() = + runBlocking { + val contacts = ContactListEvent.createFromScratch(emptyList(), null, signer) + // Address-only deletion (no `e` tag) → only the `a`-tag path can match it. + val delAddr = signer.sign(DeletionEvent.buildAddressOnly(listOf(contacts), createdAt = contacts.createdAt + 1)) + store.insert(delAddr) + + assertEquals( + listOf(delAddr.id), + store.deletionsCovering(listOf(contacts), here).map { it.id }, + "a replaceable event is covered by an address deletion at/after it", + ) + + // NIP-09 cutoff: a deletion OLDER than the event does not delete it. + val stale = EventStore(null) + stale.insert(signer.sign(DeletionEvent.buildAddressOnly(listOf(contacts), createdAt = contacts.createdAt - 1))) + assertTrue(stale.deletionsCovering(listOf(contacts), here).isEmpty(), "an older address deletion does not cover") + stale.close() + } + + @Test + fun vanishCoversAuthorsEventsWhenTargetedAndNewer() = + runBlocking { + val old = note("before the vanish") + val vanishHere = signer.sign(RequestToVanishEvent.build(here, createdAt = old.createdAt + 1)) + store.insert(vanishHere) + + assertEquals( + listOf(vanishHere.id), + store.deletionsCovering(listOf(old), here).map { it.id }, + "a relay-targeted vanish issued after the event covers it", + ) + + // Not targeting this relay → not sent here. + val otherStore = EventStore(null) + otherStore.insert(signer.sign(RequestToVanishEvent.build(elsewhere, createdAt = old.createdAt + 1))) + assertTrue(otherStore.deletionsCovering(listOf(old), here).isEmpty(), "a vanish for another relay is not sent") + + // A newer event (created after the vanish) is NOT deleted by it. + val newer = signer.sign(TextNoteEvent.build("after", createdAt = vanishHere.createdAt + 10)) + assertTrue(store.deletionsCovering(listOf(newer), here).isEmpty(), "the vanish does not cover a later event") + otherStore.close() + } + + // ---- end-to-end through the relay ---------------------------------------- + + // UP direction: we deleted it, the relay still has it → send our deletion up. + @Test + fun sendsCoveringDeletionSoRelayRemovesTheNote() = + runBlocking { + val target = note("delete me e2e") + val deletion = signer.sign(DeletionEvent.build(listOf(target), createdAt = target.createdAt + 1)) + + // Relay holds the note; we already deleted it locally (hold only the kind-5). + defaultRelay.preload(listOf(target)) + val local = hub.getOrCreate(RelayUrlNormalizer.normalize("ws://local/")) + local.preload(listOf(target, deletion)) + assertTrue(local.store.query(Filter(ids = listOf(target.id))).isEmpty(), "local deleted the note") + + // Reconcile → the note is a need id. (No local kind-1 remains.) + val diff = + withTimeout(20_000) { + client.negentropyReconcileIds(relay = defaultRelayUrl, filter = Filter(kinds = listOf(1)), localEntries = emptyList()) + } + assertEquals(setOf(target.id), diff.needIds.toSet()) + + // What SyncCommand does: fetch the need events, ask the local store which of + // our deletions cover them, publish those. + val serverEvents = defaultRelay.store.query(Filter(ids = diff.needIds)) + val covering = local.store.deletionsCovering(serverEvents, defaultRelayUrl) + assertEquals(listOf(deletion.id), covering.map { it.id }) + covering.forEach { defaultRelay.publish(it) } + + assertTrue( + defaultRelay.store.query(Filter(ids = listOf(target.id))).isEmpty(), + "relay applied the pushed deletion and removed the note", + ) + } + + // DOWN direction: the relay deleted it, we still have it → pull the relay's deletion + // down and apply it locally (the residual-have resolution). + @Test + fun appliesRelaysDeletionSoLocalRemovesTheNote() = + runBlocking { + val target = note("delete me down") + val deletion = signer.sign(DeletionEvent.build(listOf(target), createdAt = target.createdAt + 1)) + + // Relay already applied the deletion → holds only the kind-5. + defaultRelay.preload(listOf(target, deletion)) + assertTrue(defaultRelay.store.query(Filter(ids = listOf(target.id))).isEmpty(), "relay deleted the note") + + // Local still holds the note (never saw the deletion). + val local = hub.getOrCreate(RelayUrlNormalizer.normalize("ws://local-down/")) + local.preload(listOf(target)) + assertEquals(1, local.store.query(Filter(ids = listOf(target.id))).size) + + // Reconcile → the note is a HAVE (we have it, the relay lacks it). + val diff = + withTimeout(20_000) { + client.negentropyReconcileIds( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + localEntries = listOf(IdAndTime(target.createdAt, target.id)), + ) + } + assertEquals(setOf(target.id), diff.haveIds.toSet()) + + // What SyncCommand does for the down direction: take our have events, ask the + // RELAY which of ITS deletions cover them, and apply those locally. + val ourEvents = local.store.query(Filter(ids = diff.haveIds)) + val relayDeletions = deletionsCovering(ourEvents, defaultRelayUrl) { f -> defaultRelay.store.query(f) } + assertEquals(listOf(deletion.id), relayDeletions.map { it.id }) + relayDeletions.filterIsInstance().forEach { local.store.insert(it) } + + assertTrue( + local.store.query(Filter(ids = listOf(target.id))).isEmpty(), + "local applied the pulled deletion and removed the note", + ) + } + + // ---- the full accessory loop (negentropySettleDeletions) ----------------- + + // sendUp: local holds the deletion, relay still has the note → the loop pushes it + // up and the relay converges to gone. + @Test + fun settleSendsOurDeletionUp() = + runBlocking { + val target = note("settle up") + val deletion = signer.sign(DeletionEvent.build(listOf(target), createdAt = target.createdAt + 1)) + val localStore = EventStore(null) + localStore.insert(target) + localStore.insert(deletion) // deletes target locally, keeps the kind-5 + defaultRelay.preload(listOf(target)) + + val res = + withTimeout(30_000) { + client.negentropySettleDeletions( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + store = localStore, + sendUp = true, + applyDown = false, + idleTimeoutMs = 20_000, + ) + } + + assertEquals(1, res.sentUp) + assertEquals(0, res.appliedDown) + assertTrue( + defaultRelay.store.query(Filter(ids = listOf(target.id))).isEmpty(), + "relay converged: the deleted note is gone", + ) + localStore.close() + } + + // applyDown: relay deleted the note (holds only the kind-5), local still has it → + // the loop pulls the relay's deletion down and local converges to gone. + @Test + fun settleAppliesRelayDeletionDown() = + runBlocking { + val target = note("settle down") + val deletion = signer.sign(DeletionEvent.build(listOf(target), createdAt = target.createdAt + 1)) + defaultRelay.preload(listOf(target, deletion)) // relay deletes target, keeps the kind-5 + val localStore = EventStore(null) + localStore.insert(target) + + val res = + withTimeout(30_000) { + client.negentropySettleDeletions( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + store = localStore, + sendUp = false, + applyDown = true, + idleTimeoutMs = 20_000, + ) + } + + assertEquals(0, res.sentUp) + assertEquals(1, res.appliedDown) + assertTrue( + localStore.query(Filter(ids = listOf(target.id))).isEmpty(), + "local converged: the relay-deleted note is gone", + ) + localStore.close() + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropyDeletionSettleExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropyDeletionSettleExt.kt new file mode 100644 index 0000000000..67e770f1de --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropyDeletionSettleExt.kt @@ -0,0 +1,145 @@ +/* + * 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.crypto.verify +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.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.deletionsCovering +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent + +/** + * Outcome of a [negentropySettleDeletions] run. + * + * @property sentUp distinct local deletions published to the relay (up direction). + * @property appliedDown distinct relay deletions ingested into [store] (down direction). + * @property rounds reconcile rounds run before convergence (or the cap). + */ +class DeletionSettleResult( + val sentUp: Int, + val appliedDown: Int, + val rounds: Int, +) + +/** + * Converge deletions between [store] and [relay] AFTER a content sync has settled the + * two sides — the second half of a two-pass sync. NIP-77 reconciles by id, so a plain + * content sync converges everything except events a deletion physically stops from + * moving; those survive as the reconcile's residual, which this resolves: + * + * - **[sendUp]** — a residual **need** (relay has it, [store] still lacks it after the + * content pass tried to download it) means we deleted it. Publish OUR covering + * deletion up ([IEventStore.deletionsCovering]) so the relay drops it. + * - **[applyDown]** — a residual **have** ([store] has it, relay still lacks it after + * the content pass tried to upload it) means the relay deleted it. Pull the RELAY'S + * covering **kind-5** down and ingest it, so [store] drops it too. A NIP-62 vanish is + * deliberately NOT applied on pull — its blast radius is the author's whole account. + * + * Because it works off the residual — not every id — the cost is one cheap reconcile + * per round plus the (small) residual, independent of database size. It loops until a + * round resolves nothing (converged, and thereby self-verified) or [maxRounds] is hit. + * + * **Direction requires the matching content pass.** A residual need is a clean signal + * only after the content sync attempted the download ([sendUp] pairs with a `--down` + * content pass); a residual have only after it attempted the upload ([applyDown] pairs + * with `--up`). Passing a direction whose content pass didn't run makes its residual the + * full unsettled set, not a deletion signal — so drive this with the same directions the + * content pass used. + * + * Best-effort: a reconcile failure ([NegentropySyncException]) stops the loop and returns + * what already settled rather than throwing — the content sync is the primary work. + * + * @param batchSize ids per reconcile chunk and per by-id fetch. + * @param idleTimeoutMs idle watchdog for the reconciles and fetches. + * @param maxRounds hard cap on rounds; the "resolved nothing" check usually stops first. + * @param reconcileConcurrency overlapped `created_at`-window reconciles after an over-cap split. + */ +suspend fun INostrClient.negentropySettleDeletions( + relay: NormalizedRelayUrl, + filter: Filter, + store: IEventStore, + sendUp: Boolean, + applyDown: Boolean, + batchSize: Int = 500, + idleTimeoutMs: Long = 120_000L, + maxRounds: Int = 4, + reconcileConcurrency: Int = 1, +): DeletionSettleResult { + if ((!sendUp && !applyDown) || maxRounds <= 0) return DeletionSettleResult(0, 0, 0) + + val publishTimeoutSecs = (idleTimeoutMs / 1000).coerceAtLeast(1) + val sentUp = HashSet() + val appliedDown = HashSet() + var rounds = 0 + + while (rounds < maxRounds) { + rounds++ + val diff = + try { + negentropyReconcileIds( + relay = relay, + filter = filter, + localEntries = store.snapshotIdsForNegentropy(listOf(filter)), + batchSize = batchSize, + idleTimeoutMs = idleTimeoutMs, + reconcileConcurrency = reconcileConcurrency, + ) + } catch (e: NegentropySyncException) { + break + } + + var resolved = 0 + + // residual needs → publish our covering deletions up. + if (sendUp) { + for (chunk in diff.needIds.chunked(batchSize)) { + val events = fetchAll(relay, Filter(ids = chunk), idleTimeoutMs) + for (del in store.deletionsCovering(events, relay)) { + if (sentUp.add(del.id)) { + if (publishAndConfirm(del, setOf(relay), publishTimeoutSecs)) resolved++ + } + } + } + } + + // residual haves → ingest the relay's covering kind-5 (never a vanish). + if (applyDown) { + for (chunk in diff.haveIds.chunked(batchSize)) { + val ours = store.query(Filter(ids = chunk)) + val relayDeletions = deletionsCovering(ours, relay) { f -> fetchAll(relay, f, idleTimeoutMs) } + for (del in relayDeletions.filterIsInstance()) { + if (del.verify() && appliedDown.add(del.id)) { + store.insert(del) + resolved++ + } + } + } + } + + if (resolved == 0) break + } + + return DeletionSettleResult(sentUp.size, appliedDown.size, rounds) +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md new file mode 100644 index 0000000000..46ce631095 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md @@ -0,0 +1,62 @@ +# `INostrClient` accessories + +One-shot / high-level relay operations, written as **extension functions** on +`INostrClient`. They live here (and in `../reqs/`) rather than on the client class, +so they don't show up under "usages of `NostrClient`" or in method completion — you +only find them by knowing this package exists. + +**Before writing a new subscribe / REQ / publish loop, look here first.** Most of what +a caller needs (fetch a set, fetch one, page past the relay cap, publish-and-confirm, +count, negentropy sync/reconcile) already exists. + +Import as `com.vitorpamplona.quartz.nip01Core.relay.client.accessories.` (or +`...client.reqs.` for the flow/subscribe helpers). + +## One-shot reads (subscribe → collect → return) + +| Function | File | Use when | +| --- | --- | --- | +| `fetchAll(relay, filter, timeoutMs)` | `NostrClientFetchAllExt` | Get every event matching a filter in one REQ, deduped by id, until EOSE or timeout. **No verify, no store** — just the events. | +| `fetchFirst(relay, filter, timeoutMs)` | `NostrClientFetchFirstExt` | Get the first matching event and stop (returns `null` on none/timeout). | +| `fetchAllPages(relay, filters, timeoutMs)` | `NostrClientFetchAllPagesExt` | Fully retrieve a result set larger than the relay's per-REQ cap (strfry `limit`, ~500) by walking a `created_at` cursor. Bound it with the filter's `limit`. | +| `fetchAllPagesFromPool(filters, ...)` | `NostrClientFetchAllPagesPoolExt` | Same paging, across several relays at once, deduped across them. | + +## Streaming (`Flow`) + +| Function | File | Use when | +| --- | --- | --- | +| `fetchAsFlow(relay, filter)` | `../reqs/NostrClientFetchAsFlowExt` | Emit the accumulating list on each arrival; completes on EOSE. One-shot query as a flow. | +| `subscribeAsFlow(relay, filter)` | `../reqs/NostrClientSubscribeAsFlowExt` | Live subscription as a flow (stays open past EOSE; re-sends the REQ on reconnect). | +| `subscribe(subId, filters, listener)` | `../reqs/StaticSubscription`, `DynamicSubscription` | Raw live subscription with a `SubscriptionListener`. The lowest-level primitive the above build on. | + +## Publish + +| Function | File | Use when | +| --- | --- | --- | +| `publishAndConfirm(event, relays, timeout)` | `NostrClientPublishExt` | Send an EVENT and wait for `OK`; returns whether any relay accepted it. | +| `publishAndConfirmDetailed(event, relays, timeout)` | `NostrClientPublishExt` | Same, but returns the per-relay accepted/rejected map. | + +## Count (NIP-45) + +| Function | File | Use when | +| --- | --- | --- | +| `count(relay, filter, timeoutMs)` | `NostrClientCountExt` | NIP-45 `COUNT` against one relay (`null` on timeout / no support). | +| `countMerged(relays, filter, ...)` | `NostrClientCountExt` | Merged count across relays. | + +## Negentropy (NIP-77) + +| Function | File | Use when | +| --- | --- | --- | +| `negentropySync(relay, filter, ...)` | `NostrClientNegentropySyncExt` | Download everything a relay holds for a filter, diffing against `localEntries` and by-id downloading only the diff. Throws `NegentropySyncException` if the relay can't reconcile (no fallback). | +| `negentropySyncOrFetch(relay, filter, ...)` | `NostrClientNegentropySyncExt` | Same, but transparently falls back to `fetchAllPages` when the relay can't reconcile. The "just get the events" combinator. | +| `negentropySyncEvents` / `negentropySyncOrFetchEvents` | `NostrClientNegentropySyncEventsExt` | The two above as an O(1)-memory `Flow`. | +| `negentropyReconcile(relay, filter, localEntries, onNeedIds, onHaveIds)` | `NostrClientNegentropySyncExt` | **Pure diff, no I/O** — streams the two directions (`need` = relay has & we lack; `have` = we have & relay lacks) to callbacks. Compose your own download/upload on top. | +| `negentropyReconcileIds(relay, filter, localEntries)` | `NostrClientNegentropySyncExt` | Same diff, materialized into `needIds` / `haveIds` lists (small sets only). | +| `negentropySettleDeletions(relay, filter, store, sendUp, applyDown)` | `NostrClientNegentropyDeletionSettleExt` | Second pass of a two-pass sync: after a content sync settles, re-reconcile and resolve only the residual — send our covering deletions up (`sendUp`) and/or apply the relay's kind-5 down (`applyDown`), looping until stable. Cost is O(residual), not O(db). Pairs with `IEventStore.deletionsCovering`. | + +`fetchByIds`, `reconcileStreaming`, `syncPipeline` in `NostrClientNegentropySyncExt` +are `internal` implementation details — not part of the public surface. + +--- + +_Keep this table in sync when you add a public `INostrClient` extension here._ diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/EventStoreDeletionsExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/EventStoreDeletionsExt.kt new file mode 100644 index 0000000000..0c5a8203dc --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/EventStoreDeletionsExt.kt @@ -0,0 +1,104 @@ +/* + * 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.store + +import com.vitorpamplona.quartz.nip01Core.core.Address +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.isAddressable +import com.vitorpamplona.quartz.nip01Core.core.isReplaceable +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent +import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent + +/** The addressable/replaceable coordinate of [event] as a NIP-01 `a`-tag value. */ +private fun addressValue(event: Event): String { + val dTag = if (event.kind.isAddressable()) event.tags.firstOrNull { it.size > 1 && it[0] == "d" }?.get(1) ?: "" else "" + return Address.assemble(event.kind, event.pubKey, dTag) +} + +/** + * The local deletion events that would make [relay] remove one of [serverEvents] — the + * events the relay HAS that we LACK. Used by sync to push *only* the deletions that + * actually apply to what the relay holds, and nothing else (not other deletions by the + * same author). Covers every way a stored deletion can reach an event: + * + * - **NIP-09, id-based** — a kind-5 with an `e` tag naming a server event's id. + * - **NIP-09, address-based** — a kind-5 with an `a` tag naming a server event's + * addressable/replaceable coordinate, at or after that event's `created_at` + * (NIP-09 only deletes `created_at <= deletion.created_at`). + * - **NIP-62 vanish** — a kind-62 by a server event's author, targeting [relay] (its + * `relay` tags name the URL or `ALL_RELAYS`), issued after that event (a vanish + * deletes `created_at < vanish.created_at`). + * + * Deduped by event id; a single deletion covering several events is returned once. + * + * [query] is where the deletions are looked up — it is source-agnostic on purpose, so + * the same coverage rule runs in both sync directions: + * - **up** (send our deletions): `events` are the relay's, `query` is the local store — + * which of OUR deletions would delete what the relay still holds. + * - **down** (apply the relay's deletions): `events` are ours, `query` fetches from the + * relay — which of the RELAY'S deletions would delete what we still hold. + */ +suspend fun deletionsCovering( + events: List, + relay: NormalizedRelayUrl, + query: suspend (Filter) -> List, +): List { + if (events.isEmpty()) return emptyList() + val covering = LinkedHashMap() + + // 1. id-based NIP-09: a kind-5 `e`-tagging an event's id. + query(Filter(kinds = listOf(DeletionEvent.KIND), tags = mapOf("e" to events.map { it.id }))) + .forEach { covering[it.id] = it } + + // 2. address-based NIP-09: a kind-5 `a`-tagging an event's coordinate, cutoff-checked. + val byAddress = events.filter { it.kind.isAddressable() || it.kind.isReplaceable() }.groupBy(::addressValue) + if (byAddress.isNotEmpty()) { + query(Filter(kinds = listOf(DeletionEvent.KIND), tags = mapOf("a" to byAddress.keys.toList()))) + .forEach { del -> + if (del !is DeletionEvent) return@forEach + for (addr in del.deleteAddresses()) { + val hit = byAddress[addr.toValue()] ?: continue + if (hit.any { it.createdAt <= del.createdAt }) { + covering[del.id] = del + break + } + } + } + } + + // 3. NIP-62 vanish: a kind-62 by an event's author, targeting this relay, issued after it. + query(Filter(kinds = listOf(RequestToVanishEvent.KIND), authors = events.mapTo(HashSet()) { it.pubKey }.toList())) + .forEach { vanish -> + if (vanish !is RequestToVanishEvent || !vanish.shouldVanishFrom(relay)) return@forEach + if (events.any { it.pubKey == vanish.pubKey && it.createdAt < vanish.createdAt }) covering[vanish.id] = vanish + } + + return covering.values.toList() +} + +/** [deletionsCovering] with the local store as the deletion source (the "up" direction). */ +suspend fun IEventStore.deletionsCovering( + serverEvents: List, + relay: NormalizedRelayUrl, +): List = deletionsCovering(serverEvents, relay) { query(it) }