Outbox¶

Introduction¶

When handling a message, it’s typical to send messages (send commands or publish events) to various Mersal apps (including the local app that is currently handling the message). There can be two scenarios where the final outcome results in a broken state.

Message handling fails (e.g. data is not persisted) but the messages are sent.
Message handling succeeds but the sent messages fail.

Both are undesirable states; the first can result in publishing false information to other systems (e.g. RegisterStudentMessageHandler that fails to persist data but publishes a StudentRegisteredEvent to other systems). The second will lead to broken business processes (e.g a finance system that is subscribing to StudentRegisteredEvent to complete financial registration will fail to function.)

The solution is to persist outgoing messages along the business data using the same persistence transaction. A successful transaction means the business data and the outgoing messages are stored. A failed transactions means nothing is stored.

The outgoing messages are said to be stored in an outbox hence the name “Outbox pattern”

Since there needs to be a transaction, using the outbox pattern requires a persistence mechanism that supports atomic transactions (see [1] [page 145-146] for implementing this pattern with NoSQL databases).

Usage¶

Notes¶

The outbox pattern can be used in Mersal by passing an instance of OutboxConfig during initialization. The configuration requires an outbox persistence mechanism that implements OutboxStorage.

# incomplete code

from mersal.outbox import OutboxConfig
from mersal.app import Mersal

app = Mersal(outbox_config=OutboxConfig(...))

Provided implementation of OutboxStorage:

SQLAlchemyOutboxStorage
InMemoryOutboxStorage, only used for testing since it has no concept of a transaction.

When using the outbox feature, it’s important to not commit nor close the database transaction within the message handlers. That action should be handled by the outbox storage or during the unit of work step.

Examples¶

Outbox with SQLAlchemy

outbox_sqlalchemy.py¶

from dataclasses import dataclass, field
from typing import Any, cast

from mersal_alchemy.sqlalchemy_outbox_storage import (
    SQLAlchemyOutboxStorageConfig,
)
from mersal_msgspec import MsgspecSerializer
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from mersal.activation import BuiltinHandlerActivator
from mersal.app import Mersal
from mersal.outbox import OutboxConfig
from mersal.pipeline import MessageContext
from mersal.transport.in_memory import (
    InMemoryNetwork,
    InMemoryTransportConfig,
)

__all__ = (
    "AddUser",
    "AddUserMessageHandler",
    "Base",
    "PromoteUser",
    "PromoteUserMessageHandler",
    "PromotedUsersRepo",
    "User",
    "app_factory",
)


@dataclass
class AddUser:
    username: str


@dataclass
class PromoteUser:
    username: str


NETWORK = InMemoryNetwork()


@dataclass
class PromotedUsersRepo:
    users: list[str] = field(default_factory=list)


PROMOTED_USERS = PromotedUsersRepo()


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    username: Mapped[str] = mapped_column(primary_key=True)


class AddUserMessageHandler:
    def __init__(
        self,
        mersal: Mersal,
        message_context: MessageContext,
        session_factory: async_sessionmaker,
    ) -> None:
        self.mersal = mersal
        self.session = session_factory()
        transaction_context = message_context.transaction_context
        transaction_context.items["db-session"] = self.session

    async def __call__(self, message: AddUser) -> None:
        user = User(username=message.username)
        self.session.add(user)

        # Remember, do not commit session

        await self.mersal.send_local(PromoteUser(username=message.username))


class PromoteUserMessageHandler:
    async def __call__(self, message: PromoteUser) -> Any:
        # please don't do this in production
        PROMOTED_USERS.users.append(message.username)


async def app_factory() -> Mersal:
    engine = create_async_engine("sqlite+aiosqlite://", echo=True)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    session_factory = async_sessionmaker(engine)

    outbox_storage = SQLAlchemyOutboxStorageConfig(
        async_session_factory=session_factory,
        table_name="outbox",
        session_extractor=lambda transaction_context: cast("AsyncSession", transaction_context.items.get("db-session")),
    ).storage
    queue_address = "test-queue"
    transport = InMemoryTransportConfig(network=NETWORK, input_queue_address=queue_address).transport
    activator = BuiltinHandlerActivator()
    activator.register(
        AddUser,
        lambda message_context, mersal: AddUserMessageHandler(
            mersal=mersal,
            message_context=message_context,
            session_factory=session_factory,
        ),
    )
    activator.register(
        PromoteUser,
        lambda _, __: PromoteUserMessageHandler(),
    )
    return Mersal(
        "m1",
        activator,
        transport=transport,
        outbox=OutboxConfig(storage=outbox_storage),
        message_body_serializer=MsgspecSerializer(object_types={AddUser, PromoteUser}),
    )

Looking at the emphasized lines, the database session is created upon handling the message. This same session needs to be shared with the outbox storage. This is done using the transaction context as shown. Other configurations are possible.

Summary¶

Don’t close the database transaction in the message handler.
Use Mersal idempotency feature with the outbox feature. Set it to completely skip message handling.
Preferable to use the outbox feature with the unit of work pattern.

Internal Implementation¶

Once the outgoing messages are persisted in the outbox. A relay checks the outbox at a fixed interval for any stored messages (currently set at 1 second).

Ref [1] and [2] discuss the implementation and alternatives. One alternative is a push based relay. This type of relay needs to be supported by the database.

Road Map¶

See Outbox Project