fix(rpc): implement global backoff and normalized rate limits

Author: dvilelaf

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "iwa"
-version = "0.0.37"
+version = "0.0.38"
 description = "A secure, modular, and plugin-based framework for crypto agents and ops"
 readme = "README.md"
 requires-python = ">=3.12,<4.0"

src/iwa/core/chain/interface.py

@@ -35,7 +35,8 @@ def __init__(self, chain: Union[SupportedChain, str] = None):
 
         self.chain = chain
         # Enforce strict 1.0 RPS limit to prevent synchronization issues
-        self._rate_limiter = get_rate_limiter(chain.name, rate=1.0, burst=1)
+        # Use lowercase chain name for consistent file locking
+        self._rate_limiter = get_rate_limiter(chain.name.lower(), rate=1.0, burst=1)
         self._current_rpc_index = 0
         self._rpc_failure_counts: Dict[int, int] = {}
 
@@ -248,15 +249,21 @@ def _handle_rpc_error(self, error: Exception) -> Dict[str, Union[bool, int]]:
                 f"(RPC #{self._current_rpc_index}): {error}"
             )
 
+            # If it's a rate limit error, trigger global backoff immediately
+            # BEFORE rotating, to tell other processes "hey, cool it!"
+            if result["is_rate_limit"]:
+                # Backoff for 5 seconds locally and globally
+                self._rate_limiter.trigger_backoff(seconds=5.0)
+
             if self.rotate_rpc():
                 result["rotated"] = True
                 result["should_retry"] = True
                 logger.info(f"Rotated to RPC #{self._current_rpc_index} for {self.chain.name}")
             else:
                 if result["is_rate_limit"]:
-                    self._rate_limiter.trigger_backoff(seconds=5.0)
+                    # Already backed off above, but ensure retry flag is set
                     result["should_retry"] = True
-                    logger.warning("No other RPCs available, triggered backoff")
+                    logger.warning("No other RPCs available, relying on backoff")
 
         elif result["is_server_error"]:
             logger.warning(f"Server error on {self.chain.name}: {error}")
@@ -291,8 +298,13 @@ def _init_web3_under_lock(self):
         rpc_url = self.chain.rpcs[self._current_rpc_index] if self.chain.rpcs else ""
         raw_web3 = Web3(Web3.HTTPProvider(rpc_url, request_kwargs={"timeout": DEFAULT_RPC_TIMEOUT}))
 
+        # Ensure rate limiter is up to date (e.g. if we want to enforce parameters again)
+        # normalize to lowercase
+        self._rate_limiter = get_rate_limiter(self.chain.name.lower(), rate=1.0, burst=1)
+
         # Use duck typing to check if current web3 is a RateLimitedWeb3 wrapper
         if hasattr(self, "web3") and hasattr(self.web3, "set_backend"):
+            self.web3._rate_limiter = self._rate_limiter  # Update limiter ref
             self.web3.set_backend(raw_web3)
         else:
             self.web3 = RateLimitedWeb3(raw_web3, self._rate_limiter, self)

src/iwa/core/chain/rate_limiter.py

