From c9ee79beba8ec6978b06ce63c26e34d020bd309d Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Jul 2026 19:10:33 +0000 Subject: [PATCH] feat(graperank): log slow/timed-out relays with their query under --diagnose MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a --diagnose slow-relay log: every content drain that reaches its terminal (EOSE or timeout) slower than SLOW_DRAIN_LOG_MS, or times out entirely, is recorded with the offending relay URL, the failure/EOSE reason, elapsed ms, and the exact filter shape (kinds + author count + first authors). This lets a human replay that precise REQ later to understand why the relay lags. Gated on --diagnose so there is no per-group timing/collection overhead otherwise. Make the content-drain fan-out configurable via a new --drain-concurrency flag (Config.drainConcurrency), replacing the DRAIN_CONCURRENCY constant. Default stays at the validated 24: an A/B at 64 ran ~2x slower with more dead relays (a higher global fan-out re-floods busy hubs faster than the per-relay demotion catches up), so the flag is a probe knob, not a speedup. Client WebSocket pings were also tried and reverted — busy-but-alive relays don't reliably pong while their query handler runs, so pinging just cut them as dead. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01RWk2ZMrGBSr4WenKgwqmbB --- .../amethyst/cli/commands/GrapeRankCommand.kt | 6 +++ .../graperank/GrapeRankDataCrawler.kt | 40 ++++++++++++++----- 2 files changed, 37 insertions(+), 9 deletions(-) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt index e9fa3e8f84..d6f483b0ef 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt @@ -127,6 +127,11 @@ object GrapeRankCommand { // forces the per-event insert path (baseline); higher amortizes the SQLite // transaction + writer-mutex cost across the batch. val insertBatch = args.intFlag("insert-batch", 500) + // How many outbox batches drain in parallel (the worker-pool size). 24 is the + // validated default; higher fan-out re-floods busy hubs faster than the + // per-relay demotion catches up (an A/B at 64 was ~2x slower with MORE dead + // relays), so raise it only to probe specific slow relays. + val drainConcurrency = args.intFlag("drain-concurrency", 24) val doPublish = args.bool("publish") // Publish cutoff: only cards with rank >= this are published; existing // cards for targets below it (or gone from the graph) are retracted. Rank @@ -185,6 +190,7 @@ object GrapeRankCommand { timeoutMs = timeoutMs, diagnose = diagnose, insertBatchSize = insertBatch, + drainConcurrency = drainConcurrency, ), log = { System.err.println(it) }, ) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankDataCrawler.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankDataCrawler.kt index 02c791ebfa..e15634abd3 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankDataCrawler.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankDataCrawler.kt @@ -117,6 +117,12 @@ class GrapeRankDataCrawler( * [IEventStore.batchInsert]. The outbox model streams the same events from * many relays through a single SQLite writer, so batching amortizes the * per-transaction + writer-mutex cost across the batch (coerced to `>= 1`). + * @param drainConcurrency how many outbox batches drain at once (the worker + * pool size). A GLOBAL bound (memory / open sockets); the per-relay + * concurrent-sub cap is enforced separately by [AdaptiveRelayLimiter]. Keep it + * moderate: a higher global fan-out re-floods busy hubs faster than demotion + * catches up (an A/B at 64 ran ~2x slower with more dead relays), so 24 is the + * validated default and raising it is a probe, not a speedup. */ class Config( val relayListDiscoveryRelays: Set, @@ -126,6 +132,7 @@ class GrapeRankDataCrawler( val timeoutMs: Long = 10_000, val diagnose: Boolean = false, val insertBatchSize: Int = 500, + val drainConcurrency: Int = 24, ) /** What the crawl fetched — the counters the caller reports and the graph is built from. */ @@ -527,7 +534,7 @@ class GrapeRankDataCrawler( // on the producer (keeps writeRelayFreq serial) and ingest runs // only on the consumer (keeps done/builder/hopOf serial), now // overlapped with draining instead of blocked behind each batch. - val routed = Channel, Map>>>(DRAIN_CONCURRENCY * 2) + val routed = Channel, Map>>>(config.drainConcurrency * 2) val drainedOut = Channel(Channel.UNLIMITED) coroutineScope { // Producer: route each batch by outbox (serial), backpressured @@ -545,7 +552,7 @@ class GrapeRankDataCrawler( // that cleanly EOSE'd, so the consumer can tell "answered empty" // from "timed out" per user. val workers = - List(DRAIN_CONCURRENCY) { + List(config.drainConcurrency) { launch { for ((batch, filters) in routed) { val dead = HashMap() @@ -635,7 +642,7 @@ class GrapeRankDataCrawler( ) log( "[graperank] write path: $stored events stored, verify ${verifyMs}ms + insert ${insertMs}ms " + - "(summed across ${DRAIN_CONCURRENCY} consumers, batch=${config.insertBatchSize})", + "(summed across ${config.drainConcurrency} consumers, batch=${config.insertBatchSize})", ) return Stats( rounds = rounds, @@ -713,6 +720,10 @@ class GrapeRankDataCrawler( // couldn't connect). A relay absent from this set answered definitively — // so an author it was asked for but didn't return is one it simply lacks. val notAnswered = ConcurrentSet() + // --diagnose: which relays were slow (or timed out) and on which query, so a + // human can replay that exact REQ later to understand the slowness. Null when + // diagnosis is off (no per-group timing/collection overhead). + val slowDrains = if (config.diagnose) ConcurrentSet() else null val collected = mutableListOf>() coroutineScope { @@ -801,7 +812,9 @@ class GrapeRankDataCrawler( } client.subscribe(subId, mapOf(subRelay to groupFilters), groupListener) try { + val gMark = TimeSource.Monotonic.markNow() val reason = withTimeoutOrNull(config.timeoutMs) { done.await() } ?: "timeout" + val elapsedMs = gMark.elapsedNow().inWholeMilliseconds if (reason == "timeout") timedOut.add(subRelay) if (reason != "eose") notAnswered.add(subRelay) classifyDrainFailure(reason)?.let { kind -> @@ -809,6 +822,16 @@ class GrapeRankDataCrawler( if (a == DrainFailure.HARD || b == DrainFailure.HARD) DrainFailure.HARD else DrainFailure.TRANSIENT } } + // Record slow/timed-out REQs with their exact query so a + // human can replay them later and see why the relay lags. + if (slowDrains != null && (reason == "timeout" || elapsedMs > SLOW_DRAIN_LOG_MS)) { + val authors = groupFilters.flatMap { it.authors.orEmpty() } + val kinds = groupFilters.flatMap { it.kinds.orEmpty() }.distinct() + slowDrains.add( + "[slow-relay] ${subRelay.url} $reason in ${elapsedMs}ms | kinds=$kinds authors=${authors.size}: " + + authors.take(30).joinToString(",") + (if (authors.size > 30) ",…" else ""), + ) + } } finally { client.unsubscribe(subId) } @@ -826,6 +849,7 @@ class GrapeRankDataCrawler( val detail = stalled.take(12).joinToString(", ") { "${it.url}(${eventsPer[it] ?: 0}ev)" } log("[drain] timeout ${config.timeoutMs}ms: ${stalled.size} slow(no EOSE)" + (if (detail.isNotEmpty()) " | slow: $detail" else "")) } + slowDrains?.snapshot()?.forEach { log(it) } deadOut?.putAll(failures.snapshot()) answeredOut?.addAll(filters.keys.filter { it !in notAnswered }) return collected @@ -852,6 +876,10 @@ class GrapeRankDataCrawler( // message cap most relays enforce. drainGated groups filters to stay within. private const val MAX_REQ_ENTRIES = 2500 + // --diagnose: a REQ that takes longer than this to reach a terminal (EOSE or + // timeout) is logged with its relay + filter, so slow relays can be replayed. + private const val SLOW_DRAIN_LOG_MS = 4000L + // Times we re-query an unreachable user's outbox before giving up, so the // crawl still terminates on a finite graph. private const val MAX_OUTBOX_ATTEMPTS = 3 @@ -861,12 +889,6 @@ class GrapeRankDataCrawler( // (~250/drain succeeds, ~17k fails); keep the fan-out small. private const val USER_BATCH = 256 - // Global content-drain fan-out — how many outbox batches we drain at once. A - // GLOBAL bound (memory / open sockets); the per-relay concurrency limit is - // enforced separately by AdaptiveRelayLimiter. A higher global fan-out - // re-floods busy hubs faster than demotion catches up, so keep it moderate. - private const val DRAIN_CONCURRENCY = 24 - // Sharded backbone sweep: split the still-missing authors into SHARD_RELAYS // lists, one per top relay, rotating up to SHARD_ROTATIONS times; once the // remainder drops below SHARD_BROADCAST_THRESHOLD, broadcast it at once.