Idempotent Event Handlers

The Problem

An event handler processes OrderPlaced events and awards loyalty points:

# fragment
@domain.event_handler(part_of=Order)
class LoyaltyEventHandler:
    @handle(OrderPlaced)
    def award_points(self, event: OrderPlaced):
        repo = current_domain.repository_for(CustomerLoyalty)
        loyalty = repo.get(event.customer_id)
        loyalty.add_points(int(event.total))
        repo.add(loyalty)

In production, the Protean server processes events from the event store via subscriptions. The subscription tracks its position in the stream, advancing as events are processed. But the server crashes between processing the event and persisting the updated position. On restart, the subscription resumes from its last saved position, and delivers the same OrderPlaced event again.

The handler runs a second time. The customer gets double loyalty points.

This is not a rare edge case. It happens routinely in distributed systems:

Server restart during processing: The event was handled, the aggregate was persisted, but the subscription position wasn't updated before the crash.
Broker redelivery: The message broker delivered the event, but the consumer didn't acknowledge it before disconnecting. The broker redelivers.
Subscription replay: A subscription is restarted from an earlier position for debugging, migration, or recovery.
Network partitions: The handler commits to the database but the acknowledgment to the event store is lost. The event is redelivered.
Competing consumers: In a multi-instance deployment, two instances briefly process the same event before coordination starts.

Unlike commands (covered in Command Idempotency), events represent facts that already happened. You cannot reject an event or return an error to its producer. The Order was placed. The handler must process it, potentially more than once.

The Pattern

Design every event handler to produce the same result whether it processes an event once or multiple times. The aggregate's state after two deliveries should be identical to its state after one.

You make a handler idempotent with one of three strategies:

Naturally idempotent operations: Operations that produce the same result regardless of repetition (set-based, not additive).
Deduplication: Detecting and skipping duplicate deliveries.
Upsert instead of insert: Using database operations that update an existing record instead of failing on a duplicate.

The examples on this page react to events from an Order aggregate:

@domain.aggregate
class Order:
    order_id: Identifier(identifier=True)
    customer_id: Identifier(required=True)
    customer_name: String()
    total: Float(required=True)
    items: List(content_type=Dict)


@domain.event(part_of=Order)
class OrderPlaced:
    order_id: Identifier(required=True)
    customer_id: Identifier(required=True)
    customer_name: String()
    total: Float(required=True)
    items: List(content_type=Dict)
    placed_at: DateTime()


@domain.event(part_of=Order)
class OrderPaid:
    order_id: Identifier(required=True)
    customer_id: Identifier(required=True)
    items: List(content_type=Dict)


@domain.event(part_of=Order)
class OrderShipped:
    order_id: Identifier(required=True)
    shipped_at: DateTime()
    tracking_number: String()

Each handler that reacts to OrderPlaced, OrderPaid or OrderShipped is declared with part_of=Order, even when it changes another aggregate. A handler listens to the stream of the aggregate it is part of, so a handler declared with part_of=CustomerLoyalty would never receive OrderPlaced.

Events vs Commands: Why Handlers Differ

The Command Idempotency pattern covers framework-level deduplication for commands using idempotency keys and Redis-backed caches. Event handlers have different characteristics:

Aspect	Commands	Events
Semantics	Intent to change	Fact that occurred
Can be rejected?	Yes	No (it already happened)
Idempotency key	Caller-provided	Not applicable
Framework dedup	Redis-backed cache (Layers 1-2)	Subscription position; projectors can opt into `idempotent=True`
Handler responsibility	Defense-in-depth	Primary defense

Because events don't carry caller-provided idempotency keys, the handler itself is the primary defense against duplicate processing. The subscription's position tracking provides some protection, but it's not sufficient (position updates are periodic, not per-message). A projector can opt into idempotent=True, which records a marker per event and skips a redelivered one. Event handlers have no such option.

Strategy 1: Naturally Idempotent Operations

