Advanced¶

Custom ProcessingStep¶

from omni_box.core.pipeline.step import BaseProcessingStep, StepResult


class AuditStep(BaseProcessingStep):
    async def execute(self, event, context):
        await audit_log.write(event)
        return StepResult.next()

Add it to the builder before HandlerExecutionStep.

Custom FetchStrategy¶

Implement omni_box.core.pipeline.strategies.fetch.FetchStrategy[T]. The strategy receives an EventRepository[T] and a worker_id, and must return a list of events with appropriate locking semantics.

FilteredFetchStrategy(sources=[...]) is a quick way to scope a worker to a subset of source values; it composes with the default strategies.

Custom CommitStrategy¶

Implement omni_box.core.pipeline.strategies.commit.CommitStrategy[T]. It receives the final ProcessingContext and is responsible for persisting completion / failure / skipped transitions.

Partitioned tables¶

Use OutboxEventPartitionedDBBase / InboxEventPartitionedDBBase. PostgreSQL requires the partition key (created_at) to appear in every unique index; the partitioned bases handle this for you via __inbox_dedup_index_columns__ and __outbox_conflict_index_*__.

Create child partitions per time window (typically RANGE by month or day) and attach them with pg_partman or your own DDL. The repositories use INSERT ... ON CONFLICT DO NOTHING on the partitioned table, which propagates to the correct child partition.

Deduplication on a partitioned table cannot rely on that index alone, because it has to carry created_at and two retries never share a timestamp. Both repositories serialize the logical identity instead — the inbox for (message_id, consumer_group), the outbox for idempotency_key — with a transaction-scoped advisory lock (pg_advisory_xact_lock) and a lookup before the insert, so a retry, a replay or a concurrent writer gets the existing row back. bulk_create does the same per batch (one lock statement, one lookup). Non-partitioned tables keep the single INSERT ... ON CONFLICT DO NOTHING round trip.

Multi-region / multi-worker¶

• Use DistributedLockingFetchStrategy (auto-picked when the repository implements SupportsDistributedLocking).
• Tune EventProcessorBuilder.with_lease_ttl(seconds) to cover the worst-case handler runtime plus a safety margin.
• Run OmniBoxMaintenanceService.release_stale_locks periodically with a stale_timeout_seconds strictly larger than the lease TTL.
• Remember that CircuitBreakerStep is process-local — replicas trip independently. Add Redis-backed counters if you need cross-replica coordination.

Custom DLQ sink¶

from omni_box.core.pipeline.steps import DLQStep, DLQStorage


class KafkaDLQ(DLQStorage):
    def __init__(self, producer, topic):
        self._producer = producer
        self._topic = topic

    async def move_to_dlq(self, event, error: str) -> None:
        await self._producer.send_and_wait(
            self._topic,
            key=str(event.id).encode(),       # idempotent key — avoids dup on replay
            value=event.model_dump_json().encode(),
            headers=[("dlq-error", error.encode())],
        )

builder.add_step(DLQStep(KafkaDLQ(producer, "events.dlq")))

Because DLQStep is best-effort, the unique key keeps replays from inflating the DLQ.

Replacing the broker¶

OutboxPublisher and InboxConsumerRunner operate against the broker protocols:

• EventPublisher — async def publish(event, repo).
• EventConsumer — start, stop, async def getone() -> ConsumedMessage.
• AckHandle — attached to each ConsumedMessage: async def commit(), and async def rewind(), which the runner calls on a message it does not commit. The default rewind() does nothing, which suits a broker that redelivers an unacknowledged message by itself (RabbitMQ after a nack with requeue, SQS after the visibility timeout). Where a commit acknowledges every earlier message, as a Kafka offset does, rewind() must make the next getone() return this message again, or the next commit loses it.

Implement those for RabbitMQ, Pulsar, NATS, etc. The Kafka adapter is just a reference implementation; nothing about the core pipeline assumes Kafka.