api: read Hanzo KV from KV_URL; a KV outage no longer fails requests

The cache client is now the KV client (core/cache/kv.py, kv_client),
configured by KV_URL. rate_limit_check allows the request and logs when KV
is unreachable, and the RPC cache treats an outage as a miss. cloud_billing
reads KV directly instead of a missing usage_tracker.redis attribute; the
NFT refresh enqueues through an arq pool.

Co-authored-by: Hanzo Dev <[email protected]>

api/.env.example

@@ -12,8 +12,8 @@ DATABASE_URL=postgresql+asyncpg://bootnode:bootnode@localhost:5432/bootnode
 DB_POOL_SIZE=20
 DB_MAX_OVERFLOW=10
 
-# Redis
-REDIS_URL=redis://localhost:6379/0
+# Hanzo KV (RESP)
+KV_URL=redis://localhost:6379/0
 
 # Authentication
 JWT_SECRET=dev-jwt-secret-change-in-production

api/README.md

@@ -39,7 +39,7 @@ docker compose up -d
 
 ```bash
 DATABASE_URL=postgresql+asyncpg://...
-REDIS_URL=redis://...
+KV_URL=redis://...
 DATASTORE_URL=clickhouse://...
 JWT_SECRET=...
 ```

api/bootnode/api/chat.py

@@ -17,7 +17,7 @@ from pydantic import BaseModel
 
 from bootnode.api.deps import ProjectDep
 from bootnode.config import get_settings
-from bootnode.core.cache import redis_client
+from bootnode.core.cache import kv_client
 
 router = APIRouter()
 settings = get_settings()