The simplest and most reliable approach. If the handler's operation produces the same result whether executed once or ten times, duplicates are harmless.

Set-Based Operations

Operations that set state rather than add to it are naturally idempotent:

@domain.aggregate
class OrderStatus:
    order_id: Identifier(identifier=True)
    status: String(default="placed")
    shipped_at: DateTime()
    tracking_number: String()


@domain.event_handler(part_of=Order)
class OrderStatusEventHandler:
    @handle(OrderShipped)
    def on_order_shipped(self, event: OrderShipped):
        repo = current_domain.repository_for(OrderStatus)
        status = repo.get(event.order_id)
        # Setting status to "shipped" is idempotent.
        # Doing it twice produces the same result.
        status.status = "shipped"
        status.shipped_at = event.shipped_at
        status.tracking_number = event.tracking_number
        repo.add(status)

Setting status = "shipped" ten times leaves the status at "shipped". There's no accumulation, no duplication, no corruption.

More naturally idempotent patterns:

Replacing a value: user.email = event.new_email
Overwriting a flag: order.is_paid = True
Assigning a reference: task.assignee_id = event.user_id
Replacing a value object: account.address = Address(...)

When to Prefer Set-Based Design

If you can design your event handler to use set-based operations instead of additive ones, always do so. It eliminates the need for deduplication entirely.

For example, instead of incrementing a counter:

# fragment
# Additive (NOT idempotent)
def on_order_placed(self, event: OrderPlaced):
    stats.order_count += 1
    stats.total_revenue += event.total

Consider maintaining the full state. record_order keeps the orders keyed by order ID and recalculates the totals from them, so recording the same order twice changes nothing:

@domain.aggregate
class OrderStats:
    stats_id: Identifier(identifier=True)
    order_totals: Dict()
    order_count: Integer(default=0)
    total_revenue: Float(default=0.0)

    def record_order(self, order_id: str, total: float) -> None:
        # Keyed by order ID: recording the same order twice is a no-op
        self.order_totals = {**self.order_totals, order_id: total}
        # Recalculate from the stored orders instead of adding to a running total
        self.order_count = len(self.order_totals)
        self.total_revenue = sum(self.order_totals.values())


@domain.event_handler(part_of=Order)
class OrderStatsEventHandler:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = current_domain.repository_for(OrderStats)
        stats = repo.get("all-orders")
        stats.record_order(event.order_id, event.total)
        repo.add(stats)

Strategy 2: Deduplication

When the operation is inherently additive (incrementing counters, appending to lists, creating new records), you need explicit deduplication.

Check-Before-Act

The simplest deduplication: check whether the event has already been processed before processing it.

@domain.aggregate
class Inventory:
    product_id: Identifier(identifier=True)
    available_quantity: Integer(default=0)
    reserved_for_orders: List(content_type=String)

    def reserve(self, quantity: int, order_id: str) -> None:
        self.available_quantity -= quantity
        self.reserved_for_orders = [*self.reserved_for_orders, order_id]


@domain.event_handler(part_of=Order)
class InventoryEventHandler:
    @handle(OrderPlaced)
    def reserve_inventory(self, event: OrderPlaced):
        repo = current_domain.repository_for(Inventory)

        # One reservation per product: an order can list a product twice
        quantities: dict[str, int] = {}
        for item in event.items:
            product_id = item["product_id"]
            quantities[product_id] = quantities.get(product_id, 0) + item["quantity"]

        for product_id, quantity in quantities.items():
            inventory = repo.get(product_id)

            # Check if this event's reservation was already applied
            if event.order_id in inventory.reserved_for_orders:
                continue  # Already processed, skip

            inventory.reserve(quantity=quantity, order_id=event.order_id)
            repo.add(inventory)

The Inventory aggregate tracks which orders have already reserved stock. If the same OrderPlaced event arrives twice, the second delivery finds the order ID already in reserved_for_orders and skips that product. The handler adds up the quantities for each product first, so an order that lists the same product on two lines reserves the total once.

