In-Process Event Bus in a Monolith: Domain Events Done Right

What domain events are and why a monolith needs an in-process bus

This article treats the in-process event bus as a module boundary tool with explicit transactional semantics. It is not a lightweight message broker. That distinction drives every design decision that follows.

A domain event is a fact, named in the past tense, that an aggregate records after a state change: OrderPlaced, PaymentCaptured, ShipmentDispatched. It cannot be rejected, because it already happened. That separates it from a command and from an integration event.

The payload should carry identifiers and the values that changed. It should not carry a live entity reference. If handlers receive the aggregate object itself, they can mutate it or reach into the publisher's internals.

In a modular monolith, the bus lets Ordering publish OrderPlaced without knowing that Inventory, Billing and Notifications react to it. Without it, the order service calls each module directly. Its dependency list then grows with every new feature, and its owner gets pulled into reviews for code they don't own.

Adopt a bus when several modules have different owners and cross-module calls have started to tangle. If one small team owns a single module, direct calls are clearer.

How it works: the minimal bus contract and dispatch from the unit of work

The contract can stay small. A bus needs three things: publish(event), subscribe(eventType, handler), and events defined as typed, immutable classes such as OrderPlaced or InvoiceVoided. Use the event's type as the routing key, not a string topic. That way the compiler or type checker catches a renamed event, and a handler's signature documents exactly what it consumes.

Handlers should be registered through the DI container at startup, not wired by hand. Each module declares its own handlers, and the bus finds every registered handler for an event type when it is resolved. The publisher never imports or names a subscriber, so a module can add a reaction to OrderPlaced without touching the Orders module. A missing registration also shows up as a startup failure instead of a silent no-op in production.

Domain code should not depend on the bus at all. Aggregates record events in an internal list as state changes, for example when order.place() appends OrderPlaced. The unit of work, or the repository's save path, drains those lists from every tracked aggregate and hands the events to the bus. This keeps aggregates easy to unit test, since you assert on recorded events and need no mocks. It also gives you one place that controls when dispatch happens relative to the transaction, which is the decision covered next.

CopyOrders module                               Other modules
┌───────────────┐
│ Order         │  place()
│  events: [    │──────┐
│   OrderPlaced]│      │ collect
└───────────────┘      ▼
               ┌──────────────┐  publish   ┌─────┐   ┌─────────────────────┐
               │ Unit of Work │──────────▶│ Bus │──▶│ Billing handler     │
               │ (save/commit)│            └─────┘   ├─────────────────────┤
               └──────────────┘               └────▶│ Inventory handler   │
                                                     └─────────────────────┘

Copytype Ctor<T> = new (...args: any[]) => T;
type Handler<T> = (event: T) => Promise<void>;

// Typed, immutable event: the class itself is the routing key.
export class OrderPlaced {
  constructor(readonly orderId: string, readonly totalCents: number) {}
}

export class EventBus {
  private handlers = new Map<Function, Handler<any>[]>();
  subscribe<T>(type: Ctor<T>, handler: Handler<T>): void {
    this.handlers.set(type, [...(this.handlers.get(type) ?? []), handler]);
  }
  async publish(event: object): Promise<void> {
    for (const h of this.handlers.get(event.constructor) ?? []) await h(event);
  }
}

// Aggregates only record events. They hold no reference to the bus.
export abstract class AggregateRoot {
  private pending: object[] = [];
  protected record(event: object): void { this.pending.push(event); }
  pullEvents(): object[] { const e = this.pending; this.pending = []; return e; }
}

export class Order extends AggregateRoot {
  constructor(readonly id: string, private totalCents: number) { super(); }
  place(): void { this.record(new OrderPlaced(this.id, this.totalCents)); }
}

// The single place that decides when recorded events reach the bus.
export class UnitOfWork {
  private tracked: AggregateRoot[] = [];
  constructor(private bus: EventBus) {}
  track(aggregate: AggregateRoot): void { this.tracked.push(aggregate); }
  async commit(): Promise<void> {
    // ...persist tracked aggregates and commit the transaction here...
    const events = this.tracked.flatMap((a) => a.pullEvents());
    for (const e of events) await this.bus.publish(e);
  }
}
// At startup the Billing module's registration code calls
// bus.subscribe(OrderPlaced, ...); Orders never imports Billing.

