From 978441bdd50c857d9cf657a0ca0cfee75e58b884 Mon Sep 17 00:00:00 2001 From: JNK234 Date: Tue, 18 Aug 2026 17:28:28 -0500 Subject: [PATCH 1/3] feat: add proactive request throttling --- docs/CONFIGURATION.md | 29 ++ docs/SETUP.md | 57 ++++ src/main/LLMExtension.scala | 12 + src/main/providers/BaseHttpProvider.scala | 27 +- src/main/providers/RequestThrottle.scala | 361 ++++++++++++++++++++++ src/test/RequestThrottleSpec.scala | 286 +++++++++++++++++ src/test/ThrottleIntegrationSpec.scala | 202 ++++++++++++ 7 files changed, 973 insertions(+), 1 deletion(-) create mode 100644 src/main/providers/RequestThrottle.scala create mode 100644 src/test/RequestThrottleSpec.scala create mode 100644 src/test/ThrottleIntegrationSpec.scala diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 02c3e02..210ef36 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -28,6 +28,7 @@ Starting from this version, configuration is validated immediately: - No quotes required; avoid trailing spaces around `=`. - **Supported keys**: - Common: `provider`, `model`, `temperature`, `max_tokens`, `timeout_seconds` + - Rate limiting: `retry_max_retries`, `retry_max_elapsed_seconds`, `max_concurrent_requests`, `min_request_interval_ms` - Provider-specific API keys: `openai_api_key`, `anthropic_api_key`, `gemini_api_key`, `openrouter_api_key`, `together_api_key` - Provider-specific base URLs: `openai_base_url`, `anthropic_base_url`, `gemini_base_url`, `ollama_base_url`, `openrouter_base_url`, `together_base_url` - Legacy (still supported): `api_key`, `base_url` (applies to current provider) @@ -124,6 +125,34 @@ model=gpt-4o-mini # llm:set-model "claude-3-5-sonnet-20241022" ``` +### Request Throttling (staying inside a rate limit) +A model that calls the LLM once per agent per tick sends one request per agent +simultaneously. On a free tier that exceeds the quota on the first tick. + +``` +# At most 4 requests in flight; start them at least 250ms apart +max_concurrent_requests=4 +min_request_interval_ms=250 +``` + +- `max_concurrent_requests` — whole number of simultaneous requests. Unset or `0` means + **disabled** (unbounded, the default). Negative or unparseable values are reported on + stderr and treated as disabled. +- `min_request_interval_ms` — **milliseconds** between request starts. Unset or `0` + means **disabled**. Use this for a requests-per-minute quota; a concurrency cap alone + limits simultaneity, not rate. + +Excess requests queue rather than fail, occupy no thread while waiting, and are admitted +in arrival order. The cap applies per provider and endpoint, using the provider-specific +base URL key (`openai_base_url`, `gemini_base_url`, and so on) — the same value used to +build the request — so different providers never share a cap. Spellings of one endpoint +that differ only by trailing slash or letter case count as the same endpoint. + +Queue time is not added to `timeout_seconds`, so a cap set well below your agent +population may require raising `timeout_seconds` too. + +See [SETUP.md](SETUP.md#request-throttling) for how to pick values. + ## Ollama Quick Start (No API Key) Use Ollama to run models locally without any cloud credentials. diff --git a/docs/SETUP.md b/docs/SETUP.md index d0e65ac..488ac68 100644 --- a/docs/SETUP.md +++ b/docs/SETUP.md @@ -191,9 +191,66 @@ Browse the full catalog: [console.groq.com/docs/models](https://console.groq.com | `timeout_seconds` | Request timeout | No | 30 | | `retry_max_retries` | Retry attempts after a rate-limit (HTTP 429) response | No | 6 | | `retry_max_elapsed_seconds` | Total time allowed waiting out rate limits | No | 65 | +| `max_concurrent_requests` | Maximum requests in flight at once, per provider | No | 0 (off) | +| `min_request_interval_ms` | Minimum milliseconds between request starts | No | 0 (off) | *Not required for Ollama +### Request throttling + +`retry_max_retries` recovers from a rate limit *after* it happens. Throttling exists +to avoid tripping one in the first place. + +`ask turtles [ llm:chat-async ... ]` sends one request per turtle with nothing between +the agents and the network, so 200 turtles open 200 connections on the first tick. +`max_concurrent_requests` caps how many may be in flight at once; the rest queue and +run as capacity frees up. No request is dropped — they are deferred, not discarded. + +``` +# At most 4 requests in flight, started at least 250ms apart +max_concurrent_requests=4 +min_request_interval_ms=250 +``` + +**Units and semantics:** + +- `max_concurrent_requests` — a whole number of requests. **Unset or `0` disables** + throttling, giving the unbounded behavior of earlier versions. Throttling is off by + default, so existing models are unaffected until you set it. +- `min_request_interval_ms` — **milliseconds** between the starts of successive + requests. Unset or `0` disables pacing. Use it for a requests-per-minute quota: a + concurrency cap bounds how many run *at once*, not how many run *per minute*, so a + small cap recycling quickly can still exceed an RPM limit. For a 20 RPM quota, an + interval of `3000` keeps you inside it. +- A **negative or unparseable** value is a mistake rather than a choice, so it is + reported on stderr and then treated as disabled. A typo will not stall your model, + but it will not pass silently either. Use `0` when you mean to switch throttling off. + +The limit is shared per provider and endpoint. Every request to the same provider and +provider-specific base URL key (`openai_base_url`, `gemini_base_url`, and so on — the +same value used to build the request) draws on one cap, no matter how many times the +extension rebuilds its provider internally, while a different provider gets its own — +a Gemini free tier will not throttle a paid OpenAI key running alongside it. Spellings +of one endpoint that differ only by trailing slash or letter case count as the same +endpoint. API keys are never part of that identity: it is rebuilt from scheme, host, +port and path only, so a credential in a URL's userinfo or query string is discarded. + +Queued requests hold no thread while they wait, and are admitted in arrival order, so +no agent can be starved by later arrivals. + +**Throttling and `timeout_seconds`:** queue time is *not* added to the timeout budget. +How long a request waits depends on how many agents call in a tick, not on the cap, so +there is no formula that could size an allowance for it honestly. This means a model +whose fan-out greatly exceeds its cap can still exceed `timeout_seconds` while queued — +if you set a small cap for a large population, raise `timeout_seconds` to match. An +unthrottled model's timeout behavior is exactly as it was. + +Choosing a value: start from your provider's documented limits. A free tier allowing +5 requests/minute suits `max_concurrent_requests=1` with `min_request_interval_ms=12000`. +A paid key with generous limits may not need throttling at all. If a run still hits 429s +with throttling on, the cap is above the quota — lower it, or reduce how often the model +calls the LLM. + ### Rate limits and retries When a provider returns HTTP 429, the extension waits and retries with jittered diff --git a/src/main/LLMExtension.scala b/src/main/LLMExtension.scala index 450131c..d2cbd36 100644 --- a/src/main/LLMExtension.scala +++ b/src/main/LLMExtension.scala @@ -196,6 +196,18 @@ class LLMExtension extends DefaultClassManager { * trip a 30s request timeout. The await bound is therefore the request timeout plus * the retry budget, so `timeout_seconds` keeps its meaning and modelers who lower it * do not thereby lose the ability to recover from a rate limit. + * + * Queue time behind a concurrency cap is NOT added to this bound, and deliberately + * so. Queue depth is set by how many agents call in a tick, not by the cap: with 200 + * turtles and a cap of 4, the last request waits out ~50 waves. No fixed formula + * covers that, and one that pretended to would fail precisely where fan-out is + * widest — the case throttling exists for. + * + * So a heavily throttled model can still exceed this bound while queued, and that is + * a documented, intentional limitation rather than a solved problem: a modeler who + * sets a cap far below their fan-out should raise `timeout_seconds` accordingly. The + * bound is unchanged from before throttling existed, so an unthrottled model — the + * default — behaves exactly as it did. */ private def getAwaitTimeout: FiniteDuration = { val retryBudget = configStore.get(RetryPolicy.MAX_ELAPSED_SECONDS) diff --git a/src/main/providers/BaseHttpProvider.scala b/src/main/providers/BaseHttpProvider.scala index 1c80073..a213f6b 100644 --- a/src/main/providers/BaseHttpProvider.scala +++ b/src/main/providers/BaseHttpProvider.scala @@ -103,6 +103,22 @@ abstract class BaseHttpProvider(implicit ec: ExecutionContext) extends LLMProvid /** Random source for jitter. Overridable so tests can make backoff deterministic. */ protected def retryRandom: () => Double = () => scala.util.Random.nextDouble() + /** + * Proactive throttle for this provider, or None when it is switched off. + * + * Resolved per request so a config change takes effect without recreating the + * provider, matching how retryPolicy is handled. The gate is shared across + * every instance addressing the same provider and endpoint — see + * RequestThrottle.identity for why that is the right granularity. + */ + protected def requestThrottle: Option[RequestThrottle] = + RequestThrottle.forProvider( + providerName, + configStore.get(baseUrlConfigKey).getOrElse(defaultBaseUrl), + configStore.get(RequestThrottle.MAX_CONCURRENT_REQUESTS), + configStore.get(RequestThrottle.MIN_REQUEST_INTERVAL_MS) + ) + /** * Reports a rate-limit wait to the modeler. A silent multi-second stall inside * `go` is indistinguishable from a hang, so long waits are announced on stderr. @@ -243,6 +259,11 @@ abstract class BaseHttpProvider(implicit ec: ExecutionContext) extends LLMProvid * because retrying before a quota window reopens is guaranteed to fail. Waiting * is bounded by the policy's total elapsed budget rather than a fixed per-sleep * cap, so a quota window longer than the old 10s ceiling can actually be cleared. + * + * When a throttle is configured, the whole attempt sequence runs under a single + * permit. Holding it across retries rather than reacquiring per attempt is what + * keeps a retrying request from re-entering behind fresh arrivals and pushing + * the real in-flight count over the cap. */ protected def executeWithRetry( httpRequest: Request[Either[String, String]], @@ -280,7 +301,11 @@ abstract class BaseHttpProvider(implicit ec: ExecutionContext) extends LLMProvid } } - attempt(0, 0L) + requestThrottle match { + // One permit covers the complete logical request, including all retries. + case Some(throttle) => throttle.withPermit(attempt(0, 0L)) + case None => attempt(0, 0L) + } } override def setConfig(key: String, value: String): Unit = { diff --git a/src/main/providers/RequestThrottle.scala b/src/main/providers/RequestThrottle.scala new file mode 100644 index 0000000..22d8bfa --- /dev/null +++ b/src/main/providers/RequestThrottle.scala @@ -0,0 +1,361 @@ +// ABOUTME: Proactive request throttle — bounded in-flight requests and optional pacing per provider +// ABOUTME: Non-blocking FIFO gate so queued work never occupies a thread while it waits +package org.nlogo.extensions.llm.providers + +import java.util.concurrent.{ConcurrentHashMap, Executors, ScheduledExecutorService, TimeUnit} +import scala.collection.mutable +import scala.concurrent.{ExecutionContext, Future, Promise} +import scala.concurrent.duration._ +import scala.util.control.NonFatal + +/** + * Time source and delay scheduler for a throttle. + * + * Injected so pacing can be driven by an explicit clock in tests. Asserting on + * wall-clock gaps would make the suite a scheduler-noise detector: a paced + * delay that lands 20ms late says nothing about whether pacing is correct. + */ +trait ThrottleClock { + /** Current time in milliseconds. */ + def nowMs: Long + + /** A Future completing after `delayMs`, without occupying a thread. */ + def sleep(delayMs: Long): Future[Unit] +} + +/** Real time, with delays on a shared daemon scheduler. */ +object SystemThrottleClock extends ThrottleClock { + // Daemon so it never blocks JVM shutdown; one idle thread persists across + // extension reloads, mirroring the retry scheduler in BaseHttpProvider. + private lazy val scheduler: ScheduledExecutorService = + Executors.newSingleThreadScheduledExecutor { (r: Runnable) => + val t = new Thread(r, "llm-throttle-scheduler") + t.setDaemon(true) + t + } + + def nowMs: Long = System.currentTimeMillis() + + def sleep(delayMs: Long): Future[Unit] = { + if (delayMs <= 0L) return Future.unit + val p = Promise[Unit]() + scheduler.schedule( + new Runnable { def run(): Unit = p.trySuccess(()) }, + delayMs, + TimeUnit.MILLISECONDS + ) + p.future + } +} + +/** + * Limits how many requests are in flight at once, and optionally paces them. + * + * Retry-on-429 recovers from a rate limit after it happens; this exists to + * avoid tripping one. `ask turtles [ llm:chat-async ... ]` issues one request + * per turtle with nothing between the agents and the socket, so a 200-turtle + * model opens 200 connections on the first tick. + * + * Waiting is non-blocking by construction. A caller that cannot get a permit + * receives an incomplete Future and its continuation runs later, when a permit + * is handed over. No thread — least of all NetLogo's, or one belonging to + * ExecutionContext.global — ever sits parked inside this class waiting for + * capacity, which a `Semaphore.acquire()` would do. + * + * @param maxConcurrent permits available; must be positive + * @param minIntervalMs minimum gap between successive request starts, 0 to disable + * @param clock time source and delay scheduler + */ +class RequestThrottle( + val maxConcurrent: Int, + val minIntervalMs: Long, + clock: ThrottleClock = SystemThrottleClock +) { + require(maxConcurrent > 0, "maxConcurrent must be positive") + + // Guards `available`, `waiters` and `lastStartMs` together. Held only for + // queue arithmetic — never across a user callback or an HTTP send. + private val lock = new Object + private var available: Int = maxConcurrent + + // FIFO, so a waiter cannot be overtaken indefinitely — with agents calling + // every tick, barging would pass one over for the whole run. Only ever + // touched under `lock`, so a plain queue suffices. + private val waiters = mutable.Queue.empty[Promise[Unit]] + + // Start time reserved for the most recently admitted request, for pacing. + private var lastStartMs: Long = Long.MinValue + + /** + * Run `body` once a permit is free, releasing the permit however it finishes. + * + * The permit covers the whole logical request including its retries, because + * `body` is the retry loop rather than a single send. Releasing between + * attempts would let a retrying request re-enter behind newly arrived ones and + * push the real in-flight count above the cap — exactly what the cap forbids. + */ + def withPermit[A](body: => Future[A])(implicit ec: ExecutionContext): Future[A] = + acquire().flatMap { _ => + // A throw inside `body` before it returns a Future would otherwise leak + // the permit, so failure is normalised into the Future here. + val started = + try body + catch { case NonFatal(e) => Future.failed(e) } + // onComplete covers success, HTTP failure, and thrown exception alike. + started.onComplete(_ => release()) + started + } + + /** + * Obtain a permit. The returned Future completes immediately when capacity is + * free, otherwise when an earlier request releases one. + */ + private def acquire()(implicit ec: ExecutionContext): Future[Unit] = { + // ONE synchronized transition: take a permit, or join the queue. A free + // permit is taken only when nobody is already waiting, so a later arrival + // can never be admitted ahead of a waiter. + val waiting = lock.synchronized { + if (available > 0 && waiters.isEmpty) { + available -= 1 + None + } else { + val p = Promise[Unit]() + waiters.enqueue(p) + Some(p) + } + } + waiting match { + case None => paceThenProceed() + case Some(p) => p.future.flatMap(_ => paceThenProceed()) + } + } + + /** + * Delay the start of a request so successive starts are at least + * `minIntervalMs` apart. Serves a requests-per-minute quota: a cap alone + * bounds simultaneity, not rate, so N permits recycling quickly can still + * exceed an RPM limit. + */ + private def paceThenProceed()(implicit ec: ExecutionContext): Future[Unit] = { + if (minIntervalMs <= 0L) return Future.unit + + val waitMs = lock.synchronized { + val now = clock.nowMs + // Reserve this request's slot before releasing the lock, so concurrent + // callers each get a distinct one rather than all pacing off the same + // timestamp and starting together. + val earliest = + if (lastStartMs == Long.MinValue) now + else math.max(now, lastStartMs + minIntervalMs) + lastStartMs = earliest + earliest - now + } + + clock.sleep(waitMs) + } + + /** + * Return a permit, transferring it directly to the oldest waiter if there is one. + * + * Exactly ONE synchronized transition decides the permit's fate: it either + * dequeues the oldest waiter — leaving `available` untouched, so no arrival + * can ever observe that permit as free capacity — or, when the queue is empty, + * increments `available`. Both outcomes are decided under the same lock + * acquisition. + * + * Splitting these into two locked sections would be a lost wakeup, not merely + * unfair: a caller enqueuing after an empty queue was observed but before + * `available` was incremented would be handed nothing by this release, and + * nothing else would come to wake it — it would wait forever while capacity + * sat idle. Deciding both in one transition makes that interleaving + * unrepresentable. + * + * The Promise is completed after the lock is dropped, so a waiter's + * continuation never runs while this thread holds it. + */ + private def release(): Unit = { + val handoff = lock.synchronized { + if (waiters.nonEmpty) Some(waiters.dequeue()) + else { available += 1; None } + } + handoff.foreach(_.success(())) + } +} + +object RequestThrottle { + /** Maximum requests in flight at once for one provider. 0 or unset disables. */ + val MAX_CONCURRENT_REQUESTS = "max_concurrent_requests" + + /** Minimum milliseconds between request starts. 0 or unset disables pacing. */ + val MIN_REQUEST_INTERVAL_MS = "min_request_interval_ms" + + // Messages already shown, so a bad config value is reported once rather than + // once per request. forProvider runs on every request; an unconditional + // println would emit one line per agent per tick — worst in exactly the + // high-fan-out models throttling exists to help. + private val warningsShown = ConcurrentHashMap.newKeySet[String]() + + /** Print `message` to stderr the first time it is seen. */ + private def warnOnce(message: String): Unit = + if (warningsShown.add(message)) System.err.println(message) + + // Process-wide, because provider instances are not stable. LLMExtension drops + // and rebuilds `currentProvider` on set-provider, load-config, set-thinking, + // set-reasoning-effort and set-thinking-budget; an instance field would reset + // the cap mid-run and lose track of requests the previous instance still has + // open. Keyed identity survives that churn. + private val throttles = new ConcurrentHashMap[String, RequestThrottle]() + + /** + * Identity of the quota a request draws on: the provider and the endpoint it + * talks to. + * + * Two instances of the same provider must share one gate, or the cap means + * nothing. Different providers must not share one, or a Gemini free tier + * throttles a paid OpenAI key alongside it. + * + * The base URL splits genuinely separate endpoints — two OpenRouter + * configurations pointed at different hosts do not share a quota, nor does a + * local Ollama share one with a remote. + * + * The API key is deliberately NOT part of this. It is the truest identity of + * an account quota, but a secret has no business in a map key that may be + * printed in a diagnostic. Base URL is the safe proxy. + * + * The URL is normalised first. Two spellings of one endpoint — a trailing + * slash, a capitalised host — address the same quota, so treating them as + * distinct would silently hand a modeler twice the cap they configured. + */ + private[providers] def identity(providerName: String, baseUrl: String): String = + s"${providerName.toLowerCase.trim}|${normalizeEndpoint(baseUrl)}" + + /** + * Canonical form of a base URL for identity purposes. + * + * Scheme and host are case-insensitive per RFC 3986, and a trailing slash on + * the base does not change which endpoint is addressed. The path is otherwise + * left alone: `/v1` and `/v2` are genuinely different endpoints. + * + * Falls back to a trimmed, lowercased string if the value will not parse, + * since a throttle key must never be the thing that fails a request. + */ + private def normalizeEndpoint(baseUrl: String): String = { + val trimmed = Option(baseUrl).getOrElse("").trim + try { + val uri = new java.net.URI(trimmed) + val scheme = Option(uri.getScheme).map(_.toLowerCase).getOrElse("") + val host = Option(uri.getHost).map(_.toLowerCase).getOrElse("") + if (scheme.isEmpty || host.isEmpty) trimmed.toLowerCase.stripSuffix("/") + else { + // Default ports are implicit, so :443 must not split from the bare host. + val port = uri.getPort match { + case -1 => "" + case 80 if scheme == "http" => "" + case 443 if scheme == "https" => "" + case p => s":$p" + } + val path = Option(uri.getPath).getOrElse("").stripSuffix("/") + s"$scheme://$host$port$path" + } + } catch { + case NonFatal(_) => trimmed.toLowerCase.stripSuffix("/") + } + } + + /** + * The throttle for a provider+endpoint, or None when throttling is off. + * + * Unset or zero disables the gate, preserving the unbounded behaviour of + * earlier versions. A negative or unparseable value is a mistake rather than + * an intention, so it is reported on stderr and then treated as disabled: a + * typo must not stall a model, but silently ignoring a cap a modeler asked + * for is how a run keeps overrunning a quota with nothing to show why. + */ + def forProvider( + providerName: String, + baseUrl: String, + maxConcurrentRaw: Option[String], + minIntervalRaw: Option[String] + ): Option[RequestThrottle] = + parseCap(maxConcurrentRaw).map { limit => + val interval = parseInterval(minIntervalRaw) + val key = identity(providerName, baseUrl) + val replaced = new java.util.concurrent.atomic.AtomicBoolean(false) + + // compute() so a concurrent caller cannot observe a half-installed gate + // or race two replacements past each other. The remapping function does + // no I/O — it runs while the map holds a bin lock, so the notice below is + // emitted afterwards rather than inside it. + val throttle = throttles.compute(key, (_, existing) => { + if (existing != null && existing.maxConcurrent == limit && existing.minIntervalMs == interval) { + existing + } else { + if (existing != null) replaced.set(true) + new RequestThrottle(limit, interval) + } + }) + + if (replaced.get()) { + // Requests already running under the previous gate keep its permits, so + // in-flight work can briefly exceed the new cap. Reporting beats a stall + // or a silently changed limit. + warnOnce( + s"NOTE: request throttle for $providerName updated to " + + s"$MAX_CONCURRENT_REQUESTS=$limit, $MIN_REQUEST_INTERVAL_MS=$interval. " + + "Requests already in flight finish under the previous limit." + ) + } + throttle + } + + /** + * Parse the concurrency cap. None means "no gate". + * + * Zero and unset are the documented ways to disable and pass silently. + * Anything else unusable is announced, because it means the modeler intended + * a limit and is not getting one. + */ + private def parseCap(raw: Option[String]): Option[Int] = + raw.map(_.trim).filter(_.nonEmpty) match { + case None => None + case Some(v) => + v.toIntOption match { + case Some(n) if n > 0 => Some(n) + case Some(0) => None // documented "off" + case Some(n) => + warnOnce( + s"WARNING: $MAX_CONCURRENT_REQUESTS must be a positive whole number of requests, " + + s"but was '$n'. Request throttling is disabled; use 0 to disable it deliberately." + ) + None + case None => + warnOnce( + s"WARNING: $MAX_CONCURRENT_REQUESTS must be a whole number, but was '$v'. " + + "Request throttling is disabled; use 0 to disable it deliberately." + ) + None + } + } + + /** Parse the pacing interval in milliseconds. Unusable values disable pacing. */ + private def parseInterval(raw: Option[String]): Long = + raw.map(_.trim).filter(_.nonEmpty) match { + case None => 0L + case Some(v) => + v.toLongOption match { + case Some(n) if n >= 0 => n + case Some(n) => + warnOnce( + s"WARNING: $MIN_REQUEST_INTERVAL_MS must be milliseconds >= 0, but was '$n'. " + + "Pacing is disabled." + ) + 0L + case None => + warnOnce( + s"WARNING: $MIN_REQUEST_INTERVAL_MS must be a whole number of milliseconds, " + + s"but was '$v'. Pacing is disabled." + ) + 0L + } + } +} diff --git a/src/test/RequestThrottleSpec.scala b/src/test/RequestThrottleSpec.scala new file mode 100644 index 0000000..1b8acc6 --- /dev/null +++ b/src/test/RequestThrottleSpec.scala @@ -0,0 +1,286 @@ +// ABOUTME: Unit tests for RequestThrottle behavior — permits, FIFO queueing, pacing, config parsing +// ABOUTME: Uses a manual clock so pacing is asserted deterministically, never on wall-clock timing +package org.nlogo.extensions.llm.providers + +import org.scalatest.funsuite.AnyFunSuite +import scala.collection.mutable +import scala.concurrent.{Await, Future, Promise} +import scala.concurrent.ExecutionContext.Implicits.global +import scala.concurrent.duration._ + +/** + * Clock whose time only moves when a test moves it. + * + * Pacing is a statement about ordering and delay arithmetic, not about how + * punctual the JVM scheduler is. Asserting on elapsed wall-clock time turns + * these into flaky scheduler-noise detectors; with a manual clock the same + * properties are exact. + */ +class ManualClock(start: Long = 0L) extends ThrottleClock { + private var current: Long = start + private val pending = mutable.ListBuffer.empty[(Long, Promise[Unit])] + + def nowMs: Long = synchronized(current) + + def sleep(delayMs: Long): Future[Unit] = synchronized { + if (delayMs <= 0L) Future.unit + else { + val p = Promise[Unit]() + pending += ((current + delayMs, p)) + p.future + } + } + + /** Delays requested but not yet elapsed, in milliseconds from `start`. */ + def scheduledAt: List[Long] = synchronized(pending.map(_._1).toList.sorted) + + /** Move time forward, completing every delay that has now elapsed. */ + def advance(ms: Long): Unit = { + val due = synchronized { + current += ms + val (elapsed, rest) = pending.partition(_._1 <= current) + pending.clear() + pending ++= rest + elapsed.toList + } + due.foreach(_._2.trySuccess(())) + } +} + +class RequestThrottleSpec extends AnyFunSuite { + + /** Text every bad-cap warning contains, used to count them. */ + private val MAX_CONCURRENT_WARNING_MARKER = RequestThrottle.MAX_CONCURRENT_REQUESTS + + /** Block until `cond` holds, or fail. Used only to await async handoffs. */ + private def eventually(what: String)(cond: => Boolean): Unit = { + val deadline = System.nanoTime() + 5.seconds.toNanos + while (System.nanoTime() < deadline && !cond) Thread.sleep(2) + assert(cond, s"timed out waiting for $what") + } + + test("permits beyond the cap wait, then every request completes") { + // The core invariant: a burst wider than the cap is deferred, not dropped. + val throttle = new RequestThrottle(maxConcurrent = 2, minIntervalMs = 0) + val gates = List.fill(5)(Promise[Unit]()) + val running = new java.util.concurrent.atomic.AtomicInteger(0) + val peak = new java.util.concurrent.atomic.AtomicInteger(0) + + val calls = gates.map { gate => + throttle.withPermit { + val n = running.incrementAndGet() + peak.updateAndGet(p => math.max(p, n)) + gate.future.map(_ => running.decrementAndGet()) + } + } + + eventually("the cap to be filled")(running.get() == 2) + assert(peak.get() == 2, s"only 2 may run at once, saw ${peak.get()}") + + gates.foreach(_.trySuccess(())) + Await.result(Future.sequence(calls), 10.seconds) + assert(peak.get() == 2, s"the cap must hold for the whole run, saw ${peak.get()}") + } + + test("permits are released on success, failure, and a synchronous throw") { + // A leaked permit is worse than no throttle: after `maxConcurrent` leaks the + // model stops issuing requests and looks like a hang. All three exit paths + // must return the permit, so all three are exercised against a cap of 1 — + // a leak in any of them blocks the requests that follow. + val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 0) + + Await.result(throttle.withPermit(Future.successful("ok")), 5.seconds) + + intercept[RuntimeException] { + Await.result(throttle.withPermit(Future.failed(new RuntimeException("boom"))), 5.seconds) + } + + intercept[IllegalStateException] { + Await.result( + throttle.withPermit[String](throw new IllegalStateException("thrown before any Future")), + 5.seconds + ) + } + + // If any of the three leaked, this final acquisition never completes. + assert(Await.result(throttle.withPermit(Future.successful("free")), 5.seconds) == "free") + } + + test("queued work is admitted in arrival order") { + // FIFO is what rules out starvation: with agents calling every tick, a + // barging (non-fair) gate can pass over one waiter for an entire run. + val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 0) + val hold = Promise[Unit]() + val admitted = new java.util.concurrent.ConcurrentLinkedQueue[Int]() + + val blocker = throttle.withPermit(hold.future) + // withPermit takes a permit or joins the queue synchronously, before it + // returns, so constructing these in order IS enqueuing them in order — no + // waiting or polling is needed to establish arrival order. + val queuedCalls = (1 to 4).map(i => throttle.withPermit(Future { admitted.add(i); () })) + + hold.success(()) + Await.result(Future.sequence(blocker +: queuedCalls), 10.seconds) + + val order = admitted.toArray(Array.empty[Integer]).map(_.intValue()).toList + assert(order == List(1, 2, 3, 4), s"expected FIFO admission, got $order") + } + + test("queued work occupies no thread while it waits") { + // Blocking on a semaphore would park one thread per queued agent and can + // deadlock the global pool. Queue far more work than the pool has threads, + // then prove the pool still runs something else. + val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 0) + val hold = Promise[Unit]() + val blocker = throttle.withPermit(hold.future) + val queuedCalls = (1 to 64).map(_ => throttle.withPermit(Future.successful(()))) + + assert(Await.result(Future(41 + 1), 5.seconds) == 42, + "the execution context must not be starved by queued requests") + + hold.success(()) + Await.result(Future.sequence(blocker +: queuedCalls), 30.seconds) + } + + test("pacing spaces successive starts by the configured interval") { + // Deterministic: the clock only moves when this test moves it, so the + // assertion is about the delays the throttle asks for, not about timing. + val clock = new ManualClock() + val throttle = new RequestThrottle(maxConcurrent = 4, minIntervalMs = 100, clock) + + val calls = (1 to 3).map(_ => throttle.withPermit(Future.successful(()))) + eventually("all three to reserve a start slot")(clock.scheduledAt.size == 2) + + // First starts immediately; the next two are held to 100ms and 200ms. + assert(clock.scheduledAt == List(100L, 200L), s"got ${clock.scheduledAt}") + + clock.advance(200) + Await.result(Future.sequence(calls), 10.seconds) + } + + test("pacing is skipped entirely when the interval is zero") { + // Pacing off must cost nothing — no scheduled delays at all. + val clock = new ManualClock() + val throttle = new RequestThrottle(maxConcurrent = 4, minIntervalMs = 0, clock) + + val calls = (1 to 3).map(_ => throttle.withPermit(Future.successful(()))) + Await.result(Future.sequence(calls), 10.seconds) + assert(clock.scheduledAt.isEmpty, s"expected no pacing delays, got ${clock.scheduledAt}") + } + + // --- Configuration --- + + test("unset and zero disable throttling, preserving unbounded behaviour") { + assert(RequestThrottle.forProvider("p", "http://a", None, None).isEmpty) + assert(RequestThrottle.forProvider("p", "http://a", Some("0"), None).isEmpty) + assert(RequestThrottle.forProvider("p", "http://a", Some(""), None).isEmpty) + } + + test("negative and unparseable caps warn and disable throttling") { + // A typo must not stall a model, but it must not pass in silence either: + // the modeler asked for a limit and is not getting one. + Seq("-1", "lots", "2.5").foreach { bad => + val warning = captureStdErr { + assert(RequestThrottle.forProvider("p", "http://a", Some(bad), None).isEmpty, + s"'$bad' should disable throttling") + } + assert(warning.contains(RequestThrottle.MAX_CONCURRENT_REQUESTS), + s"'$bad' should be reported to the modeler, got: $warning") + } + } + + test("a bad value is reported once, not on every request") { + // forProvider runs per request, so an unconditional warning would print once + // per agent per tick — hundreds of thousands of lines in the very models + // throttling is meant to help, drowning the console and slowing the run. + val warnings = captureStdErr { + (1 to 50).foreach(_ => + RequestThrottle.forProvider("warn-once", "http://warn-once.local", Some("nonsense"), None)) + } + val count = warnings.linesIterator.count(_.contains(MAX_CONCURRENT_WARNING_MARKER)) + assert(count == 1, s"expected exactly 1 warning across 50 resolutions, got $count") + } + + test("an unusable interval warns, disables pacing, and keeps the cap") { + val warning = captureStdErr { + val throttle = RequestThrottle + .forProvider("p", "http://a", Some("2"), Some("soon")) + .getOrElse(fail("the cap should still apply")) + assert(throttle.maxConcurrent == 2) + assert(throttle.minIntervalMs == 0L) + } + assert(warning.contains(RequestThrottle.MIN_REQUEST_INTERVAL_MS), s"got: $warning") + } + + // --- Shared identity --- + + test("identity covers provider and endpoint but never a credential") { + // A secret in a map key can surface in any diagnostic that prints it, so the + // identity is rebuilt from scheme, host, port and path only. That drops the + // two places a credential can legitimately appear in a URL: userinfo, and a + // query parameter — which is exactly how Gemini passes its API key. + assert(RequestThrottle.identity("openai", "https://api.openai.com/v1") == + "openai|https://api.openai.com/v1") + + val withUserInfo = RequestThrottle.identity("p", "https://user:sk-secret@api.test/v1") + assert(!withUserInfo.contains("sk-secret"), s"userinfo must not survive: $withUserInfo") + + val withQueryKey = RequestThrottle.identity("gemini", "https://api.test/v1?key=sk-secret") + assert(!withQueryKey.contains("sk-secret"), s"a query key must not survive: $withQueryKey") + } + + test("one gate per provider and endpoint, not per instance") { + // The gate must survive LLMExtension rebuilding its provider (set-provider, + // load-config, set-thinking all do), and must not merge separate quotas. + def gate(provider: String, url: String) = + RequestThrottle.forProvider(provider, url, Some("3"), None).getOrElse(fail("expected a gate")) + + val a = gate("openai", "https://api.openai.com/v1") + val again = gate("openai", "https://api.openai.com/v1") + assert(a.eq(again), "the same provider and endpoint must share one gate") + + assert(!a.eq(gate("gemini", "https://api.openai.com/v1")), + "different providers must not share a gate") + assert(!gate("ollama", "http://localhost:11434").eq(gate("ollama", "http://remote:11434")), + "the same provider on different endpoints must not share a gate") + } + + test("endpoints differing only in trailing slash or case share one gate") { + // These address one quota, so splitting them silently doubles the effective + // cap: a modeler who writes a trailing slash gets 2x the limit they set. + def gate(url: String) = + RequestThrottle.forProvider("normalize", url, Some("1"), None).getOrElse(fail("expected a gate")) + + val canonical = gate("https://api.normalize.test/v1") + assert(canonical.eq(gate("https://api.normalize.test/v1/")), "a trailing slash must not split the gate") + assert(canonical.eq(gate("https://API.Normalize.TEST/v1")), "host case must not split the gate") + assert(canonical.eq(gate(" https://api.normalize.test/v1 ")), "surrounding space must not split the gate") + + // Genuinely different endpoints must still be kept apart. + assert(!canonical.eq(gate("https://api.normalize.test/v2")), "a different path is a different endpoint") + assert(!canonical.eq(gate("https://other.normalize.test/v1")), "a different host is a different endpoint") + } + + test("changing the configured cap replaces the gate") { + val url = "http://identity-change.local" + val first = RequestThrottle.forProvider("cfgchange", url, Some("2"), None).getOrElse(fail()) + assert(first.maxConcurrent == 2) + + val second = RequestThrottle.forProvider("cfgchange", url, Some("5"), None).getOrElse(fail()) + assert(second.maxConcurrent == 5, "a changed cap must take effect") + assert(!first.eq(second)) + } + + /** Run `body`, returning whatever it wrote to stderr. */ + private def captureStdErr(body: => Unit): String = { + val buffer = new java.io.ByteArrayOutputStream() + val original = System.err + try { + System.setErr(new java.io.PrintStream(buffer, true, "UTF-8")) + body + } finally { + System.setErr(original) + } + buffer.toString("UTF-8") + } +} diff --git a/src/test/ThrottleIntegrationSpec.scala b/src/test/ThrottleIntegrationSpec.scala new file mode 100644 index 0000000..5e53bc7 --- /dev/null +++ b/src/test/ThrottleIntegrationSpec.scala @@ -0,0 +1,202 @@ +// ABOUTME: Integration tests for the throttle gate inside BaseHttpProvider's request path +// ABOUTME: Covers cap enforcement over real sends, retry interaction, and the unthrottled default +package org.nlogo.extensions.llm.providers + +import org.nlogo.extensions.llm.models.{ChatRequest, ChatResponse, ChatMessage, Choice} +import org.scalatest.funsuite.AnyFunSuite +import sttp.client4._ +import sttp.client4.testing.{BackendStub, ResponseStub} +import sttp.model.{StatusCode, Uri} +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.{Await, Future} +import scala.concurrent.ExecutionContext.Implicits.global +import scala.concurrent.duration._ + +/** + * Provider backed by a stub backend that holds every send open until released, + * so requests admitted by the gate overlap and the true peak is observable. A + * fast stub would complete before the next send began and measure a peak of 1 + * no matter how many requests were let through. + * + * Each instance gets a distinct base URL by default, so a test's gate is its + * own — no shared-registry reset is needed to keep cases independent. + */ +class ProbeProvider( + release: CountDownLatch, + override val providerName: String = "throttle-probe", + baseUrl: String +) extends BaseHttpProvider { + + private val running = new AtomicInteger(0) + private val peakSeen = new AtomicInteger(0) + val sendCount = new AtomicInteger(0) + + /** Highest number of sends in the backend at one instant. */ + def peak: Int = peakSeen.get() + + /** Sends currently inside the backend. */ + def inBackend: Int = running.get() + + override lazy val backend: Backend[Future] = + BackendStub.asynchronousFuture.whenAnyRequest.thenRespondF { _ => + Future { + sendCount.incrementAndGet() + val n = running.incrementAndGet() + peakSeen.updateAndGet(p => math.max(p, n)) + release.await() + running.decrementAndGet() + ResponseStub.adjust("ok", StatusCode.Ok) + } + } + + override protected def retryBaseDelayMs: Long = 10L + override protected def retryRandom: () => Double = () => 0.0 + + override def defaultModel: String = "stub-model" + override protected def defaultBaseUrl: String = baseUrl + override protected def baseUrlConfigKey: String = "throttle_probe_base_url" + override protected def apiKeyConfigKey: String = "throttle_probe_api_key" + override protected def defaultMaxTokens: String = "128" + override protected def requiresApiKey: Boolean = false + + override protected def buildApiUrl(base: String): Uri = uri"$base/chat" + override protected def buildHeaders(apiKey: Option[String]): Map[String, String] = + Map("Content-Type" -> "application/json") + override protected def createProviderRequest(request: ChatRequest): ujson.Value = + ujson.Obj("messages" -> ujson.Arr(request.messages.map(m => ujson.Obj("content" -> m.content))*)) + override protected def parseProviderResponse(responseBody: String, model: String): ChatResponse = + ChatResponse( + id = "probe-response", + created = 0L, + model = model, + choices = Array(Choice(0, ChatMessage.assistant(responseBody), "stop")) + ) +} + +/** Provider returning a scripted sequence of responses, for the retry path. */ +class ScriptedProvider(responses: Seq[Response[testing.StubBody]], baseUrl: String) + extends BaseHttpProvider { + + val sendCount = new AtomicInteger(0) + + override lazy val backend: Backend[Future] = + BackendStub.asynchronousFuture.whenAnyRequest.thenRespond { + val idx = sendCount.getAndIncrement() + responses(math.min(idx, responses.length - 1)) + } + + override protected def retryBaseDelayMs: Long = 10L + override protected def retryRandom: () => Double = () => 0.0 + + override def providerName: String = "throttle-scripted" + override def defaultModel: String = "stub-model" + override protected def defaultBaseUrl: String = baseUrl + override protected def baseUrlConfigKey: String = "throttle_scripted_base_url" + override protected def apiKeyConfigKey: String = "throttle_scripted_api_key" + override protected def defaultMaxTokens: String = "128" + override protected def requiresApiKey: Boolean = false + + override protected def buildApiUrl(base: String): Uri = uri"$base/chat" + override protected def buildHeaders(apiKey: Option[String]): Map[String, String] = + Map("Content-Type" -> "application/json") + override protected def createProviderRequest(request: ChatRequest): ujson.Value = + ujson.Obj("messages" -> ujson.Arr(request.messages.map(m => ujson.Obj("content" -> m.content))*)) + override protected def parseProviderResponse(responseBody: String, model: String): ChatResponse = + ChatResponse( + id = "scripted-response", + created = 0L, + model = model, + choices = Array(Choice(0, ChatMessage.assistant(responseBody), "stop")) + ) +} + +class ThrottleIntegrationSpec extends AnyFunSuite { + + private def request = ChatRequest(model = "stub-model", messages = Seq(ChatMessage.user("hi"))) + + // A distinct endpoint per test, so each case gets its own gate out of the + // shared registry without depending on a global reset between tests. + private val urls = new AtomicInteger(0) + private def freshUrl(): String = s"http://throttle-test-${urls.incrementAndGet()}.local" + + private def eventually(what: String)(cond: => Boolean): Unit = { + val deadline = System.nanoTime() + 5.seconds.toNanos + while (System.nanoTime() < deadline && !cond) Thread.sleep(2) + assert(cond, s"timed out waiting for $what") + } + + test("a burst of agent calls is capped and every request still completes") { + // The defect this feature exists to fix: `ask turtles [ llm:chat-async ]` + // fanning every agent's request out to HTTP at once. + val release = new CountDownLatch(1) + val provider = new ProbeProvider(release, baseUrl = freshUrl()) + provider.setConfig(RequestThrottle.MAX_CONCURRENT_REQUESTS, "2") + + val calls = (1 to 10).map(_ => provider.chat(request)) + try { + eventually("the cap to be filled")(provider.inBackend == 2) + + // Every send blocks on the release latch, so none can have finished and let + // a successor through unseen: with the cap filled, exactly the admitted + // requests have been sent. An unbounded path would show all 10 here. + assert(provider.sendCount.get() == 2, + s"only 2 requests may reach the backend, saw ${provider.sendCount.get()}") + assert(provider.peak == 2, s"expected at most 2 concurrent sends, saw ${provider.peak}") + } finally { + // Never leak blocked backend work into later tests when an assertion fails. + release.countDown() + Await.ready(Future.sequence(calls), 15.seconds) + } + assert(provider.peak <= 2, s"the cap must hold for the whole burst, saw ${provider.peak}") + assert(provider.sendCount.get() == 10, "every deferred request must still be sent") + } + + test("an unthrottled provider fans out exactly as before") { + // Backward compatibility: with no throttle configured, behaviour is + // unchanged from before this feature existed. + val release = new CountDownLatch(1) + val provider = new ProbeProvider(release, baseUrl = freshUrl()) + + val calls = (1 to 5).map(_ => provider.chat(request)) + try { + eventually("all five to reach the backend")(provider.inBackend == 5) + assert(provider.peak == 5, s"unthrottled fan-out must be unbounded, saw ${provider.peak}") + } finally { + release.countDown() + Await.ready(Future.sequence(calls), 15.seconds) + } + } + + test("a retrying request holds one permit across all its attempts") { + // Reacquiring per attempt would let a retry re-enter behind fresh arrivals + // and push real in-flight work above the cap. Holding one permit across the + // whole sequence is what prevents that — and the permit must be released + // once, at the end, or the next request never starts. + val rateLimited = ResponseStub.adjust("rate limit exceeded", StatusCode.TooManyRequests) + val ok = ResponseStub.adjust("recovered", StatusCode.Ok) + val provider = new ScriptedProvider(Seq(rateLimited, rateLimited, ok), freshUrl()) + provider.setConfig(RequestThrottle.MAX_CONCURRENT_REQUESTS, "1") + + val result = Await.result(provider.chat(request), 15.seconds) + assert(result.firstMessage.map(_.content).contains("recovered")) + assert(provider.sendCount.get() == 3, "retry must still happen under the throttle") + + // With a cap of 1, a permit leaked across the retries would hang this. + val next = Await.result(provider.chat(request), 15.seconds) + assert(next.firstMessage.map(_.content).contains("recovered")) + } + + test("a request that exhausts its retries releases its permit") { + val rateLimited = ResponseStub.adjust("rate limit exceeded", StatusCode.TooManyRequests) + val provider = new ScriptedProvider(Seq(rateLimited), freshUrl()) + provider.setConfig(RequestThrottle.MAX_CONCURRENT_REQUESTS, "1") + provider.setConfig(RetryPolicy.MAX_RETRIES, "1") + + intercept[RuntimeException](Await.result(provider.chat(request), 15.seconds)) + + // The gate must still admit work; a leak here would look like a hang. + intercept[RuntimeException](Await.result(provider.chat(request), 15.seconds)) + assert(provider.sendCount.get() == 4, "two attempts per call, both calls admitted") + } +} From 000f99d100fd9e3827fa41c2914802bf54b6f6a1 Mon Sep 17 00:00:00 2001 From: JNK234 Date: Thu, 27 Aug 2026 12:51:16 -0500 Subject: [PATCH 2/3] fix: release throttle permit when pacing fails A permit is taken before pacing runs, but withPermit could only register its release once acquire() completed. A pacing delay that failed - or a scheduler that rejected it by throwing - therefore stranded the permit for good. That is worse than a failed request: capacity never returns, so with max_concurrent_requests=1 the gate is dead and every later request waits forever for a permit nobody holds. Release is now attached in acquire(), covering both the failed-Future and synchronous-throw paths. Also mark the integration stub's blocking wait with scala.concurrent.blocking. Holding a send open parks a worker of the global fork-join pool, and the unthrottled case deliberately holds five open at once. Without the hint the pool spawns no compensation thread, so on a 4-core runner the fifth send never starts and the test times out while the production code is correct. --- src/main/providers/RequestThrottle.scala | 24 ++++++++++-- src/test/RequestThrottleSpec.scala | 50 ++++++++++++++++++++++++ src/test/ThrottleIntegrationSpec.scala | 10 ++++- 3 files changed, 79 insertions(+), 5 deletions(-) diff --git a/src/main/providers/RequestThrottle.scala b/src/main/providers/RequestThrottle.scala index 22d8bfa..6de0e0b 100644 --- a/src/main/providers/RequestThrottle.scala +++ b/src/main/providers/RequestThrottle.scala @@ -124,12 +124,30 @@ class RequestThrottle( Some(p) } } - waiting match { - case None => paceThenProceed() - case Some(p) => p.future.flatMap(_ => paceThenProceed()) + // The permit is held from here on, but `withPermit` cannot register its + // release until this Future completes. Anything that fails in between would + // therefore strand the permit for good — and a stranded permit is far worse + // than a failed request: capacity never comes back, so with a cap of 1 the + // gate is dead and every later request waits forever for a permit nobody + // holds. Pacing is the reachable case (a scheduler that rejects work throws + // instead of returning a failed Future), so the release is attached here + // rather than left to the caller. + val paced = + waiting match { + case None => pacedStart() + case Some(p) => p.future.flatMap(_ => pacedStart()) + } + paced.recoverWith { case NonFatal(e) => + release() + Future.failed(e) } } + /** `paceThenProceed`, with a synchronous throw normalised into the Future. */ + private def pacedStart()(implicit ec: ExecutionContext): Future[Unit] = + try paceThenProceed() + catch { case NonFatal(e) => Future.failed(e) } + /** * Delay the start of a request so successive starts are at least * `minIntervalMs` apart. Serves a requests-per-minute quota: a cap alone diff --git a/src/test/RequestThrottleSpec.scala b/src/test/RequestThrottleSpec.scala index 1b8acc6..ad081b9 100644 --- a/src/test/RequestThrottleSpec.scala +++ b/src/test/RequestThrottleSpec.scala @@ -47,6 +47,23 @@ class ManualClock(start: Long = 0L) extends ThrottleClock { } } +/** Clock whose delays always fail, for the pacing-failure path. */ +object FailingClock extends ThrottleClock { + def nowMs: Long = 0L + def sleep(delayMs: Long): Future[Unit] = + Future.failed(new IllegalStateException("pacing delay failed")) +} + +/** + * Clock whose `sleep` throws rather than returning a failed Future, mirroring a + * scheduler that rejects work because it has been shut down. + */ +object ThrowingClock extends ThrottleClock { + def nowMs: Long = 0L + def sleep(delayMs: Long): Future[Unit] = + throw new IllegalStateException("scheduler rejected the delay") +} + class RequestThrottleSpec extends AnyFunSuite { /** Text every bad-cap warning contains, used to count them. */ @@ -158,6 +175,39 @@ class RequestThrottleSpec extends AnyFunSuite { Await.result(Future.sequence(calls), 10.seconds) } + test("a failing pacing delay releases the permit instead of leaking it") { + // The permit is taken before pacing runs, so a pacing delay that fails must + // still give it back. Otherwise one failure permanently removes capacity: + // with a cap of 1 the gate is dead, and every later request waits forever + // for a permit nobody holds. Losing a request is recoverable; losing the + // gate is not. + val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 100, FailingClock) + + intercept[IllegalStateException] { + Await.result(throttle.withPermit(Future.successful("never runs")), 5.seconds) + } + + // Capacity must have returned. A leak makes this wait out the timeout. + intercept[IllegalStateException] { + Await.result(throttle.withPermit(Future.successful("never runs either")), 5.seconds) + } + } + + test("a pacing delay that throws synchronously releases the permit") { + // SystemThrottleClock.sleep schedules on an executor, which throws + // RejectedExecutionException rather than returning a failed Future if that + // executor is shut down. That path must not lose the permit either. + val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 100, ThrowingClock) + + intercept[IllegalStateException] { + Await.result(throttle.withPermit(Future.successful("never runs")), 5.seconds) + } + + intercept[IllegalStateException] { + Await.result(throttle.withPermit(Future.successful("never runs either")), 5.seconds) + } + } + test("pacing is skipped entirely when the interval is zero") { // Pacing off must cost nothing — no scheduled delays at all. val clock = new ManualClock() diff --git a/src/test/ThrottleIntegrationSpec.scala b/src/test/ThrottleIntegrationSpec.scala index 5e53bc7..043f48c 100644 --- a/src/test/ThrottleIntegrationSpec.scala +++ b/src/test/ThrottleIntegrationSpec.scala @@ -9,7 +9,7 @@ import sttp.client4.testing.{BackendStub, ResponseStub} import sttp.model.{StatusCode, Uri} import java.util.concurrent.CountDownLatch import java.util.concurrent.atomic.AtomicInteger -import scala.concurrent.{Await, Future} +import scala.concurrent.{Await, Future, blocking} import scala.concurrent.ExecutionContext.Implicits.global import scala.concurrent.duration._ @@ -44,7 +44,13 @@ class ProbeProvider( sendCount.incrementAndGet() val n = running.incrementAndGet() peakSeen.updateAndGet(p => math.max(p, n)) - release.await() + // `blocking` is load-bearing, not decoration. Holding a send open parks + // a worker of the global fork-join pool, and these tests deliberately + // hold more sends open at once than a small machine has workers — the + // unthrottled case needs five. Without this hint the pool spawns no + // compensation thread, the fifth send never starts, and the test times + // out on a 4-core CI runner while the production code is perfectly fine. + blocking(release.await()) running.decrementAndGet() ResponseStub.adjust("ok", StatusCode.Ok) } From d220843e49590ae310dfff179310f91e635f7b88 Mon Sep 17 00:00:00 2001 From: JNK234 Date: Thu, 27 Aug 2026 12:58:02 -0500 Subject: [PATCH 3/3] fix: make provider registry tests order-independent ClaudeRequestSpec builds a ClaudeProvider in its constructor, which reads the provider registry, but nothing in the suite installed it. ProviderDefaultsSpec cleared that same global registry from its own constructor. Which suite won depended on the order ScalaTest built them in, so CI aborted ClaudeRequestSpec with "Unknown provider: anthropic". ClaudeRequestSpec now registers what it needs, and ProviderDefaultsSpec no longer resets shared state out from under suites already built against it. registerAll() inserts by name, so re-registering is all either suite needs. Both now pass in isolation, which is what makes them order-independent. --- src/test/ClaudeRequestSpec.scala | 8 ++++++++ src/test/ProviderDefaultsSpec.scala | 7 ++++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/src/test/ClaudeRequestSpec.scala b/src/test/ClaudeRequestSpec.scala index 78e0bc0..d3f389a 100644 --- a/src/test/ClaudeRequestSpec.scala +++ b/src/test/ClaudeRequestSpec.scala @@ -15,6 +15,14 @@ class InspectableClaudeProvider extends ClaudeProvider()(using scala.concurrent. class ClaudeRequestSpec extends AnyFunSuite { + // ClaudeProvider.defaultModel reads the provider registry, which is normally + // populated by LLMExtension.load(). Nothing here installs it, and the registry + // is global mutable state that another suite resets in its own constructor, so + // whether this suite could construct its provider depended on the order suites + // happened to be built in. Registering here makes the suite self-contained, the + // same way ProviderDefaultsSpec already does. + ProviderRegistrations.registerAll() + private val provider = new InspectableClaudeProvider private def request( diff --git a/src/test/ProviderDefaultsSpec.scala b/src/test/ProviderDefaultsSpec.scala index 7e7c286..73dcd2d 100644 --- a/src/test/ProviderDefaultsSpec.scala +++ b/src/test/ProviderDefaultsSpec.scala @@ -16,7 +16,12 @@ class ProviderDefaultsSpec extends AnyFunSuite { // Registrations are normally installed by LLMExtension.load(); do it here so // the suite is self-contained and order-independent. - ProviderRegistry.reset() + // + // Deliberately no ProviderRegistry.reset() first. The registry is global mutable + // state shared with every other suite, and clearing it from a constructor emptied + // it out from under suites that had already been built against it — ClaudeRequestSpec + // aborts with "Unknown provider: anthropic" when it loses that race. registerAll() + // is idempotent (it inserts by name), so re-registering is all this suite needs. ProviderRegistrations.registerAll() private val descriptors = ProviderRegistry.allNames.toSeq.sorted.flatMap(ProviderRegistry.get)