Track Processed Event IDs

For handlers that process many event types, maintain a set of processed event identifiers:

@domain.aggregate
class CustomerLoyalty:
    customer_id: Identifier(identifier=True)
    points: Integer(default=0)
    processed_events: List(content_type=String)

    def add_points(self, amount: int) -> None:
        self.points += amount


@domain.event_handler(part_of=Order)
class LoyaltyEventHandler:
    @handle(OrderPlaced)
    def award_points(self, event: OrderPlaced):
        repo = current_domain.repository_for(CustomerLoyalty)
        loyalty = repo.get(event.customer_id)

        # Use the event's unique ID for deduplication
        event_id = event._metadata.headers.id
        if event_id in loyalty.processed_events:
            return  # Already processed

        loyalty.add_points(int(event.total))
        # Assign a new list: an in-place append is not tracked as a change
        loyalty.processed_events = [*loyalty.processed_events, event_id]
        repo.add(loyalty)

The handler assigns a new list instead of calling append() on the stored one. Protean tracks a change to a List field when you assign it. An in-place append() on its own is not seen as a change, so saving an aggregate whose only change is the append does not store it.

The processed_events list is a bounded collection of recently processed event IDs. For high-volume aggregates, prune entries older than a retention window (e.g., keep only the last 1000 or entries from the last 24 hours).

Use Event Sequence Numbers

When a handler reads events strictly in event store order, the event's global position can serve as a deduplication marker. The global position grows across the whole store. The stream position (event_store.position) restarts at 0 for each aggregate, so it cannot serve as one watermark shared by several aggregates:

@domain.aggregate
class ProjectionState:
    state_id: Identifier(identifier=True)
    order_count: Integer(default=0)
    total_revenue: Float(default=0.0)
    last_processed_position: Integer(default=-1)

    def apply_order_placed(self, event: OrderPlaced) -> None:
        self.order_count += 1
        self.total_revenue += event.total


@domain.event_handler(part_of=Order)
class ProjectionStateHandler:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = current_domain.repository_for(ProjectionState)
        state = repo.get("order-projection")

        # Use the store-wide global position for deduplication
        event_position = event._metadata.event_store.global_position
        if event_position <= state.last_processed_position:
            return  # Already processed or older

        # Process the event
        state.apply_order_placed(event)
        state.last_processed_position = event_position
        repo.add(state)

This is particularly useful for projectors that rebuild from an event stream and need to track their position.

Use the watermark only where processing stops at the first failure, such as a rebuild that halts on an error. A live event store subscription does not work that way. With recovery on (the default), it records a failed event, moves on to newer events, and retries the failed one later. By then the watermark has moved past the failed event's position, so the handler skips it and the order is never counted. For a live handler, track processed event IDs instead.

Strategy 3: Upsert Operations

For handlers that create new records (like projections), use upsert semantics to handle duplicates at the database level.

Projector Upserts

Projectors are a common case where idempotency matters. When a projector processes OrderPlaced to create an OrderSummary projection, a duplicate delivery would try to create a second record with the same ID:

from protean.exceptions import ObjectNotFoundError

@domain.projection
class OrderSummaryProjection:
    order_id: Identifier(identifier=True)
    customer_name: String()
    status: String()
    total: Float()
    placed_at: DateTime()


@domain.projector(projector_for=OrderSummaryProjection, aggregates=[Order])
class OrderSummaryProjector:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = current_domain.repository_for(OrderSummaryProjection)

        # Check if the projection already exists
        try:
            existing = repo.get(event.order_id)
        except ObjectNotFoundError:
            existing = None

        if existing:
            # Already created: update instead of insert
            existing.customer_name = event.customer_name
            existing.total = event.total
            existing.placed_at = event.placed_at
            existing.status = "placed"
            repo.add(existing)
        else:
            # First time: create
            projection = OrderSummaryProjection(
                order_id=event.order_id,
                customer_name=event.customer_name,
                status="placed",
                total=event.total,
                placed_at=event.placed_at,
            )
            repo.add(projection)

