Author a PinnedAPI

Subclass PinnedAPI: lifespan yields once (setup before, teardown after), and @route.* methods are the instance's HTTP surface. Instance memory is lost on eviction. Anything that must outlive the process goes in a tray, a durable queue whose fields live in the state store.

A first classCopy Anchor Link

from fastapi import FastAPI
from pinned import PinnedAPI, PinnedWorker, route
from pinned.durability import MemoryStateStore

class Counter(PinnedAPI):
    async def lifespan(self):
        self.count = 0          # setup
        yield
        # teardown

    @route.get("/value")
    async def value(self):
        return {"id": self.id, "count": self.count}

    @route.post("/increment")
    async def increment(self):
        self.count += 1
        return {"count": self.count}


worker = PinnedWorker(hosts={"/counters": Counter}, state_store=MemoryStateStore())
app = FastAPI()
app.include_router(worker)

• PinnedWorker is a FastAPI router serving every entry in hosts. uvicorn app:app serves /counters/<id>/value.
• Signatures: PinnedAPI, route. The model: durable objects.

RoutesCopy Anchor Link

• Handlers are async def instance methods decorated with route.get/post/put/delete/patch/websocket(path, auth=...). Every request for one id runs on that process's event loop.
• Parameters are FastAPI parameters (query, body, WebSocket, ...). WebSocket handlers work as in FastAPI. id and pinned_record are reserved names. The routed id is self.id.
• A parameter typed with a Pydantic model or dict binds as the JSON body.
• Ids are client-chosen names (Ids). The first request for an id activates it.
• Return a StreamingResponse (any Starlette Response with a body iterator) to stream: the stream counts as in-flight work, advances only as the client reads, and ends when the instance closes. Streams are pull-based and unbuffered. A slow client backpressures only its own stream.

AuthCopy Anchor Link

• authorize(conn) gates every route. auth= on a route swaps the resolver (None disables it). A per-route resolver is an async def authorize_admin(self, conn) defined above the route in the class body; each call binds it to the instance: @route.post("/wipe", auth=authorize_admin).
• Depends(auth_result) reads the resolver's return value inside a handler.
• Auth runs after __init__ and before lifespan: self.id is available, lifespan state is not. A denied cold request discards its constructed instance unless another request is still authorizing on it.
• Deny by raising AuthRejected, or return a Starlette Response on an HTTP route to answer with it; a returned Response on a WebSocket route is a TypeError.

TraysCopy Anchor Link

A tray is a durable execution queue inside an instance: its fields live in the state store (PinnedWorker(state_store=...)), and its pump drives items(). A crash resumes it from the store on the next activation.

A tray with a queue and a handlerCopy Anchor Link

from fastapi import FastAPI
from pinned import (
    PinnedAPI,
    PinnedWorker,
    Tray,
    inbox,
    persistent,
    persistent_queue,
    route,
    tray,
)
from pinned.durability import MemoryStateStore


class Jobs(Tray):
    queue = persistent_queue()
    done = persistent([])

    def items(self):
        while self.queue.value:
            yield self.queue.peek()
            self.queue.popleft()      # the durable ack

    @inbox
    def add(self, job: dict) -> None:
        self.kick()               # the mark leads the payload
        self.queue.append(job)


class Runner(PinnedAPI):
    @tray(Jobs)
    async def jobs(self, job, ctx) -> None:
        ctx.tray.done.value = [*ctx.tray.done.value, job["name"]]

    @route.post("/enqueue")
    async def enqueue(self, job: dict) -> dict:
        jobs = await self.jobs.get()
        await jobs.add(job)
        return {"admitted": job["name"]}

    @route.get("/done")
    async def done(self) -> dict:
        jobs = await self.jobs.get()
        return {"done": list(jobs.done.value)}


worker = PinnedWorker(hosts={"/runners": Runner}, state_store=MemoryStateStore())
app = FastAPI()
app.include_router(worker)

Run it with uvicorn app:app --port 8000:

$ curl -X POST localhost:8000/runners/r1/enqueue -H 'content-type: application/json' -d '{"name":"build"}'
{"admitted":"build"}
$ curl localhost:8000/runners/r1/done
{"done":["build"]}

MemoryStateStore empties with the process. Postgres keeps the rows across restarts (stores):

import os

from pinned.durability.postgres import PostgresStateStore  # uv add "pinned[postgres]"

store = PostgresStateStore(dsn=os.environ["DATABASE_URL"])
worker = PinnedWorker(hosts={"/runners": Runner}, state_store=store)
# in the app lifespan, once at boot:
await store.setup()

A tray that only holds durable fields: items() yields nothing (yield from ()) and the handler raises. Read and write the fields through await self.<tray>.get() (durable host).

FieldsCopy Anchor Link

Durable fields are class attributes on the Tray, one doc entry each (reference).

DeclaringCopy Anchor Link

