Realtime and durable work#

Choose a primitive by lifetime: connection, room, resumable stream, durable attempt, multi-step workflow or exclusive live owner.

wreath.websocket#

WebSocket connection API for route handlers.

A handler registered with @app.websocket(path) receives one WebSocket per connection. The server has already validated the upgrade handshake; the handler decides whether to WebSocket.accept or WebSocket.close it, then exchanges messages:

@app.websocket("/echo")
async def echo(ws: WebSocket) -> None:
    await ws.accept()
    async for message in ws:
        await ws.send(message)

Iteration ends when the peer disconnects. Explicit receive_text() / receive_bytes() raise WebSocketDisconnect instead.

Calls#

Request/response over one socket, correlated by an identifier.

A WebSocket is a frame pipe: frames arrive in order and nothing pairs a reply with the request that caused it. Every protocol that puts a request/response contract on top of one -- and most do -- grows the same three things by hand: an identifier on the way out, a map of what is outstanding, and a deadline so a peer that never answers does not pin a caller forever.

Calls owns the read loop and demultiplexes. Both directions are declared once, as functions over decoded messages, so the protocol lives in one place instead of being spread through the handler:

calls = Calls(
    ws,
    reply_to=lambda message: message.get("reply_to"),
    label=lambda identifier, payload: {"id": identifier, **payload},
)

@calls.on_request
async def handle(message: dict) -> dict | None:
    return {"reply_to": message["id"], "ok": True}

async with calls:
    answer = await calls.call({"op": "read"}, timeout=seconds(30))

reply_to returns the identifier a frame is answering, or None when the frame is a peer-initiated request; label stamps an outgoing request with the identifier a reply will carry back. Those two are the whole protocol seam, and neither has a default: a guessed correlation field is a protocol this class does not know it is implementing.

One reader. WebSocket documents that it holds no lock and assumes one reader; this is that reader, so a handler using Calls must not also iterate the socket. Sends are serialised by a lock, because a request and a reply genuinely do race.

On close, every outstanding question fails with CallsClosed rather than waiting out its own deadline -- an answer that provably cannot arrive should not cost a caller its timeout.

Calls(ws: WebSocket, *, reply_to: Callable[[Any], str | None], label: Callable[[str, Any], Any], encode: Callable[[Any], str | bytes] = <function _json_text>, decode: Callable[[str | bytes], Any] = <function _json_value>, timeout: Any = 30.0, max_pending: int = 256) -> None

call#

async call(payload: Any, *, timeout: Any = None) -> Any

Send payload as a request and wait for the reply that names it.

Raises TimeoutError when the deadline passes and CallsClosed when the socket goes away first.

close#

async close() -> None

Stop reading and fail every outstanding call.

on_request#

on_request(handler: Callable[[Any], Awaitable[Any]]) -> Callable[[Any], Awaitable[Any]]

Register the handler for peer-initiated frames.

Returning a value sends it; returning None sends nothing, which is what a notification -- a request with no response -- means.

outstanding (property)#

outstanding: int

Calls awaiting an answer right now.

refusals (property)#

refusals: int

Calls refused because too many were already outstanding.

send#

async send(payload: Any) -> None

Send one frame, serialised against every other sender.

CallsClosed#

The socket went away while a call was outstanding.

ConnectionBackpressure#

A managed connection could not accept another outbound frame.

ConnectionSnapshot#

Point-in-time operational counters for one connection service.

ConnectionSnapshot(active: int, queued: int, accepted: int, closed: int, capacity_refusals: int, queue_refusals: int, protocol_refusals: int, heartbeat_timeouts: int, drain_timeouts: int) -> None

Heartbeat#

Protocol-supplied heartbeat frame and acknowledgement contract.

Wreath does not guess an application subprotocol. The caller supplies the frame and the predicate that recognizes its answer; the service supplies bounded scheduling, timeout, and closure.

Heartbeat(*, frame: str | bytes, acknowledge: Callable[[str | bytes], bool], interval: Any = 30.0, timeout: Any = 10.0, consume: bool = True) -> None

WebSocket#

One WebSocket connection, backed directly by the ASGI scope.

Constructed by the framework, one per connection, and handed to the @app.websocket(path) handler. The server has already validated the upgrade handshake and the route's authentication; what is left to the handler is to accept or close.

Nothing may be sent before accept -- the ASGI server rejects a websocket.send on an unaccepted connection. The first receive of any kind consumes the websocket.connect message, so accept, close, and receive are each safe as the first call.

The socket is its own async iterator: async for message in ws yields str for text frames and bytes for binary ones, and stops -- rather than raising -- when the peer disconnects. A handler that must distinguish a clean close from a dropped connection reads the code off WebSocketDisconnect from an explicit receive instead.

This class holds no lock and assumes one reader and one writer. Two tasks receiving on the same socket interleave frames unpredictably.

Args:

• path_params — Route captures; {} when the route has none.
• identity — The caller when the route declared auth; None on an open route.

WebSocket(scope: dict[str, Any], receive: Any, send: Any, path_params: dict[str, str] | None = None, *, identity: Any = None) -> None

accept#

async accept(subprotocol: str | None = None, headers: list[tuple[bytes, bytes]] | None = None) -> None

Complete the handshake. Nothing may be sent before this returns.

Consumes the websocket.connect message if it has not been consumed already, then sends websocket.accept. Calling it twice is a protocol error the server raises on, not something this class guards.

Args:

• subprotocol — One value for Sec-WebSocket-Protocol; None selects none.
• headers — Extra response headers as raw byte pairs, e.g. a Set-Cookie.

Raises:

• RuntimeError — The first message from the peer was not websocket.connect.

close#

async close(code: int = 1000, reason: str = '') -> None

Close the connection with a Close frame carrying code.

Valid before accept too, where it rejects the handshake instead: the server answers the upgrade with a plain HTTP error rather than a Close frame, so code and reason are not seen by the peer.

Only codes an endpoint may put on the wire are accepted (RFC 6455 §7.4.1): the assigned 1000-range values except 1004, 1005, 1006 and 1015, plus the 3000-4999 registered and private-use range. The three excluded ones exist to describe a closure locally and MUST NOT be sent, which is why they are refused here rather than forwarded.

Args:

• code — Close code. 1000 is normal closure; 1011 reports an internal error.
• reason — UTF-8 text for a human; peers are not required to surface it.

Raises:

• ValueError — code is not a code an endpoint may send.

header#

header(name: str | bytes, default: str | None = None) -> str | None

One handshake header, decoded latin-1, or default when absent.

name is lowercased before the lookup, and ASGI guarantees the scope's header names are already lowercase, so case does not matter. A header sent more than once resolves to the first occurrence; read headers directly for the rest.

Args:

• name — Header name, str or bytes; case does not matter.
• default — Returned when the header is absent. Not encoded or validated.

headers (property)#

headers: list[tuple[bytes, bytes]]

Handshake headers as raw (name, value) byte pairs, names lowercased.

path (property)#

path: str

The percent-decoded request path from the handshake.

query_string (property)#

query_string: bytes

The raw, still percent-encoded query string; b"" when there is none.

receive#

async receive() -> Message

The next raw ASGI message (websocket.receive or .disconnect).

The escape hatch below the typed helpers: it never raises on disconnect, so a handler that wants to inspect the close code, or to accept text and binary in one place, reads the dict itself.

Returns:

• The ASGI message dict, unmodified.

Raises:

• RuntimeError — The first message from the peer was not websocket.connect.

receive_bytes#

async receive_bytes() -> bytes

The next message, which must be a binary frame.

The mirror of receive_text, and equally strict: a text frame raises rather than being encoded.

Raises:

• WebSocketDisconnect — The peer closed or dropped; .code carries the close code.
• RuntimeError — The peer sent a text frame.

receive_json#

async receive_json() -> Any

Decode the next text or binary frame as JSON.

receive_text#

async receive_text() -> str

The next message, which must be a text frame.

Strict about the frame type: a binary frame arriving here raises rather than being decoded, because a peer sending the wrong opcode is a bug worth surfacing. The offending message is consumed and discarded either way -- use receive to accept both kinds in one place.

Raises:

• WebSocketDisconnect — The peer closed or dropped; .code carries the close code.
• RuntimeError — The peer sent a binary frame.

send#

async send(data: str | bytes) -> None

Send one message, framed by type: str as text, anything else as binary.

The dispatch is on str alone. A bytearray or memoryview takes the binary branch and is then refused by the server, which requires exact bytes -- convert before sending. Sending before accept is also the server's refusal, not this method's.

Args:

• data — str is encoded UTF-8 by the server; bytes is sent verbatim.

send_bytes#

async send_bytes(data: bytes) -> None

Send a binary frame. Explicit about the frame type, unlike send.

send_json#

async send_json(data: Any) -> None

Encode data with Wreath's JSON codec and send one text frame.

send_text#

async send_text(data: str) -> None

Send a text frame. Explicit about the frame type, unlike send.

subprotocols (property)#

subprotocols: list[str]

Subprotocols the client offered, in its order of preference.

Advisory: nothing checks that the value passed to accept came from this list. Selecting one the client did not offer is a protocol violation the peer is entitled to fail the connection over.

WebSocketDisconnect#

The peer closed or dropped the connection.

Raised by WebSocket.receive_text and WebSocket.receive_bytes. Iterating a socket does not raise it -- iteration stops instead.

Args:

• code — The peer's close code, or 1006 when it left without a Close frame.

WebSocketDisconnect(code: int = 1006) -> None

WebSocketService#

Bounded lifecycle and outbound flow control for long-lived sockets.

The service is protocol-neutral. Applications supply a message handler and optional connection key; Wreath supplies capacity admission, one bounded outbound queue per connection, serialized writes, shutdown draining, and counters. Register it with app.service(name, service) and delegate a route to serve.

overflow is explicit because all three useful policies have different operational meanings:

• "reject" refuses the producer immediately;
• "backpressure" waits for queue space, bounded by enqueue_timeout;
• "disconnect" closes a peer that cannot keep up, then refuses the send.

No mode grows memory with traffic. The default is rejection, which keeps a slow peer from delaying an unrelated producer.

WebSocketService(*, max_connections: int = 1024, queue_capacity: int = 64, overflow: "Literal['reject', 'backpressure', 'disconnect']" = 'reject', enqueue_timeout: Any = 5.0, heartbeat: Heartbeat | None = None) -> None

broadcast#

async broadcast(frame: _Frame) -> int

Enqueue one frame for each current connection; return acceptances.

drain#

async drain(deadline: float) -> None

Stop admission, close every peer, and wait until the deadline.

send#

async send(key: Hashable, frame: _Frame) -> None

Enqueue one frame under the configured overflow policy.

serve#

async serve(websocket: WebSocket, handler: Callable[[_Frame], Awaitable[_Frame | None]], *, key: Hashable | None = None, subprotocol: str | None = None) -> None

Own one connection until the peer leaves or the handler fails.

The handler is called sequentially, so inbound work has natural backpressure. A returned frame enters the same bounded outbound queue as send; returning None sends nothing. Handler failures propagate to Wreath's ordinary route error boundary after the connection is removed, rather than being swallowed by the service.

snapshot (property)#

snapshot: ConnectionSnapshot

Current gauges and cumulative refusal/closure counters.

start#

async start(supervisor: Any) -> None

Open admission when the application lifespan starts.

wreath.rooms#

WebSocket rooms: broadcast to a named group of sockets, across workers.

A room is a name and the sockets currently in it. Joining and leaving are local bookkeeping; broadcasting reaches every worker through the PostgreSQL message bus, so a four-worker deployment behaves like a one-worker one with no Redis and no extra process:

rooms = RoomRegistry(app.message_bus("main"))

@app.websocket("/chat/{room}")
async def chat(ws):
    room = ws.path_params["room"]
    await ws.accept()
    await rooms.join(room, ws)
    try:
        async for text in ws:
            await rooms.broadcast(room, text)
    finally:
        await rooms.leave(room, ws)

Without a bus it still works, single-process: broadcasts go to local members only. That is the right default for tests and single-worker deployments.

One channel, not one per room. The bus registers its subscriptions at startup and validates channel names as SQL identifiers, so a room cannot own a channel. Every worker instead subscribes once and filters locally by room name. The trade is that a worker receives traffic for rooms it holds no members of and drops it — fine at chat scale, and it keeps LISTEN count at one regardless of how many rooms exist. A deployment with very high cross-room traffic and many workers should shard by running separate registries on separate channels.

Delivery is at-most-once and unordered across workers, because the transport is ephemeral NOTIFY. A room is for live fan-out, not for durable messaging -- if a message must survive a disconnect, persist it and use wreath.messaging durable delivery.

RoomRegistry#

Named groups of WebSockets, with cross-worker broadcast over the bus.

One registry per application. Pass the message bus to reach other workers; omit it for single-process fan-out.

Membership is per worker and lives only in memory: a restart empties every room, and members and snapshot report this process's share rather than the fleet's. Nothing here is a durable channel -- see the module docstring for what NOTIFY does and does not guarantee.

Args:

• bus — A wreath.messaging bus, or None for local-only broadcast.
• channel — Bus channel carrying every room; must be a valid SQL identifier.

RoomRegistry(bus: Any = None, *, channel: str = 'wreath_rooms') -> None

broadcast#

async broadcast(room: str, payload: str | bytes, *, grade: Callable[[Any], Any] | None = None, render: Callable[[Any, Any], Any] | None = None) -> int

Send payload to every socket in room, on every worker.

Pass grade and render together to deliver one event at each subscriber's own authorization outcome -- two watchers of one incident seeing the same position at different precisions, because the authorizer said so. grade(websocket) runs per socket and answers what that connection may see; render(grade, payload) runs once per distinct grade and shapes the payload for it, returning None to send that grade nothing. Passing only one of the two is refused, since a grade nothing renders would silently deliver the ungraded payload to everyone.

Grading is local. A graded broadcast is not published to other workers, because the grade of a socket on another worker cannot be computed here and shipping the ungraded payload for the far side to shape would put the authorization decision on the wire. Fan a graded event out from each worker, or grade after receipt.

Local delivery happens first and is complete before this returns. A socket whose send raises is dropped from the room and not counted; one dead peer never aborts the rest of a broadcast.

The cross-worker publish is awaited, unlike Wreath's other bus bridges: the caller asked for this fan-out and is waiting on its result, so a bus failure is theirs to see and propagates from here. What is not guaranteed is that other workers receive it -- the transport is NOTIFY, so remote delivery is at-most-once and unordered.

Arbitrary bytes cross the bus intact. The cross-worker payload is JSON, which carries no binary type, so a bytes payload travels as a string: decoded UTF-8 when it is valid UTF-8, and base64 when it is not. Either way the far side reconstructs the original bytes, so nothing here refuses a payload for its contents and no encoding failure can leave a broadcast half-delivered. The bus form is built before the first local send, so if that ever does become able to fail it fails with nothing delivered.

