laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-09 16:14:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(cli): split rate-limit from subscription-count limit in the crawl

Browse Source 
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.
This commit is contained in:
Claude
2026-07-07 13:10:58 +00:00
parent 6d64b7b41c
commit 59d5fc752c
1 changed files with 122 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/AdaptiveRelayLimiter.kt
+122 -55
View File
@@ -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<Int> = listOf(20, 10),
private val subLadder: List<Int> = listOf(20, 10),
private val rateLadder: List<Long> = listOf(250L, 500L, 1000L, 2000L),
) : RelayConnectionListener {
private val gates = ConcurrentHashMap<NormalizedRelayUrl, Gate>()
// 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<NormalizedRelayUrl, Int>()
// Concurrency-cap demotions per relay (== index+1 into subLadder). Capped at
// subLadder.size: past the floor we stop demoting.
private val subDemotions = ConcurrentHashMap<NormalizedRelayUrl, Int>()
// 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<NormalizedRelayUrl, Int>()
private val rateDelayMs = ConcurrentHashMap<NormalizedRelayUrl, Long>()
private val nextAllowedAtMs = ConcurrentHashMap<NormalizedRelayUrl, AtomicLong>()
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 <T> 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<String, Any?> {
val cappedAt = sortedMapOf<Int, Int>()
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<Long, Int>()
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",
)
}