Delivery Guarantee: No Message Loss, No Duplication

“

Series: S10 Design Patterns · Article 10 | Level: Advanced | Reading Time: 18 min

#TL;DR

The Delivery Guarantee pattern ensures cross-Layer asynchronous messages are neither lost nor duplicated under various failure scenarios. coomia-dip uses three guarantee levels: At-Least-Once, At-Most-Once, and Exactly-Once depending on the scenario.
coomia-dip achieves end-to-end message reliability through Kafka transactional producers, idempotent consumers, the Outbox pattern, and Dead Letter Queues (DLQ).
Combined with Transactional Outbox and CDC (Change Data Capture), coomia-dip guarantees atomicity between database writes and message sends.

#Introduction: Message Reliability in Distributed Systems

In coomia-dip's multi-Layer architecture, Layers communicate via asynchronous messages. But asynchronous messaging faces multiple failure modes:

Code

Failure 1: Producer crash — data written to DB but message not sent to Kafka
Failure 2: Kafka node failure — message sent but not acknowledged
Failure 3: Consumer crash — message read from Kafka but processing incomplete
Failure 4: Network partition — network between producer and Kafka interrupted
Failure 5: Duplicate delivery — Kafka retries cause message sent twice

Different business scenarios have different message reliability requirements. The Delivery Guarantee pattern selects the appropriate guarantee level for each scenario.

#Part 1: Three Guarantee Levels

#1.1 At-Most-Once

Messages are processed at most once. May be lost but never duplicated. Suitable for loss-tolerant scenarios (e.g., monitoring metric collection).

Python

class AtMostOnceProducer:
    async def send(self, topic: str, message: dict) -> None:
        try:
            await self._kafka_producer.send(
                topic=topic, value=self._serialize(message), acks=0,
            )
        except Exception:
            pass  # Fire and forget

class AtMostOnceConsumer:
    async def consume(self, topic: str, handler: callable) -> None:
        async for message in self._kafka_consumer.subscribe(topic):
            await self._kafka_consumer.commit(message.offset)
            try:
                await handler(message.value)
            except Exception:
                pass  # Already committed, won't retry

#1.2 At-Least-Once

Messages are processed at least once. Never lost but may be duplicated. Suitable for most business scenarios — combined with idempotent consumers achieves Exactly-Once effect.

Python

class AtLeastOnceProducer:
    async def send(self, topic: str, message: dict) -> None:
        max_retries = 5
        for attempt in range(max_retries):
            try:
                await self._kafka_producer.send(
                    topic=topic, value=self._serialize(message), acks="all",
                )
                return
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

class AtLeastOnceConsumer:
    async def consume(self, topic: str, handler: callable) -> None:
        async for message in self._kafka_consumer.subscribe(topic):
            try:
                await handler(message.value)
                await self._kafka_consumer.commit(message.offset)
            except Exception:
                await self._error_handler.handle(message)

#1.3 Exactly-Once

Messages are processed exactly once. Never lost, never duplicated. coomia-dip achieves this through Kafka transactions + idempotent consumers:

Python

class ExactlyOnceProducer:
    def __init__(self, transactional_id: str):
        self._producer = KafkaProducer(
            transactional_id=transactional_id,
            enable_idempotence=True,
            acks="all",
        )

    async def send_transactional(self, messages: list[tuple[str, dict]]) -> None:
        await self._producer.begin_transaction()
        try:
            for topic, message in messages:
                await self._producer.send(topic, self._serialize(message))
            await self._producer.commit_transaction()
        except Exception:
            await self._producer.abort_transaction()
            raise

class ExactlyOnceConsumer:
    def __init__(self, idempotency_store: "IdempotencyStore"):
        self._idempotency = idempotency_store

    async def consume(self, topic: str, handler: callable) -> None:
        async for message in self._kafka_consumer.subscribe(topic):
            message_id = message.headers.get("message_id")

            if await self._idempotency.is_processed(message_id):
                await self._kafka_consumer.commit(message.offset)
                continue

            try:
                await handler(message.value)
                await self._idempotency.mark_processed(message_id)
                await self._kafka_consumer.commit(message.offset)
            except Exception:
                await self._error_handler.handle(message)

#Part 2: Transactional Outbox Pattern

#2.1 The Problem: Dual-Write Consistency