Args:

• payload — str or bytes; bytes is delivered to sockets as a binary frame.
• grade — (websocket) -> hashable, this connection's authorization outcome.
• render — (grade, payload) -> payload | None, once per distinct grade.

Returns:

• The number of sockets on this worker the payload was delivered to.

Raises:

• ValueError — exactly one of grade and render was given.

grade_errors (property)#

grade_errors: int

Sockets skipped because their grade callable raised.

A grade that raises for every socket delivers to nobody. That is the fail-closed direction and the right one, but it is indistinguishable from an empty room without a number to read.

join#

async join(room: str, websocket: Any) -> None

Add websocket to room (idempotent).

The room is created on first join. Members are held in a set, so joining twice leaves one member and one delivery per broadcast.

Args:

• websocket — Anything with an awaitable send(payload); it need not be a WebSocket.

leave#

async leave(room: str, websocket: Any) -> None

Remove websocket from room; drop the room when it empties.

Safe to call for a socket that never joined, so it belongs in a finally without a guard.

leave_all#

async leave_all(websocket: Any) -> None

members#

members(room: str) -> int

How many sockets this worker holds for room. 0 for an unknown room.

A local count, not a fleet-wide one: other workers hold their own members and this registry never asks them.

rooms#

rooms() -> list[str]

Room names this worker holds at least one socket for, sorted.

A room disappears from this list as soon as its last local member leaves, so it tracks local membership rather than rooms in existence.

snapshot#

snapshot() -> dict[str, int]

Room name -> local member count. For health and debug endpoints.

A point-in-time copy, safe to serialize. Counts are this worker's; a four-worker deployment needs four of these to see the whole fleet.

wreath.messaging#

Pub/sub + event fan-out backed by PostgreSQL — a RabbitMQ-shaped surface.

Two delivery tiers, chosen per publish/subscribe:

• Ephemeral fan-out (durable=False): NOTIFY to every live subscriber on this channel. At-most-once, sub-millisecond, no persistence — the direct "tell all replicas now" analogue. Payloads are bounded (NOTIFY caps at 8000 bytes); oversized payloads must go durable.
• Durable (durable=True): a work-queue table consumed with FOR UPDATE SKIP LOCKED + fencing (the same machinery as wreath.jobs), with NOTIFY used only as a wakeup doorbell. At-least-once, replayable, dead-letterable.

Durable fan-out delivers one copy per subscriber group, and the groups are discovered fleet-wide: each bus writes its durable subscriptions into a shared message_groups table at startup, and every publisher reads that table. Discovering them from local registrations instead — as this module once did — meant a publisher deployed before its consumer, or living in a different service, enqueued nothing for that group: no error, no dead letter, the message simply never existed for it.

Local registrations are unioned with the persisted ones rather than replaced by them, because the two failure modes are not symmetric. A duplicate copy goes to a group that has a consumer, and durable delivery is at-least-once anyway, so handlers are already idempotent. A lost copy is silent. The union also means a deployment that has not applied the new table behaves exactly as it did before.

Multi-tenancy: a dedicated system schema + tenant column, never search_path (design 01 §5).

Message#

A delivered message. Durable messages honour ack/nack/reject.

Message(channel: str, group: str | None, tenant: str, payload: Any, id: int | None = None, fence: int | None = None, attempts: int = 0, trace_context: str | None = None, _disposition: str = 'ack') -> None

ack#

ack() -> None

Mark handled — the message completes (durable) / no-op (ephemeral).

envelope#

envelope() -> MessageEnvelope | None

Return the versioned envelope, or None for a legacy payload.

nack#

nack() -> None

Fail transiently — the durable message retries with backoff.

reject#

reject() -> None

Fail permanently — the durable message is dead-lettered immediately.

MessageBus#

A named message bus on one application database. Obtain via wreath.Wreath.messaging.

MessageBus(database: Any, *, name: str, workload: str = 'write', schema: str = 'wreath', poll_interval: Any = 5.0, lease: Any = 30.0, group_refresh: Any = 30.0) -> None

component#

component() -> Any

This bus's claim on the wreath schema.

The durable queue and the group registry are wreath's furniture, not the application's data model, so they live in the wreath schema and never appear in the application's migration artifact. Wreath collects this during lifespan and brings it up to date before the bus starts, so the group registry the fan-out depends on is there rather than absent.

counters#

counters() -> Any

This bus's counters, for wreath.metrics.collect.

doorbell_reconnects (property)#

doorbell_reconnects: int

Times the doorbell's held LISTEN connection was lost, plus every attempt to (re)open one that failed -- including the attempt made at startup, since a bus that came up against a dead database has no doorbell either. A database that stays down keeps this climbing rather than reading as a single blip.

Non-zero means ephemeral fan-out was down for at least a moment and those messages are gone (at-most-once has nowhere to replay from); climbing means it still is, and durable consumers are running on the poll interval.

Kept deliberately apart from handler_errors: a bug in a subscriber must never read as a flapping database.

drain#

async drain(deadline: float) -> None

known_groups#

known_groups(channel: str) -> frozenset[str]

Every durable group a publish to channel will reach.

The deploy-time check: "will anything actually receive this?" is answerable before shipping rather than by noticing an empty queue days later. Reflects the registry as of the last refresh.

name (property)#

name: str

prune_groups#

async prune_groups(*, unseen_for: float) -> None

Drop registry rows nobody has re-registered in unseen_for seconds.

The backstop behind _deregister_groups: a consumer that was killed rather than drained never deregistered, and seen_at is how that becomes visible. Run it from a scheduled job.

publish#

async publish(channel: str, payload: Any, *, tx: Any = None, durable: bool = False, tenant: str = '', key: str | None = None, require_group: bool = False) -> None

Publish payload (JSON-serialisable) to channel.

Ephemeral (default): a single NOTIFY fans out to live subscribers. Durable: one row per subscriber group is enqueued; pass tx to publish atomically with your writes (the outbox guarantee).

require_group (durable only) raises NoSubscriberGroup when no group is known for channel anywhere in the fleet, for the caller who knows a consumer must exist. Without it the publish is a counted no-op, because shipping a producer before its consumer is normal.

Plain payloads retain their existing wire shape. Passing a MessageEnvelope opts into the marked, versioned object used for event identity, correlation, causation, and explicit trace context. During a rolling deployment an older build still decodes that object as ordinary JSON; consumers should be deployed before producers when they require envelope semantics. Message.envelope() recognizes the new shape while returning None for every legacy payload.

purge#

async purge(*, older_than: float) -> None

Delete finished messages older than older_than seconds.

As with wreath.jobs.JobRunner.purge: caller-driven, done and dead only, and the thing that keeps this table from being append-only.

schema_sql#

schema_sql() -> str

DDL for the durable messages and group-registry tables, joined.

A derivation of component(), not a second copy: the statements live there and this joins them, so the tuple and the script cannot disagree. Retained for a caller applying the DDL itself; wreath schema sql is the supported spelling and component() is what wreath applies.

start#

async start(supervisor: Any) -> None

stats#

stats() -> dict[str, int]

Every counter this bus keeps, by name.

The counters were readable one attribute at a time, which means an exporter has to know each name and gains nothing when one is added. This is the shape a metrics scrape wants.

subscribe#

subscribe(channel: str, *, group: str | None = None, concurrency: int = 1, durable: bool = False, retries: int = 5) -> Callable[[MessageHandler], MessageHandler]

Decorator registering handler(message) for channel.

Durable subscriptions require a group (the competing-consumer set that shares one copy of each message); ephemeral ones ignore it.

MessageEnvelope#

Versioned message identity and causality around a JSON payload.

The marker makes detection exact. A bus only emits this shape when the caller supplies an envelope, so existing publishers retain their byte shape and an older consumer can still decode an envelope as an ordinary JSON object during a rolling deployment.

MessageEnvelope(kind: str, payload: Any, version: int = 1, id: str = '', correlation_id: str | None = None, causation_id: str | None = None, trace_context: str | None = None) -> None

as_dict#

as_dict() -> dict[str, Any]

The explicit version-1 wire object.

decode (classmethod)#

decode(value: Any) -> MessageEnvelope | None

Decode an envelope, returning None for a legacy plain payload.

encode#

encode() -> bytes

Encode with Wreath's native JSON kernel.

NoSubscriberGroup#

A durable publish found no subscriber group, and the caller wanted one.

Only raised for publish(..., require_group=True). Publishing to a channel nobody consumes yet is legitimate -- a producer often ships before its consumer -- so the default stays a counted no-op.

wreath.streams#

Resumable streams over durable work: the producer outlives the connection.

An LLM response is a five-minute HTTP response, and HTTP has no resume. EventSource reconnects; the generation does not, so a five-minute agent task that drops at minute four restarts from zero. The answer everybody converges on is a durable session that decouples delivery from the connection: a reconnecting client, on any device, resumes from the last delivered chunk without re-invoking the producer.

Wreath already had both halves and never joined them. wreath.jobs is durable execution with attempt identity and fencing; wreath.log is a total order under a cursor that cannot skip, with a batched append. This is the join: a request that attaches to a running job's output instead of producing it.

streams = Streams(jobs=runner, log=PostgresLog(database, declaration()))

@streams.producer("chat")
async def answer(stream: StreamWriter, question: str) -> None:
    async for token in some_model(question):
        await stream.write(token)

@app.post("/chat")
async def ask(body: Ask) -> StreamHandle:
    return await streams.start("chat", key=body.stream_id, args=(body.text,))

@app.get("/chat/{key}")
async def resume(request: Request, key: str) -> Response:
    return streams.attach(
        key,
        since=request.header("last-event-id"),
        authorize=lambda stream_key: request.user.owns_stream(stream_key),
    )

No provider client ships here. Providers change monthly, wreath.http_client already pools, retries and rate-limits, and the framework-shaped part is this primitive -- which is provider-agnostic.

The four things this module is careful about#

Exactly-once is a client-side property. The wire is at-least-once. A reconnecting client can be handed a chunk it already has, and the sequence is what lets it drop the duplicate. Nothing on this side promises otherwise, and a server that did would be lying about what a transport can keep.

A retried producer must not append a second copy of chunks 1..400. So every row carries the attempt's fence, which wreath.jobs already bumps on each claim, and a reader skips any row whose fence is below the highest fence that stream has reached. That is the whole idempotency scheme -- see follow for the one rule -- and it is affordable only because the fence already existed. The reader emits an explicit superseded event when it happens, because a client that concatenates a replaced range renders duplicated text and blames the model. Streams(on_retry="truncate") is the simpler discipline the plan kept in reserve: the superseded rows are deleted at the start of the retried attempt. Both tell the reader; they differ only in whether the old bytes survive.

The chunk log is delivery, not transcript. declaration() refuses KEEP_FOREVER outright. Prompts and completions are the newest large pile of unclassified personal data, and a buffer that cannot expire is the shape that makes them unerasable; conversation history is application data in the ORM, where it has an owner and a retention decision of its own. retention_pass executes the window as a counted wreath.passes walk.

A stream nobody attaches to still costs. The producer runs regardless -- that is the entire point -- but an application starting streams nobody reads has built a token furnace with no meter on it. Streams.stats counts started against attached and check_stream_attachment turns that into a finding.

What this does not do#

No fan-out of its own. Two clients attaching to one key both read the same rows out of the same log, which is what makes a second reader free and a second worker unnecessary; there is no in-memory subscriber set to lose on a restart.

No transcript, no projections, no re-derived state. See wreath.log's module docstring for the rest of that boundary -- this is one of its four callers.

StreamCursor#

Where a client got to: which attempt, and where in the log.

