Events¶

An event says that something happened. It commits in the transaction that made it happen, and consumers receive it after the commit. Events live in the layout's database, with no broker.

Publishing¶

Declare events = Events() on the service, and publish with its SQLAlchemy connection or session. The event exists if and only if the transaction commits:

from myxo.events import Events


class Noted(msgspec.Struct, frozen=True):
    note_id: int


class Notes(Service):
    name = "notes"
    events = Events()

    async def add(self, note: NewNote) -> NoteRef:
        insert = NOTES.insert().values(text=note.text).returning(NOTES.c.id)
        async with self.engine.begin() as tx:
            note_id = (await tx.execute(insert)).scalar_one()
            await self.events.publish(tx, Noted(note_id))
        return NoteRef(note_id)

Events are frozen structs, like messages. Declare them in the interface module, since their consumers import them: Noted belongs beside NotesAPI. The layout's database must be the one this service writes to:

database = "${DATABASE_URL:-sqlite:///notes.db}"

[services.notes]
database = "${DATABASE_URL:-sqlite:///notes.db}"

With two different databases, publish would write to one and the consumers would read the other.

Consuming¶

@subscribe makes a method a handler. It runs after the commit, in any process of the layout that runs the service:

from hello.api import Noted
from myxo.events import subscribe


class Search(Service):
    name = "search"

    @subscribe(Noted)
    async def index(self, event: Noted) -> None:
        ...

Delivery works like this:

• Consumer groups. The group is the service's name, or group=. Each group sees each event at least once, however many replicas run it. One member of a group handles its events at a time, and another takes over within seconds if it dies.
• Order. A group sees events in the order of their commits. Groups advance independently of each other.
• At least once. A handler may see an event twice after a crash, so it must be harmless to repeat.
• Failure. Unavailable brings the event back after a backoff. Any other error stops the consumer, which its supervisor restarts, so a handler that always fails ends up stopping the process.
• A new group starts from the oldest event kept, so a service added later sees the last hour.

In the publisher's transaction¶

A handler that must commit with the event takes the transaction. It runs inside publish, and only in the publisher's process, where its service must be placed:

@subscribe(Noted, in_transaction=True)
async def count(self, tx, event: Noted) -> None:
    await tx.execute(...)

Use this only when the two writes must commit together. It ties the two services to one process and one database.

Publishing from outside a service¶

A command that imports data has no runtime. myxo.events.publish does the same with any connection, and runs no in_transaction handler:

import myxo.events

async with engine.begin() as tx:
    await myxo.events.publish(tx, Noted(note_id))

Reading with your own position¶

await self.events.read(after, (Noted,), wait) returns the events after a position, waiting up to wait seconds, for a reader that keeps its own place. It returns at once when the process begins to stop. Push uses it.

Retention¶

Handled events are kept an hour, for readers that resume, then pruned by one process. A group's row in myxo_groups holds its position, and pruning waits for every group. So a group that stops for good holds events back until you remove its row:

delete from myxo_groups where name = 'search';

How it runs¶

On PostgreSQL, publishing takes a lock until the commit, so that positions follow commit order and a consumer that has seen an event has seen every event before it. NOTIFY wakes the consumers. On SQLite, the single writer keeps the order, and consumers look every 50 ms.