Cache

A framework-provided key/value cache accessible as self.cache on every handler. Memory-backed for hot reads, flushed to a per-worker SQLite file via a periodic write-behind loop, and optionally shared across workers via a periodic peer-sync loop.

The cache is opt-in (cache.enabled=false by default). When disabled, handler.cache is a no-op stub so handler code can call self.cache.set(...) unconditionally — no if self.cache is not None guards.

---

Quick reference

The handler-facing surface is five methods:

Method	Sync?	What it does
set(key, value, *, ttl=None, scope=CacheScope.LOCAL)	sync	Store a value with optional TTL (seconds) and peer-sync scope. Memory write now; SQLite on the next flush cycle.
await get(key, *, as_type=None)	async	Lookup with memory → DB fallback; None on miss. as_type=SomeModel revives the JSON via model_validate() for a typed return.
peek(key)	sync	Memory-only probe, no DB fallback. TTL-aware, but a key that lives only on disk reports None.
delete(key)	sync	Drop from memory, schedule the DB-row deletion for the next flush. Returns True iff the key was in memory. Local-only — peers keep their copy.
key in cache	sync	TTL-aware membership check, memory only, no LRU bump.

Rules of thumb:

• get when you want the truth — it is the only method that consults SQLite. peek and in see only what is already warm in memory.
• get is async by contract even when the cache is disabled (the no-op stub keeps the same signature), so call sites do not change between enabled and disabled states.
• For cross-worker invalidation use TTL, never delete — see Delete is local-only.
• The four sync methods are safe inside handler.offload(); get() is loop-only. Warm the keys you need with get before offloading, then peek inside the offloaded function.

Full semantics, metrics side effects, and sharp edges: see the API reference below.

---

What it solves

Before this feature, handlers that wanted to memoize subprocess results had to hand-roll their own dict + TTL + eviction policy (see the pre-cache integration demo). Each ad-hoc cache reinvented:

• Eviction when the dict grew unbounded
• Hit/miss counters and metrics
• TTL bookkeeping
• Durability across worker restarts
• Cross-worker visibility

Drakkar’s cache gives you all of these for free, plus:

• arrange() can short-circuit tasks with PrecomputedResult built from a cache hit — skipping the subprocess entirely.
• Workers in the same cluster can share cached values via peer sync (eventually consistent, LWW).
• Operators inspect cache contents in the operator UI and scrape size/hit metrics via Prometheus.

---

How it flows

flowchart LR
    H["handler code\nself.cache.set(k, v)"] --> MEM["memory dict\n(OrderedDict, LRU)"]
    MEM -. flush every Ns .-> DB[("worker-cache.db\nSQLite WAL")]
    MEM -->|peek / get hit| R["return value"]
    H -->|get on miss| DB
    DB -->|warm| MEM
    PEER[("peer-cache.db\nother workers")] -. sync every Ns .-> DB
    CLEAN["cleanup loop\nevery Ns"] -. delete expired .-> DB

    style H fill:#fef3c7,stroke:#d97706,color:#1a1a1a
    style MEM fill:#d1fae5,stroke:#059669,color:#1a1a1a
    style DB fill:#e0f2fe,stroke:#0284c7,color:#1a1a1a
    style PEER fill:#f5f3ee,stroke:#94a3b8,color:#1a1a1a
    style CLEAN fill:#fef3c7,stroke:#d97706,color:#1a1a1a
    style R fill:#d1fae5,stroke:#059669,color:#1a1a1a