Three parts and a stream tag, because all four are load-bearing. The log cursor (xid, seq) is the resume point and cannot skip (see wreath.log's cursor contract); fence is the attempt the client last read under, which is what lets follow say your content was replaced instead of leaving the client to notice; and token binds the whole thing to one stream key.

Opaque: build one with start, take one off a StreamEvent, round-trip it through encode/decode, and never do arithmetic on it.

decode (classmethod)#

decode(value: str | None, *, key: str) -> StreamCursor

Parse encode's form for key, refusing anything else.

An empty or absent value is "from the beginning", which is what an EventSource sends on its first connection.

Everything else is refused rather than repaired, for the reason wreath.log.Cursor.decode refuses: the value arrives in a header somebody else wrote. str.isdigit rather than int(...) in a try, because int accepts leading whitespace, a sign, and Unicode digits from other scripts -- int("７") really is 7, and none of those are cursors this ever emitted.

Raises:

• ValueError — naming which check failed. A cursor for another stream
• is called out by name rather than folded into "malformed",
• because the two have different causes — one is a corrupted
• header, the other is a client with its keys crossed.

encode#

encode() -> str

The form a client echoes back as a Last-Event-ID.

start (classmethod)#

start(key: str) -> StreamCursor

The cursor before the first row of key.

StreamEvent#

One thing a reader has to say, whether or not it was a row.

chunk is payload. end, error and cancelled are the three ways a stream finishes and are rows too. superseded and timeout are synthesised by Streams.follow and carry the cursor of the last real row, so a client that stores the id it last saw is never handed one it cannot resume from.

StreamEvent(kind: str, cursor: StreamCursor, fence: int, index: int, data: bytes = b'', detail: str = '') -> None

as_dict#

as_dict() -> dict[str, Any]

A JSON-safe form, for the WebSocket transport and for tests.

data is the UTF-8 text where the payload is text and base64 under data64 where it is not, for the reason as_sse gives.

as_sse#

as_sse() -> ServerSentEvent

This event as one Server-Sent Event.

A chunk whose bytes are valid UTF-8 is event: chunk with the text as data; one that is not is event: chunk64 with base64, because SSE data is text and silently replacing an undecodable byte would corrupt a payload rather than report it. wreath.rooms makes the same call on the same evidence, and the two spell it the same way on purpose.

id (property)#

id: str

This event's Last-Event-ID.

terminal (property)#

terminal: bool

Nothing further will arrive on this stream.

Broader than "the producer finished": a timeout ends the reader while the producer may still be running, and the distinction is exactly what a client needs in order to decide between reconnecting and giving up. kind says which.

StreamHandle#

What Streams.start hands back: an id to attach to, and where it is.

A StreamingResponse rather than a plain object, because wreath.app's coercion ends in a closed isinstance check and a duck-typed value with a correct __call__(send) dies there with handlers must return a response-compatible value. Subclassing also picks up the deferred-cleanup contract that releases a database connection a handler borrowed.

The body is one JSON object, so the length is known and is sent -- a streaming response is the shape this needs, not the framing.

StreamHandle(*, key: str, task_id: str, state: str = 'queued') -> None

as_dict#

as_dict() -> dict[str, Any]

set_cache_control (inherited from StreamingResponse)#

set_cache_control(policy: CacheControl) -> None

Replace every cache-control header on this response with policy.

Identical in behaviour to Response.set_cache_control(), and defined again here only because StreamingResponse is not a Response subclass. On an SSEResponse this is also how the constructor's no-cache is displaced, since dropping first is what keeps the stricter policy from landing behind it.

set_cdn_cache_control (inherited from StreamingResponse)#

set_cdn_cache_control(policy: CacheControl) -> None

Replace CDN-Cache-Control with an RFC 9213 targeted policy.

set_link_templates (inherited from StreamingResponse)#

set_link_templates(*templates: LinkTemplate) -> None

Replace Link-Template with validated RFC 9652 templated links.

StreamWriter#

The producer's end of one stream, for one attempt.

Handed to a registered producer as its first argument. write buffers and lets wreath.log's _Buffer decide when to issue a statement, so a token-at-a-time producer costs one multi-row INSERT per flush rather than one round trip per token -- the write amplification this whole design is challenged on, and the reason append_many exists.

Every row carries this attempt's fence. That is what a reader uses to skip a superseded range, and it is why nothing here has to be idempotent.

StreamWriter(log: PostgresLog, key: str, *, fence: int, resume_from: int, storage_key: str | None = None) -> None

abandon#

abandon() -> int

Give up on what is buffered, counting it as dropped.

What a cancelled attempt calls. The rows are lost either way; this makes the loss a number on PostgresLog.dropped rather than a buffer nobody looks at again.

fail#

async fail(detail: str) -> None

Close a failed stream with a bounded diagnostic.

fence (property)#

fence: int

This attempt's fence, from wreath.jobs.JobContext.fence.

finish#

async finish() -> None

Close a successfully completed stream.

flush#

async flush() -> int

Write everything buffered now, returning how many rows landed.

flushes (property)#

flushes: int

Statements' worth of batches this attempt has issued.

written / flushes is chunks-per-flush, which is the number this design's write amplification is measured in.

index (property)#

index: int

The idx the next row will carry.

key (property)#

key: str

The stream this writes to.

pending (property)#

pending: int

Rows buffered and not yet written.

write#

async write(data: str | bytes) -> None

Append one chunk, flushing when the policy says to.

The flush policy is checked here and nowhere else, which matters for a producer that writes rarely. There is no timer task behind this: the Flush(every=...) threshold is read on the next write, so a producer emitting one line per tool call leaves that line in the buffer until the call after it. At token rates that delay is a few milliseconds and invisible; at one chunk a minute it is a minute. A producer whose gaps are long calls flush itself, exactly as a caller of wreath.log.PostgresLog.buffered does -- the buffer is driven by its owner, and a background task per stream would be a task per stream.

Args:

• data — str is encoded UTF-8; bytes travel unchanged. A reader
• delivers bytes and decides how to frame them, so a producer that
• emits binary is not forced through a text round trip.

written (property)#

written: int

Rows offered to the buffer by this attempt.

Streams#

Durable streams over one job queue and one chunk log.

Args:

• jobs — a wreath.jobs.JobRunner. Producers register as tasks on it, so a
• stream inherits retries, fencing, leases, the deadline, and
• cancellation -- there is deliberately no second execution path.
• log — a wreath.log.PostgresLog over declaration().
• poll — seconds between reads while tailing.
• idle — seconds a reader waits with nothing new before it says timeout.
• on_retry — "supersede" (keep the replaced rows, skip them on read) or
• "truncate" (delete them when the retried attempt starts). Both
• tell the reader; only one keeps the bytes.
• started_capacity — how many started keys are remembered for the
• attached-versus-started count.

Streams(*, jobs: Any, log: PostgresLog, poll: float = 0.05, idle: float = 30.0, on_retry: str = 'supersede', started_capacity: int = 4096) -> None

attach#

attach(key: str, *, since: StreamCursor | str | None = None, idle: float | None = None, poll: float | None = None, authorize: Callable[[str], bool] | None = None, public: bool = False) -> Response | SSEResponse

An SSE response replaying from since and then tailing key.

Synchronous, like wreath.progress.progress_stream: it returns a response rather than awaiting one, so a refusal is a plain response and the stream only opens once the refusals are past.

authorize(key) -> bool decides whether this caller may read that stream. It matters: a stream key is usually a conversation id, and without a guard whoever can guess one can read a generation. A refusal answers 404, identical to an unknown key, because a distinct 403 would confirm which keys exist.

A since that does not parse answers 400 naming why, rather than starting from zero -- a silent restart would hand the client a second copy of everything it already rendered, which is the failure mode this whole module exists to remove.

Pass public=True only when every stream key is intentionally public.

cancel#

async cancel(key: str, *, reason: str = 'cancelled') -> bool

End key now: write the terminal row, then cancel the job.

Terminal row first, deliberately. The invariant is that attaching to a cancelled stream returns the terminal record rather than hanging, and a cancel that fenced the job and then failed to write would leave every reader idling until its own timeout.

There is no third cancellation path: the job's fence is bumped, so whatever the running attempt does next matches no row, and its late chunks sit behind a terminal record no reader passes.

Returns:

• Whether a queue row moved. False means the job had already
• finished or dead-lettered -- the terminal row is written either way,
• because a reader still needs to be let go.

component#

component() -> Any

This module's claim on the wreath schema.

counters#

counters() -> Any

This registry's counters, for wreath.metrics.collect.

started against attached is the pair worth watching, and a divergence between them is only visible if both are scraped.

follow#

follow(key: str, *, since: StreamCursor | str | None = None, idle: float | None = None, poll: float | None = None, limit: int = 512, _storage: str | None = None) -> AsyncIterator[StreamEvent]

Replay from since, then tail, until the stream ends or goes quiet.

The one reader. SSE (attach), WebSocket (push_stream) and MCP's notification stream are three framings of this generator, so there is one cursor discipline and one supersede rule rather than three that drift.

The supersede rule, in full. A stream's rows are ordered by the log, which is commit order, and every row carries the fence of the attempt that wrote it. The reader holds the highest fence it has seen for this stream -- seeded from the table, so a client arriving after a retry never receives the replaced range at all -- and then:

• a row below that fence is skipped. Not hypothetical: a worker whose lease expired is still alive and still flushing while its replacement produces, and its late rows land after the newer ones;
• a row above it means a retry, so superseded is emitted before the new content and the fence moves up. A client concatenating a replaced range renders duplicated text and blames the model, so it is told rather than left to notice.

Args:

• since — a StreamCursor, its encoded form, or None for the start.
• A string is validated as an index into this stream; anything
• else raises.
• idle — seconds with nothing new before timeout ends the stream.
• poll — seconds between reads.
• limit — rows one replay page carries.

Raises:

• ValueError — since is not a cursor for key.

log (property)#

log: PostgresLog

The chunk log. Its component, retention_pass and dropped are the schema, the retention walk, and the honest completeness number.

producer#

producer(kind: str, *, retries: int = 2, timeout: float | None = None) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]

Register kind's producer as a durable task on the job runner.

The decorated function is producer(stream: StreamWriter, *args), and it runs detached from any connection -- in a worker, under a lease, with a fence. That is the property everything else here depends on: it keeps producing when nobody is attached, and it is re-run by an ordinary job retry when the worker holding it dies.

Registration happens at import time in every process, which is why the producer is not simply passed to start: the worker that runs it never calls start, so a callable handed over there would have registered nowhere that matters.

retries defaults low. A stream retry re-runs the producer, and for a priced external call that is a second charge; the fence makes it correct, not free.

Args:

• kind — names the task, so it is a bounded SQL-safe identifier.
• retries — retries after the first attempt.
• timeout — seconds one attempt may run, or None for the runner's
• default. Must be shorter than the lease; JobRunner.task
• checks that.

registered#

registered(kind: str) -> bool

Whether kind has a producer in this process.

retention_pass#

retention_pass(*, name: str = 'stream_chunks', **options: Any) -> Any

The counted walk that executes the chunk log's retention window.

start#

async start(kind: str, *, key: str, producer: Callable[..., Awaitable[None]] | None = None, args: Sequence[Any] = (), tenant: str = '') -> StreamHandle

Enqueue kind's producer for key and hand back something to attach to.

Idempotent in the key: the enqueue carries key as its dedup key, so a second start for a stream already in flight returns the handle of the first rather than starting a second producer. One key is one stream -- compose it (f"{conversation}:{turn}") when a conversation has several.

producer= is accepted for the shape the call site reads best, and is checked rather than used: a callable that is not the one registered under kind is refused by name, because silently preferring one of two would mean the worker ran the other.

Raises:

• ValueError — kind has no registered producer in this process, or
• producer= is not that producer, or key is not a usable key.

stats#

stats() -> dict[str, int]

Every counter this registry keeps, by name.

started against attached is the one worth an alert. The producer runs whether or not anybody is reading, which is the point of the design; an application where the two numbers diverge is paying a model for output nobody asked for twice. Both are this worker's, the way RoomRegistry.members is.

writer#

async writer(key: str, *, fence: int, attempt: int = 1) -> StreamWriter

Join a job owned by another subsystem to this stream log.

check_stream_attachment#

check_stream_attachment(streams: Streams, *, ratio: float = 0.5) -> list[str]

Findings about streams that were produced and never read.

In wreath.doctor's shape -- a list of human-readable findings, empty when there is nothing to say -- because this is a doctor question and not a metric: "the producer runs whether or not anyone reads it" is the design, and "most of them are never read" is a bill.

Args:

• ratio — the fraction of started streams that must have been attached to
• before this says nothing. Below it, the finding names both numbers.

declaration#

declaration(table: str = 'stream_chunks', *, schema: str = 'wreath', retain: float = 3600.0, flush: Flush | None = None) -> Log

The wreath.log shape a chunk buffer needs.

Four payload columns beside the log's own (xid, seq):

• fence -- the attempt that wrote this row, straight off wreath.jobs.JobContext.fence. It is what makes a retry idempotent to a reader without the producer having to be.
• idx -- this row's position within its attempt, from zero. Not a cursor: the cursor is the log's, and two attempts both have an idx 0. It is what a client counts, and what resume_from reports to a producer that can pick up where the last attempt stopped.
• kind -- chunk, or one of the three terminal kinds.
• body -- the payload, NULL on a terminal row, where the reason travels in detail instead.

Args:

• table — the backing table; interpolated, so a plain identifier.
• schema — where it lives. Wreath's own furniture goes in wreath.
• retain — seconds a chunk lives. KEEP_FOREVER is refused, by name:
• this table is delivery, and a delivery buffer that never expires is
• a transcript of prompts and completions that nobody declared, in a
• place with no erasure story.
• flush — the buffering policy StreamWriter batches under.

Raises:

• ValueError — on retain=KEEP_FOREVER.

push_stream#

async push_stream(websocket: Any, streams: Streams, key: str, *, since: StreamCursor | str | None = None, idle: float | None = None, poll: float | None = None, authorize: Callable[[str], bool] | None = None, public: bool = False) -> bool

Push key's events as JSON text frames over an accepted WebSocket.

The same reader as attach, framed for a socket that already exists -- which is the case wreath.rooms leaves you in, where a client is already connected for something else and a second HTTP request would be a second connection for no reason. Each frame carries id, so a client that reconnects resumes exactly as an EventSource would. Pass public=True only when every stream key is intentionally public.

wreath.progress#

Report and expose the progress of a long-running task.

A slow job — an import, a rollup, a batch call — wants to tell the client "42%, processing invoices". Wreath gives you the transport (SSE, WebSockets, JSON) and durable jobs give you the state; this is the small piece in between: a task writes progress to a bounded in-process registry, and a status endpoint or a stream reads it.

progress = ProgressRegistry()

@app.post("/imports") async def start_import(request): task_id = new_id() asyncio.create_task(run_import(progress.reporter(task_id), ...)) # your runner return {"task_id": task_id}

@app.get("/imports/{task_id}/status") async def status(request): return status_response( progress, request.path_params["task_id"], authorize=lambda task_id: request.user.owns_task(task_id), )

@app.get("/imports/{task_id}/stream") async def stream(request): return progress_stream( progress, request.path_params["task_id"], authorize=lambda task_id: request.user.owns_task(task_id), )

Inside the task, reporter.update(42, "processing invoices") / reporter.done() / reporter.fail(exc).

Across workers. The registry is in-process, which is exactly wrong for the case that matters most: the durable job runs on worker 3 and the browser is connected to worker 1. Give the registry the message bus and every report reaches every worker, so whichever one holds the stream can answer it:

progress = ProgressRegistry(app.messaging("bus", database="app"))

No Redis — the bus is the database you already have. Delivery is at-most-once, as ephemeral fan-out is: a worker that missed an update gets the next one, and percentages are a running commentary rather than a ledger.

The natural pairing is wreath.jobs.JobRunner.launch, which uses the job id as the task id and lets the runner set the terminal states itself.

Progress#

A snapshot of a task's progress.

Progress(percent: float, message: str = '', state: str = 'running', error: str | None = None) -> None

as_dict#

as_dict() -> dict[str, Any]

ends_stream (property)#

ends_stream: bool

Nothing further will arrive on the stream that yielded this.

Broader than terminal, which is about the task. A stream also ends when the registry forgets an entry, when an id never appears, or when a watch budget runs out -- and a client needs to tell those apart from the connection simply dropping.

terminal (property)#

terminal: bool

The task finished: done or failed.

ProgressRegistry#

A bounded map of task id -> latest Progress.

Pass the message bus to make it fleet-wide: every report is applied here and published once, and every other worker applies what it receives without relaying it onward. Omit it for single-process work, which is the right default for tests and one-worker deployments.

ProgressRegistry(bus: Any = None, *, max_tasks: int = 4096, ttl: Any = 86400, channel: str = 'wreath_progress') -> None

get#

get(task_id: str) -> Progress | None

report#

report(task_id: str, percent: float, message: str = '', *, state: str = 'running', error: str | None = None) -> None

reporter#

reporter(task_id: str) -> ProgressReporter

A handle bound to task_id to hand to the running task.

stream#

stream(task_id: str, *, interval: float = 1.0, max_duration: float | None = None)

Yield each new Progress for task_id until it is terminal.

Polls every interval seconds (thread-safe, no cross-task signalling); stops after a done/failed state, once the entry is gone, once a task that never appeared has been waited for long enough, or once max_duration seconds have passed.

An unknown id receives a short grace period for the race where a client starts watching just before task registration, then the stream emits an unknown event and closes. These helpers carry no authentication of their own; authorization belongs on the route that exposes the stream.

Every stream ends with an event saying why, so a close is never ambiguous with a dropped connection: done/failed when the task finished, expired when the registry forgot the entry mid-stream, unknown when the id never appeared, detached when max_duration ran out while the task was still going. The last three are Progress.ends_stream but not Progress.terminal in the expired/unknown sense the task never reached -- the registry stopped being able to answer, which is not the same as the work stopping.

