From 59d5fc752cc72036908467c8f0074eb6a1b43544 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Jul 2026 13:10:58 +0000 Subject: [PATCH] perf(cli): split rate-limit from subscription-count limit in the crawl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A relay pushes back for two different reasons that need two different fixes, and treating them the same mishandles the relay: - a subscription-COUNT cap ("too many subscriptions", "maximum concurrent subscription count") is fixed by fewer CONCURRENT subs — demote the per-relay concurrency cap (100 -> 20 -> 10), as before; - a RATE limit ("rate-limited: too many messages", "burst exhausted") is too many subscription CHANGES per second — fewer concurrent subs don't help; the fix is to SPACE the REQs out in time. AdaptiveRelayLimiter now routes each complaint to its own actuator by matching the notice text, and adds a per-relay rate gate: a growing minimum interval between subscription opens (250ms -> 500ms -> 1s -> 2s), enforced in withPermit before the concurrency permit. A relay can be under both controls at once. The snapshot reports each dimension separately. --- .../amethyst/cli/AdaptiveRelayLimiter.kt | 177 ++++++++++++------ 1 file changed, 122 insertions(+), 55 deletions(-) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/AdaptiveRelayLimiter.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/AdaptiveRelayLimiter.kt index d57e19c034..b2f39fc534 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/AdaptiveRelayLimiter.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/AdaptiveRelayLimiter.kt @@ -27,51 +27,72 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.delay import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong /** - * Adaptive per-relay concurrent-subscription cap. + * Adaptive per-relay back-pressure with TWO independent controls, because relays + * push back for two different reasons that need two different responses: * - * Every relay starts with a generous cap ([startCap], default 100) — we assume a - * relay can take as many concurrent REQs as we throw at it until it tells us - * otherwise. When a relay complains about concurrency (a `CLOSED rate-limited`, - * or a `NOTICE` like "too many concurrent REQs" / "too many subscriptions" / - * "burst exhausted"), we demote *that relay only* down the [ladder] - * (100 → 20 → 10). A well-behaved relay keeps the full cap; only the ones that - * push back get throttled, and only as far as they keep pushing. + * 1. **Subscription-count limit** — a max on how many subscriptions may be OPEN + * at once ("too many subscriptions", "maximum concurrent subscription count", + * "number of subscriptions exceeds limit"). The fix is fewer *concurrent* + * subs, so we demote the relay's concurrency cap down [subLadder] + * (100 → 20 → 10). + * 2. **Rate limit** — too many subscription *changes per second* ("rate-limited: + * too many messages", "burst exhausted", "slow down"). Fewer concurrent subs + * wouldn't help; the fix is to *space the REQs out in time*, so we impose a + * minimum interval between opens to that relay, growing it up [rateLadder] + * (250ms → 500ms → 1s → 2s). * - * This replaces a single blunt global concurrency number with per-relay - * back-pressure: the crawl can fan out widely across the many relays that don't - * mind, while automatically easing off the few busy hubs that do — exactly the - * signals [RelayDiagnostics] already observes, here turned into an actuator. + * Mixing the two mishandles the relay: capping concurrency does nothing for a + * rate limit, and slowing the rate does nothing for a subscription-count cap. So + * each complaint is routed to its own actuator by matching the notice text. * - * Registered as a [RelayConnectionListener] on the shared client, so demotions + * A well-behaved relay starts at [startCap] concurrent subs with no rate delay, + * and only the ones that push back get throttled — each only as far, and in the + * dimension, they keep pushing. + * + * Registered as a [RelayConnectionListener] on the shared client, so both signals * are driven straight off the incoming NOTICE/CLOSED frames (which fire on the * per-relay socket threads — all state here is concurrent). Drains gate through * [withPermit]; [Context.drain]'s `gatePerRelay` path holds a relay's permit for - * the lifetime of that relay's subscription, so at most `cap` of our - * subscriptions are ever open on it at once. + * the lifetime of that relay's subscription, and passes the rate gate before it + * opens, so we respect both limits at once. */ class AdaptiveRelayLimiter( private val startCap: Int = 100, - private val ladder: List = listOf(20, 10), + private val subLadder: List = listOf(20, 10), + private val rateLadder: List = listOf(250L, 500L, 1000L, 2000L), ) : RelayConnectionListener { private val gates = ConcurrentHashMap() - // How many concurrency complaints we've acted on per relay (== index+1 into - // the ladder). Capped at ladder.size: past the floor we stop demoting. - private val demotions = ConcurrentHashMap() + // Concurrency-cap demotions per relay (== index+1 into subLadder). Capped at + // subLadder.size: past the floor we stop demoting. + private val subDemotions = ConcurrentHashMap() + + // Rate-limit state per relay: how far down rateLadder we've stepped, the + // current min interval between opens, and the next epoch-ms an open may fire. + private val rateSteps = ConcurrentHashMap() + private val rateDelayMs = ConcurrentHashMap() + private val nextAllowedAtMs = ConcurrentHashMap() private fun gate(relay: NormalizedRelayUrl): Gate = gates.getOrPut(relay) { Gate(startCap) } - /** Run [block] holding one of [relay]'s permits, respecting its current cap. */ + /** + * Run [block] against [relay] respecting both limits: first wait out any rate + * delay (spacing opens in time), then hold one of the relay's concurrency + * permits for the duration. + */ suspend fun withPermit( relay: NormalizedRelayUrl, block: suspend () -> T, ): T { + rateGate(relay) val g = gate(relay) g.acquire() try { @@ -81,52 +102,89 @@ class AdaptiveRelayLimiter( } } + /** If [relay] is rate-limited, reserve and wait for its next allowed open slot. */ + private suspend fun rateGate(relay: NormalizedRelayUrl) { + val delayMs = rateDelayMs[relay] ?: return + if (delayMs <= 0L) return + val now = System.currentTimeMillis() + // Atomically claim the next slot: my turn is max(prevSlot, now); the next + // caller can't fire until delayMs after me. Serializes opens to this relay + // at one per delayMs, in arrival order. + val slot = nextAllowedAtMs.getOrPut(relay) { AtomicLong(now) } + var myTurn: Long + while (true) { + val prev = slot.get() + myTurn = maxOf(prev, now) + if (slot.compareAndSet(prev, myTurn + delayMs)) break + } + val wait = myTurn - now + if (wait > 0) delay(wait) + } + override fun onIncomingMessage( relay: IRelayClient, msgStr: String, msg: Message, ) { - when (msg) { - is ClosedMessage -> if (isConcurrencyComplaint(msg.message)) demote(relay.url) - is NoticeMessage -> if (isConcurrencyComplaint(msg.message)) demote(relay.url) - else -> Unit - } - } - - /** Step [relay] one rung down the cap ladder, unless it's already at the floor. */ - private fun demote(relay: NormalizedRelayUrl) { - // Fast path: relays flood identical NOTICEs, so bail once at the floor - // instead of counting them all (the demotion is monotonic and idempotent). - if ((demotions[relay] ?: 0) >= ladder.size) return - val step = demotions.merge(relay, 1, Int::plus)!! - val cap = ladder[(step - 1).coerceIn(0, ladder.size - 1)] - gate(relay).lower(cap) - if (step <= ladder.size) { - System.err.println("[limiter] ${relay.url} capped at $cap concurrent subs (complaint #$step)") - } - } - - private fun isConcurrencyComplaint(text: String): Boolean { + val text = + when (msg) { + is ClosedMessage -> msg.message + is NoticeMessage -> msg.message + else -> return + } val t = text.lowercase() - return CONCURRENCY_MARKERS.any { it in t } + // Route each complaint to the matching actuator. Not mutually exclusive: + // if a relay somehow reports both, we act on both (they don't conflict). + if (RATE_LIMIT_MARKERS.any { it in t }) throttleRate(relay.url) + if (SUB_LIMIT_MARKERS.any { it in t }) demoteConcurrency(relay.url) } - /** JSON-friendly view of which relays we throttled and how far. */ + /** Step [relay] one rung down the concurrency-cap ladder, unless already at the floor. */ + private fun demoteConcurrency(relay: NormalizedRelayUrl) { + if ((subDemotions[relay] ?: 0) >= subLadder.size) return + val step = subDemotions.merge(relay, 1, Int::plus)!! + val cap = subLadder[(step - 1).coerceIn(0, subLadder.size - 1)] + gate(relay).lower(cap) + if (step <= subLadder.size) { + System.err.println("[limiter] ${relay.url} concurrency capped at $cap subs (sub-limit #$step)") + } + } + + /** Step [relay] one rung down the rate ladder, unless already at the slowest. */ + private fun throttleRate(relay: NormalizedRelayUrl) { + if ((rateSteps[relay] ?: 0) >= rateLadder.size) return + val step = rateSteps.merge(relay, 1, Int::plus)!! + val d = rateLadder[(step - 1).coerceIn(0, rateLadder.size - 1)] + rateDelayMs[relay] = d + if (step <= rateLadder.size) { + System.err.println("[limiter] ${relay.url} rate-throttled to 1 REQ / ${d}ms (rate-limit #$step)") + } + } + + /** JSON-friendly view of which relays we throttled, in which dimension, how far. */ fun snapshot(): Map { val cappedAt = sortedMapOf() - for ((_, step) in demotions) { - val cap = ladder[(step - 1).coerceIn(0, ladder.size - 1)] + for ((_, step) in subDemotions) { + val cap = subLadder[(step - 1).coerceIn(0, subLadder.size - 1)] cappedAt.merge(cap, 1, Int::plus) } + val rateAt = sortedMapOf() + for ((_, step) in rateSteps) { + val d = rateLadder[(step - 1).coerceIn(0, rateLadder.size - 1)] + rateAt.merge(d, 1, Int::plus) + } return mapOf( "start_cap" to startCap, - "ladder" to ladder, - "throttled_relays" to demotions.size, - "capped_at" to cappedAt, + "sub_ladder" to subLadder, + "rate_ladder_ms" to rateLadder, + "concurrency_capped_relays" to subDemotions.size, + "concurrency_capped_at" to cappedAt, + "rate_limited_relays" to rateSteps.size, + "rate_limited_at_ms" to rateAt, ) } - fun hadThrottling(): Boolean = demotions.isNotEmpty() + fun hadThrottling(): Boolean = subDemotions.isNotEmpty() || rateSteps.isNotEmpty() /** * A bounded-concurrency gate whose limit can only ever be *lowered* (relays @@ -174,24 +232,33 @@ class AdaptiveRelayLimiter( } companion object { - // Substrings (matched case-insensitively) that mean "you're opening too - // many concurrent subscriptions / sending too fast" — the failure modes a - // lower per-relay cap actually fixes. Auth/blocked/restricted/unsupported - // are deliberately excluded: throttling wouldn't help those. - private val CONCURRENCY_MARKERS = + // A cap on how many subscriptions may be OPEN at once. Fix: fewer + // concurrent subs (demote the concurrency cap). + private val SUB_LIMIT_MARKERS = listOf( "too many concurrent", "concurrent req", "too many subscription", "number of subscriptions", + "subscriptions exceeds", "subscription limit", + "subscription count", + "maximum concurrent subscription", + "max subscription", "too many req", + ) + + // Too many subscription CHANGES per second. Fix: space the REQs out in + // time (a per-relay min interval), not fewer concurrent subs. + private val RATE_LIMIT_MARKERS = + listOf( "rate-limit", "rate limit", "ratelimit", + "too many messages", + "too many requests", "burst exhausted", "throttl", - "too many messages", "slow down", ) }