repo.get() raises ObjectNotFoundError when no record has the ID, so the projector catches it to tell a first delivery from a repeat. The projector creates the record on the first delivery and updates it on a duplicate delivery. The result is the same either way.

Existence Guard

For handlers that create an aggregate, check whether one already exists for the source of the event before creating it. ShipmentRepository.find_by_order_id looks the shipment up by order ID and returns None when there is none:

@domain.aggregate
class Shipment:
    shipment_id: Auto(identifier=True)
    order_id: Identifier(required=True)
    customer_id: Identifier(required=True)
    items: List(content_type=Dict)
    status: String(default="pending")


@domain.repository(part_of=Shipment)
class ShipmentRepository:
    def find_by_order_id(self, order_id: str) -> Shipment | None:
        return self.query.filter(order_id=order_id).first


@domain.event_handler(part_of=Order)
class ShipmentEventHandler:
    @handle(OrderPaid)
    def create_shipment(self, event: OrderPaid):
        repo = current_domain.repository_for(Shipment)

        # Check if a shipment already exists for this order
        existing = repo.find_by_order_id(event.order_id)
        if existing:
            return  # Shipment already created

        shipment = Shipment(
            order_id=event.order_id,
            customer_id=event.customer_id,
            items=event.items,
            status="pending",
        )
        repo.add(shipment)

Applying the Pattern: Worked Examples

Example 1: Loyalty Points (Additive Operation)

Loyalty points are additive, so awarding points twice doubles the reward. This requires explicit deduplication.

@domain.entity(part_of="CustomerLoyalty")
class PointTransaction:
    source_id: Identifier(required=True)
    amount: Integer(required=True)
    type: String(required=True)


@domain.aggregate
class CustomerLoyalty:
    customer_id: Identifier(identifier=True)
    points: Integer(default=0)
    point_transactions = HasMany(PointTransaction)

    def award_points(self, order_id: str, amount: int) -> None:
        """Award loyalty points for an order, idempotently."""
        # Check if points were already awarded for this order
        for txn in self.point_transactions:
            if txn.source_id == order_id:
                return  # Already awarded

        self.points += amount
        self.add_point_transactions(
            PointTransaction(
                source_id=order_id,
                amount=amount,
                type="earned",
            )
        )


@domain.event_handler(part_of=Order)
class LoyaltyEventHandler:
    @handle(OrderPlaced)
    def award_points(self, event: OrderPlaced):
        repo = current_domain.repository_for(CustomerLoyalty)
        loyalty = repo.get(event.customer_id)
        # The aggregate method handles idempotency internally
        loyalty.award_points(
            order_id=event.order_id,
            amount=int(event.total),
        )
        repo.add(loyalty)

The idempotency logic runs in the aggregate's award_points method, not in the handler. The handler is thin. The aggregate checks its point_transactions to detect duplicates.

Example 2: Inventory Reservation (Cross-Aggregate Side Effect)

from protean.exceptions import ValidationError

@domain.event(part_of="Inventory")
class InventoryReserved:
    product_id: Identifier(required=True)
    order_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.entity(part_of="Inventory")
class Reservation:
    order_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.aggregate
class Inventory:
    product_id: Identifier(identifier=True)
    available_quantity: Integer(default=0)
    reservations = HasMany(Reservation)

    def reserve(self, order_id: str, quantity: int) -> None:
        """Reserve inventory for an order, idempotently."""
        # Check if already reserved for this order
        for reservation in self.reservations:
            if reservation.order_id == order_id:
                return  # Already reserved

        if self.available_quantity < quantity:
            raise ValidationError(
                {
                    "quantity": [
                        f"Insufficient inventory. Available: {self.available_quantity}"
                    ]
                }
            )

        self.available_quantity -= quantity
        self.add_reservations(
            Reservation(
                order_id=order_id,
                quantity=quantity,
            )
        )

        self.raise_(
            InventoryReserved(
                product_id=self.product_id,
                order_id=order_id,
                quantity=quantity,
            )
        )