• @tray(C) declares a singleton: await self.<name>.get() is the one instance, opened at every activation. @tray.map(C) declares a keyed type.
• The handler is the decorated async def (self, item, ctx), the same for both kinds.
• await self.<name>.get() works inside lifespan before yield: the tray namespace is created on the first self.<name> access.
• items() yields the head item. The generator is re-created on each drain: derive it from durable state only.
• lifecycle() on the tray and @<name>.on_activate (api, id, tray) on the API, both async generators, bracket each hydration. A tray that drains dry unmarks its row and passivates unless always_open. The next access re-hydrates it.
• A marked stored row with no matching @tray on the class raises RuntimeError at activation, as does a row whose kind changed between @tray and @tray.map. Unmarked rows of a removed tray stay on disk unread.
• Bump Tray.version to discard a tray's durable state: a stored row of another version activates blank.

AdmissionCopy Anchor Link

• An @inbox method is sync and runs as one atomic admission segment. Mark and payload land in one row write.
• The mark is the row flag claiming in-flight work. kick() sets it, before the payload append.
• The call returns a Receipt. Awaiting it waits for the write to be durable; dropping it is fire-and-forget.
• The durable ack is the code after the yield in items(): the item is gone from the queue only once that write lands.

DispatchCopy Anchor Link

• At-least-once and crash-only. ctx.attempts is recorded before the handler runs.
• To keep one item's failure inside the instance, catch it in the handler and return.
• tray.unpark() raises RuntimeError on a tray that is not parked; kick() and @inbox calls raise it on a retired tray. tray.cancel_dispatch() sets ctx.cancelled and consumes the item. tray.delete() admits the built-in delete command: the in-flight dispatch finishes, then one atomic row delete, re-run on the next activation if the crash came first.
• A persistent_outbox() holds the item across those returns: dispatch(item) records it and bumps attempt for the same item, ack() clears it. An unacked outbox re-yields on the next loop of items(), at once and in the same drain. Pace it with await self.domain.sleep(...) in the handler.
• ctx.attempts counts crash redeliveries; outbox.attempt counts in-handler retries. Cap those in the handler: ack() the entry and record the failure.

EffectsCopy Anchor Link

• The frontier (self.frontier) is the instance's ordered log of durable effects. Storage writes and enlisted awaitables land on it in enlist order; enlisting is the append.

• Adjacent writes coalesce into one batch on a store with apply_batch (a tray's writes compress to a single row upsert); without it, one write per row. An opaque effect, an awaitable the log only awaits, ends the batch in progress.

• A set whose value matches the last known value enlists nothing. An @inbox body's writes flush as one batch.

• A flush failure poisons the log and aborts the instance. Every later enlist or wait raises RuntimeError chained from the failure. Memory never runs ahead of the store.

• Tray.drain_on_eviction = True keeps admission handlers running during wind-down; their tracked work finishes before release. Statewire uses this for authenticated command intake, which finishes before deadline cancellation reaches command tails. Execution trays pause new dispatches by default.

Keyed traysCopy Anchor Link

• @tray.map(C) reaches instances by id: get_or_create(id) creates on first use, get(id) raises KeyError for a never-created id, delete(id) opens even a passivated instance and frees the id.
• A keyed instance reads its own key as ctx.tray.id; the instance object is ctx.tray.
• always_open=True opens every stored instance of that type at activation and keeps it open. The default opens marked rows only.

Background work and lifecycleCopy Anchor Link

self.create_task(coro) runs instance-owned work.

• self.lease (InstanceLease) exposes healthy, expires_at, winding_down, and eviction_deadline.
• self.lease.wind_down(deadline=None) requests cooperative eviction. A finite, nonnegative deadline cancels remaining execution after that many seconds; 0 cancels immediately. Repeated calls can tighten the local deadline.
• await self.eviction_checkpoint() raises pinned.EvictionRequested while winding down. Call it at safe task boundaries; cleanup runs before placement release.
• configure(on_idle=, on_wind_down=, on_unhealthy=, on_restored=, buffer=) installs lifecycle callbacks. on_idle controls ordinary idle retention; wind-down evicts idle instances immediately. on_wind_down(deadline) fires when the effective deadline or wind-down state changes. Check self.lease.winding_down to distinguish cooperative wind-down from revocation when deadline is None.
• self.domain supplies now/monotonic/sleep/wait/spawn/rng/token_hex. Use it over time/asyncio so the instance runs under simulation.

Calling other instancesCopy Anchor Link

self.cluster is an httpx.AsyncClient addressing instances by data-plane path.

• await self.cluster.post(f"/threads/{target}/accept") reaches /<ns>/<id>/....
• A POST to an unplaced id places and wakes it.
• Targets on any host attached to the caller's PinnedWorker (its own namespace or a sibling's) dispatch in-process, the caller attested. Others go through pinboard, which verifies the caller and stamps caller/depth claims into the Pinboard-Attestation token. An attested call reaches only worlds the caller's own registration covers (instance to instance).
• On the receiving side, attestation(conn) names the calling instance, a passport-verified principal, or None.

match attestation(request):
    case Attestation.Instance("/threads", id): ...
    case _: raise HTTPException(status_code=403)

Where nextCopy Anchor Link

• Every option and signature: trays and durability, pinned reference.
• Stores, fencing, what survives what: Persist across restarts.
• Placement, leases, epochs: durable objects.