flowchart LR
    H["handler code\nself.cache.set(k, v)"] --> MEM["memory dict\n(OrderedDict, LRU)"]
    MEM -. flush every Ns .-> DB[("worker-cache.db\nSQLite WAL")]
    MEM -->|peek / get hit| R["return value"]
    H -->|get on miss| DB
    DB -->|warm| MEM
    PEER[("peer-cache.db\nother workers")] -. sync every Ns .-> DB
    CLEAN["cleanup loop\nevery Ns"] -. delete expired .-> DB

    style H fill:#422006,stroke:#f59e0b,color:#fef3c7
    style MEM fill:#065f46,stroke:#34d399,color:#d1fae5
    style DB fill:#172554,stroke:#60a5fa,color:#e2e8f0
    style PEER fill:#1e293b,stroke:#64748b,color:#94a3b8
    style CLEAN fill:#422006,stroke:#f59e0b,color:#fef3c7
    style R fill:#065f46,stroke:#34d399,color:#d1fae5

Three periodic loops run alongside the main processing pipeline, registered as system periodic tasks (they appear with a [system] badge on the UI’s periodic tasks page alongside user-defined @periodic methods):

Loop	Default interval	What it does
cache.flush	flush_interval_seconds (3s)	Drains the pending _dirty map to SQLite via UPSERT / DELETE in one transaction
cache.sync	peer_sync.interval_seconds (30s)	Pulls recent rows from peer workers’ -cache.db files, LWW-merges into local DB
cache.cleanup	cleanup_interval_seconds (60s)	Removes rows where expires_at_ms < now_ms; refreshes DB-size Prometheus gauges

A reader aiosqlite connection is opened alongside the writer so get() DB-fallback queries never queue behind a flush/sync/cleanup commit. WAL mode lets the reader see consistent snapshots during concurrent writes.

---

Minimal example: memoize a subprocess result

import drakkar as dk
from drakkar import CacheScope

class MyHandler(dk.BaseDrakkarHandler[MyInput, MyOutput]):
    async def arrange(self, messages, pending):
        tasks = []
        for msg in messages:
            req = msg.payload
            cache_key = f'search|{req.pattern}|{req.file_path}'

            # Fast path: synchronous memory peek (no DB, no I/O)
            cached = self.cache.peek(cache_key)
            if cached is None:
                # Slower path: memory miss, check SQLite (local + peer-synced)
                cached = await self.cache.get(cache_key)

            if cached is not None:
                # Skip the subprocess entirely — hand the result back directly
                tasks.append(dk.ExecutorTask(
                    task_id=dk.make_task_id('search'),
                    source_offsets=[msg.offset],
                    metadata={'request_id': req.request_id},
                    labels={'source': 'cache'},
                    precomputed=dk.PrecomputedResult(stdout=cached),
                ))
                continue

            # Cache miss: run the subprocess
            tasks.append(dk.ExecutorTask(
                task_id=dk.make_task_id('search'),
                args=[req.pattern, req.file_path],
                source_offsets=[msg.offset],
                metadata={'request_id': req.request_id},
                labels={'source': 'subprocess'},
            ))
        return tasks

    async def on_task_complete(self, result):
        # Persist the result back into the cache for next time.
        # Only genuine subprocess runs need this — precomputed hits are
        # already in the cache (that's where they came from).
        if result.pid is not None:
            meta = result.task.metadata
            cache_key = f'search|{meta["pattern"]}|{meta["file_path"]}'
            self.cache.set(
                cache_key,
                result.stdout,
                ttl=3600,           # 1 hour TTL
                scope=CacheScope.LOCAL,
            )
        # ... return CollectResult with sink payloads as usual

And the matching config:

cache:
  enabled: true
  # db_dir: ""           # empty → falls back to ui.recorder.db_dir
  flush_interval_seconds: 3.0
  cleanup_interval_seconds: 60.0
  max_memory_entries: 10000     # LRU cap; omit for unbounded
  peer_sync:
    enabled: true
    interval_seconds: 30.0
    batch_size: 500
    timeout_seconds: 5.0
    # Upper bound on a single sync cycle. When None (default), derived as
    # interval_seconds * 0.9 so the next scheduled tick never overlaps the
    # current one. Set an explicit value when peers live on slow storage
    # (NFS, cross-region) and you want a tighter cap than the derived
    # default — but stay strictly below interval_seconds (config load
    # fails otherwise, so a misconfiguration surfaces at startup).
    cycle_deadline_seconds: null