@@ -114,7 +114,7 @@ async def chat_completions(
     """
     # Per-project rate limiting for chat (expensive LLM calls)
     rate_key = f"chat_rate:{project.id}"
-    allowed, remaining = await redis_client.rate_limit_check(
+    allowed, remaining = await kv_client.rate_limit_check(
         rate_key, CHAT_RATE_LIMIT
     )
     if not allowed:

api/bootnode/api/deps.py

@@ -11,7 +11,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from bootnode.config import get_settings
-from bootnode.core.cache import redis_client
+from bootnode.core.cache import kv_client
 from bootnode.db.models import ApiKey, Project, User
 from bootnode.db.session import get_db
 
@@ -87,7 +87,7 @@ async def verify_api_key_or_token(
 
     # Check rate limit for API key
     rate_key = f"rate:{db_key.id}"
-    allowed, remaining = await redis_client.rate_limit_check(
+    allowed, remaining = await kv_client.rate_limit_check(
         rate_key, db_key.rate_limit
     )
 

api/bootnode/api/nfts/__init__.py

@@ -225,12 +225,12 @@ async def refresh_nft_metadata(
             detail=f"Unsupported chain/network: {chain}/{network}",
         )
 
-    # Queue refresh job via Redis/arq
+    # Queue refresh job on the KV-backed arq queue
     try:
-        import redis.asyncio as aioredis
-        from arq.connections import ArqRedis
+        from arq import create_pool
+        from arq.connections import RedisSettings
 
-        pool: ArqRedis = await aioredis.from_url(settings.redis_url)
+        pool = await create_pool(RedisSettings.from_dsn(settings.kv_url))
         await pool.enqueue_job(
             "refresh_nft_metadata_job",
             chain,

api/bootnode/api/rpc/__init__.py

@@ -9,7 +9,7 @@ from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
 
 from bootnode.api.deps import ApiKeyDep, DbDep
-from bootnode.core.cache import redis_client
+from bootnode.core.cache import kv_client
 from bootnode.core.chains import ChainRegistry, RPCClient
 from bootnode.core.chains.rpc import RPCError, get_compute_units
 from bootnode.db.models import Usage
@@ -127,7 +127,7 @@ async def _handle_single(
     cache_key = None
     if method not in NO_CACHE_METHODS:
         params_hash = get_params_hash(params)
-        cached = await redis_client.get_rpc_response(chain, network, method, params_hash)
+        cached = await kv_client.get_rpc_response(chain, network, method, params_hash)
         if cached is not None:
             # Record usage (even cached)
             response_time = int((time.time() - start_time) * 1000)
@@ -147,7 +147,7 @@ async def _handle_single(
         # Cache the result if applicable
         if cache_key and method in CACHE_TTLS:
             ttl = CACHE_TTLS[method]
-            await redis_client.set_rpc_response(chain, network, method, cache_key, result, ttl)
+            await kv_client.set_rpc_response(chain, network, method, cache_key, result, ttl)
 
         # Record usage
         response_time = int((time.time() - start_time) * 1000)

api/bootnode/config.py

@@ -3,7 +3,7 @@
 from functools import lru_cache
 from typing import Annotated, Literal
 
-from pydantic import AliasChoices, Field, RedisDsn, field_validator
+from pydantic import AliasChoices, Field, field_validator
 from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 
@@ -32,8 +32,8 @@ class Settings(BaseSettings):
     db_pool_size: int = 20
     db_max_overflow: int = 10
 
-    # Redis
-    redis_url: RedisDsn = Field(default="redis://localhost:6379/0")
+    # Hanzo KV (RESP): cache, rate limits, usage counters, job queue
+    kv_url: str = Field(default="redis://localhost:6379/0")
 
     # Authentication
     jwt_secret: str = Field(default="change-me-in-production")
@@ -114,10 +114,6 @@ class Settings(BaseSettings):
     cloudflare_zone_id: str = ""
     cloudflare_api_token: str = ""
 
-    # Message Queue (Redis-based, @hanzo/mq compatible)
-    # Uses the same Redis instance as cache - different db index for isolation
-    mq_redis_db: int = 1  # Redis DB index for MQ (separate from cache at db 0)
-
     # External Services
     ipfs_gateway: str = "https://ipfs.io/ipfs/"
 

api/bootnode/core/billing/cloud_billing.py

@@ -18,7 +18,7 @@ from bootnode.core.billing.tiers import (
     PricingTier,
     get_cloud_compute_limits,
 )
-from bootnode.core.billing.tracker import usage_tracker
+from bootnode.core.cache import kv_client
 from bootnode.db.models import Subscription
 
 logger = structlog.get_logger()
@@ -188,12 +188,12 @@ class CloudBillingService:
         """
         cost_cents = int(report.compute_hours * report.hourly_rate_cents)
 
-        # Track in Redis for real-time aggregation.
+        # Track in Hanzo KV for real-time aggregation.
         key = f"cloud:usage:{report.instance_id}"
         month_key = f"cloud:monthly:{datetime.now(UTC).strftime('%Y-%m')}"
 
-        await usage_tracker.redis.incrbyfloat(key, report.compute_hours)
-        await usage_tracker.redis.incrby(f"{month_key}:cost_cents", cost_cents)
+        await kv_client.client.incrbyfloat(key, report.compute_hours)
+        await kv_client.client.incrby(f"{month_key}:cost_cents", cost_cents)
 
         logger.debug(
             "cloud usage recorded",
@@ -226,11 +226,11 @@ class CloudBillingService:
     ) -> int:
         """Count active (non-terminated) instances for a team+platform.
 
-        Uses Redis counter set by the control-plane, or returns 0 if unavailable.
+        Uses the KV counter set by the control-plane, or returns 0 if unavailable.
         """
         key = f"cloud:active:{team_id}:{platform}"
         try:
-            val = await usage_tracker.redis.get(key)
+            val = await kv_client.client.get(key)
             return int(val) if val else 0
         except Exception:
             return 0
@@ -242,7 +242,7 @@ class CloudBillingService:
         month = datetime.now(UTC).strftime("%Y-%m")
         key = f"cloud:hours:{team_id}:{month}"
         try:
-            val = await usage_tracker.redis.get(key)
+            val = await kv_client.client.get(key)
             return float(val) if val else 0.0
         except Exception:
             return 0.0
@@ -254,7 +254,7 @@ class CloudBillingService:
         month = datetime.now(UTC).strftime("%Y-%m")
         key = f"cloud:spend:{team_id}:{month}"
         try:
-            val = await usage_tracker.redis.get(key)
+            val = await kv_client.client.get(key)
             return int(val) if val else 0
         except Exception:
             return 0

api/bootnode/core/billing/sync.py

@@ -1,6 +1,6 @@
 """Usage sync worker for Commerce billing.
 
-Periodically syncs compute unit usage from Redis to Hanzo Commerce
+Periodically syncs compute unit usage from Hanzo KV to Hanzo Commerce
 for PAYG (pay-as-you-go) metered billing. Commerce handles invoicing via Square.
 """
 
@@ -17,20 +17,20 @@ from bootnode.config import get_settings
 from bootnode.core.billing.commerce import CommerceError, commerce_client
 from bootnode.core.billing.tiers import PricingTier
 from bootnode.core.billing.tracker import usage_tracker
-from bootnode.core.cache import redis_client
+from bootnode.core.cache import kv_client
 from bootnode.db.models import Subscription
 from bootnode.db.session import async_session
 
 logger = structlog.get_logger()
 
-# Redis key for tracking sync state
+# KV keys for tracking sync state
 SYNC_LOCK_KEY = "billing:sync:lock"
 SYNC_CURSOR_KEY = "billing:sync:cursor:{project_id}"
 LAST_SYNC_KEY = "billing:sync:last"
 
 
 class UsageSyncWorker:
-    """Periodically sync usage from Redis to Hanzo Commerce."""
+    """Periodically sync usage from Hanzo KV to Hanzo Commerce."""
 
     def __init__(self, interval_seconds: int = 3600) -> None:
         """Initialize sync worker.
@@ -44,7 +44,7 @@ class UsageSyncWorker:
 
     async def _acquire_lock(self, ttl: int = 300) -> bool:
         """Acquire distributed lock for sync operation."""
-        result = await redis_client.client.set(
+        result = await kv_client.client.set(
             SYNC_LOCK_KEY,
             datetime.now(UTC).isoformat(),
             nx=True,  # Only set if not exists
@@ -54,7 +54,7 @@ class UsageSyncWorker:
 
     async def _release_lock(self) -> None:
         """Release distributed lock."""
-        await redis_client.delete(SYNC_LOCK_KEY)
+        await kv_client.delete(SYNC_LOCK_KEY)
 
     async def _get_payg_subscriptions(
         self, db: AsyncSession
@@ -79,18 +79,18 @@ class UsageSyncWorker:
     async def _get_unsync_usage(self, project_id: uuid.UUID) -> int:
         """Get compute units accumulated since last sync."""
         key = f"billing:unsync:cu:{project_id}"
-        value = await redis_client.get(key)
+        value = await kv_client.get(key)
         return int(value) if value else 0
 
     async def _mark_synced(self, project_id: uuid.UUID, compute_units: int) -> None:
         """Mark usage as synced by decrementing counter."""
         key = f"billing:unsync:cu:{project_id}"
         # Decrement by the amount we synced
-        await redis_client.client.decrby(key, compute_units)
+        await kv_client.client.decrby(key, compute_units)
         # Ensure we don't go negative
-        current = await redis_client.get(key)
+        current = await kv_client.get(key)
         if current and int(current) < 0:
-            await redis_client.set(key, "0")
+            await kv_client.set(key, "0")
 
     async def _sync_project(
         self,
@@ -203,7 +203,7 @@ class UsageSyncWorker:
                             )
 
             # Record last sync time
-            await redis_client.set(LAST_SYNC_KEY, datetime.now(UTC).isoformat())
+            await kv_client.set(LAST_SYNC_KEY, datetime.now(UTC).isoformat())
             summary["completed_at"] = datetime.now(UTC).isoformat()
 
             logger.info(
@@ -282,8 +282,8 @@ class UsageSyncWorker:
 
     async def get_status(self) -> dict[str, Any]:
         """Get sync worker status."""
-        last_sync = await redis_client.get(LAST_SYNC_KEY)
-        lock_holder = await redis_client.get(SYNC_LOCK_KEY)
+        last_sync = await kv_client.get(LAST_SYNC_KEY)
+        lock_holder = await kv_client.get(SYNC_LOCK_KEY)
 
         return {
             "running": self._running,

api/bootnode/core/billing/tracker.py

@@ -1,6 +1,6 @@
-"""Real-time usage tracking with Redis.
+"""Real-time usage tracking in Hanzo KV.
 
-Tracks compute units in Redis for real-time rate limiting and quota checks.
+Tracks compute units in Hanzo KV for real-time rate limiting and quota checks.
 Batches writes to DataStore for analytics and billing.
 """
 
@@ -14,19 +14,19 @@ import structlog
 
 from bootnode.core.billing.compute_units import get_compute_units
 from bootnode.core.billing.tiers import PricingTier, get_tier_limits
-from bootnode.core.cache import redis_client
+from bootnode.core.cache import kv_client
 from bootnode.core.datastore import datastore_client
 
 logger = structlog.get_logger()
 
-# Redis key prefixes
+# KV key prefixes
 CU_USAGE_KEY = "cu:usage:{project_id}:{period}"  # Monthly CU counter
 RATE_LIMIT_KEY = "rate:{project_id}:{window}"  # Per-second rate limit
 CU_BUFFER_KEY = "cu:buffer:{project_id}"  # Batch buffer for DataStore
 
 
 class UsageTracker:
-    """Track compute units in Redis, batch to DataStore."""
+    """Track compute units in Hanzo KV, batch to DataStore."""
 
     def __init__(self) -> None:
         self._batch_size = 100  # Flush to DataStore after N records
@@ -73,16 +73,16 @@ class UsageTracker:
         if compute_units is None:
             compute_units = get_compute_units(method)
 
-        # Increment monthly CU counter in Redis
+        # Increment monthly CU counter in KV
         period_key = self._get_period_key(project_id)
         try:
-            pipe = redis_client.client.pipeline()
+            pipe = kv_client.client.pipeline()
             pipe.incrby(period_key, compute_units)
             # Set expiry to 35 days (covers billing period + grace)
             pipe.expire(period_key, 35 * 24 * 60 * 60)
             await pipe.execute()
         except Exception as e:
-            logger.error("Failed to track usage in Redis", error=str(e))
+            logger.error("Failed to track usage in KV", error=str(e))
 
         # Buffer for DataStore batch insert
         usage_record = {
@@ -103,8 +103,8 @@ class UsageTracker:
         # Add to buffer and check if we should flush
         buffer_key = CU_BUFFER_KEY.format(project_id=project_id)
         try:
-            await redis_client.client.rpush(buffer_key, json.dumps(usage_record))
-            buffer_len = await redis_client.client.llen(buffer_key)
+            await kv_client.client.rpush(buffer_key, json.dumps(usage_record))
+            buffer_len = await kv_client.client.llen(buffer_key)
 
             if buffer_len >= self._batch_size:
                 await self._flush_buffer(project_id)
@@ -120,7 +120,7 @@ class UsageTracker:
 
         try:
             # Get and clear buffer atomically
-            pipe = redis_client.client.pipeline()
+            pipe = kv_client.client.pipeline()
             pipe.lrange(buffer_key, 0, -1)
             pipe.delete(buffer_key)
             results = await pipe.execute()
@@ -160,7 +160,7 @@ class UsageTracker:
         """
         period_key = self._get_period_key(project_id)
         try:
-            value = await redis_client.get(period_key)
+            value = await kv_client.get(period_key)
             return int(value) if value else 0
         except Exception:
             return 0
@@ -202,7 +202,7 @@ class UsageTracker:
         rate_key = self._get_rate_key(project_id)
 
         try:
-            allowed, remaining = await redis_client.rate_limit_check(
+            allowed, remaining = await kv_client.rate_limit_check(
                 rate_key,
                 limits.rate_limit_per_second,
                 window_seconds=1,
@@ -248,7 +248,7 @@ class UsageTracker:
             # Find all buffer keys
             cursor = 0
             while True:
-                cursor, keys = await redis_client.client.scan(
+                cursor, keys = await kv_client.client.scan(
                     cursor, match="cu:buffer:*", count=100
                 )
                 for key in keys:

api/bootnode/core/billing/webhooks.py

@@ -21,7 +21,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from bootnode.config import get_settings
 from bootnode.core.billing.models import PlanTier, SubscriptionStatus
 from bootnode.core.billing.tiers import PricingTier, get_tier_limits
-from bootnode.core.cache import redis_client
+from bootnode.core.cache import kv_client
 from bootnode.db.models import Project, Subscription
 from bootnode.db.session import async_session
 
@@ -578,7 +578,7 @@ class CommerceWebhookHandler:
     async def _invalidate_cache(self, project_id: UUID) -> None:
         """Invalidate cached subscription data."""
         cache_key = f"billing:subscription:{project_id}"
-        await redis_client.delete(cache_key)
+        await kv_client.delete(cache_key)
 
 
 # Global singleton

api/bootnode/core/cache/__init__.py

@@ -1,5 +1,5 @@
 """Caching layer."""
 
-from bootnode.core.cache.redis import RedisClient, redis_client
+from bootnode.core.cache.kv import KVClient, kv_client
 
-__all__ = ["RedisClient", "redis_client"]
+__all__ = ["KVClient", "kv_client"]

api/bootnode/core/cache/kv.py

@@ -1,10 +1,14 @@
-"""Redis cache client."""
+"""Hanzo KV client.
+
+Hanzo KV speaks RESP, so the redis-py client library talks to it directly.
+"""
 
 import json
 from typing import Any
 
 import redis.asyncio as redis
 import structlog
+from redis.exceptions import RedisError
 
 from bootnode.config import get_settings
 
@@ -12,31 +16,32 @@ logger = structlog.get_logger()
 settings = get_settings()
 
 
-class RedisClient:
-    """Async Redis client wrapper."""
+class KVClient:
+    """Async Hanzo KV client wrapper."""
 
     def __init__(self) -> None:
         self._client: redis.Redis | None = None
 
     async def initialize(self) -> None:
-        """Initialize Redis connection."""
+        """Initialize the KV connection pool (connects lazily on first command)."""
         self._client = redis.from_url(
-            str(settings.redis_url),
+            settings.kv_url,
             encoding="utf-8",
             decode_responses=True,
+            socket_connect_timeout=2,
         )
-        logger.info("Redis connected", url=str(settings.redis_url).split("@")[-1])
+        logger.info("KV configured", url=settings.kv_url.split("@")[-1])
 
     async def close(self) -> None:
-        """Close Redis connection."""
+        """Close the KV connection."""
         if self._client:
             await self._client.aclose()
-            logger.info("Redis connection closed")
+            logger.info("KV connection closed")
 
     @property
     def client(self) -> redis.Redis:
         if not self._client:
-            raise RuntimeError("Redis not initialized")
+            raise RuntimeError("KV not initialized")
         return self._client
 
     # Key-Value Operations
@@ -97,11 +102,18 @@ class RedisClient:
         """
         Check rate limit using sliding window.
         Returns (allowed, remaining).
+
+        A KV outage allows the request: rate limiting degrades, the request
+        path does not fail.
         """
-        pipe = self.client.pipeline()
-        pipe.incr(key)
-        pipe.expire(key, window_seconds)
-        results = await pipe.execute()
+        try:
+            pipe = self.client.pipeline()
+            pipe.incr(key)
+            pipe.expire(key, window_seconds)
+            results = await pipe.execute()
+        except RedisError as e:
+            logger.warning("KV unavailable, rate limit not enforced", key=key, error=str(e))
+            return True, limit
 
         current = results[0]
         remaining = max(0, limit - current)
@@ -134,9 +146,13 @@ class RedisClient:
         method: str,
         params_hash: str,
     ) -> Any:
-        """Get cached RPC response."""
+        """Get cached RPC response. A KV outage is a cache miss."""
         key = self.rpc_cache_key(chain, network, method, params_hash)
-        return await self.get_json(key)
+        try:
+            return await self.get_json(key)
+        except RedisError as e:
+            logger.warning("KV unavailable, RPC cache skipped", key=key, error=str(e))
+            return None
 
     async def set_rpc_response(
         self,
@@ -147,9 +163,12 @@ class RedisClient:
         response: Any,
         ttl: int = 12,  # ~1 block on most chains
     ) -> None:
-        """Cache RPC response."""
+        """Cache RPC response. A KV outage skips the write."""
         key = self.rpc_cache_key(chain, network, method, params_hash)
-        await self.set_json(key, response, ex=ttl)
+        try:
+            await self.set_json(key, response, ex=ttl)
+        except RedisError as e:
+            logger.warning("KV unavailable, RPC cache skipped", key=key, error=str(e))
 
     # Pub/Sub for Webhooks
     async def publish(self, channel: str, message: str) -> int:
@@ -164,4 +183,4 @@ class RedisClient:
 
 
 # Global singleton
-redis_client = RedisClient()
+kv_client = KVClient()

api/bootnode/main.py

@@ -11,7 +11,7 @@ from fastapi.responses import ORJSONResponse
 from bootnode import __version__
 from bootnode.api import router as api_router
 from bootnode.config import get_settings
-from bootnode.core.cache import redis_client
+from bootnode.core.cache import kv_client
 from bootnode.core.datastore import datastore_client
 from bootnode.core.kms import inject_secrets
 from bootnode.db.session import engine, init_db
@@ -42,7 +42,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # Startup
     logger.info("Starting Bootnode API", env=settings.app_env)
     await init_db()
-    await redis_client.initialize()
+    await kv_client.initialize()
 
     # Initialize DataStore - optional, won't fail startup
     try:
@@ -80,7 +80,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
             logger.warning("ZAP server stop error", error=str(e))
 
     await engine.dispose()
-    await redis_client.close()
+    await kv_client.close()
     await datastore_client.close()
 
 

api/bootnode/workers/webhook.py

@@ -1,7 +1,7 @@
 """Webhook delivery worker.
 
-Processes webhook events from Redis queue and delivers them to registered endpoints.
-Uses arq (async Redis queue) for job processing - compatible with @hanzo/mq patterns.
+Processes webhook events from the Hanzo KV queue and delivers them to registered endpoints.
+Uses arq (RESP job queue) for job processing - compatible with @hanzo/mq patterns.
 """
 
 import asyncio
@@ -321,8 +321,8 @@ class WorkerSettings:
     on_startup = startup
     on_shutdown = shutdown
 
-    # Redis connection settings
-    redis_settings = RedisSettings.from_dsn(settings.redis_url)
+    # Hanzo KV connection (arq reads it as `redis_settings`)
+    redis_settings = RedisSettings.from_dsn(settings.kv_url)
 
     # Worker settings
     max_jobs = 100

api/tests/test_fleet_admin_gate.py

@@ -17,7 +17,7 @@ import os
 
 # Base/SQLite so the mounted router's db.session engine imports without Postgres.
 os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///:memory:")
-os.environ.setdefault("REDIS_URL", "redis://localhost:6379/0")
+os.environ.setdefault("KV_URL", "redis://localhost:6379/0")
 
 import pytest
 from fastapi import FastAPI, HTTPException, status

api/tests/test_kv.py

@@ -0,0 +1,64 @@
+"""Hanzo KV client: configured by KV_URL, and an outage never fails a request."""
+
+import pytest
+
+from bootnode.config import Settings
+from bootnode.core.cache import kv as kv_module
+from bootnode.core.cache.kv import KVClient
+
+# Nothing listens on port 1, so every command fails with a connection error.
+DOWN = "redis://127.0.0.1:1/0"
+
+
+def test_kv_url_is_read_from_env(monkeypatch):
+    monkeypatch.setenv("KV_URL", "redis://kv.hanzo.svc.cluster.local:6379/0")
+    monkeypatch.setenv("REDIS_URL", "redis://ignored:6379/0")
+    assert Settings().kv_url == "redis://kv.hanzo.svc.cluster.local:6379/0"
+
+
+@pytest.fixture
+async def down(monkeypatch):
+    monkeypatch.setattr(kv_module.settings, "kv_url", DOWN)
+    client = KVClient()
+    await client.initialize()
+    yield client
+    await client.close()
+
+
+async def test_rate_limit_allows_when_kv_down(down):
+    assert await down.rate_limit_check("rate:test", 100) == (True, 100)
+
+
+async def test_rpc_cache_misses_when_kv_down(down):
+    assert await down.get_rpc_response("ethereum", "mainnet", "eth_chainId", "h") is None
+    await down.set_rpc_response("ethereum", "mainnet", "eth_chainId", "h", "0x1")
+
+
+class _Pipe:
+    def __init__(self, count: int):
+        self._count = count
+
+    def incr(self, key):
+        pass
+
+    def expire(self, key, seconds):
+        pass
+
+    async def execute(self):
+        return [self._count, True]
+
+
+class _KV:
+    def __init__(self, count: int):
+        self._count = count
+
+    def pipeline(self):
+        return _Pipe(self._count)
+
+
+async def test_rate_limit_enforced_when_kv_up():
+    client = KVClient()
+    client._client = _KV(101)
+    assert await client.rate_limit_check("rate:test", 100) == (False, 0)
+    client._client = _KV(40)
+    assert await client.rate_limit_check("rate:test", 100) == (True, 60)

compose.yml

@@ -114,7 +114,7 @@ services:
       APP_ENV: development
       DEBUG: "true"
       DATABASE_URL: postgresql+asyncpg://bootnode:bootnode@postgres:5432/bootnode
-      REDIS_URL: redis://redis:6379/0
+      KV_URL: redis://redis:6379/0
       DATASTORE_URL: clickhouse://bootnode:bootnode@datastore:8123/bootnode
       QDRANT_URL: http://qdrant:6333
       MEILISEARCH_URL: http://meilisearch:7700

infra/README.md

@@ -112,7 +112,7 @@ See `.env.example` for all configuration options.
 
 Key variables:
 - `DATABASE_URL` - PostgreSQL connection
-- `REDIS_URL` - Redis connection
+- `KV_URL` - Hanzo KV connection (RESP)
 - `DATASTORE_URL` - ClickHouse connection
 - `*_RPC` - Chain RPC endpoints
 

infra/compose.yml

@@ -98,7 +98,7 @@ services:
     environment:
       APP_ENV: ${APP_ENV:-development}
       DATABASE_URL: postgresql+asyncpg://bootnode:${POSTGRES_PASSWORD:-bootnode_secret}@postgres:5432/bootnode
-      REDIS_URL: redis://redis:6379/0
+      KV_URL: redis://redis:6379/0
       DATASTORE_URL: clickhouse://bootnode:${CLICKHOUSE_PASSWORD:-bootnode_secret}@datastore:8123/bootnode
       JWT_SECRET: ${JWT_SECRET:-change_me_in_production}
       API_KEY_SALT: ${API_KEY_SALT:-change_me_in_production}
@@ -179,7 +179,7 @@ services:
     command: ["python", "-m", "bootnode.workers.webhook"]
     environment:
       DATABASE_URL: postgresql+asyncpg://bootnode:${POSTGRES_PASSWORD:-bootnode_secret}@postgres:5432/bootnode
-      REDIS_URL: redis://redis:6379/0
+      KV_URL: redis://redis:6379/0
       DATASTORE_URL: clickhouse://bootnode:${CLICKHOUSE_PASSWORD:-bootnode_secret}@datastore:8123/bootnode
     depends_on:
       - api