Pinned trays and durability

Signatures quoted from source. Usage: Author a PinnedAPI, Persist across restarts. Related: pinned reference.

Entry pointsCopy Anchor Link

pinned.durabilityCopy Anchor Link

StateStoreCopy Anchor Link

class StateStore(Protocol):
    apply_batch: Callable[[str, str, list[tuple], int], Awaitable[None]] | None   # read-only property
    async def set_tray(self, ns: str, id: str, tray_ns: str, tray_id: str, marked: bool, doc: dict, epoch: int) -> None
    async def delete_tray(self, ns: str, id: str, tray_ns: str, tray_id: str, epoch: int) -> None
    async def get_tray(self, ns: str, id: str, tray_ns: str, tray_id: str, epoch: int) -> tuple[bool, dict] | None
    async def list(self, ns: str, id: str, epoch: int, *, tray_ns: str | None = None, marked: bool | None = None) -> list[tuple[str, str, bool, dict]]

Durable instance state as one row per tray. Every op carries the caller's epoch and acquires the instance atomically: at or above the stored epoch advances it and executes, below raises FenceLost. An instance with no stored epoch is at 0.

MemoryStateStoreCopy Anchor Link

class MemoryStateStore:   # pinned.durability
    epochs: dict[tuple[str, str], int]; trays: dict[tuple[str, str, str, str], tuple[bool, dict]]; ops: int

In-memory StateStore (deep-copied docs). set_tray raises TypeError on a non-bool mark and ValueError on a non-dict doc.

FenceLostCopy Anchor Link

class FenceLost(RuntimeError):
    def __init__(self, key: str, epoch: int) -> None
    key: str; epoch: int

A newer activation owns the record; the write is permanently fenced and the host aborts the instance.

InstanceDurabilityCopy Anchor Link

class InstanceDurability:
    def __init__(self, store: StateStore, ns: str, id: str, epoch: int, domain: Domain, *, dispatch_slots: asyncio.Semaphore | None = None) -> None
    store; ns; id; epoch; domain; dispatch_slots
    frontier: Frontier | None   # set by PinnedContext

Built by the host at activation with the placement epoch and the worker's dispatch_slots.

Frontier, ReceiptCopy Anchor Link

class Frontier:
    def __init__(self, durability: InstanceDurability, abort: Callable[[], None], spawn: Callable[[Coroutine], asyncio.Task]) -> None
    def write_tray(self, tray_ns: str, tray_id: str, marked: bool | None, doc: dict | None) -> int
    def enlist(self, effect: Awaitable[Any]) -> int
    def receipt(self) -> Receipt
    async def wait(self, position: int | None = None) -> None
    def close(self) -> None

class Receipt:
    def __init__(self, frontier: Frontier, position: int) -> None
    def __await__(self)

Instance-wide ordered log of durable effects, flushed in enlist order. Receipt awaits as frontier.wait(position); dropping it is fire-and-forget. Batch coalescing and flush-failure poisoning: Effects.

Frontier methodsCopy Anchor Link

pinned.traysCopy Anchor Link

TrayCopy Anchor Link

class Tray:
    version = 0; max_attempts = 10; dispatch_timeout = 12 * 3600.0   # class attributes, table below
    id: str; storage: TrayStorage; handler: Callable[..., Coroutine]
    parked: bool   # property
    def items(self) -> Generator[Any, None, None];  async def lifecycle(self) -> AsyncIterator[None]
    def kick(self) -> None;  def unpark(self) -> None;  def cancel_dispatch(self) -> None
    def close(self) -> asyncio.Task[None];  def delete(self) -> asyncio.Task[None]

Subclass it, declare durable fields at class level (cloned per instance, hydrated in dependency order) and implement items(); the pump drives it with an at-least-once seam. Errors, parking, and redelivery: Dispatch.

TrayStorageCopy Anchor Link

class TrayStorage:
    marked: bool   # property
    def get(self, name: str, default: Any = ABSENT) -> Any;  def set(self, name: str, value: Any) -> None;  def delete(self, name: str) -> None
    def set_marked(self, marked: bool) -> None

One tray's durable row: the marked flag plus a doc with one entry per field name. Every mutation enlists the whole row on the instance frontier in call order. An unchanged value or mark enlists nothing. A stored None stays distinct from an absent entry. RuntimeError on a write to a retired tray. ValueError on a name that is not a non-empty string or starts with ~. TypeError on a non-bool mark.

tray, tray.mapCopy Anchor Link

def tray(tray_cls: type[Tray]) -> Callable[[handler], TrayDescriptor]
def tray.map(tray_cls: type[Tray], *, always_open: bool = False) -> Callable[[handler], TrayDescriptor]

@tray(C) declares a singleton tray (always open; await self.<name>.get() is the one instance). @tray.map(C) declares a keyed tray type reached by id. The decorated async def method is the handler, called (item, ctx). The tray name is its row's tray_ns; pinned is reserved and raises ValueError.

TrayDescriptorCopy Anchor Link

class TrayDescriptor:
    def __init__(self, tray_cls: type[Tray], handler, *, always_open: bool) -> None
    tray_cls; handler; name: str; always_open: bool
    def on_activate(self, fn: Callable[..., AsyncIterator[None]]) -> Callable[..., AsyncIterator[None]]
    def peek(self, obj: Any) -> Any                        # the namespace already created on obj, else None

