Cheetah / Cheetah documentation

Run and schedule the view engine

ViewEngine is application-side machinery around an IHistoryStore and an IViewStore. It is not a server node and does not receive network traffic. The product constructs it, registers a bounded set of reducers, and calls it from a deliberately owned processing loop.

Construct and register a view

The default factory uses an in-memory view store. Name a store explicitly when the snapshot must survive a process restart.

from servercheetah.views import InMemoryViewStore, create_view_engine
from servercheetah.views.builtins import LatestPerKeyReducer

user_id = "usr_01234567-89ab-4def-8123-456789abcdef"
view_id = f"{user_id}/latest-device-status"

engine = create_view_engine(
    history=components.history,
    view_store=InMemoryViewStore(),
)

engine.register(
    user_id=user_id,
    view_id=view_id,
    reducer=LatestPerKeyReducer(
        source_streams=["device-status"],
        key_field="device_id",
    ),
)

Reducer source streams are suffixes. This registration reads usr_01234567-89ab-4def-8123-456789abcdef/device-status; it cannot use a suffix to escape into another user's stream.

view_id is the key in one engine-wide registration map, not a user-relative key. Make it unique across every user registered in that engine. A second registration of the same ID raises ValueError, including when it names another user. Unregister the old definition before installing a replacement.

The engine retains one process-local lock for every distinct view ID it has seen, including after unregistration. Frozen v1 therefore assumes a bounded application-defined registration set; do not create unbounded ephemeral IDs.

Advance and read the view

processed = await engine.process(view_id, batch_size=100)
state = await engine.get_state(view_id)

One process() call visits every declared source stream once and requests at most batch_size messages from each. It returns the number of history messages whose cursors were consumed. That count includes messages skipped by the reducer error policy; it is not the number of state changes.

The engine saves only when at least one message was consumed. A newly registered view over empty history therefore still has no snapshot, and get_state() returns None. A deliberate rebuild() over empty history is different: it publishes the reducer's initial state as the complete replacement.

get_state() also returns None when the view is not currently registered or has no saved snapshot. With a conforming view store, the returned dictionary is detached from persisted state; mutating it does not update the view.

Both batch_size and the rebuild limits must be positive. A small page limits one processing cycle's work but requires more cycles to catch up.

Put the clock in an owned host task

Registration starts no task. A simple service can supervise calls explicitly:

import asyncio
import logging

log = logging.getLogger(__name__)

async def advance_views(engine, view_ids, stop_event):
    while not stop_event.is_set():
        for current_view_id in view_ids:
            try:
                await engine.process(current_view_id, batch_size=100)
            except Exception:
                log.exception("derived-view processing failed: %s", current_view_id)
        await asyncio.sleep(2)

Calling process() per view makes failure isolation visible. process_all() is a convenience that visits registrations sequentially and returns {view_id: processed_count}. If one view raises, later registrations are not processed during that call.

A production owner should also expose last success, last failure, processing duration, consumed count, snapshot age, and backlog or lag evidence appropriate to its history backend. Snapshot updated_at_ms only says when a non-empty batch was saved; it is not proof that the scheduler is healthy or the view is caught up.

Same-process calls are serialized

process() and rebuild() for one view share a lock inside one ViewEngine. Two same-process calls cannot publish that view concurrently. Different views may be scheduled independently when the host chooses to do so.

Another process has another lock. If two processes operate the same (user_id, view_id), both can load an old checkpoint and the later save can overwrite the other result. Place each view in one designated processor or provide application-owned cross-process election and fencing.

Callbacks report successful reduction, not freshness

register() accepts a synchronous on_change(view_id, state) callback. It runs after the snapshot save when at least one reducer call succeeded. A successful no-op reducer call still notifies; an all-skipped batch does not. Cheetah does not compare the old and new dictionaries.

A callback exception is logged and does not reverse the committed snapshot. Use the callback as a notification hint, not as a second atomic write or an external side effect that must occur exactly once.

Restart and remove deliberately

Registrations and reducer objects are process-local. After restart, recreate the same registration before process(), get_state(), or rebuild() can use the saved snapshot.

unregister(view_id) removes only that process-local definition. It does not delete the stored snapshot, cancel another process, or remove retained source messages. Use the view store's user-scoped delete() operation as a separate, explicitly authorized lifecycle action.

Next: Define reducers.