Persistence¶

Cosalette's persistence system lets devices save state across restarts — accumulated values, calibration offsets, last-known-good readings, or anything that shouldn't be lost on power failure.

The system has three layers:

Store Backends¶

A Store is a key-value storage backend. The framework ships four:

The Store protocol is simple:

class Store(Protocol):
    def load(self, key: str) -> dict[str, object] | None: ...
    def save(self, key: str, data: dict[str, object]) -> None: ...

You can implement your own backend (Redis, S3, etc.) by satisfying this protocol.

JsonFileStore¶

Stores all keys as top-level entries in a single JSON file. Uses atomic writes (write to temp file, then rename) to prevent corruption.

store = JsonFileStore("./data/state.json")
# All device keys stored in one file: {"sensor": {...}, "counter": {...}}

SqliteStore¶

Stores all keys in a single SQLite database with WAL mode enabled for concurrent read access.

store = SqliteStore("./data/state.db")

Store Factories¶

When the store path depends on runtime settings, pass a callable factory instead of a concrete instance:

def make_store(settings: Gas2MqttSettings) -> Store:
    return JsonFileStore(settings.data_dir / "state.json")

app = cosalette.App(
    name="gas2mqtt",
    version="1.0.0",
    settings_class=Gas2MqttSettings,
    store=make_store,
)

The factory is called during bootstrap — after settings and adapters are resolved but before any device handlers run. Parameters are injected via the DI system (every parameter must carry a type annotation), so the factory can request settings, adapters, or both:

def make_store(settings: Gas2MqttSettings) -> Store:
    return SqliteStore(settings.db_path)

app = cosalette.App(name="gas2mqtt", store=make_store)

Default Store Resolution¶

When store= is omitted from App(...), the framework automatically creates a JsonFileStore at a path derived from the app name. Path precedence:

1. <NAME>_STORE_PATH environment variable — name upper-cased, all non-alphanumeric characters replaced by underscores (e.g. CALDATES2MQTT_STORE_PATH, SENSOR_HUB_STORE_PATH).
2. $XDG_STATE_HOME/<name>/store.json.
3. ~/.local/state/<name>/store.json (the XDG default).

This means retained-topic cleanup (ADR-048) works with zero configuration — apps that remove configured entities will have orphaned retained topics cleared on the next restart without any store wiring.

import cosalette

# Zero-config: JsonFileStore auto-resolved from the app name
# Path: MYAPP_STORE_PATH → $XDG_STATE_HOME/myapp/store.json → ~/.local/state/myapp/store.json
app = cosalette.App(name="myapp", version="1.0.0")

To opt out of persistence entirely, pass store=None:

# Explicit opt-out — no store, no retained-topic cleanup
app = cosalette.App(name="myapp", version="1.0.0", store=None)

Explicit Store instances and factory callables always take precedence over the default resolution:

# Explicit path — default resolution does not apply
app = cosalette.App(
    name="myapp",
    store=cosalette.JsonFileStore("/app/data/state.json"),
)

Configurable default backend¶

By default the auto-resolved store uses JsonFileStore. For high-write apps, call cosalette.set_default_store_backend() once at startup (before any App() is constructed) to swap the backend:

import cosalette
from cosalette import SqliteStore

# All App() instances that omit store= will now resolve a SqliteStore
cosalette.set_default_store_backend(SqliteStore)

app = cosalette.App(name="myapp", version="1.0.0")

Passing None resets to the JsonFileStore default. Explicit store= arguments on App() are always unaffected.

Container deployments¶

Inside a container the default XDG path is ephemeral. When the framework detects a container runtime (/.dockerenv, /run/.containerenv, or the container env var) and no <NAME>_STORE_PATH is set and the app's entity set may vary by config across restarts, it logs a WARNING at startup:

WARNING  Using an auto-resolved default store at <path>, which is ephemeral
         inside a container - retained-topic cleanup (ADR-048) will not survive
         restarts. Set <NAME>_STORE_PATH to a path on a mounted volume for
         durable persistence.

For durable persistence across restarts, set <NAME>_STORE_PATH to a path on a mounted volume (e.g. MYAPP_STORE_PATH=/app/data/store.json). See the Deployment guide for details.

Explicit opt-out: retained_cleanup¶

If your app uses @app.on_configure purely for non-entity-varying reasons (e.g. config validation only), the conservative heuristic will still warn. Use retained_cleanup=False to opt out explicitly — the store is kept for persist= device state, but ADR-048 cleanup and the ephemeral-store warning are both disabled:

app = cosalette.App(
    name="myapp",
    version="1.0.0",
    retained_cleanup=False,  # skip cleanup + warning; store kept for persist=
)

Pass retained_cleanup=True to force cleanup on for an app whose entity names are derived from import-time config (not callable name= specs) — a case the structural heuristic would otherwise miss.

App.retained_cleanup is a read-only property returning the override value (True, False, or None for auto). See ADR-049 for full rationale.

See ADR-049 — Default store path resolution for the design rationale and alternatives considered.

Optional: signed snapshots with retained_cleanup_snapshot_key¶