cycle_deadline_seconds caps each cache.sync cycle. When the deadline fires, _sync_once tears down the in-flight peer reads, ticks the unlabelled cache_peer_sync_timeouts_total counter, and raises TimeoutError so the periodic-task wrapper classifies the run as status=error — dashboards keyed on drakkar_periodic_task_runs_total{status='error'} pick up deadline breaches through the same alerting path as other failures. The periodic loop keeps running (the wrapper’s default on-error behavior is to log + schedule the next tick), so a single deadline fire does not disable peer sync.

---

API reference

class Cache:
    def set(
        self,
        key: str,
        value: Any,
        *,
        ttl: float | None = None,
        scope: CacheScope = CacheScope.LOCAL,
    ) -> None: ...

    def peek(self, key: str) -> Any | None: ...
    def delete(self, key: str) -> bool: ...
    def __contains__(self, key: str) -> bool: ...

    async def get(
        self,
        key: str,
        *,
        as_type: type[T] | None = None,
    ) -> Any | None: ...

Thread-safety. The four sync operations (set / peek / delete / in) are safe to call from functions running under handler.offload() — the cache guards its memory state with an internal lock, so a pool thread and the event loop can use the same cache concurrently. The async get() stays loop-only (its SQLite fallback rides a connection bound to the main loop): await it for the keys you need before offloading, which warms them into memory, then peek inside the offloaded function.

set(key, value, *, ttl=None, scope=CacheScope.LOCAL)

Stores value under key in the in-memory dict and schedules a flush to SQLite on the next flush cycle. Synchronous — no I/O happens at call time.

value can be any JSON-serializable primitive (str, int, float, bool, None, list, dict) or a Pydantic BaseModel. Pydantic models are serialized via model_dump_json(). Anything else raises TypeError at set() time (deliberately — clearer error now than a silent failure on a peer that can’t parse the value later).

ttl is in seconds (fractional OK). None means “never expires” (the entry will remain until explicitly deleted or the worker’s cache DB is wiped).

scope controls how far the entry reaches — from process memory (MEMORY) to every peer (GLOBAL). See CacheScope below.

peek(key)

Memory-only lookup. Returns the decoded value if the key is in memory and unexpired; None otherwise. Never hits the DB — intended for arrange()’s hot path where you want zero I/O.

peek bumps the key to MRU position in the LRU order. Expired entries are opportunistically evicted.

peek() is NOT counted in hit/miss metrics

drakkar_cache_hits_total and drakkar_cache_misses_total are incremented by get() only. peek() is a memory probe — counting it would double-count if a caller does peek-then-get.

async get(key, *, as_type=None)

Async lookup with DB fallback. Lookup order:

1. Memory hit — key in _memory and unexpired → return without DB access. LRU-bump.
2. DB fallback — memory miss → SELECT on the reader connection, filtered by expires_at_ms IS NULL OR expires_at_ms > now.
3. Warm memory — DB hit → rebuild the entry in _memory at MRU position (may trigger LRU eviction if max_memory_entries is set).

as_type=SomeModel revives the JSON value via SomeModel.model_validate() for typed return. Without it, get returns the raw json.loads result (primitives / lists / dicts).

Increments drakkar_cache_hits_total{source="memory"} on memory hit, drakkar_cache_hits_total{source="db"} on DB hit, drakkar_cache_misses_total on miss.

delete(key)

Removes the key from memory and schedules a DB row deletion on next flush. Returns True if the key was present in memory, False otherwise.

Delete is local-only — use TTL for cross-worker invalidation

delete(key) removes the row from this worker’s cache DB. Peers that already synced the value keep their copy until it expires via TTL or is overwritten. Worse, if a peer’s copy has a newer updated_at_ms than any local state, the next sync cycle may re-pull the deleted value back (LWW will accept it).

