perf(cache): hash-tag the cache namespace so an instance uses one shard (#1058)

On Redis Cluster a key's slot comes from the substring inside the first
{...}, when one is present. Untagged, one deployment's keys spread over
every slot, so its client opens and holds a connection to every node in
the cluster. Wrapping the namespace in braces puts them all on one slot,
and therefore one node, cutting each deployment's connection count to
the cluster by a factor of the shard count. Namespaces still hash
independently of each other, so keys stay spread across the cluster and
no shard becomes a hotspot.

The tag needs two spellings, because the two ways a key gets built treat
the string differently. cashews runs `prefix=` through format
substitution, so braces have to be doubled there to survive as literals;
keys built by concatenation need them single. A single brace passed to
cashews is read as an empty substitution field and the namespace is
dropped entirely, which would let two deployments collide on one key --
hence two clearly named helpers rather than one string, and a test that
the two paths produce identical bytes.

No key format change for a non-cluster backend, and no migration: the
old keys simply age out by TTL.

src/cache/client.py

@@ -123,6 +123,28 @@ def get_cache_namespace() -> str:
     return cast(str, settings.CACHE.NAMESPACE)
 
 
+# On Redis Cluster a key's slot is derived from the substring inside the first
+# {...}, when one is present. Tagging the namespace puts every key an instance
+# writes on a single slot, and therefore a single shard, so its client holds
+# connections to one node rather than to all of them. Namespaces still hash
+# independently of one another, so keys stay spread across the cluster.
+#
+# Two spellings, because the two ways a key gets built treat the string
+# differently: cashews runs `prefix=` through format substitution, so braces
+# have to be doubled to survive as literals, while direct construction does no
+# substitution and needs them single. Both render to the same bytes, which
+# tests/cache/test_cache_namespace_hash_tag.py asserts -- a mismatch would send
+# writes and deletes to different keys with nothing raised.
+def cache_key_namespace() -> str:
+    """Tagged namespace for keys built by string concatenation."""
+    return "{" + get_cache_namespace() + "}"
+
+
+def cache_prefix_namespace() -> str:
+    """Tagged namespace for cashews `prefix=`, which format-substitutes."""
+    return "{{" + get_cache_namespace() + "}}"
+
+
 async def init_cache() -> None:
     """Initialize and verify cache connection if enabled."""
     async with _cache_lock:
@@ -256,6 +278,8 @@ __all__ = [
     "init_cache",
     "close_cache",
     "cache",
+    "cache_key_namespace",
+    "cache_prefix_namespace",
     "safe_cache_delete",
     "safe_cache_set",
 ]

src/crud/collection.py

@@ -10,7 +10,8 @@ from sqlalchemy.orm import make_transient_to_detached
 from src import models
 from src.cache.client import (
     cache,
-    get_cache_namespace,
+    cache_key_namespace,
+    cache_prefix_namespace,
     safe_cache_delete,
     safe_cache_set,
 )
@@ -22,13 +23,13 @@ logger = getLogger(__name__)
 COLLECTION_CACHE_KEY_TEMPLATE = (
     "v2:workspace:{workspace_name}:collection:{observer}:{observed}"
 )
-COLLECTION_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2"
+COLLECTION_LOCK_PREFIX = f"{cache_prefix_namespace()}:lock:v2"
 
 
 def collection_cache_key(workspace_name: str, observer: str, observed: str) -> str:
     """Generate cache key for collection."""
     return (
-        get_cache_namespace()
+        cache_key_namespace()
         + ":"
         + COLLECTION_CACHE_KEY_TEMPLATE.format(
             workspace_name=workspace_name,
@@ -41,7 +42,7 @@ def collection_cache_key(workspace_name: str, observer: str, observed: str) -> s
 @cache(
     key=COLLECTION_CACHE_KEY_TEMPLATE,
     ttl=f"{settings.CACHE.DEFAULT_TTL_SECONDS}s",
-    prefix=get_cache_namespace(),
+    prefix=cache_prefix_namespace(),
     condition=NOT_NONE,
 )
 @cache.locked(

src/crud/peer.py

@@ -12,7 +12,12 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import make_transient_to_detached
 
 from src import models, schemas
-from src.cache.client import cache, get_cache_namespace, safe_cache_delete
+from src.cache.client import (
+    cache,
+    cache_key_namespace,
+    cache_prefix_namespace,
+    safe_cache_delete,
+)
 from src.config import settings
 from src.crud.workspace import get_or_create_workspace
 from src.exceptions import (
@@ -32,13 +37,13 @@ logger = getLogger(__name__)
 PEER_NAME_MAX_LENGTH = 512
 
 PEER_CACHE_KEY_TEMPLATE = "v2:workspace:{workspace_name}:peer:{peer_name}"
-PEER_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2"
+PEER_LOCK_PREFIX = f"{cache_prefix_namespace()}:lock:v2"
 
 
 def peer_cache_key(workspace_name: str, peer_name: str) -> str:
     """Generate cache key for peer."""
     return (
-        get_cache_namespace()
+        cache_key_namespace()
         + ":"
         + PEER_CACHE_KEY_TEMPLATE.format(
             workspace_name=workspace_name,
@@ -391,7 +396,7 @@ async def get_or_create_peers(
 @cache(
     key=PEER_CACHE_KEY_TEMPLATE,
     ttl=f"{settings.CACHE.DEFAULT_TTL_SECONDS}s",
-    prefix=get_cache_namespace(),
+    prefix=cache_prefix_namespace(),
     condition=NOT_NONE,
 )
 @cache.locked(

src/crud/session.py

@@ -29,7 +29,8 @@ from sqlalchemy.types import BigInteger, Boolean
 from src import models, schemas
 from src.cache.client import (
     cache,
-    get_cache_namespace,
+    cache_key_namespace,
+    cache_prefix_namespace,
     safe_cache_delete,
     safe_cache_set,
 )
@@ -67,13 +68,13 @@ class SessionDeletionResult:
 
 
 SESSION_CACHE_KEY_TEMPLATE = "v2:workspace:{workspace_name}:session:{session_name}"
-SESSION_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2"
+SESSION_LOCK_PREFIX = f"{cache_prefix_namespace()}:lock:v2"
 
 
 def session_cache_key(workspace_name: str, session_name: str) -> str:
     """Generate cache key for session."""
     return (
-        get_cache_namespace()
+        cache_key_namespace()
         + ":"
         + SESSION_CACHE_KEY_TEMPLATE.format(
             workspace_name=workspace_name,
@@ -85,7 +86,7 @@ def session_cache_key(workspace_name: str, session_name: str) -> str:
 @cache(
     key=SESSION_CACHE_KEY_TEMPLATE,
     ttl=f"{settings.CACHE.DEFAULT_TTL_SECONDS}s",
-    prefix=get_cache_namespace(),
+    prefix=cache_prefix_namespace(),
     condition=NOT_NONE,
 )
 @cache.locked(

src/crud/workspace.py

@@ -15,7 +15,8 @@ from sqlalchemy.orm import make_transient_to_detached
 from src import models, schemas
 from src.cache.client import (
     cache,
-    get_cache_namespace,
+    cache_key_namespace,
+    cache_prefix_namespace,
     safe_cache_delete,
     safe_cache_set,
 )
@@ -40,13 +41,13 @@ class WorkspaceDeletionResult:
 
 
 WORKSPACE_CACHE_KEY_TEMPLATE = "v2:workspace:{workspace_name}"
-WORKSPACE_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2"
+WORKSPACE_LOCK_PREFIX = f"{cache_prefix_namespace()}:lock:v2"
 
 
 def workspace_cache_key(workspace_name: str) -> str:
     """Generate cache key for workspace."""
     return (
-        get_cache_namespace()
+        cache_key_namespace()
         + ":"
         + WORKSPACE_CACHE_KEY_TEMPLATE.format(workspace_name=workspace_name)
     )
@@ -55,7 +56,7 @@ def workspace_cache_key(workspace_name: str) -> str:
 @cache(
     key=WORKSPACE_CACHE_KEY_TEMPLATE,
     ttl=f"{settings.CACHE.DEFAULT_TTL_SECONDS}s",
-    prefix=get_cache_namespace(),
+    prefix=cache_prefix_namespace(),
     condition=NOT_NONE,
 )
 @cache.locked(

tests/test_cache_key_namespace.py

@@ -0,0 +1,62 @@
+"""The namespace hash tag must survive both ways a cache key gets built.
+
+cashews format-substitutes `prefix=`, direct construction does not, so the two
+need different spellings of the same tag. If they ever diverge, a write and its
+invalidation land on different keys and nothing raises -- the cache just serves
+stale rows. These tests are what fails instead.
+"""
+
+import pytest
+from redis.crc import key_slot
+
+from src.cache.client import (
+    cache,
+    cache_key_namespace,
+    cache_prefix_namespace,
+    get_cache_namespace,
+)
+from src.crud.session import SESSION_CACHE_KEY_TEMPLATE, session_cache_key
+
+
+def test_both_spellings_render_the_same_tag():
+    ns = get_cache_namespace()
+    assert cache_key_namespace() == "{" + ns + "}"
+    # Doubled braces collapse to single ones when cashews formats the prefix.
+    assert cache_prefix_namespace().format() == cache_key_namespace()
+
+
+@pytest.mark.asyncio
+async def test_decorator_key_matches_helper_key():
+    """The key the decorator writes is the key the helper computes."""
+
+    @cache(
+        key=SESSION_CACHE_KEY_TEMPLATE,
+        prefix=cache_prefix_namespace(),
+        ttl="60s",
+    )
+    async def get_session(workspace_name: str, session_name: str) -> str:
+        # The names matter: cashews fills the key template from them.
+        return f"{workspace_name}/{session_name}"
+
+    await get_session(workspace_name="w1", session_name="s1")
+
+    written = [k async for k in cache.scan("*")]
+    assert session_cache_key("w1", "s1") in written
+
+
+def test_one_namespace_hashes_to_one_slot():
+    """Every key an instance writes shares a Redis Cluster slot."""
+    keys = [
+        session_cache_key("w1", "s1"),
+        session_cache_key("w2", "s2"),
+        f"{cache_key_namespace()}:lock:v2:anything",
+    ]
+    assert len({key_slot(k.encode()) for k in keys}) == 1
+
+
+def test_namespaces_hash_independently():
+    """Tagging must not collapse the whole fleet onto one shard."""
+    slots = {
+        key_slot(("{" + n + "}:v2:workspace:w").encode()) for n in ("a1", "b2", "c3")
+    }
+    assert len(slots) > 1