ProgressReporter#

A task's write handle for one task_id (from registry.reporter).

ProgressReporter(registry: ProgressRegistry, task_id: str) -> None

done#

done(message: str = 'done') -> None

fail#

fail(error: object, message: str = '') -> None

update#

update(percent: float, message: str = '') -> None

progress_stream#

progress_stream(registry: ProgressRegistry, task_id: str, *, interval: float = 1.0, max_duration: float | None = None, authorize: Callable[[str], bool] | None = None, public: bool = False) -> SSEResponse | Response

An SSE response streaming progress events until the task is terminal.

authorize and max_duration are status_response's and ProgressRegistry.stream's respectively; a refused caller gets the same 404 a missing task does, before any stream is opened. Pass public=True only when every task in the registry is public.

push_progress#

async push_progress(websocket: Any, registry: ProgressRegistry, task_id: str, *, interval: float = 1.0, max_duration: float | None = None, authorize: Callable[[str], bool] | None = None, public: bool = False) -> bool

Push authorised progress as JSON frames until terminal; opt public explicitly.

status_response#

status_response(registry: ProgressRegistry, task_id: str, *, authorize: Callable[[str], bool] | None = None, public: bool = False) -> Response

A JSON status for task_id (404 if unknown, expired, or refused).

authorize(task_id) -> bool decides whether this caller may watch that task. It matters more than it looks: wreath.jobs.JobRunner.launch makes the task id the job id, which is a sequence, so without a guard the ids are countable and every task's state, message, and error text is readable by whoever counts. The predicate is synchronous on purpose -- this function is not a coroutine, and a handler that needs to await something can do it before calling.

A refusal answers 404, identical to an unknown id: a distinct 403 would confirm which ids exist, which is most of what enumeration wants. Pass public=True only when every task in the registry is public.

wreath.jobs#

Durable, at-least-once background jobs backed by PostgreSQL.

A replacement for Celery/arq for teams already running Postgres: enqueue jobs (transactionally, in the same commit as your business writes), and a supervised pool of workers claims them with FOR UPDATE SKIP LOCKED + fencing tokens, retries with backoff, and dead-letters on exhaustion. NOTIFY is used only as a latency doorbell — correctness never depends on a notification arriving, the workers also poll.

Delivery is at-least-once. A crash between a job's side effect and its completion UPDATE yields a re-run on lease expiry. Make handlers idempotent; pass key= to JobRunner.enqueue for exactly-once enqueue (a unique index drops duplicates) and use it to guard non-idempotent side effects.

Multi-tenancy note (design 01 §5): jobs live in one dedicated system schema with a tenant column, never relying on search_path for isolation — a database-global NOTIFY name would otherwise wake the wrong tenant's workers.

JobContext#

Handed to a job handler as its first argument.

JobContext(job_id: int, task: str, attempt: int, fence: int, tenant: str, key: str | None, progress: Any = None, trace_context: str | None = None) -> None

report#

report(percent: float, message: str = '') -> None

Tell whoever is watching how far along this job is.

A no-op when the runner has no progress registry, so a handler can report unconditionally. Only progress -- the runner sets done and failed itself, because it is the thing that actually knows whether the job finished, is about to retry, or was dead-lettered.

task_id (property)#

task_id: str

This job's progress key. The job id, so there is one identifier.

JobRow#

One queue row as an operator reads it. No handler, no runner, no lease.

JobRow(id: int, queue: str, task: str, tenant: str, state: str, attempts: int, max_attempts: int, run_at: Any, updated_at: Any, last_error: str | None, trace_context: str | None = None) -> None

as_dict#

as_dict() -> dict[str, Any]

trace_id (property)#

trace_id: str | None

The 32-hex trace id, for pasting into a tracing UI or wreath doctor.

JobRunner#

A named queue of durable jobs on one application database.

Obtain via wreath.Wreath.jobs. Register handlers with task, enqueue with enqueue, and schedule recurring work with schedule. The runner is a supervised service — its workers, sweeper, and scheduler run for the process lifetime.

JobRunner(database: Any, *, name: str, workload: str = 'write', concurrency: int = 8, lease: Any = 30.0, poll_interval: Any = 5.0, schema: str = 'wreath', batch: int = 1, progress: Any = None, attempts: Any = None) -> None

cancel#

async cancel(job_id: int | None = None, *, key: str | None = None, reason: str = 'cancelled', tenant: str = '') -> bool

Stop a queued or leased job. Returns whether a row moved.

Name it by id, or by the key an enqueue/launch deduplicated on -- the second is what a caller has when the id belongs to whichever worker enqueued first.

The fence is bumped, and that is the whole mechanism. There is no signal to a worker in another process and this does not invent one: the row goes to dead, its fence moves, and the running attempt's completion UPDATE -- which is already WHERE id=$1 AND fence=$2 -- matches nothing when it lands. So a cancelled job stops being retried immediately, and the attempt in flight stops being able to say anything about the queue. What it does not do is interrupt that attempt's side effects, for the same reason a lease expiry does not: nothing here can reach into another process's event loop, and pretending otherwise is how a "cancelled" job charges a card.

A job that has already reached done or dead is left alone and answers False; cancelling something that finished is not an error, it is a race the caller lost.

Raises:

• ValueError — neither or both of job_id and key were given.

component#

component() -> Any

This runner's claim on the wreath schema.

The queue's tables are wreath's furniture, not the application's data model, so they live in the wreath schema and never appear in the application's migration artifact -- nobody declared a job queue. Wreath collects this during lifespan and brings it up to date before any worker starts; a deployment that cannot grant CREATE SCHEMA applies the same statements by hand and is refused at startup, by name, if it has not.

Statements are individual rather than semicolon-joined because the driver prepares every statement and PostgreSQL refuses cannot insert multiple commands into a prepared statement. See wreath.schema.

counters#

counters() -> Any

This runner's counters, for wreath.metrics.collect.

deadline_for#

deadline_for(task: str) -> float

Seconds task may run before it is cancelled.

doorbell_reconnects (property)#

doorbell_reconnects: int

Doorbell connections lost, plus every failed attempt to open one (including the attempt made on the startup path). Correctness never depended on the doorbell, which is exactly why losing it is invisible without a number: the runner keeps working at poll_interval.

There is deliberately no companion handler_errors here. The bus keeps one because its pump dispatches to subscriber callbacks; this pump only sets an event, so a counter for user-code failures could never be anything but zero and would imply a distinction the runner cannot make.

drain#

async drain(deadline: float) -> None

drive#

drive(walk: Any, *, cron: str | None = None) -> str

Run a ChunkedPass on this queue. Returns its task name.

One registration path rather than a registry the runner has to reconcile at startup: this registers the task that runs a shift, arranges for the ledger row to be seeded on first contact with the database, adds the schedule when one is given, and enqueues the first shift when the runner starts. The dependency is visible at the call site because the job runner is literally the thing that drives the walk:

jobs.drive(normalize_grades)
jobs.drive(purge_replays, cron="*/5 * * * *")

A shift must be shorter than the lease, and this is where that is checked. There is no heartbeat: a handler still running when its lease expires is reclaimed by the sweeper, picked up by a second worker, and re-claimed again when the first one's fenced completion matches nothing. A pass survives that -- the cursor's compare-and-swap serialises the duplicates -- but it must not rely on surviving something it can simply not do.

A recurring pass needs cron=: nothing else would start its next cycle, and a pass that quietly stops after one cycle is worse than one that refuses to be declared.

enqueue#

async enqueue(task: str, *args: Any, tx: Any = None, run_at: Any = None, key: str | None = None, tenant: str = '', max_attempts: int | None = None, priority: int = 0, coalesce: bool = False) -> int | None

Insert a job. Returns its id, or None if a key deduplicated it.

Pass tx (an open connection.transaction()) to enqueue atomically with your business writes — the job becomes visible only if that transaction commits (exactly-once enqueue). key sets an idempotency key: a second enqueue with the same (queue, tenant, key) is dropped.

priority orders the claim: ready jobs are taken priority DESC, run_at, so a higher number goes first. It is a lane, not a promise — a strict priority queue starves its low lane by definition, and sizing for that stays the caller's.

coalesce changes what a repeated key does. By default the second enqueue is dropped, which is right for "run this once" and wrong when the second call carries something the first did not: the arguments are replaced, the run time becomes the earlier of the two, and the priority the higher. Only while the row is still ready — a leased job is being worked, and rewriting its arguments underneath the worker would hand it inputs it never read. Returns the existing id rather than None, because the work is still pending and the caller asked for it.

The calling request's trace context rides the row when the schema has somewhere to put it, so the job names the request that caused it. It rides the row and never the NOTIFY: that payload is a doorbell, is capped at 8000 bytes, and is deliberately empty.

launch#

async launch(task: str, *args: Any, tx: Any = None, run_at: Any = None, key: str | None = None, tenant: str = '', max_attempts: int | None = None) -> TaskHandle

Enqueue task and return a handle the client can watch.

The long-mutation shape: a request that cannot finish in a request enqueues durable work and hands back an id instead of a timeout:

@app.post("/herd/imports")
async def start_import(request, path: str):
    return (await jobs.launch("import_herd", path)).as_dict()

@app.get("/herd/imports/{task_id}/stream")
async def watch(request):
    return progress_stream(
        jobs.progress,
        request.path_params["task_id"],
        authorize=lambda task_id: request.user.owns_task(task_id),
    )

The job id is the task id, so there is one identifier rather than two to correlate. With a progress registry configured the task is seeded as queued right here, so a client that starts polling immediately sees a pending task rather than a 404 it will read as a failure.

A key that deduplicates does not lose the caller: the surviving row is looked up and its handle returned, so submitting the same work twice yields the same task to watch rather than nothing at all. If that row is gone by the time it is read -- purged after completing, in the window between the conflict and the lookup -- there is nothing to watch and JobVanished is raised rather than a handle whose id is the string "None".

name (property)#

name: str

progress (property)#

progress: Any

The registry this runner reports to, for the status/stream endpoints.

purge#

async purge(*, older_than: float) -> None

Delete finished rows older than older_than seconds.

Nothing calls this for you, for the same reason nothing purges wreath.store: a background sweep duplicates across workers and swallows its own failures. Run it from a scheduled job.

Only done and dead rows go: a dead one has exhausted its attempts and is a record rather than work, and keeping either forever is what makes launch(key=...) eventually raise JobVanished -- the retention this table always assumed and never had.

schedule#

schedule(task: str, *, cron: str | None = None, recurrence: Recurrence | None = None, args: tuple[Any, ...] = (), tenant: str = '', misfire: str = 'skip') -> None

Enqueue task on a repeating schedule. Idempotent across instances.

Every app instance runs the scheduler, but each firing's enqueue carries a deterministic key so the unique index makes exactly one row win — no leader election needed.

cron= is a five-field expression read in UTC, which is what it has always meant. recurrence= takes a wreath.temporal.Recurrence, which carries its own zone:

runner.schedule("rebalance", recurrence=Recurrence.cron("0 3 * * *", tz=depot_tz))

Pass one or the other. A zone is the difference between "03:00" and "03:00 for half the year and 04:00 for the other half", and a depot operator reading a schedule means the former.

schema_sql#

schema_sql() -> str

DDL for the jobs table + indexes, semicolon-joined.

A derivation of component(), not a second copy. It was both for one stage of this work -- the same sixteen columns written out twice -- which is two spellings of one truth and exactly the shape this repository treats as a defect. Retained for a caller applying the DDL itself; wreath schema sql is the supported spelling.

start#

async start(supervisor: Any) -> None

stats#

stats() -> dict[str, int]

Every counter this runner keeps, by name. See MessageBus.stats.

task#

task(name: str, *, retries: int = 5, backoff: str = 'exp', backoff_base: float = 1.0, backoff_factor: float = 2.0, backoff_cap: float = 3600.0, backoff_jitter: float = 0.2, timeout: float | None = None, query_budget: int | None = None) -> Callable[[JobHandler], JobHandler]

Decorator registering an async handler(ctx, *args) under name.

timeout is how long the handler may run before it is cancelled and the attempt fails; None takes the runner's default, DEADLINE_FRACTION of the lease. It must end inside the lease, and that is checked here.

query_budget is how many times one model may be hydrated within a single attempt before that is a defect. Declaring it makes the attempt fail at the query that crossed the line, with a traceback naming the loop -- which is the only way to find an N+1 in work nobody is watching. Leave it unset and the attempt is merely observed: counted, and reported if a guard exists in this process, but never failed. That default is deliberate. A guard that raises inside a durable job converts a slow job into a failed one and then into a retry storm, and wreath does not know whether five hundred queries was the defect or the point of the task.

There is no heartbeat, so a handler still running when its lease expires is reclaimed by the sweeper and started again by another worker — two copies of the same job, doing the same work, and the fence only stops the loser's bookkeeping from landing. That was the whole cost of an unbounded handler: not a stuck worker, a second charge on the card. drive() has always refused a pass whose shift could outlast the lease for this reason; an ordinary task simply had no bound to check.

JobVanished#

A deduplicated JobRunner.launch found no row to hand back.

The insert conflicted -- so the work was already enqueued -- but the surviving row was gone by the time it was read, which is what a retention sweep over completed jobs looks like from here. There is no id to watch, and inventing one is worse than saying so: a task id of "None" 404s on the status endpoint and streams forever on the SSE one, and the client cannot tell that from a job that failed. Re-launch (nothing is holding the key any more) or treat the earlier run as finished.

TaskHandle#

What a caller gets back from JobRunner.launch.

Small on purpose: an id to watch and the state at hand-back time. Everything after that comes from the progress stream.

TaskHandle(task_id: str, state: str = 'queued') -> None

as_dict#

as_dict() -> dict[str, Any]

read_jobs#

async read_jobs(connection: Any, *, schema: str = 'wreath', states: tuple[str, ...] = ('dead',), queue: str | None = None, limit: int = 50) -> list[JobRow]

Rows from the queue table, newest first. For wreath jobs and forensics.

Defaults to dead because that is the state somebody goes looking for: a job that retried until it ran out, hours after the request that enqueued it finished. states=() reads every state.

Selects trace_context only where the column is there, so this runs against a database still on the schema version before propagation and reports every row with no trace rather than failing. A build reading an older schema and a job enqueued outside a traced request both give None; telling them apart is wreath schema check's job, not this one's.

wreath.workflows#

Durable multi-step workflows: ordered steps, recorded progress, compensation.

wreath.jobs gives you one durable unit of work — claimed with FOR UPDATE SKIP LOCKED, fenced, retried, dead-lettered. That is the right shape for "send this email" and the wrong one for "reserve the stock, charge the card, book the courier, and if the courier fails, refund the card and release the stock." The second shape is a saga, and the thing that makes it hard is not the ordering — it is that a worker can die between any two steps.