For cross-worker invalidation, use TTL — set a short TTL when writing, or overwrite with a dummy/tombstone value. True delete propagation would require tombstone rows and GC, which is out of scope for v1.

__contains__(key)

key in self.cache — returns True if the key is in memory and unexpired. Does not bump LRU position (membership is a probe, not an access). Does not consult the DB.

CacheScope

class CacheScope(StrEnum):
    MEMORY = 'memory'     # this worker's memory only; never flushed to SQLite
    LOCAL = 'local'       # this worker only; peers never pull
    CLUSTER = 'cluster'   # same-cluster peers pull via sync
    GLOBAL = 'global'     # all peers (any cluster) pull via sync

The scopes form a reach ladder — each step widens where the entry can travel: process memory → this worker’s DB file → same-cluster peers → all peers.

For the persisted scopes, scope filters peer sync’s SELECT when another worker pulls from your cache DB:

• Same cluster peer pulls scope IN ('cluster', 'global')
• Different cluster peer pulls scope = 'global' only
• scope = 'local' never leaves this worker

MEMORY entries never reach the DB at all. A MEMORY set stores the value in memory and records a delete tombstone for the key instead of a write — so if an earlier wider-scoped set left a row on disk, the next flush purges it. This keeps the invariant “a MEMORY key never serves disk data”: without the tombstone, an LRU eviction or a worker restart would let get() resurrect the stale persisted value. The flip side: MEMORY entries are lost on eviction and restart — see the edge case below.

---

Choosing a scope

Scope	Use when	Example
MEMORY	Hot, cheap-to-recompute values that should never touch disk; also values too churn-heavy to be worth flush traffic	Per-worker negative-lookup markers; request-coalescing flags with sub-second TTL
LOCAL	Per-worker results that peers will compute independently (or that are worker-specific)	Hot-path subprocess memoization; per-worker rate limit counters
CLUSTER	Expensive results worth sharing across workers in the same deployment	Regex match results over a shared corpus; enrichment lookups
GLOBAL	Results universal across clusters (rare — usually signals “this is really a DB sink”)	Cross-deployment config distribution (unusual)

When unsure, start with LOCAL. The default is LOCAL specifically because it’s the safe choice — LOCAL entries stay put, peers never see them, and if you later decide to share, you flip the scope on new writes without touching consumers.

---

Cache vs. real DB sink vs. compute-every-time

Question	Choose…
“Do I need this value to survive a cluster-wide crash?”	DB sink (Postgres, Redis with persistence). Cache entries live in per-worker SQLite files with TTL retention — fine for optimization, not for durability-critical state.
“Is recomputation cheap (< subprocess launch cost)?”	Compute every time. A 50us Python function is not worth caching.
“Is recomputation expensive, and would it be OK to recompute on a cold cache?”	Cache. Cold start is bounded by whatever time it takes for reasonable cache population via normal traffic.
“Do I need strict consistency across workers?”	Not cache. Cache is eventually consistent (peer sync lags + LWW + no tombstones). Use Redis / Postgres / a distributed consensus system.
“Is the value small, JSON-serializable, and idempotent?”	Cache fits well.
“Does the value contain secrets?”	Not cache. Cache DBs are downloadable from the UI (same as recorder DBs). No redaction layer applies to cache values.

---

Edge cases and sharp edges

Cold cache on worker restart

When a worker restarts, its in-memory dict is empty. The DB file (<worker>-cache.db) still has all the entries, so get() will lazily warm memory on first access per key. There is no eager preload on startup — adding it would stall startup and race with normal traffic. This is deliberate (YAGNI: normal traffic warms the cache within one flush cycle).

The peer-sync cursor resets to 0 on restart, so the worker also re-pulls everything from peers once. Bounded by TTL cleanup and LWW rejection — no runaway work.

TTL vs. delete

