spakky-outbox

spakky-outbox는 Integration Event를 비즈니스 데이터와 같은 transaction에 원자적으로 기록하고 별도 Relay로 전송하기 위한 계약을 제공합니다. 브로커가 수락 가능한 레코드는 확인될 때까지 재전송되지만, 영구적인 레코드 귀속 거부는 retry 소진 후 성공 전달 없이 abandoned 처리될 수 있습니다.

Outbox 패턴 — 원자적 기록과 Relay 상태 계약

플러그인 진입점

Plugin initialization entry point.

initialize(app)

Initialize the Outbox plugin.

Parameters:

Name	Type	Description	Default
app	SpakkyApplication	The Spakky application instance.	required

Source code in core/spakky-outbox/src/spakky/outbox/main.py

def initialize(app: SpakkyApplication) -> None:
    """Initialize the Outbox plugin.

    Args:
        app: The Spakky application instance.
    """
    app.add(OutboxConfig)
    app.add(OutboxEventBus)
    app.add(AsyncOutboxEventBus)
    app.add(OutboxRelayBackgroundService)
    app.add(AsyncOutboxRelayBackgroundService)

EventBus

Outbox Event Bus — sync and async implementations replacing IEventBus/IAsyncEventBus via @Primary.

OutboxEventBus(storage, propagator, auth_snapshot_headers=None)

Bases: IEventBus

Intercepts integration events and stores them in the Outbox table (sync).

Replaces the default DirectEventBus so that events are persisted atomically within the same database transaction as the business data.

Source code in core/spakky-outbox/src/spakky/outbox/bus/outbox_event_bus.py

def __init__(
    self,
    storage: IOutboxStorage,
    propagator: ITracePropagator,
    auth_snapshot_headers: AuthContextSnapshotHeaderInjector | None = None,
) -> None:
    self._storage = storage
    self._propagator = propagator
    self._auth_snapshot_headers = (
        auth_snapshot_headers or AuthContextSnapshotHeaderInjector()
    )

AsyncOutboxEventBus(storage, propagator, auth_snapshot_headers=None)

Bases: IAsyncEventBus

Intercepts integration events and stores them in the Outbox table (async).

Replaces the default AsyncDirectEventBus so that events are persisted atomically within the same database transaction as the business data.

Source code in core/spakky-outbox/src/spakky/outbox/bus/outbox_event_bus.py

def __init__(
    self,
    storage: IAsyncOutboxStorage,
    propagator: ITracePropagator,
    auth_snapshot_headers: AuthContextSnapshotHeaderInjector | None = None,
) -> None:
    self._storage = storage
    self._propagator = propagator
    self._auth_snapshot_headers = (
        auth_snapshot_headers or AuthContextSnapshotHeaderInjector()
    )

포트

Outbox storage port.

IOutboxStorage

Bases: ABC

Synchronous outbox message storage abstraction.

save(message) abstractmethod

Save message within the current transaction.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
def save(self, message: OutboxMessage) -> None:
    """Save message within the current transaction."""

fetch_pending(limit, max_retry) abstractmethod

Claim unpublished messages for this relay instance (with lock).

A partition key must be claimed whole: an implementation may only hand out messages of a key when it also claims that key's oldest message that is neither published nor abandoned. Otherwise two relay instances publish one key in parallel and lose the ordering the key exists to provide. Messages without a partition key carry no such constraint.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
def fetch_pending(self, limit: int, max_retry: int) -> list[OutboxMessage]:
    """Claim unpublished messages for this relay instance (with lock).

    A partition key must be claimed whole: an implementation may only hand
    out messages of a key when it also claims that key's oldest message that
    is neither published nor abandoned. Otherwise two relay instances
    publish one key in parallel and lose the ordering the key exists to
    provide. Messages without a partition key carry no such constraint.
    """

mark_published(message_id) abstractmethod

Mark a message as published.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
def mark_published(self, message_id: UUID) -> None:
    """Mark a message as published."""

increment_retry(message_id) abstractmethod

Increment the retry count of a message.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
def increment_retry(self, message_id: UUID) -> None:
    """Increment the retry count of a message."""

