# Kourier Automatic Reconnection: How the Robust AMQP Client Recovers from Network Failures

> Kourier's robust AMQP client automatically reconnects after network failures. Discover how it restores exchanges, queues, bindings, and consumers without manual intervention.

- Repository: [Guimauve Digital/kourier](https://github.com/guimauvedigital/kourier)
- Tags: internals
- Published: 2026-03-02

---

**Kourier’s robust AMQP client automatically recovers from network failures by running a continuous retry loop in `RobustAMQPConnection` that restores channel state—including exchanges, queues, bindings, and consumers—via `RobustAMQPChannel.restore()` without requiring manual intervention.**

The **guimauvedigital/kourier** library provides a fault-tolerant AMQP client designed to handle broker restarts, socket drops, and intermittent network issues transparently. This article examines the source code implementation that enables **Kourier automatic reconnection**, ensuring your applications maintain stable message processing even during infrastructure instability.

## The Core Architecture: Connection and Channel Recovery

Kourier’s resilience strategy splits responsibility between two primary classes. **`RobustAMQPConnection`** manages the low-level TCP socket and orchestrates the retry logic, while **`RobustAMQPChannel`** maintains an in-memory registry of all declared resources and replays them after reconnection.

### Connection-Level Reconnection Logic

When instantiated, `RobustAMQPConnection` launches a **reconnect subscription**—a Kotlin coroutine (`Job`) that continuously executes `connectionFactory()` in a loop. In [`amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/RobustAMQPConnection.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/RobustAMQPConnection.kt), the `connect()` method spawns this long-lived coroutine:

```kotlin
override suspend fun connect() {
    reconnectSubscription?.cancel()
    reconnectSubscription = messageListeningScope.launch {    // ← coroutine lives for connection lifetime
        connectionFactory()                               // ← handles retries
    }
    // wait for the first successful CONNECTED event ...
}

```

The `connectionFactory()` method attempts a normal connection via `super.connect()`. Upon successful handshake, it iterates over all registered robust channels and invokes their `restore()` method to recreate AMQP state. If any step throws an exception—such as when the broker remains offline—the catch block logs the error and the loop continues after an implicit back-off period. The `finally` block marks every channel as **CLOSED**, ensuring the next successful iteration knows to restore them:

```kotlin
// Simplified logic from RobustAMQPConnection.kt lines 56-78
try {
    super.connect()
    // ... authentication ...
    channels.forEach { channel ->
        channel.restore()  // Re-declare exchanges, queues, bindings
    }
} catch (e: Exception) {
    // Log and retry
} finally {
    channels.forEach { it.markClosed() }
}

```

When your application explicitly closes the connection, the subscription cancellation stops the retry loop, preventing unnecessary reconnection attempts after intentional shutdown.

### Channel-Level State Restoration

While the connection handles socket recovery, **`RobustAMQPChannel`** ensures semantic continuity by remembering everything declared on the channel. It maintains maps including `declaredExchanges`, `declaredQueues`, `boundExchanges`, and `consumedQueues` to store the complete configuration history.

When `connectionFactory()` detects a successful reconnection, it calls `channel.restore()` for each managed channel. In [`amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/RobustAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/RobustAMQPChannel.kt), the restoration process follows a strict sequence:

1. **Prepare for restore** – Captures the highest delivery tag seen (`deliveryTagOffsetBeforeRestore`) and clears pending restore flags.
2. **Re-open the channel** – Calls `open()` to establish a fresh AMQP channel on the new connection.
3. **Replay declarations** – Re-applies QoS settings, exchanges, queues, and bindings using the stored descriptors.
4. **Re-register consumers** – Snapshots the `consumedQueues` map, clears it, and restarts each consumer with the original callback functions.

Only after `restoreCompleted.complete(Unit)` signals success does the channel resume normal operation. This guarantees that message consumers resume exactly where they left off, with identical topology and flow control settings.

## Handling Stale Delivery Tags During Reconnection

A critical edge case in **Kourier automatic reconnection** involves delivery tag invalidation. When a consumer reconnects, the broker assigns new delivery tags for incoming messages. Any acknowledgments referencing pre-reconnection tags would trigger `PRECONDITION_FAILED` errors.

`RobustAMQPChannel` solves this by tracking the maximum delivery tag observed before restoration. In [`RobustAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/RobustAMQPChannel.kt) (lines 61-68), the client silently ignores `basicAck`, `basicNack`, and `basicReject` calls where the tag is less than or equal to the saved offset:

```kotlin
// Conceptual implementation from source analysis
if (deliveryTag <= deliveryTagOffsetBeforeRestore) {
    return  // Silently drop stale acknowledgments
}
// Otherwise, proceed with normal acknowledgment

```

This idempotency mechanism prevents application crashes when the network flakes during message processing, allowing safe at-least-once delivery semantics without duplicate processing risks.

## Practical Implementation: Using the Robust Client

Implementing **Kourier automatic reconnection** requires minimal code changes from standard AMQP usage. The robust wrapper handles failure detection and recovery transparently.

### Creating a Robust Connection

Instantiate `RobustAMQPConnection` with your coroutine scope and configuration. The client immediately begins its background reconnection watcher:

```kotlin
import dev.kourier.amqp.robust.RobustAMQPConnection
import dev.kourier.amqp.AMQPConfig
import kotlinx.coroutines.GlobalScope

suspend fun startRobustClient() {
    val config = AMQPConfig(
        host = "broker.example.com",
        port = 5672,
        username = "guest",
        password = "guest",
        // TLS, virtual host, timeout settings...
    )

    // Robust connection automatically retries and restores channels
    val connection = RobustAMQPConnection.create(GlobalScope, config)
    
    // Create a channel with automatic restoration support
    val channel = connection.createChannel(ChannelId(1), frameMax = 0u)
    
    // Declare resources—these are remembered for reconnection replay
    channel.queueDeclare("my-queue", durable = true, exclusive = false, autoDelete = false)
    channel.basicConsume(
        queue = "my-queue",
        consumerTag = "",
        noAck = false,
        onDelivery = { delivery ->
            println("Received: ${delivery.message.body}")
            channel.basicAck(delivery.message.deliveryTag, multiple = false)
        }
    )
}

```

### Publishing During Network Instability

Publishers benefit equally from the robust wrapper. If the connection drops during a `basicPublish` call, the underlying channel restoration completes before the operation resumes:

```kotlin
suspend fun publishWithRecovery(channel: AMQPChannel, payload: ByteArray) {
    // Automatically waits for reconnection if network fails
    channel.basicPublish(
        exchange = "",
        routingKey = "my-queue",
        mandatory = false,
        immediate = false,
        properties = emptyMap(),
        body = payload
    )
}

```

The application never needs to catch `IOException` or manually re-declare `exchanges` and `queues`—the `RobustAMQPConnection` and `RobustAMQPChannel` classes handle the entire lifecycle.

## Summary

- **`RobustAMQPConnection`** manages the reconnection loop via `connectionFactory()`, continuously retrying the socket connection and invoking channel restoration upon success.
- **`RobustAMQPChannel`** persists all AMQP state—exchanges, queues, bindings, QoS, and consumers—in memory and replays them via `restore()` after network recovery.
- **Stale delivery tags** are filtered using `deliveryTagOffsetBeforeRestore` to prevent acknowledgment errors across connection boundaries.
- The implementation requires no manual intervention; applications use standard AMQP methods while the robust layer handles failure recovery transparently in `amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/`.

## Frequently Asked Questions

### What triggers Kourier's automatic reconnection mechanism?

Any network-level interruption triggers the mechanism, including TCP socket drops, broker restarts, or physical network failures. When `super.connect()` throws an exception inside `RobustAMQPConnection.connectionFactory()`, the catch block logs the error and the coroutine loop immediately retries the connection handshake without terminating the application.

### How does Kourier preserve consumer callbacks after a network failure?

`RobustAMQPChannel` stores consumer configurations in the `consumedQueues` map alongside their callback functions. During `restore()`, the method snapshots this registry, clears the active consumer list, and re-invokes `basicConsume()` for each entry with the original `onDelivery` and `onCanceled` handlers. This ensures message processing resumes with identical business logic after reconnection.

### Does Kourier handle message acknowledgments during reconnection?

Yes. The client tracks the highest delivery tag received before disconnection using `deliveryTagOffsetBeforeRestore`. Any `basicAck`, `basicNack`, or `basicReject` calls referencing tags at or below this threshold are silently discarded rather than sent to the broker. This prevents `PRECONDITION_FAILED` errors that would otherwise occur when acknowledging messages from the previous connection session.

### How do I stop the reconnection loop in Kourier?

Call `connection.close()` explicitly on your `RobustAMQPConnection` instance. This cancels the `reconnectSubscription` coroutine (`Job`) in [`RobustAMQPConnection.kt`](https://github.com/guimauvedigital/kourier/blob/main/RobustAMQPConnection.kt) (lines 85-88), terminating the retry loop and preventing further connection attempts. Without explicit closure, the coroutine continues running indefinitely, attempting to reconnect even during application shutdown.