TTL is the recommended invalidation mechanism. Set a TTL when writing; never call delete unless you mean “remove this from my own worker’s memory, and peers may still have it.”

Reasoning: peers who synced a value before a delete keep their copy. If any peer has updated_at_ms newer than your “deleted” state, the next sync may re-pull the value back into your DB via LWW. TTL naturally expires everywhere (every worker runs its own cleanup loop).

Peer unreachable

The sync loop iterates peers one at a time, each wrapped in try/except. A failed peer (connection refused, corrupt DB, missing file, read timeout) logs a warning, increments drakkar_cache_sync_errors_total{peer=...}, and the loop continues to the next peer. The worker keeps running. One bad peer cannot break the whole sync cycle.

The peer’s cursor is unaffected by other peers’ state — each peer has its own _peer_cursors[peer_name] entry. Recovery is automatic on the next sync tick.

Cross-cluster visibility

Peer sync uses the same worker-autodiscovery mechanism as the flight recorder (see Worker Autodiscovery). Workers are grouped by cluster_name (from DrakkarConfig.cluster_name or cluster_name_env).

A peer’s scope filter depends on cluster membership:

• Peer in same cluster as self → pulls scope IN ('cluster', 'global')
• Peer in different cluster → pulls scope = 'global' only

Unclustered workers (cluster_name='') are treated as a single implicit cluster — they pull cluster-scoped entries from other unclustered workers, and global-scoped entries from anyone.

LRU eviction

max_memory_entries=10_000 (default) → the in-memory dict is capped at 10,000 entries. Production-safe default — prevents unbounded growth under write-heavy workloads. Entries beyond the cap are evicted LRU, with pending dirty-ops preserved (see below).

max_memory_entries=None → the in-memory dict grows unbounded. Set this explicitly only when you have a bounded key universe or your own eviction strategy. The config layer emits a cache_max_memory_entries_unbounded warning at startup so the intentional choice is visible in logs — unlike the default, this knob needs operator awareness.

When the cap is set and adding the N+1-th entry would exceed it, the least-recently-used key is popped from the dict. Its pending dirty-op (if any) is preserved — the DB is the source of truth, and losing a dirty SET before flush would silently drop the write.

Evicted keys fall through to the DB on the next get(), which warms memory again (and may evict a different key in turn). The drakkar_cache_evictions_total counter tracks churn — a sustained high rate means the cap is undersized for the working set.

MEMORY entries are volatile (and invisible to the entries browser)

CacheScope.MEMORY trades durability for zero disk traffic. Three consequences, all by design:

• LRU eviction loses the entry. Persisted scopes fall through to the DB after an eviction and re-warm on the next get(); a MEMORY entry has no DB row, so once evicted the next read is a plain miss. Best-effort under memory pressure.
• A worker restart loses all MEMORY entries. Same reason — there is nothing on disk to warm from.
• The /debug/cache entries browser never lists them. The browser reads the cache DB file; MEMORY entries exist only in the process dict. They are still counted in the page’s in-memory header stats and in the entries_in_memory / bytes_in_memory gauges.

Re-scoping a key to MEMORY purges its disk row on the next flush (the tombstone described in CacheScope); re-scoping back to a persisted scope resumes normal flushing. Peers that synced a wider-scoped value before the re-scope keep their copy until its TTL expires — the same local-only limitation as delete.

Secrets in cache values

The recorder redacts secrets before writing worker_config (see the warning in Observability). The cache does not apply such redaction: cache values are written verbatim. Do not put secrets in the cache if the DB file is downloadable through the UI in your deployment.

Delete is local-only (the main sharp edge)

Repeated here because it’s the single behavior most likely to surprise. See delete() above for the full story.

Rule of thumb: for cross-worker invalidation, use TTL, not delete.

---

Consistency model

