Bring your own backend

A 10xGraph server keeps its state in three places, and each one is an abstract class you can replace. This page lists them, shows how to select a custom one, and documents the rules a replacement must keep. The default for each is enough for most deployments; read on when your infrastructure is different, for example a database the built-in classes do not cover.

Valkey, Dragonfly and KeyDB speak the Redis protocol, so you do not need any of this for them: set cache_store.type to redis and point its URL at the server, and the built-in classes work.

The three classes

Each of the three has an abstract aping(). The server calls it at startup and on /health/ready, so a replacement tells the server whether it is usable in the same way as a built-in.

What the cache store backs

Select a custom backend

The checkpointer, store and media store are passed to compile() in your graph module. The server does not read the checkpointer or store keys to build them for you (see Configure the server):

app = graph.compile(checkpointer=MyCheckpointer(), store=MyStore())

The cache store is the one backend chosen by the cache_store object in 10xgraph.json. Its type picks the store:

1. "memory" (the default, also when cache_store is absent): InMemoryCacheStore, correct for one process only.
2. "redis": RedisCacheStore. url is optional: empty uses the REDIS_URL environment variable, and a value you set (a literal or ${VAR}) wins.
3. "custom": your own store. path is a module:attr path to a BaseCacheStore instance, or a zero-argument factory (a class or function) that returns one.

REDIS_URL alone never selects Redis.

{
  "agent": "graph.agent:app",
  "cache_store": {
    "type": "custom",
    "path": "graph.cache:store"
  }
}

from my_company.cache import MyCacheStore

store = MyCacheStore("https://cache.internal")

The server logs one INFO line at boot that names the store and the config key it came from, for example Cache store: RedisCacheStore (cache_store.url, localhost:6379). A store can add its own endpoint to that line by implementing describe(), which returns a string such as host:port and never credentials.

The aping() rules

All three classes use the same pattern:

• Return when the backend is reachable and usable. Raise any exception when it is not.
• Be read-only and cheap. The server may call it on every readiness probe, which is every few seconds. Do not write data, create tables or take locks.
• An in-process implementation just returns.

PgCheckpointer.aping() is the reference: it runs SELECT 1, raises RuntimeError when the recorded schema version is older than the code expects, and pings its own Redis. It never creates the schema; the server runs asetup() at boot.

aping() is abstract on BaseCheckpointer, BaseStore and BaseCacheStore. A subclass without it raises TypeError when created.

What a failed ping does:

When the cache store is down and no fail-closed feature uses it, the server logs one WARNING and starts, with locks and caps per process. The readiness response then reports "cache": "fail" and still answers 200. See Ping and health and the startup check.

The BaseCacheStore contract

Two primitives cover every feature above. Timed slots (acquire, renew, release) are a set of expiring slots per key, capped at limit. A lock is a slot set with limit=1; a rate limit is a slot set whose slots are never released and expire after the window. Keys with a TTL (get, set, delete) hold values such as idempotency records, ownership entries and revocations.

acquire returns SlotResult(granted, count, oldest_expiry, slot_id): whether a slot was granted, the live slots after the call (including yours when granted), the expiry of the oldest live slot on the store’s clock (or None), and the new slot’s id when granted (or None).

An implementation must keep these rules. The contract test suite checks them.

• acquire is atomic. Drop expired slots, and only if the live count is below limit, add a new slot expiring at now() + ttl. Two callers can never both take the last slot. limit < 1 always denies.
• A denied acquire adds nothing. The rate limiter relies on this: a client over its limit does not extend its own penalty.
• renew never re-creates a slot. It returns False when the slot expired or was released.
• release of a missing slot is not an error.
• set(mode="if_absent") writes only when the key does not exist.
• set(mode="if_greater") treats the value as a decimal number and writes only when it is greater than the stored one, or nothing is stored. It is atomic. Token revocation uses it so an older cutoff never overwrites a newer one. A value that is not a decimal number raises ValueError.
• ttl is in seconds and must be positive. set(ttl=None) keeps the key until it is deleted. A write replaces the TTL of the key it writes.
• TTLs are never dropped early. A key or slot must not disappear before its TTL. A store that evicts data under memory pressure breaks locks: two workers run the same thread. For Redis this means maxmemory-policy must be noeviction or a volatile-* policy. Any other policy is a deployment mistake.
• Every method touches exactly one key. That keeps the store safe on Redis Cluster and on sharded stores.
• Raise on errors. When the backend fails, raise any exception. The features decide whether to fail open or closed (see the table above). The store never swallows errors.
• Use the store’s clock. All times come from now(), so pods with skewed clocks agree on when a slot expires.