checkout = Workflow("checkout")

@checkout.step(compensate=release_hold) async def reserve_stock(context): return await inventory.hold(context.results["order_id"])

@checkout.step(compensate=refund) async def charge_card(context): return await payments.charge(context.results["order_id"])

await checkout.run(store=store, key=f"checkout:{order_id}")

Three properties are the whole point, and each one is a defect people ship without it:

• A completed step never runs twice. Each step's return value is recorded before the next one starts, so resume re-enters at the first step with no record. A plain for loop over callables re-runs everything from the top, and every non-idempotent side effect happens again.
• A failure undoes what succeeded, newest first. Reverse order is not cosmetic: step 3's undo routinely depends on state step 2 established. The step that raised is not compensated — it did not complete, so there is nothing of its to undo, and running an undo against state that was never established is how a recoverable failure becomes a corrupt one.
• A compensation that fails is counted, not swallowed. Compensation runs when something has already gone wrong, which is exactly when except Exception: pass is most tempting and most damaging: it leaves a half-compensated saga with no signal. The remaining undos still run, and Outcome.compensation_errors says how many did not.

One instance key runs once. key= is the same guarantee JobRunner.enqueue(key=...) gives — a second run for a key that already finished returns the first outcome rather than re-executing it.

Renaming a step of a live instance is refused. Completion is recorded per step name, so a rename would leave a record matching nothing and a resume would silently redo work and report success. resume compares the recorded step list against the definition and raises WorkflowDefinitionChanged naming the step it cannot account for. That is the one failure in this module that would otherwise be invisible.

Storage is a protocol. InMemoryWorkflowStore is for tests and single-process work; PostgresWorkflowStore is durable, and it lives in its own system schema with a tenant column for the reason jobs does — never relying on search_path for isolation.

InMemoryWorkflowStore#

Step records in a dict. For tests and single-process work.

Durable only for the life of the process, which is the honest limit: it satisfies the resume contract within a run but cannot survive the crash the contract exists for. Use PostgresWorkflowStore where that matters.

InMemoryWorkflowStore() -> None

begin#

async begin(key: str, workflow: str, steps: Sequence[str]) -> dict[str, Any]

Record the instance and its step list, or return what is already there.

Idempotent, because that is what makes key= an exactly-once start. The step list is recorded here, not derived at resume time, so a later rename is detectable at all.

complete_step#

async complete_step(key: str, name: str, result: Any) -> None

execution#

execution(key: str) -> Any

finish#

async finish(key: str, state: str, compensation_errors: int = 0) -> None

load#

async load(key: str) -> dict[str, Any] | None

uncomplete_step#

async uncomplete_step(key: str, name: str) -> None

Outcome#

What a run produced, whether it completed or compensated.

Outcome(key: str, workflow: str, state: str, results: dict[str, Any] = <factory>, completed: list[str] = <factory>, compensated: list[str] = <factory>, compensation_errors: int = 0) -> None

PostgresWorkflowStore#

Durable step records in PostgreSQL.

One row per (instance, step), plus one row per instance carrying the recorded step list. Both live in a dedicated system schema with a tenant column rather than relying on search_path, which is wreath.jobs's rule and holds for the same reason: schema-qualified state cannot be read by the wrong tenant's worker because someone set a search path wrong.

PostgresWorkflowStore(database: Any, *, schema: str = 'wreath_system', table: str = 'workflow_steps', tenant: str = '', workload: str = 'write') -> None

begin#

async begin(key: str, workflow: str, steps: Sequence[str]) -> dict[str, Any]

Insert the instance if absent, then return its record.

ON CONFLICT DO NOTHING rather than a check-then-insert: two workers starting one key race, and the unique index is the only arbiter that does not have a window between the read and the write.

complete_step#

async complete_step(key: str, name: str, result: Any) -> None

execution#

execution(key: str) -> Any

finish#

async finish(key: str, state: str, compensation_errors: int = 0) -> None

load#

async load(key: str) -> dict[str, Any] | None

schema_sql (classmethod)#

schema_sql(*, schema: str = 'wreath_system', table: str = 'workflow_steps') -> str

DDL for the instance and step tables, semicolon-joined.

CREATE SCHEMA is emitted; CREATE EXTENSION is not, and neither is any search_path assignment -- every relation here is schema-qualified at the point of use.

uncomplete_step#

async uncomplete_step(key: str, name: str) -> None

Step#

One step: a name, the work, and optionally how to undo it.

Step(name: str, run: Callable[..., Any], compensate: Callable[..., Any] | None = None, query_budget: int | None = None) -> None

StepContext#

What a step is handed. results carries every earlier step's return value.

Threading results through the context rather than through arguments is what lets a resumed run reconstruct them from the store: a step that reads context.results["make_id"] gets the recorded value after a crash, not a recomputed one.

StepContext(key: str, workflow: str, results: dict[str, Any] = <factory>) -> None

UnknownWorkflowInstance#

resume was given a key that was never started.

Workflow#

An ordered series of steps, each optionally paired with its undo.

Declaration order is execution order; there is no dependency graph, because a saga's undo chain is inherently a sequence and a graph would make "newest first" ambiguous.

Args:

• name — Recorded with every instance, so a stored record says which
• workflow it belongs to.

Workflow(name: str) -> None

resume#

async resume(*, store: Any, key: str, progress: Any = None, recorder: Any = None) -> Outcome

Carry a started instance on from its first unrecorded step.

Raises:

• UnknownWorkflowInstance — key was never started.
• WorkflowDefinitionChanged — the stored step list names a step this
• workflow no longer declares.

run#

async run(*, store: Any, key: str | None = None, compensate: bool = True, progress: Any = None, recorder: Any = None) -> Outcome

Start (or rejoin) an instance and execute it to completion.

A key that already finished returns the recorded outcome without executing anything -- exactly-once start. Omitting key generates one, which is the right choice only when nothing will ever need to resume it.

Args:

• store — Anything satisfying the store protocol.
• key — The instance identity. Generated when omitted.
• compensate — Run the undo chain when a step raises. False leaves the
• completed steps recorded so resume can carry on from them --
• which is what you want for a retryable failure, as opposed to
• one that invalidates the whole saga.
• progress — A wreath.progress.ProgressRegistry to report step
• completion to.
• recorder — A wreath.recording.WorkflowStepRecorder, or None. It
• arms nothing on its own -- a policy with no triggers records
• nothing -- and it is passed per run rather than held on the
• workflow because arming is an operator's decision about a
• deployment, not a property of the saga's definition.

Raises:

• Exception — whatever the failing step raised, re-raised after the undo
• chain has run.

status#

async status(*, store: Any, key: str) -> Outcome | None

What the store knows about an instance, or None if it never started.

The way compensation_errors is read. A run that compensated re-raises the step's own exception -- callers expect the error they would have got without a saga around it -- so the summary cannot come back as a return value. It is recorded instead, which is the more useful place anyway: a half-compensated saga has to be findable later, by someone who was not holding the traceback.

step#

step(handler: Callable[..., Any] | None = None, *, compensate: Callable[..., Any] | None = None, query_budget: int | None = None) -> Any

Register a step. Usable bare or with compensate=.

The step's name is the function's name, which is what a stored record keys on -- see WorkflowDefinitionChanged for what happens when it changes underneath a live instance.

query_budget is how many times one model may be hydrated inside this step before that is a defect; crossing it raises from the query that did. It covers the step's compensation too, which is the half nobody exercises -- an undo runs only when something has already gone wrong, so an N+1 in one is discovered during an incident or not at all. Omitted, the step is observed rather than bounded; see wreath.jobs.JobRunner.task for why that is the default.

steps (property)#

steps: tuple[Step, ...]

The declared steps, in order.

WorkflowDefinitionChanged#

A stored instance's steps no longer match the workflow's definition.

Raised by resume rather than absorbed, because the alternative is a resume that re-runs completed work and reports success. The message names the step that could not be accounted for.

WorkflowError#

Base class for every refusal this module raises.

wreath.entity#

One name, one owner, one mailbox — addressing a thing that lives on a worker.

Six places in this tree already answer "exactly one process owns this right now", each correctly, none sharing:

• SingletonRunner holds an advisory lock for as long as its work runs.
• wreath.jobs leases a row and bumps a fence on every claim.
• wreath.store.PostgresStore.claim returns a row that is the claim.
• wreath.passes takes a row lock as the chunk's first statement.
• _secondfactor.ChallengeStore consumes with one DELETE ... RETURNING.
• wreath.streams reuses the queue's fence, because writing a second one would have been absurd -- its module docstring says so.

That last one is the tell: the concept was already being shared across a module boundary, by prose. This module is the shared spelling, and it adds the thing none of the six could express -- a message aimed at whoever holds the name.

Why that last part matters#

A WebSocket gateway is stateful and sits behind a stateless load balancer. The socket for device abc lives on exactly one worker, and the request that needs to talk to it arrives on any of them. Every system with that shape grows the same three subsystems by hand: a per-connection channel on a broker, an ownership table with a heartbeat, and a correlation map with a timeout.

Ownership is the second. ask is the first and third:

entities = app.entities(database="main", bus="events")   # registers the table

@entities.answers("device")
async def talk_to_device(name: str, payload: dict) -> dict:
    return await SOCKETS[name].request(payload)   # local; we hold this one

# ... on any worker, without knowing which one holds it:
reply = await entities.ask("device", "abc", {"op": "read"}, timeout=seconds(30))

What this deliberately does not do#

It does not replace SingletonRunner. An advisory lock releases the instant its connection drops, which is a better failure detector than a lease -- there is no expiry to wait out. It costs a held connection per lock, which is right for "one process runs this loop" and wrong for a hundred thousand devices. Two mechanisms, two shapes, and folding the better one into the weaker one to save a class would be a downgrade wearing a tidy-up's clothes. Ownership is for the many-names case; SingletonRunner stays for the one-name case.

Delivery is at-most-once and ask can time out. The doorbell is wreath.messaging's ephemeral fan-out, which is a NOTIFY. That is the correct tier for a request/response with a deadline -- a durable queue would replay a question whose asker has already given up -- but it means a timeout is an ordinary outcome and not an error to be surprised by.

A lease does not stop the world. As in wreath.jobs: there is no heartbeat, so a holder still running when its lease expires is superseded by whoever takes the name next. The fence is what stops the loser's bookkeeping landing; it cannot stop the loser's side effects. Read Lease.fence before you write.

EntityRegistry#

Names with owners, and a question that reaches whoever owns one.

One bus channel for the whole registry rather than one per name: a channel per entity is a LISTEN per entity, and the point of this module is the case where there are a hundred thousand of them.

EntityRegistry(database: Any, bus: Any, *, channel: str = 'wreath_entity', table: str = 'wreath_entity', lease: Any = 30.0, max_pending: int = 1024, on_lost: Callable[[str], None] | None = None) -> None

answers#

answers(kind: str) -> Callable[[Answer], Answer]

Register the handler for one kind of name.

kind namespaces the answer, so one registry serves devices and sessions without either handler seeing the other's questions.

ask#

async ask(kind: str, name: str, payload: Any, *, timeout: Any = 10.0) -> Any

Put payload to whoever holds kind:name and wait for their answer.

Raises NotHeld when nothing holds the name -- checked before publishing, because an ephemeral publish to nobody is a silent no-op and a caller waiting the full timeout for that is a bad diagnosis. Raises Unanswered when a holder existed and did not reply in time.

counters#

counters() -> Any

This registry's counters, for wreath.metrics.collect.

held is a gauge and the rest are monotonic; both belong here, because which of the two a number is belongs to whatever renders it.

drain#

async drain(deadline: float) -> None

Release every held name, so a restart does not wait out the leases.

Without this a rolling deploy parks every name for a full lease before another worker may take it, which is a visible outage for exactly as long as the lease -- and the lease is sized for crash detection, where waiting is the only option. A clean shutdown has better information and should use it.

held (property)#

held: frozenset[str]

The names this process currently believes it holds.

hold#

async hold(kind: str, name: str) -> Lease | None

Take or renew kind:name for this process, and keep it renewed.

A held name joins the batch the renewal tick extends, so a caller never writes a keep-alive loop of its own -- which is the loop every application with a lease grows, once per name, at one round trip each.

holding#

holding(kind: str, name: str, *, grace: Any = 0.0) -> AsyncIterator[Lease | None]

Hold kind:name for the block, renewed throughout, released after.

Yields the Lease, or None when another worker holds the name -- the body decides what that means, because refusing and waiting are both reasonable and only the caller knows which.

async with entities.holding("device", name, grace=seconds(2)) as lease:
    if lease is None:
        return
    ...

grace keeps the lease for that long after the block exits instead of releasing immediately. A caller that reconnects inside the window renews rather than re-acquires, so a brief drop does not hand the name to another worker and then take it back -- two handovers and two fence bumps for something that never actually moved. The deferral is applied by the renewal tick, not by a timer per name.

lost (property)#

lost: int

Names that were taken by another worker while this one held them.

Non-zero means leases are expiring under load before the tick renews them -- a slow database, a blocked event loop, or a lease too short for either. It is the one number that says a gateway is quietly shedding ownership, which otherwise looks exactly like clients reconnecting.

outstanding (property)#

outstanding: int

Questions awaiting an answer right now.

ownership (property)#

ownership: Ownership

The lease table underneath, for a caller that only wants ownership.

refusals (property)#

refusals: int

Questions refused because too many were already in flight.

release#

async release(kind: str, name: str) -> bool

Give up kind:name now, and stop renewing it.

renewal_errors (property)#

renewal_errors: int

Renewal attempts that failed before completing a tick.

renewals (property)#

renewals: int

Completed renewal ticks. Flat means the loop has stopped.

schema_owners (property)#

schema_owners: tuple[Ownership, ...]

The objects holding this registry's tables.

Wreath.schema_components collects a claim by asking, and a registry does not own its table directly -- Ownership does. Answering here is what puts wreath_entity in the DDL the application applies at startup; without it the table is emitted by wreath schema sql and created by nothing, which is the exact defect that mechanism exists to prevent.

start#

async start(supervisor: Any) -> None

Begin renewing held names. The wreath.services.Service half.

unrouted (property)#

unrouted: int

Questions whose name had no live holder.

Lease#

Proof that this process held name when the lease was taken.

fence increases every time the name changes hands and not when the same holder renews. That is the wreath.jobs rule, deliberately: a fence that moved on renewal would invalidate the holder's own in-flight work.

Lease(name: str, owner: str, fence: int) -> None

NotHeld#

A name was addressed and no live holder answered.