A minimal in-process bus keyed by event class: aggregates record events internally, and the unit of work drains and publishes them on commit, so domain code never depends on the bus.

The core decision: in-transaction vs after-commit handlers

The diagram hides the decision that matters most: when each handler runs relative to the commit.

In-transaction handlers run synchronously inside the same unit of work. If the inventory handler throws, the order rolls back too. You get atomic consistency, but every handler's failure modes become yours, and its latency extends the transaction while row locks are held.

After-commit handlers run once the commit succeeds. A failing handler cannot undo the order. However, if the process dies between commit and dispatch, the event is gone. Section 4's outbox closes that gap.

Per-handler rule: if the system would be wrong (not just stale) when the handler's write is missing, it runs in-transaction. Everything else runs after commit. The bus exposes this as two registration phases, and the unit of work drains the first before committing and the second after.

Handlers that publish events cascade. Queue new events into the current phase's buffer instead of dispatching recursively, and cap the depth so an A→B→A cycle fails loudly.

Pushing after-commit handlers onto a thread pool does not make them independent. They share the process's CPU, memory, and connection pool, and queued work vanishes on shutdown or crash.

Copyimport logging
from collections import defaultdict, deque

MAX_DEPTH = 5  # an A->B->A cycle hits this cap and fails loudly
IN_TX, AFTER_COMMIT = "in_tx", "after_commit"

class EventBus:
    def __init__(self):
        self.handlers = {IN_TX: defaultdict(list), AFTER_COMMIT: defaultdict(list)}

    def register(self, event_type, handler, phase):
        self.handlers[phase][event_type].append(handler)

class UnitOfWork:
    def __init__(self, bus, conn):
        self.bus, self.conn = bus, conn  # conn: a DB-API connection
        self.queues = {IN_TX: deque(), AFTER_COMMIT: deque()}

    def publish(self, event, depth=0, phases=(IN_TX, AFTER_COMMIT)):
        # Buffer only; never dispatch recursively from inside a handler.
        for phase in phases:
            self.queues[phase].append((event, depth))

    def _drain(self, phase, isolate):
        # After commit, cascaded events can only reach after-commit handlers.
        phases = (IN_TX, AFTER_COMMIT) if phase == IN_TX else (AFTER_COMMIT,)
        while self.queues[phase]:
            event, depth = self.queues[phase].popleft()
            if depth > MAX_DEPTH:
                raise RuntimeError(f"event cascade too deep: {type(event).__name__}")
            for handler in self.bus.handlers[phase][type(event)]:
                emit = lambda e, d=depth: self.publish(e, d + 1, phases)
                try:
                    handler(event, emit)
                except Exception:
                    if not isolate:
                        raise  # in-tx failure must roll back the publisher
                    logging.exception("after-commit handler failed")

    def commit(self):
        try:
            self._drain(IN_TX, isolate=False)  # runs while row locks are held
            self.conn.commit()
        except Exception:
            self.conn.rollback()
            raise
        # If the process dies here, these events are lost unless an outbox exists.
        self._drain(AFTER_COMMIT, isolate=True)

A minimal unit of work that drains in-transaction handlers before commit (sharing rollback), drains after-commit handlers afterward with per-handler failure isolation, and buffers cascaded events with a depth cap so cycles fail loudly.

Implementing a transactional outbox inside the monolith

The fix is to make the event part of the data. Inside the same transaction that changes the aggregate, insert one row per event into an outbox table. The row holds id (a UUID generated at raise time), event_type, payload (serialized JSON), occurred_at, attempts, last_error, and a nullable processed_at. If the business write commits, the event commits too. If it rolls back, the event never existed.

A background dispatcher handles delivery. It selects a batch of unprocessed rows ordered by occurred_at, locks them with FOR UPDATE SKIP LOCKED so multiple instances don't collide, invokes the handlers, and sets processed_at. A crash between delivery and marking means the row is picked up again. Delivery is therefore at-least-once. Each handler must record the event id in a processed_events table keyed by (handler_name, event_id), inside its own transaction, and skip the event if that row already exists.