Keys arrive as "<feature>:<id>", for example concurrency:runs:thread:42. The store may map the feature segment to its own namespace and add a deployment prefix, but two different keys must never land on the same entry, and any string (with :, spaces or unicode) is a valid key. A key holds either slots or a value, never both.

Built-in stores

RedisCacheStore uses Lua scripts on sorted sets for the slots, SET with PX for values, and a small script for if_greater. It applies REDIS_KEY_PREFIX and keeps the Redis key layouts of earlier releases, except for the rate limiter, which uses the 10xgraph:ratelimit2 namespace.Its connection limits come from the REDIS_* settings; see the environment reference.

InMemoryCacheStore keeps everything in one process. It is correct for a single worker and wrong for several: locks, caps and revocations do not cross processes. To bound memory it caps slot keys at 50,000 and plain values at 10,000 and evicts the least recently used entry at the cap, with a warning. That is its one exception to “TTLs are never dropped early”. It never evicts lease keys (concurrency:) or revocation keys (revoked:) while they are live, so a lock or a revoked token cannot disappear.

A minimal skeleton

This skeleton shows the shape. The comments say what each method must do on your backend.

from typing import Literal

from tenxgraph_api import BaseCacheStore, SlotResult


class MyCacheStore(BaseCacheStore):
    def __init__(self, url: str) -> None:
        self._url = url
        self._client = ...  # your async client

    async def aping(self) -> None:
        # Read-only and cheap: one round trip. Raise if the backend does not answer.
        ...

    async def now(self) -> float:
        # The backend's own clock, not time.time() on this pod.
        ...

    async def acquire(self, key: str, limit: int, ttl: float) -> SlotResult:
        # One atomic step: drop expired slots, count live ones, add a slot only if
        # count < limit. Never add on denial. Return the count after the call, the
        # oldest live expiry on the store's clock, and the new slot id when granted.
        ...

    async def renew(self, key: str, slot_id: str, ttl: float) -> bool:
        # Move the expiry of a slot that still exists. Return False if it is gone.
        ...

    async def release(self, key: str, slot_id: str) -> None:
        # Remove the slot; a missing slot is fine.
        ...

    async def get(self, key: str) -> bytes | None:
        ...

    async def set(
        self,
        key: str,
        value: bytes,
        ttl: float | None,
        mode: Literal["always", "if_absent", "if_greater"] = "always",
    ) -> bool:
        # "if_absent": write only if missing. "if_greater": atomic compare of two
        # decimal numbers. Return True when you wrote.
        ...

    async def delete(self, key: str) -> None:
        ...

    def describe(self) -> str | None:
        return "cache.internal:443"  # no credentials

    async def aclose(self) -> None:
        ...

Select it with "cache_store": {"type": "custom", "path": "my_company.cache:store"} as shown above. store can be an instance or a function that takes no arguments and returns one.

Test it with the contract suite

tenxgraph_api.testing.cache_store_contract ships the checks the built-in stores pass: aping, the slot limit, 50 concurrent acquire calls granting exactly the limit, release and renew, slot expiry, get, set and delete with all three modes (including 50 concurrent if_greater writes leaving the maximum), the clock, and keys with odd characters. Each check opens its own store from your factory, works under its own keys and closes the store, so it is safe to run against a shared server.

As a pytest mixin, subclass CacheStoreContract and implement make_store():

from tenxgraph_api.testing.cache_store_contract import CacheStoreContract

from my_company.cache import MyCacheStore


class TestMyCacheStore(CacheStoreContract):
    def make_store(self):
        return MyCacheStore("https://cache.internal")

Or call it from any async code:

from tenxgraph_api.testing.cache_store_contract import run_cache_store_contract

await run_cache_store_contract(lambda: MyCacheStore("https://cache.internal"))

make_store() and the factory can return a store or an awaitable of one. A failed check raises AssertionError. The timing checks use TTLs of a few hundred milliseconds, so the store must answer well within that.

Custom checkpointers and stores

A custom BaseCheckpointer implements every abstract async method, including aping(), and aget_thread_owner when auth is on. A custom BaseStore implements its methods and aping(). Both are passed to compile(). The method lists are in the checkpointer reference and the store reference, and Extension points shows where every extension class is wired.

See also

• Production checklist for the startup check and the Redis settings
• Rate limiting for the limiter that uses the cache store
• Scaling and reliability for what the cache store shares across workers