mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-10 16:33:27 +00:00
refactor(quartz): extract GrapeRankUpdater outbox-model WoT refresh utility
Moves the `amy graperank update` logic out of the CLI and into quartz as GrapeRankUpdater, alongside GrapeRankDataCrawler in experimental/graperank, so Android and any other quartz consumer can run the same refresh. Given an INostrClient + IEventStore it reads every kind:10002 in the store, inverts them into a write-relay -> authors map (the outbox model), then runs one NIP-77 negentropy reconcile per write relay scoped to its authors: bidirectional content sync into/from the store, deletion settle over the residual (applyDown downloads the relay's kind:5 when an uploaded record was rejected because the author retracted it), and a full paged-download fallback when a relay can't reconcile. Bounds and directions are a Config; per-relay and aggregate outcomes are returned as a Result. The CLI `graperank update` is now a thin wrapper: it parses flags, builds the Config, and renders GrapeRankUpdater.Result as text/JSON — no sync logic left in cli/ (all reconcile/window/back-pressure/deletion logic lives in quartz's relay-client accessories, which GrapeRankUpdater composes). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TdEvjsZ81XuUtdJsVzmHxt
This commit is contained in:
Claude
@@ -30,28 +30,22 @@ import com.vitorpamplona.quartz.experimental.graperank.GrapeRank
|
||||
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankDataCrawler
|
||||
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankParams
|
||||
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankPublisher
|
||||
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankUpdater
|
||||
import com.vitorpamplona.quartz.experimental.graperank.TrustGraphBuilder
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
|
||||
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.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent
|
||||
import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent
|
||||
import com.vitorpamplona.quartz.nip09Deletions.DeletionIndex
|
||||
import com.vitorpamplona.quartz.nip51Lists.muteList.MuteListEvent
|
||||
import com.vitorpamplona.quartz.nip56Reports.ReportEvent
|
||||
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
|
||||
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.TrustProviderListEvent
|
||||
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.serviceProviders
|
||||
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.tags.ProviderTypes
|
||||
@@ -62,14 +56,7 @@ import com.vitorpamplona.quartz.nip85TrustedAssertions.users.tags.RankTag
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.joinAll
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Semaphore
|
||||
import kotlinx.coroutines.sync.withPermit
|
||||
import java.util.Collections
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.math.roundToInt
|
||||
|
||||
/**
|
||||
@@ -446,59 +433,18 @@ object GrapeRankCommand {
|
||||
return 0
|
||||
}
|
||||
|
||||
// ── graperank update: outbox-model refresh of the WoT record kinds ──────────
|
||||
|
||||
/** WoT record kinds a GrapeRank score is a function of, refreshed by `update`. */
|
||||
private val UPDATE_KINDS =
|
||||
listOf(
|
||||
MetadataEvent.KIND, // 0 — profiles
|
||||
ContactListEvent.KIND, // 3 — follows
|
||||
AdvertisedRelayListEvent.KIND, // 10002 — outbox relay lists
|
||||
ReportEvent.KIND, // 1984 — reports
|
||||
)
|
||||
|
||||
/** ids per reconcile chunk and per by-id fetch (mirrors [SyncCommand]). */
|
||||
private const val UPDATE_ID_CHUNK = 500
|
||||
|
||||
/** Concurrent by-id download REQs per relay (mirrors [SyncCommand]). */
|
||||
private const val UPDATE_DOWNLOAD_WORKERS = 4
|
||||
|
||||
/** Overlapped `created_at`-window reconciles after an over-cap split. */
|
||||
private const val UPDATE_RECONCILE_CONCURRENCY = 2
|
||||
|
||||
/** Cap on deletion-settle rounds; a healthy group converges in 1–2. */
|
||||
private const val UPDATE_MAX_DELETION_ROUNDS = 4
|
||||
|
||||
/**
|
||||
* `amy graperank update [flags]` — refresh every locally-known author's WoT
|
||||
* record kinds (0 / 3 / 10002 / 1984) straight from their own outbox, so the
|
||||
* next `graperank score` runs on current data without a full follow-graph crawl.
|
||||
*
|
||||
* Unlike `sync` (which walks the reachable follow graph outward from an
|
||||
* observer), this is a store-driven refresh: it reads every kind:10002 already
|
||||
* in the local store, inverts them into a `write-relay -> authors` map (the
|
||||
* outbox model — an author's events live on the relays they write to), then
|
||||
* runs ONE NIP-77 negentropy reconcile per write relay scoped to exactly the
|
||||
* authors who publish there. So each relay is asked only for the authors it
|
||||
* actually hosts, and each author is reconciled only against their own relays.
|
||||
*
|
||||
* Bidirectional by default (`--down`/`--up` narrow it):
|
||||
* - **down** downloads records the relay has and we lack (by-id REQ);
|
||||
* - **up** uploads records we have and the relay lacks (EVENT).
|
||||
*
|
||||
* After the content pass each group runs [negentropySettleDeletions] over the
|
||||
* reconcile residual (disable with `--no-sync-deletions`). Its **applyDown**
|
||||
* direction is the "download the deletion when our upload was rejected" case:
|
||||
* an event we pushed up that the relay keeps rejecting (it deleted it) surfaces
|
||||
* as a residual *have*, so we pull the relay's covering kind:5 down and apply it
|
||||
* locally — our store drops the event the author retracted. The **sendUp**
|
||||
* direction publishes OUR covering deletions for records we deleted that the
|
||||
* relay still serves. Cheap: it works off the small residual, not the whole set.
|
||||
*
|
||||
* If negentropy can't reconcile a relay (no NIP-77 support, an over-cap minimal
|
||||
* window, a mid-sync disconnect, …) the group falls back to a full paged download
|
||||
* ([Context.drainAllPages]) of the same authors+kinds, so those records are still
|
||||
* refreshed — only the deletion settle (negentropy-only) is skipped there.
|
||||
* Thin wrapper over quartz's [GrapeRankUpdater]: it reads every kind:10002 in the
|
||||
* store, inverts them into a `write-relay -> authors` map (the outbox model), and
|
||||
* runs one NIP-77 negentropy reconcile per write relay scoped to its authors —
|
||||
* bidirectional, settling deletions over the residual (its applyDown direction
|
||||
* downloads the relay's kind:5 when an uploaded record was rejected), and falling
|
||||
* back to a full paged download when a relay can't reconcile. This command only
|
||||
* parses flags and renders the [GrapeRankUpdater.Result] as text/JSON.
|
||||
*
|
||||
* Flags: `--timeout SECS` (per-group idle watchdog, default 30),
|
||||
* `--relay-concurrency N` (relays reconciled at once, default 4),
|
||||
@@ -512,53 +458,38 @@ object GrapeRankCommand {
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
val args = Args(rest)
|
||||
val timeoutMs = args.longFlag("timeout", 30L) * 1000
|
||||
val relayConcurrency = args.intFlag("relay-concurrency", 4).coerceAtLeast(1)
|
||||
val authorChunk = args.intFlag("author-chunk", 500).coerceAtLeast(1)
|
||||
val minAuthors = args.intFlag("min-authors", 1).coerceAtLeast(1)
|
||||
val reportLimit = args.intFlag("report-limit", 50).coerceAtLeast(0)
|
||||
val syncDeletions = !args.bool("no-sync-deletions")
|
||||
// Default is bidirectional; a single --down/--up narrows to that direction.
|
||||
val downFlag = args.bool("down")
|
||||
val upFlag = args.bool("up")
|
||||
val down = downFlag || !upFlag
|
||||
val up = upFlag || !downFlag
|
||||
|
||||
Context.openOrAnonymous(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
|
||||
// Every author's outbox relays, read from the kind:10002 already in the
|
||||
// store. Replaceable, so the store holds the latest per author; guard with
|
||||
// a createdAt max in case both an old and new copy linger.
|
||||
val latestRelayList = HashMap<HexKey, AdvertisedRelayListEvent>()
|
||||
for (event in ctx.store.query<Event>(Filter(kinds = listOf(AdvertisedRelayListEvent.KIND)))) {
|
||||
if (event !is AdvertisedRelayListEvent) continue
|
||||
val prev = latestRelayList[event.pubKey]
|
||||
if (prev == null || event.createdAt > prev.createdAt) latestRelayList[event.pubKey] = event
|
||||
}
|
||||
val updater =
|
||||
GrapeRankUpdater(
|
||||
client = ctx.client,
|
||||
store = ctx.store,
|
||||
config =
|
||||
GrapeRankUpdater.Config(
|
||||
down = downFlag || !upFlag,
|
||||
up = upFlag || !downFlag,
|
||||
syncDeletions = !args.bool("no-sync-deletions"),
|
||||
relayConcurrency = args.intFlag("relay-concurrency", 4),
|
||||
authorChunk = args.intFlag("author-chunk", 500),
|
||||
minAuthors = args.intFlag("min-authors", 1),
|
||||
idleTimeoutMs = args.longFlag("timeout", 30L) * 1000,
|
||||
),
|
||||
log = { System.err.println(it) },
|
||||
)
|
||||
|
||||
// Invert to write-relay -> authors (the outbox model). An author with no
|
||||
// write-marked relays contributes nothing (nowhere to reconcile them).
|
||||
val relayToAuthors = HashMap<NormalizedRelayUrl, MutableSet<HexKey>>()
|
||||
var authorsWithOutbox = 0
|
||||
for ((author, list) in latestRelayList) {
|
||||
val writes = list.writeRelaysNorm() ?: continue
|
||||
authorsWithOutbox++
|
||||
for (relay in writes) relayToAuthors.getOrPut(relay) { HashSet() }.add(author)
|
||||
}
|
||||
val result = updater.update()
|
||||
|
||||
// Drop relays hosting fewer than --min-authors of our authors; largest
|
||||
// first so the heaviest groups start while permits are free.
|
||||
val groups =
|
||||
relayToAuthors.entries
|
||||
.filter { it.value.size >= minAuthors }
|
||||
.sortedByDescending { it.value.size }
|
||||
|
||||
if (groups.isEmpty()) {
|
||||
if (result.relays == 0) {
|
||||
Output.emit(
|
||||
linkedMapOf<String, Any?>(
|
||||
"relay_lists_in_store" to latestRelayList.size,
|
||||
"authors_with_outbox" to authorsWithOutbox,
|
||||
"relay_lists_in_store" to result.relayListsInStore,
|
||||
"authors_with_outbox" to result.authorsWithOutbox,
|
||||
"relays" to 0,
|
||||
"note" to "no kind:10002 write relays in the local store — run `graperank sync` first",
|
||||
),
|
||||
@@ -566,99 +497,40 @@ object GrapeRankCommand {
|
||||
return 0
|
||||
}
|
||||
|
||||
val downloaded = AtomicInteger(0)
|
||||
val uploaded = AtomicInteger(0)
|
||||
val deletionsUp = AtomicInteger(0)
|
||||
val deletionsDown = AtomicInteger(0)
|
||||
val relaysOk = AtomicInteger(0)
|
||||
val relaysFailed = AtomicInteger(0)
|
||||
val relaysPaged = AtomicInteger(0)
|
||||
val perRelay = Collections.synchronizedList(ArrayList<Map<String, Any?>>())
|
||||
|
||||
val gate = Semaphore(relayConcurrency)
|
||||
coroutineScope {
|
||||
groups.forEach { (relay, authors) ->
|
||||
launch {
|
||||
gate.withPermit {
|
||||
val authorList = authors.toList()
|
||||
var relayDownloaded = 0
|
||||
var relayUploaded = 0
|
||||
var relayDelUp = 0
|
||||
var relayDelDown = 0
|
||||
var relayNeed = 0
|
||||
var relayHave = 0
|
||||
var failed = false
|
||||
var paged = false
|
||||
var error: String? = null
|
||||
|
||||
for (chunk in authorList.chunked(authorChunk)) {
|
||||
val filter = Filter(kinds = UPDATE_KINDS, authors = chunk)
|
||||
val res = syncGroup(ctx, relay, filter, down, up, syncDeletions, timeoutMs)
|
||||
relayDownloaded += res.downloaded
|
||||
relayUploaded += res.uploaded
|
||||
relayDelUp += res.deletionsSentUp
|
||||
relayDelDown += res.deletionsAppliedDown
|
||||
relayNeed += res.need
|
||||
relayHave += res.have
|
||||
if (res.pagedFallback) paged = true
|
||||
if (res.error != null) {
|
||||
failed = true
|
||||
error = res.error
|
||||
}
|
||||
}
|
||||
|
||||
downloaded.addAndGet(relayDownloaded)
|
||||
uploaded.addAndGet(relayUploaded)
|
||||
deletionsUp.addAndGet(relayDelUp)
|
||||
deletionsDown.addAndGet(relayDelDown)
|
||||
if (failed) relaysFailed.incrementAndGet() else relaysOk.incrementAndGet()
|
||||
if (paged) relaysPaged.incrementAndGet()
|
||||
|
||||
System.err.println(
|
||||
"[graperank update] ${relay.url}: ${authors.size} authors, " +
|
||||
"down $relayDownloaded, up $relayUploaded, del↑ $relayDelUp, del↓ $relayDelDown" +
|
||||
(if (paged) " (paged fallback)" else "") +
|
||||
(if (error != null) " (error: $error)" else ""),
|
||||
)
|
||||
perRelay.add(
|
||||
linkedMapOf<String, Any?>(
|
||||
"relay" to relay.url,
|
||||
"authors" to authors.size,
|
||||
"need" to relayNeed,
|
||||
"have" to relayHave,
|
||||
"downloaded" to relayDownloaded,
|
||||
"uploaded" to relayUploaded,
|
||||
"deletions_sent_up" to relayDelUp,
|
||||
"deletions_applied_down" to relayDelDown,
|
||||
"paged_fallback" to paged,
|
||||
"error" to error,
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Busiest relays first, capped so a many-thousand-relay run still emits a
|
||||
// bounded JSON object; totals below always cover every relay.
|
||||
val report =
|
||||
perRelay
|
||||
.sortedByDescending { (it["downloaded"] as Int) + (it["uploaded"] as Int) }
|
||||
result.perRelay
|
||||
.sortedByDescending { it.downloaded + it.uploaded }
|
||||
.take(reportLimit)
|
||||
.map {
|
||||
linkedMapOf<String, Any?>(
|
||||
"relay" to it.relay.url,
|
||||
"authors" to it.authors,
|
||||
"need" to it.need,
|
||||
"have" to it.have,
|
||||
"downloaded" to it.downloaded,
|
||||
"uploaded" to it.uploaded,
|
||||
"deletions_sent_up" to it.deletionsSentUp,
|
||||
"deletions_applied_down" to it.deletionsAppliedDown,
|
||||
"paged_fallback" to it.pagedFallback,
|
||||
"error" to it.error,
|
||||
)
|
||||
}
|
||||
|
||||
Output.emit(
|
||||
linkedMapOf<String, Any?>(
|
||||
"kinds" to UPDATE_KINDS,
|
||||
"relay_lists_in_store" to latestRelayList.size,
|
||||
"authors_with_outbox" to authorsWithOutbox,
|
||||
"relays" to groups.size,
|
||||
"relays_ok" to relaysOk.get(),
|
||||
"relays_failed" to relaysFailed.get(),
|
||||
"relays_paged_fallback" to relaysPaged.get(),
|
||||
"downloaded" to downloaded.get(),
|
||||
"uploaded" to uploaded.get(),
|
||||
"deletions_sent_up" to deletionsUp.get(),
|
||||
"deletions_applied_down" to deletionsDown.get(),
|
||||
"kinds" to GrapeRankUpdater.DEFAULT_KINDS,
|
||||
"relay_lists_in_store" to result.relayListsInStore,
|
||||
"authors_with_outbox" to result.authorsWithOutbox,
|
||||
"relays" to result.relays,
|
||||
"relays_ok" to result.relaysOk,
|
||||
"relays_failed" to result.relaysFailed,
|
||||
"relays_paged_fallback" to result.relaysPagedFallback,
|
||||
"downloaded" to result.downloaded,
|
||||
"uploaded" to result.uploaded,
|
||||
"deletions_sent_up" to result.deletionsSentUp,
|
||||
"deletions_applied_down" to result.deletionsAppliedDown,
|
||||
"report_limit" to reportLimit,
|
||||
"per_relay" to report,
|
||||
),
|
||||
@@ -667,146 +539,6 @@ object GrapeRankCommand {
|
||||
}
|
||||
}
|
||||
|
||||
/** Outcome of one relay/author-chunk reconcile in [update]. */
|
||||
private class GroupSyncResult(
|
||||
val downloaded: Int,
|
||||
val uploaded: Int,
|
||||
val deletionsSentUp: Int,
|
||||
val deletionsAppliedDown: Int,
|
||||
val need: Int,
|
||||
val have: Int,
|
||||
/** True when negentropy couldn't reconcile and we paged the filter instead. */
|
||||
val pagedFallback: Boolean,
|
||||
val error: String?,
|
||||
)
|
||||
|
||||
/**
|
||||
* Content pass + deletion settle for one relay scoped to [filter] (kinds +
|
||||
* one author chunk). Mirrors [SyncCommand]'s two-pass structure exactly — the
|
||||
* reconcile, windowing, and back-pressure all live in quartz's
|
||||
* [negentropyReconcile] / [negentropySettleDeletions]; this only routes ids to
|
||||
* [Context.drain] / [Context.publish]. Best-effort: a reconcile failure is
|
||||
* captured in [GroupSyncResult.error], never thrown, so one bad relay can't
|
||||
* abort the whole update.
|
||||
*/
|
||||
private suspend fun syncGroup(
|
||||
ctx: Context,
|
||||
relay: NormalizedRelayUrl,
|
||||
filter: Filter,
|
||||
down: Boolean,
|
||||
up: Boolean,
|
||||
syncDeletions: Boolean,
|
||||
timeoutMs: Long,
|
||||
): GroupSyncResult {
|
||||
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)
|
||||
|
||||
val result =
|
||||
try {
|
||||
coroutineScope {
|
||||
// needIds = relay has, we lack; haveIds = we have, relay lacks.
|
||||
val needBatches = Channel<List<HexKey>>(UPDATE_DOWNLOAD_WORKERS * 2)
|
||||
val haveBatches = Channel<List<HexKey>>(Channel.UNLIMITED)
|
||||
|
||||
val downloaders =
|
||||
List(UPDATE_DOWNLOAD_WORKERS) {
|
||||
launch {
|
||||
for (batch in needBatches) {
|
||||
downloaded.addAndGet(ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs).size)
|
||||
}
|
||||
}
|
||||
}
|
||||
val uploader =
|
||||
launch {
|
||||
for (batch in haveBatches) {
|
||||
for (id in batch) {
|
||||
val ev = localById[id] ?: continue
|
||||
if (ctx.publish(ev, setOf(relay)).values.any { it }) uploaded.incrementAndGet()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val reconcile =
|
||||
try {
|
||||
ctx.client.negentropyReconcile(
|
||||
relay = relay,
|
||||
filter = filter,
|
||||
localEntries = localEntries,
|
||||
batchSize = UPDATE_ID_CHUNK,
|
||||
idleTimeoutMs = timeoutMs,
|
||||
reconcileConcurrency = UPDATE_RECONCILE_CONCURRENCY,
|
||||
onHaveIds = if (up) { batch -> haveBatches.send(batch) } else null,
|
||||
onNeedIds = { batch -> if (down) needBatches.send(batch) },
|
||||
)
|
||||
} finally {
|
||||
needBatches.close()
|
||||
haveBatches.close()
|
||||
}
|
||||
|
||||
downloaders.joinAll()
|
||||
uploader.join()
|
||||
reconcile
|
||||
}
|
||||
} catch (e: NegentropySyncException) {
|
||||
// Negentropy couldn't reconcile this relay (no NIP-77, an over-cap
|
||||
// minimal window, a disconnect, …). Fall back to a full paged download
|
||||
// of the SAME authors+kinds so `update` still refreshes the records —
|
||||
// [Context.drainAllPages] walks each relay past its per-REQ cap and
|
||||
// verifies+stores every event. Only the download direction has a paging
|
||||
// analog; upload and the negentropy-only deletion settle are skipped.
|
||||
var pageError: String? = null
|
||||
if (down) {
|
||||
try {
|
||||
downloaded.addAndGet(ctx.drainAllPages(mapOf(relay to listOf(filter)), timeoutMs).size)
|
||||
} catch (pe: Exception) {
|
||||
pageError = "negentropy: ${e.message}; page fallback: ${pe::class.simpleName}: ${pe.message}"
|
||||
}
|
||||
}
|
||||
return GroupSyncResult(
|
||||
downloaded = downloaded.get(),
|
||||
uploaded = uploaded.get(),
|
||||
deletionsSentUp = 0,
|
||||
deletionsAppliedDown = 0,
|
||||
need = 0,
|
||||
have = 0,
|
||||
pagedFallback = true,
|
||||
error = pageError,
|
||||
)
|
||||
}
|
||||
|
||||
val deletions =
|
||||
if (syncDeletions) {
|
||||
ctx.client.negentropySettleDeletions(
|
||||
relay = relay,
|
||||
filter = filter,
|
||||
store = ctx.store,
|
||||
sendUp = down,
|
||||
applyDown = up,
|
||||
batchSize = UPDATE_ID_CHUNK,
|
||||
idleTimeoutMs = timeoutMs,
|
||||
maxRounds = UPDATE_MAX_DELETION_ROUNDS,
|
||||
reconcileConcurrency = UPDATE_RECONCILE_CONCURRENCY,
|
||||
)
|
||||
} else {
|
||||
DeletionSettleResult(0, 0, 0)
|
||||
}
|
||||
|
||||
return GroupSyncResult(
|
||||
downloaded = downloaded.get(),
|
||||
uploaded = uploaded.get(),
|
||||
deletionsSentUp = deletions.sentUp,
|
||||
deletionsAppliedDown = deletions.appliedDown,
|
||||
need = result.needCount,
|
||||
have = result.haveCount,
|
||||
pagedFallback = false,
|
||||
error = null,
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Build + sign one kind:30382 [ContactCardEvent] per (target, rank), fanned
|
||||
* out across CPU cores (id-hash + Schnorr sign is CPU-bound). The signed
|
||||
|
||||
+425
@@ -0,0 +1,425 @@
|
||||
/*
|
||||
* 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.experimental.graperank
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcile
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySettleDeletions
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
|
||||
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.IdAndTime
|
||||
import com.vitorpamplona.quartz.nip01Core.store.verifyAndInsert
|
||||
import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent
|
||||
import com.vitorpamplona.quartz.nip56Reports.ReportEvent
|
||||
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.joinAll
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Semaphore
|
||||
import kotlinx.coroutines.sync.withPermit
|
||||
import kotlin.concurrent.atomics.AtomicInt
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
|
||||
/**
|
||||
* Store-driven, outbox-model refresh of the record kinds a [GrapeRank] score is a
|
||||
* function of — the profiles (kind:0), follows (kind:3), outbox relay lists
|
||||
* (kind:10002), and reports (kind:1984) of every author already known to the
|
||||
* local [store].
|
||||
*
|
||||
* Where [GrapeRankDataCrawler] discovers the graph by walking follows outward from
|
||||
* an observer, this refreshes what is *already* known: it reads every kind:10002 in
|
||||
* the store, inverts them into a `write-relay -> authors` map (the outbox model — an
|
||||
* author's events live on the relays they write to), then runs one NIP-77 negentropy
|
||||
* reconcile per write relay scoped to exactly the authors who publish there. So each
|
||||
* relay is asked only for the authors it hosts, and each author is reconciled only
|
||||
* against their own relays. Run it periodically to keep a scored network current
|
||||
* without paying a full from-scratch crawl.
|
||||
*
|
||||
* Each per-relay group is bidirectional by default ([Config.down] / [Config.up]):
|
||||
* - **down** downloads records the relay has and the store lacks (by-id fetch,
|
||||
* verified + inserted into [store]);
|
||||
* - **up** uploads records the store has and the relay lacks.
|
||||
*
|
||||
* After the content pass each group settles deletions over the reconcile residual
|
||||
* ([negentropySettleDeletions], [Config.syncDeletions]). Its **applyDown** direction
|
||||
* is the "download the deletion when our upload was rejected" case: an event pushed up
|
||||
* that the relay keeps rejecting (it deleted it) surfaces as a residual *have*, so the
|
||||
* relay's covering kind:5 is pulled down and applied locally and the store drops the
|
||||
* retracted record. The **sendUp** direction publishes the store's covering deletions
|
||||
* for records deleted locally that the relay still serves. Cheap: it works off the
|
||||
* small residual, not the whole set.
|
||||
*
|
||||
* If negentropy can't reconcile a relay (no NIP-77 support, an over-cap minimal window,
|
||||
* a mid-sync disconnect, …) and [Config.pageFallback] is on, the group falls back to a
|
||||
* full paged download ([fetchAllPages]) of the same authors+kinds so those records are
|
||||
* still refreshed — only the negentropy-only deletion settle is skipped there.
|
||||
*
|
||||
* Transport-agnostic within quartz: it takes an [INostrClient] and an [IEventStore].
|
||||
* Progress is emitted through [log]; a headless caller routes it to stderr, a UI ignores it.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
class GrapeRankUpdater(
|
||||
private val client: INostrClient,
|
||||
private val store: IEventStore,
|
||||
private val config: Config = Config(),
|
||||
private val log: (String) -> Unit = {},
|
||||
) {
|
||||
/**
|
||||
* @param kinds the record kinds refreshed per author (default: the WoT set
|
||||
* 0 / 3 / 10002 / 1984).
|
||||
* @param down download records the relay has that the store lacks.
|
||||
* @param up upload records the store has that the relay lacks (also arms the
|
||||
* deletion **applyDown** path — a rejected upload pulls the relay's kind:5 down).
|
||||
* @param syncDeletions run the deletion settle over the reconcile residual.
|
||||
* @param pageFallback page the filter when negentropy can't reconcile a relay.
|
||||
* @param idChunk ids per reconcile chunk and per by-id fetch.
|
||||
* @param downloadWorkers concurrent by-id download fetches per group.
|
||||
* @param reconcileConcurrency overlapped `created_at`-window reconciles after an over-cap split.
|
||||
* @param maxDeletionRounds hard cap on deletion-settle rounds (converges in 1–2).
|
||||
* @param relayConcurrency write relays reconciled at once.
|
||||
* @param authorChunk authors per reconcile filter (a relay with more is split into several).
|
||||
* @param minAuthors skip relays hosting fewer than this many of the store's authors.
|
||||
* @param idleTimeoutMs idle watchdog for reconciles / fetches / pages.
|
||||
* @param publishTimeoutSecs OK-confirmation wait per uploaded event.
|
||||
*/
|
||||
class Config(
|
||||
val kinds: List<Int> = DEFAULT_KINDS,
|
||||
val down: Boolean = true,
|
||||
val up: Boolean = true,
|
||||
val syncDeletions: Boolean = true,
|
||||
val pageFallback: Boolean = true,
|
||||
val idChunk: Int = 500,
|
||||
val downloadWorkers: Int = 4,
|
||||
val reconcileConcurrency: Int = 2,
|
||||
val maxDeletionRounds: Int = 4,
|
||||
val relayConcurrency: Int = 4,
|
||||
val authorChunk: Int = 500,
|
||||
val minAuthors: Int = 1,
|
||||
val idleTimeoutMs: Long = 30_000L,
|
||||
val publishTimeoutSecs: Long = 15,
|
||||
)
|
||||
|
||||
/** Per-write-relay outcome of an [update]. */
|
||||
class RelayResult(
|
||||
val relay: NormalizedRelayUrl,
|
||||
val authors: Int,
|
||||
val need: Int,
|
||||
val have: Int,
|
||||
val downloaded: Int,
|
||||
val uploaded: Int,
|
||||
val deletionsSentUp: Int,
|
||||
val deletionsAppliedDown: Int,
|
||||
val pagedFallback: Boolean,
|
||||
/** null on success; the negentropy (and any page-fallback) failure otherwise. */
|
||||
val error: String?,
|
||||
)
|
||||
|
||||
/** Aggregate outcome of an [update], plus the per-relay breakdown. */
|
||||
class Result(
|
||||
val relayListsInStore: Int,
|
||||
val authorsWithOutbox: Int,
|
||||
val relays: Int,
|
||||
val relaysOk: Int,
|
||||
val relaysFailed: Int,
|
||||
val relaysPagedFallback: Int,
|
||||
val downloaded: Int,
|
||||
val uploaded: Int,
|
||||
val deletionsSentUp: Int,
|
||||
val deletionsAppliedDown: Int,
|
||||
val perRelay: List<RelayResult>,
|
||||
)
|
||||
|
||||
/**
|
||||
* Group the store's authors by their kind:10002 write relays (the outbox model).
|
||||
* The latest kind:10002 per author wins; an author with no write-marked relays
|
||||
* contributes nothing (there is nowhere to reconcile them). Public so callers can
|
||||
* inspect the plan (relay count, largest groups) before running [update].
|
||||
*/
|
||||
suspend fun writeRelayGroups(): Map<NormalizedRelayUrl, Set<HexKey>> = groupByWriteRelay(loadLatestRelayLists())
|
||||
|
||||
/** Latest kind:10002 per author from the store (replaceable — newest createdAt wins). */
|
||||
private suspend fun loadLatestRelayLists(): Map<HexKey, AdvertisedRelayListEvent> {
|
||||
val latest = HashMap<HexKey, AdvertisedRelayListEvent>()
|
||||
for (event in store.query<Event>(Filter(kinds = listOf(AdvertisedRelayListEvent.KIND)))) {
|
||||
if (event !is AdvertisedRelayListEvent) continue
|
||||
val prev = latest[event.pubKey]
|
||||
if (prev == null || event.createdAt > prev.createdAt) latest[event.pubKey] = event
|
||||
}
|
||||
return latest
|
||||
}
|
||||
|
||||
/** Invert the per-author relay lists into `write-relay -> authors`. */
|
||||
private fun groupByWriteRelay(latest: Map<HexKey, AdvertisedRelayListEvent>): Map<NormalizedRelayUrl, Set<HexKey>> {
|
||||
val relayToAuthors = HashMap<NormalizedRelayUrl, MutableSet<HexKey>>()
|
||||
for ((author, list) in latest) {
|
||||
val writes = list.writeRelaysNorm() ?: continue
|
||||
for (relay in writes) relayToAuthors.getOrPut(relay) { HashSet() }.add(author)
|
||||
}
|
||||
return relayToAuthors
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the full outbox-model refresh: [writeRelayGroups] then one per-relay sync
|
||||
* for each group hosting at least [Config.minAuthors] authors, up to
|
||||
* [Config.relayConcurrency] relays at once (largest groups first). Best-effort —
|
||||
* a relay that fails is recorded in its [RelayResult.error] and never aborts the run.
|
||||
*/
|
||||
suspend fun update(): Result {
|
||||
val latest = loadLatestRelayLists()
|
||||
val groups = groupByWriteRelay(latest)
|
||||
val authorsWithOutbox = groups.values.flatMapTo(HashSet()) { it }.size
|
||||
|
||||
val plan =
|
||||
groups.entries
|
||||
.filter { it.value.size >= config.minAuthors }
|
||||
.sortedByDescending { it.value.size }
|
||||
|
||||
val perRelay =
|
||||
if (plan.isEmpty()) {
|
||||
emptyList()
|
||||
} else {
|
||||
val gate = Semaphore(config.relayConcurrency.coerceAtLeast(1))
|
||||
coroutineScope {
|
||||
plan
|
||||
.map { (relay, authors) ->
|
||||
async { gate.withPermit { syncRelay(relay, authors) } }
|
||||
}.awaitAll()
|
||||
}
|
||||
}
|
||||
|
||||
return Result(
|
||||
relayListsInStore = latest.size,
|
||||
authorsWithOutbox = authorsWithOutbox,
|
||||
relays = plan.size,
|
||||
relaysOk = perRelay.count { it.error == null },
|
||||
relaysFailed = perRelay.count { it.error != null },
|
||||
relaysPagedFallback = perRelay.count { it.pagedFallback },
|
||||
downloaded = perRelay.sumOf { it.downloaded },
|
||||
uploaded = perRelay.sumOf { it.uploaded },
|
||||
deletionsSentUp = perRelay.sumOf { it.deletionsSentUp },
|
||||
deletionsAppliedDown = perRelay.sumOf { it.deletionsAppliedDown },
|
||||
perRelay = perRelay,
|
||||
)
|
||||
}
|
||||
|
||||
/** Sync one write relay by folding each [Config.authorChunk]-sized author slice. */
|
||||
private suspend fun syncRelay(
|
||||
relay: NormalizedRelayUrl,
|
||||
authors: Set<HexKey>,
|
||||
): RelayResult {
|
||||
var downloaded = 0
|
||||
var uploaded = 0
|
||||
var delUp = 0
|
||||
var delDown = 0
|
||||
var need = 0
|
||||
var have = 0
|
||||
var paged = false
|
||||
var error: String? = null
|
||||
|
||||
for (chunk in authors.toList().chunked(config.authorChunk.coerceAtLeast(1))) {
|
||||
val filter = Filter(kinds = config.kinds, authors = chunk)
|
||||
val res = syncGroup(relay, filter)
|
||||
downloaded += res.downloaded
|
||||
uploaded += res.uploaded
|
||||
delUp += res.deletionsSentUp
|
||||
delDown += res.deletionsAppliedDown
|
||||
need += res.need
|
||||
have += res.have
|
||||
if (res.pagedFallback) paged = true
|
||||
if (res.error != null) error = res.error
|
||||
}
|
||||
|
||||
log(
|
||||
"[graperank update] ${relay.url}: ${authors.size} authors, " +
|
||||
"down $downloaded, up $uploaded, del↑ $delUp, del↓ $delDown" +
|
||||
(if (paged) " (paged fallback)" else "") +
|
||||
(if (error != null) " (error: $error)" else ""),
|
||||
)
|
||||
return RelayResult(relay, authors.size, need, have, downloaded, uploaded, delUp, delDown, paged, error)
|
||||
}
|
||||
|
||||
/** One relay + one author chunk. Mirrors the two-pass content+deletion sync. */
|
||||
private suspend fun syncGroup(
|
||||
relay: NormalizedRelayUrl,
|
||||
filter: Filter,
|
||||
): GroupResult {
|
||||
val localEvents = store.query<Event>(filter)
|
||||
val localById = localEvents.associateBy { it.id }
|
||||
val localEntries = localEvents.map { IdAndTime(it.createdAt, it.id) }
|
||||
|
||||
val downloaded = AtomicInt(0)
|
||||
val uploaded = AtomicInt(0)
|
||||
|
||||
val reconcileResult =
|
||||
try {
|
||||
coroutineScope {
|
||||
// needIds = relay has, store lacks; haveIds = store has, relay lacks.
|
||||
val needBatches = Channel<List<HexKey>>(config.downloadWorkers * 2)
|
||||
val haveBatches = Channel<List<HexKey>>(Channel.UNLIMITED)
|
||||
|
||||
val downloaders =
|
||||
List(config.downloadWorkers.coerceAtLeast(1)) {
|
||||
launch {
|
||||
for (batch in needBatches) {
|
||||
for (event in client.fetchAll(relay, Filter(ids = batch), config.idleTimeoutMs)) {
|
||||
if (store.verifyAndInsert(event)) downloaded.addAndFetch(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
val uploader =
|
||||
launch {
|
||||
for (batch in haveBatches) {
|
||||
for (id in batch) {
|
||||
val ev = localById[id] ?: continue
|
||||
if (client.publishAndConfirm(ev, setOf(relay), config.publishTimeoutSecs)) uploaded.addAndFetch(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val result =
|
||||
try {
|
||||
client.negentropyReconcile(
|
||||
relay = relay,
|
||||
filter = filter,
|
||||
localEntries = localEntries,
|
||||
batchSize = config.idChunk,
|
||||
idleTimeoutMs = config.idleTimeoutMs,
|
||||
reconcileConcurrency = config.reconcileConcurrency,
|
||||
onHaveIds = if (config.up) { batch -> haveBatches.send(batch) } else null,
|
||||
onNeedIds = { batch -> if (config.down) needBatches.send(batch) },
|
||||
)
|
||||
} finally {
|
||||
needBatches.close()
|
||||
haveBatches.close()
|
||||
}
|
||||
|
||||
downloaders.joinAll()
|
||||
uploader.join()
|
||||
result
|
||||
}
|
||||
} catch (e: NegentropySyncException) {
|
||||
// Negentropy couldn't reconcile — page the same authors+kinds so the
|
||||
// records still refresh. Deletion settle is negentropy-only, so skipped.
|
||||
var pageError: String? = e.message ?: "negentropy sync failed"
|
||||
if (config.pageFallback && config.down) {
|
||||
pageError =
|
||||
try {
|
||||
downloaded.addAndFetch(pageDownload(relay, filter))
|
||||
null
|
||||
} catch (pe: Exception) {
|
||||
"negentropy: ${e.message}; page fallback: ${pe::class.simpleName}: ${pe.message}"
|
||||
}
|
||||
}
|
||||
return GroupResult(downloaded.load(), uploaded.load(), 0, 0, 0, 0, pagedFallback = true, error = pageError)
|
||||
}
|
||||
|
||||
val deletions =
|
||||
if (config.syncDeletions) {
|
||||
client.negentropySettleDeletions(
|
||||
relay = relay,
|
||||
filter = filter,
|
||||
store = store,
|
||||
sendUp = config.down,
|
||||
applyDown = config.up,
|
||||
batchSize = config.idChunk,
|
||||
idleTimeoutMs = config.idleTimeoutMs,
|
||||
maxRounds = config.maxDeletionRounds,
|
||||
reconcileConcurrency = config.reconcileConcurrency,
|
||||
)
|
||||
} else {
|
||||
null
|
||||
}
|
||||
|
||||
return GroupResult(
|
||||
downloaded = downloaded.load(),
|
||||
uploaded = uploaded.load(),
|
||||
deletionsSentUp = deletions?.sentUp ?: 0,
|
||||
deletionsAppliedDown = deletions?.appliedDown ?: 0,
|
||||
need = reconcileResult.needCount,
|
||||
have = reconcileResult.haveCount,
|
||||
pagedFallback = false,
|
||||
error = null,
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Paged fallback: walk [relay] past its per-REQ cap for [filter], verifying and
|
||||
* inserting each event into [store]. [fetchAllPages]'s `onEvent` can't suspend, so
|
||||
* events funnel through a bounded channel to a single inserter. Returns how many
|
||||
* were newly stored.
|
||||
*/
|
||||
private suspend fun pageDownload(
|
||||
relay: NormalizedRelayUrl,
|
||||
filter: Filter,
|
||||
): Int {
|
||||
val stored = AtomicInt(0)
|
||||
val events = Channel<Event>(Channel.UNLIMITED)
|
||||
coroutineScope {
|
||||
val inserter =
|
||||
launch {
|
||||
for (event in events) {
|
||||
if (store.verifyAndInsert(event)) stored.addAndFetch(1)
|
||||
}
|
||||
}
|
||||
try {
|
||||
client.fetchAllPages(relay, listOf(filter), config.idleTimeoutMs) { event -> events.trySend(event) }
|
||||
} finally {
|
||||
events.close()
|
||||
}
|
||||
inserter.join()
|
||||
}
|
||||
return stored.load()
|
||||
}
|
||||
|
||||
private class GroupResult(
|
||||
val downloaded: Int,
|
||||
val uploaded: Int,
|
||||
val deletionsSentUp: Int,
|
||||
val deletionsAppliedDown: Int,
|
||||
val need: Int,
|
||||
val have: Int,
|
||||
val pagedFallback: Boolean,
|
||||
val error: String?,
|
||||
)
|
||||
|
||||
companion object {
|
||||
/** The record kinds a GrapeRank score is a function of. */
|
||||
val DEFAULT_KINDS =
|
||||
listOf(
|
||||
MetadataEvent.KIND, // 0 — profiles
|
||||
ContactListEvent.KIND, // 3 — follows
|
||||
AdvertisedRelayListEvent.KIND, // 10002 — outbox relay lists
|
||||
ReportEvent.KIND, // 1984 — reports
|
||||
)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user