Business operations often need to write to a database and send a message simultaneously. Since these two operations are not in the same transaction, inconsistency can occur:

Code

Scenario: Send notification after creating an order
1. Write order to database ✓
2. Send message to Kafka ✗ (network failure)
Result: Order created but notification not sent

#2.2 Outbox Solution

Write messages to an outbox table in the database (within the same transaction as business data), then have a separate process read from the outbox and send to Kafka:

Python

class TransactionalOutbox:
    async def execute_with_outbox(
        self, db_operation: callable, messages: list[dict]
    ) -> None:
        async with self._db.transaction() as tx:
            result = await db_operation(tx)
            for msg in messages:
                await tx.execute(
                    "INSERT INTO outbox (message_id, topic, payload, status, created_at) "
                    "VALUES ($1, $2, $3, 'pending', NOW())",
                    generate_id(), msg["topic"], json.dumps(msg["payload"]),
                )
            return result

class OutboxRelay:
    async def relay_pending(self) -> int:
        pending = await self._db.fetch(
            "SELECT * FROM outbox WHERE status = 'pending' "
            "ORDER BY created_at LIMIT 100"
        )
        sent_count = 0
        for msg in pending:
            try:
                await self._kafka_producer.send(
                    topic=msg["topic"], value=msg["payload"],
                    headers={"message_id": msg["message_id"]},
                )
                await self._db.execute(
                    "UPDATE outbox SET status = 'sent', sent_at = NOW() "
                    "WHERE message_id = $1", msg["message_id"],
                )
                sent_count += 1
            except Exception as e:
                await self._db.execute(
                    "UPDATE outbox SET status = 'failed', error = $2, "
                    "retry_count = retry_count + 1 WHERE message_id = $1",
                    msg["message_id"], str(e),
                )
        return sent_count

#2.3 CDC-Based Outbox

A more efficient approach uses CDC to listen for outbox table changes instead of polling:

Python

class CDCOutboxRelay:
    async def start(self) -> None:
        async for change in self._cdc_client.subscribe("outbox"):
            if change.operation == "INSERT":
                await self._kafka_producer.send(
                    topic=change.new_row["topic"],
                    value=change.new_row["payload"],
                    headers={"message_id": change.new_row["message_id"]},
                )
                await self._db.execute(
                    "UPDATE outbox SET status = 'sent' WHERE message_id = $1",
                    change.new_row["message_id"],
                )

#Part 3: Idempotent Consumers

#3.1 Idempotency Store

Python

class IdempotencyStore:
    async def is_processed(self, message_id: str) -> bool:
        result = await self._redis.get(f"idempotency:{message_id}")
        return result is not None

    async def mark_processed(self, message_id: str, ttl_hours: int = 168) -> None:
        await self._redis.set(
            f"idempotency:{message_id}", "1", ex=ttl_hours * 3600,
        )

    async def mark_processed_with_result(
        self, message_id: str, result: dict, ttl_hours: int = 168
    ) -> None:
        await self._redis.set(
            f"idempotency:{message_id}",
            json.dumps(result),
            ex=ttl_hours * 3600,
        )

    async def get_result(self, message_id: str) -> dict | None:
        result = await self._redis.get(f"idempotency:{message_id}")
        if result and result != "1":
            return json.loads(result)
        return None

#3.2 Business-Level Idempotency

Some operations are naturally idempotent (e.g., setting state), while others are not (e.g., debiting a balance). coomia-dip provides idempotent wrappers for non-idempotent operations:

Python

class BusinessIdempotency:
    async def debit_account_idempotent(
        self, idempotency_key: str, account_id: str, amount: float
    ) -> dict:
        existing = await self._idempotency.get_result(idempotency_key)
        if existing:
            return existing

        result = await self._account_service.debit(account_id, amount)
        await self._idempotency.mark_processed_with_result(idempotency_key, result)
        return result

#Part 4: Dead Letter Queue (DLQ)

#4.1 DLQ Strategy

Messages that repeatedly fail processing enter the DLQ to avoid blocking normal message processing:

Python