On failure, increment attempts, store the error, and push a next_attempt_at forward with exponential backoff. After a fixed ceiling, mark the row dead so one poison event cannot block the batch. Alert on dead rows.

Polling has a cost: an indexed query every interval, even when the system is idle. A partial index on unprocessed rows keeps that query cheap. Prune processed rows on a schedule, or the table will grow without bound.

Ordering is per aggregate at best. Retries and concurrent dispatchers reorder events across types, so handlers must not assume any global sequence.

Copy-- PostgreSQL. Outbox rows are inserted in the same transaction as the aggregate change.
CREATE TABLE outbox (
  id              uuid        PRIMARY KEY,           -- generated when the event is raised
  event_type      text        NOT NULL,
  payload         jsonb       NOT NULL,
  occurred_at     timestamptz NOT NULL DEFAULT now(),
  attempts        int         NOT NULL DEFAULT 0,
  last_error      text,
  next_attempt_at timestamptz NOT NULL DEFAULT now(),
  processed_at    timestamptz,
  dead_at         timestamptz                        -- set once the retry ceiling is hit
);
-- Partial index keeps the idle poll cheap: only pending rows are indexed.
CREATE INDEX outbox_pending ON outbox (next_attempt_at)
  WHERE processed_at IS NULL AND dead_at IS NULL;

-- Dispatcher batch (one transaction): claim rows other instances have not locked.
SELECT id, event_type, payload FROM outbox
WHERE processed_at IS NULL AND dead_at IS NULL AND next_attempt_at <= now()
ORDER BY occurred_at LIMIT 100
FOR UPDATE SKIP LOCKED;
-- After handlers succeed for a row:
UPDATE outbox SET processed_at = now() WHERE id = $1;
-- After a failure: exponential backoff, dead after 10 attempts (old attempts used on the right).
UPDATE outbox SET attempts = attempts + 1, last_error = $2,
  next_attempt_at = now() + interval '1 second' * power(2, attempts),
  dead_at = CASE WHEN attempts + 1 >= 10 THEN now() END
WHERE id = $1;

-- Idempotency: each handler records the event id in its own transaction.
CREATE TABLE processed_events (
  handler_name text NOT NULL, event_id uuid NOT NULL,
  processed_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (handler_name, event_id)
);
INSERT INTO processed_events (handler_name, event_id) VALUES ('send_receipt', $1)
ON CONFLICT DO NOTHING;  -- 0 rows affected => already handled: skip side effects, commit.

-- Scheduled pruning so the table does not grow without bound.
DELETE FROM outbox WHERE processed_at < now() - interval '7 days';

A PostgreSQL outbox schema with a partial index, the SKIP LOCKED claim query, success and backoff/dead-letter updates, a per-handler idempotency guard, and a pruning statement.

Keeping module boundaries intact

An event type belongs in the publishing module's public contract package, next to its other exported interfaces. Subscribers import that package and nothing else from the publisher. This keeps the event a deliberate, versioned API rather than an accidental view of internal classes. When the publisher refactors its domain model, the contract package is the only surface that has to stay stable.

Payloads should carry identifiers and immutable values: an order ID, a customer ID, a total as a money value, a timestamp. Never pass entities or ORM-managed objects. A lazy-loaded entity handed to a handler running after commit may hit a closed session. It also gives the subscriber a live handle into the publisher's persistence layer.

Payload size is a real trade-off. If you send only an ID, every subscriber calls back into the publisher to fetch state, which re-creates the synchronous coupling the event was meant to remove. If you send a full snapshot, you leak internal fields that subscribers start depending on. A workable rule is to include the facts that changed and the values most subscribers need, and nothing that only exists for the publisher's own bookkeeping.

Enforce the boundary mechanically. Add an architecture test or build-level dependency rule that fails when any module imports another module's non-contract packages. Code review alone will not hold this line.

Isolate handler failures too. For after-commit handlers, the dispatcher should invoke each handler separately. It should catch, log, and record a failure without skipping the remaining handlers, so one broken subscriber cannot silently starve the others.