At class level the declared tray type. On an API instance it resolves to the namespace. on_activate registers a generator hook (api, id, tray): mount before the yield, teardown after (RuntimeError unless it yields exactly once). peek returns the namespace already created on obj. TypeError when tray_cls is not a Tray subclass, tray_cls.version is not an int, the handler is not async def, or always_open is not a bool.

TrayNamespace, TrayMapNamespaceCopy Anchor Link

class TrayNamespace:        # @tray
    async def get(self) -> Tray
    async def close(self) -> None

class TrayMapNamespace:     # @tray.map
    instances: dict[str, Tray]   # property, a copy
    def retiring(self, id: str) -> bool
    async def get_or_create(self, id: str) -> Tray;  async def get(self, id: str) -> Tray
    async def delete(self, id: str) -> None;  async def close(self) -> None

get() creates the singleton on first access. get(id) raises KeyError for a never-created id. delete(id) opens a passivated tray first and resolves once the id is free. retiring(id) is True while a prior instance's retirement drains. close() stops pumps, runs teardowns and drains flushes. RuntimeError on access without PinnedWorker(state_store=...) or after close().

TrayCtxCopy Anchor Link

class TrayCtx:
    tray: Tray; attempts: int; state: Any; cancelled: asyncio.Event; consumed: bool
    def enlist(self, effect: Awaitable[Any]) -> None
    def set_state(self, value: Any) -> None
    async def dispatch(self, key: Any, target: Callable[..., Any], *args: Any) -> str   # "admitted" | "absorbed"

Per-dispatch surface handed to the handler; all three methods raise RuntimeError after the dispatch settled.

inboxCopy Anchor Link

def inbox(fn: Callable[..., Any]) -> Callable[..., Receipt]

Marks a sync tray method whose body runs as one atomic admission segment. The call returns the Receipt for it. TypeError on an async def. RuntimeError when called on a retired tray. Under a keyed admission, queue appends and extend take dict items only (TypeError), reject a ~admit key (ValueError), and insert raises RuntimeError.

Tray storageCopy Anchor Link

persistent, persistent_queue, persistent_outboxCopy Anchor Link

def persistent(initial: Any = None) -> Persistent
def persistent_queue() -> PersistentQueue
def persistent_outbox() -> PersistentOutbox

class Persistent(Observable):
    def __init__(self, initial: Any = None) -> None
    value: Any   # assign and read

class PersistentQueue(Observable):
    def __init__(self, items: Any = ()) -> None
    value: list
    def append(self, item: Any) -> None
    def extend(self, items: Any) -> None
    def insert(self, index: int, item: Any) -> None
    def peek(self) -> Any
    def popleft(self) -> Any
    async def take(self) -> AsyncIterator[Any]
    def take_now(self, count: int | None = None) -> list[Any]
    def untake(self, count: int) -> None

class PersistentOutbox(Observable):
    def __init__(self, initial: Any = None) -> None
    value: {"item": Any, "attempt": int} | None
    attempt: int   # property; 0 when empty
    def dispatch(self, item: Any) -> int
    def ack(self) -> None

Durable fields declared at Tray class level, each one doc entry. A value equal to the declared initial, an empty queue, or an empty outbox occupies no storage. Doc values are wire values (bool, None, str, finite float, int within ±(2**53 - 1), aware datetime, list/tuple, dict with str keys). An unsupported type or a non-str key raises TypeError at write, a non-finite float, an out-of-range int, or a naive datetime a ValueError. Reserved entries: ~inbox, ~state, ~v, ~attempts, ~parked, ~dispatch, ~ignore.

PersistentOutbox.dispatch bumps attempt when the item equals the current one and resets it to 1 otherwise. ack() on an empty outbox raises RuntimeError.

observable, ObservableCopy Anchor Link

Re-exports of statepatch's observable and Observable. Op is (kind, path, value) with kind in "replace" | "add" | "remove". Wrapping an Observable in another raises TypeError.

pinned.durability.postgresCopy Anchor Link

PostgresStateStoreCopy Anchor Link

class PostgresStateStore:
    def __init__(self, conn: AsyncConnection[Any] | None = None, *, dsn: str | None = None, pool_size: int = 8, env: str = "*", hostname: str = "*", configure_transaction: Callable[[AsyncConnection[Any]], Awaitable[None]] | None = None) -> None
    async def setup(self) -> None

StateStore on Postgres: every op runs in its own transaction holding the instance's pinned_epoch row lock (SELECT ... FOR UPDATE) and writes that row only when the op's epoch is newer. Ops on one (ns, id) stay ordered, different instances proceed in parallel. A dsn opens a lazy pool of pool_size connections, a conn is a pool of one. env/hostname stamp every row with the worker's placement identity for pinboard's tray index. "*" matches single-world mode.

• configure_transaction(conn) runs after transaction-local timeouts and before epoch acquisition on every attempt, including retries.
• setup() invokes the hook before creating tables.
• Hook failures use the transaction's rollback, deadline, and retry handling. setup() rolls back hook failures.
• configure_transaction must be callable or None.

setup() creates pinned_epoch (ns, id, epoch) and pinned_tray_index (ns, id, tray_ns, tray_id, marked, doc, env, hostname). TypeError when env or hostname is not a string or pool_size not an int. ValueError unless exactly one of conn/dsn is given or when pool_size < 1. Importing without the extra raises ImportError (pip install 'pinned[postgres]').