feat(rate-limit): meter cheap reads separately and expose usage headers

.env.example

@@ -96,6 +96,7 @@ RATE_LIMIT_ENABLED='false'           # Enable/disable API key rate limiting
 RATE_LIMIT_ANON_PER_MINUTE='30'      # Anonymous (no API key) per-minute limit
 RATE_LIMIT_ANON_PER_HOUR='500'       # Anonymous per-hour limit
 RATE_LIMIT_ANON_PER_DAY='5000'       # Anonymous per-day limit
+RATE_LIMIT_READ_MULTIPLIER='10'      # Cheap reads (no GenVM) get this multiple of the tier limits
 
 # PENDING-tx queue depth caps for eth_sendRawTransaction (admission control).
 # Empty / unset = no cap (the default for self-hosted). Public shared

backend/database_handler/contract_snapshot.py

@@ -1,6 +1,7 @@
 # database_handler/contract_snapshot.py
 from .models import CurrentState
 from .errors import ContractNotFoundError
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 from typing import Optional, Dict
 import base64
@@ -87,20 +88,80 @@ def extract_deployed_code_b64(self) -> Optional[str]:
         slices out the code payload, and returns it base64-encoded. Returns None
         if missing/invalid.
         """
-        # Import here to avoid circular dependencies at module import time
-        from backend.node.genvm import get_code_slot
-
         accepted = self.states.get("accepted") or {}
 
         try:
-            code_slot_b64 = base64.b64encode(get_code_slot()).decode("ascii")
-            stored = accepted.get(code_slot_b64)
+            stored = accepted.get(_code_slot_b64())
             if not stored:
                 return None
-
-            raw = base64.b64decode(stored, validate=True)
-            code_len = int.from_bytes(raw[0:4], byteorder="little", signed=False)
-            code_bytes = raw[4 : 4 + code_len]
-            return base64.b64encode(code_bytes).decode("ascii")
+            return _decode_code_payload(stored)
         except Exception:
             return None
+
+
+def _code_slot_b64() -> str:
+    """Base64 of the deterministic storage slot the deployed code lives in."""
+    # Import here to avoid circular dependencies at module import time
+    from backend.node.genvm import get_code_slot
+
+    return base64.b64encode(get_code_slot()).decode("ascii")
+
+
+def _decode_code_payload(stored: str) -> Optional[str]:
+    """Slice the code out of a stored slot blob and re-encode it as base64.
+
+    The blob is a 4-byte little-endian length prefix followed by the code.
+    """
+    raw = base64.b64decode(stored, validate=True)
+    code_len = int.from_bytes(raw[0:4], byteorder="little", signed=False)
+    code_bytes = raw[4 : 4 + code_len]
+    return base64.b64encode(code_bytes).decode("ascii")
+
+
+def fetch_deployed_code_b64(session: Session, contract_address: str) -> Optional[str]:
+    """Read just the deployed code, without loading the contract's whole state.
+
+    ``ContractSnapshot`` pulls the entire ``data`` JSONB — every storage slot
+    the contract owns — in order to read one deterministic slot out of it. For a
+    contract holding a large vector store that is a big fetch and deserialize
+    per call, which matters because ``gen_getContractCode`` is polled heavily by
+    batch tooling. Extracting the slot in SQL keeps the state off the wire and
+    out of Python.
+
+    Postgres still has to detoast the JSONB server-side, so this narrows the
+    transfer and parse cost rather than eliminating the read entirely.
+
+    Raises ContractNotFoundError when the contract is absent or undeployed, and
+    returns None when the contract exists but holds no code.
+    """
+    slot = _code_slot_b64()
+
+    row = session.execute(
+        select(
+            func.jsonb_typeof(CurrentState.data).label("data_kind"),
+            func.jsonb_typeof(CurrentState.data["state"]).label("state_kind"),
+            CurrentState.data["state"]["accepted"][slot].astext.label("nested"),
+            CurrentState.data["state"][slot].astext.label("flat"),
+        ).where(CurrentState.id == contract_address)
+    ).one_or_none()
+
+    if row is None:
+        raise ContractNotFoundError(contract_address)
+
+    if row.data_kind != "object" or row.state_kind is None:
+        # Legacy rows store `data` as a JSON string scalar, and undeployed ones
+        # store an empty object with no `state` key. Both are rare and fiddly,
+        # so hand them to the original path rather than reimplementing its error
+        # handling in SQL.
+        return ContractSnapshot(contract_address, session).extract_deployed_code_b64()
+
+    # Current rows nest slots under `state.accepted`; the pre-migration format
+    # put them directly under `state`.
+    stored = row.nested if row.nested is not None else row.flat
+    if not stored:
+        return None
+
+    try:
+        return _decode_code_payload(stored)
+    except Exception:
+        return None

backend/protocol_rpc/endpoints.py

@@ -16,7 +16,10 @@
 from sqlalchemy.orm import Session
 import backend.validators as validators
 
-from backend.database_handler.contract_snapshot import ContractSnapshot
+from backend.database_handler.contract_snapshot import (
+    ContractSnapshot,
+    fetch_deployed_code_b64,
+)
 from backend.database_handler.llm_providers import LLMProviderRegistry
 from backend.rollup.consensus_service import ConsensusService
 from backend.database_handler.models import Base, TransactionStatus
@@ -1108,13 +1111,12 @@ async def get_contract_schema_for_code(
 
 def get_contract_code(session: Session, contract_address: str) -> str:
     try:
-        contract_snapshot = ContractSnapshot(contract_address, session)
+        code_b64 = fetch_deployed_code_b64(session, contract_address)
     except ContractNotFoundError:
         raise NotFoundError(
             message=f"Contract {contract_address} not found",
             data={"contract_address": contract_address},
         )
-    code_b64 = contract_snapshot.extract_deployed_code_b64()
     if not code_b64:
         raise InvalidAddressError(
             contract_address,

backend/protocol_rpc/fastapi_server.py

@@ -68,6 +68,17 @@ async def lifespan(app: FastAPI):
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
+    # Browsers hide non-simple response headers from JS unless they are listed
+    # here, so without this the rate limit headers are readable by curl but not
+    # by genlayer-js in the browser — the client that most needs to self-pace.
+    expose_headers=[
+        "Retry-After",
+        "X-RateLimit-Bucket",
+        "X-RateLimit-Window",
+        "X-RateLimit-Limit",
+        "X-RateLimit-Remaining",
+        "X-RateLimit-Reset",
+    ],
 )
 
 # Add rate limiting middleware (executes after CORS, before route handler)

backend/protocol_rpc/rate_limit_methods.py

@@ -0,0 +1,104 @@
+"""Classification of JSON-RPC methods for rate limiting.
+
+Requests are split into two buckets:
+
+- *cheap reads* — methods that never enter the GenVM and never touch an LLM.
+  These are served from Postgres (or are outright constants), so the cost of
+  serving one is orders of magnitude below a consensus round. They get their
+  own, much larger bucket.
+- *everything else* — the default. Keeps the pre-existing limits untouched.
+
+The allowlist below is deliberately conservative and maintained by hand rather
+than derived from any other list in the codebase. Two traps make that
+necessary:
+
+- ``DISABLE_INFO_LOGS_ENDPOINTS`` (set in the deployment env) looks like the
+  natural source, but it contains ``eth_call`` — which runs contract code in
+  the GenVM and can fan out to LLM validators. Reusing that list would make the
+  single most expensive call in the system effectively free.
+- ``gen_getContractSchema`` and ``gen_getContractSchemaForCode`` read like
+  metadata lookups, but both build a ``Node`` backed by a ``GenVMManager`` to
+  derive the schema from bytecode.
+
+Being too conservative is cheap: an omitted method simply keeps today's limits.
+Being too liberal hands out free capacity on a path that costs real money. When
+in doubt, leave a method out.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+# Bodies larger than this are not parsed for classification — they are charged
+# to the standard bucket. A cheap read is a handful of bytes; anything this
+# large is a contract deployment or a batch, neither of which is cheap.
+MAX_CLASSIFY_BODY_BYTES = 64 * 1024
+
+CHEAP_READ_METHODS = frozenset(
+    {
+        # Constants / trivial responses
+        "ping",
+        "net_version",
+        "eth_chainId",
+        "eth_syncing",
+        "eth_gasPrice",
+        "eth_maxPriorityFeePerGas",
+        "eth_blockNumber",
+        "eth_feeHistory",
+        "eth_getCode",  # returns a literal "0x"
+        "eth_estimateGas",  # returns a constant
+        "sim_getFinalityWindowTime",
+        "sim_getConsensusContract",
+        # Indexed database reads
+        "eth_getBalance",
+        "eth_getTransactionCount",
+        "eth_getTransactionByHash",
+        "eth_getTransactionReceipt",
+        "eth_getBlockByHash",
+        "eth_getBlockByNumber",
+        "gen_getContractCode",
+        "gen_getContractNonce",
+        "gen_getTransactionStatus",
+        "gen_getStudioTransactionByHash",
+        "sim_getTransactionsForAddress",
+    }
+)
+
+# Explicitly *not* cheap, recorded here so the reasoning survives future edits:
+#   eth_call, gen_call, sim_call        -> execute contract code in the GenVM
+#   gen_getContractSchema               -> builds a Node + GenVMManager
+#   gen_getContractSchemaForCode        -> builds a Node + GenVMManager
+#   sim_lintContract                    -> runs the GenVM linter
+#   eth_getLogs                         -> unbounded range scan
+#   eth_sendRawTransaction, sim_*, admin_*, dev_*  -> writes / privileged
+
+
+def is_cheap_read_payload(raw_body: bytes) -> bool:
+    """Return True if every call in this JSON-RPC body is a cheap read.
+
+    Anything ambiguous — unparseable, oversized, empty, or a batch containing a
+    single expensive call — is reported as *not* cheap, so uncertainty charges
+    the stricter bucket rather than the looser one.
+    """
+    if not raw_body or len(raw_body) > MAX_CLASSIFY_BODY_BYTES:
+        return False
+
+    try:
+        payload = json.loads(raw_body)
+    except (ValueError, UnicodeDecodeError):
+        return False
+
+    if isinstance(payload, list):
+        # A batch is only cheap if every member is. An empty batch is invalid
+        # JSON-RPC and is charged normally.
+        return bool(payload) and all(_is_cheap_call(call) for call in payload)
+
+    return _is_cheap_call(payload)
+
+
+def _is_cheap_call(call: Any) -> bool:
+    if not isinstance(call, dict):
+        return False
+    method = call.get("method")
+    return isinstance(method, str) and method in CHEAP_READ_METHODS

