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..6de0e0b --- /dev/null +++ b/src/main/providers/RequestThrottle.scala @@ -0,0 +1,379 @@ +// 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) + } + } + // 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 + * 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/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) diff --git a/src/test/RequestThrottleSpec.scala b/src/test/RequestThrottleSpec.scala new file mode 100644 index 0000000..ad081b9 --- /dev/null +++ b/src/test/RequestThrottleSpec.scala @@ -0,0 +1,336 @@ +// 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(())) + } +} + +/** 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. */ + 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("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() + 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..043f48c --- /dev/null +++ b/src/test/ThrottleIntegrationSpec.scala @@ -0,0 +1,208 @@ +// 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, blocking} +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)) + // `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) + } + } + + 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") + } +}