Events and consumers¶

An event is a fact one module states about its own data, such as "a task was created". Other modules react to it without importing the producer. An event is a typed class, class TaskCreated(Event, channel="tasks.task.created.v1"); await TaskCreated(...).publish() writes it to the transactional outbox of the running unit, so it is sent if and only if the change commits. A consumer is an async def decorated with @on(TaskCreated); each delivery runs in its own unit. Delivery is at least once, so a consumer that must apply an effect once claims the event id in its own transaction, which @on does for you by default. Events are Sandloom's asynchronous cross-module channel; the synchronous one is a contract.

What you get by default / what you declare¶

How to use it¶

Declare and publish¶

A producer lives in the module's events.py. The channel is <module>.<entity>.<verb>.vN in lower case; its first segment is the short name of the module that owns it.

"""Events of the tasks module (its own channels)."""

from uuid import UUID

from sandloom.sdk import Event


class ItemCreated(Event, channel="tasks.item.created.v1"):
    item_id: UUID
    tenant_id: UUID

Publish from code that runs inside a unit, typically right after the write it reports:

await ItemCreated(item_id=item.id, tenant_id=item.tenant_id).publish()

• Fields are plain data from a closed set of wire shapes (the table below); anything else is refused at class definition (SANDLOOM-B015). A field named channel, meta, publish or wire_schema is refused at class definition (SANDLOOM-B015: it would shadow the event's API). To keep such a wire key, alias the field: via: str = Field(alias="channel") is published as channel, and every consumer reads it back by that key. Unknown keyword arguments are refused at construction.
• Payloads hold ids and amounts, not PII. Consumers can read the rest through the owner's read contract.
• Order is yours to carry. Events have no ordering guarantee (below). If consumers must apply events in order, add a per-aggregate version: int.
• Several databases in one unit stack: publish(database="x") chooses which database's outbox gets the row. Without it, an ambiguous choice raises instead of guessing.
• .publish() outside any unit raises UnitOfWorkRequiredError (SANDLOOM-R002). There is no outbox to make it reliable. The deliberately lossy L2 form, publish_event(..., required=False), needs a # sandloom: best-effort <reason> marker (SANDLOOM-C007).

The generator writes a producer class for you: sandloom add event <module> <name>. See Add an event.

What an event field may be¶

An event payload is plain data that every reader, old or new, decodes the same way. The set of shapes is closed: each field, at every depth, is one of the allowed shapes, and a class with any other is refused when it is defined (SANDLOOM-B015), with the field's path (OrderPlaced.lines[].item) and the fix. The check judges what pydantic validates, so a custom type is fine when it validates as an allowed shape.

Why closed: a reader built for these shapes ignores a newer producer's key at every depth and returns the same value, of the same type, a constructed event holds. Each refused shape breaks that somewhere: an undiscriminated union can pick the wrong branch and drop a newer key's value, a union of overlapping scalars reads "123" back as 123, a custom __init__ or a raw validator builds a DTO with its own strict validator, a one-way, empty or shared wire name or a serializer publishes what the reader does not read back, a ser_json_* option publishes timedelta(seconds=2) as 2000.0, and JSON has no NaN. A property test builds random events from the whole set (and from what it refuses) and holds every one that defines to an exact round trip through JSON (tests/sdk/test_sdk_event_roundtrip.py).

from_wire reads JSON data (a delivery's decoded payload), never Python objects: a DTO instance in a payload is refused. A class rebuilt after a configuration change (model_rebuild) is checked and gets a new reader on its next decode.

Producers and views¶

The same Event base class declares both sides of a channel:

• A producer is a class whose channel starts with the defining module's own short name. Its schema is registered and exported; it may publish. A module defines at most one producer class per channel.
• A view is a class for another module's channel, declared by the consumer with only the fields it reads. A view is never registered, and its .publish() is refused (EventDefinitionError; SANDLOOM-C025 finds it statically). Unknown payload fields are ignored when a view is decoded.

class ItemCreated(Event, channel="tasks.item.created.v1"):
    """A view: fields of the producer (tasks/events.py)."""

    item_id: UUID
    tenant_id: UUID


@on(ItemCreated)
async def on_item_created(event: ItemCreated) -> None:
    """Runs once per event (durable), with the event's tenant bound."""
    await services.record_item()

A view's fields are checked against the producer's schema: the producer in the same app first, else the exported schema file (<shared package>/events/<channel>.json), else nowhere, which is reported as "unverified" (SANDLOOM-C025, a warning). sandloom add consumer <module> <channel> writes a view from the producer schema, so it starts compatible. role="producer" / role="view" state the role when the layout cannot derive it (advanced); a stated producer outside the channel's module is refused (SANDLOOM-B015).

Consume¶

@on(E) turns a coroutine that takes exactly one parameter, the event, into a durable consumer. Put it in the module's consumers.py; discovery registers it. The options are those of the L2 register_event_consumer:

Inside the handler, db() is the consumer unit's session; the unit commits when the handler returns and rolls back, for redelivery, when it raises. event.meta holds the delivery metadata (id, channel, schema_version, trace_id, emitted_at, tenant_id). It is not part of the payload and exists only on delivered instances: reading it on an event you built yourself raises EventMetaUnavailable.

Without a bound tenant a handler cannot touch TenantScoped tables (they fail closed); consumers never inherit the publisher's tenant context. With tenant="derive" and a tenant_id field, the tenant is bound for you; a missing or invalid value fails the delivery.

See Add a consumer.

Groups in split deployments¶

A consumer's group is namespaced by the module that declares it, so two modules, or one module deployed in two services, that reuse a name never share a group and never split each other's events. Replicas of one deployment share the group. Startup fails when two consumers of one process resolve to the same group, when an explicit group= is already used by another module, or when the group is already owned by another module or physical database (the first joiner records an owner fingerprint).

/diagnostics/consumers reports each local consumer's state, connection attempts, last error, group, module and the group's lag and pending count. Startup retries required consumers for CONSUMER_STARTUP_TIMEOUT_SECONDS; reconnect backoff is CONSUMER_RETRY_BASE_SECONDS / CONSUMER_RETRY_MAX_SECONDS. A handler failure for one event never makes the process unready: that event stays unacknowledged for redelivery or the DLQ.

At-least-once delivery and exactly-once effects¶

• The envelope id equals the outbox row id and is stable across redeliveries. Duplicate deliveries do happen (the post-commit dispatch and the relay can race); this is by design.
• With durable=True (the default) the consumer unit claims (consumer, envelope id) in its own database before your handler runs; a duplicate is skipped. The claim commits with your effect, so a crash before COMMIT redelivers and a crash after it never re-applies. This is exactly-once effect on that database's state, not exactly-once delivery.
• With durable=False the dedupe is a cache record: best effort, fail-open, TTL-bounded. Use it only for idempotent or non-critical handlers.
• An effect in another store (a cache, an external API, another database) is not covered by the claim. Make it idempotent, key it with operation_key(...), or write it conditionally.
• L2 code that registers a raw (session, envelope) handler claims explicitly: if not await durable_inbox("<consumer>").claim(session, envelope["id"]): return.

Poison events and the dead-letter queue¶

• A handler error rolls the unit back and the delivery is retried with per-message backoff. After the bus's delivery budget a dead_letter=True consumer's delivery is parked in <channel>.dlq, recording the reason and the group.
• Raise PoisonEventError (from sandloom.sdk.events) for an event that can never apply, such as a malformed payload: it is dead-lettered at once, even on a dead_letter=False consumer. That is the only way out of a consumer that otherwise retries forever.
• Inspect and replay once the cause is fixed. A replay is targeted at the group that dead-lettered the entry; other groups skip it.

sandloom streams dlq-list   <channel> [--limit N]
sandloom streams dlq-replay <channel> [--id DLQ_ID ...] [--limit N]

• On the producer side, an outbox row that exhausts OUTBOX_RELAY_MAX_ATTEMPTS is dead-lettered too (dead_at set, outbox_dead_total incremented), not retried forever. Repair it with sandloom outbox list-dead / sandloom outbox requeue. A broker outage does not burn attempts: it trips the relay's circuit breaker.
• Streams are trimmed by age (EVENT_STREAM_RETENTION_SECONDS, default 7 days) and optionally by length (EVENT_STREAM_MAXLEN). A consumer offline longer than retention misses events.

No ordering guarantee¶

There is no ordering guarantee, not even within one channel or one aggregate: concurrent units, relay redeliveries and retries interleave. Only one unit's own post-commit dispatch is sequential in emission order. Order by a per-aggregate version in the payload and apply it with a version guard (see Projections). If a version field was added as optional within a .v1, events from not-yet-upgraded producers lack it: treat such an event as "unknown order" and refresh from the owner's read contract instead of applying or dropping it.

Changing an event¶

The .vN suffix is the major schema version and is authoritative: the outbox row and the envelope take schema_version from it.

• Compatible on the same .vN: adding an optional field: one with a default of any kind, a literal (note: str | None = None) or a factory (tags: list[str] = Field(default_factory=list)). The wire schema lists it as required (it is always emitted) and marks it "x-sandloom-default": true; a reader fills it when an older payload omits it. sandloom generate compares the producer's schema with the previous committed snapshot before writing anything (sandloom test compares and never writes); any other change fails with SANDLOOM-T005. Producers and consumers of one .vN deploy independently, so the check holds both directions: a new reader accepts what an old producer sends, and an old reader accepts what a new producer sends. Loosening a constraint (le=10 to le=20) or removing a field breaks the second one, even on a model open to extra keys. A model open to extra keys (ConfigDict(extra="allow")) may already carry any key as an extra, so a field it newly declares must accept whatever the old model allowed there (score: Any = None, not score: int = 0).
• Readers ignore unknown keys at every depth. An @on consumer, a view or a projection decodes a payload with from_wire: a key its class does not declare is ignored, at the top level and in every nested DTO, even one declared ConfigDict(extra="forbid") (construction stays strict; only the reader ignores). The reader, built when the class is defined, is the class's own validation schema with extra="forbid" read as extra="ignore" everywhere and nothing strict (a DTO declared strict=True still reads a UUID published as a string), so aliases, discriminators and validators behave as in the class and run once, and every value is an instance of its declared class. That is what makes an added optional field safe for an old reader anywhere in the payload, and the wire schema says so: it carries no additionalProperties: false. A DTO open to extra keys (extra="allow") keeps them as data.
• Removing a default breaks. tags: list[str] = Field(default_factory=list) becoming tags: list[str] leaves the schema's required list unchanged, but a producer from before tags existed still sends payloads without it, and only a defaulted reader fills it: losing the default fails with SANDLOOM-T005 (adding a default to a required field is compatible). The "x-sandloom-default" mark belongs to the framework: a json_schema_extra that sets it is refused at class definition (SANDLOOM-B015).
• Anything else (a renamed, removed, retyped or newly required field) is a new channel, <module>.<entity>.<verb>.v2, with its own producer class. Publish both versions in one unit until no live consumer reads .v1:

from sandloom.sdk.events import publish_event_versions

await publish_event_versions({
    ItemCreated.channel: v1.model_dump(mode="json"),
    ItemCreatedV2.channel: v2.model_dump(mode="json"),
})

sandloom streams consumers shows the live readers per channel version. See Change an event's payload.

The bus¶

EVENT_BUS_BACKEND selects the transport of the default eventbus instance: auto (default: streams in production, memory otherwise), streams (Redis Streams: durable groups, ack, redelivery, DLQ), redis (Pub/Sub, volatile) and memory (in-process, for dev and tests). Volatile buses are refused in production, and durable consumers are refused on Pub/Sub, unless the deployment accepts the risk volatile_event_bus (SANDLOOM_ACCEPTED_RISKS). See Capabilities and infrastructure.

Testing¶

published(ItemCreated) lists what the code under test published; deliver(event) (an instance or a channel and payload, times= for duplicates) runs your consumers as the runtime would. See the testing reference.

What breaks it¶

sandloom explain <code> prints the cause and the fix.

Design rationale¶

• ADR-0005: the effect model (outbox, inbox, guarantee windows).
• ADR-0010: Public API 1.0 §3 (typed events and consumers) and §6 (derived labels).
• Event contracts reference (naming rules, the registry, evolution policy).