• MEMORY scope opts out of all of this. Memory-scoped entries are process-local: no durability window, no peer visibility, no LWW — they live and die with the worker process. Everything below concerns the persisted scopes.
• Eventually consistent across workers. A set() on worker A is visible on worker B at most ~(flush_interval + sync_interval) later (default: 3s + 30s = 33s).
• Worst-case staleness, concretely: local durability is bounded by flush_interval_seconds (default 3s); cross-worker freshness is bounded by flush_interval_seconds + peer_sync.interval_seconds (default 3s + 30s = 33s). Size intervals based on your tolerance — halving both to 1.5s/15s trades more SQLite + peer-disk traffic for tighter staleness. The cache never promises stricter consistency than those intervals; anything that needs real consistency should use a real DB sink, not this cache.
• LWW (last-write-wins) by updated_at_ms, with lexicographically smaller origin_worker_id as the tiebreaker for equal timestamps. Identical SQL (LWW_UPSERT_SQL) is used by local flushes and peer-sync merges, so local and cross-worker paths cannot diverge.
• No tombstones in v1. delete is local-only. Use TTL.
• Durability window = flush_interval_seconds (default 3s). A worker crash between set() and the next flush loses the unflushed writes. Tune shorter for stricter durability at the cost of more SQLite transactions.
• Wall-clock dependent. LWW compares updated_at_ms across workers, which assumes NTP-sync’d hosts. Wall-clock skew of a few seconds is fine; multi-minute skew would break LWW fairness.
• size_bytes is UTF-8 byte length. The in-memory bytes_in_memory gauge (and its DB counterpart bytes_in_db) sum the byte length of the JSON-encoded value (len(json_text.encode('utf-8'))) — not the Python-object size, not character count. Non-ASCII characters cost 2–4 bytes each. When sizing capacity from the gauge, budget against UTF-8 byte counts: a 100-char Cyrillic string costs 200 bytes, not 100.

---

UI walkthrough

When cache.enabled=true, the operator UI gains a Cache nav link that opens /debug/cache.

/debug/cache — entries browser

• Header stats: in-memory entries / bytes, DB entries / bytes
• Filter bar: scope dropdown, key search (case-sensitive substring match), “show only expired” toggle
• Paginated table of entries (limit default 200, max 1000): key, scope, size, age, TTL remaining, origin worker
• Side panel on click: decoded JSON value + full metadata (created_at, updated_at, expires_at, origin_worker_id)

The table lists the cache DB file, so MEMORY-scoped entries never appear in it — they show up only in the in-memory header stats. See the edge case above.

/debug/periodic — system periodic tasks

The existing periodic tasks page now shows cache.flush, cache.sync, and cache.cleanup alongside user-defined @periodic methods, each with a small [system] pill indicating framework origin. Status, last run, duration, success/error counts, and sparkline of recent runs are rendered identically to user periodic tasks — the [system] pill is the only visual difference.

Prometheus metrics page

All cache metrics appear on the /debug metrics viewer alongside the rest of the framework metrics.

Message Probe tab — cache is read-only

The Message Probe tab on /debug runs a pasted message through the full handler pipeline without touching production state. In this mode all cache writes are silently suppressed (set, delete) — the probe’s DebugCacheProxy records every call but never mutates the live cache. Reads are opt-in via the Use cache (read-only) checkbox: when enabled, get / peek / __contains__ forward to the real cache and show accurate hit/miss outcomes in the call log; when disabled, all reads return a miss even for keys that exist. Every call (including suppressed writes) is captured with op, key, scope, outcome, and the originating hook stage, so the probe report lets you audit what the handler asked of the cache without disturbing its state.

---

Related pages

• Handler — the handler class surface that exposes self.cache
• PrecomputedResult — how cache hits short-circuit tasks
• Observability — Prometheus metrics, operator UI details
• Performance — when caching helps, the PrecomputedResult fast path
• Worker Autodiscovery — the shared-db_dir convention that peer sync reuses
• Integration Tests — the docker-compose demo exercises cache fast-path, peer sync, and disk persistence