backend/protocol_rpc/rate_limit_middleware.py

@@ -12,7 +12,8 @@
 from starlette.responses import JSONResponse, Response
 
 from backend.protocol_rpc.exceptions import RateLimitExceeded
-from backend.protocol_rpc.rate_limiter import RateLimiterService
+from backend.protocol_rpc.rate_limit_methods import is_cheap_read_payload
+from backend.protocol_rpc.rate_limiter import RateLimiterService, RateLimitUsage
 
 logger = logging.getLogger(__name__)
 
@@ -63,29 +64,70 @@ async def dispatch(self, request: Request, call_next) -> Response:
 
         api_key = request.headers.get("X-API-Key")
         client_ip = self._client_ip(request)
+        is_cheap_read = await self._is_cheap_read(request)
 
+        usage: Optional[RateLimitUsage] = None
         try:
-            await rate_limiter.check_rate_limit(api_key, client_ip)
+            usage = await rate_limiter.check_rate_limit(
+                api_key, client_ip, is_cheap_read=is_cheap_read
+            )
         except RateLimitExceeded as exc:
-            retry_after = "60"
-            if exc.data and isinstance(exc.data, dict):
-                retry_after = str(exc.data.get("retry_after_seconds", 60))
             return JSONResponse(
                 status_code=429,
                 content={
                     "jsonrpc": "2.0",
                     "error": exc.to_dict(),
                     "id": None,
                 },
-                headers={"Retry-After": retry_after},
+                headers=self._denial_headers(exc),
             )
         except Exception:
             logger.warning(
                 "Rate limiter unavailable, allowing request through",
                 exc_info=True,
             )
 