@domain.event_handler(part_of=Order)
class InventoryEventHandler:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = current_domain.repository_for(Inventory)
        # One reservation per product: an order can list a product twice
        quantities: dict[str, int] = {}
        for item in event.items:
            product_id = item["product_id"]
            quantities[product_id] = quantities.get(product_id, 0) + item["quantity"]

        for product_id, quantity in quantities.items():
            inventory = repo.get(product_id)
            inventory.reserve(order_id=event.order_id, quantity=quantity)
            repo.add(inventory)

The Reservation entity within the Inventory aggregate acts as a deduplication record. The order_id on each reservation uniquely identifies the source event, making the operation idempotent.

Example 3: Notification (External Side Effect)

External side effects (sending emails, calling APIs) require special care because they cannot be rolled back:

from datetime import UTC, datetime

@domain.aggregate
class Notification:
    notification_id: Auto(identifier=True)
    recipient_id: Identifier(required=True)
    type: String(required=True)
    source_event_id: Identifier(required=True)
    status: String(default="pending")
    sent_at: DateTime()

    def mark_sent(self) -> None:
        self.status = "sent"
        self.sent_at = datetime.now(UTC)


@domain.repository(part_of=Notification)
class NotificationRepository:
    def find_by_source_event_id(self, event_id: str) -> Notification | None:
        return self.query.filter(source_event_id=event_id).first

    def pending(self) -> list[Notification]:
        return self.query.filter(status="pending").all().items


@domain.event_handler(part_of=Order)
class NotificationEventHandler:
    @handle(OrderPlaced)
    def send_confirmation(self, event: OrderPlaced):
        repo = current_domain.repository_for(Notification)

        # Use event ID as the deduplication key
        event_id = event._metadata.headers.id
        existing = repo.find_by_source_event_id(event_id)

        if existing and existing.status == "sent":
            return  # Already sent

        if not existing:
            notification = Notification(
                recipient_id=event.customer_id,
                type="order_confirmation",
                source_event_id=event_id,
                status="pending",
            )
            repo.add(notification)

        # The actual sending happens in a separate process
        # that picks up pending notifications