mark_abandoned(message_id) abstractmethod

Record that the relay gave up on a message after exhausting retries.

The message leaves the pending queue without being published, so a partition key never waits forever on a message that will not be retried again. An implementation must keep the record and the reason readable — the operator has to be able to find what was dropped.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
def mark_abandoned(self, message_id: UUID) -> None:
    """Record that the relay gave up on a message after exhausting retries.

    The message leaves the pending queue without being published, so a
    partition key never waits forever on a message that will not be retried
    again. An implementation must keep the record and the reason readable —
    the operator has to be able to find what was dropped.
    """

IAsyncOutboxStorage

Bases: ABC

Asynchronous outbox message storage abstraction.

save(message) abstractmethod async

Save message within the current transaction.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
async def save(self, message: OutboxMessage) -> None:
    """Save message within the current transaction."""

fetch_pending(limit, max_retry) abstractmethod async

Claim unpublished messages for this relay instance (with lock).

A partition key must be claimed whole: an implementation may only hand out messages of a key when it also claims that key's oldest message that is neither published nor abandoned. Otherwise two relay instances publish one key in parallel and lose the ordering the key exists to provide. Messages without a partition key carry no such constraint.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
async def fetch_pending(self, limit: int, max_retry: int) -> list[OutboxMessage]:
    """Claim unpublished messages for this relay instance (with lock).

    A partition key must be claimed whole: an implementation may only hand
    out messages of a key when it also claims that key's oldest message that
    is neither published nor abandoned. Otherwise two relay instances
    publish one key in parallel and lose the ordering the key exists to
    provide. Messages without a partition key carry no such constraint.
    """

mark_published(message_id) abstractmethod async

Mark a message as published.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
async def mark_published(self, message_id: UUID) -> None:
    """Mark a message as published."""

increment_retry(message_id) abstractmethod async

Increment the retry count of a message.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
async def increment_retry(self, message_id: UUID) -> None:
    """Increment the retry count of a message."""

mark_abandoned(message_id) abstractmethod async

Record that the relay gave up on a message after exhausting retries.

The message leaves the pending queue without being published, so a partition key never waits forever on a message that will not be retried again. An implementation must keep the record and the reason readable — the operator has to be able to find what was dropped.

Source code in core/spakky-outbox/src/spakky/outbox/ports/storage.py

@abstractmethod
async def mark_abandoned(self, message_id: UUID) -> None:
    """Record that the relay gave up on a message after exhausting retries.

    The message leaves the pending queue without being published, so a
    partition key never waits forever on a message that will not be retried
    again. An implementation must keep the record and the reason readable —
    the operator has to be able to find what was dropped.
    """

Relay

Outbox relay module.

Outbox Relay Background Services (sync and async).

OutboxRelayBackgroundService(storage, transport, config)

Bases: AbstractBackgroundService

Polls the Outbox storage and relays pending messages to the transport (sync).

Initialize with storage, transport, and config dependencies.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

def __init__(
    self,
    storage: IOutboxStorage,
    transport: IEventTransport,
    config: OutboxConfig,
) -> None:
    """Initialize with storage, transport, and config dependencies."""
    self._storage = storage
    self._transport = transport
    self._config = config

initialize()

No-op initialization for the relay service.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

@override
def initialize(self) -> None:
    """No-op initialization for the relay service."""
    return

dispose()

No-op disposal for the relay service.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

@override
def dispose(self) -> None:
    """No-op disposal for the relay service."""
    return

run()

Poll the outbox storage and relay pending messages until stopped.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

@override
def run(self) -> None:
    """Poll the outbox storage and relay pending messages until stopped."""
    while not self._stop_event.is_set():
        self._relay_batch()
        self._stop_event.wait(timeout=self._config.polling_interval_seconds)

__register_refusal(message, halted_partition_keys)

Spend one retry on a refused message, or abandon it when none is left.

A message whose budget is spent will never be fetched again, so it is abandoned rather than left behind: that releases its partition key, which would otherwise wait forever on a message nobody will retry, and keeps the record findable. A message that still has budget holds its key back until it is delivered, so nothing of that key overtakes it.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