Copy// Architecture test (ArchUnit + JUnit 5): code outside the orders module may use
// com.example.shop.orders.api (the contract), never orders.internal.
import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.noClasses;
import com.tngtech.archunit.core.importer.ClassFileImporter;
import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.function.Consumer;
import java.util.logging.Level;
import java.util.logging.Logger;

class ModuleBoundaryTest {
    @Test
    void otherModulesSeeOnlyTheOrdersContract() {
        noClasses().that().resideOutsideOfPackage("com.example.shop.orders..")
            .should().dependOnClassesThat().resideInAPackage("com.example.shop.orders.internal..")
            .check(new ClassFileImporter().importPackages("com.example.shop"));
    }
}

// Lives in orders.api: IDs and immutable values only, no entities or ORM objects.
record OrderPlaced(java.util.UUID orderId, java.util.UUID customerId,
                   java.math.BigDecimal total, String currency, java.time.Instant placedAt) {}

interface FailureStore { void record(String handler, OrderPlaced event, Exception error); }

// After-commit dispatch: each handler gets its own try/catch, so one failure
// is logged and recorded for retry without skipping the remaining handlers.
final class AfterCommitDispatcher {
    private static final Logger LOG = Logger.getLogger(AfterCommitDispatcher.class.getName());
    private final List<Consumer<OrderPlaced>> handlers;
    private final FailureStore failures;

    AfterCommitDispatcher(List<Consumer<OrderPlaced>> handlers, FailureStore failures) {
        this.handlers = List.copyOf(handlers);
        this.failures = failures;
    }

    void dispatch(OrderPlaced event) {
        for (Consumer<OrderPlaced> handler : handlers) {
            try { handler.accept(event); }
            catch (RuntimeException e) {
                LOG.log(Level.WARNING, "Handler failed for order " + event.orderId(), e);
                failures.record(handler.getClass().getName(), event, e);
            }
        }
    }
}

An ArchUnit test that fails the build when any class outside the orders module imports orders.internal, plus a contract event carrying only IDs and immutable values and an after-commit dispatcher that isolates each handler's failure.

Testing and observability for implicit control flow

Once events are in place, the call site no longer shows what happens next. order.place() reads as a self-contained operation, but three modules may react to it. Tests and telemetry have to make that hidden control flow visible again.

Testing at three levels

Aggregates. Assert on the events an aggregate records, not on downstream effects. A test that places an order should check that exactly one OrderPlaced was raised with the expected order ID and total. This keeps aggregate tests fast and independent of any subscriber.

Handlers. Test each handler in isolation by constructing the event directly and calling the handler. This also forces handlers to depend only on the event payload, not on state leaked from the publisher.

Dispatch timing. Write a small number of integration tests against a real database. One should confirm that an in-transaction handler's writes roll back when the outer transaction fails. Another should confirm that an after-commit handler never runs if the commit does not happen. These are the guarantees most likely to break silently during a refactor.

Making runtime flow visible

For every dispatch, log the event type, a correlation ID inherited from the originating request, the handler name, and the outcome (success, failure, or duration). If your tracing setup supports it, open a child span per handler. A single request then appears as one trace that fans out across modules.

At startup, have the bus emit a subscriber map that lists each event type and its registered handlers. New engineers can read it to answer "what reacts to this?" without searching the codebase. It also exposes events that no handler subscribes to.

When to avoid it, alternatives, and impact on cost and engineering time

When not to use it

Skip the bus inside a single module. If Orders needs to recalculate a total after adding a line item, a method call is clearer and the stack trace stays readable. Small codebases with one team rarely gain enough to justify the indirection. Be honest about what the bus does not provide: while modules share a database and a release cycle, you get no independent scaling, deployment, or failure isolation. A handler that leaks memory still takes down the whole process.

Alternatives

Cost versus time saved

The upfront cost is the outbox and relay, idempotency keys and deduplication in every after-commit handler, and tooling such as the subscriber map so the hidden control flow stays visible. You get the payoff when a new reaction becomes a new handler instead of an edit inside another team's module, which reduces merge conflicts and lets teams work in parallel. Extraction also gets easier later. Outbox-backed integration events can be relayed to a real broker, and publishers and most handlers need only limited changes because they already treat delivery as asynchronous and repeatable.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to in-process event bus monolith, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.