-        return await call_next(request)
+        response = await call_next(request)
+        if usage is not None:
+            response.headers.update(usage.as_headers())
+        return response
+
+    async def _is_cheap_read(self, request: Request) -> bool:
+        """Classify the request body, charging the stricter bucket on any doubt.
+
+        Reading the body here is safe because Starlette's BaseHTTPMiddleware
+        wraps the request in a _CachedRequest, which replays the buffered body
+        downstream. Calling request.stream() instead would starve the route
+        handler.
+        """
+        try:
+            raw_body = await request.body()
+        except Exception:
+            logger.warning(
+                "Could not read request body to classify rate limit bucket",
+                exc_info=True,
+            )
+            return False
+        return is_cheap_read_payload(raw_body)
+
+    @staticmethod
+    def _denial_headers(exc: RateLimitExceeded) -> dict:
+        data = exc.data if isinstance(exc.data, dict) else {}
+        retry_after = data.get("retry_after_seconds", 60)
+        headers = {"Retry-After": str(retry_after)}
+        # An invalid API key is raised before any window is evaluated, so there
+        # is no usage to report — only the windowed denials carry limits.
+        if "limit" in data:
+            headers.update(
+                {
+                    "X-RateLimit-Bucket": str(data.get("bucket", "standard")),
+                    "X-RateLimit-Window": str(data.get("window", "")),
+                    "X-RateLimit-Limit": str(data["limit"]),
+                    "X-RateLimit-Remaining": "0",
+                    "X-RateLimit-Reset": str(retry_after),
+                }
+            )
+        return headers
 
     def _client_ip(self, request: Request) -> str:
         peer_host = request.client.host if request.client else "unknown"