Cheetah / Cheetah documentation

Define reducers

An IViewReducer defines one deterministic state transition. It declares a stable name and one or more history-stream suffixes, produces an initial dictionary, and applies one retained message at a time.

from servercheetah.views import IViewReducer

class SuccessCounter(IViewReducer):
    @property
    def name(self) -> str:
        return "success-counter-v1"

    @property
    def source_streams(self) -> list[str]:
        return ["jobs"]

    def initial_state(self) -> dict:
        return {"completed": 0}

    def apply(self, event: dict, state: dict) -> dict:
        if event.get("kind") != "result":
            return state
        return {**state, "completed": state.get("completed", 0) + 1}

apply() is synchronous and runs once per consumed history message. Keep it fast, deterministic, and free of I/O or external effects. It must not send mail, charge an account, issue another command, or update unrelated storage: a message can be replayed before a checkpoint is saved.

Treat the input event as read-only. A reducer may mutate the state dictionary or return a new one, but returned state must remain deep-copyable. It must also be JSON-serializable when the Redis store may be used. State size is the application's responsibility; frozen v1 enforces no byte, key-count, or reduction-time quota.

Five supplied state shapes

All five reject an empty source_streams list. Bounded reducers reject zero or negative limits. PartialUpdateMergerReducer accepts only replace, deep_merge, or concatenate. Misconfiguration raises ValueError during construction rather than becoming silently unbounded.

Collect a bounded result list

from servercheetah.views.builtins import ListCollectorReducer

reducer = ListCollectorReducer(
    source_streams=["scrape-results"],
    max_items=500,
)

The state contains items, count, error_count, and reserved _seen_ids. items retains the newest max_items result payloads. count is the current retained list length, while error_count is cumulative within the replay behavior described below.

Keep the latest payload per key

from servercheetah.views.builtins import LatestPerKeyReducer

reducer = LatestPerKeyReducer(
    source_streams=["device-status"],
    key_field="device_id",
)

For every result whose payload contains key_field, the reducer stores entries[str(key)] = {"payload": ..., "timestamp_ms": ...}. It replaces an entry only when the new event timestamp is strictly greater. Equal timestamps keep the existing value. The state also contains last_updated_ms.

This reducer has no entry-count bound. Choose it only when the key population is already bounded or add a product reducer with an explicit eviction policy.

Collect screenshot evidence

ScreenshotCollectorReducer(source_streams, max_items=500) retains screenshot records keyed by their reported client and browser context information. It recognizes tab_id or context_id, dataUrl or data_url, screenshot_b64, and url in result payloads.

The screenshot list is bounded by max_items; the detailed error list retains the newest 100 records. error_count remains cumulative apart from duplicate handling, so the snapshot can grow in meaning even when detailed records have rolled out.

Aggregate protocol errors

ErrorAggregatorReducer(source_streams, max_recent=200, max_clients_per_code=50) reads canonical Cheetah error details from the top-level error object:

{
  "kind": "error",
  "error": {
    "code": "tab_not_found",
    "message": "Tab does not exist"
  }
}

It groups counts by code, records last time and message, keeps a bounded affected-client list, and retains bounded recent errors. Direct legacy events that put the error dictionary in payload remain accepted as a compatibility fallback; that fallback is not an alternative wire format.

Merge partial keyed updates

from servercheetah.views.builtins import PartialUpdateMergerReducer

reducer = PartialUpdateMergerReducer(
    source_streams=["inventory-updates"],
    key_field="device_id",
    strategy="deep_merge",
    max_entries=5_000,
)

The three strategies operate on each key's data dictionary:

When the entry bound is exceeded, the reducer removes entries with the oldest last_updated_ms. The state also exposes total_updates, last_updated_ms, per-entry update_count, and reserved _seen_ids.

Duplicate suppression is deliberately bounded

Four collection reducers keep recent message IDs in _seen_ids. The remembered window is derived from max_items, max_recent, or max_entries; it prevents ordinary recent replay but eventually evicts old IDs. Replaying an event outside that window can collect, merge, or count it again.

Do not call those reducers permanently idempotent and do not remove _seen_ids from a saved state. The field is visible reserved bookkeeping and uses linear membership checks. If an application needs all-history deduplication, exact counters, or a different performance bound, enforce that identity contract before history ingestion or implement a product reducer and durable index.

LatestPerKeyReducer is different: timestamp comparison makes an equal or older replay harmless for the retained entry. It still depends on trustworthy timestamps and does not resolve ties.

Next: Understand checkpoints and failures.