Maintain a Local Cache with Change Drains

A local cache, a search index, a dashboard, an agent’s working memory: each needs to know what changed in a repository since it last looked. Re-reading everything works and costs a full scan every time. A change drain returns only what changed, plus a checkpoint to resume from next time.

Every repository has a monotonic repository sequence. A drain takes the sequence you stopped at and returns changed current-state records, together with a checkpoint. These helpers do not capture every writer’s changes: treat their deltas as best-effort. When you need a complete copy, periodically reconcile against a full read, as described in incremental reads.

The two drains

from warmhub import WarmHubClient

with WarmHubClient.from_env() as client:
    sensors = client.repository("acme/sensors")

    changes = sensors.things.head_changes(since_repo_seq=-1)
    changes.items     # changed records returned by this drain
    changes.repo_seq  # the checkpoint to pass next time

• head_changes is the drain form of head. With no filters it drains the whole repository, including shape definitions.
• query_changes is the drain form of query. It adds the filters only query has — about, resolve_collections, role — on top of shape, kind, and where.

Both are on the repository handle’s things, and on client.thing with the organization and repository names passed explicitly.

How they differ from head and query:

• since_repo_seq is required and keyword-only. It is a strict lower bound: the drain returns changes after that sequence. -1 asks for a full snapshot, which is what a cold cache wants.
• The drain pages for you. It follows the cursor to the last page before returning. There is no cursor= argument; limit= sets only the page size.
• The result is a pair. ThingChanges carries items and repo_seq. The checkpoint arrives with the last page; if it is missing, the SDK raises VALIDATION_ERROR instead of returning items you cannot resume from.
• Retractions are always included. A retracted record comes back with active=False. A drain that hid retractions would leave deleted records in your cache forever.
• One row per changed record. You get each record’s current state, not a log of writes. Two writes to the same record between drains give one row.

A cache updated by deltas

This example caches Things that directly declare the local Reading shape, including declarations pinned to a version such as Reading@v1. It does not include records that only inherit Reading through shape composition.

Drain all Things and decide membership locally: a shape-filtered drain sees current matching versions, so an active record that leaves Reading produces no row and would remain stale in the cache.

from warmhub import WarmHubClient, normalize_wref

client = WarmHubClient.from_env()
sensors = client.repository("acme/sensors")

cache: dict[str, object] = {}
state = {"checkpoint": -1}


def sync() -> None:
    changes = sensors.things.query_changes(
        since_repo_seq=state["checkpoint"], kind="thing"
    )
    for item in changes.items:
        durable_id = item.metadata["durableId"]
        is_reading = any(
            normalize_wref(ref) == "Reading"
            for ref in item.declared_shapes or ()
        )
        if item.active and is_reading:
            cache[durable_id] = item
        else:
            cache.pop(durable_id, None)
    state["checkpoint"] = changes.repo_seq

Call sync() on a schedule, or whenever you are told something changed. Four details keep the delta application consistent:

• Key by durable id, not by name. item.metadata["durableId"] is fixed for the life of a record. Its name is not: after a rename the drain returns the record under its new name, and a cache keyed by name would hold it twice.
• Store only active records that directly declare Reading. Drop both retractions and active records that leave that shape.
• Advance the checkpoint after applying the items. If the process stops between the two, the next run re-reads the same changes, and applying a change twice to a dictionary is harmless. Store the checkpoint wherever you store the cache, so the two cannot drift apart.
• Keep the drain scope consistent. This cache always drains all Things. Changing the scope or the local membership rule requires rebuilding from -1; the old checkpoint cannot populate records skipped by the old rule.

Here is what each sync sees as the repository changes:

Write	Rows returned	Cache afterwards
(cold start, empty repository)	0	—
add probe-0, probe-1, probe-2	3	probe-0, probe-1, probe-2
revise probe-0, retract probe-1	2	probe-0, probe-2
rename probe-2 to probe-2-north	1	probe-0, probe-2-north
revise active probe-0 to be shapeless	1	probe-2-north
revise probe-0 to directly declare Reading again	1	probe-0, probe-2-north
revise probe-0 to the data it already holds	0	probe-0, probe-2-north

The last row is worth noticing. A write whose data matches what is stored is a noop: it allocates no version and does not move the repository sequence, so it puts nothing in anyone’s drain. Pipelines that rewrite records from source on every run do not flood their readers.

What a lost checkpoint costs

A drain from -1 requests a full current-state snapshot. A stored checkpoint lets you request later deltas. If you lose the checkpoint, rebuild the cache from a new snapshot rather than guessing a sequence.

A checkpoint does not establish completeness: the helpers do not capture every writer’s changes. Periodically read the full current state and replace the cache using the same active/direct-shape membership rule. Replacement also removes stale entries absent from the full read; merely overlaying rows would retain them. The delta loop above does not implement this reconciliation.

When a drain is refused

• INCREMENTAL_READ_UNAVAILABLE. Some filters cannot be read incrementally — a glob match=, resolve_collections=, and a cross-repository about — and incremental reads are not yet available on every repository. The error does not say which applies, so rule out the filters first, then fall back to a full read.
• FORBIDDEN. A drain needs read access to the whole repository. A token restricted to a subset of records is refused, even if the subset covers everything the drain asks for.

Filtering: incremental reads covers these refusals in full.

Be told instead of polling

A drain answers “what changed since I last asked?”. To be notified when a write lands, create a subscription and run sync() when it fires, or run an action runner. For this cache, notifications must also cover writes that remove Reading: a shape-filtered subscription alone is not enough. Poll periodically if your notification scope does not cover every change the drain needs.