class DeadLetterQueueHandler:
    def __init__(self, max_retries: int = 3, retry_delays: list[int] | None = None):
        self._max_retries = max_retries
        self._retry_delays = retry_delays or [60, 300, 3600]

    async def handle_failure(
        self, message: dict, error: Exception, retry_count: int
    ) -> None:
        if retry_count < self._max_retries:
            delay = self._retry_delays[min(retry_count, len(self._retry_delays) - 1)]
            await self._kafka_producer.send(
                topic=f"{message['topic']}.retry",
                value=message,
                headers={
                    "retry_count": str(retry_count + 1),
                    "original_topic": message["topic"],
                    "retry_at": str(time.time() + delay),
                    "error": str(error),
                },
            )
        else:
            await self._kafka_producer.send(
                topic=f"{message['topic']}.dlq",
                value=message,
                headers={
                    "retry_count": str(retry_count),
                    "original_topic": message["topic"],
                    "final_error": str(error),
                    "failed_at": datetime.utcnow().isoformat(),
                },
            )
            await self._alert_service.send(
                severity="warning",
                title=f"Message sent to DLQ: {message['topic']}",
                message=f"Message {message.get('message_id')} failed after {retry_count} retries: {error}",
            )

    async def replay_dlq(self, topic: str, filter_fn: callable | None = None) -> int:
        replayed = 0
        async for message in self._kafka_consumer.subscribe(f"{topic}.dlq"):
            if filter_fn and not filter_fn(message):
                continue
            await self._kafka_producer.send(
                topic=message.headers["original_topic"],
                value=message.value,
                headers={"replayed_from_dlq": "true"},
            )
            replayed += 1
        return replayed

#Part 5: Delivery Guarantee Configuration in coomia-dip

#5.1 Selecting Guarantee Level by Scenario

Python

DELIVERY_CONFIG = {
    "ontology.action.execute": {
        "guarantee": "exactly_once",
        "use_outbox": True,
        "idempotency_ttl_hours": 168,
        "dlq_max_retries": 5,
    },
    "ontology.state.changed": {
        "guarantee": "at_least_once",
        "use_outbox": True,
        "idempotency_ttl_hours": 24,
        "dlq_max_retries": 3,
    },
    "audit.event": {
        "guarantee": "at_least_once",
        "use_outbox": False,
        "dlq_max_retries": 10,
    },
    "metrics.collect": {
        "guarantee": "at_most_once",
        "use_outbox": False,
    },
    "reasoning.cache.invalidate": {
        "guarantee": "at_most_once",
        "use_outbox": False,
    },
}

#5.2 End-to-End Message Tracing

Python

class MessageTracer:
    async def trace_message(self, message_id: str) -> dict:
        return {
            "message_id": message_id,
            "produced_at": await self._get_produce_time(message_id),
            "kafka_partition": await self._get_partition(message_id),
            "consumed_at": await self._get_consume_time(message_id),
            "processed": await self._idempotency.is_processed(message_id),
            "retry_count": await self._get_retry_count(message_id),
            "in_dlq": await self._is_in_dlq(message_id),
            "outbox_status": await self._get_outbox_status(message_id),
        }

#Part 6: Message Ordering Guarantees

#6.1 Partition-Level Ordering

Python

class OrderedMessageProducer:
    async def send_ordered(
        self, topic: str, ordering_key: str, message: dict
    ) -> None:
        partition = self._hash_to_partition(ordering_key, self._num_partitions)
        await self._kafka_producer.send(
            topic=topic, value=self._serialize(message),
            partition=partition, key=ordering_key.encode(),
        )

    def _hash_to_partition(self, key: str, num_partitions: int) -> int:
        import hashlib
        return int(hashlib.md5(key.encode()).hexdigest(), 16) % num_partitions

#Key Takeaways

Three Guarantee Levels: At-Most-Once, At-Least-Once, and Exactly-Once suit different business scenarios
Transactional Outbox: The Outbox table solves dual-write consistency between database writes and message sends
Idempotent Consumers: Idempotency checks are key to achieving Exactly-Once semantics
Dead Letter Queue: DLQ + retry strategies ensure problem messages don't block normal processing
CDC Optimization: CDC-based Outbox Relay is more efficient with lower latency than polling
Partition Ordering: Partition keys guarantee processing order for related messages

#Next Article

In the next article, we will explore the Facade Pattern — how coomia-dip provides a clean, unified API entry point for its complex multi-Layer architecture.

S10-11: Facade Pattern: Unified API Entry Point

#Tags

#DesignPatterns #DeliveryGuarantee #ExactlyOnce #Outbox #CDC #Idempotent #DeadLetterQueue #Kafka