fix(nodes): address PR #560 review feedback for rate limiter

charliegillet · claude · charliegillet · commit c173873cd285 · 2026-04-06T11:17:17.000-07:00
- Fix critical bug: check semaphore BEFORE consuming tokens so rejected
  requests don't wastefully consume rate-limit tokens. Release semaphore
  if token check fails afterward.
- Extract _config_int() utility with min_value/max_value clamping params
  from inline _int_or_default helper in IGlobal._build_rate_limiter.
- Allow _build_rate_limiter to return None when all three rate-limit
  knobs are explicitly set to 0 (opt-out path).
- Remove unused RateLimitError import with noqa:F401 from http_driver.py.
- Add unit tests covering: normal acquire/release, per-second enforcement,
  per-minute enforcement, semaphore exhaustion, token restoration on
  semaphore rejection, and thread safety.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/nodes/src/nodes/tool_http_request/IGlobal.py b/nodes/src/nodes/tool_http_request/IGlobal.py
@@ -36,6 +36,30 @@
 
 from .rate_limiter import DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_PER_MINUTE, DEFAULT_MAX_PER_SECOND, RateLimiter
 
+
+def _config_int(cfg: dict, key: str, default: int, *, min_value: int | None = None, max_value: int | None = None) -> int:
+    """Read an integer from *cfg*, falling back to *default*.
+
+    Returns *default* when the key is missing, non-numeric, or <= 0.
+    The result is clamped to [min_value, max_value] when those bounds are given.
+    """
+    raw = cfg.get(key)
+    if raw is None:
+        val = default
+    else:
+        try:
+            val = int(raw)
+            if val <= 0:
+                val = default
+        except (TypeError, ValueError):
+            val = default
+    if min_value is not None:
+        val = max(val, min_value)
+    if max_value is not None:
+        val = min(val, max_value)
+    return val
+
+
 _METHOD_FLAGS = {
     'GET': 'allowGET',
     'POST': 'allowPOST',
@@ -94,23 +118,32 @@ def _build_guardrails(cfg: dict) -> tuple[set[str], list[re.Pattern]]:
         return enabled, patterns
 
     @staticmethod
-    def _build_rate_limiter(cfg: dict) -> RateLimiter:
-        """Create a ``RateLimiter`` from the node configuration."""
-
-        def _int_or_default(key: str, default: int) -> int:
-            raw = cfg.get(key)
+    def _build_rate_limiter(cfg: dict) -> RateLimiter | None:
+        """Create a ``RateLimiter`` from the node configuration.
+
+        Returns ``None`` when all three rate-limit knobs are explicitly set to
+        ``0`` (i.e. the user has opted out of rate limiting).
+        """
+        raw_ps = cfg.get('rateLimitPerSecond')
+        raw_pm = cfg.get('rateLimitPerMinute')
+        raw_mc = cfg.get('maxConcurrentRequests')
+
+        # If all three are explicitly set to 0, disable rate limiting entirely.
+        def _is_zero(raw: object) -> bool:
             if raw is None:
-                return default
+                return False
             try:
-                val = int(raw)
-                return val if val > 0 else default
+                return int(raw) == 0
             except (TypeError, ValueError):
-                return default
+                return False
+
+        if _is_zero(raw_ps) and _is_zero(raw_pm) and _is_zero(raw_mc):
+            return None
 
         return RateLimiter(
-            max_per_second=_int_or_default('rateLimitPerSecond', DEFAULT_MAX_PER_SECOND),
-            max_per_minute=_int_or_default('rateLimitPerMinute', DEFAULT_MAX_PER_MINUTE),
-            max_concurrent=_int_or_default('maxConcurrentRequests', DEFAULT_MAX_CONCURRENT),
+            max_per_second=_config_int(cfg, 'rateLimitPerSecond', DEFAULT_MAX_PER_SECOND, min_value=1),
+            max_per_minute=_config_int(cfg, 'rateLimitPerMinute', DEFAULT_MAX_PER_MINUTE, min_value=1),
+            max_concurrent=_config_int(cfg, 'maxConcurrentRequests', DEFAULT_MAX_CONCURRENT, min_value=1),
         )
 
     def validateConfig(self) -> None:
diff --git a/nodes/src/nodes/tool_http_request/rate_limiter.py b/nodes/src/nodes/tool_http_request/rate_limiter.py
@@ -84,20 +84,25 @@ def __init__(
 
     def acquire(self) -> None:
         """Acquire a rate-limit slot, or raise ``RateLimitError``."""
-        # 1. Check token buckets (per-second + per-minute).
-        with self._lock:
-            self._refill()
-            if self._ps_tokens < 1.0:
-                raise RateLimitError(f'Rate limit exceeded: max {self._ps_capacity} requests per second. Please retry after a short delay.')
-            if self._pm_tokens < 1.0:
-                raise RateLimitError(f'Rate limit exceeded: max {self._pm_capacity} requests per minute. Please retry after a short delay.')
-            self._ps_tokens -= 1.0
-            self._pm_tokens -= 1.0
-
-        # 2. Check concurrency limit (non-blocking).
+        # 1. Check concurrency limit first (non-blocking) so we never
+        #    consume tokens for a request that would be rejected anyway.
         if not self._semaphore.acquire(blocking=False):
             raise RateLimitError(f'Too many concurrent requests: max {self._max_concurrent} in-flight. Please wait for an ongoing request to complete.')
 
+        # 2. Check token buckets (per-second + per-minute).
+        try:
+            with self._lock:
+                self._refill()
+                if self._ps_tokens < 1.0:
+                    raise RateLimitError(f'Rate limit exceeded: max {self._ps_capacity} requests per second. Please retry after a short delay.')
+                if self._pm_tokens < 1.0:
+                    raise RateLimitError(f'Rate limit exceeded: max {self._pm_capacity} requests per minute. Please retry after a short delay.')
+                self._ps_tokens -= 1.0
+                self._pm_tokens -= 1.0
+        except RateLimitError:
+            self._semaphore.release()
+            raise
+
     def release(self) -> None:
         """Release the concurrency slot after a request completes."""
         self._semaphore.release()
diff --git a/nodes/test/test_rate_limiter.py b/nodes/test/test_rate_limiter.py
@@ -0,0 +1,187 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2024 RocketRide Inc.
+# =============================================================================
+
+"""Unit tests for the token-bucket rate limiter."""
+
+from __future__ import annotations
+
+import threading
+import time
+
+import sys
+from pathlib import Path
+
+import pytest
+
+# Add the node source directory to sys.path so we can import the module
+# without triggering the top-level nodes/__init__.py (which requires the
+# engine runtime).
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'src' / 'nodes' / 'tool_http_request'))
+
+from rate_limiter import RateLimiter, RateLimitError  # noqa: E402
+
+
+class TestAcquireRelease:
+    """Normal acquire / release cycle."""
+
+    def test_single_acquire_release(self):
+        rl = RateLimiter(max_per_second=5, max_per_minute=100, max_concurrent=2)
+        rl.acquire()
+        rl.release()
+
+    def test_multiple_sequential_acquires(self):
+        rl = RateLimiter(max_per_second=3, max_per_minute=100, max_concurrent=3)
+        for _ in range(3):
+            rl.acquire()
+        for _ in range(3):
+            rl.release()
+
+
+class TestPerSecondEnforcement:
+    """Per-second token bucket rejects once exhausted."""
+
+    def test_exceeds_per_second_limit(self):
+        rl = RateLimiter(max_per_second=2, max_per_minute=100, max_concurrent=10)
+        rl.acquire()
+        rl.acquire()
+        with pytest.raises(RateLimitError, match='per second'):
+            rl.acquire()
+        # Clean up
+        rl.release()
+        rl.release()
+
+    def test_per_second_refills_over_time(self):
+        rl = RateLimiter(max_per_second=2, max_per_minute=100, max_concurrent=10)
+        rl.acquire()
+        rl.acquire()
+        rl.release()
+        rl.release()
+        # Wait long enough for tokens to refill
+        time.sleep(1.1)
+        rl.acquire()
+        rl.release()
+
+
+class TestPerMinuteEnforcement:
+    """Per-minute token bucket rejects once exhausted."""
+
+    def test_exceeds_per_minute_limit(self):
+        rl = RateLimiter(max_per_second=100, max_per_minute=3, max_concurrent=10)
+        rl.acquire()
+        rl.acquire()
+        rl.acquire()
+        with pytest.raises(RateLimitError, match='per minute'):
+            rl.acquire()
+        for _ in range(3):
+            rl.release()
+
+
+class TestSemaphoreExhaustion:
+    """Concurrency semaphore rejects when all slots are occupied."""
+
+    def test_exceeds_concurrent_limit(self):
+        rl = RateLimiter(max_per_second=100, max_per_minute=100, max_concurrent=2)
+        rl.acquire()
+        rl.acquire()
+        with pytest.raises(RateLimitError, match='concurrent'):
+            rl.acquire()
+        rl.release()
+        rl.release()
+
+    def test_release_frees_slot(self):
+        rl = RateLimiter(max_per_second=100, max_per_minute=100, max_concurrent=1)
+        rl.acquire()
+        rl.release()
+        # Should succeed now that the slot is freed.
+        rl.acquire()
+        rl.release()
+
+
+class TestTokenRestorationOnSemaphoreRejection:
+    """Tokens must NOT be consumed when the semaphore rejects the request."""
+
+    def test_tokens_preserved_after_semaphore_rejection(self):
+        rl = RateLimiter(max_per_second=2, max_per_minute=100, max_concurrent=1)
+
+        # Use up the only concurrency slot.
+        rl.acquire()
+
+        # This should fail on the semaphore. Tokens must not be consumed.
+        with pytest.raises(RateLimitError, match='concurrent'):
+            rl.acquire()
+
+        # Release the held slot.
+        rl.release()
+
+        # We should still have 1 per-second token left (only 1 was consumed
+        # by the first successful acquire).  If the bug existed (tokens
+        # consumed before semaphore check) this second acquire would fail
+        # with a per-second error.
+        rl.acquire()
+        rl.release()
+
+    def test_semaphore_not_leaked_on_token_rejection(self):
+        """Semaphore slot is released when token-bucket check fails.
+
+        With max_concurrent=2 and max_per_second=2: after two successful
+        acquires exhaust the per-second tokens, a third acquire will pass
+        the semaphore but fail on tokens.  The implementation must release
+        the semaphore slot in that case.  We verify by releasing all held
+        slots, waiting for token refill, then acquiring both concurrent
+        slots again — which would fail if one was leaked.
+        """
+        rl = RateLimiter(max_per_second=2, max_per_minute=100, max_concurrent=2)
+
+        # Exhaust both per-second tokens (each also takes a semaphore slot).
+        rl.acquire()
+        rl.acquire()
+
+        # Release one semaphore slot so the next acquire can get past the
+        # semaphore check and fail on the token bucket instead.
+        rl.release()
+
+        # This acquire gets a semaphore slot but fails on per-second tokens.
+        with pytest.raises(RateLimitError, match='per second'):
+            rl.acquire()
+
+        # Release the remaining held slot.
+        rl.release()
+
+        # Wait for per-second tokens to fully refill (capacity=2).
+        time.sleep(1.2)
+
+        # Both semaphore slots should be free.  If the failed acquire
+        # leaked a slot, the second acquire here would raise a
+        # concurrency error.
+        rl.acquire()
+        rl.acquire()
+        rl.release()
+        rl.release()
+
+
+class TestThreadSafety:
+    """Basic smoke test for concurrent usage."""
+
+    def test_concurrent_acquires(self):
+        rl = RateLimiter(max_per_second=50, max_per_minute=500, max_concurrent=5)
+        errors: list[Exception] = []
+
+        def worker():
+            try:
+                rl.acquire()
+                time.sleep(0.01)
+                rl.release()
+            except RateLimitError:
+                pass
+            except Exception as exc:
+                errors.append(exc)
+
+        threads = [threading.Thread(target=worker) for _ in range(20)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=5)
+
+        assert not errors, f'Unexpected errors in threads: {errors}'