Distinct from Unanswered: nothing claimed the name, as opposed to something claiming it and not replying in time. The two want different operator responses -- a missing device against a wedged worker -- so they are different exceptions rather than one with a flag.

Ownership#

Which worker holds each name, leased and fenced, on one table.

The claim is a single INSERT ... ON CONFLICT DO UPDATE ... WHERE, so "a row came back" is the claim -- there is no read-then-write window in which two workers both conclude they were first. wreath.store already made that argument for keyed state; this is the same argument for names.

Ownership(database: Any, *, table: str = 'wreath_entity', lease: Any = 30.0, owner: str | None = None) -> None

component#

component() -> Any

This registry's claim on the wreath schema.

database (property)#

database: Any

The database this registry's table lives in. Read by wreath.infra.

hold#

async hold(name: str) -> Lease | None

Take or renew name, or None when someone else holds it.

Idempotent for the current holder: calling it again is the renewal, and the fence does not move. A caller renews well inside the lease, the way wreath.jobs expects a handler to finish well inside its own.

holder#

async holder(name: str) -> str | None

Which worker holds name, or None when nobody live does.

lease_seconds (property)#

lease_seconds: float

How long a claim survives without renewal.

owner (property)#

owner: str

This process's identity, as it appears in the table.

purge#

async purge() -> str

Drop expired rows. Run it from a durable job, as with every store.

release#

async release(name: str) -> bool

Give up name, if this process still holds it.

Scoped to the owner in the WHERE, so a process whose lease already expired and was taken by someone else cannot delete the new holder's row on its way out -- which is the shutdown race a naive delete loses.

release_all#

async release_all() -> int

Give up every name this process holds. Returns how many went.

One statement, for the same reason as renew_all -- a drain that walked its names would make shutdown time proportional to the fleet.

release_many#

async release_many(names: Sequence[str]) -> int

Give up the named leases this process holds, in one statement.

The predicate is IN ($2, $3, ...) with one placeholder per name, not = ANY($1) with an array: the driver refuses to bind a sequence, on purpose -- [1, 2] is equally int4[], int8[] or numeric[], and [] names no element type at all. So this is built per call rather than prepared. That costs a plan, and the alternative costs a round trip per name; this path runs only when a grace period expires, where the list is short and the calls are rare.

renew_all#

async renew_all() -> frozenset[str]

Extend every live lease this process holds. Returns the names kept.

One statement, whatever the worker holds. A renewal loop that issued a statement per name would put a round trip per entity per tick on the database, which is precisely the cost that stops this scaling to the fleet it exists for. WHERE owner = $1 is the whole selector, and the RETURNING set is the answer.

A name absent from the result was taken by someone else while this process was not looking. That is the only way to learn it: there is no heartbeat, and a lease that lapsed under load looks exactly like one that was never held.

schema_sql#

schema_sql() -> str

DDL for the backing table, semicolon-joined.

Unanswered#

A holder was asked and did not reply inside the deadline.

A TimeoutError subclass, because a caller that already treats a slow dependency as a timeout should not have to learn a second name.

wreath.queue#

One bounded in-process queue, for every hand-off that needed one.

A log record going from the projector thread to the writer, a finished trace going to the exporter, a WebSocket frame going to an application task: all of them want the same thing, which is a queue that cannot grow without bound and that says out loud when it has dropped something.

from wreath.queue import Queue

records = Queue(capacity=4096)
records.offer(record)                 # False, and counts, when full
batch = records.drain(limit=256)      # one call, not a pop per item
item = await records.get()            # no Future when an item is already there

Loss is counted, never silent. Offering to a full queue returns False and increments dropped. That is the only policy compatible with the promise a bounded hand-off makes -- bounded memory, bounded latency, and a number an operator can look at when a consumer falls behind. drop_oldest=True evicts the front instead, for a consumer that would rather have the newest; it still counts, because something was lost either way. put_nowait is the other posture: raise QueueFull rather than lose the item.

Safe across threads. Unlike wreath.kv this genuinely crosses them, so the ring is locked. await get() may only be used from one event loop, which is the loop the first waiter parks on; offering from any other thread wakes that loop through call_soon_threadsafe.

The awaiting half is written once here and runs on both arms. A queue that already has an item resolves without suspending the caller and without building a Future -- the native get() hands back a pre-resolved awaitable, and only a genuinely empty queue reaches _blocked below.

Awaiting#

The awaiting half, over whichever ring it is mixed into.

A mixin rather than a base class with the ring built in, because the FIFO ring and the priority heap are separate C-level layouts and a class cannot inherit from both. Queue and PriorityQueue below are the two compositions callers want.

Carries no instance layout of its own -- __slots__ is empty here, and every concrete class that mixes this in must declare _loop, _waiters and _active_waiters in its own __slots__. A mixin that declared them would give itself a tp_basicsize larger than object's, and combining two such bases is the "multiple bases have instance lay-out conflict" this class exists to avoid. Queue, PriorityQueue and RoundRobin all do so.

Awaiting(*args: Any, **kwargs: Any) -> None

PriorityQueue#

A bounded priority queue with counted loss and an awaitable get.

Lowest number first, and items sharing a priority come out in the order they went in. That stability matters more than it sounds: most items in a real workload share a priority, so without it the common case is unordered and a test that pins ordering is flaky rather than wrong.

The item itself is never compared, so anything is queueable -- including objects that define no ordering at all, and objects whose __lt__ would run arbitrary Python at an inconvenient moment.

Args:

• capacity — items held before offering starts dropping. Must be positive.
• drop_lowest — when full, displace the worst queued item rather than
• refusing the new one, so a high-priority arrival is never lost to a
• backlog of low-priority work. Refuses only when the new item is
• itself the worst. Costs a scan of the heap's leaves, which is the
• honest price of asking a min-heap for its maximum.

Queue#

A bounded queue with counted loss and an awaitable get.

First in, first out by default, which is what a hand-off almost always wants: the oldest item has been waiting longest.

lifo=True makes it a stack instead. That is not a fairness choice but a latency one -- under a backlog a LIFO serves the newest item first, so the items it serves are the freshest, at the cost of the oldest possibly never being served at all. Reach for it where a stale item is worthless anyway (a live metric, a frame of a video feed) and never where every item must eventually be handled.

Args:

• capacity — items held before offering starts dropping. Must be positive.
• drop_oldest — evict the front when full instead of refusing the new item.
• lifo — take the newest item first rather than the oldest.

QueueEmpty#

Raised by get_nowait() when the queue holds nothing.

QueueFull#

Raised by put_nowait() when the queue is at capacity.

RoundRobin#

Fair scheduling across named lanes, each its own bounded queue.

One queue shared by many producers is fair only in the sense that it is first-come-first-served, which is exactly the wrong fairness when the producers are tenants: whoever offers fastest gets the most service, and one busy tenant starves the rest. A lane per tenant and a rotating cursor converts that into "everyone gets a turn".

work = RoundRobin(capacity=1024, max_lanes=256)
work.offer("tenant-a", job)
job = work.get_nowait()          # the next lane in the rotation
job = await work.get()           # ... or wait for whichever lane fills

capacity is per lane, not total, which is the point: one lane filling up drops that lane's work and nobody else's. max_lanes bounds the number of lanes, because lanes are created on demand and a lane name that arrives from a request is otherwise an unbounded allocation an unauthenticated caller controls -- the same failure mode every other bound here exists to prevent.

It carries the same surface as the other three: offer, get, get_nowait, peek, drain, snapshot, clear, close, closed, capacity, offered, dropped, len().

The rotation is deliberately plain Python. It moves a cursor over a list once per item and is nowhere near the cost of the queue operation it wraps, so there is nothing here for C to do. The waiting half is not rewritten either: Awaiting is the same mixin Queue and PriorityQueue use, so cancellation and cross-thread wake-ups are solved once for all four.

RoundRobin(capacity: int = 4096, *, max_lanes: int = 1024, drop_oldest: bool = False, lanes: Iterable[str] = ()) -> None

capacity (property)#

capacity: int

Items each lane holds before dropping. Per lane, not in total.

clear#

clear() -> int

Drop everything on every lane, returning how many went.

close#

close() -> None

Refuse further items; draining the backlog still works.

closed (property)#

closed: bool

Whether further items are refused.

drain#

drain(limit: int | None = None) -> list[Any]

Take up to limit items, one lane at a time in rotation.

Interleaved rather than lane-by-lane, so a batch is as fair as the stream: draining each lane to exhaustion in turn would hand a consumer one tenant's entire backlog before any of the next tenant's.

dropped (property)#

dropped: int

Items lost to a full lane, across every lane.

get#

get() -> Any

An awaitable for the next item, waiting for one if every lane is empty.

Resolves without suspending when something is already queued, exactly as Queue.get does.

get_nowait#

get_nowait() -> Any

The next item in the rotation, or QueueEmpty when every lane is empty.

The cursor advances past whichever lane answered, so the next call starts after it. Advancing only on a hit would let one busy lane hold the cursor and reinvent the starvation this exists to prevent.

lanes (property)#

lanes: tuple[str, ...]

The lane names, in rotation order.

offer#

offer(lane: str, item: Any) -> bool

Enqueue item on lane, reporting whether it was kept.

offered (property)#

offered: int

Items ever offered, across every lane.

peek#

peek(default: Any = None) -> Any

The item the rotation would serve next, without taking it.

put_nowait#

put_nowait(lane: str, item: Any) -> None

Enqueue item on lane, raising QueueFull rather than dropping it.

snapshot#

snapshot() -> dict[str, list[Any]]

What each lane is holding, without consuming any of it.

waiting (property)#

waiting: bool

Whether a getter is parked.

Setting it arms every lane, because a getter waits on any lane and the rings only call _wake for a lane that is armed.

wreath.services#

A minimal supervisor for process-lifetime background services.

The supervisor owns the long-lived asyncio tasks that the durable-jobs and messaging coordinators run (workers, durable-subscription consumers, and the lease sweeper). It is started during the application lifespan after databases come up and before user startup handlers, and stopped in reverse: stop fetching new work, drain bounded in-flight handlers, then cancel anything that outlives the grace period.

Design 01 §3 (lifespan) is the contract implemented here. A "service" is any object exposing async start(supervisor) and async drain(deadline).

Service#

A process-lifetime background service managed by the supervisor.

Structural, not nominal: any object with these two coroutines is a service. Implementations are expected to spawn their long-lived tasks through Supervisor.spawn rather than owning them, so the supervisor can cancel what outlives the grace period.

Service(*args, **kwargs)

drain#

async drain(deadline: float) -> None

Stop taking new work and finish what is in flight, by deadline.

Returning does not have to mean every task exited: whatever is still running when this returns is cancelled. Raising is survivable and counted in Supervisor.drain_errors; it never aborts a sibling's drain.

Args:

• deadline — An event-loop clock time (loop.time() units), not a duration.

start#

async start(supervisor: Supervisor) -> None

Begin running. Raising here aborts startup and rolls back siblings.

Args:

• supervisor — The owner, for spawn and for reading stopping.

Supervisor#

Owns and lifecycles the background tasks of registered services.

Register every service with add before start; services come up in registration order and drain in reverse, so a producer quiesces before whatever it feeds. stop is the only path that resolves the shutdown, and it always completes -- a service whose drain fails increments drain_errors instead of aborting its siblings.

Args:

• drain_timeout — Seconds allowed for the whole drain, shared by every service.

Supervisor(*, drain_timeout: float = 10.0) -> None

add#

add(service: Service) -> None

Register a service to be started and drained with this supervisor.

Raises:

• RuntimeError — The supervisor has already started; registration is closed.

empty (property)#

empty: bool

Whether no service is registered. An empty supervisor still starts.

is_stopping#

is_stopping() -> bool

Whether shutdown has begun. The non-awaiting read of stopping.

spawn#

spawn(name: str, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]

Own a long-lived task; it is cancelled on stop if it outlives drain.

The supervisor holds a strong reference until the task completes, which is what keeps a fire-and-forget worker from being garbage collected mid-flight. A task that raises is not observed until stop reaps it, and its exception is counted in drain_errors rather than re-raised.

Args:

• name — Task name, as it appears in asyncio diagnostics.

Returns:

• The task, for a caller that wants to await or cancel it itself.

start#

async start() -> None

Start every registered service, in registration order.

Idempotent: a second call on a started supervisor returns immediately. Startup is all-or-nothing. If one service's start raises -- including by cancellation -- the services that already came up are drained in reverse, every spawned task is cancelled, and the original exception propagates unchanged. The supervisor is left un-started, so it can be started again once the cause is fixed.

Raises:

• Exception — Whatever a service's start raised, after the rollback.

stop#

async stop() -> None

Signal shutdown, drain in reverse registration order, cancel the rest.

Idempotent: a call on a supervisor that never started returns immediately. This never raises on a service's behalf -- a drain that fails, and a spawned task that died of something other than the supervisor's own cancellation, each increment drain_errors. Read that counter to tell "everything quiesced" from "shutdown finished anyway"; the shutdown itself completes either way. CancelledError still propagates, so a cancelled shutdown stays cancelled.

stopping (property)#

stopping: asyncio.Event

Set once shutdown begins; loops should stop fetching when it is set.

wreath.notifications#

Notifications: one declaration, several channels, one preference per person.

Every application eventually rebuilds the same paragraph of behaviour -- email me immediately, push me only when I am away, never tell me the same thing twice, and batch the noisy ones -- and rebuilds it badly, because it started from a send_email(...) call at the point where something happened.

The fix is upstream of the channels. A notification here is a kind, declared once with a name and a delivery policy, and sent as an instance of that kind:

notify = Notifications(channels=[Email(sender), WebPush(keys, subscriptions)])

@notify.kind("photo_shared", digest=hours(1))
@dataclass
class PhotoShared:
    photo_id: str
    actor: str

    def title(self) -> str:
        return f"{self.actor} shared a photo with you"

await notify.send(PhotoShared(photo.id, actor.name), to=recipient)

That declaration is what makes the rest possible at all. A kind has a stable identity, so two of them can be recognised as the same thing and collapsed; a stream of bare send calls cannot be, which is why applications that start there never get deduplication and never get digests.

Delivery is a wreath.jobs job, one per channel per recipient, so retries, backoff and dead-lettering are the ones wreath already has rather than a second set. Nothing here retries in-process.

What this is not#

Not an ESP. Warm-up scheduling, reputation management, per-domain throttling and IP rotation are a product, and a framework that starts down that road ends up maintaining a deliverability team's worth of heuristics. Wreath's job is to emit correct, signed, unsubscribable mail through whatever transport is configured -- see wreath.doctor.check_email_deliverability for the part that tells you when the DNS does not back it up.

Reference: wreath.users for the sending side of account mail, and wreath.doctor for the deliverability check.

Channel#

One way of reaching someone.

Channel(*args, **kwargs)

deliver#