The ADR-048 snapshot has no integrity protection by default — anything with write access to the configured Store backend (a shared SQLite file, a network-backed store, etc.) can silently rewrite it. Pass a retained_cleanup_snapshot_key to have the snapshot HMAC-SHA256 signed on save and verified (timing-safe) on load:

import cosalette
import os

from pydantic import SecretStr

app = cosalette.App(
    name="myapp",
    version="1.0.0",
    retained_cleanup_snapshot_key=SecretStr(os.environ["CLEANUP_SNAPSHOT_KEY"]),
)

Source the key the same way you would MqttSettings.password (env var, secrets manager, etc.) — never via the same Store this feature protects, or an attacker who can rewrite the snapshot could rewrite the key too. A missing, unrecognized-algorithm, or mismatched signature — including a snapshot written before a key was configured — is treated as no previous snapshot: cleanup is skipped for that run and a freshly signed snapshot is written. None (the default) preserves today's unsigned behavior exactly.

This only adds tamper detection against an untrusted-but-writable Store; it does not change the module's single-writer assumption — see ADR-063 for the full threat model and rationale.

DeviceStore¶

DeviceStore is a per-device scoped wrapper around a Store backend. It provides a familiar dict-like interface:

@app.telemetry("sensor", interval=60)
async def sensor(store: DeviceStore) -> dict[str, object]:
    # Dict-like access
    store["count"] = store.get("count", 0) + 1
    store.setdefault("offset", 0.0)

    # Check what's stored
    all_data = store.to_dict()

    return {"count": store["count"]}

The framework automatically:

1. Creates a DeviceStore scoped to the device name
2. Loads existing data before the first handler call
3. Injects it via the DI system (declare store: DeviceStore)
4. Saves on shutdown (safety net, regardless of policy)

Dirty Tracking¶

DeviceStore tracks whether it has been modified since the last save. This enables the SaveOnChange policy to avoid unnecessary I/O:

store["value"] = 42      # store.dirty → True
store.save()             # store.dirty → False
store.mark_dirty()       # Force dirty (e.g., after mutating a nested object)

Save Policies (PersistPolicy)¶

A PersistPolicy controls when the store is saved during the telemetry loop. Three policies ship with the framework:

SaveOnPublish¶

Save after each successful MQTT publish. The most common choice — persisted state always matches what's been broadcast.

@app.telemetry("sensor", interval=60, persist=SaveOnPublish())
async def sensor(store: DeviceStore) -> dict[str, object]:
    store["count"] = store.get("count", 0) + 1
    return {"count": store["count"]}

SaveOnChange¶

Save whenever the store has been modified, regardless of whether MQTT publishing occurred. Most aggressive — minimises data loss.

@app.telemetry("sensor", interval=60, persist=SaveOnChange())
async def sensor(store: DeviceStore) -> dict[str, object]:
    store["count"] = store.get("count", 0) + 1
    return {"count": store["count"]}

SaveOnShutdown¶

Save only on graceful shutdown. Lightest I/O — no saves during normal operation. Risk: data loss on hard crash or power loss.

@app.telemetry("sensor", interval=60, persist=SaveOnShutdown())

Default Behaviour¶

If you set store= on the App but don't specify persist= on a device, the framework saves only on shutdown (equivalent to SaveOnShutdown()).

The framework always saves on shutdown regardless of policy — the persist= parameter only controls additional saves during operation.

Composing Policies¶

Policies compose with | (OR) and & (AND), just like publish strategies:

# Save on publish OR when dirty (maximum safety)
persist = SaveOnPublish() | SaveOnChange()

# Save only when BOTH conditions are true
persist = SaveOnPublish() & SaveOnChange()

| creates an AnySavePolicy (save if any child says yes). & creates an AllSavePolicy (save only if all children agree).

When to Use Which Policy¶

Testing with MemoryStore¶

Use MemoryStore in tests to avoid filesystem access:

from cosalette import MemoryStore, DeviceStore
from cosalette.testing import AppHarness

async def test_sensor_persists_count():
    backend = MemoryStore()
    harness = AppHarness.create(store=backend)

    @harness.app.telemetry("sensor", interval=10)
    async def sensor(store: DeviceStore) -> dict[str, object]:
        store["count"] = store.get("count", 0) + 1
        return {"count": store["count"]}

    await harness.run()
    assert backend.load("sensor") == {"count": 1}

You can also pre-seed the store to test load behaviour:

backend = MemoryStore()
backend.save("sensor", {"count": 99})

# Handler will see store["count"] == 99 on first call

Persistence and Device Handlers¶

The persist= parameter is only available on @app.telemetry, because the framework controls the telemetry loop and knows when publishes occur.

For @app.device handlers (which own their loop), inject DeviceStore and call store.save() manually when appropriate:

@app.device("controller")
async def controller(ctx: DeviceContext, store: DeviceStore):
    while not ctx.shutdown_requested:
        # ... do work ...
        store["last_run"] = ctx.clock.now()
        store.save()  # Manual save
        yield  # reaction boundary
        await ctx.sleep(60)

The framework still saves on shutdown via the finally block.

See Also¶

• Publish Strategies — the publish= parameter that persist= mirrors
• Signal Filters — another composable utility
• Testing Guide — testing with MemoryStore
• ADR-015: Persistence — architectural decision record
• ADR-037: Lazy Store Resolution — callable store factories