Storage¶

Storage¶

SurrealDB-backed storage for events, runs, asset records, and key-value data.

Storage.memory()¶

Create an in-memory storage instance (useful for tests).

storage = rs.Storage.memory()

Storage.embedded(path)¶

Create an embedded storage backed by RocksDB at the given path.

storage = rs.Storage.embedded(".rivers/storage")

Storage.connect(endpoint, *, username=None, password=None, namespace=None, database=None)¶

Connect to a remote SurrealDB endpoint (e.g. ws://host:8000). Used by the K8s serve / execute CLI commands.

storage = rs.Storage.connect("ws://surrealdb.rivers.svc.cluster.local:8000")

Each parameter resolves via: explicit kwarg → RIVERS_SURREAL_* env var → default. So in Kubernetes, where Helm/operator inject the env vars from a Secret, callers can simply Storage.connect(endpoint).

When username and password both resolve to non-empty values, they authenticate as a database-scoped user against namespace / database — matching a DEFINE USER ... ON DATABASE definition.

Schema versioning & migration¶

Each persistent database carries a schema stamp — the schema version it was last migrated to, plus a min_reader and min_writer floor (the oldest rivers build allowed to read, and to write, that version). Opening a store checks this stamp against the build's schema version and refuses an incompatible open instead of silently migrating.

The open contract (floor ≤ build ≤ version):

The reader/writer split lets a write-breaking migration for newer writers run without locking out an older read-only UI: a Read open is gated by min_reader, a read/write open by min_writer (and min_writer ≥ min_reader always holds).

An uninitialized store (no stamp) is bootstrapped by whichever process opens it first — the UI included — so a fresh deployment shows an empty UI without waiting for a code location.

rivers db migrate¶

Applies pending schema migrations, bringing the database to the running build's schema version. Idempotent (a no-op when already current), and serialized across processes by a short-lived lease so two openers can't migrate the same store at once. Run it after upgrading rivers when a code location or the UI reports that the database needs migration.

rivers db migrate                              # embedded (--storage-path, default .rivers/storage/)
rivers db migrate --surreal-endpoint ws://surrealdb:8000   # remote

In K8s, run it as an explicit init/job step before rolling out upgraded code locations. rivers dev instead offers to migrate interactively when it finds the database behind the build.

Storage.migrate_embedded(path) / Storage.migrate_remote(endpoint, ...)¶

The programmatic form of rivers db migrate — open-migrate-close. migrate_remote takes the same username / password / namespace / database resolution as Storage.connect.

rs.Storage.migrate_embedded(".rivers/storage")
rs.Storage.migrate_remote("ws://surrealdb.rivers.svc.cluster.local:8000")

storage.type¶

Returns a StorageType enum (StorageType.Memory, StorageType.Embedded, or StorageType.Remote).

Sync Methods¶

All query methods block the calling thread until the result is ready.

Dynamic keys registered before the reserved-character guard existed (|/,) still classify, but cannot round-trip display-string paths (UI, gRPC). The storage migration logs a warning for all such keys and records them under the reserved_dynamic_keys entry in the kv table. Delete each offending key and re-register it under a clean name.

compute_staleness()¶

def compute_staleness(self) -> dict[str, tuple[str, list[StaleCause]]]

Compute current staleness for every asset, keyed by asset_key. Each entry is (status, causes) where status is one of "UpToDate", "Stale", or "Missing". Staleness is no longer persisted on AssetRecord — call this to get the live result.

Async Methods¶

Every sync method has an async_ prefixed counterpart that returns a Python awaitable. These use pyo3_async_runtimes::tokio::future_into_py under the hood — the Rust future runs on Tokio without holding the GIL, making them safe to use in async Python code.

import asyncio
import rivers as rs

async def main():
    storage = rs.Storage.memory()
    await storage.async_kv_set("key", b"value")
    result = await storage.async_kv_get("key")
    print(result)  # b"value"

asyncio.run(main())

Data Classes¶

StoredEvent¶

StoredLog¶

One step execution's captured output. Logs live in the dedicated run_logs table (not the events stream); streams the step didn't produce are None.

A failed attempt writes its Python traceback in a row of its own, with the time of the attempt's StepFailure or StepRetry event. traceback is JSON: exceptions is the exception chain, oldest first, and each frame keeps the source lines around it; text is the traceback as Python prints it. The process that ran the step captures it, so the source lines come from that machine. The run page in the UI shows it.

AssetRecord¶

StaleCause¶

RunRecord¶

config holds the launch document the run was launched with; None means the definitions as they are.

LaunchedBy¶

Tagged union describing where a run originated. Build via classmethod factories and discriminate on .kind.

rs.LaunchedBy.manual()
rs.LaunchedBy.schedule("nightly")
rs.LaunchedBy.sensor("file_sensor")
rs.LaunchedBy.backfill("bf-2024-q1")
rs.LaunchedBy.condition()

The user is populated by the authenticated web UI, which stamps the session identity onto the launch and carries it end-to-end onto the run's (and backfill's) launched_by — rs.LaunchedBy.manual() from Python has no user. See the authentication guide.

UserRef¶

Identity of the user who triggered a manual launch — the stable subject plus display snapshots.

StoredTick¶

PoolLimit¶

Configuration for a concurrency pool.

PoolInfo¶

Runtime info for a concurrency pool (config + current usage).

PoolBlockDetail¶

Detail for a single pool that blocked a concurrency claim.

BlockReason¶

Why a step was blocked from claiming concurrency slots.

ConcurrencyClaimStatus¶

Result of attempting to claim concurrency slots.

SlotHolder¶

A run/step currently holding pool slots — surfaced by get_pool_slot_holders().

Pool Methods¶

Run-queue Methods¶

The claim/release methods (_claim_concurrency_slots, _free_concurrency_slots, _free_concurrency_slots_for_run) and the lease management methods (_renew_slot_lease, _free_expired_leases) are internal — called by the executor during step scheduling and background lease management, not by user code.

Executor Integration¶

Pool limits are enforced during execution. When a step with pool configuration runs, the executor:

1. Claims slots before execution — blocks (with backoff polling) until all required pools have capacity.
2. Starts lease renewal — a background task renews the lease every lease_duration / 3 seconds.
3. Executes the step — the Python function runs while slots are held.
4. Releases slots on completion or failure — slots are freed immediately.
5. Run-level cleanup — after all steps finish, free_concurrency_slots_for_run removes any leaked slots.

Per-backend behavior:

• InProcess: Sequential claim → execute → release per step.
• Async: Each concurrent task claims independently inside its JoinSet task.
• Parallel: Pool-requiring steps run in loky subprocesses with claim-gated concurrency via a tokio JoinSet — each task does claim → submit+collect → release so the pool limit naturally throttles how many steps execute simultaneously. Non-pool steps use the existing batch pipeline.

Mapped (fan-out) steps claim per instance, not per parent step.

Lease Expiry¶

Concurrency slots use lease-based expiry. Each claimed slot has a lease_expires_at timestamp. Expired leases are automatically excluded from capacity checks, so a crashed process's slots are reclaimed without manual intervention. The _renew_slot_lease method extends the lease during execution, and _free_expired_leases performs garbage collection of stale rows. The coordinator daemon calls free_expired_leases on every tick as defense-in-depth GC.