async deliver(recipient: Recipient, note: Any, kind: KindSpec) -> None

Deliver one notification, or raise. Retries belong to the caller.

name (property)#

name: str

The channel's name, as preferences and only= refer to it.

Chat#

Deliver a notification through a tenant-bound Wreath ChatOps destination.

Chat(chat: Any, destination: Callable[[Recipient], Any], render: Callable[[Any], str] | None = None, name: str = 'chat') -> None

deliver#

async deliver(recipient: Recipient, note: Any, kind: KindSpec) -> None

Email#

Deliver by email through a wreath.users sender.

The message body comes from the notification itself: a kind may define subject() and body(), and falls back to its title() or its repr.

Email(sender: MessageSender, name: str = 'email') -> None

deliver#

async deliver(recipient: Recipient, note: Any, kind: KindSpec) -> None

Build a Message for note and hand it to the sender.

Raises:

• ValueError — the recipient has no email address.

InApp#

Deliver into a live wreath.rooms room, for an in-product inbox.

InApp(rooms: Any, name: str = 'inapp') -> None

deliver#

async deliver(recipient: Recipient, note: Any, kind: KindSpec) -> None

Publish to the room named for this recipient.

room (staticmethod)#

room(recipient: Recipient) -> str

The tenant-qualified room a recipient's live inbox joins.

Notifications#

Declares kinds, applies preferences, and fans out to channels.

Args:

• channels — The channels to deliver on, in order.
• preferences — Per-recipient opt-outs. Permissive by default.
• enqueue — How to run a delivery. Defaults to awaiting it inline, which is
• right for a test and wrong for production -- pass
• jobs.enqueue-shaped callable so retries and dead-lettering are the
• durable ones.
• rate_limit — The most notifications one recipient may be sent per hour,
• across all kinds. 0 disables the cap.

Notifications(channels: Sequence[Channel], *, preferences: Preferences | None = None, enqueue: Callable[[Callable[[], Awaitable[None]]], Awaitable[None]] | None = None, rate_limit: int = 200) -> None

kind#

kind(name: str, *, digest: Any = 0.0, mail_class: MailClass = <MailClass.TRANSACTIONAL: transactional>, unsubscribe: Callable[[Recipient], Unsubscribe] | None = None, only: Iterable[str] = ()) -> Callable[[type], type]

Declare a notification kind. Use as a class decorator.

Raises:

• ValueError — name is already declared, or a marketing kind was
• declared without a way to unsubscribe from it -- which is the
• RFC 8058 requirement, refused here rather than at send time so
• it is a startup error instead of a delivery failure.

send#

async send(note: Any, *, to: Recipient, now: float | None = None) -> SendResult

Deliver note to one recipient on every channel they allow.

spec_for#

spec_for(note: Any) -> KindSpec

The declared policy for note.

Raises:

• LookupError — note's type was never declared with kind. Refused
• rather than defaulted, because the default would have to guess
• the mail class, and both wrong guesses are expensive.

PushDelivery#

Posts encrypted pushes, one pooled HTTPClient per push service.

wreath.http_client.HTTPClient is bound to a single origin for its life, which is what lets it pool connections and hold a rate policy. Push endpoints do not come from one origin -- a recipient list reaches Google, Mozilla and Apple in the same fan-out -- so this keeps a client per origin and reuses it, rather than building one per message and throwing away the connection that made pooling worth having.

Args:

• limits — Passed through to each client.

PushDelivery(**limits: Any) -> None

aclose#

async aclose() -> None

Close every pooled client.

send#

async send(endpoint: str, body: bytes, headers: dict[str, str]) -> PushResult

POST one encrypted payload and classify the answer.

PushError#

A subscription, key, or payload that cannot produce a valid push.

PushResult#

What a push service said.

expired is the one a caller must act on: 404 or 410 means the subscription is permanently gone and every future send to it will fail the same way. Delete it.

PushResult(status: int, expired: bool, detail: str = '') -> None

delivered (property)#

delivered: bool

PushSubscription#

What a browser's PushSubscription.toJSON() carries.

Store it whole. endpoint identifies the push service and the subscription, so it is the natural key, and p256dh/auth are the material without which a payload cannot be encrypted for this recipient.

PushSubscription(endpoint: str, p256dh: str, auth: str) -> None

auth_secret#

auth_secret() -> bytes

client_key#

client_key() -> bytes

from_json (classmethod)#

from_json(payload: dict[str, object]) -> PushSubscription

Build from the exact shape PushSubscription.toJSON() produces.

PushSubscriptions#

Where a recipient's push subscriptions live.

PushSubscriptions(*args, **kwargs)

for_recipient#

async for_recipient(key: str) -> Sequence[PushSubscription]

Every subscription registered for this recipient.

remove#

async remove(endpoint: str) -> None

Forget the subscription with this endpoint.

Called when a push service answers 404 or 410. Required: a store that never prunes becomes a slow leak plus an error rate nobody attributes, and the push services expect the deletion.

Recipient#

Who a notification is for, and how they can be reached.

key is the stable identifier deduplication and digesting are keyed on, so it must not change when an address does.

Recipient(key: str, email: str | None = None, principal: Any = None) -> None

SendResult#

What one send did, per channel.

SendResult(delivered: tuple[str, ...] = (), declined: tuple[str, ...] = (), deduplicated: bool = False, failed: dict[str, str] = <factory>) -> None

reached (property)#

reached: bool

VapidKeys#

The application server's identity to a push service (RFC 8292).

One keypair, generated once and kept for the life of the application: the public half is what a browser pins when it subscribes, so rotating it invalidates every existing subscription.

Args:

• private — The P-256 private scalar.
• subject — A mailto: or https: URL a push service operator can use to
• reach you about your traffic. Required by RFC 8292 -- some services
• reject a token without one.

VapidKeys(private: int, subject: str) -> None

application_server_key (property)#

application_server_key: str

The base64url public key to hand pushManager.subscribe in a browser.

from_bytes (classmethod)#

from_bytes(private: bytes, subject: str) -> VapidKeys

Rebuild from the 32 bytes private_bytes returned.

generate (classmethod)#

generate(subject: str) -> VapidKeys

Mint a fresh keypair. Store private_bytes; it is not recoverable.

private_bytes (property)#

private_bytes: bytes

public_bytes (property)#

public_bytes: bytes

The uncompressed public key, which is what a browser subscribes to.

WebPush#

Deliver by Web Push, encrypted per RFC 8291.

Args:

• keys — The application server's VAPID identity.
• subscriptions — Where recipients' subscriptions live.
• post — How to POST a body. Defaults to wreath.http_client; injectable
• so a test never opens a socket.
• ttl — How long a push service should hold an undelivered message.

WebPush(keys: VapidKeys, subscriptions: PushSubscriptions, post: Callable[..., Awaitable[PushResult]] | None = None, ttl: int = 3600, name: str = 'webpush') -> None

deliver#

async deliver(recipient: Recipient, note: Any, kind: KindSpec) -> None

Encrypt and send to every subscription this recipient has.

A 404 or 410 prunes the subscription and is not an error: the endpoint is permanently gone, which is information rather than a failure. Anything else raises, so the caller's job retries it.

wreath.cache#

Immutable read-mostly application cache with atomic snapshot publication.

Distinct from wreath.CacheControl (HTTP response caching), this is an in-process store for configuration, reference data, and database-backed read-mostly datasets:

cache: SnapshotCache[int, Widget] = SnapshotCache()
await cache.refresh(load_widgets)        # single-flight
widget = cache.get(widget_id)            # no I/O; explicit miss

Readers always observe one complete generation; a refresh publishes a new one atomically and leaves the previous generation intact on failure.

The read path is a dict lookup that CPython already services in C, and a native port was measured against kv.c and came out slower, so there is one implementation rather than a selection. See wreath._snapshot.

BoundedCache#

A bounded LRU cache with optional per-entry TTL.

This is wreath.kv.KV, with one convenience on top. It kept its own name because that is the spelling the response cache, the idempotency layer and wreath port's TTLCache/LRUCache translation all already use -- but it is no longer a second implementation, or a second set of semantics, or a second thing to learn. Everything it does, KV does.

get, set, delete, clear, snapshot, __contains__, __len__, ttl and max_entries are inherited from KV. The convenience adds only as_dict; clock is KV's own parameter.

Args:

• max_entries — hard ceiling; the least-recently-used entry is evicted once
• a set would exceed it. Must be positive.
• ttl — seconds an entry stays fresh, or None to never expire by time.
• Expiry is lazy -- checked on read -- so nothing runs in the
• background.
• clock — monotonic time source, injectable for deterministic tests.

stats (property)#

stats: CacheStats

A point-in-time view of this cache's activity.

The one thing KV does not offer under this name: it exposes the same four counters individually, and wreath.kv.stats(table) builds the same value from any table. This is the property spelling those callers use.

RefreshWatch#

The handle refresh_on returns. Call it to stop watching.

It also carries what an operator needs when reloads are failing. A snapshot cache degrades upwards: a broken loader leaves every read succeeding against the last good generation, so the usual signals -- errors, latency, a falling hit rate -- all stay quiet while the data silently ages. These two attributes are the only place that shows.

RefreshWatch(stop: Callable[[], None]) -> None

SnapshotCache#

A read-mostly cache with atomic snapshot publication.

replace builds and publishes a new generation. get/require/ get_many read the current generation with no I/O. max_entries bounds a published generation; a violation raises before anything is published, so the previous generation survives intact.

SnapshotCache(*, max_entries: int | None = 65536, max_bytes: int | None = 67108864) -> None

generation (property)#

generation: int

get#

get(key: K, default: V | None = None) -> V | None

get_many#

get_many(keys: Iterable[K], default: V | None = None) -> list[V | None]

Look up each key, preserving input order and duplicate positions.

A duplicate key yields a value at each position it was requested; a miss yields default at that position.

refresh#

async refresh(loader: Callable[..., Mapping[K, V] | Iterable[tuple[K, V]] | Awaitable[Any]], *args: object, **kwargs: object) -> int

Single-flight refresh: call loader and publish the result.

Concurrent refreshes coalesce — while one is in flight, the others wait and observe the generation it publishes rather than each loading again. loader may be a coroutine function or a plain callable returning the new entries. A loader failure leaves the previous generation in place.

refreshed_at (property)#

refreshed_at: float

replace#

replace(entries: Mapping[K, V] | Iterable[tuple[K, V]]) -> int

Publish entries as a new generation and return its number.

The new generation is fully materialized and size-checked before it is published, so an oversized refresh fails without disturbing readers.

require#

require(key: K) -> V

WriteBroadcast#

Relay write announcements over an at-most-once message bus.

WriteBroadcast(bus: Any, *, channel: str = 'wreath_writes') -> None

close#

close() -> None

Stop carrying and applying writes without removing the bus listener.

invalidate_across_workers#

invalidate_across_workers(bus: Any, *, channel: str = 'wreath_writes') -> WriteBroadcast

Make ORM-driven cache invalidation fleet-wide, over the message bus:

invalidate_across_workers(app.messaging("bus", database="app"))

@app.get("/herd/report")
@cached(ttl=300, invalidate_on=[Llama])
async def herd_report(request): ...

Without it, a write on worker A clears only worker A. With it, the model names the committing session announces are carried to every worker on one channel and applied there, so four workers behave like one -- and still no Redis, because the bus is the database you already have.

Returns the WriteBroadcast carrying them, whose close() stops it. Call once per process, before startup: the bus collects its subscriptions before it begins listening.

Delivery is at-most-once, as ephemeral fan-out is. A worker that misses the notification holds its entries until they expire, which is what the ttl is now for -- a backstop rather than the mechanism.

refresh_on#

refresh_on(cache: Any, models: Iterable[Any], *, load: Callable[[], Any]) -> RefreshWatch

Reload cache whenever one of models is written:

countries: SnapshotCache[str, Country] = SnapshotCache()
await countries.refresh(load_countries)
refresh_on(countries, [Country], load=load_countries)

The snapshot cache holds reference data, so the right response to a write is to reload rather than to drop -- a dropped generation would leave readers with an explicit miss on data that has not gone anywhere.

This is the same announcement the response cache listens to, so invalidate_across_workers makes it fleet-wide too: a write on any worker reloads every worker's reference data, over one bus channel.

Refresh is single-flight and a failing loader leaves the previous generation in place, so a database blip cannot empty a cache that readers depend on. The reload is scheduled rather than awaited -- the write has already committed and must not wait on it.

Returns a RefreshWatch: call it to stop watching, and read refresh_errors to find out whether the reloads are actually working.

wreath.kv#

One bounded in-process key/value table, for everything that needed one.

A response cache, a session table, an idempotency ledger, a JWKS cache and a claim ledger are five features that need the same small thing: values under a key, a hard ceiling on how many, and a deadline. Written five times, the same three decisions get re-made five times -- and they are the decisions that are easy to get quietly wrong:

• eviction has to be least recently used, not insertion order, or a hot key is thrown away while a cold one survives;
• expiry has to be lazy, because a background sweep thread duplicates across workers and swallows its own failures;
• len() has to count what the table will still return, not what it happens to be holding, or it reports debris as occupancy.

from wreath.kv import KV

sessions = KV(max_entries=10_000, ttl=1800.0)
sessions.set(session_id, payload)
payload = sessions.get(session_id)          # None once the deadline passes

This is one worker's memory. Enough on its own for a single-worker deployment or a sticky load balancer; behind anything else it is a fast path in front of a shared store rather than a substitute for one, because a second worker's memory knows none of it. wreath.store.PostgresStore is the shared half, and the two agree on the semantics that matter -- notably that a claim's window opens on first write and a later write does not move it, which set spells keep_deadline=True.

Not synchronised. Every caller is event-loop-local, exactly as the pure BoundedCache this replaces was, and adding a lock would charge every reader for a race no reader has. wreath.queue is the primitive for handing work between threads.

The native table is open-addressed with a one-byte tag per slot, scanned 32 lanes at a time; _native/simd.h documents the encoding and _native/kv.c the rebuild policy.

KV#

A bounded key/value table with LRU eviction and lazy TTL expiry.

Not synchronised: it is built for one event loop, the way the pure BoundedCache it replaces was. Use wreath.queue to hand work between threads.

KV(*args, **kwargs)

Stats#

A point-in-time view of a table's activity.

Stats(hits: int, misses: int, evictions: int, expirations: int, size: int) -> None

hit_rate (property)#

hit_rate: float

Hits as a fraction of reads, or 0.0 before the first read.

stats#

stats(table: Any) -> Stats

The counters table has accumulated, as one value.

A function rather than a method so it reads the same on both arms without the native type having to build a dataclass in C. size is the live entry count, so it agrees with len(table) and not with table.slots.

wreath.log#

One append-only log, declared once, read back from a cursor that cannot skip.

