Persist and recover runs

One port over an append-only log with conditional writes. Every adapter needs the same three primitives: append at an expected sequence, claim under a condition, and read a range....

One port over an append-only log with conditional writes. Every adapter needs the same three primitives: append at an expected sequence, claim under a condition, and read a range. That is what lets one persistence model serve Postgres, MySQL and DynamoDB with no second code path.

Four adapters, all shipped

from psych_runtime.store.memory import InMemoryStore
from psych_runtime.store.postgres import PostgresStore
from psych_runtime.store.mysql import MySQLStore
from psych_runtime.store.dynamodb import DynamoDBStore

store = MemoryStore()  # tests, scripts, single process

store = PostgresStore(dsn="postgresql://localhost/psych")  # or pool=<asyncpg.Pool>
await store.migrate()

store = MySQLStore(dsn="mysql://...")  # or pool=<aiomysql.Pool>
await store.migrate()

store = DynamoDBStore(
    endpoint_url=None,  # set for DynamoDB Local
    region_name="us-east-1",
    table_prefix="psych",
)

PostgresStore takes exactly one of dsn or pool. Passing your own pool is the right move in an application that already has one.

Install the extra you need: psych-runtime[postgres], psych-runtime[mysql], psych-runtime[dynamodb], or psych-runtime[all].

MemoryStore is real, and keeps nothing

It is a genuine Store implementation, not a stub, guarded by one lock. It is correct and it is entirely in-process: nothing survives the process. Right for a first run, a test and a script. Wrong for anything whose Runs must outlive it.

Every adapter implements the port structurally. There is no base class to inherit, because the port is a Protocol.

The queue is the store

A runnable Run is one whose lease is unheld or expired. claim() is the queue. There is no broker, no separate queue table to keep in step, and nothing to reconcile when the two disagree, because there is no two.

worker = psych_runtime.Worker(store, runtime, concurrency=8, lease_seconds=30.0)
await worker.run()

Scale by running more Workers, in more processes, on more machines. They all claim from the same store.

Conditional writes only

No transactions, no joins, no SELECT ... FOR UPDATE. That constraint is what makes DynamoDB a first-class target rather than a compromise, and it is why a SeqConflict is a normal outcome rather than an error: two writers raced for one sequence number and one lost. The loser re-reads and retries.

Migrations

Sequential and forward-only, under psych_runtime/store/migrations/ (Postgres) and psych_runtime/store/migrations_mysql/. await store.migrate() applies them. DynamoDB creates its tables instead.

Adding one: check the highest existing number first, and never edit a migration that has shipped.

Retention is yours

Psych does not expire, garbage-collect or size-cap what accumulates. That lifecycle policy is yours, the same as the database itself. Psych uses your database through this port and never provisions or migrates it beyond its own tables.

Adding an adapter

The contract suite is the deliverable as much as the adapter. Divergent implementations with no shared test are divergent behaviours waiting to be discovered in production.

# tests/functional/test_store_mystore.py
from psych_runtime.store.contract import StoreContractSuite


class TestMyStore(StoreContractSuite):
    @pytest.fixture
    async def store(self):
        yield MyStore(...)

Nothing else in the suite changes. That is what keeps every implementation honest against one definition of correct rather than an independent reading of the design per adapter.

No mocked store. Store tests run against real Postgres, real MySQL and real DynamoDB Local. The in-memory adapter tests other components and never proves the contract.

The port

append(run_id, seq, record), read(run_id, after, limit), head(run_id), put_version, get_version, create_run, get_run, claim, renew, release, set_runnable_at, expired_leases, overdue_deadlines. Plus an optional Queue and the separate BlobStore port (see psych-blobs).

The same Spec on all four

One of Psych's ten definition-of-done items: the same Spec runs identically against all four adapters. If yours behaves differently, the contract suite is where to prove it.

Gotchas

await store.migrate() before the first Run, not lazily on first use.
DynamoDB caps one item at 400KB. An oversized tool result needs a BlobStore; see psych-blobs.
Lease length is a Worker setting, not a store one. Too short and a slow model call loses its lease; too long and a dead Worker's Runs sit idle.
In development this repository ships scripts/dev-services.sh, which runs the three databases as ordinary processes where there is no container runtime.

On this page