def __register_refusal(
    self,
    message: OutboxMessage,
    halted_partition_keys: set[str],
) -> None:
    """Spend one retry on a refused message, or abandon it when none is left.

    A message whose budget is spent will never be fetched again, so it is
    abandoned rather than left behind: that releases its partition key,
    which would otherwise wait forever on a message nobody will retry, and
    keeps the record findable. A message that still has budget holds its key
    back until it is delivered, so nothing of that key overtakes it.
    """
    if message.retry_count + 1 >= self._config.max_retry_count:
        self._storage.mark_abandoned(message.id)
        return
    self._storage.increment_retry(message.id)
    if message.partition_key is not None:
        halted_partition_keys.add(message.partition_key)

__publish_one_at_a_time(messages, halted_partition_keys)

Replay a batch whose flush failed, confirming one message at a time.

A batch flush reports that the broker refused something but not what, so neither the retry budget nor the partition key hold-back can be aimed without replaying. Confirming one message per flush puts the broker's verdict on the message that earned it: the refused message spends its budget and holds back the rest of its key, while its neighbours publish. Only a refusal counts — a transport that cannot reach the broker at all stops the replay untouched, because an outage is no message's fault. The replay can re-deliver a record the failed flush had already accepted — delivery stays at-least-once, which consumers already assume.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

def __publish_one_at_a_time(
    self,
    messages: list[OutboxMessage],
    halted_partition_keys: set[str],
) -> None:
    """Replay a batch whose flush failed, confirming one message at a time.

    A batch flush reports that the broker refused something but not what, so
    neither the retry budget nor the partition key hold-back can be aimed
    without replaying. Confirming one message per flush puts the broker's
    verdict on the message that earned it: the refused message spends its
    budget and holds back the rest of its key, while its neighbours publish.
    Only a refusal counts — a transport that cannot reach the broker at all
    stops the replay untouched, because an outage is no message's fault.
    The replay can re-deliver a record the failed flush had already accepted
    — delivery stays at-least-once, which consumers already assume.
    """
    for message in messages:
        if message.partition_key in halted_partition_keys:
            continue
        try:
            self._transport.send(
                message.event_name,
                message.payload,
                message.headers,
                message.partition_key,
            )
            self._transport.flush()
        except EventDeliveryRejectedError:
            logger.exception("Broker refused outbox message %s", message.id)
            self.__register_refusal(message, halted_partition_keys)
            continue
        except Exception:
            # The transport itself is failing — a closed client, a lost
            # connection, a timeout. That is no single message's fault, so
            # the rest of the batch is left pending instead of spending
            # budgets that would abandon healthy messages during an outage.
            logger.exception(
                "Transport failed while confirming outbox message %s",
                message.id,
            )
            return
        self._storage.mark_published(message.id)

AsyncOutboxRelayBackgroundService(storage, transport, config)

Bases: AbstractAsyncBackgroundService

Polls the Outbox storage and relays pending messages to the transport (async).

Initialize with async storage, transport, and config dependencies.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

def __init__(
    self,
    storage: IAsyncOutboxStorage,
    transport: IAsyncEventTransport,
    config: OutboxConfig,
) -> None:
    """Initialize with async storage, transport, and config dependencies."""
    self._storage = storage
    self._transport = transport
    self._config = config

initialize_async() async

No-op async initialization for the relay service.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

@override
async def initialize_async(self) -> None:
    """No-op async initialization for the relay service."""
    return

dispose_async() async

No-op async disposal for the relay service.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

@override
async def dispose_async(self) -> None:
    """No-op async disposal for the relay service."""
    return

run_async() async

Poll the outbox storage and relay pending messages asynchronously.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

@override
async def run_async(self) -> None:
    """Poll the outbox storage and relay pending messages asynchronously."""
    while not self._stop_event.is_set():
        await self._relay_batch()
        try:
            await wait_for(
                self._stop_event.wait(),
                timeout=self._config.polling_interval_seconds,
            )
            break
        except TimeoutError:
            continue

__register_refusal(message, halted_partition_keys) async

Spend one retry on a refused message, or abandon it when none is left.