Instead of sending the email directly (which can't be undone), the handler creates a Notification record. If the event is redelivered, the handler finds the existing notification and skips.

A separate process sends pending notifications and marks them as sent:

def send_email(recipient_id: str, template: str) -> None:
    """Hand the message to your email provider."""
    print(f"Sending {template} to {recipient_id}")


def send_pending_notifications() -> None:
    repo = current_domain.repository_for(Notification)
    for notification in repo.pending():
        send_email(notification.recipient_id, notification.type)
        # Marked as sent, so the next run and a redelivered event both skip it
        notification.mark_sent()
        repo.add(notification)

Once a notification is marked as sent and saved, a later run of send_pending_notifications does not send it again. The sender calls the provider before it saves the status. If the save fails or the process stops between the two, the notification stays pending and the next run sends it again. To rule that out, pass the notification's ID to the provider as an idempotency key, if the provider supports one.

Event Handler vs Projector Idempotency

Projectors and event handlers have different idempotency characteristics:

Projectors

Projectors maintain read-optimized projections. They're typically idempotent by nature because they set the projection's state rather than accumulate it:

@domain.projection
class OrderDashboardProjection:
    order_id: Identifier(identifier=True)
    customer_name: String()
    status: String()
    total: Float()
    placed_at: DateTime()
    shipped_at: DateTime()


@domain.projector(projector_for=OrderDashboardProjection, aggregates=[Order])
class OrderDashboardProjector:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = current_domain.repository_for(OrderDashboardProjection)
        # Upsert pattern: create or update
        try:
            projection = repo.get(event.order_id)
        except ObjectNotFoundError:
            projection = OrderDashboardProjection(order_id=event.order_id)

        projection.customer_name = event.customer_name
        projection.status = "placed"
        projection.total = event.total
        projection.placed_at = event.placed_at
        repo.add(projection)

    @handle(OrderShipped)
    def on_order_shipped(self, event: OrderShipped):
        repo = current_domain.repository_for(OrderDashboardProjection)
        try:
            projection = repo.get(event.order_id)
        except ObjectNotFoundError:
            return  # The order was never projected, so there is nothing to ship
        projection.status = "shipped"
        projection.shipped_at = event.shipped_at
        repo.add(projection)

Each handler sets fields to specific values. Processing the same event twice produces the same projection state.

Event Handlers

Event handlers that modify domain aggregates need more care because aggregates have business rules, invariants, and state machines:

@domain.aggregate
class Account:
    account_id: Identifier(identifier=True)
    balance: Float(default=0.0)
    processed_transfers: List(content_type=String)

    def credit(self, amount: float) -> None:
        self.balance += amount


@domain.event(part_of=Account)
class MoneyDebited:
    account_id: Identifier(required=True)
    target_account_id: Identifier(required=True)
    transfer_id: Identifier(required=True)
    amount: Float(required=True)


@domain.event_handler(part_of=Account)
class AccountEventHandler:
    @handle(MoneyDebited)
    def credit_target(self, event: MoneyDebited):
        repo = current_domain.repository_for(Account)
        target = repo.get(event.target_account_id)

        # Must deduplicate because credit is additive
        if event.transfer_id in target.processed_transfers:
            return

        target.credit(event.amount)
        target.processed_transfers = [*target.processed_transfers, event.transfer_id]
        repo.add(target)

The key difference: projectors typically use set-based operations (naturally idempotent), while event handlers often perform additive operations (require explicit deduplication).

When Perfect Idempotency Is Impractical

Some scenarios make perfect idempotency difficult:

Time-Dependent Operations

If the handler's behavior depends on the current time, two executions at different times may produce different results:

# fragment
def on_order_placed(self, event: OrderPlaced):
    # This is NOT idempotent: running at different times changes the result
    if datetime.now() - event.placed_at > timedelta(hours=24):
        order.flag_as_delayed()

Mitigation: use the event's timestamp, not the current time, for time-dependent decisions:

# fragment
def on_order_placed(self, event: OrderPlaced):
    # Better: use event time, not wall clock
    if event.expected_delivery_date < event.placed_at + timedelta(hours=24):
        order.flag_as_delayed()

External API Calls

If the handler calls an external API that's not idempotent, duplicate processing produces duplicate external effects:

# fragment
def on_order_placed(self, event: OrderPlaced):
    # NOT idempotent: sends duplicate SMS on redelivery
    sms_service.send(event.customer_phone, "Your order is confirmed!")

Mitigation: use the status flag pattern (track whether the side effect occurred) or the outbox pattern (create a record, send asynchronously with deduplication).

Signs Your Handler Isn't Idempotent

Counter increments: stats.count += 1 is not idempotent.
List appends without checking: items.append(new_item) without checking if the item is already present.
Creating records without existence checks: repo.add(new_record) without checking if a record for this event already exists.
Direct external API calls: Sending emails, SMS, or API requests without deduplication.
Generating new identifiers: If the handler creates a new aggregate with Auto(identifier=True), a duplicate delivery creates a second aggregate with a different ID.

Summary

Strategy	When to Use	Complexity	Robustness
Set-based operations	Handler sets state, doesn't accumulate	Low	High
Check-before-act	Additive operations, moderate volume	Medium	High
Processed event tracking	Multiple event types, need audit trail	Medium	High
Global position tracking	Rebuilds that stop at the first failure	Medium	Medium
Upsert	Creating or updating projections	Low	High
Status flags	External side effects	Medium	High

Every event handler must produce the same result whether it processes an event once or multiple times. Prefer naturally idempotent operations. When that's not possible, deduplicate explicitly. The handler, not the framework, is the primary defense against duplicate events.