An audit trail, a change feed, a stream of a job's output, a meter of billable events: four features that need the same small thing, which is rows in commit order and a cursor a reader can come back with. Written four times, the hard parts are re-derived four times -- and they are the parts that fail silently:

• the cursor, which is the whole of it. A bigserial is allocated before commit, so a reader that remembers max(seq) skips every row a slower transaction commits afterwards with a lower number. The gap is invisible, intermittent and load-dependent;
• the table name reaches SQL by interpolation, so it must be a plain identifier;
• the schema is offered (Log.component) and never applied, because schema changes belong in the migration history with the rest of the schema;
• statements are prepared lazily, because a log is built while the application is being described and the database is not up yet;
• retention is declared, never defaulted -- a log with no answer to "how long do these rows live" is a disk-space incident with a delay fuse -- and it is executed by retention_pass, a wreath.passes walk, so the rows it removes are a number in a ledger rather than a policy nobody drove;
• appends batch. append_many decomposes a batch into powers-of-two multi-row inserts, because a producer whose rows are small and frequent would otherwise pay one round trip per row;
• an unflushed batch lost to a worker's death is counted, never absorbed.

Declare the shape and pick what it holds:

chunks = Log(
    table="stream_chunks",
    columns=(Column("body", "bytea", null=False),),
    retain=hours(24),
)
log = PostgresLog(database, chunks)

await log.append("conversation-7", body=b"hello ")
page = await log.read("conversation-7", after=Cursor.start())

What is not shared is the payload and its meaning: what the columns hold, what a stream name identifies, and what a reader does with a row. Each caller keeps those.

At-least-once, de-duplicated by the reader. A row is delivered as many times as a reader asks for it. Cursor is what makes de-duplication possible, and promising exactly-once here would be a guarantee no transport can keep.

Not a message bus. wreath.messaging is the bus, and a NOTIFY remains the doorbell that says something moved, go read the log. A notification is not the delta: a subscriber that was disconnected for one never learns it happened, and treating the payload as the change is how that subscriber silently misses data.

Not the flight recorder. wreath._ring_file is a memory-mapped ring sized to survive the process that wrote it and to be read with no network hop; its rows are gone on the next wrap. This is durable, ordered, shared between workers, and read with SQL. They look similar from a distance and answer different questions, so they stay apart.

Not event sourcing. No projections, no aggregates, no rebuilding state from the log. Read-side consumers only.

Batch#

A page of records and the cursor to come back with.

cursor is where to resume even when records is empty: an empty batch still means "nothing below the horizon that you have not seen", and throwing it away would make a quiet log replay its tail forever.

Batch(records: tuple[Record, ...], cursor: Cursor) -> None

Cursor#

Where a reader got to: a transaction id and a sequence number.

Both halves are load-bearing. The sequence alone is not a cursor, and neither is the sequence gated on visibility: sequence order is allocation order, and rows do not commit in the order they allocate. Ordering by (xid, seq) and remembering both is what makes "everything after this" mean it. wreath.log's cursor contract has the proof and the two wrong answers it rules out.

Opaque to a caller: build one with start or take it from a Batch, round-trip it through encode/decode, and never do arithmetic on it.

decode (classmethod)#

decode(value: str | None) -> Cursor

Parse encode's form, refusing anything else.

Client-supplied by design -- a Last-Event-ID header is whatever the other end sent -- so this refuses rather than repairs. A cursor that parsed loosely would be an index into the log built from request text.

encode#

encode() -> str

A form safe to hand a client and take back, e.g. a Last-Event-ID.

start (classmethod)#

start() -> Cursor

The cursor before the first row of any log.

Flush#

When a buffered append gives up and writes.

Bytes or milliseconds, whichever comes first -- one alone is a bad trade in one direction or the other. A byte threshold on a slow producer holds the first token until the last one arrives; a time threshold on a fast one writes a row per token, which is the write amplification buffering exists to avoid.

capacity bounds the buffer in items. Offering past it drops, and the drop is counted rather than absorbed: bounded memory is the promise a buffer makes, and a silent drop is how a log starts lying about completeness.

Flush(bytes: int = 8192, every: float = 0.05, capacity: int = 4096) -> None

Log#

The declaration of an append-only log: what it holds and how long.

Args:

• table — the backing table. Interpolated, so it must be a plain identifier.
• columns — the payload. Their meaning belongs to the caller.
• retain — seconds a row lives, or KEEP_FOREVER. Required, with no
• default — a chunk buffer and an audit trail want opposite answers and
• neither should inherit the other's by omission.
• stream — the column naming the partition a reader follows.
• dedup — add a nullable dedup column with a unique index, for a caller
• whose producer may retry -- a metered event, a webhook receipt.
• flush — buffering policy for PostgresLog.buffered.
• schema — the schema the table lives in. Wreath's own furniture goes in
• wreath; a caller may name another.
• prefix — prepended to prepared-statement names.

Log(table: str, retain: float | None, columns: tuple[Column, ...] = (), stream: str = 'stream', dedup: bool = False, flush: Flush = <factory>, schema: str = 'wreath', prefix: str = 'wreath_log') -> None

qualified_table (property)#

qualified_table: str

The table as it reaches SQL: schema-qualified unless the schema is "".

schema_claim#

schema_claim(name: str) -> Any

This declaration's claim on the wreath schema, under name.

name is the caller's, because one Log shape backs several subsystems and each needs its own version marker and its own advisory lock -- sharing one would make an upgrade to the audit trail block a worker that only streams.

Named apart from the zero-argument component() protocol for the reason wreath.store.Keyed.schema_claim gives: a declaration needs the argument, and the walk cannot pass one.

schema_sql#

schema_sql() -> str

DDL for the backing table, semicolon-joined.

A derivation of statements(); wreath schema sql is the supported spelling for a caller applying it by hand.

statements#

statements() -> tuple[str, ...]

DDL for the backing table, one statement per element.

xid8 rather than xid: the 32-bit type wraps, and a cursor built from a wrapping counter compares wrongly exactly once every four billion transactions, which is both rare enough to survive review and certain enough to happen.

PostgresLog#

An append-only log in a table every worker shares.

Holds the disciplines named in this module's docstring, and generates the statements a log always wants -- append, read, tail, purge -- from the declaration.

The horizon is what makes the cursor honest. A read never returns a row whose transaction may still be in flight: it stops at pg_snapshot_xmin(pg_current_snapshot()), the oldest transaction id not yet settled. Every row below that horizon has finished, so nothing can later appear behind a cursor that has passed it.

The cost is latency, and it is worth naming: a long-running transaction -- anybody's, anywhere in the database -- pins the horizon and stalls every log reader until it ends. That is a property of the database, not of this module, and it is visible as a log that stops advancing while appends continue. PostgresLog.horizon_lag reports it.

Args:

• database — a wreath.postgres.Database.
• declaration — what the log holds.
• read_workload — the pool reads go to. "write" by default, because a
• reader that has just appended must see its own rows.

PostgresLog(database: Any, declaration: Log, *, read_workload: str = 'write') -> None

append#

async append(stream: str, /, *, connection: Any = None, **values: Any) -> Cursor

Append one row to stream, returning where it landed.

The returned cursor is this row's position, not a resume point: a reader handed it would skip this row. It is the append's receipt -- proof the row committed and where it sits in the order -- and callers that want to read from here pass it as after.

connection is what makes an append atomic with a write. Without it the statement takes its own pooled connection and therefore its own transaction, which commits whether or not the caller's does -- so a change feed built that way describes writes that rolled back, and an audit trail built that way records them. Pass the connection the caller is already writing on and the row shares its fate.

Each caller chooses. A change feed and an audit trail must pass one; a buffered stream of output must not, because its loss is survivable and its latency is not.

append_many#

async append_many(rows: Sequence[tuple[str, Mapping[str, Any]]], /, *, connection: Any = None) -> int

Append many rows as multi-row inserts, returning how many landed.

The point of the whole thing. append is one statement per row, and a producer with a buffer in front of it -- a chunked stream, a metered event per request, an ORM flush of a hundred audited instances -- turns that into a round trip per row, which is the write amplification the buffer exists to remove. This decomposes the batch into powers of two and issues one prepared INSERT per rung, so a thousand rows cost six statements rather than a thousand.

rows is (stream, values) pairs, because the audit trail's rows go to one stream per audited row and a single-stream signature could not carry them.

connection carries the same meaning it does on append, and it is what keeps the batch atomic with the write it describes: every rung runs on the connection the caller is already writing on, inside the caller's transaction, so the records commit if and only if that transaction does. Without it each rung takes its own pooled connection and its own transaction, and a batch that fails halfway leaves the rungs before it committed -- which is what a buffered producer wants and what an audit trail must not have.

append_once#

async append_once(stream: str, /, *, dedup: str, **values: Any) -> Cursor | None

Append unless dedup is already recorded for stream.

None means the event was already there -- a retried producer, a redelivered webhook -- which is a fact about the caller's world rather than an error in this one.

Raises:

• ValueError — when the log was not declared with dedup=True.

buffered#

buffered(stream: str) -> _Buffer

A byte-or-millisecond buffer over append for one stream.

For a producer whose rows are small and frequent -- a token at a time, a metered event per request -- where a row per item is write amplification that a flush threshold removes. Rows still in the buffer when the process dies are lost and counted; a caller who cannot afford that appends directly.

database (property)#

database: Any

The wreath.postgres.Database this log runs against.

Exposed because two legitimate operations need a connection of their own rather than a pooled statement: an erasure, which carries a transaction-scoped setting, and a retention walk, which paces itself. wreath.store.PostgresStore has no equivalent because neither applies to a keyed store.

declaration (property)#

declaration: Log

The Log this was built from. Frozen, so it is safe to share.

drop_stream#

async drop_stream(stream: str, *, connection: Any = None) -> str

Drop one partition outright, whatever its retention.

The erasure path: a stream is a subject, and a subject may ask to be forgotten. Deliberately separate from purge so that a KEEP_FOREVER log still has an answer, and so that dropping evidence is always an explicit act naming what it drops.

connection runs the delete on a connection the caller already holds -- which is what an erasure needs, because it has to carry a transaction of its own (see wreath.audit_log.AuditTrail.forget).

dropped (property)#

dropped: int

Buffered rows lost -- to a full buffer, or to a flush that failed.

Never resets. A log that has dropped anything is a log whose completeness claim has a number attached to it, which is the only honest way to make one.

horizon#

async horizon() -> int

The transaction id below which every row has settled.

horizon_lag#

async horizon_lag() -> int

How far the newest appended row sits above the horizon.

Zero on a healthy log. A number that grows and does not come back down is a transaction somebody left open: readers stall while writers carry on, which looks like the log has stopped rather than like the database has a held snapshot.

purge#

async purge() -> str

Drop rows past their retention.

Nothing calls this for you, and on a log large enough to need retention this is the wrong shape: it is one unbounded DELETE over a table that only ever grows, which is a lock held for as long as it takes and a WAL spike nobody scheduled. retention_pass is the same deletion as a durable, resumable, paced walk whose rows are counted in the pass ledger. Reach for this on a small log, and for the pass otherwise.

Raises:

• ValueError — on a KEEP_FOREVER log, where there is nothing to drop.

read#

async read(stream: str | None = None, *, after: Cursor, limit: int = 512) -> Batch

Records after after, stopping at the horizon.

stream=None reads the whole log rather than one partition -- the shape a change feed wants, where every row matters to somebody.

The returned Batch.cursor is where to resume. When the batch is empty it is after unchanged, so a quiet log does not rewind.

retention_pass#

retention_pass(*, name: str, **options: Any) -> Any

This log's retention walk. See wreath.log.retention_pass.

schema_claim#

schema_claim(name: str) -> Any

This log's claim on the wreath schema, under name.

schema_sql#

schema_sql() -> str

DDL for the backing table, semicolon-joined.

table (property)#

table: str

The backing table, schema-qualified as it reaches SQL.

define (inherited from _Statements)#

define(name: str, sql: str, *, workload: str = 'write') -> None

Register sql under name for lazy preparation.

sql (inherited from _Statements)#

sql(name: str) -> str

The text registered under name. Useful when explaining a plan.

statement (inherited from _Statements)#

statement(name: str) -> Any

The prepared statement for name, preparing it on first use.

workload (inherited from _Statements)#

workload(name: str) -> str

The pool the statement registered under name runs against.

Record#

One row, with its cursor and whatever payload columns the log declared.

Record(cursor: Cursor, stream: str, values: Mapping[str, Any]) -> None

retention_pass#

retention_pass(declaration: Log, *, name: str, chunk: int = 1000, within: Any = '5s', shift: Any = '10s', pace: Any = None, schema: str = 'wreath', tenant: str = '') -> Any

A wreath.passes.ChunkedPass that drops this log's rows once they age out.

retain= is a declaration and this is what executes it. purge() is the one statement it would otherwise take -- an unbounded DELETE over a table that only ever grows, which on a log large enough to need retention is a lock held for minutes and a WAL spike nobody scheduled. A pass is the same deletion done as a durable, resumable, paced keyset walk, and the rows it removes are counted: ShiftResult.rows per drive, PassStatus.rows_done cumulatively in the ledger. That is what makes "we have a retention policy" a number rather than a claim.

The walk is (at, seq): at because that is the ordered domain the retention window is measured in and the column Log.statements indexes for exactly this, and seq appended because two rows can share a clock_timestamp() and a chunk boundary that is not unique either skips its siblings or loops on them forever.

Takes the declaration rather than a PostgresLog, because a ChunkedPass is handed its database when it is driven -- so a connection passed here would have nowhere to go. Drive it with wreath.jobs.JobRunner.drive(pass_, cron=...).

Retention does not use the erasure door, deliberately. An audit trail's append-only trigger refuses DELETE unless the transaction has set wreath.audit_erasure, and a retention walk could be taught to set it too. It is not, and it must not be: that setting is the whole of what stands between an audit record and a background job, and a scheduled walk holding it would mean the permission to delete evidence is granted every five minutes to a process nobody is watching. So the two doors stay separate -- AuditTrail.forget names one subject and carries the setting for exactly one transaction, and retention is refused outright on a KEEP_FOREVER log, which is what an audit trail is. A log that both retains and wears the append-only trigger is a contradiction the trigger will report at the first chunk, and it is the declaration that is wrong.

Args:

• declaration — the Log whose rows are being aged out.
• name — this pass's identity in the ledger.
• chunk — rows one chunk deletes.
• within — a chunk's time budget.
• shift — how long one stretch of work runs.
• pace — how much of the machine the walk may be. DutyCycle() by default.
• schema — where the pass ledger lives; match the job runner's.
• tenant — the ledger row's tenant.

Raises:

• ValueError — on a KEEP_FOREVER log. There is nothing to age out, and a
• pass that silently walked one would be a scheduled deletion of
• evidence.