@@ -38,23 +38,22 @@ def __init__(
             rate: Maximum requests per second (refill rate)
             burst: Maximum tokens (bucket size)
         """
-        self.chain_name = chain_name
+        # Normalize chain name to lowercase to prevent split buckets
+        self.chain_name = chain_name.lower().strip()
         self.rate = rate
         self.burst = burst
 
         # Ensure cache directory exists
         CACHE_DIR.mkdir(parents=True, exist_ok=True)
 
         # Sanitize chain name for filename
-        safe_name = "".join(c for c in chain_name if c.isalnum() or c in ('-', '_'))
+        safe_name = "".join(c for c in self.chain_name if c.isalnum() or c in ('-', '_'))
         self.state_file: Path = CACHE_DIR / f"rate_limit_{safe_name}.json"
 
         # Use a string path for FileLock to ensure compatibility
         lock_path = CACHE_DIR / f"rate_limit_{safe_name}.lock"
         self._lock = FileLock(str(lock_path))
 
-        self._backoff_until = 0.0
-
     def _load_state(self) -> dict:
         """Load state from file, initializing if necessary."""
         try:
@@ -68,6 +67,7 @@ def _load_state(self) -> dict:
         return {
             "tokens": float(self.burst),
             "last_update": time.time(),
+            "backoff_until": 0.0,
         }
 
     def _save_state(self, state: dict):
@@ -86,18 +86,18 @@ def acquire(self, timeout: float = 30.0) -> bool:
             try:
                 # Acquire file lock to read/write state safely across processes
                 with self._lock.acquire(timeout=timeout):
+                    state = self._load_state()
                     now = time.time()
 
-                    # Check backoff (in-memory only for now, could be persisted if needed globally)
-                    if now < self._backoff_until:
-                        wait_time = self._backoff_until - now
+                    # Check global persisted backoff
+                    backoff_until = state.get("backoff_until", 0.0)
+                    if now < backoff_until:
+                        wait_time = backoff_until - now
                         if now + wait_time > deadline:
                             return False
                     else:
-                        state = self._load_state()
-
-                        tokens = state["tokens"]
-                        last_update = state["last_update"]
+                        tokens = state.get("tokens", float(self.burst))
+                        last_update = state.get("last_update", now)
 
                         # Refill tokens based on elapsed time
                         elapsed = now - last_update
@@ -136,20 +136,31 @@ def acquire(self, timeout: float = 30.0) -> bool:
                 time.sleep(0.1)
 
     def trigger_backoff(self, seconds: float = 5.0):
-        """Trigger rate limit backoff."""
-        # Simple in-memory backoff for now
-        self._backoff_until = time.time() + seconds
-        logger.warning(f"RPC rate limit triggered, backing off for {seconds}s")
+        """Trigger rate limit backoff globally."""
+        backoff_until = time.time() + seconds
+        logger.warning(f"RPC rate limit triggered for {self.chain_name}, backing off for {seconds}s (GLOBAL)")
+
+        try:
+            with self._lock.acquire(timeout=2.0):
+                state = self._load_state()
+                # Only update if our backoff is further in future
+                current_backoff = state.get("backoff_until", 0.0)
+                if backoff_until > current_backoff:
+                    state["backoff_until"] = backoff_until
+                    self._save_state(state)
+        except Exception as e:
+            logger.error(f"Failed to persist backoff state: {e}")
 
     def get_status(self) -> dict:
         """Get current rate limiter status."""
         try:
             with self._lock.acquire(timeout=1.0):
                 state = self._load_state()
                 return {
-                    "tokens": state["tokens"],
+                    "tokens": state.get("tokens", 0),
                     "rate": self.rate,
                     "burst": self.burst,
+                    "backoff_until": state.get("backoff_until", 0.0),
                 }
         except TimeoutError:
             return {"error": "Could not acquire lock to read status"}
@@ -165,35 +176,35 @@ def get_rate_limiter(chain_name: str, rate: float = None, burst: int = None) ->
 
     If the limiter already exists but with different parameters, update it.
     """
+    # Normalize chain name for cache lookup
+    normalized_name = chain_name.lower().strip()
+
     with _rate_limiters_lock:
-        if chain_name not in _rate_limiters:
-            _rate_limiters[chain_name] = RPCRateLimiter(
-                chain_name=chain_name,
+        if normalized_name not in _rate_limiters:
+            _rate_limiters[normalized_name] = RPCRateLimiter(
+                chain_name=normalized_name,
                 rate=rate or RPCRateLimiter.DEFAULT_RATE,
                 burst=burst or RPCRateLimiter.DEFAULT_BURST,
             )
         else:
             # Update existing limiter if stricter parameters are requested
-            limiter = _rate_limiters[chain_name]
+            limiter = _rate_limiters[normalized_name]
 
             if rate is not None and rate != limiter.rate:
                 logger.info(
-                    f"Updating rate limiter for {chain_name}: "
+                    f"Updating rate limiter for {normalized_name}: "
                     f"rate {limiter.rate}->{rate}"
                 )
                 limiter.rate = rate
 
             if burst is not None and burst != limiter.burst:
                 logger.info(
-                    f"Updating rate limiter for {chain_name}: "
+                    f"Updating rate limiter for {normalized_name}: "
                     f"burst {limiter.burst}->{burst}"
                 )
                 limiter.burst = burst
 
-            # If parameters updated, we might want to force a state save/refresh,
-            # but next acquire will handle it naturally.
-
-        return _rate_limiters[chain_name]
+        return _rate_limiters[normalized_name]
 
 
 class RateLimitedEth:

src/tests/test_rpc_rate_limit.py

@@ -1,3 +1,4 @@
+
 import json
 import time
 from pathlib import Path
@@ -13,54 +14,63 @@ def clean_rate_limiters():
     yield
     _rate_limiters.clear()
 
-def test_get_rate_limiter_updates_existing_instance(clean_rate_limiters):
-    """Verify that get_rate_limiter updates configuration if instance exists."""
-    chain_name = "UpdateTest"
+def test_rate_limiter_name_normalization(clean_rate_limiters):
+    """Verify that different casings map to the same file."""
 
-    # Mock file operations to prevent real file creation
     with patch("iwa.core.chain.rate_limiter.FileLock"), \
          patch("iwa.core.chain.rate_limiter.Path.mkdir"), \
          patch("iwa.core.chain.rate_limiter.Path.open"), \
-         patch("json.load", return_value={"tokens": 50.0, "last_update": time.time()}), \
+         patch("iwa.core.chain.rate_limiter.Path.exists", return_value=True), \
+         patch("json.load", return_value={"tokens": 50.0, "last_update": time.time(), "backoff_until": 0.0}), \
          patch("json.dump"):
 
-        # 1. Create with default values (25.0)
-        limiter1 = get_rate_limiter(chain_name)
-        assert limiter1.rate == 25.0
-        assert limiter1.burst == 50
+        l1 = get_rate_limiter("Gnosis")
+        l2 = get_rate_limiter("gnosis")
 
-        # 2. Request again with stricter values
-        limiter2 = get_rate_limiter(chain_name, rate=1.0, burst=1)
+        # Should be same object reference because key is normalized
+        assert l1 is l2
+        assert l1.chain_name == "gnosis"
 
-        # 3. Verify it is the same instance
-        assert limiter2 is limiter1
+def test_rate_limiter_backoff(clean_rate_limiters):
+    """Verify backoff triggers and is respected."""
+    chain_name = "BackoffTest"
 
-        # 4. Verify parameters were updated
-        assert limiter2.rate == 1.0, f"Rate should be updated to 1.0, got {limiter2.rate}"
-        assert limiter2.burst == 1, f"Burst should be updated to 1, got {limiter2.burst}"
+    with patch("iwa.core.chain.rate_limiter.FileLock"), \
+         patch("iwa.core.chain.rate_limiter.Path.mkdir") as mock_mkdir, \
+         patch("iwa.core.chain.rate_limiter.Path.open") as mock_open, \
+         patch("iwa.core.chain.rate_limiter.Path.exists", return_value=True), \
+         patch("json.dump") as mock_dump:
 
+        # Setup mock read to return NO backoff initially
+        mock_read_state = {"tokens": 10.0, "last_update": time.time(), "backoff_until": 0.0}
 
-def test_rate_limiter_uses_file_lock(clean_rate_limiters):
-    """Verify that RPCRateLimiter uses file locks."""
-    chain_name = "LockTest"
+        with patch("json.load", return_value=mock_read_state) as mock_load:
+            limiter = get_rate_limiter(chain_name)
 
-    with patch("iwa.core.chain.rate_limiter.FileLock") as MockLock, \
-         patch("iwa.core.chain.rate_limiter.Path.mkdir"), \
-         patch("iwa.core.chain.rate_limiter.Path.open") as mock_open, \
-         patch("json.load", return_value={"tokens": 1.0, "last_update": time.time() - 10}), \
-         patch("json.dump"):
+            # 1. Acquire should succeed
+            assert limiter.acquire() is True
+
+            # 2. Trigger Backoff
+            limiter.trigger_backoff(seconds=5.0)
+
+            # Verify we tried to save new state with backoff
+            args, _ = mock_dump.call_args
+            saved_state = args[0]
+            assert saved_state["backoff_until"] > time.time()
 
-        limiter = RPCRateLimiter(chain_name, rate=1.0, burst=1)
+            # 3. Simulate another process/call seeing the backoff
+            # Update read mock to return the backed-off state
+            future_backoff = time.time() + 5.0
+            mock_read_state["backoff_until"] = future_backoff
 
-        # Call acquire
-        limiter.acquire()
+            # Acquire should now fail (or block, but we expect False if it exceeds timeout,
+            # here we assume immediate return False if backoff > timeout,
+            # but wait, acquire blocks. Let's mock a short timeout)
 
-        # Verify lock was initialized with correct file path
-        # Note: In the implementation we convert Path to str for FileLock
-        MockLock.assert_called_once()
+            # Our implementation blocks until backoff expires OR timeout.
+            # If we pass timeout=0.1, and backoff is 5s, it should return False
+            start = time.time()
+            result = limiter.acquire(timeout=0.1)
+            duration = time.time() - start
 
-        # Verify lock was acquired
-        lock_instance = MockLock.return_value
-        # Note: implementation uses ctx manager `with self._lock.acquire():` which calls __enter__
-        # Or directly `with self._lock:`
-        assert lock_instance.acquire.called or lock_instance.__enter__.called
+            assert result is False, "Should fail to acquire during backoff"