A message whose budget is spent will never be fetched again, so it is abandoned rather than left behind: that releases its partition key, which would otherwise wait forever on a message nobody will retry, and keeps the record findable. A message that still has budget holds its key back until it is delivered, so nothing of that key overtakes it.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

async def __register_refusal(
    self,
    message: OutboxMessage,
    halted_partition_keys: set[str],
) -> None:
    """Spend one retry on a refused message, or abandon it when none is left.

    A message whose budget is spent will never be fetched again, so it is
    abandoned rather than left behind: that releases its partition key,
    which would otherwise wait forever on a message nobody will retry, and
    keeps the record findable. A message that still has budget holds its key
    back until it is delivered, so nothing of that key overtakes it.
    """
    if message.retry_count + 1 >= self._config.max_retry_count:
        await self._storage.mark_abandoned(message.id)
        return
    await self._storage.increment_retry(message.id)
    if message.partition_key is not None:
        halted_partition_keys.add(message.partition_key)

__publish_one_at_a_time(messages, halted_partition_keys) async

Replay a batch whose flush failed, confirming one message at a time.

A batch flush reports that the broker refused something but not what, so neither the retry budget nor the partition key hold-back can be aimed without replaying. Confirming one message per flush puts the broker's verdict on the message that earned it: the refused message spends its budget and holds back the rest of its key, while its neighbours publish. Only a refusal counts — a transport that cannot reach the broker at all stops the replay untouched, because an outage is no message's fault. The replay can re-deliver a record the failed flush had already accepted — delivery stays at-least-once, which consumers already assume.

Source code in core/spakky-outbox/src/spakky/outbox/relay/relay.py

async def __publish_one_at_a_time(
    self,
    messages: list[OutboxMessage],
    halted_partition_keys: set[str],
) -> None:
    """Replay a batch whose flush failed, confirming one message at a time.

    A batch flush reports that the broker refused something but not what, so
    neither the retry budget nor the partition key hold-back can be aimed
    without replaying. Confirming one message per flush puts the broker's
    verdict on the message that earned it: the refused message spends its
    budget and holds back the rest of its key, while its neighbours publish.
    Only a refusal counts — a transport that cannot reach the broker at all
    stops the replay untouched, because an outage is no message's fault.
    The replay can re-deliver a record the failed flush had already accepted
    — delivery stays at-least-once, which consumers already assume.
    """
    for message in messages:
        if message.partition_key in halted_partition_keys:
            continue
        try:
            await self._transport.send(
                message.event_name,
                message.payload,
                message.headers,
                message.partition_key,
            )
            await self._transport.flush()
        except EventDeliveryRejectedError:
            logger.exception("Broker refused outbox message %s", message.id)
            await self.__register_refusal(message, halted_partition_keys)
            continue
        except Exception:
            # The transport itself is failing — a closed client, a lost
            # connection, a timeout. That is no single message's fault, so
            # the rest of the batch is left pending instead of spending
            # budgets that would abandon healthy messages during an outage.
            logger.exception(
                "Transport failed while confirming outbox message %s",
                message.id,
            )
            return
        await self._storage.mark_published(message.id)

공통

Outbox configuration.

OutboxConfig()

Bases: BaseSettings

Outbox plugin configuration loaded from environment variables.

Load outbox configuration from environment variables.

Source code in core/spakky-outbox/src/spakky/outbox/common/config.py

def __init__(self) -> None:
    """Load outbox configuration from environment variables."""
    super().__init__()

Outbox message model.

OutboxMessage(id, event_name, payload, headers, created_at, published_at=None, retry_count=0, claimed_at=None, partition_key=None, abandoned_at=None) dataclass

Persistence-agnostic Outbox message model.

partition_key = field(default=None) class-attribute instance-attribute

Key pinning the message to one broker partition. None spreads round-robin.

abandoned_at = field(default=None) class-attribute instance-attribute

When the relay gave up on this message. None means it is still pending.

에러

Outbox error classes.

AbstractSpakkyOutboxError

Bases: AbstractSpakkyFrameworkError, ABC

Base exception for Spakky Outbox errors.