From fa66f8956250960adb9b94c9a2982edf8bab4dd7 Mon Sep 17 00:00:00 2001 From: Rajat Ahuja Date: Mon, 13 Jul 2026 00:44:51 -0400 Subject: [PATCH] feat: add support for redis cluster --- .env.template | 2 ++ pyproject.toml | 2 +- src/cache/client.py | 6 +++++- src/config.py | 10 ++++++++++ src/crud/collection.py | 1 + src/crud/peer.py | 1 + src/crud/session.py | 1 + src/crud/workspace.py | 1 + uv.lock | 10 +++++----- 9 files changed, 27 insertions(+), 7 deletions(-) diff --git a/.env.template b/.env.template index 167fa857..2516a239 100644 --- a/.env.template +++ b/.env.template @@ -287,9 +287,11 @@ LLM_OPENAI_API_KEY=your-api-key-here # ============================================================================= # CACHE_ENABLED=false # CACHE_URL="redis://localhost:6379/0?suppress=true" +# CACHE_CLUSTER=false # true when CACHE_URL is a Redis Cluster (e.g. Memorystore for Redis Cluster) # CACHE_NAMESPACE="honcho" # Inherits from NAMESPACE if not set # CACHE_DEFAULT_TTL_SECONDS=300 # CACHE_DEFAULT_LOCK_TTL_SECONDS=5 +# CACHE_LOCK_WAIT_CHECK_INTERVAL_SECONDS=0.1 # ============================================================================= # CORS Settings diff --git a/pyproject.toml b/pyproject.toml index a79a3d85..6766c2c8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ dependencies = [ "lancedb>=0.25.3", "pyarrow>=19.0.0", "redis>=7.0.0,<8.0.0", - "cashews[redis]==7.4.4", + "cashews[redis]==7.5.0", "scikit-learn>=1.6.0", "prometheus_client>=0.21.0", "cloudevents>=1.12.0,<2.0", diff --git a/src/cache/client.py b/src/cache/client.py index 319fab6e..29d8c187 100644 --- a/src/cache/client.py +++ b/src/cache/client.py @@ -45,11 +45,15 @@ async def init_cache() -> None: cache.setup("mem://", pickle_type=PicklerType.SQLALCHEMY) return - # Setup cache with Redis backend + # Setup cache with Redis backend. CACHE_CLUSTER selects the + # cluster-aware client, which follows the MOVED redirects a Redis + # Cluster returns for keys hashed to another shard; the standalone + # client treats those as command errors. try: cache.setup( settings.CACHE.URL, pickle_type=PicklerType.SQLALCHEMY, + cluster=settings.CACHE.CLUSTER, ) except Exception as setup_err: diff --git a/src/config.py b/src/config.py index 773ffcd6..e4dde41a 100644 --- a/src/config.py +++ b/src/config.py @@ -1234,6 +1234,10 @@ class CacheSettings(HonchoSettings): ENABLED: bool = False URL: str = "redis://localhost:6379/0?suppress=true" + # URL points at a Redis Cluster (OSS cluster protocol, e.g. GCP Memorystore + # for Redis Cluster). A standalone client cannot follow the MOVED redirects + # such deployments return for keys hashed to another shard. + CLUSTER: bool = False NAMESPACE: str | None = None DEFAULT_TTL_SECONDS: Annotated[int, Field(default=300, ge=1, le=86_400)] = ( 300 # how long to keep items in cache @@ -1243,6 +1247,12 @@ class CacheSettings(HonchoSettings): 5 # how long to hold a lock on a resource when fetching DB after cache miss ) + # Polling interval while waiting for another worker's fetch lock. cashews + # defaults to 0, which busy-spins the event loop for the whole wait. + LOCK_WAIT_CHECK_INTERVAL_SECONDS: Annotated[ + float, Field(default=0.1, gt=0, le=5) + ] = 0.1 + class SurprisalSettings(BaseModel): """Settings for tree-based surprisal sampling during dreams.""" diff --git a/src/crud/collection.py b/src/crud/collection.py index 2790cec1..63a775e3 100644 --- a/src/crud/collection.py +++ b/src/crud/collection.py @@ -48,6 +48,7 @@ def collection_cache_key(workspace_name: str, observer: str, observed: str) -> s key=COLLECTION_CACHE_KEY_TEMPLATE, ttl=f"{settings.CACHE.DEFAULT_LOCK_TTL_SECONDS}s", prefix=COLLECTION_LOCK_PREFIX, + check_interval=settings.CACHE.LOCK_WAIT_CHECK_INTERVAL_SECONDS, ) async def _fetch_collection( db: AsyncSession, diff --git a/src/crud/peer.py b/src/crud/peer.py index 21792b0f..f7936761 100644 --- a/src/crud/peer.py +++ b/src/crud/peer.py @@ -152,6 +152,7 @@ async def get_or_create_peers( key=PEER_CACHE_KEY_TEMPLATE, ttl=f"{settings.CACHE.DEFAULT_LOCK_TTL_SECONDS}s", prefix=PEER_LOCK_PREFIX, + check_interval=settings.CACHE.LOCK_WAIT_CHECK_INTERVAL_SECONDS, ) async def _fetch_peer( db: AsyncSession, diff --git a/src/crud/session.py b/src/crud/session.py index 40cdacac..4310cb3a 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -72,6 +72,7 @@ def session_cache_key(workspace_name: str, session_name: str) -> str: key=SESSION_CACHE_KEY_TEMPLATE, ttl=f"{settings.CACHE.DEFAULT_LOCK_TTL_SECONDS}s", prefix=SESSION_LOCK_PREFIX, + check_interval=settings.CACHE.LOCK_WAIT_CHECK_INTERVAL_SECONDS, ) async def _fetch_session( db: AsyncSession, diff --git a/src/crud/workspace.py b/src/crud/workspace.py index 52662c79..c9040f55 100644 --- a/src/crud/workspace.py +++ b/src/crud/workspace.py @@ -60,6 +60,7 @@ def workspace_cache_key(workspace_name: str) -> str: key=WORKSPACE_CACHE_KEY_TEMPLATE, ttl=f"{settings.CACHE.DEFAULT_LOCK_TTL_SECONDS}s", prefix=WORKSPACE_LOCK_PREFIX, + check_interval=settings.CACHE.LOCK_WAIT_CHECK_INTERVAL_SECONDS, ) async def _fetch_workspace( db: AsyncSession, workspace_name: str diff --git a/uv.lock b/uv.lock index c16f23c4..bd4ffae4 100644 --- a/uv.lock +++ b/uv.lock @@ -8,7 +8,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-06-27T20:44:32.746059Z" +exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. exclude-newer-span = "P5D" [manifest] @@ -275,11 +275,11 @@ wheels = [ [[package]] name = "cashews" -version = "7.4.4" +version = "7.5.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/8c/5d/26eb556824a7ac9e24f751645961d2078b7b15be105f7fc39eda5308896f/cashews-7.4.4.tar.gz", hash = "sha256:dca761c60192bfe354abd6e9eb98d6f62c817e675df3fbe7d1bdfaa4303d1320", size = 92948, upload-time = "2025-12-06T22:31:56.187Z" } +sdist = { url = "https://files.pythonhosted.org/packages/44/73/31598b352165cd0f0b777df1eb67e33f334e29fcd7eb4f4bb48a41b9affe/cashews-7.5.0.tar.gz", hash = "sha256:3f88b8c5ced0ea4826915a1ff67055b647252dd65ef25f4813316a6341f00b37", size = 97699, upload-time = "2026-03-02T22:28:52.462Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/bf/65/29d94c27dfa3cdb213ae62a328c6efe6cd37b888d334e90ecaa22eadafe9/cashews-7.4.4-py3-none-any.whl", hash = "sha256:d5b8fc3cb590ed388823388b972947fd5659e2a94109af107cb508a3240f5ef0", size = 79893, upload-time = "2025-12-06T22:31:53.918Z" }, + { url = "https://files.pythonhosted.org/packages/0d/14/06cca741567a2ec458fb1db9d053e72477d9da2be387c1d705cb1060b2c6/cashews-7.5.0-py3-none-any.whl", hash = "sha256:e79cb4e5cc164d8f2d2856b166d45dcc2dd8d53b95874d2c6d07dfdb1c9ac3c4", size = 82413, upload-time = "2026-03-02T22:28:50.98Z" }, ] [package.optional-dependencies] @@ -1217,7 +1217,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "alembic", specifier = ">=1.14.0" }, - { name = "cashews", extras = ["redis"], specifier = "==7.4.4" }, + { name = "cashews", extras = ["redis"], specifier = "==7.5.0" }, { name = "cloudevents", specifier = ">=1.12.0,<2.0" }, { name = "fastapi", extras = ["standard"], specifier = ">=0.131.0" }, { name = "fastapi-pagination", specifier = ">=0.14.2" },